[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/config.go

@@ -15,6 +15,7 @@ import (
 	"k8s.io/client-go/dynamic"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
+	"go.uber.org/zap"
 )
 
 type mode string
@@ -33,6 +34,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,26 +59,61 @@ type K8sObjectsConfig struct {
 type Config struct {
 	k8sconfig.APIConfig `mapstructure:",squash"`
 
-	Objects []*K8sObjectsConfig `mapstructure:"objects"`
+	Objects   []*K8sObjectsConfig `mapstructure:"objects"`
+	ErrorMode ErrorMode           `mapstructure:"error_mode"`
+	logger    *zap.Logger
 
 	// For mocking purposes only.
 	makeDiscoveryClient func() (discovery.ServerResourcesInterface, error)
 	makeDynamicClient   func() (dynamic.Interface, error)
 }
 
 func (c *Config) Validate() error {
+	zapCfg := zap.NewProductionConfig()
+	logger, err := zapCfg.Build()
+	if err != nil {
+		return fmt.Errorf("failed to create logger: %w", err)
+	}
+	c.logger = logger
+
+	if c.ErrorMode == "" {
+		c.ErrorMode = PropagateError
+	}
+
+	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)
+	}
+
 	validObjects, err := c.getValidObjects()
 	if err != nil {
-		return err
+		if c.ErrorMode == PropagateError {
+			return err
+		}
+		if c.ErrorMode == IgnoreError {
+			c.logger.Warn("Error getting valid objects", zap.Error(err))
+		}
 	}
+
+	var validConfigs []*K8sObjectsConfig
+
 	for _, object := range c.Objects {
 		gvrs, ok := validObjects[object.Name]
 		if !ok {
-			availableResource := make([]string, len(validObjects))
-			for k := range validObjects {
-				availableResource = append(availableResource, k)
+			if c.ErrorMode == PropagateError {
+				return fmt.Errorf("resource not found: %s", object.Name)
+			}
+			if c.ErrorMode == IgnoreError {
+				availableResource := make([]string, 0, len(validObjects))
+				for k := range validObjects {
+					availableResource = append(availableResource, k)
+				}
+				c.logger.Warn("Resource not found",
+					zap.String("resource", object.Name),
+					zap.Strings("available_resources", availableResource))
 			}
-			return fmt.Errorf("resource %v not found. Valid resources are: %v", object.Name, availableResource)
+			continue
 		}
 
 		gvr := gvrs[0]
@@ -95,7 +139,21 @@ func (c *Config) Validate() error {
 		}
 
 		object.gvr = gvr
+		validConfigs = append(validConfigs, object)
 	}
+
+	c.Objects = validConfigs
+
+	if len(validConfigs) == 0 {
+		msg := "no valid Kubernetes objects found to watch"
+		if c.ErrorMode == PropagateError {
+			return errors.New(msg)
+		}
+		if c.ErrorMode == IgnoreError {
+			c.logger.Warn(msg)
+		}
+	}
+
 	return nil
 }
 

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

@@ -161,6 +161,13 @@ func TestLoadConfig(t *testing.T) {
 			assert.NoError(t, xconfmap.Validate(cfg))
 			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)
+			}
 		})
 	}
 }

receiver/k8sobjectsreceiver/receiver.go

@@ -36,6 +36,7 @@ type k8sobjectsreceiver struct {
 	obsrecv         *receiverhelper.ObsReport
 	mu              sync.Mutex
 	cancel          context.CancelFunc
+	logger          *zap.Logger
 }
 
 func newReceiver(params receiver.Settings, config *Config, consumer consumer.Logs) (receiver.Logs, error) {
@@ -50,6 +51,10 @@ func newReceiver(params receiver.Settings, config *Config, consumer consumer.Log
 		return nil, err
 	}
 
+	if err := config.Validate(); err != nil {
+		return nil, err
+	}
+
 	for _, object := range config.Objects {
 		object.exclude = make(map[apiWatch.EventType]bool)
 		for _, item := range object.ExcludeWatchType {
@@ -63,6 +68,7 @@ func newReceiver(params receiver.Settings, config *Config, consumer consumer.Log
 		config:   config,
 		obsrecv:  obsrecv,
 		mu:       sync.Mutex{},
+		logger:   params.Logger,
 	}, nil
 }
 
@@ -72,7 +78,7 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, _ component.Host) error
 		return err
 	}
 	kr.client = client
-	kr.setting.Logger.Info("Object Receiver started")
+	kr.logger.Info("Object Receiver started")
 
 	cctx, cancel := context.WithCancel(ctx)
 	kr.cancel = cancel
@@ -84,7 +90,7 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, _ component.Host) error
 }
 
 func (kr *k8sobjectsreceiver) Shutdown(context.Context) error {
-	kr.setting.Logger.Info("Object Receiver stopped")
+	kr.logger.Info("Object Receiver stopped")
 	if kr.cancel != nil {
 		kr.cancel()
 	}
@@ -99,7 +105,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.logger.Info("Started collecting",
+		zap.Any("gvr", object.gvr),
+		zap.Any("mode", object.Mode),
+		zap.Any("namespaces", object.Namespaces))
 
 	switch object.Mode {
 	case PullMode:
@@ -144,7 +153,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.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)
@@ -175,7 +186,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.logger.Error("could not retrieve a resourceVersion",
+				zap.String("resource", cfgCopy.gvr.String()),
+				zap.Error(err))
 			cancel()
 			return
 		}
@@ -195,7 +208,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.logger.Error("error in watching object",
+			zap.String("resource", config.gvr.String()),
+			zap.Error(err))
 		return true
 	}
 
@@ -208,25 +223,28 @@ 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.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.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.logger.Debug("dropping excluded data",
+					zap.String("type", string(data.Type)))
 				continue
 			}
 
 			logs, err := watchObjectsToLogData(&data, time.Now(), config)
 			if err != nil {
-				kr.setting.Logger.Error("error converting objects to log data", zap.Error(err))
+				kr.logger.Error("error converting objects to log data", zap.Error(err))
 			} else {
 				obsCtx := kr.obsrecv.StartLogsOp(ctx)
 				err := kr.consumer.ConsumeLogs(obsCtx, logs)