[exporter/cassandra] Added authorization by username and password (#2…

…7841)

Added authorization by username and password for the Casandra exporter

Fixes #27827

---------

Co-authored-by: Curtis Robert <92119472+crobert-1@users.noreply.github.com>

.chloggen/cassandraexporter-add-authorization.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: cassandraexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: added authorization by username and password
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [27827]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

exporter/cassandraexporter/README.md

@@ -25,6 +25,7 @@ The following settings can be optionally configured:
 - `replication` (default = class: SimpleStrategy, replication_factor: 1): The strategy of
   replication. https://cassandra.apache.org/doc/4.1/cassandra/architecture/dynamo.html#replication-strategy
 - `compression` (default = LZ4Compressor): https://cassandra.apache.org/doc/latest/cassandra/operating/compression.html
+- `auth` (default = username: "", password: "") Authorization for the Cassandra.
 
 ## Example
 
@@ -41,4 +42,7 @@ exporters:
       replication_factor: 1
     compression:
       algorithm: "ZstdCompressor"
+    auth:
+      username: "your-username"
+      password: "your-password"
 ```

exporter/cassandraexporter/config.go

@@ -2,7 +2,11 @@
 // SPDX-License-Identifier: Apache-2.0
 
 package cassandraexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/cassandraexporter"
-import "time"
+import (
+	"time"
+
+	"go.opentelemetry.io/collector/config/configopaque"
+)
 
 type Config struct {
 	DSN         string        `mapstructure:"dsn"`
@@ -13,6 +17,7 @@ type Config struct {
 	LogsTable   string        `mapstructure:"logs_table"`
 	Replication Replication   `mapstructure:"replication"`
 	Compression Compression   `mapstructure:"compression"`
+	Auth        Auth          `mapstructure:"auth"`
 }
 
 type Replication struct {
@@ -23,3 +28,8 @@ type Replication struct {
 type Compression struct {
 	Algorithm string `mapstructure:"algorithm"`
 }
+
+type Auth struct {
+	UserName string              `mapstructure:"username"`
+	Password configopaque.String `mapstructure:"password"`
+}

exporter/cassandraexporter/exporter_logs.go

@@ -6,6 +6,7 @@ package cassandraexporter // import "github.com/open-telemetry/opentelemetry-col
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
@@ -24,13 +25,16 @@ type logsExporter struct {
 }
 
 func newLogsExporter(logger *zap.Logger, cfg *Config) (*logsExporter, error) {
-	cluster := gocql.NewCluster(cfg.DSN)
-	session, err := cluster.CreateSession()
+	cluster, err := newCluster(cfg)
+	if err != nil {
+		return nil, err
+	}
 	cluster.Keyspace = cfg.Keyspace
 	cluster.Consistency = gocql.Quorum
 	cluster.Port = cfg.Port
 	cluster.Timeout = cfg.Timeout
 
+	session, err := cluster.CreateSession()
 	if err != nil {
 		return nil, err
 	}
@@ -40,7 +44,10 @@ func newLogsExporter(logger *zap.Logger, cfg *Config) (*logsExporter, error) {
 
 func initializeLogKernel(cfg *Config) error {
 	ctx := context.Background()
-	cluster := gocql.NewCluster(cfg.DSN)
+	cluster, err := newCluster(cfg)
+	if err != nil {
+		return err
+	}
 	cluster.Consistency = gocql.Quorum
 	cluster.Port = cfg.Port
 	cluster.Timeout = cfg.Timeout
@@ -64,6 +71,25 @@ func initializeLogKernel(cfg *Config) error {
 	return nil
 }
 
+func newCluster(cfg *Config) (*gocql.ClusterConfig, error) {
+	cluster := gocql.NewCluster(cfg.DSN)
+	if cfg.Auth.UserName != "" && cfg.Auth.Password == "" {
+		return nil, errors.New("empty auth.password")
+	}
+	if cfg.Auth.Password != "" && cfg.Auth.UserName == "" {
+		return nil, errors.New("empty auth.username")
+	}
+	if cfg.Auth.UserName != "" && cfg.Auth.Password != "" {
+		cluster.Authenticator = gocql.PasswordAuthenticator{
+			Username: cfg.Auth.UserName,
+			Password: string(cfg.Auth.Password),
+		}
+	}
+	cluster.Consistency = gocql.Quorum
+	cluster.Port = cfg.Port
+	return cluster, nil
+}
+
 func (e *logsExporter) Start(_ context.Context, _ component.Host) error {
 	initializeErr := initializeLogKernel(e.cfg)
 	return initializeErr

exporter/cassandraexporter/exporter_logs_test.go

@@ -0,0 +1,66 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package cassandraexporter
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/gocql/gocql"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewCluster(t *testing.T) {
+	testCases := map[string]struct {
+		cfg                   *Config
+		expectedAuthenticator gocql.Authenticator
+		expectedErr           error
+	}{
+		"empty_auth": {
+			cfg:                   withDefaultConfig(),
+			expectedAuthenticator: nil,
+		},
+		"empty_username": {
+			cfg: withDefaultConfig(func(config *Config) {
+				config.Auth.Password = "pass"
+			}),
+			expectedAuthenticator: nil,
+			expectedErr:           errors.New("empty auth.username"),
+		},
+		"empty_password": {
+			cfg: withDefaultConfig(func(config *Config) {
+				config.Auth.UserName = "user"
+			}),
+			expectedAuthenticator: nil,
+			expectedErr:           errors.New("empty auth.password"),
+		},
+		"success_auth": {
+			cfg: withDefaultConfig(func(config *Config) {
+				config.Auth.UserName = "user"
+				config.Auth.Password = "pass"
+			}),
+			expectedAuthenticator: gocql.PasswordAuthenticator{
+				Username: "user",
+				Password: "pass",
+			},
+		},
+	}
+	for name, test := range testCases {
+		t.Run(name, func(t *testing.T) {
+			c, err := newCluster(test.cfg)
+			if err == nil {
+				require.Equal(t, test.expectedAuthenticator, c.Authenticator)
+			}
+			require.Equal(t, test.expectedErr, err)
+		})
+	}
+}
+
+func withDefaultConfig(fns ...func(*Config)) *Config {
+	cfg := createDefaultConfig().(*Config)
+	for _, fn := range fns {
+		fn(cfg)
+	}
+	return cfg
+}

exporter/cassandraexporter/exporter_traces.go

@@ -23,13 +23,16 @@ type tracesExporter struct {
 }
 
 func newTracesExporter(logger *zap.Logger, cfg *Config) (*tracesExporter, error) {
-	cluster := gocql.NewCluster(cfg.DSN)
-	session, err := cluster.CreateSession()
+	cluster, err := newCluster(cfg)
+	if err != nil {
+		return nil, err
+	}
 	cluster.Keyspace = cfg.Keyspace
 	cluster.Consistency = gocql.Quorum
 	cluster.Port = cfg.Port
 	cluster.Timeout = cfg.Timeout
 
+	session, err := cluster.CreateSession()
 	if err != nil {
 		return nil, err
 	}
@@ -39,7 +42,10 @@ func newTracesExporter(logger *zap.Logger, cfg *Config) (*tracesExporter, error)
 
 func initializeTraceKernel(cfg *Config) error {
 	ctx := context.Background()
-	cluster := gocql.NewCluster(cfg.DSN)
+	cluster, err := newCluster(cfg)
+	if err != nil {
+		return err
+	}
 	cluster.Consistency = gocql.Quorum
 	cluster.Port = cfg.Port
 	cluster.Timeout = cfg.Timeout

exporter/cassandraexporter/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.92.0
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/collector/component v0.92.1-0.20240112172857-83d463ceba06
+	go.opentelemetry.io/collector/config/configopaque v0.92.1-0.20240112172857-83d463ceba06
 	go.opentelemetry.io/collector/confmap v0.92.1-0.20240112172857-83d463ceba06
 	go.opentelemetry.io/collector/exporter v0.92.1-0.20240112172857-83d463ceba06
 	go.opentelemetry.io/collector/pdata v1.0.2-0.20240112172857-83d463ceba06