add interval to experimental entity events

Author: povilasv

.chloggen/entity-interval.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: "enhancement"
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: hostmetricsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "add reporting interval to entity event"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34240]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

pkg/experimentalmetricmetadata/entity_events.go

@@ -4,6 +4,8 @@
 package experimentalmetricmetadata // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata"
 
 import (
+	"time"
+
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 )
@@ -18,6 +20,7 @@ const (
 
 	semconvOtelEntityID         = "otel.entity.id"
 	semconvOtelEntityType       = "otel.entity.type"
+	semconvOtelEntityInterval   = "otel.entity.interval"
 	semconvOtelEntityAttributes = "otel.entity.attributes"
 
 	semconvOtelEntityEventAsScope = "otel.entity.event_as_log"
@@ -173,6 +176,21 @@ func (s EntityStateDetails) SetEntityType(t string) {
 	s.orig.Attributes().PutStr(semconvOtelEntityType, t)
 }
 
+// SetInterval sets the reporting period
+// i.e. how frequently the information about this entity is reported via EntityState events even if the entity does not change.
+func (s EntityStateDetails) SetInterval(t time.Duration) {
+	s.orig.Attributes().PutInt(semconvOtelEntityInterval, t.Milliseconds())
+}
+
+// Interval returns the reporting period
+func (s EntityStateDetails) Interval() time.Duration {
+	t, ok := s.orig.Attributes().Get(semconvOtelEntityInterval)
+	if !ok {
+		return 0
+	}
+	return time.Duration(t.Int()) * time.Millisecond
+}
+
 // EntityDeleteDetails represents the details of an EntityDelete event.
 type EntityDeleteDetails struct {
 	orig plog.LogRecord

pkg/experimentalmetricmetadata/entity_events_test.go

@@ -19,6 +19,7 @@ func Test_Entity_State(t *testing.T) {
 	event.ID().PutStr("k8s.pod.uid", "123")
 	state := event.SetEntityState()
 	state.SetEntityType("k8s.pod")
+	state.SetInterval(1 * time.Hour)
 	state.Attributes().PutStr("label1", "value1")
 
 	actual := slice.At(0)
@@ -34,6 +35,8 @@ func Test_Entity_State(t *testing.T) {
 	assert.Equal(t, "value1", v.Str())
 
 	assert.Equal(t, "k8s.pod", actual.EntityStateDetails().EntityType())
+
+	assert.Equal(t, 1*time.Hour, actual.EntityStateDetails().Interval())
 }
 
 func Test_Entity_Delete(t *testing.T) {

receiver/hostmetricsreceiver/receiver.go

@@ -66,6 +66,10 @@ func (hmr *hostEntitiesReceiver) sendEntityEvent(ctx context.Context) {
 	state := entityEvent.SetEntityState()
 	state.SetEntityType(entityType)
 
+	if hmr.cfg.MetadataCollectionInterval != 0 {
+		state.SetInterval(hmr.cfg.MetadataCollectionInterval)
+	}
+
 	logs := out.ConvertAndMoveToLogs()
 
 	err := hmr.nextLogs.ConsumeLogs(ctx, logs)

receiver/k8sclusterreceiver/internal/metadata/entities.go

@@ -4,13 +4,15 @@
 package metadata // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/metadata"
 
 import (
+	"time"
+
 	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	metadataPkg "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata"
 )
 
 // GetEntityEvents processes metadata updates and returns entity events that describe the metadata changes.
-func GetEntityEvents(oldMetadata, newMetadata map[metadataPkg.ResourceID]*KubernetesMetadata, timestamp pcommon.Timestamp) metadataPkg.EntityEventsSlice {
+func GetEntityEvents(oldMetadata, newMetadata map[metadataPkg.ResourceID]*KubernetesMetadata, timestamp pcommon.Timestamp, reportingInterval time.Duration) metadataPkg.EntityEventsSlice {
 	out := metadataPkg.NewEntityEventsSlice()
 
 	for id, oldObj := range oldMetadata {
@@ -30,6 +32,9 @@ func GetEntityEvents(oldMetadata, newMetadata map[metadataPkg.ResourceID]*Kubern
 		entityEvent.ID().PutStr(newObj.ResourceIDKey, string(newObj.ResourceID))
 		state := entityEvent.SetEntityState()
 		state.SetEntityType(newObj.EntityType)
+		if reportingInterval != 0 {
+			state.SetInterval(reportingInterval)
+		}
 
 		attrs := state.Attributes()
 		for k, v := range newObj.Metadata {

receiver/k8sclusterreceiver/internal/metadata/entities_test.go

@@ -190,7 +190,7 @@ func Test_GetEntityEvents(t *testing.T) {
 
 				// Convert and test expected events.
 				timestamp := pcommon.NewTimestampFromTime(time.Now())
-				events := GetEntityEvents(tt.old, tt.new, timestamp)
+				events := GetEntityEvents(tt.old, tt.new, timestamp, 1*time.Hour)
 				require.Equal(t, tt.events.Len(), events.Len())
 				for i := 0; i < events.Len(); i++ {
 					actual := events.At(i)
@@ -202,6 +202,7 @@ func Test_GetEntityEvents(t *testing.T) {
 						estate := expected.EntityStateDetails()
 						astate := actual.EntityStateDetails()
 						assert.EqualValues(t, estate.EntityType(), astate.EntityType())
+						assert.EqualValues(t, 1*time.Hour, astate.Interval())
 						assert.EqualValues(t, estate.Attributes().AsRaw(), astate.Attributes().AsRaw())
 					}
 				}

receiver/k8sclusterreceiver/watcher.go

@@ -366,7 +366,7 @@ func (rw *resourceWatcher) syncMetadataUpdate(oldMetadata, newMetadata map[exper
 
 	if rw.entityLogConsumer != nil {
 		// Represent metadata update as entity events.
-		entityEvents := metadata.GetEntityEvents(oldMetadata, newMetadata, timestamp)
+		entityEvents := metadata.GetEntityEvents(oldMetadata, newMetadata, timestamp, rw.config.MetadataCollectionInterval)
 
 		// Convert entity events to log representation.
 		logs := entityEvents.ConvertAndMoveToLogs()

receiver/k8sclusterreceiver/watcher_test.go

@@ -229,7 +229,7 @@ func TestSyncMetadataAndEmitEntityEvents(t *testing.T) {
 	origPod := pods[0]
 	updatedPod := getUpdatedPod(origPod)
 
-	rw := newResourceWatcher(receivertest.NewNopSettings(), &Config{}, metadata.NewStore())
+	rw := newResourceWatcher(receivertest.NewNopSettings(), &Config{MetadataCollectionInterval: 2 * time.Hour}, metadata.NewStore())
 	rw.entityLogConsumer = logsConsumer
 
 	step1 := time.Now()
@@ -266,6 +266,7 @@ func TestSyncMetadataAndEmitEntityEvents(t *testing.T) {
 	lr := logsConsumer.AllLogs()[0].ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0)
 	expected := map[string]any{
 		"otel.entity.event.type": "entity_state",
+		"otel.entity.interval":   int64(7200000), // 2h in milliseconds
 		"otel.entity.type":       "k8s.pod",
 		"otel.entity.id":         map[string]any{"k8s.pod.uid": "pod0"},
 		"otel.entity.attributes": map[string]any{"pod.creation_timestamp": "0001-01-01T00:00:00Z"},