mirror of
https://github.com/zalando-incubator/kube-metrics-adapter.git
synced 2025-01-09 18:01:36 +00:00
Compare commits
44 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
e61071a36d | ||
|
c108fab55a | ||
|
be7567efea | ||
|
b677e814be | ||
|
deec4727ee | ||
|
cf369639d8 | ||
|
262a35c2ec | ||
|
607386834b | ||
|
4bdce4da4d | ||
|
801e5d7a47 | ||
|
4df21ae2b4 | ||
|
7f639baeee | ||
|
88ddb6f10e | ||
|
d9cf2e0858 | ||
|
c3b18e784b | ||
|
c3a03fd758 | ||
|
05704c0a6b | ||
|
f3a608fcf7 | ||
|
223ec9fd89 | ||
|
81255aa956 | ||
|
dac44965e3 | ||
|
dbed1570ba | ||
|
ef24244074 | ||
|
cadf2dff3c | ||
|
db83268343 | ||
|
78c2ef742b | ||
|
f0b817629c | ||
|
3e7b66070c | ||
|
f3d33b6132 | ||
|
7b7af889fb | ||
|
2bcaa80ce0 | ||
|
a2e6980b25 | ||
|
f0783dcdd8 | ||
|
560fb40dae | ||
|
41229e370d | ||
|
338393c11e | ||
|
8e7d1bc0d6 | ||
|
dc7fb5875b | ||
|
8f70d3493e | ||
|
97ec13d010 | ||
|
5e20dcd961 | ||
|
f27dbc3939 | ||
|
6443613930 | ||
|
d9c2e50f2c |
35
.github/CODEOWNERS
vendored
35
.github/CODEOWNERS
vendored
@ -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
|
14
.github/dependabot.yml
vendored
Normal file
14
.github/dependabot.yml
vendored
Normal file
@ -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
|
@ -2,7 +2,7 @@ language: go
|
||||
dist: xenial
|
||||
|
||||
go:
|
||||
- "1.13.x"
|
||||
- "1.14.x"
|
||||
|
||||
env:
|
||||
- GO111MODULE=on GOLANGCI_RELEASE="v1.21.0"
|
||||
|
@ -1,3 +1,2 @@
|
||||
Mikkel Larsen <mikkel.larsen@zalando.de>
|
||||
Arjun Naik <arjun.naik@zalando.de>
|
||||
Team Teapot <team-teapot@zalando.de>
|
||||
|
10
README.md
10
README.md
@ -142,7 +142,7 @@ example the following JSON data would be expected:
|
||||
```
|
||||
|
||||
The json-path query support depends on the
|
||||
[github.com/oliveagle/jsonpath](https://github.com/oliveagle/jsonpath) library.
|
||||
[github.com/spyzhov/ajson](https://github.com/spyzhov/ajson) library.
|
||||
See the README for possible queries. It's expected that the metric you query
|
||||
returns something that can be turned into a `float64`.
|
||||
|
||||
@ -166,6 +166,14 @@ will create a URL like this:
|
||||
http://<podIP>:9090/metrics?foo=bar&baz=bop
|
||||
```
|
||||
|
||||
There are also configuration options for custom (connect and request) timeouts when querying pods for metrics:
|
||||
```yaml
|
||||
metric-config.pods.requests-per-second.json-path/request-timeout: 2s
|
||||
metric-config.pods.requests-per-second.json-path/connect-timeout: 500ms
|
||||
```
|
||||
|
||||
The default for both of the above values is 15 seconds.
|
||||
|
||||
## Prometheus collector
|
||||
|
||||
The Prometheus collector is a generic collector which can map Prometheus
|
||||
|
30
go.mod
30
go.mod
@ -2,31 +2,33 @@ 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.35.0
|
||||
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/kubernetes-incubator/custom-metrics-apiserver v0.0.0-20200618121405-54026617ec44
|
||||
github.com/lib/pq v1.2.0 // indirect
|
||||
github.com/mattn/go-colorable v0.1.4 // indirect
|
||||
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/prometheus/client_golang v1.7.1
|
||||
github.com/prometheus/common v0.14.0
|
||||
github.com/sirupsen/logrus v1.7.0
|
||||
github.com/spf13/cobra v0.0.7
|
||||
github.com/stretchr/testify v1.5.1
|
||||
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-20200220183623-bac4c82f6975 // indirect
|
||||
golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45
|
||||
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.17.3
|
||||
k8s.io/apimachinery v0.17.4
|
||||
k8s.io/client-go v0.17.3
|
||||
k8s.io/component-base v0.17.3
|
||||
k8s.io/api v0.18.8
|
||||
k8s.io/apimachinery 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/metrics v0.17.3
|
||||
k8s.io/metrics v0.18.8
|
||||
)
|
||||
|
||||
// use version that provides: https://github.com/kubernetes-sigs/custom-metrics-apiserver/pull/65
|
||||
replace github.com/kubernetes-incubator/custom-metrics-apiserver => github.com/mikkeloscar/custom-metrics-apiserver v0.0.0-20200626202535-cd2d550cf55d
|
||||
|
||||
go 1.13
|
||||
|
@ -7,7 +7,6 @@ import (
|
||||
|
||||
"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"
|
||||
@ -36,14 +35,12 @@ func (p *HTTPCollectorPlugin) NewCollector(_ *v2beta2.HorizontalPodAutoscaler, c
|
||||
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
|
||||
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
|
||||
@ -65,13 +62,16 @@ func (p *HTTPCollectorPlugin) NewCollector(_ *v2beta2.HorizontalPodAutoscaler, c
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
collector.metricsGetter = httpmetrics.NewJSONPathMetricsGetter(httpmetrics.DefaultMetricsHTTPClient(), aggFunc, jsonPath)
|
||||
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
|
||||
jsonPath *jsonpath.Compiled
|
||||
interval time.Duration
|
||||
metricType v2beta2.MetricSourceType
|
||||
metricsGetter *httpmetrics.JSONPathMetricsGetter
|
||||
|
@ -1,7 +1,6 @@
|
||||
package httpmetrics
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
@ -9,38 +8,50 @@ import (
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"github.com/oliveagle/jsonpath"
|
||||
"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 *jsonpath.Compiled
|
||||
jsonPath string
|
||||
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 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
|
||||
}
|
||||
|
||||
func DefaultMetricsHTTPClient() *http.Client {
|
||||
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: 15 * time.Second,
|
||||
Timeout: connectTimeout,
|
||||
}).DialContext,
|
||||
MaxIdleConns: 50,
|
||||
IdleConnTimeout: 90 * time.Second,
|
||||
ExpectContinueTimeout: 1 * time.Second,
|
||||
},
|
||||
Timeout: 15 * 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.
|
||||
@ -51,57 +62,45 @@ func (g *JSONPathMetricsGetter) GetMetric(metricsURL url.URL) (float64, error) {
|
||||
}
|
||||
|
||||
// parse data
|
||||
var jsonData interface{}
|
||||
err = json.Unmarshal(data, &jsonData)
|
||||
root, err := ajson.Unmarshal(data)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
res, err := g.jsonPath.Lookup(jsonData)
|
||||
nodes, err := root.JSONPath(g.jsonPath)
|
||||
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 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")
|
||||
}
|
||||
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)
|
||||
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
|
||||
}
|
||||
|
||||
return out, 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) {
|
||||
|
@ -1,59 +1,21 @@
|
||||
package httpmetrics
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"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 {
|
||||
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")
|
||||
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)
|
||||
_, err := w.Write(response)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
return httptest.NewServer(http.HandlerFunc(h))
|
||||
@ -61,29 +23,62 @@ func makeTestHTTPServer(t *testing.T, values ...int64) *httptest.Server {
|
||||
|
||||
func TestJSONPathMetricsGetter(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
input []int64
|
||||
output float64
|
||||
aggregator AggregatorFunc
|
||||
name string
|
||||
jsonResponse []byte
|
||||
jsonPath string
|
||||
result float64
|
||||
aggregator AggregatorFunc
|
||||
err error
|
||||
}{
|
||||
{
|
||||
name: "basic average",
|
||||
input: []int64{3, 4, 5},
|
||||
output: 4,
|
||||
aggregator: Average,
|
||||
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.input...)
|
||||
server := makeTestHTTPServer(t, tc.jsonResponse)
|
||||
defer server.Close()
|
||||
path, err := jsonpath.Compile("$.value")
|
||||
getter, err := NewJSONPathMetricsGetter(DefaultMetricsHTTPClient(), tc.aggregator, tc.jsonPath)
|
||||
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)
|
||||
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.output, metric)
|
||||
require.Equal(t, tc.result, metric)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
@ -4,8 +4,8 @@ import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/oliveagle/jsonpath"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
)
|
||||
|
||||
@ -32,18 +32,13 @@ func (g PodMetricsJSONPathGetter) GetMetric(pod *v1.Pod) (float64, error) {
|
||||
func NewPodMetricsJSONPathGetter(config map[string]string) (*PodMetricsJSONPathGetter, error) {
|
||||
getter := PodMetricsJSONPathGetter{}
|
||||
var (
|
||||
jsonPath *jsonpath.Compiled
|
||||
jsonPath string
|
||||
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
|
||||
jsonPath = v
|
||||
}
|
||||
|
||||
if v, ok := config["scheme"]; ok {
|
||||
@ -72,7 +67,37 @@ func NewPodMetricsJSONPathGetter(config map[string]string) (*PodMetricsJSONPathG
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
getter.metricGetter = NewJSONPathMetricsGetter(DefaultMetricsHTTPClient(), aggregator, jsonPath)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
|
@ -3,8 +3,8 @@ package httpmetrics
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/oliveagle/jsonpath"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
@ -22,12 +22,11 @@ func TestNewPodJSONPathMetricsGetter(t *testing.T) {
|
||||
"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},
|
||||
metricGetter: &JSONPathMetricsGetter{jsonPath: configNoAggregator["json-key"]},
|
||||
scheme: "http",
|
||||
path: "/metrics",
|
||||
port: 9090,
|
||||
@ -40,12 +39,11 @@ func TestNewPodJSONPathMetricsGetter(t *testing.T) {
|
||||
"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},
|
||||
metricGetter: &JSONPathMetricsGetter{jsonPath: configAggregator["json-key"], aggregator: Average},
|
||||
scheme: "http",
|
||||
path: "/metrics",
|
||||
port: 9090,
|
||||
@ -78,12 +76,11 @@ func TestNewPodJSONPathMetricsGetter(t *testing.T) {
|
||||
"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},
|
||||
metricGetter: &JSONPathMetricsGetter{jsonPath: configWithRawQuery["json-key"]},
|
||||
scheme: "http",
|
||||
path: "/metrics",
|
||||
port: 9090,
|
||||
@ -106,8 +103,6 @@ func TestBuildMetricsURL(t *testing.T) {
|
||||
"port": port,
|
||||
"raw-query": rawQuery,
|
||||
}
|
||||
_, err := jsonpath.Compile(configWithRawQuery["json-key"])
|
||||
require.NoError(t, err)
|
||||
getterWithRawQuery, err1 := NewPodMetricsJSONPathGetter(configWithRawQuery)
|
||||
require.NoError(t, err1)
|
||||
|
||||
@ -122,8 +117,6 @@ func TestBuildMetricsURL(t *testing.T) {
|
||||
"path": path,
|
||||
"port": port,
|
||||
}
|
||||
_, err2 := jsonpath.Compile(configWithNoQuery["json-key"])
|
||||
require.NoError(t, err2)
|
||||
getterWithNoQuery, err3 := NewPodMetricsJSONPathGetter(configWithNoQuery)
|
||||
require.NoError(t, err3)
|
||||
|
||||
@ -131,3 +124,67 @@ func TestBuildMetricsURL(t *testing.T) {
|
||||
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)
|
||||
}
|
||||
|
@ -1,6 +1,7 @@
|
||||
package collector
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
@ -8,6 +9,7 @@ import (
|
||||
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"
|
||||
@ -80,37 +82,25 @@ func (c *PodCollector) GetMetrics() ([]CollectedMetric, error) {
|
||||
LabelSelector: labels.Set(c.podLabelSelector.MatchLabels).String(),
|
||||
}
|
||||
|
||||
pods, err := c.client.CoreV1().Pods(c.namespace).List(opts)
|
||||
pods, err := c.client.CoreV1().Pods(c.namespace).List(context.TODO(), opts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
values := make([]CollectedMetric, 0, len(pods.Items))
|
||||
|
||||
// TODO: get metrics in parallel
|
||||
ch := make(chan CollectedMetric)
|
||||
errCh := make(chan error)
|
||||
for _, pod := range pods.Items {
|
||||
value, err := c.Getter.GetMetric(&pod)
|
||||
if err != nil {
|
||||
c.logger.Errorf("Failed to get metrics from pod '%s/%s': %v", pod.Namespace, pod.Name, err)
|
||||
continue
|
||||
}
|
||||
go c.getPodMetric(pod, ch, errCh)
|
||||
}
|
||||
|
||||
metricValue := CollectedMetric{
|
||||
Type: c.metricType,
|
||||
Custom: custom_metrics.MetricValue{
|
||||
DescribedObject: custom_metrics.ObjectReference{
|
||||
APIVersion: "v1",
|
||||
Kind: "Pod",
|
||||
Name: pod.Name,
|
||||
Namespace: pod.Namespace,
|
||||
},
|
||||
Metric: custom_metrics.MetricIdentifier{Name: c.metric.Name, Selector: c.podLabelSelector},
|
||||
Timestamp: metav1.Time{Time: time.Now().UTC()},
|
||||
Value: *resource.NewMilliQuantity(int64(value*1000), resource.DecimalSI),
|
||||
},
|
||||
values := make([]CollectedMetric, 0, len(pods.Items))
|
||||
for i := 0; i < len(pods.Items); i++ {
|
||||
select {
|
||||
case err := <-errCh:
|
||||
c.logger.Error(err)
|
||||
case resp := <-ch:
|
||||
values = append(values, resp)
|
||||
}
|
||||
|
||||
values = append(values, metricValue)
|
||||
}
|
||||
|
||||
return values, nil
|
||||
@ -120,16 +110,39 @@ func (c *PodCollector) Interval() time.Duration {
|
||||
return c.interval
|
||||
}
|
||||
|
||||
func (c *PodCollector) getPodMetric(pod corev1.Pod, ch chan CollectedMetric, errCh chan error) {
|
||||
value, err := c.Getter.GetMetric(&pod)
|
||||
if err != nil {
|
||||
errCh <- fmt.Errorf("Failed to get metrics from pod '%s/%s': %v", pod.Namespace, pod.Name, err)
|
||||
return
|
||||
}
|
||||
|
||||
ch <- CollectedMetric{
|
||||
Type: c.metricType,
|
||||
Custom: custom_metrics.MetricValue{
|
||||
DescribedObject: custom_metrics.ObjectReference{
|
||||
APIVersion: "v1",
|
||||
Kind: "Pod",
|
||||
Name: pod.Name,
|
||||
Namespace: pod.Namespace,
|
||||
},
|
||||
Metric: custom_metrics.MetricIdentifier{Name: c.metric.Name, Selector: c.podLabelSelector},
|
||||
Timestamp: metav1.Time{Time: time.Now().UTC()},
|
||||
Value: *resource.NewMilliQuantity(int64(value*1000), resource.DecimalSI),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getPodLabelSelector(client kubernetes.Interface, hpa *autoscalingv2.HorizontalPodAutoscaler) (*metav1.LabelSelector, error) {
|
||||
switch hpa.Spec.ScaleTargetRef.Kind {
|
||||
case "Deployment":
|
||||
deployment, err := client.AppsV1().Deployments(hpa.Namespace).Get(hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
deployment, err := client.AppsV1().Deployments(hpa.Namespace).Get(context.TODO(), hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return deployment.Spec.Selector, nil
|
||||
case "StatefulSet":
|
||||
sts, err := client.AppsV1().StatefulSets(hpa.Namespace).Get(hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
sts, err := client.AppsV1().StatefulSets(hpa.Namespace).Get(context.TODO(), hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -1,11 +1,13 @@
|
||||
package collector
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@ -55,7 +57,7 @@ func TestPodCollector(t *testing.T) {
|
||||
for _, m := range metrics {
|
||||
values = append(values, m.Custom.Value.Value())
|
||||
}
|
||||
require.Equal(t, tc.result, values)
|
||||
require.ElementsMatch(t, tc.result, values)
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -68,9 +70,13 @@ type testMetricsHandler struct {
|
||||
calledCounter uint
|
||||
t *testing.T
|
||||
metricsPath string
|
||||
sync.RWMutex
|
||||
}
|
||||
|
||||
func (h *testMetricsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
|
||||
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]})
|
||||
@ -110,7 +116,7 @@ func makeTestPods(t *testing.T, testServer string, metricName string, port strin
|
||||
PodIP: testServer,
|
||||
},
|
||||
}
|
||||
_, err := client.CoreV1().Pods(testNamespace).Create(testPod)
|
||||
_, err := client.CoreV1().Pods(testNamespace).Create(context.TODO(), testPod, v1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
}
|
||||
@ -124,7 +130,7 @@ func makeTestDeployment(t *testing.T, client kubernetes.Interface) *appsv1.Deplo
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := client.AppsV1().Deployments(testNamespace).Create(&deployment)
|
||||
_, err := client.AppsV1().Deployments(testNamespace).Create(context.TODO(), &deployment, v1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
return &deployment
|
||||
|
||||
@ -144,7 +150,7 @@ func makeTestHPA(t *testing.T, client kubernetes.Interface) *autoscalingv2.Horiz
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers("test-namespace").Create(hpa)
|
||||
_, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers("test-namespace").Create(context.TODO(), hpa, v1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
return hpa
|
||||
}
|
||||
|
@ -1,6 +1,7 @@
|
||||
package collector
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@ -87,16 +88,16 @@ func NewSkipperCollector(client kubernetes.Interface, plugin CollectorPlugin, hp
|
||||
}, nil
|
||||
}
|
||||
|
||||
func getAnnotationWeight(backendWeights string, backend string) float64 {
|
||||
var weightsMap map[string]int
|
||||
func getAnnotationWeight(backendWeights string, backend string) (float64, error) {
|
||||
var weightsMap map[string]float64
|
||||
err := json.Unmarshal([]byte(backendWeights), &weightsMap)
|
||||
if err != nil {
|
||||
return 0
|
||||
return 0, err
|
||||
}
|
||||
if weight, ok := weightsMap[backend]; ok {
|
||||
return float64(weight) / 100
|
||||
return float64(weight) / 100, nil
|
||||
}
|
||||
return 0
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
func getWeights(ingressAnnotations map[string]string, backendAnnotations []string, backend string) (float64, error) {
|
||||
@ -106,7 +107,11 @@ func getWeights(ingressAnnotations map[string]string, backendAnnotations []strin
|
||||
for _, anno := range backendAnnotations {
|
||||
if weightsMap, ok := ingressAnnotations[anno]; ok {
|
||||
annotationsPresent = true
|
||||
maxWeight = math.Max(maxWeight, getAnnotationWeight(weightsMap, backend))
|
||||
weight, err := getAnnotationWeight(weightsMap, backend)
|
||||
if err != nil {
|
||||
return 0.0, err
|
||||
}
|
||||
maxWeight = math.Max(maxWeight, weight)
|
||||
}
|
||||
}
|
||||
|
||||
@ -125,7 +130,7 @@ 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{})
|
||||
ingress, err := c.client.NetworkingV1beta1().Ingresses(c.objectReference.Namespace).Get(context.TODO(), c.objectReference.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -207,13 +212,13 @@ func targetRefReplicas(client kubernetes.Interface, hpa *autoscalingv2.Horizonta
|
||||
var replicas int32
|
||||
switch hpa.Spec.ScaleTargetRef.Kind {
|
||||
case "Deployment":
|
||||
deployment, err := client.AppsV1().Deployments(hpa.Namespace).Get(hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
deployment, err := client.AppsV1().Deployments(hpa.Namespace).Get(context.TODO(), hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
replicas = deployment.Status.Replicas
|
||||
case "StatefulSet":
|
||||
sts, err := client.AppsV1().StatefulSets(hpa.Namespace).Get(hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
sts, err := client.AppsV1().StatefulSets(hpa.Namespace).Get(context.TODO(), hpa.Spec.ScaleTargetRef.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
@ -1,6 +1,7 @@
|
||||
package collector
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"testing"
|
||||
@ -10,7 +11,7 @@ import (
|
||||
appsv1 "k8s.io/api/apps/v1"
|
||||
autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"k8s.io/api/extensions/v1beta1"
|
||||
v1beta1 "k8s.io/api/networking/v1beta1"
|
||||
"k8s.io/apimachinery/pkg/api/resource"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
@ -32,7 +33,7 @@ func TestTargetRefReplicasDeployments(t *testing.T) {
|
||||
|
||||
// Create an HPA with the deployment as ref
|
||||
hpa, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers(deployment.Namespace).
|
||||
Create(newHPA(defaultNamespace, name, "Deployment"))
|
||||
Create(context.TODO(), newHPA(defaultNamespace, name, "Deployment"), metav1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
|
||||
replicas, err := targetRefReplicas(client, hpa)
|
||||
@ -49,7 +50,7 @@ func TestTargetRefReplicasStatefulSets(t *testing.T) {
|
||||
|
||||
// Create an HPA with the statefulSet as ref
|
||||
hpa, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers(statefulSet.Namespace).
|
||||
Create(newHPA(defaultNamespace, name, "StatefulSet"))
|
||||
Create(context.TODO(), newHPA(defaultNamespace, name, "StatefulSet"), metav1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
|
||||
replicas, err := targetRefReplicas(client, hpa)
|
||||
@ -73,7 +74,7 @@ func newHPA(namesapce string, refName string, refKind string) *autoscalingv2.Hor
|
||||
}
|
||||
|
||||
func newDeployment(client *fake.Clientset, namespace string, name string, replicas, readyReplicas int32) (*appsv1.Deployment, error) {
|
||||
return client.AppsV1().Deployments(namespace).Create(&appsv1.Deployment{
|
||||
return client.AppsV1().Deployments(namespace).Create(context.TODO(), &appsv1.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: name,
|
||||
Namespace: namespace,
|
||||
@ -83,11 +84,11 @@ func newDeployment(client *fake.Clientset, namespace string, name string, replic
|
||||
ReadyReplicas: replicas,
|
||||
Replicas: readyReplicas,
|
||||
},
|
||||
})
|
||||
}, metav1.CreateOptions{})
|
||||
}
|
||||
|
||||
func newStatefulSet(client *fake.Clientset, namespace string, name string) (*appsv1.StatefulSet, error) {
|
||||
return client.AppsV1().StatefulSets(namespace).Create(&appsv1.StatefulSet{
|
||||
return client.AppsV1().StatefulSets(namespace).Create(context.TODO(), &appsv1.StatefulSet{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: name,
|
||||
Namespace: namespace,
|
||||
@ -96,7 +97,7 @@ func newStatefulSet(client *fake.Clientset, namespace string, name string) (*app
|
||||
ReadyReplicas: 1,
|
||||
Replicas: 2,
|
||||
},
|
||||
})
|
||||
}, metav1.CreateOptions{})
|
||||
}
|
||||
|
||||
func TestSkipperCollector(t *testing.T) {
|
||||
@ -111,7 +112,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
expectError bool
|
||||
fakedAverage bool
|
||||
namespace string
|
||||
backendWeights map[string]map[string]int
|
||||
backendWeights map[string]map[string]float64
|
||||
replicas int32
|
||||
readyReplicas int32
|
||||
backendAnnotations []string
|
||||
@ -137,7 +138,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
collectedMetric: 1000,
|
||||
namespace: "default",
|
||||
backend: "backend1",
|
||||
backendWeights: map[string]map[string]int{testBackendWeightsAnnotation: {"backend2": 60, "backend1": 40}},
|
||||
backendWeights: map[string]map[string]float64{testBackendWeightsAnnotation: {"backend2": 60.0, "backend1": 40}},
|
||||
replicas: 1,
|
||||
readyReplicas: 1,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -151,7 +152,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
collectedMetric: 1000,
|
||||
namespace: "default",
|
||||
backend: "backend1",
|
||||
backendWeights: map[string]map[string]int{testBackendWeightsAnnotation: {"backend2": 60, "backend1": 40}},
|
||||
backendWeights: map[string]map[string]float64{testBackendWeightsAnnotation: {"backend2": 60, "backend1": 40}},
|
||||
replicas: 1,
|
||||
readyReplicas: 1,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -166,7 +167,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
fakedAverage: true,
|
||||
namespace: "default",
|
||||
backend: "backend1",
|
||||
backendWeights: map[string]map[string]int{testBackendWeightsAnnotation: {"backend2": 50, "backend1": 50}},
|
||||
backendWeights: map[string]map[string]float64{testBackendWeightsAnnotation: {"backend2": 50, "backend1": 50}},
|
||||
replicas: 5,
|
||||
readyReplicas: 5,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -180,7 +181,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
collectedMetric: 1500,
|
||||
namespace: "default",
|
||||
backend: "backend1",
|
||||
backendWeights: map[string]map[string]int{testBackendWeightsAnnotation: {"backend2": 50, "backend1": 50}},
|
||||
backendWeights: map[string]map[string]float64{testBackendWeightsAnnotation: {"backend2": 50, "backend1": 50}},
|
||||
replicas: 5, // this is not taken into account
|
||||
readyReplicas: 5,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -194,7 +195,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
collectedMetric: 0,
|
||||
namespace: "default",
|
||||
backend: "backend1",
|
||||
backendWeights: map[string]map[string]int{testBackendWeightsAnnotation: {"backend2": 100, "backend1": 0}},
|
||||
backendWeights: map[string]map[string]float64{testBackendWeightsAnnotation: {"backend2": 100, "backend1": 0}},
|
||||
replicas: 5,
|
||||
readyReplicas: 5,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -209,7 +210,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
fakedAverage: true,
|
||||
namespace: "default",
|
||||
backend: "backend1",
|
||||
backendWeights: map[string]map[string]int{
|
||||
backendWeights: map[string]map[string]float64{
|
||||
testBackendWeightsAnnotation: {"backend2": 20, "backend1": 80},
|
||||
testStacksetWeightsAnnotation: {"backend2": 0, "backend1": 100},
|
||||
},
|
||||
@ -226,7 +227,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
collectedMetric: 1500,
|
||||
namespace: "default",
|
||||
backend: "backend1",
|
||||
backendWeights: map[string]map[string]int{
|
||||
backendWeights: map[string]map[string]float64{
|
||||
testBackendWeightsAnnotation: {"backend2": 20, "backend1": 80},
|
||||
testStacksetWeightsAnnotation: {"backend2": 0, "backend1": 100},
|
||||
},
|
||||
@ -243,7 +244,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
collectedMetric: 0,
|
||||
namespace: "default",
|
||||
backend: "backend3",
|
||||
backendWeights: map[string]map[string]int{testBackendWeightsAnnotation: {"backend2": 100, "backend1": 0}},
|
||||
backendWeights: map[string]map[string]float64{testBackendWeightsAnnotation: {"backend2": 100, "backend1": 0}},
|
||||
replicas: 1,
|
||||
readyReplicas: 1,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -257,7 +258,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
collectedMetric: 1500,
|
||||
namespace: "default",
|
||||
backend: "backend3",
|
||||
backendWeights: map[string]map[string]int{},
|
||||
backendWeights: map[string]map[string]float64{},
|
||||
replicas: 1,
|
||||
readyReplicas: 1,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -271,7 +272,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
expectError: true,
|
||||
namespace: "default",
|
||||
backend: "",
|
||||
backendWeights: map[string]map[string]int{testBackendWeightsAnnotation: {"backend2": 100, "backend1": 0}},
|
||||
backendWeights: map[string]map[string]float64{testBackendWeightsAnnotation: {"backend2": 100, "backend1": 0}},
|
||||
replicas: 1,
|
||||
readyReplicas: 1,
|
||||
backendAnnotations: []string{testBackendWeightsAnnotation},
|
||||
@ -300,7 +301,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
fakedAverage: true,
|
||||
namespace: "default",
|
||||
backend: "backend2",
|
||||
backendWeights: map[string]map[string]int{
|
||||
backendWeights: map[string]map[string]float64{
|
||||
testBackendWeightsAnnotation: {"backend2": 20, "backend1": 80},
|
||||
testStacksetWeightsAnnotation: {"backend1": 100},
|
||||
},
|
||||
@ -313,12 +314,12 @@ func TestSkipperCollector(t *testing.T) {
|
||||
metric: 1500,
|
||||
ingressName: "dummy-ingress",
|
||||
hostnames: []string{"example.org"},
|
||||
expectedQuery: `scalar(sum(rate(skipper_serve_host_duration_seconds_count{host=~"example_org"}[1m])) * 0.2000)`,
|
||||
expectedQuery: `scalar(sum(rate(skipper_serve_host_duration_seconds_count{host=~"example_org"}[1m])) * 0.2050)`,
|
||||
collectedMetric: 1500,
|
||||
namespace: "default",
|
||||
backend: "backend2",
|
||||
backendWeights: map[string]map[string]int{
|
||||
testBackendWeightsAnnotation: {"backend2": 20, "backend1": 80},
|
||||
backendWeights: map[string]map[string]float64{
|
||||
testBackendWeightsAnnotation: {"backend2": 20.5, "backend1": 79.5},
|
||||
testStacksetWeightsAnnotation: {"backend1": 100},
|
||||
},
|
||||
replicas: 5,
|
||||
@ -351,7 +352,7 @@ func TestSkipperCollector(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func makeIngress(client kubernetes.Interface, namespace, ingressName, backend string, hostnames []string, backendWeights map[string]map[string]int) error {
|
||||
func makeIngress(client kubernetes.Interface, namespace, ingressName, backend string, hostnames []string, backendWeights map[string]map[string]float64) error {
|
||||
annotations := make(map[string]string)
|
||||
for anno, weights := range backendWeights {
|
||||
sWeights, err := json.Marshal(weights)
|
||||
@ -382,7 +383,7 @@ func makeIngress(client kubernetes.Interface, namespace, ingressName, backend st
|
||||
Host: hostname,
|
||||
})
|
||||
}
|
||||
_, err := client.ExtensionsV1beta1().Ingresses(namespace).Create(ingress)
|
||||
_, err := client.NetworkingV1beta1().Ingresses(namespace).Create(context.TODO(), ingress, metav1.CreateOptions{})
|
||||
return err
|
||||
}
|
||||
|
||||
|
@ -119,7 +119,7 @@ func (p *HPAProvider) Run(ctx context.Context) {
|
||||
func (p *HPAProvider) updateHPAs() error {
|
||||
p.logger.Info("Looking for HPAs")
|
||||
|
||||
hpas, err := p.client.AutoscalingV2beta2().HorizontalPodAutoscalers(metav1.NamespaceAll).List(metav1.ListOptions{})
|
||||
hpas, err := p.client.AutoscalingV2beta2().HorizontalPodAutoscalers(metav1.NamespaceAll).List(context.TODO(), metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -70,7 +70,7 @@ func TestUpdateHPAs(t *testing.T) {
|
||||
fakeClient := fake.NewSimpleClientset()
|
||||
|
||||
var err error
|
||||
hpa, err = fakeClient.AutoscalingV2beta2().HorizontalPodAutoscalers("default").Create(hpa)
|
||||
hpa, err = fakeClient.AutoscalingV2beta2().HorizontalPodAutoscalers("default").Create(context.TODO(), hpa, metav1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
|
||||
collectorFactory := collector.NewCollectorFactory()
|
||||
@ -86,7 +86,7 @@ func TestUpdateHPAs(t *testing.T) {
|
||||
|
||||
// update HPA
|
||||
hpa.Annotations["metric-config.pods.requests-per-second.json-path/port"] = "8080"
|
||||
_, err = fakeClient.AutoscalingV2beta2().HorizontalPodAutoscalers("default").Update(hpa)
|
||||
_, err = fakeClient.AutoscalingV2beta2().HorizontalPodAutoscalers("default").Update(context.TODO(), hpa, metav1.UpdateOptions{})
|
||||
require.NoError(t, err)
|
||||
|
||||
err = provider.updateHPAs()
|
||||
@ -134,7 +134,7 @@ func TestUpdateHPAsDisregardingIncompatibleHPA(t *testing.T) {
|
||||
fakeClient := fake.NewSimpleClientset()
|
||||
|
||||
var err error
|
||||
_, err = fakeClient.AutoscalingV2beta2().HorizontalPodAutoscalers("default").Create(hpa)
|
||||
_, err = fakeClient.AutoscalingV2beta2().HorizontalPodAutoscalers("default").Create(context.TODO(), hpa, metav1.CreateOptions{})
|
||||
require.NoError(t, err)
|
||||
|
||||
collectorFactory := collector.NewCollectorFactory()
|
||||
|
@ -69,9 +69,15 @@ func (s *MetricStore) insertCustomMetric(value custom_metrics.MetricValue) {
|
||||
Resource: "pods",
|
||||
}
|
||||
case "Ingress":
|
||||
// group can be either `extentions` or `networking.k8s.io`
|
||||
group := "extensions"
|
||||
gv, err := schema.ParseGroupVersion(value.DescribedObject.APIVersion)
|
||||
if err == nil {
|
||||
group = gv.Group
|
||||
}
|
||||
groupResource = schema.GroupResource{
|
||||
Resource: "ingresses",
|
||||
Group: "extensions",
|
||||
Group: group,
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -126,6 +126,8 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
|
||||
return err
|
||||
}
|
||||
|
||||
config.GenericConfig.OpenAPIConfig.Info.Title = "kube-metrics-adapter"
|
||||
|
||||
var clientConfig *rest.Config
|
||||
if len(o.RemoteKubeConfigFile) > 0 {
|
||||
loadingRules := &clientcmd.ClientConfigLoadingRules{ExplicitPath: o.RemoteKubeConfigFile}
|
||||
|
Reference in New Issue
Block a user