feat(signozspanmetricsconnector): addition of traces to metrics connector Part-2

Makefile

@@ -3,6 +3,7 @@ REPONAME ?= signoz
 IMAGE_NAME ?= signoz-otel-collector
 MIGRATOR_IMAGE_NAME ?= signoz-schema-migrator
 CONFIG_FILE ?= ./config/default-config.yaml
+CONFIG_FILE_WITH_CONNECTOR ?= ./config/default-config-with-connector.yaml
 DOCKER_TAG ?= latest
 
 GOOS ?= $(shell go env GOOS)
@@ -52,6 +53,10 @@ build-all: amd64 arm64
 run:
 	go run cmd/signozotelcollector/main.go --config ${CONFIG_FILE}
 
+.PHONY: run-with-connector
+run-with-connector:
+	go run cmd/signozotelcollector/main.go --config ${CONFIG_FILE_WITH_CONNECTOR}
+
 .PHONY: fmt
 fmt:
 	@echo Running go fmt on query service ...

components/components.go

@@ -8,7 +8,6 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/roundrobinconnector"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/routingconnector"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/servicegraphconnector"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/sumconnector"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/alertmanagerexporter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter"
@@ -171,6 +170,7 @@ import (
 	"go.uber.org/multierr"
 
 	"github.com/SigNoz/signoz-otel-collector/connectors/signozmeterconnector"
+	"github.com/SigNoz/signoz-otel-collector/connectors/signozspanmetricsconnector"
 	"github.com/SigNoz/signoz-otel-collector/exporter/clickhouselogsexporter"
 	"github.com/SigNoz/signoz-otel-collector/exporter/clickhousetracesexporter"
 	"github.com/SigNoz/signoz-otel-collector/exporter/jsontypeexporter"
@@ -391,7 +391,7 @@ func Components() (otelcol.Factories, error) {
 		roundrobinconnector.NewFactory(),
 		routingconnector.NewFactory(),
 		servicegraphconnector.NewFactory(),
-		spanmetricsconnector.NewFactory(),
+		signozspanmetricsconnector.NewFactory(),
 		sumconnector.NewFactory(),
 		signozmeterconnector.NewFactory(),
 	}

config/default-config-with-connector.yaml

@@ -0,0 +1,97 @@
+receivers:
+  otlp/spanmetrics:
+    protocols:
+      grpc:
+        endpoint: "localhost:12345"
+  otlp:
+    protocols:
+      grpc:
+      http:
+  jaeger:
+    protocols:
+      grpc:
+      thrift_http:
+  hostmetrics:
+    collection_interval: 30s
+    scrapers:
+      cpu:
+      load:
+      memory:
+      disk:
+      filesystem:
+      network:
+processors:
+  batch:
+    send_batch_size: 1000
+    timeout: 10s
+
+connectors:
+  signozspanmetrics/delta:
+    metrics_flush_interval: 60s
+    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 50ms, 100ms, 250ms, 500ms, 1000ms, 1400ms, 2000ms, 5s, 10s, 20s, 40s, 60s ]
+    dimensions_cache_size: 100000
+    aggregation_temporality: AGGREGATION_TEMPORALITY_DELTA
+    enable_exp_histogram: true
+    skip_spans_older_than: 24h
+    dimensions:
+      - name: service.namespace
+        default: default
+      - name: deployment.environment
+        default: default
+      # This is added to ensure the uniqueness of the timeseries
+      # Otherwise, identical timeseries produced by multiple replicas of
+      # collectors result in incorrect APM metrics
+      - name: signoz.collector.id
+      - name: service.version
+      - name: browser.platform
+      - name: browser.mobile
+      - name: k8s.cluster.name
+      - name: k8s.node.name
+      - name: k8s.namespace.name
+      - name: host.name
+      - name: host.type
+      - name: container.name
+  # memory_limiter:
+  #   # 80% of maximum memory up to 2G
+  #   limit_mib: 1500
+  #   # 25% of limit up to 2G
+  #   spike_limit_mib: 512
+  #   check_interval: 5s
+  #
+  #   # 50% of the maximum memory
+  #   limit_percentage: 50
+  #   # 20% of max memory usage spike expected
+  #   spike_limit_percentage: 20
+  # queued_retry:
+  #   num_workers: 4
+  #   queue_size: 100
+  #   retry_on_failure: true
+extensions:
+  zpages: {}
+exporters:
+  clickhousetraces:
+    datasource: tcp://localhost:9000/signoz_traces
+    retry_on_failure:
+      enabled: true
+      initial_interval: 5s
+      max_interval: 30s
+      max_elapsed_time: 300s
+    sending_queue:
+      enabled: true
+      queue_size: 100
+      num_consumers: 5
+  signozclickhousemetrics:
+    dsn: tcp://localhost:9000/signoz_metrics
+  prometheus:
+    endpoint: "0.0.0.0:8889"
+service:
+  extensions: [zpages]
+  pipelines:
+    traces:
+      receivers: [jaeger, otlp]
+      processors: [batch]
+      exporters: [signozspanmetrics/delta, clickhousetraces]
+    metrics:
+      receivers: [otlp, hostmetrics, signozspanmetrics/delta]
+      processors: [batch]
+      exporters: [signozclickhousemetrics]

connectors/signozspanmetricsconnector/config.go

@@ -0,0 +1,81 @@
+package signozspanmetricsconnector
+
+import (
+	"time"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+// Config defines the configuration options for spanmetricsconnector.
+type Config struct {
+	// Dimensions defines the list of additional dimensions on top of the provided:
+	// - service.name
+	// - span.kind
+	// - span.kind
+	// - status.code
+	// - collector.instance.id This dimensions never added unless enable feature-gate connector.spanmetrics.includeCollectorInstanceID
+	// The dimensions will be fetched from the span's attributes. Examples of some conventionally used attributes:
+	// https://github.com/open-telemetry/opentelemetry-collector/blob/main/model/semconv/opentelemetry.go.
+	Dimensions []Dimension `mapstructure:"dimensions"`
+
+	// LatencyHistogramBuckets is the list of durations representing latency histogram buckets.
+	// See defaultLatencyHistogramBucketsMs in processor.go for the default value.
+	LatencyHistogramBuckets []time.Duration `mapstructure:"latency_histogram_buckets"`
+
+	// ExcludePatterns defines the list of patterns to exclude from the metrics.
+	ExcludePatterns []ExcludePattern `mapstructure:"exclude_patterns"`
+
+	// DimensionsCacheSize defines the size of cache for storing Dimensions, which helps to avoid cache memory growing
+	// indefinitely over the lifetime of the collector.
+	// Optional. See defaultDimensionsCacheSize in connector.go for the default value.
+	// Deprecated [v0.130.0]:  Please use AggregationCardinalityLimit instead
+	DimensionsCacheSize int `mapstructure:"dimensions_cache_size"`
+
+	AggregationTemporality string `mapstructure:"aggregation_temporality"`
+
+	// skipSanitizeLabel if enabled, labels that start with _ are not sanitized
+	skipSanitizeLabel bool
+
+	// MetricsEmitInterval is the time period between when metrics are flushed or emitted to the configured MetricsExporter.
+	MetricsFlushInterval time.Duration `mapstructure:"metrics_flush_interval"`
+
+	// TimeBucketInterval is the time interval for bucketing spans based on their start timestamp.
+	// Spans are grouped into time buckets based on when they started, not when they are processed.
+	// Default is 1 minute.
+	TimeBucketInterval time.Duration `mapstructure:"time_bucket_interval"`
+
+	EnableExpHistogram bool `mapstructure:"enable_exp_histogram"`
+
+	MaxServicesToTrack             int `mapstructure:"max_services_to_track"`
+	MaxOperationsToTrackPerService int `mapstructure:"max_operations_to_track_per_service"`
+
+	// SkipSpansOlderThan defines the staleness window for skipping late-arriving spans.
+	// Spans with start time older than now - SkipSpansOlderThan are skipped.
+	// Default is 24 hours if not set.
+	SkipSpansOlderThan time.Duration `mapstructure:"skip_spans_older_than"`
+}
+
+// GetAggregationTemporality converts the string value given in the config into a AggregationTemporality.
+// Returns cumulative, unless delta is correctly specified.
+func (c Config) GetAggregationTemporality() pmetric.AggregationTemporality {
+	if c.AggregationTemporality == delta {
+		return pmetric.AggregationTemporalityDelta
+	}
+	return pmetric.AggregationTemporalityCumulative
+}
+
+// GetTimeBucketInterval returns the configured time bucket interval, or the default if not set.
+func (c Config) GetTimeBucketInterval() time.Duration {
+	if c.TimeBucketInterval == 0 {
+		return defaultTimeBucketInterval
+	}
+	return c.TimeBucketInterval
+}
+
+// GetSkipSpansOlderThan returns the configured staleness window or a default of 24 hours.
+func (c Config) GetSkipSpansOlderThan() time.Duration {
+	if c.SkipSpansOlderThan <= 0 {
+		return defaultSkipSpansOlderThan
+	}
+	return c.SkipSpansOlderThan
+}