Add Redis TLS server name and cluster mode support (#522)

- Add set_tls_server_name to override the TLS server name (SNI) on the
  Redis client connection.
- Add cluster_mode to run the producer against a Redis Cluster.
- Include vin and tx_type in Redis publish-error logs.

Author: agbpatro

config/config.go

@@ -7,7 +7,9 @@ import (
 	_ "embed" //Used for default CAs
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"strings"
 	"time"
@@ -155,13 +157,15 @@ type Kinesis struct {
 
 // 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"`
+	Addrs                 []string      `json:"addrs"`
+	Username              string        `json:"username,omitempty"`
+	Password              string        `json:"password,omitempty"`
+	DB                    int           `json:"db,omitempty"`
+	TLS                   *TLS          `json:"tls,omitempty"`
+	OverrideTLSServerName bool          `json:"override_tls_server_name,omitempty"`
+	Pool                  *RedisPool    `json:"pool,omitempty"`
+	PublishTimeout        time.Duration `json:"publish_timeout,omitempty"`
+	ClusterMode           bool          `json:"cluster_mode,omitempty"`
 
 	// PublishVINTopics, when true, always publishes on the VIN set-key
 	// channel.
@@ -191,6 +195,9 @@ func (r *Redis) options() (*goredis.UniversalOptions, error) {
 	if err != nil {
 		return nil, err
 	}
+	if r.OverrideTLSServerName && tlsConfig != nil {
+		tlsConfig.ServerName = r.getServerName()
+	}
 	password := r.Password
 	if envPassword := os.Getenv("REDIS_PASSWORD"); envPassword != "" {
 		password = envPassword
@@ -203,6 +210,9 @@ func (r *Redis) options() (*goredis.UniversalOptions, error) {
 		DB:        r.DB,
 		TLSConfig: tlsConfig,
 	}
+	if r.ClusterMode {
+		options.IsClusterMode = true
+	}
 	if r.Pool != nil {
 		options.PoolSize = r.Pool.PoolSize
 		options.MinIdleConns = r.Pool.MinIdleConns
@@ -217,6 +227,24 @@ func (r *Redis) options() (*goredis.UniversalOptions, error) {
 	return options, nil
 }
 
+func (r *Redis) getServerName() string {
+	if len(r.Addrs) == 0 {
+		return ""
+	}
+
+	addr := r.Addrs[0]
+	// A scheme-qualified addr (e.g. rediss://host:6379) parses cleanly here.
+	if parsedURL, err := url.Parse(addr); err == nil && parsedURL.Hostname() != "" {
+		return parsedURL.Hostname()
+	}
+	// A bare host:port (the common case) is mis-parsed by url.Parse as
+	// scheme:opaque, so strip the port directly.
+	if host, _, err := net.SplitHostPort(addr); err == nil {
+		return host
+	}
+	return addr
+}
+
 //go:embed files/eng_ca.crt
 var defaultEngCA []byte
 

datastore/redis/redis.go

@@ -86,6 +86,9 @@ func NewProducer(options *redis.UniversalOptions, publishTimeout time.Duration,
 		timeout:             timeout,
 	}
 
+	if err := client.Ping(context.Background()).Err(); err != nil {
+		return nil, fmt.Errorf("redis_connect_error %w", err)
+	}
 	producer.logger.ActivityLog("redis_registered", logrus.LogInfo{"namespace": namespace, "publish_vin_topics": publishVINTopics, "subscriber_set_prefix": subscriberSetPrefix})
 	return producer, nil
 }
@@ -97,15 +100,15 @@ func (p *Producer) Produce(entry *telemetry.Record) {
 	payload := entry.Payload()
 	channels, err := p.channelsForRecord(entry)
 	if err != nil {
-		p.ReportError("redis_err", err, logrus.LogInfo{"channel": "", "txid": entry.Txid})
+		p.ReportError("redis_err", err, logrus.LogInfo{"vin": entry.Vin, "tx_type": entry.TxType, "txid": entry.Txid})
 		metricsRegistry.errorCount.Inc(map[string]string{"record_type": entry.TxType})
 		return
 	}
 	for _, channel := range channels {
 		ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
 		defer cancel()
 		if err := p.client.Publish(ctx, channel, payload).Err(); err != nil {
-			p.ReportError("redis_err", err, logrus.LogInfo{"channel": channel, "txid": entry.Txid})
+			p.ReportError("redis_err", err, logrus.LogInfo{"channel": channel, "vin": entry.Vin, "tx_type": entry.TxType, "txid": entry.Txid})
 			metricsRegistry.errorCount.Inc(map[string]string{"record_type": entry.TxType})
 			return
 		}