[discovery] Apply entity events schema the discovery receiver logs (#…

…4638)

* [discovery] Apply entity events schema the discovery receiver logs

This change updates the schema of the logs emitted by the discovery receiver to represent entity events. No information is dropped, but it was reshuffled to adopt for the new schema:
- Entity ID is made of one key/value pair moved from `discovery.endpoint.id` resource attribute
- All other resource attributes are moved to the entity state event attributes
- Log body was moved to the `discovery.event.message` entity state attribute

The Discovery config provider is updated to use the entity events.

The keys used in the entity ID and entity attributes are not finalized and are expected to be changed in the following PRs.

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)
 				}