[receiver/k8sobjectsreceiver] Refactor Validate to only validate static config (#39469)

#### Description

Refactor Validate() to only validate static config without external
calls to Kubernetes API server as discussed in
#38851 (comment)

Once this is merged, the PR linked above should introduce `error_mode`
implemented in `start` too.

#### Link to tracking issue
It is related to the issue and follow-up PR mentioned above, but it
doesn't fix the issue.

#### Testing

I've refactored the tests in config_test.go, so we are not testing
object discovery and added new tests for additional failure modes.
I've build a dev images, and validated that functionality is the same as
before the refactor.

#### Documentation

This is a "breaking change" in a sense we are changing what we are
validating in Validate. The new validation logic will not fail when
objects are missing. The receiver will fail the same way in this case,
but in `start`.

Author: krisztianfekete

.chloggen/k8sobjectsrecevier_validate_refactor.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: breaking
+
+# 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: Check for K8s API objects existence on receiver startup and not during config validation.
+
+# 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

@@ -58,28 +58,7 @@ type Config struct {
 }
 
 func (c *Config) Validate() error {
-	validObjects, err := c.getValidObjects()
-	if err != nil {
-		return err
-	}
 	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)
-			}
-			return fmt.Errorf("resource %v not found. Valid resources are: %v", object.Name, availableResource)
-		}
-
-		gvr := gvrs[0]
-		for i := range gvrs {
-			if gvrs[i].Group == object.Group {
-				gvr = gvrs[i]
-				break
-			}
-		}
-
 		if object.Mode == "" {
 			object.Mode = defaultMode
 		} else if _, ok := modeMap[object.Mode]; !ok {
@@ -93,8 +72,6 @@ func (c *Config) Validate() error {
 		if object.Mode == PullMode && len(object.ExcludeWatchType) != 0 {
 			return errors.New("the Exclude config can only be used with watch mode")
 		}
-
-		object.gvr = gvr
 	}
 	return nil
 }

receiver/k8sobjectsreceiver/config_test.go

@@ -12,8 +12,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
-	"go.opentelemetry.io/collector/confmap/xconfmap"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	apiWatch "k8s.io/apimachinery/pkg/watch"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
@@ -40,29 +38,17 @@ func TestLoadConfig(t *testing.T) {
 						Interval:      time.Hour,
 						FieldSelector: "status.phase=Running",
 						LabelSelector: "environment in (production),tier in (frontend)",
-						gvr: &schema.GroupVersionResource{
-							Group:    "",
-							Version:  "v1",
-							Resource: "pods",
-						},
 					},
 					{
-						Name:            "events",
-						Mode:            WatchMode,
-						Namespaces:      []string{"default"},
-						Group:           "events.k8s.io",
-						ResourceVersion: "",
+						Name:       "events",
+						Mode:       WatchMode,
+						Namespaces: []string{"default"},
+						Group:      "events.k8s.io",
 						ExcludeWatchType: []apiWatch.EventType{
 							apiWatch.Deleted,
 						},
-						gvr: &schema.GroupVersionResource{
-							Group:    "events.k8s.io",
-							Version:  "v1",
-							Resource: "events",
-						},
 					},
 				},
-				makeDiscoveryClient: getMockDiscoveryClient,
 			},
 		},
 		{
@@ -77,24 +63,13 @@ func TestLoadConfig(t *testing.T) {
 						Mode:            PullMode,
 						ResourceVersion: "1",
 						Interval:        time.Hour,
-						gvr: &schema.GroupVersionResource{
-							Group:    "",
-							Version:  "v1",
-							Resource: "pods",
-						},
 					},
 					{
 						Name:     "events",
 						Mode:     PullMode,
 						Interval: time.Hour,
-						gvr: &schema.GroupVersionResource{
-							Group:    "",
-							Version:  "v1",
-							Resource: "events",
-						},
 					},
 				},
-				makeDiscoveryClient: getMockDiscoveryClient,
 			},
 		},
 		{
@@ -110,34 +85,17 @@ func TestLoadConfig(t *testing.T) {
 						Namespaces:      []string{"default"},
 						Group:           "events.k8s.io",
 						ResourceVersion: "",
-						gvr: &schema.GroupVersionResource{
-							Group:    "events.k8s.io",
-							Version:  "v1",
-							Resource: "events",
-						},
 					},
 					{
 						Name:            "events",
 						Mode:            WatchMode,
 						Namespaces:      []string{"default"},
 						Group:           "events.k8s.io",
 						ResourceVersion: "2",
-						gvr: &schema.GroupVersionResource{
-							Group:    "events.k8s.io",
-							Version:  "v1",
-							Resource: "events",
-						},
 					},
 				},
-				makeDiscoveryClient: getMockDiscoveryClient,
 			},
 		},
-		{
-			id: component.NewIDWithName(metadata.Type, "invalid_resource"),
-		},
-		{
-			id: component.NewIDWithName(metadata.Type, "exclude_deleted_with_pull"),
-		},
 	}
 
 	for _, tt := range tests {
@@ -147,58 +105,81 @@ func TestLoadConfig(t *testing.T) {
 
 			factory := NewFactory()
 			cfg := factory.CreateDefaultConfig().(*Config)
-			cfg.makeDiscoveryClient = getMockDiscoveryClient
 
 			sub, err := cm.Sub(tt.id.String())
 			require.NoError(t, err)
 			require.NoError(t, sub.Unmarshal(cfg))
 
-			if tt.expected == nil {
-				err = xconfmap.Validate(cfg)
-				assert.Error(t, err)
-				return
-			}
-			assert.NoError(t, xconfmap.Validate(cfg))
 			assert.Equal(t, tt.expected.AuthType, cfg.AuthType)
 			assert.Equal(t, tt.expected.Objects, cfg.Objects)
 		})
 	}
 }
 
-func TestValidateResourceConflict(t *testing.T) {
-	mockClient := newMockDynamicClient()
-	rCfg := createDefaultConfig().(*Config)
-	rCfg.makeDynamicClient = mockClient.getMockDynamicClient
-	rCfg.makeDiscoveryClient = getMockDiscoveryClient
-
-	// Validate it should choose first gvr if group is not specified
-	rCfg.Objects = []*K8sObjectsConfig{
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		desc        string
+		cfg         *Config
+		expectedErr string
+	}{
 		{
-			Name: "myresources",
-			Mode: PullMode,
+			desc: "invalid mode",
+			cfg: &Config{
+				Objects: []*K8sObjectsConfig{
+					{
+						Name: "pods",
+						Mode: "invalid_mode",
+					},
+				},
+			},
+			expectedErr: "invalid mode: invalid_mode",
 		},
-	}
-
-	err := rCfg.Validate()
-	require.NoError(t, err)
-	assert.Equal(t, "group1", rCfg.Objects[0].gvr.Group)
-
-	// Validate it should choose gvr for specified group
-	rCfg.Objects = []*K8sObjectsConfig{
 		{
-			Name:  "myresources",
-			Mode:  PullMode,
-			Group: "group2",
+			desc: "exclude watch type with pull mode",
+			cfg: &Config{
+				Objects: []*K8sObjectsConfig{
+					{
+						Name: "pods",
+						Mode: PullMode,
+						ExcludeWatchType: []apiWatch.EventType{
+							apiWatch.Deleted,
+						},
+					},
+				},
+			},
+			expectedErr: "the Exclude config can only be used with watch mode",
+		},
+		{
+			desc: "default mode is set",
+			cfg: &Config{
+				Objects: []*K8sObjectsConfig{
+					{
+						Name: "pods",
+					},
+				},
+			},
+		},
+		{
+			desc: "default interval for pull mode",
+			cfg: &Config{
+				Objects: []*K8sObjectsConfig{
+					{
+						Name: "pods",
+						Mode: PullMode,
+					},
+				},
+			},
 		},
 	}
 
-	err = rCfg.Validate()
-	require.NoError(t, err)
-	assert.Equal(t, "group2", rCfg.Objects[0].gvr.Group)
-}
-
-func TestClientRequired(t *testing.T) {
-	rCfg := createDefaultConfig().(*Config)
-	err := rCfg.Validate()
-	require.Error(t, err)
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			err := tt.cfg.Validate()
+			if tt.expectedErr != "" {
+				assert.EqualError(t, err, tt.expectedErr)
+				return
+			}
+			assert.NoError(t, err)
+		})
+	}
 }

receiver/k8sobjectsreceiver/go.mod

@@ -12,7 +12,6 @@ require (
 	go.opentelemetry.io/collector/component v1.30.0
 	go.opentelemetry.io/collector/component/componenttest v0.124.0
 	go.opentelemetry.io/collector/confmap v1.30.0
-	go.opentelemetry.io/collector/confmap/xconfmap v0.124.0
 	go.opentelemetry.io/collector/consumer v1.30.0
 	go.opentelemetry.io/collector/consumer/consumertest v0.124.0
 	go.opentelemetry.io/collector/pdata v1.30.0

receiver/k8sobjectsreceiver/receiver.go

@@ -30,6 +30,7 @@ import (
 type k8sobjectsreceiver struct {
 	setting         receiver.Settings
 	config          *Config
+	objects         []*K8sObjectsConfig
 	stopperChanList []chan struct{}
 	client          dynamic.Interface
 	consumer        consumer.Logs
@@ -50,17 +51,21 @@ func newReceiver(params receiver.Settings, config *Config, consumer consumer.Log
 		return nil, err
 	}
 
-	for _, object := range config.Objects {
-		object.exclude = make(map[apiWatch.EventType]bool)
-		for _, item := range object.ExcludeWatchType {
-			object.exclude[item] = true
+	objects := make([]*K8sObjectsConfig, len(config.Objects))
+	for i, obj := range config.Objects {
+		copied := *obj // Copy the object
+		objects[i] = &copied
+		objects[i].exclude = make(map[apiWatch.EventType]bool)
+		for _, item := range objects[i].ExcludeWatchType {
+			objects[i].exclude[item] = true
 		}
 	}
 
 	return &k8sobjectsreceiver{
 		setting:  params,
-		consumer: consumer,
 		config:   config,
+		objects:  objects,
+		consumer: consumer,
 		obsrecv:  obsrecv,
 		mu:       sync.Mutex{},
 	}, nil
@@ -72,12 +77,39 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, _ component.Host) error
 		return err
 	}
 	kr.client = client
+
+	// Validate objects against K8s API
+	validObjects, err := kr.config.getValidObjects()
+	if err != nil {
+		return err
+	}
+
+	for _, object := range kr.objects {
+		gvrs, ok := validObjects[object.Name]
+		if !ok {
+			availableResource := make([]string, len(validObjects))
+			for k := range validObjects {
+				availableResource = append(availableResource, k)
+			}
+			return fmt.Errorf("resource %v not found. Valid resources are: %v", object.Name, availableResource)
+		}
+
+		gvr := gvrs[0]
+		for i := range gvrs {
+			if gvrs[i].Group == object.Group {
+				gvr = gvrs[i]
+				break
+			}
+		}
+		object.gvr = gvr
+	}
+
 	kr.setting.Logger.Info("Object Receiver started")
 
 	cctx, cancel := context.WithCancel(ctx)
 	kr.cancel = cancel
 
-	for _, object := range kr.config.Objects {
+	for _, object := range kr.objects {
 		kr.start(cctx, object)
 	}
 	return nil

receiver/k8sobjectsreceiver/receiver_test.go

@@ -21,8 +21,11 @@ import (
 func TestNewReceiver(t *testing.T) {
 	t.Parallel()
 
+	mockClient := newMockDynamicClient()
 	rCfg := createDefaultConfig().(*Config)
-	rCfg.makeDynamicClient = newMockDynamicClient().getMockDynamicClient
+	rCfg.makeDynamicClient = mockClient.getMockDynamicClient
+	rCfg.makeDiscoveryClient = getMockDiscoveryClient
+
 	r, err := newReceiver(
 		receivertest.NewNopSettings(metadata.Type),
 		rCfg,