@@ -39,6 +39,8 @@ const (
3939
4040// identifyingAttrKeys are the keys of attributes that are used to identify an entity.
4141var identifyingAttrKeys = []string {
42+ serviceTypeAttr ,
43+ semconv .AttributeServiceName ,
4244 semconv .AttributeK8SPodUID ,
4345 semconv .AttributeContainerID ,
4446 semconv .AttributeK8SNodeUID ,
@@ -135,7 +137,7 @@ func (et *endpointTracker) stop() {
135137
136138func (et * endpointTracker ) emitEntityStateEvents (observerCID component.ID , endpoints []observer.Endpoint ) {
137139 if et .pLogs != nil {
138- entityEvents , numFailed , err := entityStateEvents (observerCID , endpoints , et .correlations , time .Now ())
140+ entityEvents , numFailed , err := entityEvents (observerCID , endpoints , et .correlations , time .Now (), experimentalmetricmetadata . EventTypeState )
139141 if err != nil {
140142 et .logger .Warn (fmt .Sprintf ("failed converting %v endpoints to entity state events" , numFailed ), zap .Error (err ))
141143 }
@@ -145,9 +147,10 @@ func (et *endpointTracker) emitEntityStateEvents(observerCID component.ID, endpo
145147 }
146148}
147149
148- func (et * endpointTracker ) emitEntityDeleteEvents (endpoints []observer.Endpoint ) {
150+ func (et * endpointTracker ) emitEntityDeleteEvents (observerCID component. ID , endpoints []observer.Endpoint ) {
149151 if et .pLogs != nil {
150- entityEvents , numFailed , err := entityDeleteEvents (endpoints , time .Now ())
152+ entityEvents , numFailed , err := entityEvents (observerCID , endpoints , et .correlations , time .Now (),
153+ experimentalmetricmetadata .EventTypeDelete )
151154 if err != nil {
152155 et .logger .Warn (fmt .Sprintf ("failed converting %v endpoints to entity delete events" , numFailed ), zap .Error (err ))
153156 }
@@ -226,18 +229,18 @@ func (n *notify) OnRemove(removed []observer.Endpoint) {
226229 n .endpointTracker .correlations .MarkStale (endpoint .ID )
227230 }
228231 }
229- n .endpointTracker .emitEntityDeleteEvents (matchingEndpoints )
232+ n .endpointTracker .emitEntityDeleteEvents (n . observerID , matchingEndpoints )
230233}
231234
232235func (n * notify ) OnChange (changed []observer.Endpoint ) {
233236 n .endpointTracker .updateEndpoints (changed , n .observerID )
234237}
235238
236- // entityStateEvents converts observer endpoints to entity state events excluding those
239+ // entityEvents converts observer endpoints to entity state events excluding those
237240// that don't have a discovery status attribute yet.
238- func entityStateEvents (observerID component.ID , endpoints []observer.Endpoint , correlations * correlationStore ,
239- ts time.Time ) (ees experimentalmetricmetadata.EntityEventsSlice , failed int , err error ) {
240- entityEvents := experimentalmetricmetadata .NewEntityEventsSlice ()
241+ func entityEvents (observerID component.ID , endpoints []observer.Endpoint , correlations * correlationStore ,
242+ ts time.Time , eventType experimentalmetricmetadata. EventType ) (ees experimentalmetricmetadata.EntityEventsSlice , failed int , err error ) {
243+ events := experimentalmetricmetadata .NewEntityEventsSlice ()
241244 for _ , endpoint := range endpoints {
242245 if endpoint .Details == nil {
243246 failed ++
@@ -251,17 +254,11 @@ func entityStateEvents(observerID component.ID, endpoints []observer.Endpoint, c
251254 continue
252255 }
253256
254- entityEvent := entityEvents .AppendEmpty ()
255- entityEvent .SetTimestamp (pcommon .NewTimestampFromTime (ts ))
256- entityState := entityEvent .SetEntityState ()
257- entityState .SetEntityType (entityType )
258- attrs := entityState .Attributes ()
259- if envAttrs , e := endpointEnvToAttrs (endpoint .Details .Type (), endpoint .Details .Env ()); e != nil {
257+ attrs , e := endpointEnvToAttrs (endpoint .Details .Type (), endpoint .Details .Env ())
258+ if e != nil {
260259 err = multierr .Combine (err , fmt .Errorf ("failed determining attributes for %q: %w" , endpoint .ID , e ))
261260 failed ++
262- } else {
263- // this must be the first mutation of attrs since it's destructive
264- envAttrs .CopyTo (attrs )
261+ continue
265262 }
266263 attrs .PutStr ("type" , string (endpoint .Details .Type ()))
267264 attrs .PutStr (discovery .EndpointIDAttr , string (endpoint .ID ))
@@ -273,31 +270,21 @@ func entityStateEvents(observerID component.ID, endpoints []observer.Endpoint, c
273270 }
274271 attrs .PutStr (serviceTypeAttr , deduceServiceType (attrs ))
275272 attrs .PutStr (semconv .AttributeServiceName , deduceServiceName (attrs ))
276- extractIdentifyingAttrs (attrs , entityEvent .ID ())
277- }
278- return entityEvents , failed , err
279- }
280-
281- func entityDeleteEvents (endpoints []observer.Endpoint , ts time.Time ) (ees experimentalmetricmetadata.EntityEventsSlice , failed int , err error ) {
282- entityEvents := experimentalmetricmetadata .NewEntityEventsSlice ()
283- for _ , endpoint := range endpoints {
284- if endpoint .Details == nil {
285- failed ++
286- err = multierr .Combine (err , fmt .Errorf ("endpoint %q has no details" , endpoint .ID ))
287- continue
288- }
289273
290- entityEvent := entityEvents .AppendEmpty ()
291- entityEvent .SetTimestamp (pcommon .NewTimestampFromTime (ts ))
292- entityEvent .SetEntityDelete ()
293- if envAttrs , e := endpointEnvToAttrs (endpoint .Details .Type (), endpoint .Details .Env ()); e != nil {
294- err = multierr .Combine (err , fmt .Errorf ("failed determining attributes for %q: %w" , endpoint .ID , e ))
295- failed ++
296- } else {
297- extractIdentifyingAttrs (envAttrs , entityEvent .ID ())
274+ event := events .AppendEmpty ()
275+ event .SetTimestamp (pcommon .NewTimestampFromTime (ts ))
276+ extractIdentifyingAttrs (attrs , event .ID ())
277+ switch eventType {
278+ case experimentalmetricmetadata .EventTypeState :
279+ entityState := event .SetEntityState ()
280+ entityState .SetEntityType (entityType )
281+ attrs .MoveTo (entityState .Attributes ())
282+ case experimentalmetricmetadata .EventTypeDelete :
283+ deleteEvent := event .SetEntityDelete ()
284+ deleteEvent .SetEntityType (entityType )
298285 }
299286 }
300- return entityEvents , failed , err
287+ return events , failed , err
301288}
302289
303290func endpointEnvToAttrs (endpointType observer.EndpointType , endpointEnv observer.EndpointEnv ) (pcommon.Map , error ) {
0 commit comments