[connector/servicegraphconnector] Adds a new config option to enable messaging_system's metric generation

.chloggen/servicegraph-msg-system-metrics.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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: servicegraphprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Added a configuration option to enable the servicegraphprocessor to generate traces_service_graph_request_messaging_system_seconds metrics."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30856]
+
+# (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: []

connector/servicegraphconnector/README.md

@@ -142,6 +142,8 @@ The following settings can be optionally configured:
   - Default: Metrics are flushed on every received batch of traces.
 - `database_name_attribute`: the attribute name used to identify the database name from span attributes.
   - Default: `db.name`
+- `enable_messaging_system_latency_histogram`: enable to produce histogram for interactions with messaging systems.
+  - Default: `false`
 
 ## Example configurations
 

connector/servicegraphconnector/config.go

@@ -50,6 +50,8 @@ type Config struct {
 	// DatabaseNameAttribute is the attribute name used to identify the database name from span attributes.
 	// The default value is db.name.
 	DatabaseNameAttribute string `mapstructure:"database_name_attribute"`
+
+	EnableMessagingSystemLatencyHistogram bool `mapstructure:"enable_messaging_system_latency_histogram"`
 }
 
 type StoreConfig struct {

connector/servicegraphconnector/connector.go

@@ -64,16 +64,19 @@ type serviceGraphConnector struct {
 
 	startTime time.Time
 
-	seriesMutex                          sync.Mutex
-	reqTotal                             map[string]int64
-	reqFailedTotal                       map[string]int64
-	reqClientDurationSecondsCount        map[string]uint64
-	reqClientDurationSecondsSum          map[string]float64
-	reqClientDurationSecondsBucketCounts map[string][]uint64
-	reqServerDurationSecondsCount        map[string]uint64
-	reqServerDurationSecondsSum          map[string]float64
-	reqServerDurationSecondsBucketCounts map[string][]uint64
-	reqDurationBounds                    []float64
+	seriesMutex                                   sync.Mutex
+	reqTotal                                      map[string]int64
+	reqFailedTotal                                map[string]int64
+	reqClientDurationSecondsCount                 map[string]uint64
+	reqClientDurationSecondsSum                   map[string]float64
+	reqClientDurationSecondsBucketCounts          map[string][]uint64
+	reqServerDurationSecondsCount                 map[string]uint64
+	reqServerDurationSecondsSum                   map[string]float64
+	reqServerDurationSecondsBucketCounts          map[string][]uint64
+	reqMessagingSystemDurationSecondsCount        map[string]uint64
+	reqMessagingSystemDurationSecondsSum          map[string]float64
+	reqMessagingSystemDurationSecondsBucketCounts map[string][]uint64
+	reqDurationBounds                             []float64
 
 	metricMutex sync.RWMutex
 	keyToMetric map[string]metricSeries
@@ -124,19 +127,22 @@ func newConnector(set component.TelemetrySettings, config component.Config, next
 		logger:          set.Logger,
 		metricsConsumer: next,
 
-		startTime:                            time.Now(),
-		reqTotal:                             make(map[string]int64),
-		reqFailedTotal:                       make(map[string]int64),
-		reqClientDurationSecondsCount:        make(map[string]uint64),
-		reqClientDurationSecondsSum:          make(map[string]float64),
-		reqClientDurationSecondsBucketCounts: make(map[string][]uint64),
-		reqServerDurationSecondsCount:        make(map[string]uint64),
-		reqServerDurationSecondsSum:          make(map[string]float64),
-		reqServerDurationSecondsBucketCounts: make(map[string][]uint64),
-		reqDurationBounds:                    bounds,
-		keyToMetric:                          make(map[string]metricSeries),
-		shutdownCh:                           make(chan any),
-		telemetryBuilder:                     telemetryBuilder,
+		startTime:                                     time.Now(),
+		reqTotal:                                      make(map[string]int64),
+		reqFailedTotal:                                make(map[string]int64),
+		reqClientDurationSecondsCount:                 make(map[string]uint64),
+		reqClientDurationSecondsSum:                   make(map[string]float64),
+		reqClientDurationSecondsBucketCounts:          make(map[string][]uint64),
+		reqServerDurationSecondsCount:                 make(map[string]uint64),
+		reqServerDurationSecondsSum:                   make(map[string]float64),
+		reqServerDurationSecondsBucketCounts:          make(map[string][]uint64),
+		reqMessagingSystemDurationSecondsCount:        make(map[string]uint64),
+		reqMessagingSystemDurationSecondsSum:          make(map[string]float64),
+		reqMessagingSystemDurationSecondsBucketCounts: make(map[string][]uint64),
+		reqDurationBounds:                             bounds,
+		keyToMetric:                                   make(map[string]metricSeries),
+		shutdownCh:                                    make(chan any),
+		telemetryBuilder:                              telemetryBuilder,
 	}, nil
 }
 
@@ -383,6 +389,10 @@ func (p *serviceGraphConnector) aggregateMetricsForEdge(e *store.Edge) {
 		p.updateErrorMetrics(metricKey)
 	}
 	p.updateDurationMetrics(metricKey, e.ServerLatencySec, e.ClientLatencySec)
+
+	if p.config.EnableMessagingSystemLatencyHistogram && e.ConnectionType == store.MessagingSystem {
+		p.updateMessagingSystemDurationMetrics(metricKey, e.ServerLatencySec)
+	}
 }
 
 func (p *serviceGraphConnector) updateSeries(key string, dimensions pcommon.Map) {
@@ -434,6 +444,16 @@ func (p *serviceGraphConnector) updateClientDurationMetrics(key string, duration
 	p.reqClientDurationSecondsBucketCounts[key][index]++
 }
 
+func (p *serviceGraphConnector) updateMessagingSystemDurationMetrics(key string, duration float64) {
+	index := sort.SearchFloat64s(p.reqDurationBounds, duration) // Search bucket index
+	if _, ok := p.reqMessagingSystemDurationSecondsBucketCounts[key]; !ok {
+		p.reqMessagingSystemDurationSecondsBucketCounts[key] = make([]uint64, len(p.reqDurationBounds)+1)
+	}
+	p.reqMessagingSystemDurationSecondsSum[key] += duration
+	p.reqMessagingSystemDurationSecondsCount[key]++
+	p.reqMessagingSystemDurationSecondsBucketCounts[key][index]++
+}
+
 func buildDimensions(e *store.Edge) pcommon.Map {
 	dims := pcommon.NewMap()
 	dims.PutStr("client", e.ClientService)
@@ -529,6 +549,12 @@ func (p *serviceGraphConnector) collectLatencyMetrics(ilm pmetric.ScopeMetrics)
 		return err
 	}
 
+	if p.config.EnableMessagingSystemLatencyHistogram {
+		if err := p.collectMessagingSystemLatencyMetrics(ilm); err != nil {
+			return err
+		}
+	}
+
 	return p.collectClientLatencyMetrics(ilm)
 }
 
@@ -591,6 +617,63 @@ func (p *serviceGraphConnector) collectServerLatencyMetrics(ilm pmetric.ScopeMet
 	return nil
 }
 
+func (p *serviceGraphConnector) collectMessagingSystemLatencyMetrics(ilm pmetric.ScopeMetrics) error {
+	if len(p.reqServerDurationSecondsCount) > 0 {
+		mDuration := ilm.Metrics().AppendEmpty()
+		mDuration.SetName("traces_service_graph_request_messaging_system_seconds")
+		// TODO: Support other aggregation temporalities
+		mDuration.SetEmptyHistogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
+		timestamp := pcommon.NewTimestampFromTime(time.Now())
+
+		for key := range p.reqServerDurationSecondsCount {
+			dpDuration := mDuration.Histogram().DataPoints().AppendEmpty()
+			dpDuration.SetStartTimestamp(pcommon.NewTimestampFromTime(p.startTime))
+			dpDuration.SetTimestamp(timestamp)
+			dpDuration.ExplicitBounds().FromRaw(p.reqDurationBounds)
+			dpDuration.BucketCounts().FromRaw(p.reqMessagingSystemDurationSecondsBucketCounts[key])
+			dpDuration.SetCount(p.reqMessagingSystemDurationSecondsCount[key])
+			dpDuration.SetSum(p.reqMessagingSystemDurationSecondsSum[key])
+
+				// TODO: Support exemplars
+				dimensions, ok := p.dimensionsForSeries(key)
+				if !ok {
+					return fmt.Errorf("failed to find dimensions for key %s", key)
+				}
+
+				dimensions.CopyTo(dpDuration.Attributes())
+			}
+	}
+	return nil
+}
+
+func (p *serviceGraphConnector) collectMessagingSystemLatencyMetrics(ilm pmetric.ScopeMetrics) error {
+	for key := range p.reqServerDurationSecondsCount {
+		mDuration := ilm.Metrics().AppendEmpty()
+		mDuration.SetName("traces_service_graph_request_messaging_system_seconds")
+		// TODO: Support other aggregation temporalities
+		mDuration.SetEmptyHistogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
+
+		timestamp := pcommon.NewTimestampFromTime(time.Now())
+
+		dpDuration := mDuration.Histogram().DataPoints().AppendEmpty()
+		dpDuration.SetStartTimestamp(pcommon.NewTimestampFromTime(p.startTime))
+		dpDuration.SetTimestamp(timestamp)
+		dpDuration.ExplicitBounds().FromRaw(p.reqDurationBounds)
+		dpDuration.BucketCounts().FromRaw(p.reqMessagingSystemDurationSecondsBucketCounts[key])
+		dpDuration.SetCount(p.reqMessagingSystemDurationSecondsCount[key])
+		dpDuration.SetSum(p.reqMessagingSystemDurationSecondsSum[key])
+
+		// TODO: Support exemplars
+		dimensions, ok := p.dimensionsForSeries(key)
+		if !ok {
+			return fmt.Errorf("failed to find dimensions for key %s", key)
+		}
+
+		dimensions.CopyTo(dpDuration.Attributes())
+	}
+	return nil
+}
+
 func (p *serviceGraphConnector) buildMetricKey(clientName, serverName, connectionType, failed string, edgeDimensions map[string]string) string {
 	var metricKey strings.Builder
 	metricKey.WriteString(clientName + metricKeySeparator + serverName + metricKeySeparator + connectionType + metricKeySeparator + failed)
@@ -665,6 +748,11 @@ func (p *serviceGraphConnector) cleanCache() {
 		delete(p.reqServerDurationSecondsCount, key)
 		delete(p.reqServerDurationSecondsSum, key)
 		delete(p.reqServerDurationSecondsBucketCounts, key)
+		if p.config.EnableMessagingSystemLatencyHistogram {
+			delete(p.reqMessagingSystemDurationSecondsCount, key)
+			delete(p.reqMessagingSystemDurationSecondsSum, key)
+			delete(p.reqMessagingSystemDurationSecondsBucketCounts, key)
+		}
 	}
 	p.seriesMutex.Unlock()
 

connector/servicegraphconnector/connector_test.go

@@ -601,3 +601,33 @@ func TestVirtualNodeClientLabels(t *testing.T) {
 	)
 	require.NoError(t, err)
 }
+
+func TestEnableMessagingSystemHistograms(t *testing.T) {
+	cfg := &Config{
+		EnableMessagingSystemLatencyHistogram: true,
+		Store: StoreConfig{
+			MaxItems: 10,
+		},
+	}
+	set := componenttest.NewNopTelemetrySettings()
+	set.Logger = zaptest.NewLogger(t)
+	conn, err := newConnector(set, cfg, newMockMetricsExporter())
+	assert.NoError(t, conn.Shutdown(context.Background()))
+	assert.NoError(t, err)
+	assert.NoError(t, conn.Start(context.Background(), componenttest.NewNopHost()))
+
+	td, err := golden.ReadTraces("testdata/messaging-system-trace.yaml")
+	assert.NoError(t, err)
+	assert.NoError(t, conn.ConsumeTraces(context.Background(), td))
+
+	conn.store.Expire()
+
+	metrics := conn.metricsConsumer.(*mockMetricsExporter).GetMetrics()
+	require.Len(t, metrics, 1)
+
+	expectedMetrics, err := golden.ReadMetrics("testdata/messaging-system-trace-expected-metrics.yaml")
+	assert.NoError(t, err)
+
+	err = pmetrictest.CompareMetrics(expectedMetrics, metrics[0], pmetrictest.IgnoreStartTimestamp(), pmetrictest.IgnoreTimestamp())
+	require.NoError(t, err)
+}

connector/servicegraphconnector/go.mod

@@ -25,6 +25,8 @@ require (
 	go.uber.org/zap v1.27.0
 )
 
+require github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.104.0 // indirect
+
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect