receiver/k8sobjects: close race window between sendInitialState and watch

Use the list ResourceVersion returned by sendInitialState directly as
the watch starting point on the first iteration, eliminating a redundant
second List() call and the race window it created.

Previously, sendInitialState() did a List() and then getResourceVersion()
did another List() to determine where to start watching. Events occurring
between the two calls could be missed or duplicated.

Now sendInitialState() returns the list's own ResourceVersion (always >=
any individual object RV), which is used directly on the first watch
iteration. Subsequent iterations (e.g. after 410 Gone) fall back to
getResourceVersion() as before.

Also update the checkpoint with the list RV after sendInitialState so
the persisted value accurately reflects the snapshot point of the
listing, not just the highest individual object RV.

Co-authored-by: Dhruv Shah <dhruv.shah@sumologic.com>

Author: dhruv-shah-sumo

internal/k8sinventory/go.mod

@@ -5,8 +5,8 @@ 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.54.1-0.20260326211300-c04f6776a74c
-	go.opentelemetry.io/collector/extension/xextension v0.148.1-0.20260326211300-c04f6776a74c
+	go.opentelemetry.io/collector/component v1.54.1-0.20260330144813-4d17eb8959de
+	go.opentelemetry.io/collector/extension/xextension v0.148.1-0.20260330144813-4d17eb8959de
 	go.uber.org/zap v1.27.1
 	k8s.io/api v0.35.3
 	k8s.io/apimachinery v0.35.3
@@ -35,11 +35,11 @@ require (
 	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.148.1-0.20260326211300-c04f6776a74c // indirect
-	go.opentelemetry.io/collector/extension v1.54.1-0.20260326211300-c04f6776a74c // indirect
-	go.opentelemetry.io/collector/featuregate v1.54.1-0.20260326211300-c04f6776a74c // indirect
-	go.opentelemetry.io/collector/internal/componentalias v0.148.1-0.20260326211300-c04f6776a74c // indirect
-	go.opentelemetry.io/collector/pdata v1.54.1-0.20260326211300-c04f6776a74c // indirect
+	go.opentelemetry.io/collector/component/componenttest v0.148.1-0.20260330144813-4d17eb8959de // indirect
+	go.opentelemetry.io/collector/extension v1.54.1-0.20260330144813-4d17eb8959de // indirect
+	go.opentelemetry.io/collector/featuregate v1.54.1-0.20260330144813-4d17eb8959de // indirect
+	go.opentelemetry.io/collector/internal/componentalias v0.148.1-0.20260330144813-4d17eb8959de // indirect
+	go.opentelemetry.io/collector/pdata v1.54.1-0.20260330144813-4d17eb8959de // indirect
 	go.opentelemetry.io/otel v1.42.0 // indirect
 	go.opentelemetry.io/otel/metric v1.42.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.42.0 // indirect

internal/k8sinventory/go.sum

@@ -101,7 +101,7 @@ func (o *Observer) startCheckpointFlusher(ctx context.Context, namespace string)
 	setLatestRV = func(rv string) {
 		// Buffer the latest rv in the checkpointer; Flush will write it to storage.
 		if err := o.checkpointer.SetCheckpoint(context.Background(), namespace, o.config.Gvr.Resource, rv); err != nil {
-			o.logger.Warn("failed to buffer resourceVersion",
+			o.logger.Error("failed to buffer resourceVersion",
 				zap.String("namespace", namespace),
 				zap.String("resource", o.config.Gvr.Resource),
 				zap.Error(err))
@@ -110,7 +110,7 @@ func (o *Observer) startCheckpointFlusher(ctx context.Context, namespace string)
 
 	flush = func() {
 		if err := o.checkpointer.Flush(context.Background()); err != nil {
-			o.logger.Warn("failed to flush checkpoints",
+			o.logger.Error("failed to flush checkpoints",
 				zap.String("namespace", namespace),
 				zap.String("resource", o.config.Gvr.Resource),
 				zap.Error(err))
@@ -154,22 +154,40 @@ func (o *Observer) startWatch(ctx context.Context, resource dynamic.ResourceInte
 	setLatestRV, flushCheckpoint, stopFlusher := o.startCheckpointFlusher(cancelCtx, namespace)
 	defer stopFlusher()
 
+	// initialListRV holds the list resourceVersion returned by sendInitialState.
+	// It is used as the watch starting point on the first iteration, eliminating
+	// a second List() call and closing the race window between the two listings.
+	// It is cleared after the first iteration so subsequent restarts (e.g. after
+	// a 410 Gone) fall back to getResourceVersion() as normal.
+	var initialListRV string
 	if o.config.IncludeInitialState {
-		o.sendInitialState(ctx, resource, namespace, setLatestRV)
-		// Force-flush immediately so the rv from the initial state is durable before
-		// the watch loop starts. Without this, a 410 arriving within the first 5s
-		// would delete a checkpoint that was never written due to the flush interval delay.
+		initialListRV = o.sendInitialState(ctx, resource, namespace, setLatestRV)
+		// Update the checkpoint with the list's own RV, which is >= any individual
+		// object RV and represents the precise snapshot point of the listing.
+		if initialListRV != "" {
+			setLatestRV(initialListRV)
+		}
+		// Force-flush immediately so the rv is durable before the watch loop starts.
 		flushCheckpoint()
 	}
 
 	wait.UntilWithContext(cancelCtx, func(newCtx context.Context) {
-		resourceVersion, err := o.getResourceVersion(newCtx, resource, namespace)
-		if err != nil {
-			o.logger.Error("could not retrieve a resourceVersion",
-				zap.String("resource", o.config.Gvr.String()),
-				zap.Error(err))
-			cancel()
-			return
+		var resourceVersion string
+		if initialListRV != "" {
+			// First iteration: reuse the list RV from sendInitialState directly,
+			// avoiding a redundant List() call and the race window it creates.
+			resourceVersion = initialListRV
+			initialListRV = ""
+		} else {
+			var err error
+			resourceVersion, err = o.getResourceVersion(newCtx, resource, namespace)
+			if err != nil {
+				o.logger.Error("could not retrieve a resourceVersion",
+					zap.String("resource", o.config.Gvr.String()),
+					zap.Error(err))
+				cancel()
+				return
+			}
 		}
 
 		done := o.doWatch(ctx, resourceVersion, namespace, watchFunc, stopperChan, setLatestRV)
@@ -186,7 +204,7 @@ func (o *Observer) startWatch(ctx context.Context, resource dynamic.ResourceInte
 		// This handles 410 Gone errors where the persisted resourceVersion is too old
 		if o.checkpointer != nil {
 			if err := o.checkpointer.DeleteCheckpoint(context.Background(), namespace, o.config.Gvr.Resource); err != nil {
-				o.logger.Warn("failed to delete persisted resourceVersion after watch restart",
+				o.logger.Error("failed to delete persisted resourceVersion after watch restart",
 					zap.String("namespace", namespace),
 					zap.String("resource", o.config.Gvr.Resource),
 					zap.Error(err))
@@ -201,7 +219,9 @@ func (o *Observer) startWatch(ctx context.Context, resource dynamic.ResourceInte
 // those objects were already seen before the last checkpoint. setLatestRV is
 // called for every event that is emitted so the flusher tracks the high-water
 // mark across the initial listing as well as the watch loop.
-func (o *Observer) sendInitialState(ctx context.Context, resource dynamic.ResourceInterface, namespace string, setLatestRV func(string)) {
+// It returns the list's own ResourceVersion, which the caller should use as
+// the watch starting point to avoid a redundant List() call.
+func (o *Observer) sendInitialState(ctx context.Context, resource dynamic.ResourceInterface, namespace string, setLatestRV func(string)) string {
 	o.logger.Info("sending initial state",
 		zap.String("resource", o.config.Gvr.String()),
 		zap.Strings("namespaces", o.config.Namespaces))
@@ -231,13 +251,15 @@ func (o *Observer) sendInitialState(ctx context.Context, resource dynamic.Resour
 		o.logger.Error("error in listing objects for initial state",
 			zap.String("resource", o.config.Gvr.String()),
 			zap.Error(err))
-		return
+		return ""
 	}
 
+	listRV := objects.GetResourceVersion()
+
 	if len(objects.Items) == 0 {
 		o.logger.Debug("no objects found for initial state",
 			zap.String("resource", o.config.Gvr.String()))
-		return
+		return listRV
 	}
 
 	emitted := 0
@@ -271,8 +293,11 @@ func (o *Observer) sendInitialState(ctx context.Context, resource dynamic.Resour
 	}
 
 	o.logger.Info("initial state sent",
+		zap.String("namespace", namespace),
+		zap.String("list_rv", listRV),
 		zap.String("resource", o.config.Gvr.String()),
 		zap.Int("object_count", emitted))
+	return listRV
 }
 
 // doWatch returns true when watching is done, false when watching should be restarted.

internal/k8sinventory/watch/observer.go

@@ -367,6 +367,76 @@ func TestObserverInitialStateNoObjects(t *testing.T) {
 	wg.Wait()
 }
 
+// TestSendInitialStateReturnsListRV verifies that sendInitialState returns the
+// list's own ResourceVersion, not just the highest individual object RV.
+// The list RV is always >= any individual object RV and is the correct starting
+// point for the subsequent watch to avoid a race window between two List calls.
+func TestSendInitialStateReturnsListRV(t *testing.T) {
+	mockClient := newMockDynamicClient()
+	// Set list RV to "999", higher than any individual object RV below.
+	mockClient.setListResourceVersion("999")
+	mockClient.createPods(
+		generatePod("pod1", "default", nil, "100"),
+		generatePod("pod2", "default", nil, "200"),
+	)
+
+	cfg := Config{
+		Config: k8sinventory.Config{
+			Gvr:        schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
+			Namespaces: []string{"default"},
+		},
+		IncludeInitialState: true,
+	}
+
+	obs, err := New(mockClient, cfg, zap.NewNop(), nil, nil)
+	require.NoError(t, err)
+
+	resource := mockClient.Resource(cfg.Gvr)
+	listRV := obs.sendInitialState(t.Context(), resource.Namespace("default"), "default", func(string) {})
+	assert.Equal(t, "999", listRV, "sendInitialState should return the list's own ResourceVersion")
+}
+
+// TestInitialStateListRVPersistedAsCheckpoint verifies that after sendInitialState
+// the checkpoint is updated with the list's RV (via setLatestRV in startWatch),
+// which is more accurate than the highest individual object RV.
+func TestInitialStateListRVPersistedAsCheckpoint(t *testing.T) {
+	mockClient := newMockDynamicClient()
+	storageClient := storagetest.NewInMemoryClient(component.KindReceiver, component.MustNewID("test"), "test")
+
+	// List RV "999" is higher than any individual pod RV — after startup the
+	// checkpoint should hold "999", not "200".
+	mockClient.setListResourceVersion("999")
+	mockClient.createPods(
+		generatePod("pod1", "default", nil, "100"),
+		generatePod("pod2", "default", nil, "200"),
+	)
+
+	cfg := Config{
+		Config: k8sinventory.Config{
+			Gvr:        schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
+			Namespaces: []string{"default"},
+		},
+		IncludeInitialState:    true,
+		PersistResourceVersion: true,
+	}
+
+	obs, err := New(mockClient, cfg, zap.NewNop(), storageClient, nil)
+	require.NoError(t, err)
+
+	wg := sync.WaitGroup{}
+	stopChan := obs.Start(t.Context(), &wg)
+
+	time.Sleep(200 * time.Millisecond)
+
+	close(stopChan)
+	wg.Wait()
+
+	cp := newCheckpointer(storageClient, zap.NewNop())
+	rv, err := cp.GetCheckpoint(t.Context(), "default", "pods")
+	require.NoError(t, err)
+	assert.Equal(t, "999", rv, "checkpoint should hold the list RV, not a lower individual object RV")
+}
+
 func verifyReceivedEvents(t *testing.T, numEvents int, receivedEventsChan chan *apiWatch.Event, stopChan chan struct{}) {
 	receivedEvents := 0
 

internal/k8sinventory/watch/observer_test.go

@@ -33,7 +33,7 @@ the K8s API server. This can be one of `none` (for no auth), `serviceAccount`
 - `namespaces` (default = `all`): An array of `namespaces` to collect events from.
 This receiver will continuously watch all the `namespaces` mentioned in the array for
 new events.
-- `storage` (default: none): specifies the storage extension to use for persisting resourceVersions. Required when `persist_resource_version` is enabled.
+- `storage` (default: none): specifies the storage extension to use for persisting resourceVersions. Required when `persist_resource_version` is enabled. The attached storage should be a persistent volume, which is accessible by all the nodes in the given cluster. If local volume is used, persistence will break when the pod is moved to another node.
 - `persist_resource_version` (default: `false`): when set to `true`, the receiver persists the resourceVersion after processing each event. On restart, the receiver resumes watching from the persisted resourceVersion, preventing duplicate events. Requires a `storage` extension to be configured.
 
 Examples:

receiver/k8seventsreceiver/README.md

@@ -69,7 +69,7 @@ use this config to specify the group to select. By default, it will select the f
 For example, `events` resource is available in both `v1` and `events.k8s.io/v1` APIGroup. In
 this case, it will select `v1` by default.
 - `k8s_leader_elector` (default: none): if specified, will enable Leader Election by using `k8sleaderelector` extension
-- `storage` (default: none): specifies the storage extension to use for persisting resourceVersions. Required when `persist_resource_version` is enabled.
+- `storage` (default: none): specifies the storage extension to use for persisting resourceVersions. Required when `persist_resource_version` is enabled. The attached storage should be a persistent volume, which is accessible by all the nodes in the given cluster. If local volume is used, persistence will break when the pod is moved to another node.
 - `persist_resource_version` (default: `false`): when set to `true` (watch-mode only), the receiver persists the resourceVersion after processing each event. On restart, the receiver resumes watching from the persisted resourceVersion, preventing duplicate events. Requires a `storage` extension to be configured.
 
 

receiver/k8sobjectsreceiver/README.md

@@ -102,7 +102,7 @@ func (c *Config) Validate() error {
 		}
 
 		if object.PersistResourceVersion && object.ResourceVersion != "" {
-			return errors.New("resource_version cannot be set when persist_resource_version is enabled; the persisted version takes priority")
+			return errors.New("resource_version cannot be set when persist_resource_version is enabled")
 		}
 
 		if len(object.ExcludeNamespaces) != 0 && len(object.Namespaces) != 0 {

receiver/k8sobjectsreceiver/config.go

@@ -246,7 +246,7 @@ func TestConfigValidationPersistResourceVersion(t *testing.T) {
 					},
 				},
 			},
-			expectedErr: "resource_version cannot be set when persist_resource_version is enabled; the persisted version takes priority",
+			expectedErr: "resource_version cannot be set when persist_resource_version is enabled",
 		},
 		{
 			desc: "resource_version set without persist_resource_version is valid",

receiver/k8sobjectsreceiver/config_test.go

@@ -20,7 +20,7 @@ require (
 	go.opentelemetry.io/collector/confmap v1.54.1-0.20260330144813-4d17eb8959de
 	go.opentelemetry.io/collector/consumer v1.54.1-0.20260330144813-4d17eb8959de
 	go.opentelemetry.io/collector/consumer/consumertest v0.148.1-0.20260330144813-4d17eb8959de
-	go.opentelemetry.io/collector/extension/xextension v0.148.1-0.20260326211300-c04f6776a74c
+	go.opentelemetry.io/collector/extension/xextension v0.148.1-0.20260330144813-4d17eb8959de
 	go.opentelemetry.io/collector/filter v0.148.1-0.20260330144813-4d17eb8959de
 	go.opentelemetry.io/collector/pdata v1.54.1-0.20260330144813-4d17eb8959de
 	go.opentelemetry.io/collector/receiver v1.54.1-0.20260330144813-4d17eb8959de