[receiver/k8sobjectsreceiver] Handle missing objects via error_mode

.chloggen/k8sobjectsreceiver_errormode.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: bug_fix
+
+# 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: Introduces `error_mode`, so users can choose between propagating, ignoring, or silencing missing objects.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38803]
+
+# (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: [user]

receiver/k8sobjectsreceiver/README.md

@@ -42,6 +42,10 @@ Brief description of configuration properties:
 the K8s API server. This can be one of `none` (for no auth), `serviceAccount`
 (to use the standard service account token provided to the agent pod), or
 `kubeConfig` to use credentials from `~/.kube/config`.
+- `error_mode` (default = `propagate`): Determines how to handle errors when the receiver is unable to pull or watch objects due to missing resources. This can be one of `propagate`, `ignore`, or `silent`.
+  - `propagate` will propagate the error to the collector as an Error.
+  - `ignore` will log and ignore the error and continue.
+  - `silent` will ignore the error and continue without logging.
 - `name`: Name of the resource object to collect
 - `mode`: define in which way it collects this type of object, either "pull" or "watch".
   - `pull` mode will read all objects of this type use the list API at an interval.

receiver/k8sobjectsreceiver/config.go

@@ -33,6 +33,14 @@ var modeMap = map[mode]bool{
 	WatchMode: true,
 }
 
+type ErrorMode string
+
+const (
+	PropagateError ErrorMode = "propagate"
+	IgnoreError    ErrorMode = "ignore"
+	SilentError    ErrorMode = "silent"
+)
+
 type K8sObjectsConfig struct {
 	Name             string               `mapstructure:"name"`
 	Group            string               `mapstructure:"group"`
@@ -50,14 +58,21 @@ type K8sObjectsConfig struct {
 type Config struct {
 	k8sconfig.APIConfig `mapstructure:",squash"`
 
-	Objects []*K8sObjectsConfig `mapstructure:"objects"`
+	Objects   []*K8sObjectsConfig `mapstructure:"objects"`
+	ErrorMode ErrorMode           `mapstructure:"error_mode"`
 
 	// For mocking purposes only.
 	makeDiscoveryClient func() (discovery.ServerResourcesInterface, error)
 	makeDynamicClient   func() (dynamic.Interface, error)
 }
 
 func (c *Config) Validate() error {
+	switch c.ErrorMode {
+	case PropagateError, IgnoreError, SilentError:
+	default:
+		return fmt.Errorf("invalid error_mode %q: must be one of 'propagate', 'ignore', or 'silent'", c.ErrorMode)
+	}
+
 	for _, object := range c.Objects {
 		if object.Mode == "" {
 			object.Mode = defaultMode

receiver/k8sobjectsreceiver/config.yaml

@@ -0,0 +1,5 @@
+receivers:
+  k8sobjects:
+    error_mode: ignore  # Can be "propagate", "ignore", or "silent"
+    objects:
+      # ... other configuration ...

receiver/k8sobjectsreceiver/config_test.go

@@ -112,6 +112,13 @@ func TestLoadConfig(t *testing.T) {
 
 			assert.Equal(t, tt.expected.AuthType, cfg.AuthType)
 			assert.Equal(t, tt.expected.Objects, cfg.Objects)
+
+			err = cfg.Validate()
+			if tt.expected == nil {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
 		})
 	}
 }
@@ -125,6 +132,7 @@ func TestValidate(t *testing.T) {
 		{
 			desc: "invalid mode",
 			cfg: &Config{
+				ErrorMode: PropagateError,
 				Objects: []*K8sObjectsConfig{
 					{
 						Name: "pods",
@@ -137,6 +145,7 @@ func TestValidate(t *testing.T) {
 		{
 			desc: "exclude watch type with pull mode",
 			cfg: &Config{
+				ErrorMode: PropagateError,
 				Objects: []*K8sObjectsConfig{
 					{
 						Name: "pods",
@@ -152,6 +161,7 @@ func TestValidate(t *testing.T) {
 		{
 			desc: "default mode is set",
 			cfg: &Config{
+				ErrorMode: PropagateError,
 				Objects: []*K8sObjectsConfig{
 					{
 						Name: "pods",
@@ -162,6 +172,7 @@ func TestValidate(t *testing.T) {
 		{
 			desc: "default interval for pull mode",
 			cfg: &Config{
+				ErrorMode: PropagateError,
 				Objects: []*K8sObjectsConfig{
 					{
 						Name: "pods",

receiver/k8sobjectsreceiver/factory.go

@@ -30,6 +30,7 @@ func createDefaultConfig() component.Config {
 		APIConfig: k8sconfig.APIConfig{
 			AuthType: k8sconfig.AuthTypeServiceAccount,
 		},
+		ErrorMode: PropagateError,
 	}
 }
 

receiver/k8sobjectsreceiver/factory_test.go

@@ -27,6 +27,7 @@ func TestDefaultConfig(t *testing.T) {
 		APIConfig: k8sconfig.APIConfig{
 			AuthType: k8sconfig.AuthTypeServiceAccount,
 		},
+		ErrorMode: PropagateError,
 	}, rCfg)
 }
 

receiver/k8sobjectsreceiver/receiver.go

@@ -59,6 +59,10 @@ func newReceiver(params receiver.Settings, config *Config, consumer consumer.Log
 		for _, item := range objects[i].ExcludeWatchType {
 			objects[i].exclude[item] = true
 		}
+		// Set default interval if in PullMode and interval is 0
+		if objects[i].Mode == PullMode && objects[i].Interval == 0 {
+			objects[i].Interval = defaultPullInterval
+		}
 	}
 
 	return &k8sobjectsreceiver{
@@ -84,14 +88,19 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, _ component.Host) error
 		return err
 	}
 
+	var validConfigs []*K8sObjectsConfig
 	for _, object := range kr.objects {
 		gvrs, ok := validObjects[object.Name]
 		if !ok {
-			availableResource := make([]string, len(validObjects))
+			availableResource := make([]string, 0, len(validObjects))
 			for k := range validObjects {
 				availableResource = append(availableResource, k)
 			}
-			return fmt.Errorf("resource %v not found. Valid resources are: %v", object.Name, availableResource)
+			err := fmt.Errorf("resource not found: %s. Available resources in cluster: %v", object.Name, availableResource)
+			if handlerErr := kr.handleError(err, ""); handlerErr != nil {
+				return handlerErr
+			}
+			continue
 		}
 
 		gvr := gvrs[0]
@@ -101,15 +110,21 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, _ component.Host) error
 				break
 			}
 		}
+
 		object.gvr = gvr
+		validConfigs = append(validConfigs, object)
 	}
 
-	kr.setting.Logger.Info("Object Receiver started")
+	if len(validConfigs) == 0 {
+		err := errors.New("no valid Kubernetes objects found to watch")
+		return err
+	}
 
+	kr.setting.Logger.Info("Object Receiver started")
 	cctx, cancel := context.WithCancel(ctx)
 	kr.cancel = cancel
 
-	for _, object := range kr.objects {
+	for _, object := range validConfigs {
 		kr.start(cctx, object)
 	}
 	return nil
@@ -131,7 +146,10 @@ func (kr *k8sobjectsreceiver) Shutdown(context.Context) error {
 
 func (kr *k8sobjectsreceiver) start(ctx context.Context, object *K8sObjectsConfig) {
 	resource := kr.client.Resource(*object.gvr)
-	kr.setting.Logger.Info("Started collecting", zap.Any("gvr", object.gvr), zap.Any("mode", object.Mode), zap.Any("namespaces", object.Namespaces))
+	kr.setting.Logger.Info("Started collecting",
+		zap.Any("gvr", object.gvr),
+		zap.Any("mode", object.Mode),
+		zap.Any("namespaces", object.Namespaces))
 
 	switch object.Mode {
 	case PullMode:
@@ -176,7 +194,9 @@ func (kr *k8sobjectsreceiver) startPull(ctx context.Context, config *K8sObjectsC
 		case <-ticker.C:
 			objects, err := resource.List(ctx, listOption)
 			if err != nil {
-				kr.setting.Logger.Error("error in pulling object", zap.String("resource", config.gvr.String()), zap.Error(err))
+				kr.setting.Logger.Error("error in pulling object",
+					zap.String("resource", config.gvr.String()),
+					zap.Error(err))
 			} else if len(objects.Items) > 0 {
 				logs := pullObjectsToLogData(objects, time.Now(), config)
 				obsCtx := kr.obsrecv.StartLogsOp(ctx)
@@ -207,7 +227,9 @@ func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjects
 	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))
+			kr.setting.Logger.Error("could not retrieve a resourceVersion",
+				zap.String("resource", cfgCopy.gvr.String()),
+				zap.Error(err))
 			cancel()
 			return
 		}
@@ -227,7 +249,9 @@ func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjects
 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))
+		kr.setting.Logger.Error("error in watching object",
+			zap.String("resource", config.gvr.String()),
+			zap.Error(err))
 		return true
 	}
 
@@ -240,19 +264,22 @@ func (kr *k8sobjectsreceiver) doWatch(ctx context.Context, config *K8sObjectsCon
 				errObject := apierrors.FromObject(data.Object)
 				//nolint:errorlint
 				if errObject.(*apierrors.StatusError).ErrStatus.Code == http.StatusGone {
-					kr.setting.Logger.Info("received a 410, grabbing new resource version", zap.Any("data", data))
+					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()))
+				kr.setting.Logger.Warn("Watch channel closed unexpectedly",
+					zap.String("resource", config.gvr.String()))
 				return true
 			}
 
 			if config.exclude[data.Type] {
-				kr.setting.Logger.Debug("dropping excluded data", zap.String("type", string(data.Type)))
+				kr.setting.Logger.Debug("dropping excluded data",
+					zap.String("type", string(data.Type)))
 				continue
 			}
 
@@ -321,3 +348,30 @@ func newTicker(ctx context.Context, repeat time.Duration) *time.Ticker {
 	ticker.C = nc
 	return ticker
 }
+
+// handleError handles errors according to the configured error mode
+func (kr *k8sobjectsreceiver) handleError(err error, msg string) error {
+	if err == nil {
+		return nil
+	}
+
+	switch kr.config.ErrorMode {
+	case PropagateError:
+		if msg != "" {
+			return fmt.Errorf("%s: %w", msg, err)
+		}
+		return err
+	case IgnoreError:
+		if msg != "" {
+			kr.setting.Logger.Info(msg, zap.Error(err))
+		} else {
+			kr.setting.Logger.Info(err.Error())
+		}
+		return nil
+	case SilentError:
+		return nil
+	default:
+		// This shouldn't happen as we validate ErrorMode during config validation
+		return fmt.Errorf("invalid error_mode %q: %w", kr.config.ErrorMode, err)
+	}
+}