Add support for passing URL query params to pod metric endpoints (#109)

Adds a new metric-config option named `rawQuery`. The value of
this option will be appended to the metric `path` as URL query
parameters to be used by the pod's application. E.g.,:
```
metric-config.pods.requests-per-second.json-path/rawQuery: "foo=bar&baz=bop"
```
will apppend `?foo=bar&baz=bop` to the URL.

Signed-off-by: Abe Friesen <2319792+doyshinda@users.noreply.github.com>

README.md

@@ -155,6 +155,17 @@ values of JSONPath expressions that evaluate to arrays/slices of numbers.
 It's optional but when the expression evaluates to an array/slice, it's absence will
 produce an error. The supported aggregation functions are `avg`, `max`, `min` and `sum`.
 
+The `raw-query` configuration option specifies the query params to send along to the endpoint:
+```yaml
+  metric-config.pods.requests-per-second.json-path/path: /metrics
+  metric-config.pods.requests-per-second.json-path/port: "9090"
+  metric-config.pods.requests-per-second.json-path/raw-query: "foo=bar&baz=bop"
+```
+will create a URL like this:
+```
+http://<podIP>:9090/metrics?foo=bar&baz=bop
+```
+
 ## Prometheus collector
 
 The Prometheus collector is a generic collector which can map Prometheus
@@ -369,10 +380,11 @@ metadata:
     # instead of using the ones specified via CLI. Respectively:
     #  - --influxdb-address
     #  - --influxdb-token
-    #  - --influxdb-org-id
+    #  - --influxdb-org
     metric-config.external.flux-query.influxdb/address: "http://influxdbv2.my-namespace.svc"
     metric-config.external.flux-query.influxdb/token: "secret-token"
-    metric-config.external.flux-query.influxdb/org-id: "deadbeef"
+    # This could be either the organization name or the ID.
+    metric-config.external.flux-query.influxdb/org: "deadbeef"
     # metric-config.<metricType>.<metricName>.<collectorName>/<configKey>
     # <configKey> == query-name
     metric-config.external.flux-query.influxdb/queue_depth: |

go.mod

@@ -2,7 +2,7 @@ module github.com/zalando-incubator/kube-metrics-adapter
 
 require (
 	github.com/NYTimes/gziphandler v1.0.1 // indirect
-	github.com/aws/aws-sdk-go v1.16.6
+	github.com/aws/aws-sdk-go v1.29.4
 	github.com/coreos/go-systemd v0.0.0-20180705093442-88bfeed483d3 // indirect
 	github.com/golang/groupcache v0.0.0-20180513044358-24b0969c4cb7 // indirect
 	github.com/googleapis/gnostic v0.2.0 // indirect
@@ -15,7 +15,7 @@ require (
 	github.com/sirupsen/logrus v1.4.2
 	github.com/soheilhy/cmux v0.1.4 // indirect
 	github.com/spf13/cobra v0.0.3
-	github.com/stretchr/testify v1.3.0
+	github.com/stretchr/testify v1.4.0
 	github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6 // indirect
 	github.com/zalando-incubator/cluster-lifecycle-manager v0.0.0-20180921141935-824b77fb1f84
 	golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45

go.sum

@@ -40,8 +40,8 @@ github.com/apache/arrow/go/arrow v0.0.0-20190426170622-338c62a2a205/go.mod h1:W8
 github.com/apex/log v1.1.0 h1:J5rld6WVFi6NxA6m8GJ1LJqu3+GiTFIt3mYv27gdQWI=
 github.com/apex/log v1.1.0/go.mod h1:yA770aXIDQrhVOIGurT/pVdfCpSq1GQV/auzMN5fzvY=
 github.com/aws/aws-sdk-go v1.15.64/go.mod h1:E3/ieXAlvM0XWO57iftYVDLLvQ824smPP3ATZkfNZeM=
-github.com/aws/aws-sdk-go v1.16.6 h1:pig/KdfESvIv4gUu1B8nVAJAURxbPCTt6e5u79Nqxqc=
-github.com/aws/aws-sdk-go v1.16.6/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
+github.com/aws/aws-sdk-go v1.29.4 h1:w3O/LGvLCliVFJ2fGrpaWDGbRHj1f+aipB1MMfInN24=
+github.com/aws/aws-sdk-go v1.29.4/go.mod h1:1KvfttTE3SPKMpo8g2c6jL3ZKfXtFvKscTgahTma5Xg=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/blakesmith/ar v0.0.0-20150311145944-8bd4349a67f2 h1:oMCHnXa6CCCafdPDbMh/lWRhRByN0VFLvv+g+ayx1SI=
@@ -106,6 +106,8 @@ github.com/go-openapi/swag v0.19.2 h1:jvO6bCMBEilGwMfHhrd61zIID4oIFdwb76V17SM88d
 github.com/go-openapi/swag v0.19.2/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-sql-driver/mysql v1.4.0 h1:7LxgVwFb2hIQtMm87NdgAVfXjnt4OePseqT1tKx+opk=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
+github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gGcHOs=
+github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/gogo/protobuf v1.2.2-0.20190723190241-65acae22fc9d h1:3PaI8p3seN09VjbTYC/QWlUZdZ1qS1zGjy7LH2Wt07I=
 github.com/gogo/protobuf v1.2.2-0.20190723190241-65acae22fc9d/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
@@ -256,6 +258,8 @@ github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/term v0.0.0-20180730021639-bffc007b7fd5/go.mod h1:eCbImbZ95eXtAUIbLAuAVnBnwf83mjf6QIVH8SHYwqQ=
 github.com/pmezard/go-difflib v0.0.0-20151028094244-d8ed2627bdf0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -301,6 +305,8 @@ github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6 h1:lYIiVDtZnyTWlNwiAxLj0bbpTcx1BWCFhXjfsvmPdNc=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
@@ -351,6 +357,8 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190812203447-cdfb69ac37fc h1:gkKoSkUmnU6bpS/VhkuO27bzQeSA51uaEfbOW5dNb68=
 golang.org/x/net v0.0.0-20190812203447-cdfb69ac37fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
+golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be h1:vEDujvNQGv4jgYKudGeI/+DAX4Jffq6hpD55MmoEvKs=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181017192945-9dcd33a902f4/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=

pkg/collector/influxdb_collector.go

@@ -18,7 +18,7 @@ const (
 	InfluxDBMetricName        = "flux-query"
 	influxDBAddressKey        = "address"
 	influxDBTokenKey          = "token"
-	influxDBOrgIDKey          = "org-id"
+	influxDBOrgKey            = "org"
 	influxDBQueryNameLabelKey = "query-name"
 )
 
@@ -26,26 +26,26 @@ type InfluxDBCollectorPlugin struct {
 	kubeClient kubernetes.Interface
 	address    string
 	token      string
-	orgID      string
+	org        string
 }
 
-func NewInfluxDBCollectorPlugin(client kubernetes.Interface, address, token, orgID string) (*InfluxDBCollectorPlugin, error) {
+func NewInfluxDBCollectorPlugin(client kubernetes.Interface, address, token, org string) (*InfluxDBCollectorPlugin, error) {
 	return &InfluxDBCollectorPlugin{
 		kubeClient: client,
 		address:    address,
 		token:      token,
-		orgID:      orgID,
+		org:        org,
 	}, nil
 }
 
 func (p *InfluxDBCollectorPlugin) NewCollector(hpa *v2beta2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
-	return NewInfluxDBCollector(p.address, p.orgID, p.token, config, interval)
+	return NewInfluxDBCollector(p.address, p.token, p.org, config, interval)
 }
 
 type InfluxDBCollector struct {
 	address string
 	token   string
-	orgID   string
+	org     string
 
 	influxDBClient *influxdb.Client
 	interval       time.Duration
@@ -54,7 +54,7 @@ type InfluxDBCollector struct {
 	query          string
 }
 
-func NewInfluxDBCollector(address string, token string, orgID string, config *MetricConfig, interval time.Duration) (*InfluxDBCollector, error) {
+func NewInfluxDBCollector(address string, token string, org string, config *MetricConfig, interval time.Duration) (*InfluxDBCollector, error) {
 	collector := &InfluxDBCollector{
 		interval:   interval,
 		metric:     config.Metric,
@@ -87,8 +87,8 @@ func NewInfluxDBCollector(address string, token string, orgID string, config *Me
 	if v, ok := config.Config[influxDBTokenKey]; ok {
 		token = v
 	}
-	if v, ok := config.Config[influxDBOrgIDKey]; ok {
-		orgID = v
+	if v, ok := config.Config[influxDBOrgKey]; ok {
+		org = v
 	}
 	influxDbClient, err := influxdb.New(address, token)
 	if err != nil {
@@ -96,7 +96,7 @@ func NewInfluxDBCollector(address string, token string, orgID string, config *Me
 	}
 	collector.address = address
 	collector.token = token
-	collector.orgID = orgID
+	collector.org = org
 	collector.influxDBClient = influxDbClient
 	return collector, nil
 }
@@ -109,7 +109,7 @@ type queryResult struct {
 
 // getValue returns the first result gathered from an InfluxDB instance.
 func (c *InfluxDBCollector) getValue() (resource.Quantity, error) {
-	res, err := c.influxDBClient.QueryCSV(context.Background(), c.query, c.orgID)
+	res, err := c.influxDBClient.QueryCSV(context.Background(), c.query, c.org)
 	if err != nil {
 		return resource.Quantity{}, err
 	}

pkg/collector/influxdb_collector_test.go

@@ -36,7 +36,7 @@ func TestInfluxDBCollector_New(t *testing.T) {
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
-		if got, want := c.orgID, "deadbeef"; want != got {
+		if got, want := c.org, "deadbeef"; want != got {
 			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
 		}
 		if got, want := c.address, "http://localhost:9999"; want != got {
@@ -69,7 +69,7 @@ func TestInfluxDBCollector_New(t *testing.T) {
 				"range3m":    `from(bucket: "?") |> range(start: -3m)`,
 				"address":    "http://localhost:9999",
 				"token":      "sEcr3TT0ken",
-				"org-id":     "deadbeef1234",
+				"org":        "deadbeef1234",
 				"query-name": "range3m",
 			},
 		}
@@ -77,7 +77,7 @@ func TestInfluxDBCollector_New(t *testing.T) {
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
-		if got, want := c.orgID, "deadbeef1234"; want != got {
+		if got, want := c.org, "deadbeef1234"; want != got {
 			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
 		}
 		if got, want := c.address, "http://localhost:9999"; want != got {

pkg/collector/json_path_collector.go

@@ -25,6 +25,7 @@ type JSONPathMetricsGetter struct {
 	port       int
 	aggregator string
 	client     *http.Client
+	rawQuery   string
 }
 
 // NewJSONPathMetricsGetter initializes a new JSONPathMetricsGetter.
@@ -49,6 +50,10 @@ func NewJSONPathMetricsGetter(config map[string]string) (*JSONPathMetricsGetter,
 		getter.path = v
 	}
 
+	if v, ok := config["raw-query"]; ok {
+		getter.rawQuery = v
+	}
+
 	if v, ok := config["port"]; ok {
 		n, err := strconv.Atoi(v)
 		if err != nil {
@@ -83,7 +88,7 @@ func defaultHTTPClient() *http.Client {
 // endpoint and extracting the desired value using the specified json path
 // query.
 func (g *JSONPathMetricsGetter) GetMetric(pod *corev1.Pod) (float64, error) {
-	data, err := g.getPodMetrics(pod, g.scheme, g.path, g.port)
+	data, err := g.getPodMetrics(pod)
 	if err != nil {
 		return 0, err
 	}
@@ -140,20 +145,12 @@ func castSlice(in []interface{}) ([]float64, error) {
 }
 
 // getPodMetrics returns the content of the pods metrics endpoint.
-func (g *JSONPathMetricsGetter) getPodMetrics(pod *corev1.Pod, scheme, path string, port int) ([]byte, error) {
+func (g *JSONPathMetricsGetter) getPodMetrics(pod *corev1.Pod) ([]byte, error) {
 	if pod.Status.PodIP == "" {
 		return nil, fmt.Errorf("pod %s/%s does not have a pod IP", pod.Namespace, pod.Namespace)
 	}
 
-	if scheme == "" {
-		scheme = "http"
-	}
-
-	metricsURL := url.URL{
-		Scheme: scheme,
-		Host:   fmt.Sprintf("%s:%d", pod.Status.PodIP, port),
-		Path:   path,
-	}
+	metricsURL := g.buildMetricsURL(pod.Status.PodIP)
 
 	request, err := http.NewRequest(http.MethodGet, metricsURL.String(), nil)
 	if err != nil {
@@ -178,6 +175,22 @@ func (g *JSONPathMetricsGetter) getPodMetrics(pod *corev1.Pod, scheme, path stri
 	return data, nil
 }
 
+// buildMetricsURL will build the full URL needed to hit the pod metric endpoint.
+func (g *JSONPathMetricsGetter) buildMetricsURL(podIP string) url.URL {
+	var scheme = g.scheme
+
+	if scheme == "" {
+		scheme = "http"
+	}
+
+	return url.URL{
+		Scheme:   scheme,
+		Host:     fmt.Sprintf("%s:%d", podIP, g.port),
+		Path:     g.path,
+		RawQuery: g.rawQuery,
+	}
+}
+
 // reduce will reduce a slice of numbers given a aggregator function's name. If it's empty or not recognized, an error is returned.
 func reduce(values []float64, aggregator string) (float64, error) {
 	switch aggregator {

pkg/collector/json_path_collector_test.go

@@ -1,6 +1,7 @@
 package collector
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/oliveagle/jsonpath"
@@ -70,6 +71,25 @@ func TestNewJSONPathMetricsGetter(t *testing.T) {
 
 	_, err4 := NewJSONPathMetricsGetter(configErrorPort)
 	require.Error(t, err4)
+
+	configWithRawQuery := map[string]string{
+		"json-key":  "$.values",
+		"scheme":    "http",
+		"path":      "/metrics",
+		"port":      "9090",
+		"raw-query": "foo=bar&baz=bop",
+	}
+	jpath5, _ := jsonpath.Compile(configWithRawQuery["json-key"])
+	getterWithRawQuery, err5 := NewJSONPathMetricsGetter(configWithRawQuery)
+
+	require.NoError(t, err5)
+	compareMetricsGetter(t, &JSONPathMetricsGetter{
+		jsonPath: jpath5,
+		scheme:   "http",
+		path:     "/metrics",
+		port:     9090,
+		rawQuery: "foo=bar&baz=bop",
+	}, getterWithRawQuery)
 }
 
 func TestCastSlice(t *testing.T) {
@@ -110,3 +130,44 @@ func TestReduce(t *testing.T) {
 	_, err5 := reduce([]float64{1, 2, 3}, "inexistent_function")
 	require.Errorf(t, err5, "slice of numbers was returned by JSONPath, but no valid aggregator function was specified: %v", "inexistent_function")
 }
+
+func TestBuildMetricsURL(t *testing.T) {
+	scheme := "http"
+	ip := "1.2.3.4"
+	port := "9090"
+	path := "/v1/test/"
+	rawQuery := "foo=bar&baz=bop"
+
+	// Test building URL with rawQuery
+	configWithRawQuery := map[string]string{
+		"json-key":  "$.value",
+		"scheme":    scheme,
+		"path":      path,
+		"port":      port,
+		"raw-query": rawQuery,
+	}
+	_, err := jsonpath.Compile(configWithRawQuery["json-key"])
+	require.NoError(t, err)
+	getterWithRawQuery, err1 := NewJSONPathMetricsGetter(configWithRawQuery)
+	require.NoError(t, err1)
+
+	expectedURLWithQuery := fmt.Sprintf("%s://%s:%s%s?%s", scheme, ip, port, path, rawQuery)
+	receivedURLWithQuery := getterWithRawQuery.buildMetricsURL(ip)
+	require.Equal(t, receivedURLWithQuery.String(), expectedURLWithQuery)
+
+	// Test building URL without rawQuery
+	configWithNoQuery := map[string]string{
+		"json-key": "$.value",
+		"scheme":   scheme,
+		"path":     path,
+		"port":     port,
+	}
+	_, err2 := jsonpath.Compile(configWithNoQuery["json-key"])
+	require.NoError(t, err2)
+	getterWithNoQuery, err3 := NewJSONPathMetricsGetter(configWithNoQuery)
+	require.NoError(t, err3)
+
+	expectedURLNoQuery := fmt.Sprintf("%s://%s:%s%s", scheme, ip, port, path)
+	receivedURLNoQuery := getterWithNoQuery.buildMetricsURL(ip)
+	require.Equal(t, receivedURLNoQuery.String(), expectedURLNoQuery)
+}

pkg/server/start.go

@@ -92,7 +92,7 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
 		"address of InfluxDB 2.x server to query (e.g. http://localhost:9999)")
 	flags.StringVar(&o.InfluxDBToken, "influxdb-token", o.InfluxDBToken, ""+
 		"token for InfluxDB 2.x server to query")
-	flags.StringVar(&o.InfluxDBOrgID, "influxdb-org-id", o.InfluxDBOrgID, ""+
+	flags.StringVar(&o.InfluxDBOrg, "influxdb-org", o.InfluxDBOrg, ""+
 		"organization ID for InfluxDB 2.x server to query")
 	flags.StringVar(&o.ZMONKariosDBEndpoint, "zmon-kariosdb-endpoint", o.ZMONKariosDBEndpoint, ""+
 		"url of ZMON KariosDB endpoint to query for ZMON checks")
@@ -183,7 +183,7 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
 	}
 
 	if o.InfluxDBAddress != "" {
-		influxdbPlugin, err := collector.NewInfluxDBCollectorPlugin(client, o.InfluxDBAddress, o.InfluxDBToken, o.InfluxDBOrgID)
+		influxdbPlugin, err := collector.NewInfluxDBCollectorPlugin(client, o.InfluxDBAddress, o.InfluxDBToken, o.InfluxDBOrg)
 		if err != nil {
 			return fmt.Errorf("failed to initialize InfluxDB collector plugin: %v", err)
 		}
@@ -309,8 +309,8 @@ type AdapterServerOptions struct {
 	InfluxDBAddress string
 	// InfluxDBToken is the token used for querying InfluxDB
 	InfluxDBToken string
-	// InfluxDBOrgID is the organization ID used for querying InfluxDB
-	InfluxDBOrgID string
+	// InfluxDBOrg is the organization ID used for querying InfluxDB
+	InfluxDBOrg string
 	// ZMONKariosDBEndpoint enables ZMON check queries to the specified
 	// kariosDB endpoint
 	ZMONKariosDBEndpoint string