yusheng-guo/kube-metrics-adapter
1
0
Fork 0
mirror of https://github.com/zalando-incubator/kube-metrics-adapter.git synced 2024-12-22 11:06:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added weighting of ingress request metrics based on stackset and backend weights

Browse Source 
Signed-off-by: Arjun Naik <arjun.rn@gmail.com>
This commit is contained in:
Arjun Naik 2019-01-09 14:38:56 +01:00
parent b685c7db1c
commit 37003d6513
4 changed files with 179 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
pkg/collector/collector.go
View File

@ -6,6 +6,7 @@ import (
"time"
autoscalingv2beta2 "k8s.io/api/autoscaling/v2beta2"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/metrics/pkg/apis/custom_metrics"
"k8s.io/metrics/pkg/apis/external_metrics"
)
@ -164,8 +165,9 @@ func getObjectReference(hpa *autoscalingv2beta2.HorizontalPodAutoscaler, metricN
}
type MetricTypeName struct {
Type autoscalingv2beta2.MetricSourceType
Name string
Type autoscalingv2beta2.MetricSourceType
Name string
MetricLabels *metav1.LabelSelector
}
type CollectedMetric struct {
@ -279,6 +281,7 @@ func ParseHPAMetrics(hpa *autoscalingv2beta2.HorizontalPodAutoscaler) ([]*Metric
typeName.Name = metric.Pods.Metric.Name
case autoscalingv2beta2.ObjectMetricSourceType:
typeName.Name = metric.Object.Metric.Name
typeName.MetricLabels = metric.Object.Metric.Selector
ref = custom_metrics.ObjectReference{
APIVersion: metric.Object.DescribedObject.APIVersion,
Kind: metric.Object.DescribedObject.Kind,

3
pkg/collector/max_collector.go
View File

@ -38,8 +38,7 @@ func (c *WeightedMaxCollector) GetMetrics() ([]CollectedMetric, error) {
}
}
max.Custom.Value = *resource.NewQuantity(int64(float64(max.Custom.Value.MilliValue())*c.weight), resource.DecimalSI)
max.Custom.Value = *resource.NewMilliQuantity(int64(float64(max.Custom.Value.MilliValue())*c.weight), resource.DecimalSI)
return []CollectedMetric{max}, nil
}

58
pkg/collector/skipper_collector.go
View File

@ -3,6 +3,7 @@ package collector
import (
"encoding/json"
"fmt"
"math"
"strings"
"time"
@ -13,8 +14,11 @@ import (
)
const (
rpsQuery = `scalar(sum(rate(skipper_serve_host_duration_seconds_count{host="%s"}[1m])))`
rpsMetricName = "requests-per-second"
rpsQuery = `scalar(sum(rate(skipper_serve_host_duration_seconds_count{host="%s"}[1m])))`
rpsMetricName = "requests-per-second"
stackTrafficWeights = "zalando.org/stack-traffic-weights"
backendWeights = "zalando.org/backend-weights"
hostLabel = "backend"
)
// SkipperCollectorPlugin is a collector plugin for initializing metrics
@ -34,15 +38,9 @@ func NewSkipperCollectorPlugin(client kubernetes.Interface, prometheusPlugin *Pr
// NewCollector initializes a new skipper collector from the specified HPA.
func (c *SkipperCollectorPlugin) NewCollector(hpa *autoscalingv2beta2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
backend := ""
if strings.HasPrefix(config.Name, rpsMetricName) {
if len(config.Name) > len(rpsMetricName) {
backend = config.Name[len(rpsMetricName):]
}
}
switch config.Name {
case rpsMetricName:
return NewSkipperCollector(c.client, c.plugin, hpa, config, interval, backend)
return NewSkipperCollector(c.client, c.plugin, hpa, config, interval)
default:
return nil, fmt.Errorf("metric '%s' not supported", config.Name)
}
@ -58,14 +56,12 @@ type SkipperCollector struct {
interval time.Duration
plugin CollectorPlugin
config MetricConfig
backend string
}
// NewSkipperCollector initializes a new SkipperCollector.
func NewSkipperCollector(
client kubernetes.Interface, plugin CollectorPlugin, hpa *autoscalingv2beta2.HorizontalPodAutoscaler,
config *MetricConfig, interval time.Duration, backend string,
) (*SkipperCollector, error) {
config *MetricConfig, interval time.Duration) (*SkipperCollector, error) {
return &SkipperCollector{
client: client,
objectReference: config.ObjectReference,
@ -74,11 +70,30 @@ func NewSkipperCollector(
interval: interval,
plugin: plugin,
config: *config,
backend: backend,
}, nil
}
const stackTrafficWeight = "zalando.org/stack-traffic-weights"
func getAnnotationWeight(annotations map[string]string, backendName, annotation string) float64 {
if weightAnnotation, ok := annotations[annotation]; ok {
var stackWeights map[string]int
err := json.Unmarshal([]byte(weightAnnotation), &stackWeights)
if err == nil {
if _, ok := stackWeights[backendName]; ok {
return float64(stackWeights[backendName]) / 100
}
}
}
return 0.0
}
func getMaxWeights(annotations map[string]string, backendName string) float64 {
stacksetWeight := getAnnotationWeight(annotations, backendName, stackTrafficWeights)
backendWeight := getAnnotationWeight(annotations, backendName, backendWeights)
if stacksetWeight == 0.0 && backendWeight == 0.0 {
return 1.0
}
return math.Max(stacksetWeight, backendWeight)
}
// getCollector returns a collector for getting the metrics.
func (c *SkipperCollector) getCollector() (Collector, error) {
@ -88,16 +103,7 @@ func (c *SkipperCollector) getCollector() (Collector, error) {
return nil, err
}
backendWeight := 1.0
var stackWeights map[string]int
if weightAnnotation, ok := ingress.Annotations[stackTrafficWeight]; ok {
err = json.Unmarshal([]byte(weightAnnotation), &stackWeights)
if err == nil {
if _, ok := stackWeights[c.backend]; ok {
backendWeight = float64(stackWeights[c.backend]) / 100
}
}
}
backendWeight := getMaxWeights(ingress.Annotations, c.config.MetricLabels.MatchLabels[hostLabel])
config := c.config
@ -117,10 +123,8 @@ func (c *SkipperCollector) getCollector() (Collector, error) {
collectors = append(collectors, collector)
}
if len(collectors) > 1 {
if len(collectors) >= 1 {
collector = NewWeightedMaxCollector(c.interval, backendWeight, collectors...)
} else if len(collectors) == 1 {
collector = collectors[0]
} else {
return nil, fmt.Errorf("no hosts defined on ingress %s/%s, unable to create collector", c.objectReference.Namespace, c.objectReference.Name)
}

204
pkg/collector/skipper_collector_test.go
View File

@ -1,87 +1,167 @@
package collector
import (
"testing"
"github.com/stretchr/testify/require"
"k8s.io/api/apps/v1"
"encoding/json"
"fmt"
"github.com/stretchr/testify/assert"
autoscalingv2beta2 "k8s.io/api/autoscaling/v2beta2"
"k8s.io/api/extensions/v1beta1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/metrics/pkg/apis/custom_metrics"
"testing"
"time"
)
func TestTargetRefReplicasDeployments(t *testing.T) {
client := fake.NewSimpleClientset()
name := "some-app"
defaultNamespace := "default"
deployment, err := newDeployment(client, defaultNamespace, name)
require.NoError(t, err)
// Create an HPA with the deployment as ref
hpa, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers(deployment.Namespace).
Create(newHPA(defaultNamespace, name, "Deployment"))
require.NoError(t, err)
replicas, err := targetRefReplicas(client, hpa)
require.NoError(t, err)
require.Equal(t, deployment.Status.Replicas, replicas)
type testCollector struct {
metrics []int
interval time.Duration
}
func TestTargetRefReplicasStatefulSets(t *testing.T) {
client := fake.NewSimpleClientset()
name := "some-app"
defaultNamespace := "default"
statefulSet, err := newStatefulSet(client, defaultNamespace, name)
require.NoError(t, err)
// Create an HPA with the statefulSet as ref
hpa, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers(statefulSet.Namespace).
Create(newHPA(defaultNamespace, name, "StatefulSet"))
require.NoError(t, err)
replicas, err := targetRefReplicas(client, hpa)
require.NoError(t, err)
require.Equal(t, statefulSet.Status.Replicas, replicas)
}
func newHPA(namesapce string, refName string, refKind string) *autoscalingv2beta2.HorizontalPodAutoscaler {
return &autoscalingv2beta2.HorizontalPodAutoscaler{
ObjectMeta: metav1.ObjectMeta{
Name: namesapce,
func (c testCollector) GetMetrics() ([]CollectedMetric, error) {
return []CollectedMetric{
{
Custom: custom_metrics.MetricValue{Value: *resource.NewQuantity(1000, resource.DecimalSI)},
},
}, nil
}
func (c testCollector) Interval() time.Duration {
return c.interval
}
type testCollectorPlugin struct {
metrics []int
}
func (c testCollectorPlugin) NewCollector(hpa *autoscalingv2beta2.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
return &testCollector{interval: interval, metrics: c.metrics}, nil
}
func makeTestCollectorPlugin(metrics []int) *testCollectorPlugin {
return &testCollectorPlugin{metrics: metrics}
}
func makeHpa(ingressName string) *autoscalingv2beta2.HorizontalPodAutoscaler {
return &autoscalingv2beta2.HorizontalPodAutoscaler{
Spec: autoscalingv2beta2.HorizontalPodAutoscalerSpec{
ScaleTargetRef: autoscalingv2beta2.CrossVersionObjectReference{
Name: refName,
Kind: refKind,
Metrics: []autoscalingv2beta2.MetricSpec{
{
Type: autoscalingv2beta2.ObjectMetricSourceType,
Object: &autoscalingv2beta2.ObjectMetricSource{
DescribedObject: autoscalingv2beta2.CrossVersionObjectReference{
Kind: "Ingress",
APIVersion: "extensions/v1",
Name: ingressName,
},
Metric: autoscalingv2beta2.MetricIdentifier{
Name: rpsMetricName,
},
Target: autoscalingv2beta2.MetricTarget{},
},
},
},
},
Status: autoscalingv2beta2.HorizontalPodAutoscalerStatus{},
}
}
func newDeployment(client *fake.Clientset, namespace string, name string) (*v1.Deployment, error) {
return client.AppsV1().Deployments(namespace).Create(&v1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: namespace,
func makeMetricConfig(ingressName, backend string) *MetricConfig {
selector, _ := metav1.ParseToLabelSelector(fmt.Sprintf("%s=%s", hostLabel, backend))
return &MetricConfig{
ObjectReference: custom_metrics.ObjectReference{Kind: "Ingress", Namespace: "default", Name: ingressName},
MetricTypeName: MetricTypeName{
Name: rpsMetricName,
MetricLabels: selector,
},
Spec: v1.DeploymentSpec{},
Status: v1.DeploymentStatus{
ReadyReplicas: 1,
Replicas: 2,
}
}
func makeIngress(client *fake.Clientset, ingressName string, bWeights, sWeights map[string]int) {
backendWeightAnno, _ := json.Marshal(bWeights)
stacksetWeightAnno, _ := json.Marshal(sWeights)
client.Extensions().Ingresses("default").Create(&v1beta1.Ingress{
ObjectMeta: metav1.ObjectMeta{
Name: ingressName,
Annotations: map[string]string{
stackTrafficWeights: string(stacksetWeightAnno),
backendWeights: string(backendWeightAnno),
},
},
Spec: v1beta1.IngressSpec{
Rules: []v1beta1.IngressRule{
{
Host: "dummy.example.com",
},
},
},
})
}
func newStatefulSet(client *fake.Clientset, namespace string, name string) (*v1.StatefulSet, error) {
return client.AppsV1().StatefulSets(namespace).Create(&v1.StatefulSet{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: namespace,
func TestNewSkipperCollectorSimple(t *testing.T) {
for _, tc := range []struct {
bWeights map[string]int
sWeights map[string]int
metrics []int
msg string
ingressName string
backend string
expectedMetric int
}{
{
msg: "backend without weights",
sWeights: map[string]int{"another-backend": 60},
bWeights: map[string]int{"another-backend": 60},
ingressName: "dummy-ingress",
backend: "dummy-backend",
metrics: []int{1000},
expectedMetric: 1000,
},
Status: v1.StatefulSetStatus{
ReadyReplicas: 1,
Replicas: 2,
{
msg: "weighted backed",
sWeights: map[string]int{"dummy-backend": 60},
bWeights: map[string]int{"dummy-backend": 60},
ingressName: "dummy-ingress",
backend: "dummy-backend",
metrics: []int{1000},
expectedMetric: 600,
},
})
{
msg: "missing weights annotation",
ingressName: "dummy-ingress",
backend: "dummy-backend",
metrics: []int{1000},
expectedMetric: 1000,
},
{
msg: "missing stackset weights annotation",
bWeights: map[string]int{"dummy-backend": 40},
ingressName: "dummy-ingress",
backend: "dummy-backend",
metrics: []int{1000},
expectedMetric: 400,
},
{
msg: "missing backend weights annotation",
sWeights: map[string]int{"dummy-backend": 100},
ingressName: "dummy-ingress",
backend: "dummy-backend",
metrics: []int{1000},
expectedMetric: 1000,
},
} {
t.Run(tc.msg, func(tt *testing.T) {
client := fake.NewSimpleClientset()
makeIngress(client, tc.ingressName, tc.bWeights, tc.sWeights)
plugin := makeTestCollectorPlugin(tc.metrics)
hpa := makeHpa(tc.ingressName)
config := makeMetricConfig(tc.ingressName, tc.backend)
collector, err := NewSkipperCollector(client, plugin, hpa, config, time.Minute)
assert.NoError(t, err, "failed to create skipper collector: %v", err)
collectedMetrics, err := collector.GetMetrics()
assert.NoError(t, err, "failed to get metrics from collector: %v", err)
assert.Len(t, collectedMetrics, len(tc.metrics), "number of metrics returned is not %d", len(tc.metrics))
assert.EqualValues(t, collectedMetrics[0].Custom.Value.Value(), tc.expectedMetric, "the collected metric is not 1000")
})
}
}