[discovery] Apply entity events schema the discovery receiver logs

CHANGELOG.md

@@ -12,6 +12,7 @@
   - Remove `severity_text` field from log evaluation statements ([#4583](https://github.com/signalfx/splunk-otel-collector/pull/4583))
   - Remove `first_only`  field from match struct. Events are always emitted only once for first matching metric or log statement ([#4593](https://github.com/signalfx/splunk-otel-collector/pull/4593))
   - Combine matching conditions with different statuses in one list ([#4588](https://github.com/signalfx/splunk-otel-collector/pull/4588))
+  - Apply entity events schema to the logs emitted by the receiver ([#4638](https://github.com/signalfx/splunk-otel-collector/pull/4638))
 
 ## v0.97.0
 

go.mod

@@ -530,7 +530,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.97.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk v0.97.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchperresourceattr v0.97.0 // indirect
-	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata v0.97.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata v0.97.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/signalfx v0.97.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin v0.97.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/winperfcounters v0.97.0 // indirect

internal/common/discovery/discovery.go

@@ -27,6 +27,12 @@ const (
 	ReceiverNameAttr   = "discovery.receiver.name"
 	ReceiverTypeAttr   = "discovery.receiver.type"
 	StatusAttr         = "discovery.status"
+	MessageAttr        = "discovery.message"
+
+	OtelEntityAttributesAttr = "otel.entity.attributes"
+	OtelEntityIDAttr         = "otel.entity.id"
+	OtelEntityEventTypeAttr  = "otel.entity.event.type"
+	OtelEntityEventTypeState = "entity_state"
 
 	DiscoExtensionsKey = "extensions/splunk.discovery"
 	DiscoReceiversKey  = "receivers/splunk.discovery"

internal/confmapprovider/discovery/discoverer.go

@@ -641,8 +641,8 @@ func (d *discoverer) Capabilities() consumer.Capabilities {
 	return consumer.Capabilities{}
 }
 
-// ConsumeLogs will walk through all discovery receiver-emitted logs and store all receiver and observer statuses,
-// including reported receiver configs from their discovery.receiver.config attribute. It is a consumer.Logs method.
+// ConsumeLogs will walk through all discovery receiver-emitted entity events and store all receiver and observer
+// statuses, including reported receiver configs from their discovery.receiver.config attribute. It is a consumer.Logs method.
 func (d *discoverer) ConsumeLogs(_ context.Context, ld plog.Logs) error {
 	if ld.LogRecordCount() == 0 {
 		return nil
@@ -660,12 +660,26 @@ func (d *discoverer) ConsumeLogs(_ context.Context, ld plog.Logs) error {
 			observerID            component.ID
 			err                   error
 		)
-		rlog := rlogs.At(i)
-		rAttrs := rlog.Resource().Attributes()
-		if rName, ok := rAttrs.Get(discovery.ReceiverNameAttr); ok {
+
+		// We assume that every resource log has a single log record as per the current implementation of the discovery receiver.
+		lr := rlogs.At(i).ScopeLogs().At(0).LogRecords().At(0)
+
+		// Only interested in entity events that are of type entity_state.
+		if entityEventType, ok := lr.Attributes().Get(discovery.OtelEntityEventTypeAttr); !ok || entityEventType.Str() != discovery.OtelEntityEventTypeState {
+			continue
+		}
+
+		oea, ok := lr.Attributes().Get(discovery.OtelEntityAttributesAttr)
+		if !ok {
+			d.logger.Debug("invalid entity event without attributes", zap.Any("log record", lr))
+			continue
+		}
+		entityAttrs := oea.Map()
+
+		if rName, hasName := entityAttrs.Get(discovery.ReceiverNameAttr); hasName {
 			receiverName = rName.Str()
 		}
-		rType, ok := rAttrs.Get(discovery.ReceiverTypeAttr)
+		rType, ok := entityAttrs.Get(discovery.ReceiverTypeAttr)
 		if !ok {
 			// nothing we can do without this one
 			continue
@@ -675,10 +689,11 @@ func (d *discoverer) ConsumeLogs(_ context.Context, ld plog.Logs) error {
 			d.logger.Debug("invalid receiver type", zap.Error(err))
 			continue
 		}
-		if rConfig, ok := rAttrs.Get(discovery.ReceiverConfigAttr); ok {
+		if rConfig, hasConfig := entityAttrs.Get(discovery.ReceiverConfigAttr); hasConfig {
 			receiverConfig = rConfig.Str()
 		}
-		if rObsID, ok := rAttrs.Get(discovery.ObserverIDAttr); ok {
+
+		if rObsID, hasObsID := entityAttrs.Get(discovery.ObserverIDAttr); hasObsID {
 			obsID = rObsID.Str()
 		}
 
@@ -693,8 +708,14 @@ func (d *discoverer) ConsumeLogs(_ context.Context, ld plog.Logs) error {
 			}
 		}
 
+		entityIDAttr, ok := lr.Attributes().Get(discovery.OtelEntityIDAttr)
+		if !ok {
+			d.logger.Debug("invalid entity event without id", zap.Any("log record", lr))
+			continue
+		}
+
 		endpointID := "unavailable"
-		if eid, k := rAttrs.Get(discovery.EndpointIDAttr); k {
+		if eid, k := entityIDAttr.Map().Get(discovery.EndpointIDAttr); k {
 			endpointID = eid.AsString()
 		}
 
@@ -733,21 +754,24 @@ func (d *discoverer) ConsumeLogs(_ context.Context, ld plog.Logs) error {
 		currentReceiverStatus := d.discoveredReceivers[receiverID]
 		currentObserverStatus := d.discoveredObservers[observerID]
 
-		// We assume that every resource log has a single log record as per the current implementation of the discovery receiver.
-		lr := rlog.ScopeLogs().At(0).LogRecords().At(0)
 		if currentReceiverStatus != discovery.Successful || currentObserverStatus != discovery.Successful {
-			if rStatusAttr, ok := lr.Attributes().Get(discovery.StatusAttr); ok {
+			if rStatusAttr, ok := entityAttrs.Get(discovery.StatusAttr); ok {
 				rStatus := discovery.StatusType(rStatusAttr.Str())
 				if valid, e := discovery.IsValidStatus(rStatus); !valid {
-					d.logger.Debug("invalid status from log record", zap.Error(e), zap.Any("lr", lr.Body().AsRaw()))
+					d.logger.Debug("invalid status from log record", zap.Error(e), zap.Any("lr", lr))
 					continue
 				}
+				var msg string
+				msgAttr, hasMsg := entityAttrs.Get(discovery.MessageAttr)
+				if hasMsg {
+					msg = msgAttr.AsString()
+				}
 				receiverStatus := determineCurrentStatus(currentReceiverStatus, rStatus)
 				switch receiverStatus {
 				case discovery.Failed:
-					d.logger.Info(fmt.Sprintf("failed to discover %q using %q endpoint %q: %s", receiverID, observerID, endpointID, lr.Body().AsString()))
+					d.logger.Info(fmt.Sprintf("failed to discover %q using %q endpoint %q: %s", receiverID, observerID, endpointID, msg))
 				case discovery.Partial:
-					fmt.Fprintf(os.Stderr, "Partially discovered %q using %q endpoint %q: %s\n", receiverID, observerID, endpointID, lr.Body().AsString())
+					fmt.Fprintf(os.Stderr, "Partially discovered %q using %q endpoint %q: %s\n", receiverID, observerID, endpointID, msg)
 				case discovery.Successful:
 					fmt.Fprintf(os.Stderr, "Successfully discovered %q using %q endpoint %q.\n", receiverID, observerID, endpointID)
 				}