Support networking.k8s.io/v1beta1 Ingresses

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

.golangci.yml

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

.travis.yml

@@ -5,7 +5,7 @@ go:
   - "1.13.x"
 
 env:
-- GO111MODULE=on GOLANGCI_RELEASE="v1.21.0"
+- GO111MODULE=on GOLANGCI_RELEASE="v1.16.0"
 
 before_install:
   - GO111MODULE=off go get github.com/mattn/goveralls
@@ -13,8 +13,8 @@ before_install:
   - 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

Makefile

@@ -19,8 +19,7 @@ test:
 	go test -v $(GOPKGS)
 
 check:
-	go mod download
-	golangci-lint run --timeout=2m ./...
+	golangci-lint run ./...
 
 build.local: build/$(BINARY)
 build.linux: build/linux/$(BINARY)

SECURITY.md

@@ -1,8 +1,7 @@
 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 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:
+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:
 
-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.
+Email: Tech-Security@zalando.de
+OR
+Submit your vulnerability report through our bug bounty program at: https://hackerone.com/zalando

example/deploy/hpa.yaml

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

go.mod

@@ -2,31 +2,32 @@ module github.com/zalando-incubator/kube-metrics-adapter
 
 require (
 	github.com/NYTimes/gziphandler v1.0.1 // indirect
-	github.com/aws/aws-sdk-go v1.30.2
+	github.com/aws/aws-sdk-go v1.16.6
+	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
-	github.com/influxdata/influxdb-client-go v0.1.5
-	github.com/kubernetes-incubator/custom-metrics-apiserver v0.0.0-20200323093244-5046ce1afe6b
-	github.com/lib/pq v1.2.0 // indirect
-	github.com/mattn/go-colorable v0.1.4 // indirect
+	github.com/grpc-ecosystem/grpc-gateway v1.4.1 // indirect
+	github.com/imdario/mergo v0.3.6 // indirect
+	github.com/inconshreveable/mousetrap v1.0.0 // indirect
+	github.com/kubernetes-incubator/custom-metrics-apiserver v0.0.0-20190918110929-3d9be26a50eb
 	github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852
-	github.com/onsi/ginkgo v1.11.0 // indirect
-	github.com/onsi/gomega v1.8.1 // indirect
-	github.com/prometheus/client_golang v1.5.1
-	github.com/prometheus/common v0.9.1
-	github.com/sirupsen/logrus v1.5.0
-	github.com/spf13/cobra v0.0.7
-	github.com/stretchr/testify v1.5.1
+	github.com/prometheus/client_golang v0.9.2
+	github.com/prometheus/common v0.0.0-20181126121408-4724e9255275
+	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/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/crypto v0.0.0-20200220183623-bac4c82f6975 // indirect
 	golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45
-	golang.org/x/tools v0.0.0-20200204192400-7124308813f3 // indirect
-	honnef.co/go/tools v0.0.1-2020.1.3 // indirect
-	k8s.io/api v0.17.3
-	k8s.io/apimachinery v0.17.4
-	k8s.io/client-go v0.17.3
-	k8s.io/component-base v0.17.3
-	k8s.io/klog v1.0.0
-	k8s.io/metrics v0.17.3
+	gopkg.in/inf.v0 v0.9.1 // indirect
+	k8s.io/api v0.0.0-20190918155943-95b840bb6a1f
+	k8s.io/apimachinery v0.0.0-20190913080033-27d36303b655
+	k8s.io/apiserver v0.0.0-20190918160949-bfa5e2e684ad // indirect
+	k8s.io/client-go v0.0.0-20190918160344-1fbdaa4c8d90
+	k8s.io/component-base v0.0.0-20190918160511-547f6c5d7090
+	k8s.io/klog v0.4.0
+	k8s.io/metrics v0.0.0-20190226180357-f3f09b9076d1
 )
 
 go 1.13

main.go

@@ -18,7 +18,6 @@ package main
 
 import (
 	"flag"
-	_ "net/http/pprof"
 	"os"
 	"runtime"
 

pkg/annotations/parser_test.go

@@ -65,42 +65,6 @@ 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/collector.go

@@ -46,14 +46,6 @@ 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
@@ -147,7 +139,7 @@ func (c *CollectorFactory) NewCollector(hpa *autoscalingv2.HorizontalPodAutoscal
 		}
 	}
 
-	return nil, &PluginNotFoundError{metricTypeName: config.MetricTypeName}
+	return nil, fmt.Errorf("no plugin found for %s", config.MetricTypeName)
 }
 
 type MetricTypeName struct {
@@ -173,7 +165,6 @@ 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.
@@ -215,15 +206,12 @@ func ParseHPAMetrics(hpa *autoscalingv2.HorizontalPodAutoscaler) ([]*MetricConfi
 			MetricTypeName:  typeName,
 			ObjectReference: ref,
 			Config:          map[string]string{},
-			MetricSpec:      metric,
 		}
 
 		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
-			}
+			config.Config = metric.External.Metric.Selector.MatchLabels
 		}
 
 		annotationConfigs, present := parser.GetAnnotationConfig(typeName.Metric.Name, typeName.Type)

pkg/collector/http_collector.go

@@ -1,103 +0,0 @@
-package collector
-
-import (
-	"fmt"
-	"net/url"
-	"time"
-
-	"github.com/zalando-incubator/kube-metrics-adapter/pkg/collector/httpmetrics"
-
-	"github.com/oliveagle/jsonpath"
-	"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, err := jsonpath.Compile(value)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse json path: %v", err)
-	}
-	collector.jsonPath = jsonPath
-	if value, ok = config.Config[HTTPEndpointAnnotationKey]; !ok {
-		return nil, fmt.Errorf("config value %s not found", HTTPEndpointAnnotationKey)
-	}
-	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
-		}
-	}
-	collector.metricsGetter = httpmetrics.NewJSONPathMetricsGetter(httpmetrics.DefaultMetricsHTTPClient(), aggFunc, jsonPath)
-	return collector, nil
-}
-
-type HTTPCollector struct {
-	endpoint      *url.URL
-	jsonPath      *jsonpath.Compiled
-	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

@@ -1,94 +0,0 @@
-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

@@ -1,65 +0,0 @@
-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

@@ -1,57 +0,0 @@
-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

@@ -1,129 +0,0 @@
-package httpmetrics
-
-import (
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"net"
-	"net/http"
-	"net/url"
-	"time"
-
-	"github.com/oliveagle/jsonpath"
-)
-
-// 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
-	aggregator AggregatorFunc
-	client     *http.Client
-}
-
-// NewJSONPathMetricsGetter initializes a new JSONPathMetricsGetter.
-func NewJSONPathMetricsGetter(httpClient *http.Client, aggregatorFunc AggregatorFunc, compiledPath *jsonpath.Compiled) *JSONPathMetricsGetter {
-	return &JSONPathMetricsGetter{client: httpClient, aggregator: aggregatorFunc, jsonPath: compiledPath}
-}
-
-func DefaultMetricsHTTPClient() *http.Client {
-	client := &http.Client{
-		Transport: &http.Transport{
-			DialContext: (&net.Dialer{
-				Timeout: 15 * time.Second,
-			}).DialContext,
-			MaxIdleConns:          50,
-			IdleConnTimeout:       90 * time.Second,
-			ExpectContinueTimeout: 1 * time.Second,
-		},
-		Timeout: 15 * time.Second,
-	}
-	return client
-}
-
-// 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
-	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
-	case []interface{}:
-		if g.aggregator == nil {
-			return 0, fmt.Errorf("no aggregator function has been specified")
-		}
-		s, err := castSlice(res)
-		if err != nil {
-			return 0, err
-		}
-		return g.aggregator(s...), nil
-	default:
-		return 0, fmt.Errorf("unsupported type %T", res)
-	}
-}
-
-// castSlice takes a slice of interface and returns a slice of float64 if all
-// values in slice were castable, else returns an error
-func castSlice(in []interface{}) ([]float64, error) {
-	var out []float64
-
-	for _, v := range in {
-		switch v := v.(type) {
-		case int:
-			out = append(out, float64(v))
-		case float32:
-			out = append(out, float64(v))
-		case float64:
-			out = append(out, v)
-		default:
-			return nil, fmt.Errorf("slice was returned by JSONPath, but value inside is unsupported: %T", v)
-		}
-	}
-
-	return out, nil
-}
-
-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

@@ -1,89 +0,0 @@
-package httpmetrics
-
-import (
-	"encoding/json"
-	"fmt"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
-	"testing"
-
-	"github.com/oliveagle/jsonpath"
-	"github.com/stretchr/testify/require"
-)
-
-func TestCastSlice(t *testing.T) {
-	res1, err1 := castSlice([]interface{}{1, 2, 3})
-	require.NoError(t, err1)
-	require.Equal(t, []float64{1.0, 2.0, 3.0}, res1)
-
-	res2, err2 := castSlice([]interface{}{float32(1.0), float32(2.0), float32(3.0)})
-	require.NoError(t, err2)
-	require.Equal(t, []float64{1.0, 2.0, 3.0}, res2)
-
-	res3, err3 := castSlice([]interface{}{float64(1.0), float64(2.0), float64(3.0)})
-	require.NoError(t, err3)
-	require.Equal(t, []float64{1.0, 2.0, 3.0}, res3)
-
-	res4, err4 := castSlice([]interface{}{1, 2, "some string"})
-	require.Errorf(t, err4, "slice was returned by JSONPath, but value inside is unsupported: %T", "string")
-	require.Equal(t, []float64(nil), res4)
-}
-
-type testValueResponse struct {
-	Value int64 `json:"value"`
-}
-
-type testValueArrayResponse struct {
-	Value []int64 `json:"value"`
-}
-
-func makeTestHTTPServer(t *testing.T, values ...int64) *httptest.Server {
-	h := func(w http.ResponseWriter, r *http.Request) {
-		require.Equal(t, r.URL.Path, "/metrics")
-		w.Header().Set("Content-Type", "application/json")
-		var (
-			response []byte
-			err      error
-		)
-		if len(values) == 1 {
-			response, err = json.Marshal(testValueResponse{Value: values[0]})
-			require.NoError(t, err)
-		} else {
-			response, err = json.Marshal(testValueArrayResponse{Value: values})
-			require.NoError(t, err)
-		}
-		_, 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
-		input      []int64
-		output     float64
-		aggregator AggregatorFunc
-	}{
-		{
-			name:       "basic average",
-			input:      []int64{3, 4, 5},
-			output:     4,
-			aggregator: Average,
-		},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			server := makeTestHTTPServer(t, tc.input...)
-			defer server.Close()
-			path, err := jsonpath.Compile("$.value")
-			require.NoError(t, err)
-			getter := NewJSONPathMetricsGetter(DefaultMetricsHTTPClient(), tc.aggregator, path)
-			url, err := url.Parse(fmt.Sprintf("%s/metrics", server.URL))
-			require.NoError(t, err)
-			metric, err := getter.GetMetric(*url)
-			require.NoError(t, err)
-			require.Equal(t, tc.output, metric)
-		})
-	}
-}

pkg/collector/httpmetrics/pod_metrics.go

@@ -1,93 +0,0 @@
-package httpmetrics
-
-import (
-	"fmt"
-	"net/url"
-	"strconv"
-
-	"github.com/oliveagle/jsonpath"
-	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   *jsonpath.Compiled
-		aggregator AggregatorFunc
-		err        error
-	)
-
-	if v, ok := config["json-key"]; ok {
-		path, err := jsonpath.Compile(v)
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse json path definition: %v", err)
-		}
-
-		jsonPath = path
-	}
-
-	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
-		}
-	}
-	getter.metricGetter = NewJSONPathMetricsGetter(DefaultMetricsHTTPClient(), aggregator, jsonPath)
-	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

@@ -1,133 +0,0 @@
-package httpmetrics
-
-import (
-	"fmt"
-	"testing"
-
-	"github.com/oliveagle/jsonpath"
-	"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",
-	}
-	jpath1, _ := jsonpath.Compile(configNoAggregator["json-key"])
-	getterNoAggregator, err1 := NewPodMetricsJSONPathGetter(configNoAggregator)
-
-	require.NoError(t, err1)
-	compareMetricsGetter(t, &PodMetricsJSONPathGetter{
-		metricGetter: &JSONPathMetricsGetter{jsonPath: jpath1},
-		scheme:       "http",
-		path:         "/metrics",
-		port:         9090,
-	}, getterNoAggregator)
-
-	configAggregator := map[string]string{
-		"json-key":   "$.values",
-		"scheme":     "http",
-		"path":       "/metrics",
-		"port":       "9090",
-		"aggregator": "avg",
-	}
-	jpath2, _ := jsonpath.Compile(configAggregator["json-key"])
-	getterAggregator, err2 := NewPodMetricsJSONPathGetter(configAggregator)
-
-	require.NoError(t, err2)
-	compareMetricsGetter(t, &PodMetricsJSONPathGetter{
-		metricGetter: &JSONPathMetricsGetter{jsonPath: jpath2, 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",
-	}
-	jpath5, _ := jsonpath.Compile(configWithRawQuery["json-key"])
-	getterWithRawQuery, err5 := NewPodMetricsJSONPathGetter(configWithRawQuery)
-
-	require.NoError(t, err5)
-	compareMetricsGetter(t, &PodMetricsJSONPathGetter{
-		metricGetter: &JSONPathMetricsGetter{jsonPath: jpath5},
-		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,
-	}
-	_, err := jsonpath.Compile(configWithRawQuery["json-key"])
-	require.NoError(t, err)
-	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,
-	}
-	_, err2 := jsonpath.Compile(configWithNoQuery["json-key"])
-	require.NoError(t, err2)
-	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)
-}

pkg/collector/influxdb_collector.go

@@ -1,152 +0,0 @@
-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

@@ -1,155 +0,0 @@
-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

@@ -0,0 +1,133 @@
+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
+}

pkg/collector/max_collector.go

@@ -0,0 +1,67 @@
+package collector
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/resource"
+)
+
+// MaxWeightedCollector is a simple aggregator collector that returns the maximum value
+// of metrics from all collectors.
+type MaxWeightedCollector struct {
+	collectors []Collector
+	interval   time.Duration
+	weight     float64
+}
+
+// NewMaxWeightedCollector initializes a new MaxWeightedCollector.
+func NewMaxWeightedCollector(interval time.Duration, weight float64, collectors ...Collector) *MaxWeightedCollector {
+	return &MaxWeightedCollector{
+		collectors: collectors,
+		interval:   interval,
+		weight:     weight,
+	}
+}
+
+// GetMetrics gets metrics from all collectors and return the higest value.
+func (c *MaxWeightedCollector) GetMetrics() ([]CollectedMetric, error) {
+	errors := make([]error, 0)
+	collectedMetrics := make([]CollectedMetric, 0)
+	for _, collector := range c.collectors {
+		values, err := collector.GetMetrics()
+		if err != nil {
+			if _, ok := err.(NoResultError); ok {
+				errors = append(errors, err)
+				continue
+			}
+			return nil, err
+		}
+		collectedMetrics = append(collectedMetrics, values...)
+	}
+	if len(collectedMetrics) == 0 {
+		if len(errors) == 0 {
+			return nil, fmt.Errorf("no metrics collected, cannot determine max")
+		}
+		errorStrings := make([]string, len(errors))
+		for i, e := range errors {
+			errorStrings[i] = e.Error()
+		}
+		allErrors := strings.Join(errorStrings, ",")
+		return nil, fmt.Errorf("could not determine maximum due to errors: %s", allErrors)
+	}
+	max := collectedMetrics[0]
+	for _, value := range collectedMetrics {
+		if value.Custom.Value.MilliValue() > max.Custom.Value.MilliValue() {
+			max = value
+		}
+	}
+	max.Custom.Value = *resource.NewMilliQuantity(int64(c.weight*float64(max.Custom.Value.MilliValue())), resource.DecimalSI)
+	return []CollectedMetric{max}, nil
+}
+
+// Interval returns the interval at which the collector should run.
+func (c *MaxWeightedCollector) Interval() time.Duration {
+	return c.interval
+}

pkg/collector/max_collector_test.go

@@ -0,0 +1,99 @@
+package collector
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/metrics/pkg/apis/custom_metrics"
+)
+
+type dummyCollector struct {
+	value int64
+}
+
+func (c dummyCollector) Interval() time.Duration {
+	return time.Second
+}
+
+func (c dummyCollector) GetMetrics() ([]CollectedMetric, error) {
+	switch c.value {
+	case 0:
+		return nil, NoResultError{query: "invalid query"}
+	case -1:
+		return nil, fmt.Errorf("test error")
+	default:
+		quantity := resource.NewQuantity(c.value, resource.DecimalSI)
+		return []CollectedMetric{
+			{
+				Custom: custom_metrics.MetricValue{
+					Value: *quantity,
+				},
+			},
+		}, nil
+	}
+}
+
+func TestMaxCollector(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		values   []int64
+		expected int
+		weight   float64
+		errored  bool
+	}{
+		{
+			name:     "basic",
+			values:   []int64{100, 10, 9},
+			expected: 100,
+			weight:   1,
+			errored:  false,
+		},
+		{
+			name:     "weighted",
+			values:   []int64{100, 10, 9},
+			expected: 20,
+			weight:   0.2,
+			errored:  false,
+		},
+		{
+			name:    "with error",
+			values:  []int64{10, 9, -1},
+			weight:  0.5,
+			errored: true,
+		},
+		{
+			name:     "some invalid results",
+			values:   []int64{0, 1, 0, 10, 9},
+			expected: 5,
+			weight:   0.5,
+			errored:  false,
+		},
+		{
+			name:    "both invalid results and errors",
+			values:  []int64{0, 1, 0, -1, 10, 9},
+			weight:  0.5,
+			errored: true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			collectors := make([]Collector, len(tc.values))
+			for i, v := range tc.values {
+				collectors[i] = dummyCollector{value: v}
+			}
+			wc := NewMaxWeightedCollector(time.Second, tc.weight, collectors...)
+			metrics, err := wc.GetMetrics()
+			if tc.errored {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Len(t, metrics, 1)
+				require.EqualValues(t, tc.expected, metrics[0].Custom.Value.Value())
+			}
+
+		})
+
+	}
+}

pkg/collector/pod_collector.go

@@ -2,12 +2,11 @@ package collector
 
 import (
 	"fmt"
-	"net/http"
 	"time"
 
 	log "github.com/sirupsen/logrus"
-	"github.com/zalando-incubator/kube-metrics-adapter/pkg/collector/httpmetrics"
 	autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -31,14 +30,17 @@ func (p *PodCollectorPlugin) NewCollector(hpa *autoscalingv2.HorizontalPodAutosc
 
 type PodCollector struct {
 	client           kubernetes.Interface
-	Getter           httpmetrics.PodMetricsGetter
+	Getter           PodMetricsGetter
 	podLabelSelector *metav1.LabelSelector
 	namespace        string
 	metric           autoscalingv2.MetricIdentifier
 	metricType       autoscalingv2.MetricSourceType
 	interval         time.Duration
 	logger           *log.Entry
-	httpClient       *http.Client
+}
+
+type PodMetricsGetter interface {
+	GetMetric(pod *corev1.Pod) (float64, error)
 }
 
 func NewPodCollector(client kubernetes.Interface, hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (*PodCollector, error) {
@@ -58,11 +60,11 @@ func NewPodCollector(client kubernetes.Interface, hpa *autoscalingv2.HorizontalP
 		logger:           log.WithFields(log.Fields{"Collector": "Pod"}),
 	}
 
-	var getter httpmetrics.PodMetricsGetter
+	var getter PodMetricsGetter
 	switch config.CollectorName {
 	case "json-path":
 		var err error
-		getter, err = httpmetrics.NewPodMetricsJSONPathGetter(config.Config)
+		getter, err = NewJSONPathMetricsGetter(config.Config)
 		if err != nil {
 			return nil, err
 		}

pkg/collector/pod_collector_test.go

@@ -1,150 +0,0 @@
-package collector
-
-import (
-	"encoding/json"
-	"fmt"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-	appsv1 "k8s.io/api/apps/v1"
-	autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
-	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/kubernetes/fake"
-)
-
-const (
-	testNamespace         = "test-namespace"
-	applicationLabelName  = "application"
-	applicationLabelValue = "test-application"
-	testDeploymentName    = "test-application"
-	testInterval          = 10 * time.Second
-)
-
-func TestPodCollector(t *testing.T) {
-	for _, tc := range []struct {
-		name    string
-		metrics [][]int64
-		result  []int64
-	}{
-		{
-			name:    "simple",
-			metrics: [][]int64{{1}, {3}, {8}, {5}, {2}},
-			result:  []int64{1, 3, 8, 5, 2},
-		},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			client := fake.NewSimpleClientset()
-			plugin := NewPodCollectorPlugin(client)
-			makeTestDeployment(t, client)
-			host, port, metricsHandler := makeTestHTTPServer(t, tc.metrics)
-			makeTestPods(t, host, port, "test-metric", client, 5)
-			testHPA := makeTestHPA(t, client)
-			testConfig := makeTestConfig(port)
-			collector, err := plugin.NewCollector(testHPA, testConfig, testInterval)
-			require.NoError(t, err)
-			metrics, err := collector.GetMetrics()
-			require.NoError(t, err)
-			require.Equal(t, len(metrics), int(metricsHandler.calledCounter))
-			var values []int64
-			for _, m := range metrics {
-				values = append(values, m.Custom.Value.Value())
-			}
-			require.Equal(t, tc.result, values)
-		})
-	}
-}
-
-type testMetricResponse struct {
-	Values []int64 `json:"values"`
-}
-type testMetricsHandler struct {
-	values        [][]int64
-	calledCounter uint
-	t             *testing.T
-	metricsPath   string
-}
-
-func (h *testMetricsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	require.Equal(h.t, h.metricsPath, r.URL.Path)
-	require.Less(h.t, int(h.calledCounter), len(h.values))
-	response, err := json.Marshal(testMetricResponse{Values: h.values[h.calledCounter]})
-	require.NoError(h.t, err)
-	w.Header().Set("Content-Type", "application/json")
-	_, err = w.Write(response)
-	require.NoError(h.t, err)
-	h.calledCounter++
-}
-
-func makeTestHTTPServer(t *testing.T, values [][]int64) (string, string, *testMetricsHandler) {
-	metricsHandler := &testMetricsHandler{values: values, t: t, metricsPath: "/metrics"}
-	server := httptest.NewServer(metricsHandler)
-	url, err := url.Parse(server.URL)
-	require.NoError(t, err)
-	return url.Hostname(), url.Port(), metricsHandler
-}
-
-func makeTestConfig(port string) *MetricConfig {
-	return &MetricConfig{
-		CollectorName: "json-path",
-		Config:        map[string]string{"json-key": "$.values", "port": port, "path": "/metrics", "aggregator": "sum"},
-	}
-}
-
-func makeTestPods(t *testing.T, testServer string, metricName string, port string, client kubernetes.Interface, replicas int) {
-	for i := 0; i < replicas; i++ {
-		testPod := &corev1.Pod{
-			ObjectMeta: v1.ObjectMeta{
-				Name:   fmt.Sprintf("test-pod-%d", i),
-				Labels: map[string]string{applicationLabelName: applicationLabelValue},
-				Annotations: map[string]string{
-					fmt.Sprintf("metric-config.pods.%s.json-path/port", metricName): port,
-				},
-			},
-			Status: corev1.PodStatus{
-				PodIP: testServer,
-			},
-		}
-		_, err := client.CoreV1().Pods(testNamespace).Create(testPod)
-		require.NoError(t, err)
-	}
-}
-
-func makeTestDeployment(t *testing.T, client kubernetes.Interface) *appsv1.Deployment {
-	deployment := appsv1.Deployment{
-		ObjectMeta: v1.ObjectMeta{Name: testDeploymentName},
-		Spec: appsv1.DeploymentSpec{
-			Selector: &v1.LabelSelector{
-				MatchLabels: map[string]string{applicationLabelName: applicationLabelValue},
-			},
-		},
-	}
-	_, err := client.AppsV1().Deployments(testNamespace).Create(&deployment)
-	require.NoError(t, err)
-	return &deployment
-
-}
-
-func makeTestHPA(t *testing.T, client kubernetes.Interface) *autoscalingv2.HorizontalPodAutoscaler {
-	hpa := &autoscalingv2.HorizontalPodAutoscaler{
-		ObjectMeta: v1.ObjectMeta{
-			Name:      "test-hpa",
-			Namespace: testNamespace,
-		},
-		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
-			ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
-				Kind:       "Deployment",
-				Name:       testDeploymentName,
-				APIVersion: "apps/v1",
-			},
-		},
-	}
-	_, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers("test-namespace").Create(hpa)
-	require.NoError(t, err)
-	return hpa
-}

pkg/collector/prometheus_collector.go

@@ -3,7 +3,6 @@ package collector
 import (
 	"context"
 	"fmt"
-	"math"
 	"net/http"
 	"time"
 
@@ -128,7 +127,7 @@ func NewPrometheusCollector(client kubernetes.Interface, promAPI promv1.API, hpa
 
 func (c *PrometheusCollector) GetMetrics() ([]CollectedMetric, error) {
 	// TODO: use real context
-	value, _, err := c.promAPI.Query(context.Background(), c.query, time.Now().UTC())
+	value, err := c.promAPI.Query(context.Background(), c.query, time.Now().UTC())
 	if err != nil {
 		return nil, err
 	}
@@ -147,7 +146,7 @@ func (c *PrometheusCollector) GetMetrics() ([]CollectedMetric, error) {
 		sampleValue = scalar.Value
 	}
 
-	if math.IsNaN(float64(sampleValue)) {
+	if sampleValue.String() == "NaN" {
 		return nil, &NoResultError{query: c.query}
 	}
 

pkg/collector/skipper_collector.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"math"
-	"regexp"
 	"strings"
 	"time"
 
@@ -17,7 +16,7 @@ import (
 )
 
 const (
-	rpsQuery                  = `scalar(sum(rate(skipper_serve_host_duration_seconds_count{host=~"%s"}[1m])) * %.4f)`
+	rpsQuery                  = `scalar(sum(rate(skipper_serve_host_duration_seconds_count{host="%s"}[1m])))`
 	rpsMetricName             = "requests-per-second"
 	rpsMetricBackendSeparator = ","
 )
@@ -73,7 +72,8 @@ type SkipperCollector struct {
 }
 
 // NewSkipperCollector initializes a new SkipperCollector.
-func NewSkipperCollector(client kubernetes.Interface, plugin CollectorPlugin, hpa *autoscalingv2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration, backendAnnotations []string, backend string) (*SkipperCollector, error) {
+func NewSkipperCollector(client kubernetes.Interface, plugin CollectorPlugin, hpa *autoscalingv2.HorizontalPodAutoscaler,
+	config *MetricConfig, interval time.Duration, backendAnnotations []string, backend string) (*SkipperCollector, error) {
 	return &SkipperCollector{
 		client:             client,
 		objectReference:    config.ObjectReference,
@@ -125,36 +125,59 @@ func getWeights(ingressAnnotations map[string]string, backendAnnotations []strin
 
 // getCollector returns a collector for getting the metrics.
 func (c *SkipperCollector) getCollector() (Collector, error) {
-	ingress, err := c.client.ExtensionsV1beta1().Ingresses(c.objectReference.Namespace).Get(c.objectReference.Name, metav1.GetOptions{})
-	if err != nil {
-		return nil, err
+	var annotations map[string]string
+	var hosts []string
+
+	switch c.objectReference.APIVersion {
+	case "extensions/v1beta1":
+		ingress, err := c.client.ExtensionsV1beta1().Ingresses(c.objectReference.Namespace).Get(c.objectReference.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+		annotations = ingress.Annotations
+		for _, rule := range ingress.Spec.Rules {
+			hosts = append(hosts, rule.Host)
+		}
+	case "networking.k8s.io/v1beta1":
+		ingress, err := c.client.NetworkingV1beta1().Ingresses(c.objectReference.Namespace).Get(c.objectReference.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+		annotations = ingress.Annotations
+		for _, rule := range ingress.Spec.Rules {
+			hosts = append(hosts, rule.Host)
+		}
 	}
 
-	backendWeight, err := getWeights(ingress.Annotations, c.backendAnnotations, c.backend)
+	backendWeight, err := getWeights(annotations, c.backendAnnotations, c.backend)
 	if err != nil {
 		return nil, err
 	}
 	config := c.config
 
-	var escapedHostnames []string
-	for _, rule := range ingress.Spec.Rules {
-		escapedHostnames = append(escapedHostnames, regexp.QuoteMeta(strings.Replace(rule.Host, ".", "_", -1)))
+	var collector Collector
+	collectors := make([]Collector, 0, len(hosts))
+	for _, host := range hosts {
+		host := strings.Replace(host, ".", "_", -1)
+		config.Config = map[string]string{
+			"query": fmt.Sprintf(rpsQuery, host),
+		}
+
+		config.PerReplica = false // per replica is handled outside of the prometheus collector
+		collector, err := c.plugin.NewCollector(c.hpa, &config, c.interval)
+		if err != nil {
+			return nil, err
+		}
+
+		collectors = append(collectors, collector)
 	}
 
-	if len(escapedHostnames) == 0 {
+	if len(collectors) > 0 {
+		collector = NewMaxWeightedCollector(c.interval, backendWeight, collectors...)
+	} else {
 		return nil, fmt.Errorf("no hosts defined on ingress %s/%s, unable to create collector", c.objectReference.Namespace, c.objectReference.Name)
 	}
 
-	config.Config = map[string]string{
-		"query": fmt.Sprintf(rpsQuery, strings.Join(escapedHostnames, "|"), backendWeight),
-	}
-
-	config.PerReplica = false // per replica is handled outside of the prometheus collector
-	collector, err := c.plugin.NewCollector(c.hpa, &config, c.interval)
-	if err != nil {
-		return nil, err
-	}
-
 	return collector, nil
 }
 
@@ -174,27 +197,23 @@ func (c *SkipperCollector) GetMetrics() ([]CollectedMetric, error) {
 		return nil, fmt.Errorf("expected to only get one metric value, got %d", len(values))
 	}
 
-	value := values[0]
-
-	// For Kubernetes <v1.14 we have to fall back to manual average
-	if c.config.MetricSpec.Object.Target.AverageValue == nil {
-		// get current replicas for the targeted scale object. This is used to
-		// calculate an average metric instead of total.
-		// targetAverageValue will be available in Kubernetes v1.12
-		// https://github.com/kubernetes/kubernetes/pull/64097
-		replicas, err := targetRefReplicas(c.client, c.hpa)
-		if err != nil {
-			return nil, err
-		}
-
-		if replicas < 1 {
-			return nil, fmt.Errorf("unable to get average value for %d replicas", replicas)
-		}
-
-		avgValue := float64(value.Custom.Value.MilliValue()) / float64(replicas)
-		value.Custom.Value = *resource.NewMilliQuantity(int64(avgValue), resource.DecimalSI)
+	// get current replicas for the targeted scale object. This is used to
+	// calculate an average metric instead of total.
+	// targetAverageValue will be available in Kubernetes v1.12
+	// https://github.com/kubernetes/kubernetes/pull/64097
+	replicas, err := targetRefReplicas(c.client, c.hpa)
+	if err != nil {
+		return nil, err
 	}
 
+	if replicas < 1 {
+		return nil, fmt.Errorf("unable to get average value for %d replicas", replicas)
+	}
+
+	value := values[0]
+	avgValue := float64(value.Custom.Value.MilliValue()) / float64(replicas)
+	value.Custom.Value = *resource.NewMilliQuantity(int64(avgValue), resource.DecimalSI)
+
 	return []CollectedMetric{value}, nil
 }
 

pkg/provider/conversion.go

@@ -0,0 +1,247 @@
+package provider
+
+import (
+	autoscalingv2beta1 "k8s.io/api/autoscaling/v2beta1"
+	autoscaling "k8s.io/api/autoscaling/v2beta2"
+	core "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/conversion"
+)
+
+// from: https://github.com/kubernetes/kubernetes/blob/v1.14.4/pkg/apis/autoscaling/v2beta1/conversion.go
+
+func Convert_autoscaling_MetricTarget_To_v2beta1_CrossVersionObjectReference(in *autoscaling.MetricTarget, out *autoscalingv2beta1.CrossVersionObjectReference, s conversion.Scope) error {
+	return nil
+}
+
+func Convert_v2beta1_CrossVersionObjectReference_To_autoscaling_MetricTarget(in *autoscalingv2beta1.CrossVersionObjectReference, out *autoscaling.MetricTarget, s conversion.Scope) error {
+	return nil
+}
+
+func Convert_v2beta1_ResourceMetricStatus_To_autoscaling_ResourceMetricStatus(in *autoscalingv2beta1.ResourceMetricStatus, out *autoscaling.ResourceMetricStatus, s conversion.Scope) error {
+	out.Name = core.ResourceName(in.Name)
+	utilization := in.CurrentAverageUtilization
+	averageValue := in.CurrentAverageValue
+	out.Current = autoscaling.MetricValueStatus{
+		AverageValue:       &averageValue,
+		AverageUtilization: utilization,
+	}
+	return nil
+}
+
+func Convert_autoscaling_ResourceMetricStatus_To_v2beta1_ResourceMetricStatus(in *autoscaling.ResourceMetricStatus, out *autoscalingv2beta1.ResourceMetricStatus, s conversion.Scope) error {
+	out.Name = v1.ResourceName(in.Name)
+	out.CurrentAverageUtilization = in.Current.AverageUtilization
+	if in.Current.AverageValue != nil {
+		out.CurrentAverageValue = *in.Current.AverageValue
+	}
+	return nil
+}
+
+func Convert_v2beta1_ResourceMetricSource_To_autoscaling_ResourceMetricSource(in *autoscalingv2beta1.ResourceMetricSource, out *autoscaling.ResourceMetricSource, s conversion.Scope) error {
+	out.Name = core.ResourceName(in.Name)
+	utilization := in.TargetAverageUtilization
+	averageValue := in.TargetAverageValue
+
+	var metricType autoscaling.MetricTargetType
+	if utilization == nil {
+		metricType = autoscaling.AverageValueMetricType
+	} else {
+		metricType = autoscaling.UtilizationMetricType
+	}
+	out.Target = autoscaling.MetricTarget{
+		Type:               metricType,
+		AverageValue:       averageValue,
+		AverageUtilization: utilization,
+	}
+	return nil
+}
+
+func Convert_autoscaling_ResourceMetricSource_To_v2beta1_ResourceMetricSource(in *autoscaling.ResourceMetricSource, out *autoscalingv2beta1.ResourceMetricSource, s conversion.Scope) error {
+	out.Name = v1.ResourceName(in.Name)
+	out.TargetAverageUtilization = in.Target.AverageUtilization
+	out.TargetAverageValue = in.Target.AverageValue
+	return nil
+}
+
+func Convert_autoscaling_ExternalMetricSource_To_v2beta1_ExternalMetricSource(in *autoscaling.ExternalMetricSource, out *autoscalingv2beta1.ExternalMetricSource, s conversion.Scope) error {
+	out.MetricName = in.Metric.Name
+	out.TargetValue = in.Target.Value
+	out.TargetAverageValue = in.Target.AverageValue
+	out.MetricSelector = in.Metric.Selector
+	return nil
+}
+
+func Convert_v2beta1_ExternalMetricSource_To_autoscaling_ExternalMetricSource(in *autoscalingv2beta1.ExternalMetricSource, out *autoscaling.ExternalMetricSource, s conversion.Scope) error {
+	value := in.TargetValue
+	averageValue := in.TargetAverageValue
+
+	var metricType autoscaling.MetricTargetType
+	if value == nil {
+		metricType = autoscaling.AverageValueMetricType
+	} else {
+		metricType = autoscaling.ValueMetricType
+	}
+
+	out.Target = autoscaling.MetricTarget{
+		Type:         metricType,
+		Value:        value,
+		AverageValue: averageValue,
+	}
+
+	out.Metric = autoscaling.MetricIdentifier{
+		Name:     in.MetricName,
+		Selector: in.MetricSelector,
+	}
+	return nil
+}
+
+func Convert_autoscaling_ObjectMetricSource_To_v2beta1_ObjectMetricSource(in *autoscaling.ObjectMetricSource, out *autoscalingv2beta1.ObjectMetricSource, s conversion.Scope) error {
+	if in.Target.Value != nil {
+		out.TargetValue = *in.Target.Value
+	}
+	out.AverageValue = in.Target.AverageValue
+
+	out.Target = autoscalingv2beta1.CrossVersionObjectReference{
+		Kind:       in.DescribedObject.Kind,
+		Name:       in.DescribedObject.Name,
+		APIVersion: in.DescribedObject.APIVersion,
+	}
+	out.MetricName = in.Metric.Name
+	out.Selector = in.Metric.Selector
+
+	return nil
+}
+
+func Convert_v2beta1_ObjectMetricSource_To_autoscaling_ObjectMetricSource(in *autoscalingv2beta1.ObjectMetricSource, out *autoscaling.ObjectMetricSource, s conversion.Scope) error {
+	var metricType autoscaling.MetricTargetType
+	if in.AverageValue == nil {
+		metricType = autoscaling.ValueMetricType
+	} else {
+		metricType = autoscaling.AverageValueMetricType
+	}
+	out.Target = autoscaling.MetricTarget{
+		Type:         metricType,
+		Value:        &in.TargetValue,
+		AverageValue: in.AverageValue,
+	}
+	out.DescribedObject = autoscaling.CrossVersionObjectReference{
+		Kind:       in.Target.Kind,
+		Name:       in.Target.Name,
+		APIVersion: in.Target.APIVersion,
+	}
+	out.Metric = autoscaling.MetricIdentifier{
+		Name:     in.MetricName,
+		Selector: in.Selector,
+	}
+	return nil
+}
+
+func Convert_autoscaling_PodsMetricSource_To_v2beta1_PodsMetricSource(in *autoscaling.PodsMetricSource, out *autoscalingv2beta1.PodsMetricSource, s conversion.Scope) error {
+	if in.Target.AverageValue != nil {
+		targetAverageValue := *in.Target.AverageValue
+		out.TargetAverageValue = targetAverageValue
+	}
+
+	out.MetricName = in.Metric.Name
+	out.Selector = in.Metric.Selector
+
+	return nil
+}
+
+func Convert_v2beta1_PodsMetricSource_To_autoscaling_PodsMetricSource(in *autoscalingv2beta1.PodsMetricSource, out *autoscaling.PodsMetricSource, s conversion.Scope) error {
+	targetAverageValue := &in.TargetAverageValue
+	var metricType autoscaling.MetricTargetType
+	metricType = autoscaling.AverageValueMetricType
+
+	out.Target = autoscaling.MetricTarget{
+		Type:         metricType,
+		AverageValue: targetAverageValue,
+	}
+	out.Metric = autoscaling.MetricIdentifier{
+		Name:     in.MetricName,
+		Selector: in.Selector,
+	}
+	return nil
+}
+
+func Convert_autoscaling_ExternalMetricStatus_To_v2beta1_ExternalMetricStatus(in *autoscaling.ExternalMetricStatus, out *autoscalingv2beta1.ExternalMetricStatus, s conversion.Scope) error {
+	if &in.Current.AverageValue != nil {
+		out.CurrentAverageValue = in.Current.AverageValue
+	}
+	out.MetricName = in.Metric.Name
+	if in.Current.Value != nil {
+		out.CurrentValue = *in.Current.Value
+	}
+	out.MetricSelector = in.Metric.Selector
+	return nil
+}
+
+func Convert_v2beta1_ExternalMetricStatus_To_autoscaling_ExternalMetricStatus(in *autoscalingv2beta1.ExternalMetricStatus, out *autoscaling.ExternalMetricStatus, s conversion.Scope) error {
+	value := in.CurrentValue
+	averageValue := in.CurrentAverageValue
+	out.Current = autoscaling.MetricValueStatus{
+		Value:        &value,
+		AverageValue: averageValue,
+	}
+	out.Metric = autoscaling.MetricIdentifier{
+		Name:     in.MetricName,
+		Selector: in.MetricSelector,
+	}
+	return nil
+}
+
+func Convert_autoscaling_ObjectMetricStatus_To_v2beta1_ObjectMetricStatus(in *autoscaling.ObjectMetricStatus, out *autoscalingv2beta1.ObjectMetricStatus, s conversion.Scope) error {
+	if in.Current.Value != nil {
+		out.CurrentValue = *in.Current.Value
+	}
+	out.Target = autoscalingv2beta1.CrossVersionObjectReference{
+		Kind:       in.DescribedObject.Kind,
+		Name:       in.DescribedObject.Name,
+		APIVersion: in.DescribedObject.APIVersion,
+	}
+	out.MetricName = in.Metric.Name
+	out.Selector = in.Metric.Selector
+	if in.Current.AverageValue != nil {
+		currentAverageValue := *in.Current.AverageValue
+		out.AverageValue = &currentAverageValue
+	}
+	return nil
+}
+
+func Convert_v2beta1_ObjectMetricStatus_To_autoscaling_ObjectMetricStatus(in *autoscalingv2beta1.ObjectMetricStatus, out *autoscaling.ObjectMetricStatus, s conversion.Scope) error {
+	out.Current = autoscaling.MetricValueStatus{
+		Value:        &in.CurrentValue,
+		AverageValue: in.AverageValue,
+	}
+	out.DescribedObject = autoscaling.CrossVersionObjectReference{
+		Kind:       in.Target.Kind,
+		Name:       in.Target.Name,
+		APIVersion: in.Target.APIVersion,
+	}
+	out.Metric = autoscaling.MetricIdentifier{
+		Name:     in.MetricName,
+		Selector: in.Selector,
+	}
+	return nil
+}
+
+func Convert_autoscaling_PodsMetricStatus_To_v2beta1_PodsMetricStatus(in *autoscaling.PodsMetricStatus, out *autoscalingv2beta1.PodsMetricStatus, s conversion.Scope) error {
+	if in.Current.AverageValue != nil {
+		out.CurrentAverageValue = *in.Current.AverageValue
+	}
+	out.MetricName = in.Metric.Name
+	out.Selector = in.Metric.Selector
+	return nil
+}
+
+func Convert_v2beta1_PodsMetricStatus_To_autoscaling_PodsMetricStatus(in *autoscalingv2beta1.PodsMetricStatus, out *autoscaling.PodsMetricStatus, s conversion.Scope) error {
+	out.Current = autoscaling.MetricValueStatus{
+		AverageValue: &in.CurrentAverageValue,
+	}
+	out.Metric = autoscaling.MetricIdentifier{
+		Name:     in.MetricName,
+		Selector: in.Selector,
+	}
+	return nil
+}