[jaeger-v2] Align Cassandra Storage Config With OTEL

cmd/jaeger/config-cassandra.yaml

@@ -20,14 +20,24 @@ extensions:
     backends:
       some_storage:
         cassandra:
-          keyspace: "jaeger_v1_dc1"
-          username: "cassandra"
-          password: "cassandra"
+          schema: 
+            keyspace: "jaeger_v1_dc1"
+          connection:
+            auth: 
+              username: "cassandra"
+              password: "cassandra"
+            tls:
+              insecure: true
       another_storage:
         cassandra:
-          keyspace: "jaeger_v1_dc1"
-          username: "cassandra"
-          password: "cassandra"
+          schema: 
+            keyspace: "jaeger_v1_dc1"
+          connection: 
+            auth:
+              username: "cassandra"
+              password: "cassandra"
+            tls:
+              insecure: true
 receivers:
   otlp:
     protocols:

cmd/jaeger/internal/extension/jaegerstorage/config_test.go

@@ -83,7 +83,7 @@ backends:
 `)
 	cfg := createDefaultConfig().(*Config)
 	require.NoError(t, conf.Unmarshal(cfg))
-	assert.NotEmpty(t, cfg.Backends["some_storage"].Cassandra.Primary.Servers)
+	assert.NotEmpty(t, cfg.Backends["some_storage"].Cassandra.Primary.Connection.Servers)
 }
 
 func TestConfigDefaultElasticsearch(t *testing.T) {

pkg/cassandra/config/config.go

@@ -5,48 +5,44 @@
 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
+// 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"`
+	Schema     Schema     `mapstructure:"schema"`
+	Connection Connection `mapstructure:"connection"`
 }
 
-func DefaultConfiguration() Configuration {
-	return Configuration{
-		Servers:            []string{"127.0.0.1"},
-		Port:               9042,
-		MaxRetryAttempts:   3,
-		Keyspace:           "jaeger_v1_test",
-		ProtoVersion:       4,
-		ConnectionsPerHost: 2,
-		ReconnectInterval:  60 * time.Second,
-	}
+type Connection struct {
+	Servers              []string               `mapstructure:"servers" valid:"required,url" `
+	LocalDC              string                 `mapstructure:"local_dc"`
+	Port                 int                    `mapstructure:"port"`
+	DisableAutoDiscovery bool                   `mapstructure:"disable_auto_discovery"`
+	ConnectionsPerHost   int                    `mapstructure:"connections_per_host"`
+	ReconnectInterval    time.Duration          `mapstructure:"reconnect_interval"`
+	SocketKeepAlive      time.Duration          `mapstructure:"socket_keep_alive"`
+	MaxRetryAttempts     int                    `mapstructure:"max_retry_attempts"`
+	TLS                  configtls.ClientConfig `mapstructure:"tls"`
+	QueryTimeout         time.Duration          `mapstructure:"query_timeout"`
+	ConnectTimeout       time.Duration          `mapstructure:"connection_timeout"`
+	Authenticator        Authenticator          `mapstructure:"auth"`
+	ProtoVersion         int                    `mapstructure:"proto_version"`
+	Consistency          string                 `mapstructure:"consistency"`
+}
+
+type Schema struct {
+	Keyspace           string `mapstructure:"keyspace"`
+	DisableCompression bool   `mapstructure:"disable_compression"`
 }
 
 // Authenticator holds the authentication properties needed to connect to a Cassandra cluster
@@ -62,42 +58,59 @@
 	AllowedAuthenticators []string `yaml:"allowed_authenticators" mapstructure:"allowed_authenticators"`
 }
 
+func DefaultConfiguration() Configuration {
+	return Configuration{
+		Connection: Connection{
+			Servers:            []string{"127.0.0.1"},
+			Port:               9042,
+			MaxRetryAttempts:   3,
+			ProtoVersion:       4,
+			ConnectionsPerHost: 2,
+			ReconnectInterval:  60 * time.Second,
+		},
+		Schema: Schema{
+			Keyspace: "jaeger_v1_test",
+		},
+	}
+}
+
 // ApplyDefaults copies settings from source unless its own value is non-zero.
 func (c *Configuration) ApplyDefaults(source *Configuration) {
-	if c.ConnectionsPerHost == 0 {
-		c.ConnectionsPerHost = source.ConnectionsPerHost
+	if c.Connection.ConnectionsPerHost == 0 {
+		c.Connection.ConnectionsPerHost = source.Connection.ConnectionsPerHost
 	}
-	if c.MaxRetryAttempts == 0 {
-		c.MaxRetryAttempts = source.MaxRetryAttempts
+	if c.Connection.MaxRetryAttempts == 0 {
+		c.Connection.MaxRetryAttempts = source.Connection.MaxRetryAttempts
 	}
-	if c.Timeout == 0 {
-		c.Timeout = source.Timeout
+	if c.Connection.QueryTimeout == 0 {
+		c.Connection.QueryTimeout = source.Connection.QueryTimeout
 	}
-	if c.ReconnectInterval == 0 {
-		c.ReconnectInterval = source.ReconnectInterval
+	if c.Connection.ReconnectInterval == 0 {
+		c.Connection.ReconnectInterval = source.Connection.ReconnectInterval
 	}
-	if c.Port == 0 {
-		c.Port = source.Port
+	if c.Connection.Port == 0 {
+		c.Connection.Port = source.Connection.Port
 	}
-	if c.Keyspace == "" {
-		c.Keyspace = source.Keyspace
+	if c.Connection.ProtoVersion == 0 {
+		c.Connection.ProtoVersion = source.Connection.ProtoVersion
 	}
-	if c.ProtoVersion == 0 {
-		c.ProtoVersion = source.ProtoVersion
+	if c.Connection.SocketKeepAlive == 0 {
+		c.Connection.SocketKeepAlive = source.Connection.SocketKeepAlive
 	}
-	if c.SocketKeepAlive == 0 {
-		c.SocketKeepAlive = source.SocketKeepAlive
+
+	if c.Schema.Keyspace == "" {
+		c.Schema.Keyspace = source.Schema.Keyspace
 	}
 }
 
 // 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,68 +122,64 @@
 }
 
 // NewCluster creates a new gocql cluster from the configuration
-func (c *Configuration) NewCluster(logger *zap.Logger) (*gocql.ClusterConfig, error) {
-	cluster := gocql.NewCluster(c.Servers...)
-	cluster.Keyspace = c.Keyspace
-	cluster.NumConns = c.ConnectionsPerHost
-	cluster.Timeout = c.Timeout
-	cluster.ConnectTimeout = c.ConnectTimeout
-	cluster.ReconnectInterval = c.ReconnectInterval
-	cluster.SocketKeepalive = c.SocketKeepAlive
-	if c.ProtoVersion > 0 {
-		cluster.ProtoVersion = c.ProtoVersion
+func (c *Configuration) NewCluster() (*gocql.ClusterConfig, error) {
+	cluster := gocql.NewCluster(c.Connection.Servers...)
+	cluster.Keyspace = c.Schema.Keyspace
+	cluster.NumConns = c.Connection.ConnectionsPerHost
+	cluster.Timeout = c.Connection.QueryTimeout
+	cluster.ConnectTimeout = c.Connection.ConnectTimeout
+	cluster.ReconnectInterval = c.Connection.ReconnectInterval
+	cluster.SocketKeepalive = c.Connection.SocketKeepAlive
+	if c.Connection.ProtoVersion > 0 {
+		cluster.ProtoVersion = c.Connection.ProtoVersion
 	}
-	if c.MaxRetryAttempts > 1 {
-		cluster.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: c.MaxRetryAttempts - 1}
+	if c.Connection.MaxRetryAttempts > 1 {
+		cluster.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: c.Connection.MaxRetryAttempts - 1}
 	}
-	if c.Port != 0 {
-		cluster.Port = c.Port
+	if c.Connection.Port != 0 {
+		cluster.Port = c.Connection.Port
 	}
 
-	if !c.DisableCompression {
+	if !c.Schema.DisableCompression {
 		cluster.Compressor = gocql.SnappyCompressor{}
 	}
 
-	if c.Consistency == "" {
+	if c.Connection.Consistency == "" {
 		cluster.Consistency = gocql.LocalOne
 	} else {
-		cluster.Consistency = gocql.ParseConsistency(c.Consistency)
+		cluster.Consistency = gocql.ParseConsistency(c.Connection.Consistency)
 	}
 
 	fallbackHostSelectionPolicy := gocql.RoundRobinHostPolicy()
-	if c.LocalDC != "" {
-		fallbackHostSelectionPolicy = gocql.DCAwareRoundRobinPolicy(c.LocalDC)
+	if c.Connection.LocalDC != "" {
+		fallbackHostSelectionPolicy = gocql.DCAwareRoundRobinPolicy(c.Connection.LocalDC)
 	}
 	cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(fallbackHostSelectionPolicy, gocql.ShuffleReplicas())
 
-	if c.Authenticator.Basic.Username != "" && c.Authenticator.Basic.Password != "" {
+	if c.Connection.Authenticator.Basic.Username != "" && c.Connection.Authenticator.Basic.Password != "" {
 		cluster.Authenticator = gocql.PasswordAuthenticator{
-			Username:              c.Authenticator.Basic.Username,
-			Password:              c.Authenticator.Basic.Password,
-			AllowedAuthenticators: c.Authenticator.Basic.AllowedAuthenticators,
+			Username:              c.Connection.Authenticator.Basic.Username,
+			Password:              c.Connection.Authenticator.Basic.Password,
+			AllowedAuthenticators: c.Connection.Authenticator.Basic.AllowedAuthenticators,
 		}
 	}
-	tlsCfg, err := c.TLS.Config(logger)
+	tlsCfg, err := c.Connection.TLS.LoadTLSConfig(context.Background())
 	if err != nil {
 		return nil, err
 	}
-	if c.TLS.Enabled {
+	if !c.Connection.TLS.Insecure {
 		cluster.SslOpts = &gocql.SslOptions{
 			Config: tlsCfg,
 		}
 	}
 	// If tunneling connection to C*, disable cluster autodiscovery features.
-	if c.DisableAutoDiscovery {
+	if c.Connection.DisableAutoDiscovery {
 		cluster.DisableInitialHostLookup = true
 		cluster.IgnorePeerAddr = true
 	}
 	return cluster, nil
 }
 
-func (c *Configuration) Close() error {
-	return c.TLS.Close()
-}
-
 func (c *Configuration) String() string {
 	return fmt.Sprintf("%+v", *c)
 }

pkg/cassandra/config/config_test.go

@@ -0,0 +1,48 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidate_ReturnsErrorWhenInvalid(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  *Configuration
+	}{
+		{
+			name: "missing required fields",
+			cfg:  &Configuration{},
+		},
+		{
+			name: "require fields in invalid format",
+			cfg: &Configuration{
+				Connection: Connection{
+					Servers: []string{"not a url"},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := test.cfg.Validate()
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestValidate_DoesNotReturnErrorWhenRequiredFieldsSet(t *testing.T) {
+	cfg := Configuration{
+		Connection: Connection{
+			Servers: []string{"localhost:9200"},
+		},
+	}
+
+	err := cfg.Validate()
+	require.NoError(t, err)
+}

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
 }
 
@@ -195,7 +195,9 @@ func TestNewFactoryWithConfig(t *testing.T) {
 		opts := &Options{
 			Primary: NamespaceConfig{
 				Configuration: cassandraCfg.Configuration{
-					Servers: []string{"localhost:9200"},
+					Connection: cassandraCfg.Connection{
+						Servers: []string{"localhost:9200"},
+					},
 				},
 			},
 		}
@@ -215,7 +217,9 @@ func TestNewFactoryWithConfig(t *testing.T) {
 		opts := &Options{
 			Primary: NamespaceConfig{
 				Configuration: cassandraCfg.Configuration{
-					Servers: []string{"localhost:9200"},
+					Connection: cassandraCfg.Connection{
+						Servers: []string{"localhost:9200"},
+					},
 				},
 			},
 		}