[receiver/k8sobjects] Add logic to properly handle 410 responses

.chloggen/k8sobject-handle-410.yaml

@@ -0,0 +1,27 @@
+# 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. filelogreceiver)
+component: k8sobjectsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds logic to properly handle 410 response codes when watching. This improves the reliability of the receiver.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26098]
+
+# (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:
+
+# 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: []

receiver/k8sobjectsreceiver/receiver.go

@@ -6,6 +6,7 @@ package k8sobjectsreceiver // import "github.com/open-telemetry/opentelemetry-co
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"sync"
 	"time"
 
@@ -14,7 +15,9 @@ import (
 	"go.opentelemetry.io/collector/obsreport"
 	"go.opentelemetry.io/collector/receiver"
 	"go.uber.org/zap"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	apiWatch "k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/tools/cache"
@@ -141,37 +144,69 @@ func (kr *k8sobjectsreceiver) startPull(ctx context.Context, config *K8sObjectsC
 }
 
 func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjectsConfig, resource dynamic.ResourceInterface) {
-
 	stopperChan := make(chan struct{})
 	kr.mu.Lock()
 	kr.stopperChanList = append(kr.stopperChanList, stopperChan)
 	kr.mu.Unlock()
 
-	resourceVersion, err := getResourceVersion(ctx, config, resource)
-	if err != nil {
-		kr.setting.Logger.Error("could not retrieve an initial resourceVersion", zap.String("resource", config.gvr.String()), zap.Error(err))
-		return
-	}
 	watchFunc := func(options metav1.ListOptions) (apiWatch.Interface, error) {
 		options.FieldSelector = config.FieldSelector
 		options.LabelSelector = config.LabelSelector
 		return resource.Watch(ctx, options)
 	}
 
+	cancelCtx, cancel := context.WithCancel(ctx)
+	cfgCopy := config
+	wait.UntilWithContext(cancelCtx, func(newCtx context.Context) {
+		resourceVersion, err := getResourceVersion(newCtx, cfgCopy, resource)
+		if err != nil {
+			kr.setting.Logger.Error("could not retrieve a resourceVersion", zap.String("resource", cfgCopy.gvr.String()), zap.Error(err))
+			cancel()
+			return
+		}
+
+		done := kr.doWatch(newCtx, cfgCopy, resourceVersion, watchFunc, stopperChan)
+		if done {
+			cancel()
+			return
+		}
+
+		// need to restart with a fresh resource version
+		cfgCopy.ResourceVersion = ""
+	}, 0)
+}
+
+// doWatch returns true when watching is done, false when watching should be restarted.
+func (kr *k8sobjectsreceiver) doWatch(ctx context.Context, config *K8sObjectsConfig, resourceVersion string, watchFunc func(options metav1.ListOptions) (apiWatch.Interface, error), stopperChan chan struct{}) bool {
 	watcher, err := watch.NewRetryWatcher(resourceVersion, &cache.ListWatch{WatchFunc: watchFunc})
 	if err != nil {
 		kr.setting.Logger.Error("error in watching object", zap.String("resource", config.gvr.String()), zap.Error(err))
-		return
+		return true
 	}
 
+	defer watcher.Stop()
 	res := watcher.ResultChan()
 	for {
 		select {
 		case data, ok := <-res:
+			if data.Type == apiWatch.Error {
+				errObject := apierrors.FromObject(data.Object)
+				var statusErr *apierrors.StatusError
+				// FromObject guarantees an unwrapped StatusError or UnexpectedObjectError so this type checking is best
+				// nolint:errorlint
+				statusErr, ok = errObject.(*apierrors.StatusError)
+				if ok && statusErr.ErrStatus.Code == http.StatusGone {
+					kr.setting.Logger.Info("received a 410, grabbing new resource version", zap.Any("data", data))
+					// we received a 410 so we need to restart
+					return false
+				}
+			}
+
 			if !ok {
 				kr.setting.Logger.Warn("Watch channel closed unexpectedly", zap.String("resource", config.gvr.String()))
-				return
+				return true
 			}
+
 			logs, err := watchObjectsToLogData(&data, time.Now(), config)
 			if err != nil {
 				kr.setting.Logger.Error("error converting objects to log data", zap.Error(err))
@@ -182,10 +217,9 @@ func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjects
 			}
 		case <-stopperChan:
 			watcher.Stop()
-			return
+			return true
 		}
 	}
-
 }
 
 func getResourceVersion(ctx context.Context, config *K8sObjectsConfig, resource dynamic.ResourceInterface) (string, error) {