Adds resourceVersion support to k8sObject receiver (#46543)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Implements optional resourceVersion checkpointing to prevent duplicate
events on collector restart.
Resource version persistence is developed as part of watch mechanism
which is being used by k8sObjects receiver and k8sEvents receiver. The
feature is not enabled for `pull` mode in any of the receivers.

  Features:
  - Opt-in via declaring storage extension for persistence
- Namespace-aware checkpointing: Creates per stream checkpoint when
separate namespaces are mentioned. Per namespace keys are maintained in
the storage. The format is
`latestResourceVersion/configmaps.kube-system` for per namespace watch
stream. In case namespaces are not specified, a global clusterwide watch
stream is created that creates a different checkpoint key with format:
`latestResourceVersion/pods, latestResourceVersion/nodes`
  - Watch mode only (validated at config time)
- If persistence is enabled, then `resource_version` provided in the
config will be ignored. The persisted version will be used to kick-off
the watch stream.
- In case of stale persisted version, a List() API will be called to get
the latest resource version available which is the existing way of
handling stale resource version in k8s inventory.

  Configuration:
```
  receivers:
    k8sobjects:
      storage: file_storage
      objects:
        - name: pods
          mode: watch
```


  How it works:
1. At the start of each watch stream, `getResourceVersion()` is called
which checks if persistence is enabled then it retrieves the persisted
version.
- If persisted version is empty then it calls List() API to get the list
of available objects and picks the latest version and persists it.
- If persisted version is non-empty then it is supplied to kick-off the
watch stream.
- In case, the persisted version supplied to start watch stream in the
step above is stale then checkpoint is deleted. This prompts the List()
API call again to get the latest available resource version for the
given stream.
2. After processing each watch event, saves resourceVersion to storage.
3. In case persistence operation fails for any reason, the watch stream
continues.



#### Link to tracking issue
Fixes
#46017

<!--Describe what testing was performed and which tests were added.-->
#### Testing
- Unit tests
- Manually built the updated image using ` make docker-otelcontribcol`
and deployed in local. Tested various restarts and 410 handling
scenarios for global watch streams, per namespace watch streams.

<!--Describe the documentation added.-->
#### Documentation
- Updated schema yaml files for k8sobject and k8sevents receiver.
<!--Please delete paragraphs that you did not use before submitting.-->

Example - how the new config would look like?

```
extensions:
  file_storage:
    directory: /var/lib/otelcol/storage
    timeout: 10s

receivers:
  k8sobjects:
    auth_type: serviceAccount
    storage: file_storage
    include_initial_state: true
    objects:
      - name: pods
        mode: watch
        namespaces: [default]
      - name: events
        mode: watch
        namespaces: [default]

exporters:
  nop:

service:
  telemetry:
    logs:
      level: debug
  extensions: [file_storage]
  pipelines:
    logs:
      receivers: [k8sobjects]
      exporters: [nop]

```

---------

Signed-off-by: Dhruv Shah <dhruv.shah@sumologic.com>

Author: dhruv-shah-sumo

.chloggen/k8sobjects-persist-resourceversion.yaml

@@ -0,0 +1,11 @@
+change_type: enhancement
+
+component: receiver/k8sobjects
+
+note: When `storage` is configured, watch-mode objects automatically resume from the last seen resourceVersion across restarts, preventing event duplication.
+
+issues: [46543]
+
+subtext:
+
+change_logs: [user]

internal/k8sinventory/go.mod

@@ -3,22 +3,29 @@ 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/component v1.56.1-0.20260415114935-307e3abdbae9
+	go.opentelemetry.io/collector/extension/xextension v0.150.1-0.20260415114935-307e3abdbae9
 	go.uber.org/zap v1.27.1
 	k8s.io/api v0.35.3
 	k8s.io/apimachinery v0.35.3
 	k8s.io/client-go v0.35.3
 )
 
 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.9.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,16 +34,27 @@ 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/componenttest v0.150.1-0.20260415114935-307e3abdbae9 // indirect
+	go.opentelemetry.io/collector/extension v1.56.1-0.20260415114935-307e3abdbae9 // indirect
+	go.opentelemetry.io/collector/featuregate v1.56.1-0.20260415114935-307e3abdbae9 // indirect
+	go.opentelemetry.io/collector/internal/componentalias v0.150.1-0.20260415114935-307e3abdbae9 // indirect
+	go.opentelemetry.io/collector/pdata v1.56.1-0.20260415114935-307e3abdbae9 // indirect
+	go.opentelemetry.io/otel v1.43.0 // indirect
+	go.opentelemetry.io/otel/metric v1.43.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.43.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.43.0 // indirect
+	go.opentelemetry.io/otel/trace v1.43.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
 	golang.org/x/net v0.51.0 // indirect
 	golang.org/x/oauth2 v0.30.0 // indirect
-	golang.org/x/sys v0.41.0 // indirect
+	golang.org/x/sys v0.42.0 // indirect
 	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
@@ -48,3 +66,5 @@ require (
 	sigs.k8s.io/structured-merge-diff/v6 v6.3.0 // indirect
 	sigs.k8s.io/yaml v1.6.0 // indirect
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage => ../../extension/storage

internal/k8sinventory/go.sum

@@ -0,0 +1,181 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package watch // import "github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sinventory/watch"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"sync"
+
+	"go.opentelemetry.io/collector/extension/xextension/storage"
+	"go.uber.org/zap"
+)
+
+type checkpointer struct {
+	client storage.Client
+	logger *zap.Logger
+
+	// pending holds the latest resourceVersion per storage key, buffered in
+	// memory across all watch streams. Flush() drains it to persistent storage.
+	mu      sync.Mutex
+	pending map[string]string
+}
+
+const checkpointKeyFormat = "latestResourceVersion/%s"
+
+func newCheckpointer(client storage.Client, logger *zap.Logger) *checkpointer {
+	return &checkpointer{
+		client:  client,
+		logger:  logger,
+		pending: make(map[string]string),
+	}
+}
+
+func (c *checkpointer) GetCheckpoint(ctx context.Context, namespace, objectType string) (string, error) {
+	if c.client == nil {
+		return "", errors.New("storage client is nil")
+	}
+
+	checkPointKey := c.getCheckpointKey(namespace, objectType)
+	c.logger.Debug("Retrieving checkpoint, key: "+checkPointKey,
+		zap.String("namespace", namespace),
+		zap.String("objectType", 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
+}
+
+// SetCheckpoint buffers the latest resourceVersion for the given namespace and
+// objectType in memory. Call Flush to persist all buffered values to storage.
+// Only updates the in-memory value if the new resourceVersion is numerically
+// greater than the current one, acting as a high-watermark. This guards against
+// out-of-order resourceVersions from List() responses (which are ordered by
+// object key, not by resourceVersion).
+func (c *checkpointer) SetCheckpoint(
+	_ context.Context,
+	namespace, objectType, resourceVersion string,
+) error {
+	key := c.getCheckpointKey(namespace, objectType)
+	if key == "" {
+		return fmt.Errorf("checkpoint key is empty: %s, %s", namespace, objectType)
+	}
+
+	newRV, err := strconv.ParseInt(resourceVersion, 10, 64)
+	if err != nil {
+		return fmt.Errorf("invalid resourceVersion %q: %w", resourceVersion, err)
+	}
+
+	c.mu.Lock()
+	if existing, ok := c.pending[key]; ok {
+		if existingRV, err := strconv.ParseInt(existing, 10, 64); err == nil && newRV <= existingRV {
+			c.mu.Unlock()
+			return nil
+		}
+	}
+	c.pending[key] = resourceVersion
+	c.mu.Unlock()
+
+	c.logger.Debug("buffered resourceVersion checkpoint",
+		zap.String("key", key),
+		zap.String("resourceVersion", resourceVersion))
+
+	return nil
+}
+
+// Flush writes all buffered checkpoints to persistent storage. Only the latest
+// value per key is written, discarding any intermediate updates since the last
+// flush. It is safe to call concurrently from multiple goroutines.
+func (c *checkpointer) Flush(ctx context.Context) error {
+	if c.client == nil {
+		return errors.New("storage client is nil")
+	}
+
+	c.mu.Lock()
+	if len(c.pending) == 0 {
+		c.mu.Unlock()
+		return nil
+	}
+	snapshot := c.pending
+	// Setting c.pending to an empty map to avoid unnecessary writes when there are no pending updates
+	// to be flushed to the disk.
+	c.pending = make(map[string]string)
+	c.mu.Unlock()
+
+	failed := false
+	for key, rv := range snapshot {
+		if err := c.client.Set(ctx, key, []byte(rv)); err != nil {
+			c.logger.Error("failed to flush checkpoint",
+				zap.String("key", key),
+				zap.String("resourceVersion", rv),
+				zap.Error(err))
+			failed = true
+			continue
+		}
+		c.logger.Debug("flushed resourceVersion checkpoint",
+			zap.String("key", key),
+			zap.String("resourceVersion", rv))
+	}
+	if failed {
+		return errors.New("one or more checkpoints failed to be stored")
+	}
+	return nil
+}
+
+// DeleteCheckpoint deletes the persisted checkpoint for a given namespace and object type.
+// This is used when the persisted resourceVersion is no longer valid (e.g., after a 410 Gone error).
+func (c *checkpointer) DeleteCheckpoint(
+	ctx context.Context,
+	namespace, objectType 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.Delete(ctx, key); err != nil {
+		return fmt.Errorf("failed to delete resourceVersion with key %s: %w", key, err)
+	}
+
+	c.logger.Debug("Checkpoint deleted with key: "+key,
+		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)
+}