[Backport 7.81.x] [CONTINT] Retry event checkpoint read (#53871)

Backport 9c3331f from #53752.

 ___

### What does this PR do?
Adds retry-with-backoff when reading the `kubernetes_apiserver` event-collection checkpoint from its ConfigMap. 

### Motivation
A transient error reading the checkpoint was being treated the same as "checkpoint doesn't exist," which resets the watermark to 0 and causing the initial watch to potentially redeliver all events. For example:
```
2026-07-16 14:38:39 UTC | CLUSTER | ERROR | (/home/datadog/go/src/github.com/DataDog/datadog-agent-checkpoint-baseline/pkg/util/kubernetes/apiserver/apiserver.go:509 in getOrCreateConfigMap) | Could not get the ConfigMap baseline-token: configmaps "baseline-token" not found, trying to create it.
2026-07-16 14:38:39 UTC | CLUSTER | INFO | (/home/datadog/go/src/github.com/DataDog/datadog-agent-checkpoint-baseline/pkg/util/kubernetes/apiserver/apiserver.go:519 in getOrCreateConfigMap) | Created the ConfigMap baseline-token
```

### Describe how you validated your changes
Reproduced the behavior on a kind cluster running the DCA via the Operator by denying the DCA GET requests to the checkpoint ConfigMap temporarily to hit the retry. 


### Additional Notes

Co-authored-by: kacper.murzyn <kacper.murzyn@datadoghq.com>

Author: dd-octo-sts[bot]

pkg/collector/corechecks/cluster/kubernetesapiserver/kubernetes_apiserver.go

@@ -373,15 +373,37 @@ func (k *KubeASCheck) startEventCollection() error {
 	k.mu.Unlock()
 
 	// If the checkpoint is present, seed the watermark from it so the initial list only forwards newer events.
-	if resVer, _, err := k.ac.GetTokenFromConfigmap(eventTokenKey); err != nil {
-		log.Warnf("Could not read persisted event checkpoint, starting fresh: %s", err)
+	resVer, err := k.readEventCheckpointWithRetry()
+	if err != nil {
+		log.Warnf("Could not read persisted event checkpoint after retries, starting fresh: %s", err)
 	} else {
 		ec.SetCheckpoint(resVer)
 	}
 
 	return ec.Start(stopCh)
 }
 
+// readEventCheckpointWithRetry reads the persisted event checkpoint, retrying with backoff
+func (k *KubeASCheck) readEventCheckpointWithRetry() (string, error) {
+	const maxAttempts = 5
+
+	delay := time.Second
+	var lastErr error
+	for attempt := 1; attempt <= maxAttempts; attempt++ {
+		resVer, _, err := k.ac.GetTokenFromConfigmap(eventTokenKey)
+		if err == nil {
+			return resVer, nil
+		}
+		lastErr = err
+		if attempt < maxAttempts {
+			log.Warnf("Could not read persisted event checkpoint (attempt %d/%d): %s", attempt, maxAttempts, err)
+			time.Sleep(delay)
+			delay *= 2
+		}
+	}
+	return "", lastErr
+}
+
 // stopEventCollection stops the running EventCollector by closing its stop
 // channel. It is idempotent.
 func (k *KubeASCheck) stopEventCollection() {

pkg/util/kubernetes/apiserver/BUILD.bazel

@@ -45,6 +45,7 @@ go_library(
         "@io_k8s_api//discovery/v1:discovery",
         "@io_k8s_apiextensions_apiserver//pkg/client/clientset/clientset",
         "@io_k8s_apiextensions_apiserver//pkg/client/informers/externalversions",
+        "@io_k8s_apimachinery//pkg/api/errors",
         "@io_k8s_apimachinery//pkg/api/meta",
         "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
         "@io_k8s_apimachinery//pkg/apis/meta/v1/unstructured",

pkg/util/kubernetes/apiserver/apiserver.go

@@ -21,6 +21,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -485,6 +486,9 @@ func (c *APIClient) ComponentStatuses() (*v1.ComponentStatusList, error) {
 func (c *APIClient) getOrCreateConfigMap(name, namespace string) (cmEvent *v1.ConfigMap, err error) {
 	cmEvent, err = c.Cl.CoreV1().ConfigMaps(namespace).Get(context.TODO(), name, metav1.GetOptions{})
 	if err != nil {
+		if !apierrors.IsNotFound(err) {
+			return nil, fmt.Errorf("could not get the ConfigMap %s: %w", name, err)
+		}
 		log.Errorf("Could not get the ConfigMap %s: %s, trying to create it.", name, err.Error())
 		cmEvent, err = c.Cl.CoreV1().ConfigMaps(namespace).Create(context.TODO(), &v1.ConfigMap{
 			ObjectMeta: metav1.ObjectMeta{