[receiver/k8sclusterreceiver] Report entity state periodically (#24434)

Resolves
#24413

This is part of the work to move to entity events-as-log-records in K8s
cluster receiver:
#19741

Overall design document:

https://docs.google.com/document/d/1Tg18sIck3Nakxtd3TFFcIjrmRO_0GLMdHXylVqBQmJA/

Note that metadata.GetMetadataUpdate() computes deltas between the old
and new state. If the old and new states are equal
metadata.GetMetadataUpdate() returns an empty slice. This means periodic
polling when no state is changed will NOT result in any additional
metadata update calls. So, we expect no changes in the behavior of
existing metadata listeners, such as signalfxexporter.

As opposed to that entity events are always emitted even if no state is
changed. So, this periodic collection will result in periodic emitting
of entity events. This is the desirable effect of this PR.

### Testing

I was unable to think of a good automated test for this capability that
does not require a major refactoring of other code and executes
reasonably quickly. For this reason I am not including any tests in this
PR. I have tested the functionality manually and can see the entities
periodically collected according to the config setting.

If anyone has ideas about how to write good unit tests for this PR I am
open to suggestions.

.chloggen/k8sclusterreceiver-metadata-collection-interval.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# 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: k8sclustereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Report entity state periodically
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24413]
+
+# (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:

receiver/k8sclusterreceiver/README.md

@@ -38,6 +38,13 @@ The following settings are optional:
 for events using K8s API. However, the metrics collected are emitted only
 once every collection interval. `collection_interval` will determine the
 frequency at which metrics are emitted by this receiver.
+- `metadata_collection_interval` (default = `5m`): Collection interval for metadata
+for K8s entities such as pods, nodes, etc.
+Metadata of the particular entity in the cluster is collected when the entity changes.
+In addition metadata of all entities is collected periodically even if no changes happen.
+This setting controls the interval between periodic collections.
+Setting the duration to 0 will disable periodic collection (however will not impact
+metadata collection on changes).
 - `node_conditions_to_report` (default = `[Ready]`): An array of node
 conditions this receiver should report. See
 [here](https://kubernetes.io/docs/concepts/architecture/nodes/#condition) for

receiver/k8sclusterreceiver/config.go

@@ -28,6 +28,13 @@ type Config struct {
 
 	// Whether OpenShift supprot should be enabled or not.
 	Distribution string `mapstructure:"distribution"`
+
+	// Collection interval for metadata.
+	// Metadata of the particular entity in the cluster is collected when the entity changes.
+	// In addition metadata of all entities is collected periodically even if no changes happen.
+	// Setting the duration to 0 will disable periodic collection (however will not impact
+	// metadata collection on changes).
+	MetadataCollectionInterval time.Duration `mapstructure:"metadata_collection_interval"`
 }
 
 func (cfg *Config) Validate() error {

receiver/k8sclusterreceiver/config_test.go

@@ -43,6 +43,7 @@ func TestLoadConfig(t *testing.T) {
 				APIConfig: k8sconfig.APIConfig{
 					AuthType: k8sconfig.AuthTypeServiceAccount,
 				},
+				MetadataCollectionInterval: 30 * time.Minute,
 			},
 		},
 		{
@@ -54,6 +55,7 @@ func TestLoadConfig(t *testing.T) {
 				APIConfig: k8sconfig.APIConfig{
 					AuthType: k8sconfig.AuthTypeServiceAccount,
 				},
+				MetadataCollectionInterval: 5 * time.Minute,
 			},
 		},
 	}

receiver/k8sclusterreceiver/factory.go

@@ -20,8 +20,9 @@ const (
 	distributionOpenShift  = "openshift"
 
 	// Default config values.
-	defaultCollectionInterval = 10 * time.Second
-	defaultDistribution       = distributionKubernetes
+	defaultCollectionInterval         = 10 * time.Second
+	defaultDistribution               = distributionKubernetes
+	defaultMetadataCollectionInterval = 5 * time.Minute
 )
 
 var defaultNodeConditionsToReport = []string{"Ready"}
@@ -34,6 +35,7 @@ func createDefaultConfig() component.Config {
 		APIConfig: k8sconfig.APIConfig{
 			AuthType: k8sconfig.AuthTypeServiceAccount,
 		},
+		MetadataCollectionInterval: defaultMetadataCollectionInterval,
 	}
 }
 

receiver/k8sclusterreceiver/factory_test.go

@@ -38,6 +38,7 @@ func TestFactory(t *testing.T) {
 		APIConfig: k8sconfig.APIConfig{
 			AuthType: k8sconfig.AuthTypeServiceAccount,
 		},
+		MetadataCollectionInterval: 5 * time.Minute,
 	}, rCfg)
 
 	r, err := f.CreateTracesReceiver(

receiver/k8sclusterreceiver/testdata/config.yaml

@@ -4,6 +4,7 @@ k8s_cluster/all_settings:
   node_conditions_to_report: [ "Ready", "MemoryPressure" ]
   allocatable_types_to_report: [ "cpu","memory" ]
   metadata_exporters: [ nop ]
+  metadata_collection_interval: 30m
 k8s_cluster/partial_settings:
   collection_interval: 30s
   distribution: openshift

receiver/k8sclusterreceiver/watcher.go

@@ -105,7 +105,7 @@ func (rw *resourceWatcher) initialize() error {
 }
 
 func (rw *resourceWatcher) prepareSharedInformerFactory() error {
-	factory := informers.NewSharedInformerFactoryWithOptions(rw.client, 0)
+	factory := informers.NewSharedInformerFactoryWithOptions(rw.client, rw.config.MetadataCollectionInterval)
 
 	// Map of supported group version kinds by name of a kind.
 	// If none of the group versions are supported by k8s server for a specific kind,

receiver/k8sclusterreceiver/watcher_test.go

@@ -198,6 +198,7 @@ func TestPrepareSharedInformerFactory(t *testing.T) {
 				client:        newFakeClientWithAllResources(),
 				logger:        obsLogger,
 				dataCollector: collection.NewDataCollector(receivertest.NewNopCreateSettings(), []string{}, []string{}),
+				config:        &Config{},
 			}
 
 			assert.NoError(t, rw.prepareSharedInformerFactory())