feat(inputs.kube_inventory): Support filtering pods and nodes by node name (#13993)

plugins/inputs/kube_inventory/README.md

@@ -55,6 +55,9 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## Namespace to use. Set to "" to use all namespaces.
   # namespace = "default"
 
+  ## Node name to filter to. No filtering by default.
+  # node_name = ""
+
   ## Use bearer token for authorization. ('bearer_token' takes priority)
   ##
   ## Ignored if url is empty and in-cluster config is used.
@@ -112,7 +115,6 @@ list "persistentvolumes" and "nodes". You will then need to make an [aggregated
 ClusterRole][agg] that will eventually be bound to a user or group.
 
 [rbac]: https://kubernetes.io/docs/reference/access-authn-authz/rbac/
-
 [agg]: https://kubernetes.io/docs/reference/access-authn-authz/rbac/#aggregated-clusterroles
 
 ```yaml
@@ -365,11 +367,11 @@ tls_key = "/run/telegraf-kubernetes-key"
 
 The node status ready can mean 3 different values.
 
-| Tag value | Corresponding field value |  Meaning |
-| --------- | ------------------------- |  -------
-| ready     | 0                         |  NotReady|
-| ready     | 1                         |  Ready   |
-| ready     | 2                         |  Unknown |
+| Tag value | Corresponding field value | Meaning  |
+| --------- | ------------------------- | -------- |
+| ready     | 0                         | NotReady |
+| ready     | 1                         | Ready    |
+| ready     | 2                         | Unknown  |
 
 ### pv `phase_type`
 

plugins/inputs/kube_inventory/client.go

@@ -86,10 +86,14 @@ func (c *client) getIngress(ctx context.Context) (*netv1.IngressList, error) {
 	return c.NetworkingV1().Ingresses(c.namespace).List(ctx, metav1.ListOptions{})
 }
 
-func (c *client) getNodes(ctx context.Context) (*corev1.NodeList, error) {
+func (c *client) getNodes(ctx context.Context, name string) (*corev1.NodeList, error) {
 	ctx, cancel := context.WithTimeout(ctx, c.timeout)
 	defer cancel()
-	return c.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+	var fieldSelector string
+	if name != "" {
+		fieldSelector = "metadata.name=" + name
+	}
+	return c.CoreV1().Nodes().List(ctx, metav1.ListOptions{FieldSelector: fieldSelector})
 }
 
 func (c *client) getPersistentVolumes(ctx context.Context) (*corev1.PersistentVolumeList, error) {
@@ -104,10 +108,14 @@ func (c *client) getPersistentVolumeClaims(ctx context.Context) (*corev1.Persist
 	return c.CoreV1().PersistentVolumeClaims(c.namespace).List(ctx, metav1.ListOptions{})
 }
 
-func (c *client) getPods(ctx context.Context) (*corev1.PodList, error) {
+func (c *client) getPods(ctx context.Context, nodeName string) (*corev1.PodList, error) {
 	ctx, cancel := context.WithTimeout(ctx, c.timeout)
 	defer cancel()
-	return c.CoreV1().Pods(c.namespace).List(ctx, metav1.ListOptions{})
+	var fieldSelector string
+	if nodeName != "" {
+		fieldSelector = "spec.nodeName=" + nodeName
+	}
+	return c.CoreV1().Pods(c.namespace).List(ctx, metav1.ListOptions{FieldSelector: fieldSelector})
 }
 
 func (c *client) getServices(ctx context.Context) (*corev1.ServiceList, error) {

plugins/inputs/kube_inventory/kube_inventory.go

@@ -36,10 +36,10 @@ type KubernetesInventory struct {
 	ResourceInclude   []string        `toml:"resource_include"`
 	MaxConfigMapAge   config.Duration `toml:"max_config_map_age"`
 
-	SelectorInclude []string `toml:"selector_include"`
-	SelectorExclude []string `toml:"selector_exclude"`
-
-	Log telegraf.Logger `toml:"-"`
+	SelectorInclude []string        `toml:"selector_include"`
+	SelectorExclude []string        `toml:"selector_exclude"`
+	NodeName        string          `toml:"node_name"`
+	Log             telegraf.Logger `toml:"-"`
 
 	tls.ClientConfig
 	client *client

plugins/inputs/kube_inventory/node.go

@@ -9,7 +9,7 @@ import (
 )
 
 func collectNodes(ctx context.Context, acc telegraf.Accumulator, ki *KubernetesInventory) {
-	list, err := ki.client.getNodes(ctx)
+	list, err := ki.client.getNodes(ctx, ki.NodeName)
 	if err != nil {
 		acc.AddError(err)
 		return

plugins/inputs/kube_inventory/pod.go

@@ -10,7 +10,8 @@ import (
 )
 
 func collectPods(ctx context.Context, acc telegraf.Accumulator, ki *KubernetesInventory) {
-	list, err := ki.client.getPods(ctx)
+	list, err := ki.client.getPods(ctx, ki.NodeName)
+
 	if err != nil {
 		acc.AddError(err)
 		return

plugins/inputs/kube_inventory/sample.conf

@@ -7,6 +7,9 @@
   ## Namespace to use. Set to "" to use all namespaces.
   # namespace = "default"
 
+  ## Node name to filter to. No filtering by default.
+  # node_name = ""
+
   ## Use bearer token for authorization. ('bearer_token' takes priority)
   ##
   ## Ignored if url is empty and in-cluster config is used.