[connector/servicegraph] Simplify servicegraph connector config arguments

connector/servicegraphconnector/config.go

@@ -45,7 +45,7 @@ type Config struct {
 	// MetricsFlushInterval is the interval at which metrics are flushed to the exporter.
 	// If set to 0, metrics are flushed on every received batch of traces.
 	// Default is 60s if unset.
-	MetricsFlushInterval *time.Duration `mapstructure:"metrics_flush_interval"`
+	MetricsFlushInterval time.Duration `mapstructure:"metrics_flush_interval"`
 
 	// DatabaseNameAttributes is the attribute name list of attributes need to match used to identify the database name from span attributes, the higher the front, the higher the priority.
 	// The default value is {"db.name"}.
@@ -78,5 +78,13 @@ func (c *Config) Validate() error {
 		return errors.New("use either `latency_histogram_buckets` or `exponential_histogram_max_size`")
 	}
 
+	if c.StoreExpirationLoop <= 0 {
+		return errors.New("`store_expiration_loop` must be positive")
+	}
+
+	if c.CacheLoop <= 0 {
+		return errors.New("`cache_loop` must be positive")
+	}
+
 	return nil
 }

connector/servicegraphconnector/config.schema.yaml

@@ -38,7 +38,6 @@ properties:
       format: duration
   metrics_flush_interval:
     description: MetricsFlushInterval is the interval at which metrics are flushed to the exporter. If set to 0, metrics are flushed on every received batch of traces. Default is 60s if unset.
-    x-pointer: true
     type: string
     format: duration
   metrics_timestamp_offset:

connector/servicegraphconnector/config_test.go

@@ -35,9 +35,11 @@ func TestLoadConfig(t *testing.T) {
 				TTL:      time.Second,
 				MaxItems: 10,
 			},
-			CacheLoop:              time.Minute,
-			StoreExpirationLoop:    2 * time.Second,
-			DatabaseNameAttributes: []string{"db.name"},
+			CacheLoop:                 time.Minute,
+			StoreExpirationLoop:       2 * time.Second,
+			VirtualNodePeerAttributes: []string{"peer.service", "db.name", "db.system"},
+			MetricsFlushInterval:      60 * time.Second,
+			DatabaseNameAttributes:    []string{"db.name"},
 		},
 		cfg.Connectors[component.NewID(metadata.Type)],
 	)

connector/servicegraphconnector/connector.go

@@ -20,9 +20,6 @@ import (
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/processor"
-	conventionsv125 "go.opentelemetry.io/otel/semconv/v1.25.0"
-	conventionsv128 "go.opentelemetry.io/otel/semconv/v1.28.0"
-	conventions "go.opentelemetry.io/otel/semconv/v1.38.0"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/servicegraphconnector/internal/metadata"
@@ -41,19 +38,12 @@ const (
 
 var (
 	legacyDefaultLatencyHistogramBuckets = []float64{
-		2, 4, 6, 8, 10, 50, 100, 200, 400, 800, 1000, 1400, 2000, 5000, 10_000, 15_000,
+		2, 4, 6, 8, 10, 50, 100, 200, 400, 800, 1000, 1400, 2000, 5000, 10000, 15000,
 	}
+
 	defaultLatencyHistogramBuckets = []float64{
 		0.002, 0.004, 0.006, 0.008, 0.01, 0.05, 0.1, 0.2, 0.4, 0.8, 1, 1.4, 2, 5, 10, 15,
 	}
-
-	defaultPeerAttributes = []string{
-		string(conventions.PeerServiceKey), string(conventionsv125.DBNameKey), string(conventionsv128.DBSystemKey),
-	}
-
-	defaultDatabaseNameAttributes = []string{string(conventionsv125.DBNameKey)}
-
-	defaultMetricsFlushInterval = 60 * time.Second // 1 DPM
 )
 
 type metricSeries struct {
@@ -96,6 +86,16 @@ type serviceGraphConnector struct {
 func newConnector(set component.TelemetrySettings, config component.Config, next consumer.Metrics) (*serviceGraphConnector, error) {
 	pConfig := config.(*Config)
 
+	if pConfig.MetricsFlushInterval.Nanoseconds() <= 0 {
+		set.Logger.Warn("MetricsFlushInterval is set to 0, metrics will be flushed on every received batch of traces")
+	}
+
+	telemetryBuilder, err := metadata.NewTelemetryBuilder(set)
+	if err != nil {
+		return nil, err
+	}
+
+	// Compute histogram bounds based on configuration
 	var bounds []float64
 	if pConfig.ExponentialHistogramMaxSize == 0 {
 		bounds = defaultLatencyHistogramBuckets
@@ -107,33 +107,6 @@ func newConnector(set component.TelemetrySettings, config component.Config, next
 		}
 	}
 
-	if pConfig.CacheLoop <= 0 {
-		pConfig.CacheLoop = time.Minute
-	}
-
-	if pConfig.StoreExpirationLoop <= 0 {
-		pConfig.StoreExpirationLoop = 2 * time.Second
-	}
-
-	if pConfig.VirtualNodePeerAttributes == nil {
-		pConfig.VirtualNodePeerAttributes = defaultPeerAttributes
-	}
-
-	if len(pConfig.DatabaseNameAttributes) == 0 {
-		pConfig.DatabaseNameAttributes = defaultDatabaseNameAttributes
-	}
-
-	if pConfig.MetricsFlushInterval == nil {
-		pConfig.MetricsFlushInterval = &defaultMetricsFlushInterval
-	} else if pConfig.MetricsFlushInterval.Nanoseconds() <= 0 {
-		set.Logger.Warn("MetricsFlushInterval is set to 0, metrics will be flushed on every received batch of traces")
-	}
-
-	telemetryBuilder, err := metadata.NewTelemetryBuilder(set)
-	if err != nil {
-		return nil, err
-	}
-
 	return &serviceGraphConnector{
 		config:          pConfig,
 		logger:          set.Logger,
@@ -160,7 +133,7 @@ func newConnector(set component.TelemetrySettings, config component.Config, next
 func (p *serviceGraphConnector) Start(context.Context, component.Host) error {
 	p.store = store.NewStore(p.config.Store.TTL, p.config.Store.MaxItems, p.onComplete, p.onExpire)
 
-	go p.metricFlushLoop(*p.config.MetricsFlushInterval)
+	go p.metricFlushLoop(p.config.MetricsFlushInterval)
 
 	go p.cacheLoop(p.config.CacheLoop)
 
@@ -221,7 +194,7 @@ func (p *serviceGraphConnector) ConsumeTraces(ctx context.Context, td ptrace.Tra
 	}
 
 	// If metricsFlushInterval is not set, flush metrics immediately.
-	if *p.config.MetricsFlushInterval <= 0 {
+	if p.config.MetricsFlushInterval <= 0 {
 		if err := p.flushMetrics(ctx); err != nil {
 			// Not return error here to avoid impacting traces.
 			p.logger.Error("failed to flush metrics", zap.Error(err))
@@ -792,6 +765,9 @@ func durationToFloat(d time.Duration) float64 {
 }
 
 func mapDurationsToFloat(vs []time.Duration) []float64 {
+	if vs == nil {
+		return nil
+	}
 	vsm := make([]float64, len(vs))
 	for i, v := range vs {
 		vsm[i] = durationToFloat(v)