add functionality to scrape pods per kubernetes node

go.mod

@@ -136,5 +136,5 @@ require (
 	gopkg.in/olivere/elastic.v5 v5.0.70
 	gopkg.in/yaml.v2 v2.2.4
 	gotest.tools v2.2.0+incompatible // indirect
-	k8s.io/apimachinery v0.17.1 // indirect
+	k8s.io/apimachinery v0.17.1
 )

go.sum

@@ -578,6 +578,7 @@ k8s.io/apimachinery v0.17.1 h1:zUjS3szTxoUjTDYNvdFkYt2uMEXLcthcbp+7uZvWhYM=
 k8s.io/apimachinery v0.17.1/go.mod h1:b9qmWdKlLuU9EBh+06BtLcSf/Mu89rWL33naRxs1uZg=
 k8s.io/gengo v0.0.0-20190128074634-0689ccc1d7d6/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/klog v0.0.0-20181102134211-b9b56d5dfc92/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
+k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
 k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
 k8s.io/kube-openapi v0.0.0-20191107075043-30be4d16710a/go.mod h1:1TqjTSzOxsLGIKfj0lK8EeCP7K1iUG65v09OM0/WG5E=
 rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=

plugins/inputs/prometheus/README.md

@@ -36,6 +36,11 @@ in Prometheus format.
   ## Restricts Kubernetes monitoring to a single namespace
   ##   ex: monitor_kubernetes_pods_namespace = "default"
   # monitor_kubernetes_pods_namespace = ""
+  # label selector to target pods which have the label
+  # kubernetes_label_selector = "env=dev,app=nginx"
+  # field selector to target pods
+  # eg. To scrape pods on a specific node
+  # kubernetes_field_selector = "spec.nodeName=$HOSTNAME"
 
   ## Use bearer token for authorization. ('bearer_token' takes priority)
   # bearer_token = "/path/to/bearer/token"

plugins/inputs/prometheus/kubernetes.go

@@ -82,8 +82,11 @@ func (p *Prometheus) start(ctx context.Context) error {
 // pod, causing errors in the logs. This is only true if the pod going offline is not
 // directed to do so by K8s.
 func (p *Prometheus) watch(ctx context.Context, client *k8s.Client) error {
+
+	selectors := podSelector(p)
+
 	pod := &corev1.Pod{}
-	watcher, err := client.Watch(ctx, p.PodNamespace, &corev1.Pod{})
+	watcher, err := client.Watch(ctx, p.PodNamespace, &corev1.Pod{}, selectors...)
 	if err != nil {
 		return err
 	}
@@ -135,6 +138,21 @@ func podReady(statuss []*corev1.ContainerStatus) bool {
 	return true
 }
 
+func podSelector(p *Prometheus) []k8s.Option {
+	options := []k8s.Option{}
+
+	if len(p.KubernetesLabelSelector) > 0 {
+		options = append(options, k8s.QueryParam("labelSelector", p.KubernetesLabelSelector))
+	}
+
+	if len(p.KubernetesFieldSelector) > 0 {
+		options = append(options, k8s.QueryParam("fieldSelector", p.KubernetesFieldSelector))
+	}
+
+	return options
+
+}
+
 func registerPod(pod *corev1.Pod, p *Prometheus) {
 	if p.kubernetesPods == nil {
 		p.kubernetesPods = map[string]URLAndAddress{}

plugins/inputs/prometheus/kubernetes_test.go

@@ -1,6 +1,7 @@
 package prometheus
 
 import (
+	"github.com/ericchiang/k8s"
 	"testing"
 
 	"github.com/influxdata/telegraf/testutil"
@@ -95,8 +96,54 @@ func TestDeletePods(t *testing.T) {
 	assert.Equal(t, 0, len(prom.kubernetesPods))
 }
 
+func TestPodSelector(t *testing.T) {
+
+	cases := []struct {
+		expected      []k8s.Option
+		labelselector string
+		fieldselector string
+	}{
+		{
+			expected: []k8s.Option{
+				k8s.QueryParam("labelSelector", "key1=val1,key2=val2,key3"),
+				k8s.QueryParam("fieldSelector", "spec.nodeName=ip-1-2-3-4.acme.com"),
+			},
+			labelselector: "key1=val1,key2=val2,key3",
+			fieldselector: "spec.nodeName=ip-1-2-3-4.acme.com",
+		},
+		{
+			expected: []k8s.Option{
+				k8s.QueryParam("labelSelector", "key1"),
+				k8s.QueryParam("fieldSelector", "spec.nodeName=ip-1-2-3-4.acme.com"),
+			},
+			labelselector: "key1",
+			fieldselector: "spec.nodeName=ip-1-2-3-4.acme.com",
+		},
+		{
+			expected: []k8s.Option{
+				k8s.QueryParam("labelSelector", "key1"),
+				k8s.QueryParam("fieldSelector", "somefield"),
+			},
+			labelselector: "key1",
+			fieldselector: "somefield",
+		},
+	}
+
+	for _, c := range cases {
+		prom := &Prometheus{
+			Log:                     testutil.Logger{},
+			KubernetesLabelSelector: c.labelselector,
+			KubernetesFieldSelector: c.fieldselector,
+		}
+
+		output := podSelector(prom)
+
+		assert.Equal(t, len(output), len(c.expected))
+	}
+}
+
 func pod() *v1.Pod {
-	p := &v1.Pod{Metadata: &metav1.ObjectMeta{}, Status: &v1.PodStatus{}}
+	p := &v1.Pod{Metadata: &metav1.ObjectMeta{}, Status: &v1.PodStatus{}, Spec: &v1.PodSpec{}}
 	p.Status.PodIP = str("127.0.0.1")
 	p.Metadata.Name = str("myPod")
 	p.Metadata.Namespace = str("default")

plugins/inputs/prometheus/prometheus.go

@@ -15,6 +15,8 @@ import (
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/tls"
 	"github.com/influxdata/telegraf/plugins/inputs"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
 )
 
 const acceptHeader = `application/vnd.google.protobuf;proto=io.prometheus.client.MetricFamily;encoding=delimited;q=0.7,text/plain;version=0.0.4;q=0.3,*/*;q=0.1`
@@ -29,6 +31,12 @@ type Prometheus struct {
 	// Location of kubernetes config file
 	KubeConfig string
 
+	// Label Selector/s for Kubernetes
+	KubernetesLabelSelector string `toml:"kubernetes_label_selector"`
+
+	// Field Selector/s for Kubernetes
+	KubernetesFieldSelector string `toml:"kubernetes_field_selector"`
+
 	// Bearer Token authorization file path
 	BearerToken       string `toml:"bearer_token"`
 	BearerTokenString string `toml:"bearer_token_string"`
@@ -90,6 +98,11 @@ var sampleConfig = `
   ## Restricts Kubernetes monitoring to a single namespace
   ##   ex: monitor_kubernetes_pods_namespace = "default"
   # monitor_kubernetes_pods_namespace = ""
+  # label selector to target pods which have the label
+  # kubernetes_label_selector = "env=dev,app=nginx"
+  # field selector to target pods
+  # eg. To scrape pods on a specific node
+  # kubernetes_field_selector = "spec.nodeName=$HOSTNAME"
 
   ## Use bearer token for authorization. ('bearer_token' takes priority)
   # bearer_token = "/path/to/bearer/token"
@@ -124,6 +137,23 @@ func (p *Prometheus) Init() error {
 	if p.MetricVersion != 2 {
 		p.Log.Warnf("Use of deprecated configuration: 'metric_version = 1'; please update to 'metric_version = 2'")
 	}
+
+	if len(p.KubernetesLabelSelector) > 0 {
+		_, err := labels.Parse(p.KubernetesLabelSelector)
+		if err != nil {
+			p.Log.Errorf("label selector validation failed %q: %v", p.KubernetesLabelSelector, err)
+			return err
+		}
+	}
+
+	if len(p.KubernetesFieldSelector) > 0 {
+		_, err := fields.ParseSelector(p.KubernetesFieldSelector)
+		if err != nil {
+			p.Log.Errorf("field selector validation failed %s: %v", p.KubernetesFieldSelector, err)
+			return err
+		}
+	}
+
 	return nil
 }