Add scraping for Prometheus endpoint in Kubernetes

plugins/inputs/prometheus/README.md

@@ -14,6 +14,17 @@ in Prometheus format.
   ## An array of Kubernetes services to scrape metrics from.
   # kubernetes_services = ["http://my-service-dns.my-namespace:9100/metrics"]
 
+  ## Kubernetes config file to create client from.
+  # kube_config = "/path/to/kubernetes.config"
+
+  ## Scrape Kubernetes pods for the following prometheus annotations:
+  ## - prometheus.io/scrape: Enable scraping for this pod
+  ## - prometheus.io/scheme: If the metrics endpoint is secured then you will need to
+  ##     set this to `https` & most likely set the tls config.
+  ## - prometheus.io/path: If the metrics path is not /metrics, define it with this annotation.
+  ## - prometheus.io/port: If port is not 9102 use this annotation
+  # monitor_kubernetes_pods = true
+
   ## Use bearer token for authorization
   # bearer_token = /path/to/bearer/token
 
@@ -37,6 +48,18 @@ by looking up all A records assigned to the hostname as described in
 This method can be used to locate all
 [Kubernetes headless services](https://kubernetes.io/docs/concepts/services-networking/service/#headless-services).
 
+#### Kubernetes scraping
+
+Enabling this option will allow the plugin to scrape for prometheus annotation on Kubernetes
+pods. Currently, you can run this plugin in your kubernetes cluster, or we use the kubeconfig
+file to determine where to monitor.
+Currently the following annotation are supported:
+
+* `prometheus.io/scrape` Enable scraping for this pod.
+* `prometheus.io/scheme` If the metrics endpoint is secured then you will need to set this to `https` & most likely set the tls config. (default 'http')
+* `prometheus.io/path` Override the path for the metrics endpoint on the service. (default '/metrics')
+* `prometheus.io/port` Used to override the port. (default 9102)
+
 #### Bearer Token
 
 If set, the file specified by the `bearer_token` parameter will be read on

plugins/inputs/prometheus/kubernetes.go

@@ -0,0 +1,201 @@
+package prometheus
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/url"
+	"os/user"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/ericchiang/k8s"
+	corev1 "github.com/ericchiang/k8s/apis/core/v1"
+	"gopkg.in/yaml.v2"
+)
+
+type payload struct {
+	eventype string
+	pod      *corev1.Pod
+}
+
+// loadClient parses a kubeconfig from a file and returns a Kubernetes
+// client. It does not support extensions or client auth providers.
+func loadClient(kubeconfigPath string) (*k8s.Client, error) {
+	data, err := ioutil.ReadFile(kubeconfigPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed reading '%s': %v", kubeconfigPath, err)
+	}
+
+	// Unmarshal YAML into a Kubernetes config object.
+	var config k8s.Config
+	if err := yaml.Unmarshal(data, &config); err != nil {
+		return nil, err
+	}
+	return k8s.NewClient(&config)
+}
+
+func (p *Prometheus) start(ctx context.Context) error {
+	client, err := k8s.NewInClusterClient()
+	if err != nil {
+		u, err := user.Current()
+		if err != nil {
+			return fmt.Errorf("Failed to get current user - %v", err)
+		}
+		configLocation := filepath.Join(u.HomeDir, ".kube/config")
+		if p.KubeConfig != "" {
+			configLocation = p.KubeConfig
+		}
+		client, err = loadClient(configLocation)
+		if err != nil {
+			return err
+		}
+	}
+
+	p.wg = sync.WaitGroup{}
+	in := make(chan payload)
+
+	p.wg.Add(1)
+	go func() {
+		defer p.wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				break
+			case <-time.After(time.Second):
+				err := p.watch(ctx, client, in)
+				if err == nil {
+					break
+				}
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (p *Prometheus) watch(ctx context.Context, client *k8s.Client, in chan payload) error {
+	pod := &corev1.Pod{}
+	watcher, err := client.Watch(ctx, "", &corev1.Pod{})
+	if err != nil {
+		log.Printf("E! [inputs.prometheus] unable to watch resources: %v", err)
+		return err
+	}
+	defer watcher.Close()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+			pod = &corev1.Pod{}
+			// An error here means we need to reconnect the watcher.
+			eventType, err := watcher.Next(pod)
+			if err != nil {
+				log.Printf("D! [inputs.prometheus] unable to watch next: %v", err)
+				return err
+			}
+
+			switch eventType {
+			case k8s.EventAdded:
+				registerPod(pod, p)
+			case k8s.EventDeleted:
+				unregisterPod(pod, p)
+			case k8s.EventModified:
+			}
+		}
+	}
+}
+
+func registerPod(pod *corev1.Pod, p *Prometheus) {
+	targetURL := getScrapeURL(pod)
+	if targetURL == nil {
+		return
+	}
+
+	log.Printf("D! [inputs.prometheus] will scrape metrics from %s", *targetURL)
+	// add annotation as metrics tags
+	tags := pod.GetMetadata().GetAnnotations()
+	tags["pod_name"] = pod.GetMetadata().GetName()
+	tags["namespace"] = pod.GetMetadata().GetNamespace()
+	// add labels as metrics tags
+	for k, v := range pod.GetMetadata().GetLabels() {
+		tags[k] = v
+	}
+	URL, err := url.Parse(*targetURL)
+	if err != nil {
+		log.Printf("E! [inputs.prometheus] could not parse URL %s: %v", *targetURL, err)
+		return
+	}
+	podURL := p.AddressToURL(URL, URL.Hostname())
+	p.lock.Lock()
+	p.kubernetesPods = append(p.kubernetesPods,
+		URLAndAddress{
+			URL:         podURL,
+			Address:     URL.Hostname(),
+			OriginalURL: URL,
+			Tags:        tags})
+	p.lock.Unlock()
+}
+
+func getScrapeURL(pod *corev1.Pod) *string {
+	scrape := pod.GetMetadata().GetAnnotations()["prometheus.io/scrape"]
+	if scrape != "true" {
+		return nil
+	}
+	ip := pod.Status.GetPodIP()
+	if ip == "" {
+		// return as if scrape was disabled, we will be notified again once the pod
+		// has an IP
+		return nil
+	}
+
+	scheme := pod.GetMetadata().GetAnnotations()["prometheus.io/scheme"]
+	path := pod.GetMetadata().GetAnnotations()["prometheus.io/path"]
+	port := pod.GetMetadata().GetAnnotations()["prometheus.io/port"]
+
+	if scheme == "" {
+		scheme = "http"
+	}
+	if port == "" {
+		port = "9102"
+	}
+	if path == "" {
+		path = "/metrics"
+	}
+
+	u := &url.URL{
+		Scheme: scheme,
+		Host:   net.JoinHostPort(ip, port),
+		Path:   path,
+	}
+
+	x := u.String()
+
+	return &x
+}
+
+func unregisterPod(pod *corev1.Pod, p *Prometheus) {
+	url := getScrapeURL(pod)
+	if url == nil {
+		return
+	}
+
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	log.Printf("D! [inputs.prometheus] registered a delete request for %s in namespace %s",
+		pod.GetMetadata().GetName(), pod.GetMetadata().GetNamespace())
+	var result []URLAndAddress
+	for _, v := range p.kubernetesPods {
+		if v.URL.String() != *url {
+			result = append(result, v)
+		} else {
+			log.Printf("D! [inputs.prometheus] will stop scraping for %s", *url)
+		}
+
+	}
+	p.kubernetesPods = result
+}