[receiver/discovery] Emit evaluation entities by the endpoint tracker (#4728)

This change consolidates the entity emitting in one place: endpoint tracker. Once the evaluation succeeds, the endpoint attributes are updated in the correlation store and the new entity state is emitted right away. Every subsequent state is emitted with all the added attributes.

The tests are kept with minimal changes to ensure that no attributes are lost in the emitted entities.

Currently, we emit superset of all the attributes that were previously emitted on regular basis or on the evaluation. Later, we will reduce the set and remove redundant attributes.

Author: dmitryax

internal/receiver/discoveryreceiver/correlation.go

@@ -50,6 +50,7 @@ type correlationStore interface {
 	GetOrCreate(endpointID observer.EndpointID, receiverID component.ID) correlation
 	Attrs(endpointID observer.EndpointID) map[string]string
 	UpdateAttrs(endpointID observer.EndpointID, attrs map[string]string)
+	EmitCh() chan correlation
 	Endpoints(updatedBefore time.Time) []observer.Endpoint
 	// Start the reaping loop to prevent unnecessary endpoint buildup
 	Start()
@@ -71,6 +72,7 @@ type store struct {
 	attrs         *sync.Map
 	// sentinel for terminating reaper loop
 	sentinel     chan struct{}
+	emitCh       chan correlation
 	reapInterval time.Duration
 	ttl          time.Duration
 }
@@ -84,6 +86,7 @@ func newCorrelationStore(logger *zap.Logger, ttl time.Duration) correlationStore
 		reapInterval:  30 * time.Second,
 		ttl:           ttl,
 		sentinel:      make(chan struct{}, 1),
+		emitCh:        make(chan correlation),
 	}
 }
 
@@ -118,6 +121,11 @@ func (s *store) MarkStale(endpointID observer.EndpointID) {
 	}
 }
 
+// EmitCh returns a channel to emit endpoints immediately.
+func (s *store) EmitCh() chan correlation {
+	return s.emitCh
+}
+
 // Endpoints returns all active endpoints that have not been updated since the provided time.
 func (s *store) Endpoints(updatedBefore time.Time) []observer.Endpoint {
 	var endpoints []observer.Endpoint

internal/receiver/discoveryreceiver/endpoint_tracker.go

@@ -92,6 +92,8 @@ func (et *endpointTracker) startEmitLoop() {
 	timer := time.NewTicker(et.emitInterval)
 	for {
 		select {
+		case corr := <-et.correlations.EmitCh():
+			et.emitEntityStateEvents(corr.observerID, []observer.Endpoint{corr.endpoint})
 		case <-timer.C:
 			for obs := range et.observables {
 				et.emitEntityStateEvents(obs, et.correlations.Endpoints(time.Now().Add(-et.emitInterval)))
@@ -114,7 +116,7 @@ func (et *endpointTracker) stop() {
 
 func (et *endpointTracker) emitEntityStateEvents(observerCID component.ID, endpoints []observer.Endpoint) {
 	if et.pLogs != nil {
-		entityEvents, numFailed, err := entityStateEvents(observerCID, endpoints, time.Now())
+		entityEvents, numFailed, err := entityStateEvents(observerCID, endpoints, et.correlations, time.Now())
 		if err != nil {
 			et.logger.Warn(fmt.Sprintf("failed converting %v endpoints to log records", numFailed), zap.Error(err))
 		}
@@ -209,7 +211,7 @@ func (n *notify) OnChange(changed []observer.Endpoint) {
 	n.endpointTracker.updateEndpoints(changed, n.observerID)
 }
 
-func entityStateEvents(observerID component.ID, endpoints []observer.Endpoint, ts time.Time) (ees experimentalmetricmetadata.EntityEventsSlice, failed int, err error) {
+func entityStateEvents(observerID component.ID, endpoints []observer.Endpoint, correlations correlationStore, ts time.Time) (ees experimentalmetricmetadata.EntityEventsSlice, failed int, err error) {
 	entityEvents := experimentalmetricmetadata.NewEntityEventsSlice()
 	for _, endpoint := range endpoints {
 		entityEvent := entityEvents.AppendEmpty()
@@ -230,6 +232,9 @@ func entityStateEvents(observerID component.ID, endpoints []observer.Endpoint, t
 		attrs.PutStr("endpoint", endpoint.Target)
 		attrs.PutStr(observerNameAttr, observerID.Name())
 		attrs.PutStr(observerTypeAttr, observerID.Type().String())
+		for k, v := range correlations.Attrs(endpoint.ID) {
+			attrs.PutStr(k, v)
+		}
 	}
 	return entityEvents, failed, err
 }

internal/receiver/discoveryreceiver/endpoint_tracker_test.go

@@ -164,7 +164,7 @@ func TestEndpointToPLogsHappyPath(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			events, failed, err := entityStateEvents(
 				component.MustNewIDWithName("observer_type", "observer.name"),
-				[]observer.Endpoint{test.endpoint}, t0,
+				[]observer.Endpoint{test.endpoint}, newCorrelationStore(zap.NewNop(), time.Hour), t0,
 			)
 			require.NoError(t, err)
 			require.Zero(t, failed)
@@ -276,7 +276,7 @@ func TestEndpointToPLogsInvalidEndpoints(t *testing.T) {
 			// Validate entity_state event
 			events, failed, err := entityStateEvents(
 				component.MustNewIDWithName("observer_type", "observer.name"),
-				[]observer.Endpoint{test.endpoint}, t0,
+				[]observer.Endpoint{test.endpoint}, newCorrelationStore(zap.NewNop(), time.Hour), t0,
 			)
 			if test.expectedError != "" {
 				require.Error(t, err)
@@ -343,7 +343,7 @@ func FuzzEndpointToPlogs(f *testing.F) {
 							Transport: observer.Transport(transport),
 						},
 					},
-				}, t0,
+				}, newCorrelationStore(zap.NewNop(), time.Hour), t0,
 			)
 
 			expectedLogs := expectedPLogs(endpointID)
@@ -661,6 +661,7 @@ func TestEntityEmittingLifecycle(t *testing.T) {
 	require.Equal(t, 1, gotLogs.LogRecordCount())
 	// TODO: Use plogtest.IgnoreTimestamp once available
 	expectedEvents, failed, err := entityStateEvents(obsID, []observer.Endpoint{portEndpoint},
+		newCorrelationStore(zap.NewNop(), time.Hour),
 		gotLogs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Timestamp().AsTime())
 	require.NoError(t, err)
 	require.Zero(t, failed)

internal/receiver/discoveryreceiver/evaluator.go

@@ -24,7 +24,6 @@ import (
 	"github.com/antonmedv/expr/vm"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer"
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v2"
 
@@ -117,10 +116,10 @@ func (e *evaluator) evaluateMatch(match Match, pattern string, status discovery.
 }
 
 // correlateResourceAttributes sets correlation attributes including embedded base64 config content, if configured.
-func (e *evaluator) correlateResourceAttributes(cfg *Config, to pcommon.Map, corr correlation) {
+func (e *evaluator) correlateResourceAttributes(cfg *Config, to map[string]string, corr correlation) {
 	observerID := corr.observerID.String()
 	if observerID != "" && observerID != discovery.NoType.String() {
-		to.PutStr(discovery.ObserverIDAttr, observerID)
+		to[discovery.ObserverIDAttr] = observerID
 	}
 
 	if e.config.EmbedReceiverConfig {
@@ -140,6 +139,6 @@ func (e *evaluator) correlateResourceAttributes(cfg *Config, to pcommon.Map, cor
 		if err != nil {
 			e.logger.Error("failed embedding receiver config", zap.String("observer", observerID), zap.Error(err))
 		}
-		to.PutStr(discovery.ReceiverConfigAttr, base64.StdEncoding.EncodeToString(cfgYaml))
+		to[discovery.ReceiverConfigAttr] = base64.StdEncoding.EncodeToString(cfgYaml)
 	}
 }

internal/receiver/discoveryreceiver/evaluator_test.go

@@ -24,7 +24,6 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.uber.org/zap"
 )
 
@@ -124,12 +123,11 @@ func TestCorrelateResourceAttrs(t *testing.T) {
 				},
 			}
 
-			to := pcommon.NewMap()
-
+			to := map[string]string{}
 			require.Empty(t, eval.correlations.Attrs(endpointID))
 			eval.correlateResourceAttributes(cfg, to, corr)
 
-			expectedResourceAttrs := map[string]any{
+			expectedResourceAttrs := map[string]string{
 				"discovery.observer.id": "type/name",
 			}
 
@@ -147,7 +145,7 @@ watch_observers:
 `))
 			}
 
-			require.Equal(t, expectedResourceAttrs, to.AsRaw())
+			require.Equal(t, expectedResourceAttrs, to)
 		})
 	}
 }

internal/receiver/discoveryreceiver/metric_evaluator.go

@@ -19,11 +19,9 @@ import (
 	"fmt"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/pcommon"
-	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -47,12 +45,10 @@ var (
 // Status match rules. If so, they emit log records for the matching metric.
 type metricEvaluator struct {
 	*evaluator
-	pLogs chan plog.Logs
 }
 
-func newMetricEvaluator(logger *zap.Logger, cfg *Config, pLogs chan plog.Logs, correlations correlationStore) *metricEvaluator {
+func newMetricEvaluator(logger *zap.Logger, cfg *Config, correlations correlationStore) *metricEvaluator {
 	return &metricEvaluator{
-		pLogs: pLogs,
 		evaluator: newEvaluator(logger, cfg, correlations,
 			// TODO: provide more capable env w/ resource and metric attributes
 			func(pattern string) map[string]any {
@@ -67,15 +63,13 @@ func (m *metricEvaluator) Capabilities() consumer.Capabilities {
 }
 
 func (m *metricEvaluator) ConsumeMetrics(_ context.Context, md pmetric.Metrics) error {
-	if pLogs := m.evaluateMetrics(md); pLogs.LogRecordCount() > 0 {
-		m.pLogs <- pLogs
-	}
+	m.evaluateMetrics(md)
 	return nil
 }
 
 // evaluateMetrics parses the provided Metrics and returns plog.Logs with a single log record if it matches
 // against the first applicable configured Status match rule.
-func (m *metricEvaluator) evaluateMetrics(md pmetric.Metrics) plog.Logs {
+func (m *metricEvaluator) evaluateMetrics(md pmetric.Metrics) {
 	if ce := m.logger.Check(zapcore.DebugLevel, "evaluating metrics"); ce != nil {
 		if mbytes, err := jsonMarshaler.MarshalMetrics(md); err == nil {
 			ce.Write(zap.ByteString("metrics", mbytes))
@@ -84,22 +78,22 @@ func (m *metricEvaluator) evaluateMetrics(md pmetric.Metrics) plog.Logs {
 		}
 	}
 	if md.MetricCount() == 0 {
-		return plog.NewLogs()
+		return
 	}
 
 	receiverID, endpointID := statussources.MetricsToReceiverIDs(md)
 	if receiverID == discovery.NoType || endpointID == "" {
 		m.logger.Debug("unable to evaluate metrics from receiver without corresponding name or Endpoint.ID", zap.Any("metrics", md))
-		return plog.NewLogs()
+		return
 	}
 
 	rEntry, ok := m.config.Receivers[receiverID]
 	if !ok {
 		m.logger.Info("No matching configured receiver for metric status evaluation", zap.String("receiver", receiverID.String()))
-		return plog.NewLogs()
+		return
 	}
 	if rEntry.Status == nil || len(rEntry.Status.Metrics) == 0 {
-		return plog.NewLogs()
+		return
 	}
 
 	for _, match := range rEntry.Status.Metrics {
@@ -108,20 +102,26 @@ func (m *metricEvaluator) evaluateMetrics(md pmetric.Metrics) plog.Logs {
 			continue
 		}
 
-		entityEvents := experimentalmetricmetadata.NewEntityEventsSlice()
-		entityEvent := entityEvents.AppendEmpty()
-		entityEvent.ID().PutStr(discovery.EndpointIDAttr, string(endpointID))
-		entityState := entityEvent.SetEntityState()
-
-		res.Attributes().CopyTo(entityState.Attributes())
 		corr := m.correlations.GetOrCreate(endpointID, receiverID)
-		m.correlateResourceAttributes(m.config, entityState.Attributes(), corr)
+		attrs := m.correlations.Attrs(endpointID)
 
-		// Remove the endpoint ID from the attributes as it's set in the entity ID.
-		entityState.Attributes().Remove(discovery.EndpointIDAttr)
+		// If the status is already the same as desired, we don't need to update the entity state.
+		if match.Status == discovery.StatusType(attrs[discovery.StatusAttr]) {
+			return
+		}
 
-		entityState.Attributes().PutStr(eventTypeAttr, metricMatch)
-		entityState.Attributes().PutStr(receiverRuleAttr, rEntry.Rule.String())
+		res.Attributes().Range(func(k string, v pcommon.Value) bool {
+			// skip endpoint ID attr since it's set in the entity ID
+			if k == discovery.EndpointIDAttr {
+				return true
+			}
+			attrs[k] = v.AsString()
+			return true
+		})
+		m.correlateResourceAttributes(m.config, attrs, corr)
+
+		attrs[eventTypeAttr] = metricMatch
+		attrs[receiverRuleAttr] = rEntry.Rule.String()
 
 		desiredRecord := match.Record
 		if desiredRecord == nil {
@@ -131,19 +131,17 @@ func (m *metricEvaluator) evaluateMetrics(md pmetric.Metrics) plog.Logs {
 		if desiredRecord.Body != "" {
 			desiredMsg = desiredRecord.Body
 		}
-		entityState.Attributes().PutStr(discovery.MessageAttr, desiredMsg)
+		attrs[discovery.MessageAttr] = desiredMsg
 		for k, v := range desiredRecord.Attributes {
-			entityState.Attributes().PutStr(k, v)
-		}
-		entityState.Attributes().PutStr(metricNameAttr, metric.Name())
-		entityState.Attributes().PutStr(discovery.StatusAttr, string(match.Status))
-		if ts := m.timestampFromMetric(metric); ts != nil {
-			entityEvent.SetTimestamp(*ts)
+			attrs[k] = v
 		}
+		attrs[metricNameAttr] = metric.Name()
+		attrs[discovery.StatusAttr] = string(match.Status)
+		m.correlations.UpdateAttrs(endpointID, attrs)
 
-		return entityEvents.ConvertAndMoveToLogs()
+		m.correlations.EmitCh() <- corr
+		return
 	}
-	return plog.NewLogs()
 }
 
 // findMatchedMetric finds the metric that matches the provided match rule and return it along with the resource if found.
@@ -166,42 +164,3 @@ func (m *metricEvaluator) findMatchedMetric(md pmetric.Metrics, match Match, rec
 	}
 	return pcommon.NewResource(), pmetric.NewMetric(), false
 }
-
-func (m *metricEvaluator) timestampFromMetric(metric pmetric.Metric) *pcommon.Timestamp {
-	var ts *pcommon.Timestamp
-	switch dt := metric.Type(); dt {
-	case pmetric.MetricTypeGauge:
-		dps := metric.Gauge().DataPoints()
-		if dps.Len() > 0 {
-			t := dps.At(0).Timestamp()
-			ts = &t
-		}
-	case pmetric.MetricTypeSum:
-		dps := metric.Sum().DataPoints()
-		if dps.Len() > 0 {
-			t := dps.At(0).Timestamp()
-			ts = &t
-		}
-	case pmetric.MetricTypeHistogram:
-		dps := metric.Histogram().DataPoints()
-		if dps.Len() > 0 {
-			t := dps.At(0).Timestamp()
-			ts = &t
-		}
-	case pmetric.MetricTypeExponentialHistogram:
-		dps := metric.ExponentialHistogram().DataPoints()
-		if dps.Len() > 0 {
-			t := dps.At(0).Timestamp()
-			ts = &t
-		}
-	case pmetric.MetricTypeSummary:
-		dps := metric.Summary().DataPoints()
-		if dps.Len() > 0 {
-			t := dps.At(0).Timestamp()
-			ts = &t
-		}
-	default:
-		m.logger.Debug("cannot get timestamp from data type", zap.String("data type", dt.String()))
-	}
-	return ts
-}