[internal/k8sinventory] Extract checkpoint package and update Observer interface

.chloggen/k8sobject_component.yaml

@@ -0,0 +1,29 @@
+# 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. receiver/filelog)
+component: internal/k8sinventory
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Extract checkpoint package and update Observer interface to return startup errors.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [43602]
+
+# (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: |
+  - Extracted `internal/k8sinventory/checkpoint` as a shared package so checkpoint logic can be reused across observer implementations.
+  - The `Observer` interface's `Start` method now returns an error, allowing callers to detect and propagate startup failures.
+
+# 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: [api]

internal/k8sinventory/watch/checkpointer.go → internal/k8sinventory/checkpoint/checkpointer.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package watch // import "github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sinventory/watch"
+package checkpoint // import "github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sinventory/checkpoint"
 
 import (
 	"context"
@@ -14,32 +14,37 @@ import (
 	"go.uber.org/zap"
 )
 
-type checkpointer struct {
+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
+
+	// persistedRVs holds RVs loaded from storage at startup for dedup.
+	// Populated by Load(); read by AlreadySeen().
+	persistedMu  sync.RWMutex
+	persistedRVs map[string]int64
 }
 
 const checkpointKeyFormat = "latestResourceVersion/%s"
 
-func newCheckpointer(client storage.Client, logger *zap.Logger) *checkpointer {
-	return &checkpointer{
+func New(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) {
+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)
+	checkPointKey := c.checkpointKey(namespace, objectType)
 	c.logger.Debug("Retrieving checkpoint, key: "+checkPointKey,
 		zap.String("namespace", namespace),
 		zap.String("objectType", objectType))
@@ -67,11 +72,11 @@ func (c *checkpointer) GetCheckpoint(ctx context.Context, namespace, objectType
 // 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(
+func (c *Checkpointer) SetCheckpoint(
 	_ context.Context,
 	namespace, objectType, resourceVersion string,
 ) error {
-	key := c.getCheckpointKey(namespace, objectType)
+	key := c.checkpointKey(namespace, objectType)
 	if key == "" {
 		return fmt.Errorf("checkpoint key is empty: %s, %s", namespace, objectType)
 	}
@@ -101,7 +106,7 @@ func (c *checkpointer) SetCheckpoint(
 // 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 {
+func (c *Checkpointer) Flush(ctx context.Context) error {
 	if c.client == nil {
 		return errors.New("storage client is nil")
 	}
@@ -137,17 +142,53 @@ func (c *checkpointer) Flush(ctx context.Context) error {
 	return nil
 }
 
+// Load reads the persisted RV for each namespace into memory.
+// Must be called before processing initial-list events so AlreadySeen() has data to compare against.
+func (c *Checkpointer) Load(ctx context.Context, namespaces []string, objectType string) error {
+	c.persistedMu.Lock()
+	defer c.persistedMu.Unlock()
+	c.persistedRVs = make(map[string]int64)
+	for _, ns := range namespaces {
+		rv, err := c.GetCheckpoint(ctx, ns, objectType)
+		if err != nil {
+			return err
+		}
+		if rv == "" {
+			continue
+		}
+		parsed, err := strconv.ParseInt(rv, 10, 64)
+		if err != nil {
+			return err
+		}
+		c.persistedRVs[ns] = parsed
+	}
+	return nil
+}
+
+// AlreadySeen reports whether the given resourceVersion is ≤ the persisted checkpoint
+// for the namespace, meaning the object was already processed before the last restart.
+func (c *Checkpointer) AlreadySeen(resourceVersion, namespace string) (bool, error) {
+	objRV, err := strconv.ParseInt(resourceVersion, 10, 64)
+	if err != nil {
+		return false, err
+	}
+	c.persistedMu.RLock()
+	persisted, exists := c.persistedRVs[namespace]
+	c.persistedMu.RUnlock()
+	return exists && objRV <= persisted, 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(
+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)
+	key := c.checkpointKey(namespace, objectType)
 	if key == "" {
 		return fmt.Errorf("checkpoint key is empty: %s, %s", namespace, objectType)
 	}
@@ -163,10 +204,10 @@ func (c *checkpointer) DeleteCheckpoint(
 	return nil
 }
 
-// getCheckpointKey generates a unique storage key
+// checkpointKey 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 {
+func (*Checkpointer) checkpointKey(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 == "" {