Inject K8S_NODE_NAME environment variable when using the kubeletstats receiver

Signed-off-by: Israel Blancas <[email protected]>

Author: Israel Blancas

.chloggen/2779-kubeletstatsreiver-inject-en-vars.yaml

@@ -0,0 +1,16 @@
+# 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. collector, target allocator, auto-instrumentation, opamp, github action)
+component: collector
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Inject environment K8S_NODE_NAME environment variable for the Kubelet Stats Receiver.
+
+# One or more tracking issues related to the change
+issues: [2779]
+
+# (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:

apis/v1beta1/config.go

@@ -226,6 +226,42 @@ func (c *Config) getPortsForComponentKinds(logger logr.Logger, componentKinds ..
 	return ports, nil
 }
 
+// getEnvironmentVariablesForComponentKinds gets the environment variables for the given ComponentKind(s).
+func (c *Config) getEnvironmentVariablesForComponentKinds(logger logr.Logger, componentKinds ...ComponentKind) ([]corev1.EnvVar, error) {
+	var envVars []corev1.EnvVar
+	enabledComponents := c.GetEnabledComponents()
+	for _, componentKind := range componentKinds {
+		var retriever components.ParserRetriever
+		var cfg AnyConfig
+
+		switch componentKind {
+		case KindReceiver:
+			retriever = receivers.ReceiverFor
+			cfg = c.Receivers
+		case KindExporter:
+			continue
+		case KindProcessor:
+			continue
+		case KindExtension:
+			continue
+		}
+		for componentName := range enabledComponents[componentKind] {
+			parser := retriever(componentName)
+			if parsedEnvVars, err := parser.GetEnvironmentVariables(logger, cfg.Object[componentName]); err != nil {
+				return nil, err
+			} else {
+				envVars = append(envVars, parsedEnvVars...)
+			}
+		}
+	}
+
+	sort.Slice(envVars, func(i, j int) bool {
+		return envVars[i].Name < envVars[j].Name
+	})
+
+	return envVars, nil
+}
+
 // applyDefaultForComponentKinds applies defaults to the endpoints for the given ComponentKind(s).
 func (c *Config) applyDefaultForComponentKinds(logger logr.Logger, componentKinds ...ComponentKind) error {
 	if err := c.Service.ApplyDefaults(); err != nil {
@@ -286,6 +322,10 @@ func (c *Config) GetAllPorts(logger logr.Logger) ([]corev1.ServicePort, error) {
 	return c.getPortsForComponentKinds(logger, KindReceiver, KindExporter)
 }
 
+func (c *Config) GetEnvironmentVariables(logger logr.Logger) ([]corev1.EnvVar, error) {
+	return c.getEnvironmentVariablesForComponentKinds(logger, KindReceiver)
+}
+
 func (c *Config) GetAllRbacRules(logger logr.Logger) ([]rbacv1.PolicyRule, error) {
 	return c.getRbacRulesForComponentKinds(logger, KindReceiver, KindExporter, KindProcessor)
 }

internal/components/builder.go

@@ -38,6 +38,7 @@ type Settings[ComponentConfigType any] struct {
 	livenessGen     ProbeGenerator[ComponentConfigType]
 	readinessGen    ProbeGenerator[ComponentConfigType]
 	defaultsApplier Defaulter[ComponentConfigType]
+	envVarGen       EnvVarGenerator[ComponentConfigType]
 }
 
 func NewEmptySettings[ComponentConfigType any]() *Settings[ComponentConfigType] {
@@ -124,7 +125,11 @@ func (b Builder[ComponentConfigType]) WithReadinessGen(readinessGen ProbeGenerat
 		o.readinessGen = readinessGen
 	})
 }
-
+func (b Builder[ComponentConfigType]) WithEnvVarGen(envVarGen EnvVarGenerator[ComponentConfigType]) Builder[ComponentConfigType] {
+	return append(b, func(o *Settings[ComponentConfigType]) {
+		o.envVarGen = envVarGen
+	})
+}
 func (b Builder[ComponentConfigType]) WithDefaultsApplier(defaultsApplier Defaulter[ComponentConfigType]) Builder[ComponentConfigType] {
 	return append(b, func(o *Settings[ComponentConfigType]) {
 		o.defaultsApplier = defaultsApplier

internal/components/component.go

@@ -49,6 +49,10 @@ type RBACRuleGenerator[ComponentConfigType any] func(logger logr.Logger, config
 // It's expected that type Config is the configuration used by a parser.
 type ProbeGenerator[ComponentConfigType any] func(logger logr.Logger, config ComponentConfigType) (*corev1.Probe, error)
 
+// EnvVarGenerator is a function that generates a list of environment variables for a given config.
+// It's expected that type Config is the configuration used by a parser.
+type EnvVarGenerator[ComponentConfigType any] func(logger logr.Logger, config ComponentConfigType) ([]corev1.EnvVar, error)
+
 // Defaulter is a function that applies given defaults to the passed Config.
 // It's expected that type Config is the configuration used by a parser.
 type Defaulter[ComponentConfigType any] func(logger logr.Logger, defaultAddr string, defaultPort int32, config ComponentConfigType) (map[string]interface{}, error)
@@ -105,6 +109,9 @@ type Parser interface {
 	// GetLivenessProbe returns a liveness probe set for the collector
 	GetLivenessProbe(logger logr.Logger, config interface{}) (*corev1.Probe, error)
 
+	// GetEnvironmentVariables returns a list of environment variables for the collector
+	GetEnvironmentVariables(logger logr.Logger, config interface{}) ([]corev1.EnvVar, error)
+
 	// GetReadinessProbe returns a readiness probe set for the collector
 	GetReadinessProbe(logger logr.Logger, config interface{}) (*corev1.Probe, error)
 

internal/components/generic_parser.go

@@ -34,6 +34,7 @@ type GenericParser[T any] struct {
 	settings        *Settings[T]
 	portParser      PortParser[T]
 	rbacGen         RBACRuleGenerator[T]
+	envVarGen       EnvVarGenerator[T]
 	livenessGen     ProbeGenerator[T]
 	readinessGen    ProbeGenerator[T]
 	defaultsApplier Defaulter[T]
@@ -88,6 +89,17 @@ func (g *GenericParser[T]) GetRBACRules(logger logr.Logger, config interface{})
 	return g.rbacGen(logger, parsed)
 }
 
+func (g *GenericParser[T]) GetEnvironmentVariables(logger logr.Logger, config interface{}) ([]corev1.EnvVar, error) {
+	if g.envVarGen == nil {
+		return nil, nil
+	}
+	var parsed T
+	if err := mapstructure.Decode(config, &parsed); err != nil {
+		return nil, err
+	}
+	return g.envVarGen(logger, parsed)
+}
+
 func (g *GenericParser[T]) Ports(logger logr.Logger, name string, config interface{}) ([]corev1.ServicePort, error) {
 	if g.portParser == nil {
 		return nil, nil

internal/components/multi_endpoint.go

@@ -116,6 +116,10 @@ func (m *MultiPortReceiver) GetRBACRules(logr.Logger, interface{}) ([]rbacv1.Pol
 	return nil, nil
 }
 
+func (m *MultiPortReceiver) GetEnvironmentVariables(logger logr.Logger, config interface{}) ([]corev1.EnvVar, error) {
+	return nil, nil
+}
+
 type MultiPortBuilder[ComponentConfigType any] []Builder[ComponentConfigType]
 
 func NewMultiPortReceiverBuilder(name string) MultiPortBuilder[*MultiProtocolEndpointConfig] {

internal/components/receivers/helpers.go

@@ -138,6 +138,7 @@ var (
 			MustBuild(),
 		components.NewBuilder[kubeletStatsConfig]().WithName("kubeletstats").
 			WithRbacGen(generateKubeletStatsRbacRules).
+			WithEnvVarGen(generateKubeletStatsEnvVars).
 			MustBuild(),
 		NewScraperParser("prometheus"),
 		NewScraperParser("sshcheck"),

internal/components/receivers/kubeletstats.go

@@ -16,6 +16,7 @@ package receivers
 
 import (
 	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 )
 
@@ -42,6 +43,18 @@ type kubeletStatsConfig struct {
 	AuthType            string   `mapstructure:"auth_type"`
 }
 
+func generateKubeletStatsEnvVars(_ logr.Logger, config kubeletStatsConfig) ([]corev1.EnvVar, error) {
+	// The documentation mentions that the K8S_NODE_NAME environment variable is required when using the serviceAccount auth type.
+	// Also, it mentions that it is a good idea to use it for the Read Only Endpoint. Added always to make it easier for users.
+	// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/receiver/kubeletstatsreceiver/README.md
+	if config.AuthType == "serviceAccount" {
+		return []corev1.EnvVar{
+			{Name: "K8S_NODE_NAME", ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{FieldPath: "spec.nodeName"}}},
+		}, nil
+	}
+	return nil, nil
+}
+
 func generateKubeletStatsRbacRules(_ logr.Logger, config kubeletStatsConfig) ([]rbacv1.PolicyRule, error) {
 	// The Kubelet Stats Receiver needs get permissions on the nodes/stats resources always.
 	prs := []rbacv1.PolicyRule{

internal/manifests/collector/container.go

@@ -177,6 +177,12 @@ func Container(cfg config.Config, logger logr.Logger, otelcol v1beta1.OpenTeleme
 		)
 	}
 
+	if envVars, err := otelcol.Spec.Config.GetEnvironmentVariables(logger); err != nil {
+		logger.Error(err, "could not get the environment variables from the config")
+	} else {
+		envVars = append(envVars, envVars...)
+	}
+
 	envVars = append(envVars, proxy.ReadProxyVarsFromEnv()...)
 	return corev1.Container{
 		Name:            naming.Container(),