[receiver/k8scluster] add support for observing resources for a specific namespace

.chloggen/k8sclusterreceiver-namespaced.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: k8sclusterreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for limiting observed resources to a specific namespace.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [9401]
+
+# (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: This change allows to make use of this receiver with `Roles`/`RoleBindings`, as opposed to giving the collector cluster-wide read access.
+
+# 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/k8sclusterreceiver/README.md

@@ -62,6 +62,7 @@ The following allocatable resource types are available.
   - storage
 - `metrics`: Allows to enable/disable metrics.
 - `resource_attributes`: Allows to enable/disable resource attributes.
+- `namespace`: Allows to observe resources for a particular namespace only. If this option is set to a non-empty string, `Nodes`, `Namespaces` and `ClusterResourceQuotas` will not be observed. 
 
 Example:
 
@@ -273,6 +274,97 @@ subjects:
 EOF
 ```
 
+As an alternative to setting up a `ClusterRole`/`ClusterRoleBinding`, it is also possible to limit the observed resources to a
+particular namespace by setting the `namespace` option of the receiver. This allows the collector to only rely on `Roles`/`RoleBindings`, 
+instead of granting the collector cluster-wide read access to resources.
+Note however, that in this case the following resources will not be observed by the `k8sclusterreceiver`:
+
+- `Nodes`
+- `Namespaces`
+- `ClusterResourceQuotas`
+
+To use this approach, use the commands below to create the required `Role` and `RoleBinding`:
+
+```bash
+<<EOF | kubectl apply -f -
+metadata:
+  name: otelcontribcol
+  labels:
+    app: otelcontribcol
+  namespace: default
+rules:
+  - apiGroups:
+      - ""
+    resources:
+      - events
+      - pods
+      - pods/status
+      - replicationcontrollers
+      - replicationcontrollers/status
+      - services
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - apps
+    resources:
+      - daemonsets
+      - deployments
+      - replicasets
+      - statefulsets
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - extensions
+    resources:
+      - daemonsets
+      - deployments
+      - replicasets
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - batch
+    resources:
+      - jobs
+      - cronjobs
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - autoscaling
+    resources:
+      - horizontalpodautoscalers
+    verbs:
+      - get
+      - list
+      - watch
+EOF
+```
+
+```bash
+<<EOF | kubectl apply -f -
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: otelcontribcol
+  namespace: default
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: otelcontribcol
+subjects:
+  - kind: ServiceAccount
+    name: otelcontribcol
+    namespace: default
+EOF
+```
+
 ### Deployment
 
 Create a [Deployment](https://kubernetes.io/docs/concepts/workloads/controllers/deployment/) to deploy the collector.

receiver/k8sclusterreceiver/config.go

@@ -39,6 +39,11 @@ type Config struct {
 
 	// MetricsBuilderConfig allows customizing scraped metrics/attributes representation.
 	metadata.MetricsBuilderConfig `mapstructure:",squash"`
+
+	// Namespace to fetch resources from. If this is set, certain cluster-wide resources such as Nodes or Namespaces
+	// will not be able to be observed. Setting this option is recommended in environments where due to security restrictions
+	// the collector can not be granted cluster-wide permissions.
+	Namespace string `mapstructure:"namespace"`
 }
 
 func (cfg *Config) Validate() error {
@@ -48,5 +53,6 @@ func (cfg *Config) Validate() error {
 	default:
 		return fmt.Errorf("\"%s\" is not a supported distribution. Must be one of: \"openshift\", \"kubernetes\"", cfg.Distribution)
 	}
+
 	return nil
 }

receiver/k8sclusterreceiver/e2e_test.go

@@ -7,6 +7,7 @@ package k8sclusterreceiver
 
 import (
 	"context"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -25,28 +26,30 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
 )
 
-const expectedFile = "./testdata/e2e/expected.yaml"
+const expectedFileClusterScoped = "./testdata/e2e/cluster-scoped/expected.yaml"
+const expectedFileNamespaceScoped = "./testdata/e2e/namespace-scoped/expected.yaml"
+
+const testObjectsDirClusterScoped = "./testdata/e2e/cluster-scoped/testobjects"
+const testObjectsDirNamespaceScoped = "./testdata/e2e/namespace-scoped/testobjects"
 const testKubeConfig = "/tmp/kube-config-otelcol-e2e-testing"
-const testObjectsDir = "./testdata/e2e/testobjects/"
 
-// TestE2E tests the k8s cluster receiver with a real k8s cluster.
+// TestE2EClusterScoped tests the k8s cluster receiver with a real k8s cluster.
 // The test requires a prebuilt otelcontribcol image uploaded to a kind k8s cluster defined in
 // `/tmp/kube-config-otelcol-e2e-testing`. Run the following command prior to running the test locally:
 //
 //	kind create cluster --kubeconfig=/tmp/kube-config-otelcol-e2e-testing
 //	make docker-otelcontribcol
 //	KUBECONFIG=/tmp/kube-config-otelcol-e2e-testing kind load docker-image otelcontribcol:latest
-func TestE2E(t *testing.T) {
-
+func TestE2EClusterScoped(t *testing.T) {
 	var expected pmetric.Metrics
-	expected, err := golden.ReadMetrics(expectedFile)
+	expected, err := golden.ReadMetrics(expectedFileClusterScoped)
 	require.NoError(t, err)
 
 	k8sClient, err := k8stest.NewK8sClient(testKubeConfig)
 	require.NoError(t, err)
 
 	// k8s test objs
-	testObjs, err := k8stest.CreateObjects(k8sClient, testObjectsDir)
+	testObjs, err := k8stest.CreateObjects(k8sClient, testObjectsDirClusterScoped)
 	require.NoErrorf(t, err, "failed to create objects")
 
 	t.Cleanup(func() {
@@ -58,17 +61,16 @@ func TestE2E(t *testing.T) {
 	defer shutdownSink()
 
 	testID := uuid.NewString()[:8]
-	collectorObjs := k8stest.CreateCollectorObjects(t, k8sClient, testID, "")
+	collectorObjs := k8stest.CreateCollectorObjects(t, k8sClient, testID, filepath.Join(".", "testdata", "e2e", "cluster-scoped", "collector"))
 
 	t.Cleanup(func() {
 		for _, obj := range append(collectorObjs) {
 			require.NoErrorf(t, k8stest.DeleteObject(k8sClient, obj), "failed to delete object %s", obj.GetName())
 		}
 	})
 
-	wantEntries := 10 // Minimal number of metrics to wait for.
+	wantEntries := expected.ResourceMetrics().Len() // Minimal number of metrics to wait for.
 	waitForData(t, wantEntries, metricsConsumer)
-	// golden.WriteMetrics(t, expectedFile, metricsConsumer.AllMetrics()[len(metricsConsumer.AllMetrics())-1])
 
 	replaceWithStar := func(string) string { return "*" }
 	shortenNames := func(value string) string {
@@ -125,6 +127,127 @@ func TestE2E(t *testing.T) {
 			"k8s.job.desired_successful_pods",
 			"k8s.job.failed_pods",
 			"k8s.job.max_parallel_pods",
+			"k8s.hpa.current_replicas",
+			"k8s.job.successful_pods"),
+		pmetrictest.ChangeResourceAttributeValue("container.id", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("container.image.name", containerImageShorten),
+		pmetrictest.ChangeResourceAttributeValue("container.image.tag", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.cronjob.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.daemonset.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.deployment.name", shortenNames),
+		pmetrictest.ChangeResourceAttributeValue("k8s.deployment.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.hpa.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.job.name", shortenNames),
+		pmetrictest.ChangeResourceAttributeValue("k8s.job.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.namespace.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.node.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.pod.name", shortenNames),
+		pmetrictest.ChangeResourceAttributeValue("k8s.pod.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.replicaset.name", shortenNames),
+		pmetrictest.ChangeResourceAttributeValue("k8s.replicaset.uid", replaceWithStar),
+		pmetrictest.ChangeResourceAttributeValue("k8s.statefulset.uid", replaceWithStar),
+		pmetrictest.IgnoreScopeVersion(),
+		pmetrictest.IgnoreResourceMetricsOrder(),
+		pmetrictest.IgnoreMetricsOrder(),
+		pmetrictest.IgnoreScopeMetricsOrder(),
+		pmetrictest.IgnoreMetricDataPointsOrder(),
+	),
+	)
+}
+
+// TestE2ENamespaceScoped tests the k8s cluster receiver with a real k8s cluster.
+// The test requires a prebuilt otelcontribcol image uploaded to a kind k8s cluster defined in
+// `/tmp/kube-config-otelcol-e2e-testing`. Run the following command prior to running the test locally:
+//
+//	kind create cluster --kubeconfig=/tmp/kube-config-otelcol-e2e-testing
+//	make docker-otelcontribcol
+//	KUBECONFIG=/tmp/kube-config-otelcol-e2e-testing kind load docker-image otelcontribcol:latest
+func TestE2ENamespaceScoped(t *testing.T) {
+
+	var expected pmetric.Metrics
+	expected, err := golden.ReadMetrics(expectedFileNamespaceScoped)
+	require.NoError(t, err)
+
+	k8sClient, err := k8stest.NewK8sClient(testKubeConfig)
+	require.NoError(t, err)
+
+	// k8s test objs
+	testObjs, err := k8stest.CreateObjects(k8sClient, testObjectsDirNamespaceScoped)
+	require.NoErrorf(t, err, "failed to create objects")
+
+	t.Cleanup(func() {
+		require.NoErrorf(t, k8stest.DeleteObjects(k8sClient, testObjs), "failed to delete objects")
+	})
+
+	metricsConsumer := new(consumertest.MetricsSink)
+	shutdownSink := startUpSink(t, metricsConsumer)
+	defer shutdownSink()
+
+	testID := uuid.NewString()[:8]
+	collectorObjs := k8stest.CreateCollectorObjects(t, k8sClient, testID, filepath.Join(".", "testdata", "e2e", "namespace-scoped", "collector"))
+
+	t.Cleanup(func() {
+		for _, obj := range append(collectorObjs) {
+			require.NoErrorf(t, k8stest.DeleteObject(k8sClient, obj), "failed to delete object %s", obj.GetName())
+		}
+	})
+
+	wantEntries := expected.ResourceMetrics().Len() // Minimal number of metrics to wait for.
+	waitForData(t, wantEntries, metricsConsumer)
+
+	replaceWithStar := func(string) string { return "*" }
+	shortenNames := func(value string) string {
+		if strings.HasPrefix(value, "coredns") {
+			return "coredns"
+		}
+		if strings.HasPrefix(value, "kindnet") {
+			return "kindnet"
+		}
+		if strings.HasPrefix(value, "kube-apiserver") {
+			return "kube-apiserver"
+		}
+		if strings.HasPrefix(value, "kube-proxy") {
+			return "kube-proxy"
+		}
+		if strings.HasPrefix(value, "kube-scheduler") {
+			return "kube-scheduler"
+		}
+		if strings.HasPrefix(value, "kube-controller-manager") {
+			return "kube-controller-manager"
+		}
+		if strings.HasPrefix(value, "local-path-provisioner") {
+			return "local-path-provisioner"
+		}
+		if strings.HasPrefix(value, "otelcol") {
+			return "otelcol"
+		}
+		if strings.HasPrefix(value, "test-k8scluster-receiver-cronjob") {
+			return "test-k8scluster-receiver-cronjob"
+		}
+		if strings.HasPrefix(value, "test-k8scluster-receiver-job") {
+			return "test-k8scluster-receiver-job"
+		}
+		return value
+	}
+	containerImageShorten := func(value string) string {
+		return value[(strings.LastIndex(value, "/") + 1):]
+	}
+	require.NoError(t, pmetrictest.CompareMetrics(expected, metricsConsumer.AllMetrics()[len(metricsConsumer.AllMetrics())-1],
+		pmetrictest.IgnoreTimestamp(),
+		pmetrictest.IgnoreStartTimestamp(),
+		pmetrictest.IgnoreMetricValues(
+			"k8s.container.cpu_request",
+			"k8s.container.memory_limit",
+			"k8s.container.memory_request",
+			"k8s.container.restarts",
+			"k8s.cronjob.active_jobs",
+			"k8s.deployment.available",
+			"k8s.deployment.desired",
+			"k8s.job.active_pods",
+			"k8s.job.desired_successful_pods",
+			"k8s.job.failed_pods",
+			"k8s.job.max_parallel_pods",
+			"k8s.hpa.current_replicas",
 			"k8s.job.successful_pods"),
 		pmetrictest.ChangeResourceAttributeValue("container.id", replaceWithStar),
 		pmetrictest.ChangeResourceAttributeValue("container.image.name", containerImageShorten),
@@ -167,9 +290,12 @@ func startUpSink(t *testing.T, mc *consumertest.MetricsSink) func() {
 }
 
 func waitForData(t *testing.T, entriesNum int, mc *consumertest.MetricsSink) {
-	timeoutMinutes := 3
+	timeoutMinutes := 6
 	require.Eventuallyf(t, func() bool {
-		return len(mc.AllMetrics()) > entriesNum
+		if len(mc.AllMetrics()) == 0 {
+			return false
+		}
+		return mc.AllMetrics()[len(mc.AllMetrics())-1].ResourceMetrics().Len() == entriesNum
 	}, time.Duration(timeoutMinutes)*time.Minute, 1*time.Second,
 		"failed to receive %d entries,  received %d metrics in %d minutes", entriesNum,
 		len(mc.AllMetrics()), timeoutMinutes)