Merge pull request #5 from vshn/feat/http-sd

Add http service discovery

Author: glrf

README.md

@@ -36,7 +36,15 @@ To test the `exporter-filterproxy` in Kubernetes using the Kubernetes service di
 make kind
 ```
 
-This will start a local kind cluster and deploy a `exporter-filterproxy` that proxies the metrics of the CoreDNS service.
+This will start a local kind cluster and deploy a `exporter-filterproxy` that proxies the metrics of the CoreDNS service, as well as a Prometheus instance.
+
+Once the cluster started you can expose the test Prometheus using port-forward
+
+```
+kubectl port-forward svc/prometheus 9090
+```
+
+You should see two discovered targets through the exporter-filterproxy when you open `http://localhost:9090/targets`
 
 
 ## Configuration

discovery.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	"github.com/vshn/exporter-filterproxy/target"
+)
+
+type targetConfigFetcher interface {
+	FetchTargetConfigs(ctx context.Context, baseTarget string, basePath string) ([]target.StaticConfig, error)
+}
+
+type multiTargetConfigFetcher map[string]targetConfigFetcher
+
+func (mf multiTargetConfigFetcher) FetchTargetConfigs(ctx context.Context, baseTarget string, basePath string) ([]target.StaticConfig, error) {
+
+	configs := []target.StaticConfig{}
+	for path, f := range mf {
+		c, err := f.FetchTargetConfigs(ctx, baseTarget, basePath+path)
+		if err != nil {
+			log.Printf("Failed to fetch targets: %s", err.Error())
+			continue
+		}
+		configs = append(configs, c...)
+	}
+
+	return configs, nil
+}

discovery_test.go

@@ -0,0 +1,86 @@
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/prometheus/common/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/vshn/exporter-filterproxy/target"
+)
+
+func TestMultiTargetConfigFetcher(t *testing.T) {
+
+	fa := fakeTargetConfigFetcher{
+		t:      t,
+		path:   "/a",
+		target: "proxy.example.com",
+		configs: []target.StaticConfig{
+			{
+				Targets: []string{"proxy.example.com"},
+				Labels: map[model.LabelName]model.LabelValue{
+					"foo": "a1",
+				},
+			},
+			{
+				Targets: []string{"proxy.example.com"},
+				Labels: map[model.LabelName]model.LabelValue{
+					"foo": "a2",
+				},
+			},
+		},
+	}
+	fb := fakeTargetConfigFetcher{
+		t:      t,
+		path:   "/b",
+		target: "proxy.example.com",
+		configs: []target.StaticConfig{
+			{
+				Targets: []string{"proxy.example.com"},
+				Labels: map[model.LabelName]model.LabelValue{
+					"foo": "b1",
+				},
+			},
+			{
+				Targets: []string{"proxy.example.com"},
+				Labels: map[model.LabelName]model.LabelValue{
+					"foo": "b2",
+				},
+			},
+		},
+	}
+
+	mf := multiTargetConfigFetcher{
+		"/a": fa,
+		"/b": fb,
+	}
+
+	conf, err := mf.FetchTargetConfigs(context.TODO(), "proxy.example.com", "")
+	require.NoError(t, err)
+	require.Len(t, conf, 4)
+
+	seenFoolabels := []string{}
+	for _, c := range conf {
+		require.Len(t, c.Targets, 1)
+		assert.Equal(t, "proxy.example.com", c.Targets[0])
+		require.Len(t, c.Labels, 1)
+
+		seenFoolabels = append(seenFoolabels, string(c.Labels["foo"]))
+	}
+	assert.ElementsMatch(t, seenFoolabels, []string{"b1", "b2", "a1", "a2"})
+
+}
+
+type fakeTargetConfigFetcher struct {
+	configs []target.StaticConfig
+	t       *testing.T
+	path    string
+	target  string
+}
+
+func (f fakeTargetConfigFetcher) FetchTargetConfigs(ctx context.Context, baseTarget string, basePath string) ([]target.StaticConfig, error) {
+	assert.Equal(f.t, f.path, basePath)
+	assert.Equal(f.t, f.target, baseTarget)
+	return f.configs, nil
+}

handle.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log"
@@ -67,6 +68,33 @@ func multiHandler(prefix string, fetcher multiMetricsFetcher) http.HandlerFunc {
 	})
 }
 
+func serviceDiscoveryHandler(prefix string, fetcher targetConfigFetcher) http.HandlerFunc {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+
+		configs, err := fetcher.FetchTargetConfigs(r.Context(), r.Host, strings.TrimSuffix(r.URL.Path, "/"))
+		if err != nil {
+			log.Printf("Failed to discover Endpoints: %s", err.Error())
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		res, err := json.Marshal(configs)
+		if err != nil {
+			log.Printf("Failed to marshal Endpoints: %s", err.Error())
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		_, err = w.Write(res)
+
+		if err != nil {
+			log.Printf("Failed to write: %s", err.Error())
+			return
+		}
+	})
+}
+
 func parseURLParams(values url.Values) (map[string]string, error) {
 	res := map[string]string{}
 	for k, v := range values {

handle_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 
 	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/vshn/exporter-filterproxy/target"
@@ -152,3 +153,52 @@ func (f fakeMultiMetricsFetcher) FetchMetricsFor(ctx context.Context, endpoint s
 func deref[T any](x T) *T {
 	return &x
 }
+
+func TestDiscoveryHandler(t *testing.T) {
+
+	f := fakeTargetConfigFetcher{
+		configs: []target.StaticConfig{
+			{
+				Targets: []string{"proxy.example.com"},
+				Labels: map[model.LabelName]model.LabelValue{
+					"__metrics_path__": "/foo/a.b.c.d",
+					"metrics_path":     "/foo",
+					"instance":         "a.b.c.d",
+				},
+			},
+			{
+				Targets: []string{"proxy.example.com"},
+				Labels: map[model.LabelName]model.LabelValue{
+					"__metrics_path__": "/foo/d.e.f.g",
+					"metrics_path":     "/foo",
+					"instance":         "d.e.f.g",
+				},
+			},
+			{
+				Targets: []string{"proxy.example.com"},
+				Labels: map[model.LabelName]model.LabelValue{
+					"__metrics_path__": "/bar",
+					"metrics_path":     "/bar",
+				},
+			},
+		},
+		t:      t,
+		path:   "",
+		target: "proxy.example.com",
+	}
+	h := serviceDiscoveryHandler("", f)
+
+	req, err := http.NewRequest("GET", "/", nil)
+	req.Host = "proxy.example.com"
+	require.NoError(t, err)
+	rr := httptest.NewRecorder()
+
+	h.ServeHTTP(rr, req)
+	assert.Equal(t, http.StatusOK, rr.Code)
+
+	assert.Equal(t, "application/json", rr.Header().Get("Content-Type"))
+
+	assert.Equal(t, expectedDiscoveryRes, rr.Body.String())
+}
+
+var expectedDiscoveryRes = `[{"targets":["proxy.example.com"],"labels":{"__metrics_path__":"/foo/a.b.c.d","instance":"a.b.c.d","metrics_path":"/foo"}},{"targets":["proxy.example.com"],"labels":{"__metrics_path__":"/foo/d.e.f.g","instance":"d.e.f.g","metrics_path":"/foo"}},{"targets":["proxy.example.com"],"labels":{"__metrics_path__":"/bar","metrics_path":"/bar"}}]`

main.go

@@ -25,6 +25,8 @@ func main() {
 		return
 	}
 
+	targetDiscovery := multiTargetConfigFetcher{}
+
 	for name, endpoint := range conf.Endpoints {
 
 		authToken, err := getAuthToken(endpoint.Auth)
@@ -36,11 +38,11 @@ func main() {
 		switch {
 		case endpoint.Target != "":
 			log.Printf("Registering static endpoint %q at %s", name, endpoint.Path)
+			sf := target.NewStaticFetcher(endpoint.Target, authToken, endpoint.RefreshInterval, endpoint.InsecureSkipVerify)
 			mux.HandleFunc(endpoint.Path,
-				handler(
-					target.NewStaticFetcher(endpoint.Target, authToken, endpoint.RefreshInterval, endpoint.InsecureSkipVerify),
-				),
+				handler(sf),
 			)
+			targetDiscovery[endpoint.Path] = sf
 		case endpoint.KubernetesTarget != nil:
 			log.Printf("Registering kube endpoint %q at %s", name, endpoint.Path)
 			kf, err := target.NewKubernetesEndpointFetcher(
@@ -62,13 +64,21 @@ func main() {
 			mux.HandleFunc(endpoint.Path+"/",
 				multiHandler(endpoint.Path, kf),
 			)
+			mux.HandleFunc(endpoint.Path,
+				serviceDiscoveryHandler(endpoint.Path, kf),
+			)
+			targetDiscovery[endpoint.Path] = kf
 		default:
 			log.Fatalf("No target set for endpoint %s", name)
 			return
 		}
 
 	}
 
+	mux.HandleFunc("/",
+		serviceDiscoveryHandler("", targetDiscovery),
+	)
+
 	srv := &http.Server{
 		Addr:         conf.Addr,
 		ReadTimeout:  5 * time.Second,

sample-config/prometheus.yml

@@ -5,10 +5,7 @@ global:
 
 
 scrape_configs:
-  - job_name: 'node-cpu-2'
-    metrics_path: /node
-    params:
-      cpu: ['2']
-    static_configs:
-      - targets: ['filterproxy:8082']
+  - job_name: 'exporter-filterproxy'
+    http_sd_configs:
+      - url: 'http://filterproxy:8082'
 

target/kubernetes.go

@@ -11,6 +11,7 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/model"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 
@@ -123,6 +124,28 @@ func (f *KubernetesEndpointFetcher) FetchMetricsFor(ctx context.Context, endpoin
 	return f.cache[endpoint], nil
 }
 
+func (f *KubernetesEndpointFetcher) FetchTargetConfigs(ctx context.Context, baseTarget string, basePath string) ([]StaticConfig, error) {
+	staticConfig := []StaticConfig{}
+
+	endpoints, err := f.discover(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, ip := range endpoints {
+		conf := StaticConfig{
+			Targets: []string{baseTarget},
+			Labels: map[model.LabelName]model.LabelValue{
+				"__metrics_path__": model.LabelValue(fmt.Sprintf("%s/%s", basePath, ip)),
+				"metrics_path":     model.LabelValue(basePath),
+				"instance":         model.LabelValue(ip),
+			},
+		}
+		staticConfig = append(staticConfig, conf)
+	}
+	return staticConfig, nil
+}
+
 func (f *KubernetesEndpointFetcher) now() time.Time {
 	if f.clock != nil {
 		return f.clock()

target/kubernetes_test.go

@@ -2,6 +2,7 @@ package target
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"net/http"
 	"net/http/httptest"
@@ -386,3 +387,38 @@ func newTestEndpoint(port int, ips ...string) *corev1.Endpoints {
 		},
 	}
 }
+
+func TestKube_FetchTargetConfigs(t *testing.T) {
+	podIps := []string{"127.0.18.1", "127.0.18.2", "127.0.18.4", "127.0.18.8"}
+
+	f := KubernetesEndpointFetcher{
+		endpointname: "test-ep",
+		namespace:    "fetch-test",
+		port:         8119,
+		kube: newTestKubeEnv(
+			newTestEndpoint(8119, podIps...),
+		),
+	}
+
+	tconfs, err := f.FetchTargetConfigs(context.TODO(), "proxy.example.com", "/buzz")
+	require.NoError(t, err)
+	require.Len(t, tconfs, 4)
+
+	confMap := map[string]StaticConfig{}
+
+	for i, tconf := range tconfs {
+		require.Len(t, tconf.Targets, 1)
+		assert.Equal(t, "proxy.example.com", tconf.Targets[0])
+		assert.Len(t, tconf.Labels, 3)
+
+		inst := string(tconf.Labels["instance"])
+		assert.Contains(t, podIps, inst)
+		confMap[inst] = tconfs[i]
+	}
+
+	for _, pip := range podIps {
+		require.Contains(t, confMap, pip)
+		assert.EqualValues(t, fmt.Sprintf("/buzz/%s", pip), confMap[pip].Labels["__metrics_path__"])
+		assert.EqualValues(t, "/buzz", confMap[pip].Labels["metrics_path"])
+	}
+}