From dd1ace73b0cedc1b0bd68f515d23eadca0dfc51b Mon Sep 17 00:00:00 2001 From: mg03 Date: Mon, 2 Mar 2020 18:51:31 -0800 Subject: [PATCH] Add label and field selectors to prometheus input k8s discovery (#6969) --- go.mod | 2 +- go.sum | 1 + plugins/inputs/prometheus/README.md | 5 ++ plugins/inputs/prometheus/kubernetes.go | 20 +++++++- plugins/inputs/prometheus/kubernetes_test.go | 49 +++++++++++++++++++- plugins/inputs/prometheus/prometheus.go | 28 +++++++++++ 6 files changed, 102 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 84b24db20..93d92db27 100644 --- a/go.mod +++ b/go.mod @@ -134,5 +134,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 ) diff --git a/go.sum b/go.sum index f940ab12a..fc3ef5dcc 100644 --- a/go.sum +++ b/go.sum @@ -588,6 +588,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= diff --git a/plugins/inputs/prometheus/README.md b/plugins/inputs/prometheus/README.md index 7b2e054a2..b4e587452 100644 --- a/plugins/inputs/prometheus/README.md +++ b/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" diff --git a/plugins/inputs/prometheus/kubernetes.go b/plugins/inputs/prometheus/kubernetes.go index 617509384..16f69cbd1 100644 --- a/plugins/inputs/prometheus/kubernetes.go +++ b/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{} diff --git a/plugins/inputs/prometheus/kubernetes_test.go b/plugins/inputs/prometheus/kubernetes_test.go index b926f7393..8568ac946 100644 --- a/plugins/inputs/prometheus/kubernetes_test.go +++ b/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") diff --git a/plugins/inputs/prometheus/prometheus.go b/plugins/inputs/prometheus/prometheus.go index 1f0862760..35c2d3d2c 100644 --- a/plugins/inputs/prometheus/prometheus.go +++ b/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,21 @@ 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 { + return fmt.Errorf("label selector validation failed %q: %v", p.KubernetesLabelSelector, err) + } + } + + if len(p.KubernetesFieldSelector) > 0 { + _, err := fields.ParseSelector(p.KubernetesFieldSelector) + if err != nil { + return fmt.Errorf("field selector validation failed %s: %v", p.KubernetesFieldSelector, err) + } + } + return nil }