cassandra exporter keyspace check and dynamic timeout (open-telemetry…

…#27681) Check keyspace if exist and dynamic timeout for connection open-telemetry#27633 --------- Co-authored-by: Curtis Robert <[email protected]> Co-authored-by: Ziqi Zhao <[email protected]> Co-authored-by: Alex Boten <[email protected]>
codeboten · Nov 28, 2023 · 4aeacae · 4aeacae
1 parent e74c823
commit 4aeacae
Show file tree

Hide file tree

Showing 9 changed files with 51 additions and 12 deletions.
diff --git a/.chloggen/cassandra-exists-and-timeout.yaml b/.chloggen/cassandra-exists-and-timeout.yaml
@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'bug_fix'
+
+# 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: "Exist check for keyspace and dynamic timeout"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [27633]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user, api]
diff --git a/exporter/cassandraexporter/README.md b/exporter/cassandraexporter/README.md
@@ -18,12 +18,13 @@ The following settings can be optionally configured:
 
 - `dsn` The Cassandra server DSN (Data Source Name), for example `127.0.0.1`.
  reference: [https://pkg.go.dev/github.com/gocql/gocql](https://pkg.go.dev/github.com/gocql/gocql)
-- `port` (default = 9042) The Cassandra server Port 
+- `port` (default = 9042): The Cassandra server port
+- `timeout` (default = 10s): The Cassandra server connection timeout
 - `keyspace` (default = otel): The keyspace name.
 - `trace_table` (default = otel_spans): The table name for traces.
-- `replication` (default = class: SimpleStrategy, replication_factor: 1) The strategy of
+- `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
+- `compression` (default = LZ4Compressor): https://cassandra.apache.org/doc/latest/cassandra/operating/compression.html
 
 ## Example
 
@@ -32,6 +33,7 @@ exporters:
  cassandra:
  dsn: 127.0.0.1
  port: 9042
+ timeout: 10s
  keyspace: "otel"
  trace_table: "otel_spans"
  replication:

diff --git a/exporter/cassandraexporter/config.go b/exporter/cassandraexporter/config.go
@@ -2,15 +2,17 @@
 // SPDX-License-Identifier: Apache-2.0
 
 package cassandraexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/cassandraexporter"
+import "time"
 
 type Config struct {
- DSN string `mapstructure:"dsn"`
- Port int `mapstructure:"port"`
- Keyspace string `mapstructure:"keyspace"`
- TraceTable string `mapstructure:"trace_table"`
- LogsTable string `mapstructure:"logs_table"`
- Replication Replication `mapstructure:"replication"`
- Compression Compression `mapstructure:"compression"`
+ DSN string `mapstructure:"dsn"`
+ Port int `mapstructure:"port"`
+ Timeout time.Duration `mapstructure:"timeout"`
+ Keyspace string `mapstructure:"keyspace"`
+ TraceTable string `mapstructure:"trace_table"`
+ LogsTable string `mapstructure:"logs_table"`
+ Replication Replication `mapstructure:"replication"`
+ Compression Compression `mapstructure:"compression"`
 }
 
 type Replication struct {

diff --git a/exporter/cassandraexporter/constants.go b/exporter/cassandraexporter/constants.go
@@ -5,7 +5,7 @@ package cassandraexporter // import "github.com/open-telemetry/opentelemetry-col
 
 const (
  // language=SQL
- createDatabaseSQL = `CREATE KEYSPACE %s WITH REPLICATION = { 'class' : '%s', 'replication_factor' : %d };`
+ createDatabaseSQL = `CREATE KEYSPACE IF NOT EXISTS %s WITH REPLICATION = { 'class' : '%s', 'replication_factor' : %d };`
  // language=SQL
  createEventTypeSQL = `CREATE TYPE IF NOT EXISTS %s.Events (Timestamp Date, Name text, Attributes map<text, text>);`
  // language=SQL

diff --git a/exporter/cassandraexporter/example/otel-collector-config.yml b/exporter/cassandraexporter/example/otel-collector-config.yml
@@ -6,6 +6,8 @@ receivers:
 exporters:
  cassandra:
  dsn: 127.0.0.1
+ port: 9042
+ timeout: 10s
  keyspace: "otel"
  trace_table: "otel_spans"
  logs_table: "otel_logs"
@@ -14,7 +16,6 @@ exporters:
  replication_factor: 1
  compression:
  algorithm: "ZstdCompressor"
-
 service:
  pipelines:
  traces:

diff --git a/exporter/cassandraexporter/exporter_logs.go b/exporter/cassandraexporter/exporter_logs.go
@@ -29,6 +29,7 @@ func newLogsExporter(logger *zap.Logger, cfg *Config) (*logsExporter, error) {
  cluster.Keyspace = cfg.Keyspace
  cluster.Consistency = gocql.Quorum
  cluster.Port = cfg.Port
+ cluster.Timeout = cfg.Timeout
 
  if err != nil {
  return nil, err
@@ -42,6 +43,7 @@ func initializeLogKernel(cfg *Config) error {
  cluster := gocql.NewCluster(cfg.DSN)
  cluster.Consistency = gocql.Quorum
  cluster.Port = cfg.Port
+ cluster.Timeout = cfg.Timeout
 
  session, err := cluster.CreateSession()
  if err != nil {

diff --git a/exporter/cassandraexporter/exporter_traces.go b/exporter/cassandraexporter/exporter_traces.go
@@ -28,6 +28,7 @@ func newTracesExporter(logger *zap.Logger, cfg *Config) (*tracesExporter, error)
  cluster.Keyspace = cfg.Keyspace
  cluster.Consistency = gocql.Quorum
  cluster.Port = cfg.Port
+ cluster.Timeout = cfg.Timeout
 
  if err != nil {
  return nil, err
@@ -41,6 +42,7 @@ func initializeTraceKernel(cfg *Config) error {
  cluster := gocql.NewCluster(cfg.DSN)
  cluster.Consistency = gocql.Quorum
  cluster.Port = cfg.Port
+ cluster.Timeout = cfg.Timeout
 
  session, err := cluster.CreateSession()
  if err != nil {

diff --git a/exporter/cassandraexporter/factory.go b/exporter/cassandraexporter/factory.go
@@ -6,6 +6,7 @@ package cassandraexporter // import "github.com/open-telemetry/opentelemetry-col
 import (
  "context"
  "fmt"
+ "time"
 
  "go.opentelemetry.io/collector/component"
  "go.opentelemetry.io/collector/exporter"
@@ -26,6 +27,7 @@ func createDefaultConfig() component.Config {
  return &Config{
  DSN: "127.0.0.1",
  Port: 9042,
+ Timeout: 10 * time.Second,
  Keyspace: "otel",
  TraceTable: "otel_spans",
  LogsTable: "otel_logs",

diff --git a/exporter/cassandraexporter/testdata/config.yaml b/exporter/cassandraexporter/testdata/config.yaml
@@ -2,6 +2,7 @@ cassandra:
  dsn: 127.0.0.1
  keyspace: "otel"
  trace_table: "otel_spans"
+ timeout: 10s
  logs_table: "otel_logs"
  replication:
  class: "SimpleStrategy"