Add resourceVersion persistence feature to k8s receivers

This commit implements resourceVersion persistence for k8sobjects and
k8sevents receivers to prevent duplicate events on collector restart.

Key changes:
- Add checkpointer component for storing/retrieving resourceVersions
- Modify observer to compare list vs persisted versions and use highest
- Add storage client initialization in receivers
- Update config schemas with new persist_resource_version and storage fields
- Add comprehensive unit tests for persistence logic
- Consolidate test files into main test files

The feature is opt-in via persist_resource_version config flag and
requires a storage extension to be configured.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: dhruv-shah-sumo

internal/k8sinventory/go.mod

@@ -3,22 +3,28 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sinv
 go 1.25.0
 
 require (
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage v0.147.0
 	github.com/stretchr/testify v1.11.1
+	go.opentelemetry.io/collector/extension/xextension v0.147.0
 	go.uber.org/zap v1.27.1
 	k8s.io/api v0.35.2
 	k8s.io/apimachinery v0.35.2
 	k8s.io/client-go v0.35.2
 )
 
 require (
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/fxamacker/cbor/v2 v2.9.0 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.23.0 // indirect
 	github.com/google/gnostic-models v0.7.0 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/go-version v1.8.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
@@ -27,6 +33,18 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
+	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+	go.opentelemetry.io/collector/component v1.53.0 // indirect
+	go.opentelemetry.io/collector/component/componenttest v0.147.0 // indirect
+	go.opentelemetry.io/collector/extension v1.53.0 // indirect
+	go.opentelemetry.io/collector/featuregate v1.53.0 // indirect
+	go.opentelemetry.io/collector/internal/componentalias v0.147.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.53.0 // indirect
+	go.opentelemetry.io/otel v1.40.0 // indirect
+	go.opentelemetry.io/otel/metric v1.40.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.40.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.40.0 // indirect
+	go.opentelemetry.io/otel/trace v1.40.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
@@ -36,7 +54,7 @@ require (
 	golang.org/x/term v0.40.0 // indirect
 	golang.org/x/text v0.34.0 // indirect
 	golang.org/x/time v0.9.0 // indirect
-	google.golang.org/protobuf v1.36.10 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

internal/k8sinventory/go.sum

@@ -0,0 +1,89 @@
+package watch
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.opentelemetry.io/collector/extension/xextension/storage"
+	"go.uber.org/zap"
+)
+
+type checkpointer struct {
+	client storage.Client
+	logger *zap.Logger
+}
+
+const checkpointKeyFormat = "latestResourceVersion/%s"
+
+func newCheckpointer(client storage.Client, logger *zap.Logger) *checkpointer {
+	return &checkpointer{
+		client: client,
+		logger: logger,
+	}
+}
+
+func (c *checkpointer) GetResourceVersion(ctx context.Context,
+	namespace, objectType string) (string, error) {
+	if c.client == nil {
+		return "", errors.New("storage client is nil")
+	}
+
+	checkPointKey := c.getCheckpointKey(namespace, objectType)
+	data, err := c.client.Get(ctx, checkPointKey)
+	if err != nil {
+		c.logger.Warn("Error retrieving checkpoint",
+			zap.String("namespace", namespace),
+			zap.String("objectType", objectType),
+			zap.Error(err))
+		return "", fmt.Errorf("failed to retrieve checkpoint: %w", err)
+	}
+
+	// If key is not found, data and error is nil
+	if len(data) == 0 {
+		c.logger.Debug("No checkpoint found, starting from the beginning",
+			zap.String("key", checkPointKey))
+		return "", nil
+	}
+	return string(data), nil
+}
+
+func (c *checkpointer) SetResourceVersion(
+	ctx context.Context,
+	namespace, objectType, resourceVersion string) error {
+	if c.client == nil {
+		return errors.New("storage client is nil")
+	}
+
+	key := c.getCheckpointKey(namespace, objectType)
+	if key == "" {
+		return fmt.Errorf("checkpoint key is empty: %s, %s", namespace, objectType)
+	}
+
+	if err := c.client.Set(ctx, key, []byte(resourceVersion)); err != nil {
+		return fmt.Errorf("failed to store resourceVersion with key %s: %w", key, err)
+	}
+
+	c.logger.Debug("Checkpoint saved",
+		zap.String("namespace", namespace),
+		zap.String("objectType", objectType))
+
+	return nil
+}
+
+// getCheckpointKey generates a unique storage key
+// returns resourceVersion key for global watch stream (without namespace) or
+// per namespace watch stream.
+func (*checkpointer) getCheckpointKey(namespace, objectType string) string {
+	// when watch stream is cluster-wide or cluster-scoped resource (no namespace),
+	// the resource version is persisted per object type only.
+	if namespace == "" {
+		// example: latestResourceVersion/nodes, latestResourceVersion/namespaces
+		return fmt.Sprintf(checkpointKeyFormat, objectType)
+	}
+
+	// when watch stream is created per namespace, the resource version is persisted
+	// per object type per namespace.
+	// example: latestResourceVersion/pods.default, latestResourceVersion/configmaps.kube-system
+	return fmt.Sprintf("%s.%s", fmt.Sprintf(checkpointKeyFormat, objectType), namespace)
+}