Add Redis pub/sub datastore

- Add a Redis dispatcher that publishes telemetry records over Redis
  pub/sub, selectable per record type via the "redis" dispatch rule.
- Route each record to the per-VIN subscriber channels in a sorted set
  (configurable prefix); entries scored with an epoch lease expiry are
  purged when past due. With publish_vin_topics, also publish to the VIN
  channel <namespace>_<txtype>_{<vin>}.

Author: agbpatro

README.md

@@ -81,6 +81,16 @@ For ease of installation and operation, run Fleet Telemetry on Kubernetes or a s
       "V": "custom_stream_name"
     }
   },
+  "redis": { // Redis pub/sub config
+    "addrs": ["redis:6379"], // one or more addresses (cluster/sentinel supported)
+    "username": string - optional,
+    "password": string - optional,
+    "db": int - optional,
+    "subscriber_set_prefix": string - optional; names the per-VIN subscriber sorted set. When empty, the sorted set is not consulted,
+    "publish_vin_topics": bool - additionally publish each record to the VIN channel namespace_topic_{vin},
+    "tls": { "ca_file": string, "server_cert": string, "server_key": string }, // optional
+    "pool": { "pool_size": int, "min_idle_conns": int, "conn_max_lifetime": int } // optional
+  },
   "rate_limit": {
     "enabled": bool,
     "message_limit": int - ex.: 1000
@@ -149,7 +159,7 @@ spec:
 Vehicles must be running firmware version 2023.20.6 or later.  Some older model S/X are not supported.
 
 ## Personalized Backends/Dispatchers
-Dispatchers handle vehicle data processing upon its arrival at Fleet Telemetry servers. They can be of any type, from distributed message queues to  STDOUT logger.  Here is a list of the currently supported [dispatchers](./telemetry/producer.go#L10-L19)::
+Dispatchers handle vehicle data processing upon its arrival at Fleet Telemetry servers. They can be of any type, from distributed message queues to  STDOUT logger.  Here is a list of the currently supported [dispatchers](./telemetry/producer.go#L13-L26)::
 * Kafka (preferred): Configure with the config.json file.  See implementation here: [config/config.go](./config/config.go)
   * Topics will need to be created for \*prefix\*`_V`,\*prefix\*`_connectivity` and \*prefix\*`_alerts`. The default prefix is `tesla`
 * Kinesis: Configure with standard [AWS env variables and config files](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html). The default AWS credentials and config files are: `~/.aws/credentials` and `~/.aws/config`.
@@ -161,12 +171,17 @@ Dispatchers handle vehicle data processing upon its arrival at Fleet Telemetry s
 * ZMQ: Configure with the config.json file.  See implementation here: [config/config.go](./config/config.go)
 * MQTT: Configure using the config.json file. See implementation in [config/config.go](./config/config.go)
   * See detailed MQTT information in the [MQTT README](./datastore/mqtt/README.md)
+* Redis: Publishes records to Redis Pub/Sub channels. Configure with the config.json file. See implementation here: [datastore/redis/redis.go](./datastore/redis/redis.go)
+  * When `subscriber_set_prefix` is set, the producer reads a per-VIN sorted set at `<subscriber_set_prefix>_`\*namespace\*`_`\*topic\*`_{`\*vin\*`}` (e.g. `consumer_tesla_V_{<vin>}`). Sorted-set scores are subscriber lease expiries (epoch seconds); entries scored below the current time are purged before each publish, and the payload is published to every surviving member channel.
+  * If `publish_vin_topics` is enabled, the payload is additionally published to the VIN channel \*namespace\*`_`\*topic\*`_{`\*vin\*`}` (e.g. `tesla_V_{<vin>}`). Member channels and the VIN channel both receive the record.
+  * At least one of `subscriber_set_prefix` or `publish_vin_topics` must be configured — otherwise no records could ever be published and the server fails to start.
+  * Supports TLS via the `tls` block and connection-pool tuning via the `pool` block.
 * Logger: This is a simple STDOUT logger that serializes the protos to json.
 
 >NOTE: To add a new dispatcher, please provide integration tests and updated documentation. To serialize dispatcher data as json instead of protobufs, add a config `transmit_decoded_records` and set value to `true` as shown [here](config/test_configs_test.go#L186)
 
 ## Reliable Acks
-Fleet Telemetry can send ack messages back to the vehicle. This is useful for applications that need to ensure the data was received and processed. To enable this feature, set `reliable_ack_sources` to one of configured dispatchers (`kafka`,`kinesis`,`pubsub`,`zmq`, `mqtt`) in the config file. Reliable acks can only be set to one dispatcher per recordType. See [here](./test/integration/config.json#L8) for sample config.
+Fleet Telemetry can send ack messages back to the vehicle. This is useful for applications that need to ensure the data was received and processed. To enable this feature, set `reliable_ack_sources` to one of configured dispatchers (`kafka`,`kinesis`,`pubsub`,`zmq`, `mqtt`, `redis`) in the config file. Reliable acks can only be set to one dispatcher per recordType. See [here](./test/integration/config.json#L8) for sample config.
 
 ## Detecting Vehicle Connectivity Changes
 On the vehicle, Fleet Telemetry client behave similarly to how the connectivity engine for vehicle commands. Therefore we can use Fleet Telemetry connectivity event to assume when a vehicle is online. Note that it is a proxy, but if configured properly Fleet Telemetry connectivity time should match vehicle connectivity state in 99%+. To enable connectivity events simply add the `connectivity` records in the list of events in [server_config.json](./examples/server_config.json) file:

config/config.go

@@ -16,12 +16,14 @@ import (
 	githubairbrake "github.com/airbrake/gobrake/v5"
 
 	confluent "github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	goredis "github.com/redis/go-redis/v9"
 	githublogrus "github.com/sirupsen/logrus"
 
 	"github.com/teslamotors/fleet-telemetry/datastore/googlepubsub"
 	"github.com/teslamotors/fleet-telemetry/datastore/kafka"
 	"github.com/teslamotors/fleet-telemetry/datastore/kinesis"
 	"github.com/teslamotors/fleet-telemetry/datastore/mqtt"
+	redisdatastore "github.com/teslamotors/fleet-telemetry/datastore/redis"
 	"github.com/teslamotors/fleet-telemetry/datastore/simple"
 	"github.com/teslamotors/fleet-telemetry/datastore/zmq"
 	logrus "github.com/teslamotors/fleet-telemetry/logger"
@@ -71,6 +73,9 @@ type Config struct {
 	// ZMQ configures a zeromq socket
 	ZMQ *zmq.Config `json:"zmq,omitempty"`
 
+	// Redis configures a Redis pub/sub producer
+	Redis *Redis `json:"redis,omitempty"`
+
 	// Namespace defines a prefix for the kafka/pubsub topic
 	Namespace string `json:"namespace,omitempty"`
 
@@ -148,6 +153,65 @@ type Kinesis struct {
 	Streams      map[string]string `json:"streams,omitempty"`
 }
 
+// Redis is a configuration for the Redis pub/sub producer.
+type Redis struct {
+	Addrs          []string      `json:"addrs"`
+	Username       string        `json:"username,omitempty"`
+	Password       string        `json:"password,omitempty"`
+	DB             int           `json:"db,omitempty"`
+	TLS            *TLS          `json:"tls,omitempty"`
+	Pool           *RedisPool    `json:"pool,omitempty"`
+	PublishTimeout time.Duration `json:"publish_timeout,omitempty"`
+
+	// PublishVINTopics, when true, always publishes on the VIN set-key
+	// channel.
+	PublishVINTopics bool `json:"publish_vin_topics,omitempty"`
+
+	// SubscriberSetPrefix names the per-VIN sorted set of subscriber channels
+	// (<prefix>_<namespace>_<txtype>_{<vin>}). When empty, the sorted set is not
+	// consulted and records route only via the VIN channel if set.
+	SubscriberSetPrefix string `json:"subscriber_set_prefix,omitempty"`
+}
+
+// RedisPool configures the Redis connection pool. Zero values use go-redis defaults.
+type RedisPool struct {
+	PoolSize        int           `json:"pool_size,omitempty"`
+	MinIdleConns    int           `json:"min_idle_conns,omitempty"`
+	MaxIdleConns    int           `json:"max_idle_conns,omitempty"`
+	MaxActiveConns  int           `json:"max_active_conns,omitempty"`
+	ReadTimeout     time.Duration `json:"read_timeout,omitempty"`
+	WriteTimeout    time.Duration `json:"write_timeout,omitempty"`
+	PoolTimeout     time.Duration `json:"pool_timeout,omitempty"`
+	ConnMaxIdleTime time.Duration `json:"conn_max_idle_time,omitempty"`
+	ConnMaxLifetime time.Duration `json:"conn_max_lifetime,omitempty"`
+}
+
+func (r *Redis) options() (*goredis.UniversalOptions, error) {
+	tlsConfig, err := r.TLS.ClientTLSConfig()
+	if err != nil {
+		return nil, err
+	}
+	options := &goredis.UniversalOptions{
+		Addrs:     r.Addrs,
+		Username:  r.Username,
+		Password:  r.Password,
+		DB:        r.DB,
+		TLSConfig: tlsConfig,
+	}
+	if r.Pool != nil {
+		options.PoolSize = r.Pool.PoolSize
+		options.MinIdleConns = r.Pool.MinIdleConns
+		options.MaxIdleConns = r.Pool.MaxIdleConns
+		options.MaxActiveConns = r.Pool.MaxActiveConns
+		options.PoolTimeout = r.Pool.PoolTimeout
+		options.ReadTimeout = r.Pool.ReadTimeout
+		options.WriteTimeout = r.Pool.WriteTimeout
+		options.ConnMaxIdleTime = r.Pool.ConnMaxIdleTime
+		options.ConnMaxLifetime = r.Pool.ConnMaxLifetime
+	}
+	return options, nil
+}
+
 //go:embed files/eng_ca.crt
 var defaultEngCA []byte
 
@@ -161,39 +225,40 @@ type TLS struct {
 	ServerKey  string `json:"server_key"`
 }
 
-// AirbrakeTLSConfig return the TLS config needed for connecting with airbrake server
-func (c *Config) AirbrakeTLSConfig() (*tls.Config, error) {
-	if c.Airbrake.TLS == nil {
+// ClientTLSConfig builds a client-side *tls.Config from an optional cert pair
+// (mutual TLS) and CA file (server verification). A nil receiver yields a nil
+// config so callers can treat "no TLS" uniformly.
+func (t *TLS) ClientTLSConfig() (*tls.Config, error) {
+	if t == nil {
 		return nil, nil
 	}
-	caPath := c.Airbrake.TLS.CAFile
-	certPath := c.Airbrake.TLS.ServerCert
-	keyPath := c.Airbrake.TLS.ServerKey
 	tlsConfig := &tls.Config{}
-	if certPath != "" && keyPath != "" {
-		cert, err := tls.LoadX509KeyPair(certPath, keyPath)
+	if t.ServerCert != "" && t.ServerKey != "" {
+		cert, err := tls.LoadX509KeyPair(t.ServerCert, t.ServerKey)
 		if err != nil {
-			return nil, fmt.Errorf("can't properly load cert pair (%s, %s): %s", certPath, keyPath, err.Error())
+			return nil, fmt.Errorf("can't properly load cert pair (%s, %s): %s", t.ServerCert, t.ServerKey, err.Error())
 		}
 		tlsConfig.Certificates = []tls.Certificate{cert}
-		// TODO remove the lint bypass
-		// nolint:staticcheck
-		tlsConfig.BuildNameToCertificate()
 	}
 
-	if caPath != "" {
-		clientCACert, err := os.ReadFile(caPath)
+	if t.CAFile != "" {
+		caCert, err := os.ReadFile(t.CAFile)
 		if err != nil {
-			return nil, fmt.Errorf("can't properly load ca cert (%s): %s", caPath, err.Error())
+			return nil, fmt.Errorf("can't properly load ca cert (%s): %s", t.CAFile, err.Error())
 		}
-		clientCertPool := x509.NewCertPool()
-		clientCertPool.AppendCertsFromPEM(clientCACert)
-		tlsConfig.RootCAs = clientCertPool
+		caCertPool := x509.NewCertPool()
+		caCertPool.AppendCertsFromPEM(caCert)
+		tlsConfig.RootCAs = caCertPool
 	}
 
 	return tlsConfig, nil
 }
 
+// AirbrakeTLSConfig return the TLS config needed for connecting with airbrake server
+func (c *Config) AirbrakeTLSConfig() (*tls.Config, error) {
+	return c.Airbrake.TLS.ClientTLSConfig()
+}
+
 // VinsToTrack to track incoming signals in promemetheus
 func (c *Config) VinsToTrack() map[string]struct{} {
 	output := make(map[string]struct{}, 0)
@@ -346,6 +411,21 @@ func (c *Config) ConfigureProducers(airbrakeHandler *airbrake.Handler, logger *l
 		producers[telemetry.MQTT] = mqttProducer
 	}
 
+	if _, ok := requiredDispatchers[telemetry.Redis]; ok {
+		if c.Redis == nil {
+			return nil, nil, errors.New("expected Redis to be configured")
+		}
+		options, err := c.Redis.options()
+		if err != nil {
+			return nil, nil, err
+		}
+		redisProducer, err := redisdatastore.NewProducer(options, c.Redis.PublishTimeout, c.Redis.PublishVINTopics, c.Redis.SubscriberSetPrefix, c.Namespace, c.prometheusEnabled(), c.MetricCollector, airbrakeHandler, c.AckChan, reliableAckSources[telemetry.Redis], logger)
+		if err != nil {
+			return nil, nil, err
+		}
+		producers[telemetry.Redis] = redisProducer
+	}
+
 	dispatchProducerRules := make(map[string][]telemetry.Producer)
 	for recordName, dispatchRules := range c.Records {
 		var dispatchFuncs []telemetry.Producer