[receiver/discovery] Do not emit endpoint that are not evaluated yet (#…

…5032)

This change fixes the issue when "unknown" service can be shown in the inventory service right at the collector startup. We don't want to send entity events for endpoints that are not evaluated yet

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## Unreleased
+
+### 🧰 Bug fixes 🧰
+
+- (Splunk) `receiver/discovery`: Do not emit entity events for discovered endpoints that are not evaluated yet
+  to avoid showing "unknown" services on the Service Inventory page ([#5032](https://github.com/signalfx/splunk-otel-collector/pull/5032))
+
 ## v0.103.0
 
 This Splunk OpenTelemetry Collector release includes changes from the [opentelemetry-collector v0.103.0](https://github.com/open-telemetry/opentelemetry-collector/releases/tag/v0.103.0) and the [opentelemetry-collector-contrib v0.103.0](https://github.com/open-telemetry/opentelemetry-collector-contrib/releases/tag/v0.103.0) releases where appropriate.

internal/receiver/discoveryreceiver/endpoint_tracker.go

@@ -233,6 +233,8 @@ func (n *notify) OnChange(changed []observer.Endpoint) {
 	n.endpointTracker.updateEndpoints(changed, n.observerID)
 }
 
+// entityStateEvents converts observer endpoints to entity state events excluding those
+// that don't have a discovery status attribute yet.
 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 {
@@ -242,6 +244,12 @@ func entityStateEvents(observerID component.ID, endpoints []observer.Endpoint, c
 			continue
 		}
 
+		endpointAttrs := correlations.Attrs(endpoint.ID)
+		if _, ok := endpointAttrs[discovery.StatusAttr]; !ok {
+			// If the endpoint doesn't have a status attribute, it's not ready to be emitted.
+			continue
+		}
+
 		entityEvent := entityEvents.AppendEmpty()
 		entityEvent.SetTimestamp(pcommon.NewTimestampFromTime(ts))
 		entityState := entityEvent.SetEntityState()
@@ -259,7 +267,7 @@ func entityStateEvents(observerID component.ID, endpoints []observer.Endpoint, c
 		attrs.PutStr("endpoint", endpoint.Target)
 		attrs.PutStr(observerNameAttr, observerID.Name())
 		attrs.PutStr(observerTypeAttr, observerID.Type().String())
-		for k, v := range correlations.Attrs(endpoint.ID) {
+		for k, v := range endpointAttrs {
 			attrs.PutStr(k, v)
 		}
 		attrs.PutStr(serviceTypeAttr, deduceServiceType(attrs))

internal/receiver/discoveryreceiver/endpoint_tracker_test.go

@@ -63,6 +63,7 @@ func TestEndpointToPLogsHappyPath(t *testing.T) {
 				attrs.PutStr("type", "pod")
 				attrs.PutStr("service.type", "unknown")
 				attrs.PutStr("service.name", "my-mysql")
+				attrs.PutStr(discovery.StatusAttr, "successful")
 				return plogs
 			}(),
 		},
@@ -92,6 +93,7 @@ func TestEndpointToPLogsHappyPath(t *testing.T) {
 				attrs.PutStr("type", "port")
 				attrs.PutStr("service.type", "unknown")
 				attrs.PutStr("service.name", "redis-cart")
+				attrs.PutStr(discovery.StatusAttr, "successful")
 				return plogs
 			}(),
 		},
@@ -114,6 +116,7 @@ func TestEndpointToPLogsHappyPath(t *testing.T) {
 				attrs.PutStr("type", "hostport")
 				attrs.PutStr("service.type", "unknown")
 				attrs.PutStr("service.name", "process.name")
+				attrs.PutStr(discovery.StatusAttr, "successful")
 				return plogs
 			}(),
 		},
@@ -143,6 +146,7 @@ func TestEndpointToPLogsHappyPath(t *testing.T) {
 				attrs.PutStr("type", "container")
 				attrs.PutStr("service.type", "unknown")
 				attrs.PutStr("service.name", "container.name")
+				attrs.PutStr(discovery.StatusAttr, "successful")
 				return plogs
 			}(),
 		},
@@ -174,15 +178,18 @@ func TestEndpointToPLogsHappyPath(t *testing.T) {
 				attrs.PutStr("type", "k8s.node")
 				attrs.PutStr("service.type", "unknown")
 				attrs.PutStr("service.name", "unknown")
+				attrs.PutStr(discovery.StatusAttr, "successful")
 				return plogs
 			}(),
 		},
 	} {
 		test := tt
 		t.Run(test.name, func(t *testing.T) {
+			corr := newCorrelationStore(zap.NewNop(), time.Hour)
+			corr.(*store).attrs.Store(test.endpoint.ID, map[string]string{discovery.StatusAttr: "successful"})
 			events, failed, err := entityStateEvents(
 				component.MustNewIDWithName("observer_type", "observer.name"),
-				[]observer.Endpoint{test.endpoint}, newCorrelationStore(zap.NewNop(), time.Hour), t0,
+				[]observer.Endpoint{test.endpoint}, corr, t0,
 			)
 			require.NoError(t, err)
 			require.Zero(t, failed)
@@ -226,6 +233,7 @@ func TestEndpointToPLogsInvalidEndpoints(t *testing.T) {
 				attrs.PutStr("type", "empty.details.env")
 				attrs.PutStr("service.type", "unknown")
 				attrs.PutStr("service.name", "unknown")
+				attrs.PutStr(discovery.StatusAttr, "successful")
 				return plogs
 			}(),
 		},
@@ -248,6 +256,7 @@ func TestEndpointToPLogsInvalidEndpoints(t *testing.T) {
 				attrs.PutStr("type", "unexpected.env")
 				attrs.PutStr("service.type", "unknown")
 				attrs.PutStr("service.name", "unknown")
+				attrs.PutStr(discovery.StatusAttr, "successful")
 				return plogs
 			}(),
 		},
@@ -290,10 +299,11 @@ func TestEndpointToPLogsInvalidEndpoints(t *testing.T) {
 	} {
 		test := tt
 		t.Run(test.name, func(t *testing.T) {
-			// Validate entity_state event
+			corr := newCorrelationStore(zap.NewNop(), time.Hour)
+			corr.(*store).attrs.Store(test.endpoint.ID, map[string]string{discovery.StatusAttr: "successful"})
 			events, failed, err := entityStateEvents(
 				component.MustNewIDWithName("observer_type", "observer.name"),
-				[]observer.Endpoint{test.endpoint}, newCorrelationStore(zap.NewNop(), time.Hour), t0,
+				[]observer.Endpoint{test.endpoint}, corr, t0,
 			)
 			if test.expectedError != "" {
 				require.Error(t, err)
@@ -337,6 +347,8 @@ func FuzzEndpointToPlogs(f *testing.F) {
 			if err != nil {
 				observerTypeSanitized = discovery.NoType.Type()
 			}
+			corr := newCorrelationStore(zap.NewNop(), time.Hour)
+			corr.(*store).attrs.Store(observer.EndpointID(endpointID), map[string]string{discovery.StatusAttr: "successful"})
 			events, failed, err := entityStateEvents(
 				component.MustNewIDWithName(observerTypeSanitized.String(), observerName), []observer.Endpoint{
 					{
@@ -361,7 +373,7 @@ func FuzzEndpointToPlogs(f *testing.F) {
 							Transport: observer.Transport(transport),
 						},
 					},
-				}, newCorrelationStore(zap.NewNop(), time.Hour), t0,
+				}, corr, t0,
 			)
 
 			expectedLogs := expectedPLogs()
@@ -372,6 +384,7 @@ func FuzzEndpointToPlogs(f *testing.F) {
 			entityIDAttr.Map().PutInt("source.port", int64(port))
 			attrs := lr.Attributes().PutEmptyMap(discovery.OtelEntityAttributesAttr)
 			attrs.PutStr(discovery.EndpointIDAttr, endpointID)
+			attrs.PutStr(discovery.StatusAttr, "successful")
 			attrs.PutStr(observerNameAttr, observerName)
 			attrs.PutStr(observerTypeAttr, observerTypeSanitized.String())
 			attrs.PutStr("endpoint", target)
@@ -655,13 +668,14 @@ func TestEntityEmittingLifecycle(t *testing.T) {
 	ch := make(chan plog.Logs, 10)
 	obsID := component.MustNewIDWithName("fake_observer", "")
 	obs := &fakeObservable{}
+	corr := newCorrelationStore(logger, cfg.CorrelationTTL)
 	et := &endpointTracker{
 		config:       cfg,
 		observables:  map[component.ID]observer.Observable{obsID: obs},
 		logger:       logger,
 		pLogs:        ch,
-		correlations: newCorrelationStore(logger, cfg.CorrelationTTL),
-		emitInterval: 50 * time.Millisecond,
+		correlations: corr,
+		emitInterval: 20 * time.Millisecond,
 		stopCh:       make(chan struct{}),
 	}
 	et.start()
@@ -676,13 +690,19 @@ func TestEntityEmittingLifecycle(t *testing.T) {
 
 	obs.onAdd([]observer.Endpoint{portEndpoint})
 
+	// Ensure that no entities are emitted without discovery.status attribute.
+	time.Sleep(30 * time.Millisecond)
+	require.Empty(t, ch)
+
+	// Once the status attribute is set, the entity should be emitted.
+	corr.(*store).attrs.Store(portEndpoint.ID, map[string]string{discovery.StatusAttr: "successful"})
+
 	// Wait for at least 2 entity events to be emitted to confirm periodic emitting is working.
-	require.Eventually(t, func() bool { return len(ch) >= 2 }, 1*time.Second, 50*time.Millisecond)
+	require.Eventually(t, func() bool { return len(ch) >= 2 }, 1*time.Second, 60*time.Millisecond)
 
 	gotLogs := <-ch
 	require.Equal(t, 1, gotLogs.LogRecordCount())
-	expectedEvents, failed, err := entityStateEvents(obsID, []observer.Endpoint{portEndpoint},
-		newCorrelationStore(logger, time.Hour), t0)
+	expectedEvents, failed, err := entityStateEvents(obsID, []observer.Endpoint{portEndpoint}, corr, t0)
 	require.NoError(t, err)
 	require.Zero(t, failed)
 	require.NoError(t, plogtest.CompareLogs(expectedEvents.ConvertAndMoveToLogs(), gotLogs, plogtest.IgnoreTimestamp()))
@@ -732,6 +752,7 @@ func TestEntityStateEvents(t *testing.T) {
 	cStore := newCorrelationStore(logger, cfg.CorrelationTTL)
 	cStore.UpdateAttrs(portEndpoint.ID, map[string]string{
 		discovery.ReceiverTypeAttr: "redis",
+		discovery.StatusAttr:       "successful",
 		"attr1":                    "val1",
 		"attr2":                    "val2",
 	})
@@ -747,10 +768,11 @@ func TestEntityStateEvents(t *testing.T) {
 	assert.Equal(t, t0, event.Timestamp().AsTime())
 	assert.Equal(t, map[string]any{"k8s.pod.uid": "uid", "source.port": int64(1)}, event.ID().AsRaw())
 	assert.Equal(t, map[string]any{
-		observerNameAttr: "observer.name",
-		observerTypeAttr: "observer_type",
-		"endpoint":       "port.target",
-		"name":           "port.name",
+		observerNameAttr:     "observer.name",
+		observerTypeAttr:     "observer_type",
+		discovery.StatusAttr: "successful",
+		"endpoint":           "port.target",
+		"name":               "port.name",
 		"annotations": map[string]any{
 			"annotation.one": "value.one",
 			"annotation.two": "value.two",

tests/receivers/discovery/host_observer_test.go

@@ -22,17 +22,6 @@ import (
 	"github.com/signalfx/splunk-otel-collector/tests/testutils"
 )
 
-func TestDiscoveryReceiverWithHostObserverProvidesEndpointLogs(t *testing.T) {
-	testutils.SkipIfNotContainerTest(t)
-	if testutils.CollectorImageIsForArm(t) {
-		t.Skip("host_observer missing process info on arm")
-	}
-	testutils.AssertAllLogsReceived(
-		t, "host_observer_endpoints.yaml",
-		"host_observer_endpoints_config.yaml", nil, nil,
-	)
-}
-
 func TestDiscoveryReceiverWithHostObserverAndSimplePrometheusReceiverProvideStatusLogs(t *testing.T) {
 	testutils.SkipIfNotContainerTest(t)
 	if testutils.CollectorImageIsForArm(t) {