Update dependencies

Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de>

.github/CODEOWNERS

@@ -1,35 +0,0 @@
-# These owners will be the default owners for everything in
-# the repo. 
-*       @arjunrn
-
-
-
-# Samples for assigning codeowners below:
-# Order is important; the last matching pattern takes the most
-# precedence. When someone opens a pull request that only
-# modifies JS files, only @js-owner and not the global
-# owner(s) will be requested for a review.
-# *.js    @js-owner
-
-# You can also use email addresses if you prefer. They'll be
-# used to look up users just like we do for commit author
-# emails.
-# *.go docs@example.com
-
-# In this example, @doctocat owns any files in the build/logs
-# directory at the root of the repository and any of its
-# subdirectories.
-# /build/logs/ @doctocat
-
-# The `docs/*` pattern will match files like
-# `docs/getting-started.md` but not further nested files like
-# `docs/build-app/troubleshooting.md`.
-# docs/*  docs@example.com
-
-# In this example, @octocat owns any file in an apps directory
-# anywhere in your repository.
-# apps/ @octocat
-
-# In this example, @doctocat owns any file in the `/docs`
-# directory in the root of your repository.
-# /docs/ @doctocat

.github/dependabot.yml

@@ -0,0 +1,14 @@
+version: 2
+updates:
+- package-ecosystem: gomod
+  directory: "/"
+  schedule:
+    interval: monthly
+    time: "07:00"
+  open-pull-requests-limit: 10
+- package-ecosystem: docker
+  directory: "/"
+  schedule:
+    interval: monthly
+    time: "07:00"
+  open-pull-requests-limit: 10

.golangci.yml

@@ -1,7 +1,4 @@
 run:
-  skip-files:
-  - "pkg/provider/generated.conversion.go"
-  - "pkg/provider/conversion.go"
 linters-settings:
   golint:
       min-confidence: 0.9

.travis.yml

@@ -2,19 +2,19 @@ language: go
 dist: xenial
 
 go:
-  - "1.11.x"
+  - "1.14.x"
 
 env:
-- GO111MODULE=on GOLANGCI_RELEASE="v1.16.0"
+- GO111MODULE=on GOLANGCI_RELEASE="v1.21.0"
 
 before_install:
-  - go get github.com/mattn/goveralls
-  - go get github.com/lawrencewoodman/roveralls
+  - GO111MODULE=off go get github.com/mattn/goveralls
+  - GO111MODULE=off go get github.com/lawrencewoodman/roveralls
   - curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s -- -b $(go env GOPATH)/bin ${GOLANGCI_RELEASE}
 
 script:
-  - make check
   - make test
   - make build.docker
+  - make check
   - roveralls
   - goveralls -v -coverprofile=roveralls.coverprofile -service=travis-ci

MAINTAINERS

@@ -1,3 +1,2 @@
 Mikkel Larsen <mikkel.larsen@zalando.de>
-Arjun Naik <arjun.naik@zalando.de>
 Team Teapot <team-teapot@zalando.de>

Makefile

@@ -8,30 +8,45 @@ SOURCES       = $(shell find . -name '*.go')
 DOCKERFILE    ?= Dockerfile
 GOPKGS        = $(shell go list ./...)
 BUILD_FLAGS   ?= -v
+OPENAPI       ?= pkg/api/generated/openapi/zz_generated.openapi.go
 LDFLAGS       ?= -X main.version=$(VERSION) -w -s
 
+
 default: build.local
 
 clean:
 	rm -rf build
+	rm -rf $(OPENAPI)
 
 test:
 	go test -v $(GOPKGS)
 
 check:
-	golangci-lint run ./...
+	go mod download
+	golangci-lint run --timeout=2m ./...
+
+
+$(OPENAPI): go.mod
+	go run k8s.io/kube-openapi/cmd/openapi-gen \
+		--go-header-file hack/boilerplate.go.txt \
+		--logtostderr \
+		-i k8s.io/metrics/pkg/apis/custom_metrics,k8s.io/metrics/pkg/apis/custom_metrics/v1beta1,k8s.io/metrics/pkg/apis/custom_metrics/v1beta2,k8s.io/metrics/pkg/apis/external_metrics,k8s.io/metrics/pkg/apis/external_metrics/v1beta1,k8s.io/metrics/pkg/apis/metrics,k8s.io/metrics/pkg/apis/metrics/v1beta1,k8s.io/apimachinery/pkg/apis/meta/v1,k8s.io/apimachinery/pkg/api/resource,k8s.io/apimachinery/pkg/version,k8s.io/api/core/v1 \
+		-p pkg/api/generated/openapi \
+		-o . \
+		-O zz_generated.openapi \
+		-r /dev/null
 
 build.local: build/$(BINARY)
 build.linux: build/linux/$(BINARY)
 build.osx: build/osx/$(BINARY)
 
-build/$(BINARY): go.mod $(SOURCES)
+build/$(BINARY): go.mod $(SOURCES) $(OPENAPI)
 	CGO_ENABLED=0 go build -o build/$(BINARY) $(BUILD_FLAGS) -ldflags "$(LDFLAGS)" .
 
-build/linux/$(BINARY): go.mod $(SOURCES)
+build/linux/$(BINARY): go.mod $(SOURCES) $(OPENAPI)
 	GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build $(BUILD_FLAGS) -o build/linux/$(BINARY) -ldflags "$(LDFLAGS)" .
 
-build/osx/$(BINARY): go.mod $(SOURCES)
+build/osx/$(BINARY): go.mod $(SOURCES) $(OPENAPI)
 	GOOS=darwin GOARCH=amd64 CGO_ENABLED=0 go build $(BUILD_FLAGS) -o build/osx/$(BINARY) -ldflags "$(LDFLAGS)" .
 
 build.docker: build.linux

SECURITY.md

@@ -1,7 +1,8 @@
 We acknowledge that every line of code that we write may potentially contain security issues.
+We are trying to deal with it responsibly and provide patches as quickly as possible.
 
-We are trying to deal with it responsibly and provide patches as quickly as  possible. If you have anything to report to us please use the following channels:
+We host our bug bounty program on HackerOne, it is currently private, therefore if you would like to report a vulnerability and get rewarded for it, please ask to join our program by filling this form:
 
-Email: Tech-Security@zalando.de
-OR
-Submit your vulnerability report through our bug bounty program at: https://hackerone.com/zalando
+https://corporate.zalando.com/en/services-and-contact#security-form
+
+You can also send you report via this form if you do not want to join our bug bounty program and just want to report a vulnerability or security issue.

delivery.yaml

@@ -7,6 +7,8 @@ pipeline:
     - /go/pkg/mod       # pkg cache for Go modules
     - ~/.cache/go-build # Go build cache
   type: script
+  env:
+    GOFLAGS: "-mod=readonly"
   commands:
   - desc: test
     cmd: |
@@ -18,9 +20,11 @@ pipeline:
     cmd: |
       if [[ $CDP_TARGET_BRANCH == master && ! $CDP_PULL_REQUEST_NUMBER ]]; then
         IMAGE=registry-write.opensource.zalan.do/teapot/kube-metrics-adapter
-        VERSION=$(git describe --tags --always --dirty)
+        VERSION=$(git describe --tags --always)
       else
         IMAGE=registry-write.opensource.zalan.do/teapot/kube-metrics-adapter-test
         VERSION=$CDP_BUILD_VERSION
       fi
+      IMAGE=$IMAGE VERSION=$VERSION make build.docker
+      git diff --stat --exit-code
       IMAGE=$IMAGE VERSION=$VERSION make build.push

example/deploy/hpa.yaml

@@ -1,4 +1,4 @@
-apiVersion: autoscaling/v2beta1
+apiVersion: autoscaling/v2beta2
 kind: HorizontalPodAutoscaler
 metadata:
   name: custom-metrics-consumer
@@ -25,24 +25,36 @@ spec:
   # - type: Resource
   #   resource:
   #     name: cpu
-  #     targetAverageUtilization: 50
+  #     current:
+  #       averageUtilization: 50
+
   - type: Pods
     pods:
-      metricName: queue-length
-      targetAverageValue: 1k
+      metric:
+        name: queue-length
+      target:
+        averageValue: 1k
+        type: AverageValue
+
   - type: Object
     object:
-      metricName: requests-per-second
-      target:
+      describedObject:
         apiVersion: extensions/v1beta1
         kind: Ingress
         name: custom-metrics-consumer
-      targetValue: 10 # this will be treated as targetAverageValue
+      metric:
+        name: requests-per-second
+      target:
+        averageValue: "10"
+        type: AverageValue
   - type: External
     external:
-      metricName: sqs-queue-length
-      metricSelector:
-        matchLabels:
-          queue-name: foobar
-          region: eu-central-1
-      targetAverageValue: 30
+      metric:
+        name: sqs-queue-length
+        selector:
+          matchLabels:
+            queue-name: foobar
+            region: eu-central-1
+      target:
+        averageValue: "30"
+        type: AverageValue

go.mod

@@ -1,91 +1,37 @@
 module github.com/zalando-incubator/kube-metrics-adapter
 
 require (
-	bitbucket.org/ww/goautoneg v0.0.0-20120707110453-75cd24fc2f2c // indirect
-	github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 // indirect
-	github.com/BurntSushi/toml v0.3.0 // indirect
 	github.com/NYTimes/gziphandler v1.0.1 // indirect
-	github.com/PuerkitoBio/purell v1.1.0 // indirect
-	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
-	github.com/Sirupsen/logrus v1.0.6 // indirect
-	github.com/aws/aws-sdk-go v1.16.6
-	github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 // indirect
-	github.com/boltdb/bolt v1.3.1 // indirect
-	github.com/coreos/bbolt v1.3.0 // indirect
-	github.com/coreos/etcd v3.3.9+incompatible // indirect
-	github.com/coreos/go-semver v0.2.0 // indirect
-	github.com/coreos/go-systemd v0.0.0-20180705093442-88bfeed483d3 // indirect
-	github.com/coreos/pkg v0.0.0-20180108230652-97fdf19511ea // indirect
-	github.com/dgrijalva/jwt-go v3.2.0+incompatible // indirect
-	github.com/docker/docker v1.13.1 // indirect
-	github.com/elazarl/go-bindata-assetfs v1.0.0 // indirect
-	github.com/emicklei/go-restful v2.8.0+incompatible // indirect
-	github.com/emicklei/go-restful-swagger12 v0.0.0-20170926063155-7524189396c6 // indirect
-	github.com/evanphx/json-patch v4.1.1-0.20190203004735-bbf30d639737+incompatible // indirect
-	github.com/fsnotify/fsnotify v1.4.7 // indirect
-	github.com/ghodss/yaml v1.0.0 // indirect
-	github.com/go-openapi/jsonpointer v0.0.0-20180322222829-3a0015ad55fa // indirect
-	github.com/go-openapi/jsonreference v0.0.0-20180322222742-3fb327e6747d // indirect
-	github.com/go-openapi/spec v0.0.0-20180801175345-384415f06ee2 // indirect
-	github.com/go-openapi/swag v0.0.0-20180715190254-becd2f08beaf // indirect
-	github.com/gogo/protobuf v1.1.1 // indirect
-	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
-	github.com/golang/groupcache v0.0.0-20180513044358-24b0969c4cb7 // indirect
-	github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c // indirect
-	github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf // indirect
+	github.com/aws/aws-sdk-go v1.35.19
+	github.com/go-openapi/spec v0.19.12
 	github.com/googleapis/gnostic v0.2.0 // indirect
-	github.com/gorilla/websocket v1.3.0 // indirect
-	github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7 // indirect
-	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway v1.4.1 // indirect
-	github.com/hashicorp/golang-lru v0.0.0-20180201235237-0fb14efe8c47 // indirect
-	github.com/hpcloud/tail v1.0.0 // indirect
-	github.com/imdario/mergo v0.3.6 // indirect
-	github.com/inconshreveable/mousetrap v1.0.0 // indirect
-	github.com/jonboulle/clockwork v0.1.0 // indirect
-	github.com/json-iterator/go v1.1.5 // indirect
-	github.com/kubernetes-incubator/custom-metrics-apiserver v0.0.0-20190116221620-b7016fc85e1c
-	github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
-	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
-	github.com/modern-go/reflect2 v1.0.1 // indirect
-	github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852
-	github.com/onsi/ginkgo v1.6.0 // indirect
-	github.com/onsi/gomega v1.4.1 // indirect
-	github.com/pborman/uuid v0.0.0-20180122190007-c65b2f87fee3 // indirect
-	github.com/peterbourgon/diskv v2.0.1+incompatible // indirect
-	github.com/prometheus/client_golang v0.9.0-pre1.0.20180824101016-4eb539fa85a2
-	github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 // indirect
-	github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e
-	github.com/prometheus/procfs v0.0.0-20180725123919-05ee40e3a273 // indirect
-	github.com/sirupsen/logrus v1.3.0
-	github.com/soheilhy/cmux v0.1.4 // indirect
-	github.com/spf13/cobra v0.0.3
-	github.com/spf13/pflag v1.0.2 // indirect
-	github.com/stretchr/testify v1.2.2
-	github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6 // indirect
-	github.com/ugorji/go v1.1.1 // indirect
-	github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18 // indirect
+	github.com/influxdata/influxdb-client-go v0.1.5
+	github.com/kubernetes-incubator/custom-metrics-apiserver v0.0.0-20201023074945-51cc7b53320e
+	github.com/lib/pq v1.2.0 // indirect
+	github.com/mailru/easyjson v0.7.6 // indirect
+	github.com/mattn/go-colorable v0.1.4 // indirect
+	github.com/onsi/gomega v1.8.1 // indirect
+	github.com/prometheus/client_golang v1.8.0
+	github.com/prometheus/common v0.14.0
+	github.com/sirupsen/logrus v1.7.0
+	github.com/spf13/cobra v0.0.7
+	github.com/spyzhov/ajson v0.4.2
+	github.com/stretchr/testify v1.6.1
 	github.com/zalando-incubator/cluster-lifecycle-manager v0.0.0-20180921141935-824b77fb1f84
-	golang.org/x/crypto v0.0.0-20181015023909-0c41d7ab0a0e // indirect
-	golang.org/x/net v0.0.0-20180824152047-4bcd98cce591 // indirect
-	golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be
-	golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f // indirect
-	golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2 // indirect
-	google.golang.org/appengine v1.2.0 // indirect
-	google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8 // indirect
-	google.golang.org/grpc v1.14.0 // indirect
-	gopkg.in/airbrake/gobrake.v2 v2.0.9 // indirect
-	gopkg.in/fsnotify.v1 v1.4.7 // indirect
-	gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 // indirect
-	gopkg.in/inf.v0 v0.9.1 // indirect
-	gopkg.in/natefinch/lumberjack.v2 v2.0.0-20170531160350-a96e63847dc3 // indirect
-	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
-	gopkg.in/yaml.v2 v2.2.1 // indirect
-	k8s.io/api v0.0.0-20190226173710-145d52631d00
-	k8s.io/apimachinery v0.0.0-20190221084156-01f179d85dbc
-	k8s.io/apiserver v0.0.0-20190226174732-cf2f1d68202d
-	k8s.io/client-go v2.0.0-alpha.0.0.20190226174127-78295b709ec6+incompatible
-	k8s.io/kube-openapi v0.0.0-20180731170545-e3762e86a74c // indirect
-	k8s.io/metrics v0.0.0-20190226180357-f3f09b9076d1
+	golang.org/x/net v0.0.0-20201031054903-ff519b6c9102 // indirect
+	golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45
+	golang.org/x/text v0.3.4 // indirect
+	golang.org/x/tools v0.0.0-20200204192400-7124308813f3 // indirect
+	gonum.org/v1/netlib v0.0.0-20190331212654-76723241ea4e // indirect
+	honnef.co/go/tools v0.0.1-2020.1.3 // indirect
+	k8s.io/api v0.18.8
+	k8s.io/apimachinery v0.18.8
+	k8s.io/apiserver v0.18.8
+	k8s.io/client-go v0.18.8
+	k8s.io/component-base v0.18.8
+	k8s.io/klog v1.0.0
+	k8s.io/kube-openapi v0.0.0-20200410145947-61e04a5be9a6
+	k8s.io/metrics v0.18.8
 )
+
+go 1.15

hack/boilerplate.go.txt

@@ -0,0 +1,16 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+

hack/tools.go

@@ -0,0 +1,6 @@
+// +build tools
+
+// This package imports things required by build scripts, to force `go mod` to see them as dependencies
+package tools
+
+import _ "k8s.io/kube-openapi/cmd/openapi-gen"

main.go

@@ -18,12 +18,13 @@ package main
 
 import (
 	"flag"
+	_ "net/http/pprof"
 	"os"
 	"runtime"
 
 	"github.com/zalando-incubator/kube-metrics-adapter/pkg/server"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apiserver/pkg/util/logs"
+	"k8s.io/component-base/logs"
 )
 
 func main() {

pkg/annotations/parser_test.go

@@ -65,6 +65,42 @@ func TestParser(t *testing.T) {
 			},
 			PerReplica: false,
 		},
+		{
+			Name: "influxdb metrics",
+			Annotations: map[string]string{
+				"metric-config.external.flux-query.influxdb/range1m": `from(bucket: "?") |> range(start: -1m)`,
+				"metric-config.external.flux-query.influxdb/range2m": `from(bucket: "?") |> range(start: -2m)`,
+				"metric-config.external.flux-query.influxdb/range3m": `from(bucket: "?") |> range(start: -3m)`,
+				"metric-config.external.flux-query.influxdb/address": "http://localhost:9999",
+				"metric-config.external.flux-query.influxdb/token":   "sEcr3TT0ken",
+				"metric-config.external.flux-query.influxdb/org-id":  "deadbeef",
+			},
+			MetricName: "flux-query",
+			MetricType: autoscalingv2.ExternalMetricSourceType,
+			ExpectedConfig: map[string]string{
+				"range1m": `from(bucket: "?") |> range(start: -1m)`,
+				"range2m": `from(bucket: "?") |> range(start: -2m)`,
+				"range3m": `from(bucket: "?") |> range(start: -3m)`,
+				"address": "http://localhost:9999",
+				"token":   "sEcr3TT0ken",
+				"org-id":  "deadbeef",
+			},
+		},
+		{
+			Name: "http metrics",
+			Annotations: map[string]string{
+				"metric-config.external.http.json/json-key":   "$.metric.value",
+				"metric-config.external.http.json/endpoint":   "http://metric-source.source-namespace.svc.cluster.local:8000/metrics",
+				"metric-config.external.http.json/aggregator": "avg",
+			},
+			MetricName: "http",
+			MetricType: autoscalingv2.ExternalMetricSourceType,
+			ExpectedConfig: map[string]string{
+				"json-key":   "$.metric.value",
+				"endpoint":   "http://metric-source.source-namespace.svc.cluster.local:8000/metrics",
+				"aggregator": "avg",
+			},
+		},
 	} {
 		t.Run(tc.Name, func(t *testing.T) {
 			hpaMap := make(AnnotationConfigMap)

pkg/collector/aws_collector.go

@@ -56,11 +56,11 @@ func NewAWSSQSCollector(sessions map[string]*session.Session, config *MetricConf
 		return nil, fmt.Errorf("selector for queue is not specified")
 	}
 
-	name, ok := config.Metric.Selector.MatchLabels[sqsQueueNameLabelKey]
+	name, ok := config.Config[sqsQueueNameLabelKey]
 	if !ok {
 		return nil, fmt.Errorf("sqs queue name not specified on metric")
 	}
-	region, ok := config.Metric.Selector.MatchLabels[sqsQueueRegionLabelKey]
+	region, ok := config.Config[sqsQueueRegionLabelKey]
 	if !ok {
 		return nil, fmt.Errorf("sqs queue region is not specified on metric")
 	}

pkg/collector/collector.go

@@ -46,6 +46,14 @@ type CollectorPlugin interface {
 	NewCollector(hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error)
 }
 
+type PluginNotFoundError struct {
+	metricTypeName MetricTypeName
+}
+
+func (p *PluginNotFoundError) Error() string {
+	return fmt.Sprintf("no plugin found for %s", p.metricTypeName)
+}
+
 func (c *CollectorFactory) RegisterPodsCollector(metricCollector string, plugin CollectorPlugin) error {
 	if metricCollector == "" {
 		c.podsPlugins.Any = plugin
@@ -139,7 +147,7 @@ func (c *CollectorFactory) NewCollector(hpa *autoscalingv2.HorizontalPodAutoscal
 		}
 	}
 
-	return nil, fmt.Errorf("no plugin found for %s", config.MetricTypeName)
+	return nil, &PluginNotFoundError{metricTypeName: config.MetricTypeName}
 }
 
 type MetricTypeName struct {
@@ -165,6 +173,7 @@ type MetricConfig struct {
 	ObjectReference custom_metrics.ObjectReference
 	PerReplica      bool
 	Interval        time.Duration
+	MetricSpec      autoscalingv2.MetricSpec
 }
 
 // ParseHPAMetrics parses the HPA object into a list of metric configurations.
@@ -206,10 +215,15 @@ func ParseHPAMetrics(hpa *autoscalingv2.HorizontalPodAutoscaler) ([]*MetricConfi
 			MetricTypeName:  typeName,
 			ObjectReference: ref,
 			Config:          map[string]string{},
+			MetricSpec:      metric,
 		}
 
-		if metric.Type == autoscalingv2.ExternalMetricSourceType {
-			config.Config = metric.External.Metric.Selector.MatchLabels
+		if metric.Type == autoscalingv2.ExternalMetricSourceType &&
+			metric.External.Metric.Selector != nil &&
+			metric.External.Metric.Selector.MatchLabels != nil {
+			for k, v := range metric.External.Metric.Selector.MatchLabels {
+				config.Config[k] = v
+			}
 		}
 
 		annotationConfigs, present := parser.GetAnnotationConfig(typeName.Metric.Name, typeName.Type)
@@ -217,7 +231,11 @@ func ParseHPAMetrics(hpa *autoscalingv2.HorizontalPodAutoscaler) ([]*MetricConfi
 			config.CollectorName = annotationConfigs.CollectorName
 			config.Interval = annotationConfigs.Interval
 			config.PerReplica = annotationConfigs.PerReplica
-			config.Config = annotationConfigs.Configs
+			// configs specified in annotations takes precedence
+			// over labels
+			for k, v := range annotationConfigs.Configs {
+				config.Config[k] = v
+			}
 		}
 		metricConfigs = append(metricConfigs, config)
 	}

pkg/collector/http_collector.go

@@ -0,0 +1,103 @@
+package collector
+
+import (
+	"fmt"
+	"net/url"
+	"time"
+
+	"github.com/zalando-incubator/kube-metrics-adapter/pkg/collector/httpmetrics"
+
+	"k8s.io/api/autoscaling/v2beta2"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/metrics/pkg/apis/external_metrics"
+)
+
+const (
+	HTTPMetricName            = "http"
+	HTTPEndpointAnnotationKey = "endpoint"
+	HTTPJsonPathAnnotationKey = "json-key"
+	identifierLabel           = "identifier"
+)
+
+type HTTPCollectorPlugin struct{}
+
+func NewHTTPCollectorPlugin() (*HTTPCollectorPlugin, error) {
+	return &HTTPCollectorPlugin{}, nil
+}
+
+func (p *HTTPCollectorPlugin) NewCollector(_ *v2beta2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
+	collector := &HTTPCollector{}
+	var (
+		value string
+		ok    bool
+	)
+	if value, ok = config.Config[HTTPJsonPathAnnotationKey]; !ok {
+		return nil, fmt.Errorf("config value %s not found", HTTPJsonPathAnnotationKey)
+	}
+	jsonPath := value
+
+	if value, ok = config.Config[HTTPEndpointAnnotationKey]; !ok {
+		return nil, fmt.Errorf("config value %s not found", HTTPEndpointAnnotationKey)
+	}
+	var err error
+	collector.endpoint, err = url.Parse(value)
+	if err != nil {
+		return nil, err
+	}
+	collector.interval = interval
+	collector.metricType = config.Type
+	if config.Metric.Selector == nil || config.Metric.Selector.MatchLabels == nil {
+		return nil, fmt.Errorf("no label selector specified for metric: %s", config.Metric.Name)
+	}
+	if _, ok := config.Metric.Selector.MatchLabels[identifierLabel]; !ok {
+		return nil, fmt.Errorf("%s is not specified as a label for metric %s", identifierLabel, config.Metric.Name)
+	}
+	collector.metric = config.Metric
+	var aggFunc httpmetrics.AggregatorFunc
+
+	if val, ok := config.Config["aggregator"]; ok {
+		aggFunc, err = httpmetrics.ParseAggregator(val)
+		if err != nil {
+			return nil, err
+		}
+	}
+	jsonPathGetter, err := httpmetrics.NewJSONPathMetricsGetter(httpmetrics.DefaultMetricsHTTPClient(), aggFunc, jsonPath)
+	if err != nil {
+		return nil, err
+	}
+	collector.metricsGetter = jsonPathGetter
+	return collector, nil
+}
+
+type HTTPCollector struct {
+	endpoint      *url.URL
+	interval      time.Duration
+	metricType    v2beta2.MetricSourceType
+	metricsGetter *httpmetrics.JSONPathMetricsGetter
+	metric        v2beta2.MetricIdentifier
+}
+
+func (c *HTTPCollector) GetMetrics() ([]CollectedMetric, error) {
+	metric, err := c.metricsGetter.GetMetric(*c.endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	value := CollectedMetric{
+		Type: c.metricType,
+		External: external_metrics.ExternalMetricValue{
+			MetricName:   c.metric.Name,
+			MetricLabels: c.metric.Selector.MatchLabels,
+			Timestamp: metav1.Time{
+				Time: time.Now(),
+			},
+			Value: *resource.NewMilliQuantity(int64(metric*1000), resource.DecimalSI),
+		},
+	}
+	return []CollectedMetric{value}, nil
+}
+
+func (c *HTTPCollector) Interval() time.Duration {
+	return c.interval
+}

pkg/collector/http_collector_test.go

@@ -0,0 +1,94 @@
+package collector
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/api/autoscaling/v2beta2"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type testExternalMetricsHandler struct {
+	values []int64
+	test   *testing.T
+}
+
+func (t testExternalMetricsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	response, err := json.Marshal(testMetricResponse{t.values})
+	require.NoError(t.test, err)
+	w.Header().Set("Content-Type", "application/json")
+	_, err = w.Write(response)
+	require.NoError(t.test, err)
+}
+
+func makeHTTPTestServer(t *testing.T, values []int64) string {
+	server := httptest.NewServer(&testExternalMetricsHandler{values: values, test: t})
+	return server.URL
+}
+
+func TestHTTPCollector(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		values     []int64
+		output     int
+		aggregator string
+	}{
+		{
+			name:       "basic",
+			values:     []int64{3},
+			output:     3,
+			aggregator: "sum",
+		},
+		{
+			name:       "sum",
+			values:     []int64{3, 5, 6},
+			aggregator: "sum",
+			output:     14,
+		},
+		{
+			name:       "average",
+			values:     []int64{3, 5, 6},
+			aggregator: "sum",
+			output:     14,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			testServer := makeHTTPTestServer(t, tc.values)
+			plugin, err := NewHTTPCollectorPlugin()
+			require.NoError(t, err)
+			testConfig := makeTestHTTPCollectorConfig(testServer, tc.aggregator)
+			collector, err := plugin.NewCollector(nil, testConfig, testInterval)
+			require.NoError(t, err)
+			metrics, err := collector.GetMetrics()
+			require.NoError(t, err)
+			require.NotNil(t, metrics)
+			require.Len(t, metrics, 1)
+			require.EqualValues(t, metrics[0].External.Value.Value(), tc.output)
+		})
+	}
+}
+
+func makeTestHTTPCollectorConfig(endpoint, aggregator string) *MetricConfig {
+	config := &MetricConfig{
+		MetricTypeName: MetricTypeName{
+			Type: v2beta2.ExternalMetricSourceType,
+			Metric: v2beta2.MetricIdentifier{
+				Name: "test-metric",
+				Selector: &v1.LabelSelector{
+					MatchLabels: map[string]string{identifierLabel: "test-metric"},
+				},
+			},
+		},
+		Config: map[string]string{
+			HTTPJsonPathAnnotationKey: "$.values",
+			HTTPEndpointAnnotationKey: endpoint,
+		},
+	}
+	if aggregator != "" {
+		config.Config["aggregator"] = aggregator
+	}
+	return config
+}

pkg/collector/httpmetrics/aggregator.go

@@ -0,0 +1,65 @@
+package httpmetrics
+
+import (
+	"fmt"
+	"math"
+)
+
+type AggregatorFunc func(...float64) float64
+
+// Average implements the average mathematical function over a slice of float64
+func Average(values ...float64) float64 {
+	sum := Sum(values...)
+	return sum / float64(len(values))
+}
+
+// Minimum implements the absolute minimum mathematical function over a slice of float64
+func Minimum(values ...float64) float64 {
+	// initialized with positive infinity, all finite numbers are smaller than it
+	curMin := math.Inf(1)
+	for _, v := range values {
+		if v < curMin {
+			curMin = v
+		}
+	}
+	return curMin
+}
+
+// Maximum implements the absolute maximum mathematical function over a slice of float64
+func Maximum(values ...float64) float64 {
+	// initialized with negative infinity, all finite numbers are bigger than it
+	curMax := math.Inf(-1)
+	for _, v := range values {
+		if v > curMax {
+			curMax = v
+		}
+	}
+	return curMax
+}
+
+// Sum implements the summation mathematical function over a slice of float64
+func Sum(values ...float64) float64 {
+	res := 0.0
+
+	for _, v := range values {
+		res += v
+	}
+
+	return res
+}
+
+// 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 ParseAggregator(aggregator string) (AggregatorFunc, error) {
+	switch aggregator {
+	case "avg":
+		return Average, nil
+	case "min":
+		return Minimum, nil
+	case "max":
+		return Maximum, nil
+	case "sum":
+		return Sum, nil
+	default:
+		return nil, fmt.Errorf("aggregator function: %s is unknown", aggregator)
+	}
+}

pkg/collector/httpmetrics/aggregator_test.go

@@ -0,0 +1,57 @@
+package httpmetrics
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestReduce(t *testing.T) {
+	for _, tc := range []struct {
+		input      []float64
+		output     float64
+		aggregator string
+		parseError bool
+	}{
+		{
+			input:      []float64{1, 2, 3},
+			output:     2.0,
+			aggregator: "avg",
+			parseError: false,
+		},
+		{
+			input:      []float64{1, 2, 3},
+			output:     1.0,
+			aggregator: "min",
+			parseError: false,
+		},
+		{
+			input:      []float64{1, 2, 3},
+			output:     3.0,
+			aggregator: "max",
+			parseError: false,
+		},
+		{
+			input:      []float64{1, 2, 3},
+			output:     6.0,
+			aggregator: "sum",
+			parseError: false,
+		},
+		{
+			input:      []float64{1, 2, 3},
+			aggregator: "non-existent",
+			parseError: true,
+		},
+	} {
+		t.Run(fmt.Sprintf("Test function: %s", tc.aggregator), func(t *testing.T) {
+			aggFunc, err := ParseAggregator(tc.aggregator)
+			if tc.parseError {
+				require.Error(t, err)
+			} else {
+				val := aggFunc(tc.input...)
+				require.Equal(t, tc.output, val)
+			}
+		})
+	}
+}

pkg/collector/httpmetrics/json_path.go

@@ -0,0 +1,128 @@
+package httpmetrics
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/spyzhov/ajson"
+)
+
+// JSONPathMetricsGetter is a metrics getter which looks up pod metrics by
+// querying the pods metrics endpoint and lookup the metric value as defined by
+// the json path query.
+type JSONPathMetricsGetter struct {
+	jsonPath   string
+	aggregator AggregatorFunc
+	client     *http.Client
+}
+
+// NewJSONPathMetricsGetter initializes a new JSONPathMetricsGetter.
+func NewJSONPathMetricsGetter(httpClient *http.Client, aggregatorFunc AggregatorFunc, jsonPath string) (*JSONPathMetricsGetter, error) {
+	// check that jsonPath parses
+	_, err := ajson.ParseJSONPath(jsonPath)
+	if err != nil {
+		return nil, err
+	}
+	return &JSONPathMetricsGetter{client: httpClient, aggregator: aggregatorFunc, jsonPath: jsonPath}, nil
+}
+
+var DefaultRequestTimeout = 15 * time.Second
+var DefaultConnectTimeout = 15 * time.Second
+
+func CustomMetricsHTTPClient(requestTimeout time.Duration, connectTimeout time.Duration) *http.Client {
+	client := &http.Client{
+		Transport: &http.Transport{
+			DialContext: (&net.Dialer{
+				Timeout: connectTimeout,
+			}).DialContext,
+			MaxIdleConns:          50,
+			IdleConnTimeout:       90 * time.Second,
+			ExpectContinueTimeout: 1 * time.Second,
+		},
+		Timeout: requestTimeout,
+	}
+	return client
+}
+
+func DefaultMetricsHTTPClient() *http.Client {
+	return CustomMetricsHTTPClient(DefaultRequestTimeout, DefaultConnectTimeout)
+}
+
+// GetMetric gets metric from pod by fetching json metrics from the pods metric
+// endpoint and extracting the desired value using the specified json path
+// query.
+func (g *JSONPathMetricsGetter) GetMetric(metricsURL url.URL) (float64, error) {
+	data, err := g.fetchMetrics(metricsURL)
+	if err != nil {
+		return 0, err
+	}
+
+	// parse data
+	root, err := ajson.Unmarshal(data)
+	if err != nil {
+		return 0, err
+	}
+
+	nodes, err := root.JSONPath(g.jsonPath)
+	if err != nil {
+		return 0, err
+	}
+
+	if len(nodes) != 1 {
+		return 0, fmt.Errorf("unexpected json: expected single numeric or array value")
+	}
+
+	node := nodes[0]
+	if node.IsArray() {
+		if g.aggregator == nil {
+			return 0, fmt.Errorf("no aggregator function has been specified")
+		}
+		values := make([]float64, 0, len(nodes))
+		items, _ := node.GetArray()
+		for _, item := range items {
+			value, err := item.GetNumeric()
+			if err != nil {
+				return 0, fmt.Errorf("did not find numeric type: %w", err)
+			}
+			values = append(values, value)
+		}
+		return g.aggregator(values...), nil
+	} else if node.IsNumeric() {
+		res, _ := node.GetNumeric()
+		return res, nil
+	}
+
+	value, err := node.Value()
+	if err != nil {
+		return 0, fmt.Errorf("failed to check value of jsonPath result: %w", err)
+	}
+	return 0, fmt.Errorf("unsupported type %T", value)
+}
+
+func (g *JSONPathMetricsGetter) fetchMetrics(metricsURL url.URL) ([]byte, error) {
+	request, err := http.NewRequest(http.MethodGet, metricsURL.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := g.client.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	data, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unsuccessful response: %s", resp.Status)
+	}
+
+	return data, nil
+}

pkg/collector/httpmetrics/json_path_test.go

@@ -0,0 +1,84 @@
+package httpmetrics
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func makeTestHTTPServer(t *testing.T, response []byte) *httptest.Server {
+	h := func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(t, r.URL.Path, "/metrics")
+		w.Header().Set("Content-Type", "application/json")
+		_, err := w.Write(response)
+		require.NoError(t, err)
+	}
+	return httptest.NewServer(http.HandlerFunc(h))
+}
+
+func TestJSONPathMetricsGetter(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		jsonResponse []byte
+		jsonPath     string
+		result       float64
+		aggregator   AggregatorFunc
+		err          error
+	}{
+		{
+			name:         "basic single value",
+			jsonResponse: []byte(`{"value":3}`),
+			jsonPath:     "$.value",
+			result:       3,
+			aggregator:   Average,
+		},
+		{
+			name:         "basic average",
+			jsonResponse: []byte(`{"value":[3,4,5]}`),
+			jsonPath:     "$.value",
+			result:       4,
+			aggregator:   Average,
+		},
+		{
+			name:         "dotted key",
+			jsonResponse: []byte(`{"metric.value":5}`),
+			jsonPath:     "$['metric.value']",
+			result:       5,
+			aggregator:   Average,
+		},
+		{
+			name:         "json path not resulting in array or number should lead to error",
+			jsonResponse: []byte(`{"metric.value":5}`),
+			jsonPath:     "$['invalid.metric.values']",
+			err:          errors.New("unexpected json: expected single numeric or array value"),
+		},
+		{
+			name:         "invalid json should error",
+			jsonResponse: []byte(`{`),
+			jsonPath:     "$['invalid.metric.values']",
+			err:          errors.New("unexpected end of file"),
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			server := makeTestHTTPServer(t, tc.jsonResponse)
+			defer server.Close()
+			getter, err := NewJSONPathMetricsGetter(DefaultMetricsHTTPClient(), tc.aggregator, tc.jsonPath)
+			require.NoError(t, err)
+			url, err := url.Parse(fmt.Sprintf("%s/metrics", server.URL))
+			require.NoError(t, err)
+			metric, err := getter.GetMetric(*url)
+			if tc.err != nil {
+				require.Error(t, err)
+				require.Equal(t, tc.err.Error(), err.Error())
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.result, metric)
+		})
+	}
+}

pkg/collector/httpmetrics/pod_metrics.go

@@ -0,0 +1,118 @@
+package httpmetrics
+
+import (
+	"fmt"
+	"net/url"
+	"strconv"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+)
+
+type PodMetricsGetter interface {
+	GetMetric(pod *v1.Pod) (float64, error)
+}
+
+type PodMetricsJSONPathGetter struct {
+	scheme       string
+	path         string
+	rawQuery     string
+	port         int
+	metricGetter *JSONPathMetricsGetter
+}
+
+func (g PodMetricsJSONPathGetter) GetMetric(pod *v1.Pod) (float64, error) {
+	if pod.Status.PodIP == "" {
+		return 0, fmt.Errorf("pod %s/%s does not have a pod IP", pod.Namespace, pod.Name)
+	}
+	metricsURL := g.buildMetricsURL(pod.Status.PodIP)
+	return g.metricGetter.GetMetric(metricsURL)
+}
+
+func NewPodMetricsJSONPathGetter(config map[string]string) (*PodMetricsJSONPathGetter, error) {
+	getter := PodMetricsJSONPathGetter{}
+	var (
+		jsonPath   string
+		aggregator AggregatorFunc
+		err        error
+	)
+
+	if v, ok := config["json-key"]; ok {
+		jsonPath = v
+	}
+
+	if v, ok := config["scheme"]; ok {
+		getter.scheme = v
+	}
+
+	if v, ok := config["path"]; ok {
+		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 {
+			return nil, err
+		}
+		getter.port = n
+	}
+
+	if v, ok := config["aggregator"]; ok {
+		aggregator, err = ParseAggregator(v)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	requestTimeout := DefaultRequestTimeout
+	connectTimeout := DefaultConnectTimeout
+
+	if v, ok := config["request-timeout"]; ok {
+		d, err := time.ParseDuration(v)
+		if err != nil {
+			return nil, err
+		}
+		if d < 0 {
+			return nil, fmt.Errorf("Invalid request-timeout config value: %s", v)
+		}
+		requestTimeout = d
+	}
+
+	if v, ok := config["connect-timeout"]; ok {
+		d, err := time.ParseDuration(v)
+		if err != nil {
+			return nil, err
+		}
+		if d < 0 {
+			return nil, fmt.Errorf("Invalid connect-timeout config value: %s", v)
+		}
+		connectTimeout = d
+	}
+
+	jsonPathGetter, err := NewJSONPathMetricsGetter(CustomMetricsHTTPClient(requestTimeout, connectTimeout), aggregator, jsonPath)
+	if err != nil {
+		return nil, err
+	}
+	getter.metricGetter = jsonPathGetter
+	return &getter, nil
+}
+
+// buildMetricsURL will build the full URL needed to hit the pod metric endpoint.
+func (g *PodMetricsJSONPathGetter) 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,
+	}
+}

pkg/collector/httpmetrics/pod_metrics_test.go

@@ -0,0 +1,190 @@
+package httpmetrics
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func compareMetricsGetter(t *testing.T, first, second *PodMetricsJSONPathGetter) {
+	require.Equal(t, first.metricGetter.jsonPath, second.metricGetter.jsonPath)
+	require.Equal(t, first.scheme, second.scheme)
+	require.Equal(t, first.path, second.path)
+	require.Equal(t, first.port, second.port)
+}
+
+func TestNewPodJSONPathMetricsGetter(t *testing.T) {
+	configNoAggregator := map[string]string{
+		"json-key": "$.value",
+		"scheme":   "http",
+		"path":     "/metrics",
+		"port":     "9090",
+	}
+	getterNoAggregator, err1 := NewPodMetricsJSONPathGetter(configNoAggregator)
+
+	require.NoError(t, err1)
+	compareMetricsGetter(t, &PodMetricsJSONPathGetter{
+		metricGetter: &JSONPathMetricsGetter{jsonPath: configNoAggregator["json-key"]},
+		scheme:       "http",
+		path:         "/metrics",
+		port:         9090,
+	}, getterNoAggregator)
+
+	configAggregator := map[string]string{
+		"json-key":   "$.values",
+		"scheme":     "http",
+		"path":       "/metrics",
+		"port":       "9090",
+		"aggregator": "avg",
+	}
+	getterAggregator, err2 := NewPodMetricsJSONPathGetter(configAggregator)
+
+	require.NoError(t, err2)
+	compareMetricsGetter(t, &PodMetricsJSONPathGetter{
+		metricGetter: &JSONPathMetricsGetter{jsonPath: configAggregator["json-key"], aggregator: Average},
+		scheme:       "http",
+		path:         "/metrics",
+		port:         9090,
+	}, getterAggregator)
+
+	configErrorJSONPath := map[string]string{
+		"json-key": "{}",
+		"scheme":   "http",
+		"path":     "/metrics",
+		"port":     "9090",
+	}
+
+	_, err3 := NewPodMetricsJSONPathGetter(configErrorJSONPath)
+	require.Error(t, err3)
+
+	configErrorPort := map[string]string{
+		"json-key": "$.values",
+		"scheme":   "http",
+		"path":     "/metrics",
+		"port":     "a9090",
+	}
+
+	_, err4 := NewPodMetricsJSONPathGetter(configErrorPort)
+	require.Error(t, err4)
+
+	configWithRawQuery := map[string]string{
+		"json-key":  "$.values",
+		"scheme":    "http",
+		"path":      "/metrics",
+		"port":      "9090",
+		"raw-query": "foo=bar&baz=bop",
+	}
+	getterWithRawQuery, err5 := NewPodMetricsJSONPathGetter(configWithRawQuery)
+
+	require.NoError(t, err5)
+	compareMetricsGetter(t, &PodMetricsJSONPathGetter{
+		metricGetter: &JSONPathMetricsGetter{jsonPath: configWithRawQuery["json-key"]},
+		scheme:       "http",
+		path:         "/metrics",
+		port:         9090,
+		rawQuery:     "foo=bar&baz=bop",
+	}, getterWithRawQuery)
+}
+
+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,
+	}
+	getterWithRawQuery, err1 := NewPodMetricsJSONPathGetter(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,
+	}
+	getterWithNoQuery, err3 := NewPodMetricsJSONPathGetter(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)
+}
+
+func TestCustomTimeouts(t *testing.T) {
+	scheme := "http"
+	port := "9090"
+	path := "/v1/test/"
+
+	// Test no custom options results in default timeouts
+	defaultConfig := map[string]string{
+		"json-key": "$.value",
+		"scheme":   scheme,
+		"path":     path,
+		"port":     port,
+	}
+	defaultTime := time.Duration(15000) * time.Millisecond
+
+	defaultGetter, err1 := NewPodMetricsJSONPathGetter(defaultConfig)
+	require.NoError(t, err1)
+	require.Equal(t, defaultGetter.metricGetter.client.Timeout, defaultTime)
+
+	// Test with custom request timeout
+	configWithRequestTimeout := map[string]string{
+		"json-key":        "$.value",
+		"scheme":          scheme,
+		"path":            path,
+		"port":            port,
+		"request-timeout": "978ms",
+	}
+	exectedTimeout := time.Duration(978) * time.Millisecond
+	customRequestGetter, err2 := NewPodMetricsJSONPathGetter(configWithRequestTimeout)
+	require.NoError(t, err2)
+	require.Equal(t, customRequestGetter.metricGetter.client.Timeout, exectedTimeout)
+
+	// Test with custom connect timeout. Unfortunately, it seems there's no way to access the
+	// connect timeout of the client struct to actually verify it's set :/
+	configWithConnectTimeout := map[string]string{
+		"json-key":        "$.value",
+		"scheme":          scheme,
+		"path":            path,
+		"port":            port,
+		"connect-timeout": "512ms",
+	}
+	_, err3 := NewPodMetricsJSONPathGetter(configWithConnectTimeout)
+	require.NoError(t, err3)
+
+	configWithInvalidTimeout := map[string]string{
+		"json-key":        "$.value",
+		"scheme":          scheme,
+		"path":            path,
+		"port":            port,
+		"request-timeout": "-256ms",
+	}
+	_, err4 := NewPodMetricsJSONPathGetter(configWithInvalidTimeout)
+	require.Error(t, err4)
+
+	configWithInvalidTimeout = map[string]string{
+		"json-key":        "$.value",
+		"scheme":          scheme,
+		"path":            path,
+		"port":            port,
+		"connect-timeout": "-256ms",
+	}
+	_, err5 := NewPodMetricsJSONPathGetter(configWithInvalidTimeout)
+	require.Error(t, err5)
+}

pkg/collector/influxdb_collector.go

@@ -0,0 +1,152 @@
+package collector
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/influxdata/influxdb-client-go"
+	"k8s.io/api/autoscaling/v2beta2"
+	autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/metrics/pkg/apis/external_metrics"
+)
+
+const (
+	InfluxDBMetricName        = "flux-query"
+	influxDBAddressKey        = "address"
+	influxDBTokenKey          = "token"
+	influxDBOrgKey            = "org"
+	influxDBQueryNameLabelKey = "query-name"
+)
+
+type InfluxDBCollectorPlugin struct {
+	kubeClient kubernetes.Interface
+	address    string
+	token      string
+	org        string
+}
+
+func NewInfluxDBCollectorPlugin(client kubernetes.Interface, address, token, org string) (*InfluxDBCollectorPlugin, error) {
+	return &InfluxDBCollectorPlugin{
+		kubeClient: client,
+		address:    address,
+		token:      token,
+		org:        org,
+	}, nil
+}
+
+func (p *InfluxDBCollectorPlugin) NewCollector(hpa *v2beta2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
+	return NewInfluxDBCollector(p.address, p.token, p.org, config, interval)
+}
+
+type InfluxDBCollector struct {
+	address string
+	token   string
+	org     string
+
+	influxDBClient *influxdb.Client
+	interval       time.Duration
+	metric         autoscalingv2.MetricIdentifier
+	metricType     autoscalingv2.MetricSourceType
+	query          string
+}
+
+func NewInfluxDBCollector(address string, token string, org string, config *MetricConfig, interval time.Duration) (*InfluxDBCollector, error) {
+	collector := &InfluxDBCollector{
+		interval:   interval,
+		metric:     config.Metric,
+		metricType: config.Type,
+	}
+	switch configType := config.Type; configType {
+	case autoscalingv2.ObjectMetricSourceType:
+		return nil, fmt.Errorf("InfluxDB does not support object, but only external custom metrics")
+	case autoscalingv2.ExternalMetricSourceType:
+		// `metricSelector` is flattened into the MetricConfig.Config.
+		queryName, ok := config.Config[influxDBQueryNameLabelKey]
+		if !ok {
+			return nil, fmt.Errorf("selector for Flux query is not specified, "+
+				"please add metricSelector.matchLabels.%s: <...> to .yml description", influxDBQueryNameLabelKey)
+		}
+		if query, ok := config.Config[queryName]; ok {
+			// TODO(affo): validate the query once this is done:
+			//  https://github.com/influxdata/influxdb-client-go/issues/73.
+			collector.query = query
+		} else {
+			return nil, fmt.Errorf("no Flux query defined for metric \"%s\"", config.Metric.Name)
+		}
+	default:
+		return nil, fmt.Errorf("unknown metric type: %v", configType)
+	}
+	// Use custom InfluxDB config if defined in HPA annotation.
+	if v, ok := config.Config[influxDBAddressKey]; ok {
+		address = v
+	}
+	if v, ok := config.Config[influxDBTokenKey]; ok {
+		token = v
+	}
+	if v, ok := config.Config[influxDBOrgKey]; ok {
+		org = v
+	}
+	influxDbClient, err := influxdb.New(address, token)
+	if err != nil {
+		return nil, err
+	}
+	collector.address = address
+	collector.token = token
+	collector.org = org
+	collector.influxDBClient = influxDbClient
+	return collector, nil
+}
+
+// queryResult is for unmarshaling the result from InfluxDB.
+// The FluxQuery should make it so that the resulting table contains the column "metricvalue".
+type queryResult struct {
+	MetricValue float64
+}
+
+// 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.org)
+	if err != nil {
+		return resource.Quantity{}, err
+	}
+	defer res.Close()
+	// Keeping just the first result.
+	if res.Next() {
+		qr := queryResult{}
+		if err := res.Unmarshal(&qr); err != nil {
+			return resource.Quantity{}, fmt.Errorf("error in unmarshaling query result: %v", err)
+		}
+		return *resource.NewMilliQuantity(int64(qr.MetricValue*1000), resource.DecimalSI), nil
+	}
+	if err := res.Err; err != nil {
+		return resource.Quantity{}, fmt.Errorf("error in query result: %v", err)
+	}
+	return resource.Quantity{}, fmt.Errorf("empty result returned")
+}
+
+func (c *InfluxDBCollector) GetMetrics() ([]CollectedMetric, error) {
+	v, err := c.getValue()
+	if err != nil {
+		return nil, err
+	}
+	cm := CollectedMetric{
+		Type: c.metricType,
+		External: external_metrics.ExternalMetricValue{
+			MetricName:   c.metric.Name,
+			MetricLabels: c.metric.Selector.MatchLabels,
+			Timestamp: metav1.Time{
+				Time: time.Now().UTC(),
+			},
+			Value: v,
+		},
+	}
+	return []CollectedMetric{cm}, nil
+}
+
+func (c *InfluxDBCollector) Interval() time.Duration {
+	return c.interval
+}

pkg/collector/influxdb_collector_test.go

@@ -0,0 +1,155 @@
+package collector
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"k8s.io/api/autoscaling/v2beta2"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestInfluxDBCollector_New(t *testing.T) {
+	t.Run("simple", func(t *testing.T) {
+		m := &MetricConfig{
+			MetricTypeName: MetricTypeName{
+				Type: v2beta2.ExternalMetricSourceType,
+				Metric: v2beta2.MetricIdentifier{
+					Name: "flux-query",
+					// This is actually useless, because the selector should be flattened in Config when parsing.
+					Selector: &v1.LabelSelector{
+						MatchLabels: map[string]string{
+							"query-name": "range2m",
+						},
+					},
+				},
+			},
+			CollectorName: "influxdb",
+			Config: map[string]string{
+				"range1m":    `from(bucket: "?") |> range(start: -1m)`,
+				"range2m":    `from(bucket: "?") |> range(start: -2m)`,
+				"range3m":    `from(bucket: "?") |> range(start: -3m)`,
+				"query-name": "range2m",
+			},
+		}
+		c, err := NewInfluxDBCollector("http://localhost:9999", "secret", "deadbeef", m, time.Second)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		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 {
+			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
+		}
+		if got, want := c.token, "secret"; want != got {
+			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
+		}
+		if got, want := c.query, `from(bucket: "?") |> range(start: -2m)`; want != got {
+			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
+		}
+	})
+	t.Run("override params", func(t *testing.T) {
+		m := &MetricConfig{
+			MetricTypeName: MetricTypeName{
+				Type: v2beta2.ExternalMetricSourceType,
+				Metric: v2beta2.MetricIdentifier{
+					Name: "flux-query",
+					Selector: &v1.LabelSelector{
+						MatchLabels: map[string]string{
+							"query-name": "range2m",
+						},
+					},
+				},
+			},
+			CollectorName: "influxdb",
+			Config: map[string]string{
+				"range1m":    `from(bucket: "?") |> range(start: -1m)`,
+				"range2m":    `from(bucket: "?") |> range(start: -2m)`,
+				"range3m":    `from(bucket: "?") |> range(start: -3m)`,
+				"address":    "http://localhost:9999",
+				"token":      "sEcr3TT0ken",
+				"org":        "deadbeef1234",
+				"query-name": "range3m",
+			},
+		}
+		c, err := NewInfluxDBCollector("http://localhost:8888", "secret", "deadbeef", m, time.Second)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		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 {
+			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
+		}
+		if got, want := c.token, "sEcr3TT0ken"; want != got {
+			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
+		}
+		if got, want := c.query, `from(bucket: "?") |> range(start: -3m)`; want != got {
+			t.Errorf("unexpected value -want/+got:\n\t-%s\n\t+%s", want, got)
+		}
+	})
+	// Errors.
+	for _, tc := range []struct {
+		name            string
+		mTypeName       MetricTypeName
+		config          map[string]string
+		errorStartsWith string
+	}{
+		{
+			name: "object metric",
+			mTypeName: MetricTypeName{
+				Type: v2beta2.ObjectMetricSourceType,
+			},
+			errorStartsWith: "InfluxDB does not support object",
+		},
+		{
+			name: "no selector",
+			mTypeName: MetricTypeName{
+				Type: v2beta2.ExternalMetricSourceType,
+				Metric: v2beta2.MetricIdentifier{
+					Name: "flux-query",
+				},
+			},
+			//  The selector should be flattened into the config by the parsing step, but it isn't.
+			config: map[string]string{
+				"range1m": `from(bucket: "?") |> range(start: -1m)`,
+				"range2m": `from(bucket: "?") |> range(start: -2m)`,
+				"range3m": `from(bucket: "?") |> range(start: -3m)`,
+			},
+			errorStartsWith: "selector for Flux query is not specified",
+		},
+		{
+			name: "referencing non-existing query",
+			mTypeName: MetricTypeName{
+				Type: v2beta2.ExternalMetricSourceType,
+				Metric: v2beta2.MetricIdentifier{
+					Name: "flux-query",
+				},
+			},
+			config: map[string]string{
+				"range1m":    `from(bucket: "?") |> range(start: -1m)`,
+				"range2m":    `from(bucket: "?") |> range(start: -2m)`,
+				"range3m":    `from(bucket: "?") |> range(start: -3m)`,
+				"query-name": "rangeXm",
+			},
+			errorStartsWith: "no Flux query defined for metric",
+		},
+	} {
+		t.Run("error - "+tc.name, func(t *testing.T) {
+			m := &MetricConfig{
+				MetricTypeName: tc.mTypeName,
+				CollectorName:  "influxdb",
+				Config:         tc.config,
+			}
+			_, err := NewInfluxDBCollector("http://localhost:9999", "secret", "deadbeef", m, time.Second)
+			if err == nil {
+				t.Fatal("expected error got none")
+			}
+			if want, got := tc.errorStartsWith, err.Error(); !strings.HasPrefix(got, want) {
+				t.Fatalf("%s should start with %s", got, want)
+			}
+		})
+	}
+}

pkg/collector/json_path_collector.go

@@ -1,133 +0,0 @@
-package collector
-
-import (
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"net/http"
-	"net/url"
-	"strconv"
-	"time"
-
-	"github.com/oliveagle/jsonpath"
-	corev1 "k8s.io/api/core/v1"
-)
-
-// JSONPathMetricsGetter is a metrics getter which looks up pod metrics by
-// querying the pods metrics endpoint and lookup the metric value as defined by
-// the json path query.
-type JSONPathMetricsGetter struct {
-	jsonPath *jsonpath.Compiled
-	scheme   string
-	path     string
-	port     int
-}
-
-// NewJSONPathMetricsGetter initializes a new JSONPathMetricsGetter.
-func NewJSONPathMetricsGetter(config map[string]string) (*JSONPathMetricsGetter, error) {
-	getter := &JSONPathMetricsGetter{}
-
-	if v, ok := config["json-key"]; ok {
-		pat, err := jsonpath.Compile(v)
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse json path definition: %v", err)
-		}
-
-		getter.jsonPath = pat
-	}
-
-	if v, ok := config["scheme"]; ok {
-		getter.scheme = v
-	}
-
-	if v, ok := config["path"]; ok {
-		getter.path = v
-	}
-
-	if v, ok := config["port"]; ok {
-		n, err := strconv.Atoi(v)
-		if err != nil {
-			return nil, err
-		}
-		getter.port = n
-	}
-
-	return getter, nil
-}
-
-// GetMetric gets metric from pod by fetching json metrics from the pods metric
-// endpoint and extracting the desired value using the specified json path
-// query.
-func (g *JSONPathMetricsGetter) GetMetric(pod *corev1.Pod) (float64, error) {
-	data, err := getPodMetrics(pod, g.scheme, g.path, g.port)
-	if err != nil {
-		return 0, err
-	}
-
-	// parse data
-	var jsonData interface{}
-	err = json.Unmarshal(data, &jsonData)
-	if err != nil {
-		return 0, err
-	}
-
-	res, err := g.jsonPath.Lookup(jsonData)
-	if err != nil {
-		return 0, err
-	}
-
-	switch res := res.(type) {
-	case int:
-		return float64(res), nil
-	case float32:
-		return float64(res), nil
-	case float64:
-		return res, nil
-	default:
-		return 0, fmt.Errorf("unsupported type %T", res)
-	}
-}
-
-// getPodMetrics returns the content of the pods metrics endpoint.
-func getPodMetrics(pod *corev1.Pod, scheme, path string, port int) ([]byte, error) {
-	if pod.Status.PodIP == "" {
-		return nil, fmt.Errorf("pod %s/%s does not have a pod IP", pod.Namespace, pod.Namespace)
-	}
-
-	httpClient := &http.Client{
-		Timeout:   15 * time.Second,
-		Transport: &http.Transport{},
-	}
-
-	if scheme == "" {
-		scheme = "http"
-	}
-
-	metricsURL := url.URL{
-		Scheme: scheme,
-		Host:   fmt.Sprintf("%s:%d", pod.Status.PodIP, port),
-		Path:   path,
-	}
-
-	request, err := http.NewRequest(http.MethodGet, metricsURL.String(), nil)
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := httpClient.Do(request)
-	if err != nil {
-		return nil, err
-	}
-	defer resp.Body.Close()
-
-	data, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return nil, err
-	}
-
-	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("unsuccessful response: %s", resp.Status)
-	}
-
-	return data, nil
-}