[jaeger-v2] Align Cassandra Storage Config With OTEL

cmd/jaeger/config-cassandra.yaml

@@ -23,11 +23,15 @@ extensions:
           keyspace: "jaeger_v1_dc1"
           username: "cassandra"
           password: "cassandra"
+          tls:
+            insecure: true
       another_storage:
         cassandra:
           keyspace: "jaeger_v1_dc1"
           username: "cassandra"
           password: "cassandra"
+          tls:
+            insecure: true
 receivers:
   otlp:
     protocols:

pkg/cassandra/config/config.go

@@ -5,36 +5,36 @@
 package config
 
 import (
+	"context"
 	"fmt"
 	"time"
 
 	"github.com/asaskevich/govalidator"
 	"github.com/gocql/gocql"
-	"go.uber.org/zap"
+	"go.opentelemetry.io/collector/config/configtls"
 
 	"github.com/jaegertracing/jaeger/pkg/cassandra"
 	gocqlw "github.com/jaegertracing/jaeger/pkg/cassandra/gocql"
-	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 )
 
 // Configuration describes the configuration properties needed to connect to a Cassandra cluster
 type Configuration struct {
-	Servers              []string       `valid:"required,url" mapstructure:"servers"`
-	Keyspace             string         `mapstructure:"keyspace"`
-	LocalDC              string         `mapstructure:"local_dc"`
-	ConnectionsPerHost   int            `mapstructure:"connections_per_host"`
-	Timeout              time.Duration  `mapstructure:"-"`
-	ConnectTimeout       time.Duration  `mapstructure:"connection_timeout"`
-	ReconnectInterval    time.Duration  `mapstructure:"reconnect_interval"`
-	SocketKeepAlive      time.Duration  `mapstructure:"socket_keep_alive"`
-	MaxRetryAttempts     int            `mapstructure:"max_retry_attempts"`
-	ProtoVersion         int            `mapstructure:"proto_version"`
-	Consistency          string         `mapstructure:"consistency"`
-	DisableCompression   bool           `mapstructure:"disable_compression"`
-	Port                 int            `mapstructure:"port"`
-	Authenticator        Authenticator  `mapstructure:",squash"`
-	DisableAutoDiscovery bool           `mapstructure:"-"`
-	TLS                  tlscfg.Options `mapstructure:"tls"`
+	Servers              []string               `valid:"required,url" mapstructure:"servers"`
+	Keyspace             string                 `mapstructure:"keyspace"`
+	LocalDC              string                 `mapstructure:"local_dc"`
+	ConnectionsPerHost   int                    `mapstructure:"connections_per_host"`
+	Timeout              time.Duration          `mapstructure:"-"`
+	ConnectTimeout       time.Duration          `mapstructure:"connection_timeout"`
+	ReconnectInterval    time.Duration          `mapstructure:"reconnect_interval"`
+	SocketKeepAlive      time.Duration          `mapstructure:"socket_keep_alive"`
+	MaxRetryAttempts     int                    `mapstructure:"max_retry_attempts"`
+	ProtoVersion         int                    `mapstructure:"proto_version"`
+	Consistency          string                 `mapstructure:"consistency"`
+	DisableCompression   bool                   `mapstructure:"disable_compression"`
+	Port                 int                    `mapstructure:"port"`
+	Authenticator        Authenticator          `mapstructure:",squash"`
+	DisableAutoDiscovery bool                   `mapstructure:"-"`
+	TLS                  configtls.ClientConfig `mapstructure:"tls"`
 }
 
 func DefaultConfiguration() Configuration {
@@ -92,12 +92,12 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
 
 // SessionBuilder creates new cassandra.Session
 type SessionBuilder interface {
-	NewSession(logger *zap.Logger) (cassandra.Session, error)
+	NewSession() (cassandra.Session, error)
 }
 
 // NewSession creates a new Cassandra session
-func (c *Configuration) NewSession(logger *zap.Logger) (cassandra.Session, error) {
-	cluster, err := c.NewCluster(logger)
+func (c *Configuration) NewSession() (cassandra.Session, error) {
+	cluster, err := c.NewCluster()
 	if err != nil {
 		return nil, err
 	}
@@ -109,7 +109,7 @@ func (c *Configuration) NewSession(logger *zap.Logger) (cassandra.Session, error
 }
 
 // NewCluster creates a new gocql cluster from the configuration
-func (c *Configuration) NewCluster(logger *zap.Logger) (*gocql.ClusterConfig, error) {
+func (c *Configuration) NewCluster() (*gocql.ClusterConfig, error) {
 	cluster := gocql.NewCluster(c.Servers...)
 	cluster.Keyspace = c.Keyspace
 	cluster.NumConns = c.ConnectionsPerHost
@@ -150,11 +150,11 @@ func (c *Configuration) NewCluster(logger *zap.Logger) (*gocql.ClusterConfig, er
 			AllowedAuthenticators: c.Authenticator.Basic.AllowedAuthenticators,
 		}
 	}
-	tlsCfg, err := c.TLS.Config(logger)
+	tlsCfg, err := c.TLS.LoadTLSConfig(context.Background())
 	if err != nil {
 		return nil, err
 	}
-	if c.TLS.Enabled {
+	if !c.TLS.Insecure {
 		cluster.SslOpts = &gocql.SslOptions{
 			Config: tlsCfg,
 		}
@@ -167,10 +167,6 @@ func (c *Configuration) NewCluster(logger *zap.Logger) (*gocql.ClusterConfig, er
 	return cluster, nil
 }
 
-func (c *Configuration) Close() error {
-	return c.TLS.Close()
-}
-
 func (c *Configuration) String() string {
 	return fmt.Sprintf("%+v", *c)
 }

plugin/storage/cassandra/factory.go

@@ -137,14 +137,14 @@ func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger)
 	f.archiveMetricsFactory = metricsFactory.Namespace(metrics.NSOptions{Name: "cassandra-archive", Tags: nil})
 	f.logger = logger
 
-	primarySession, err := f.primaryConfig.NewSession(logger)
+	primarySession, err := f.primaryConfig.NewSession()
 	if err != nil {
 		return err
 	}
 	f.primarySession = primarySession
 
 	if f.archiveConfig != nil {
-		archiveSession, err := f.archiveConfig.NewSession(logger)
+		archiveSession, err := f.archiveConfig.NewSession()
 		if err != nil {
 			return err
 		}
@@ -251,12 +251,7 @@ func (f *Factory) Close() error {
 		f.archiveSession.Close()
 	}
 
-	var errs []error
-	if cfg := f.Options.Get(archiveStorageConfig); cfg != nil {
-		errs = append(errs, cfg.TLS.Close())
-	}
-	errs = append(errs, f.Options.GetPrimary().TLS.Close())
-	return errors.Join(errs...)
+	return nil
 }
 
 func (f *Factory) Purge(_ context.Context) error {

plugin/storage/cassandra/factory_test.go

@@ -34,7 +34,7 @@ func newMockSessionBuilder(session *mocks.Session, err error) *mockSessionBuilde
 	}
 }
 
-func (m *mockSessionBuilder) NewSession(*zap.Logger) (cassandra.Session, error) {
+func (m *mockSessionBuilder) NewSession() (cassandra.Session, error) {
 	return m.session, m.err
 }
 

plugin/storage/cassandra/options.go

@@ -247,11 +247,12 @@ func (cfg *NamespaceConfig) initFromViper(v *viper.Viper) {
 	cfg.Authenticator.Basic.AllowedAuthenticators = strings.Split(authentication, ",")
 	cfg.DisableCompression = v.GetBool(cfg.namespace + suffixDisableCompression)
 	var err error
-	cfg.TLS, err = tlsFlagsConfig.InitFromViper(v)
+	tlsCfg, err := tlsFlagsConfig.InitFromViper(v)
 	if err != nil {
 		// TODO refactor to be able to return error
 		log.Fatal(err)
 	}
+	cfg.TLS = tlsCfg.ToOtelClientConfig()
 }
 
 // GetPrimary returns primary configuration.

plugin/storage/cassandra/options_test.go

@@ -102,7 +102,7 @@ func TestDefaultTlsHostVerify(t *testing.T) {
 	opts.InitFromViper(v)
 
 	primary := opts.GetPrimary()
-	assert.False(t, primary.TLS.SkipHostVerify)
+	assert.False(t, primary.TLS.InsecureSkipVerify)
 }
 
 func TestEmptyBlackWhiteLists(t *testing.T) {

plugin/storage/cassandra/savetracetest/main.go

@@ -30,7 +30,7 @@ func main() {
 		ProtoVersion:       4,
 		Keyspace:           "jaeger_v1_test",
 	}
-	cqlSession, err := cConfig.NewSession(logger)
+	cqlSession, err := cConfig.NewSession()
 	if err != nil {
 		logger.Fatal("Cannot create Cassandra session", zap.Error(err))
 	}