From 7967642cf0619a699c14dcd6dbfe19a5d44fa7fb Mon Sep 17 00:00:00 2001 From: Piotr Szczesniak Date: Fri, 28 Aug 2015 07:54:45 +0200 Subject: [PATCH] Implemented first version of InitialResources plugin. --- cmd/kube-apiserver/app/plugins.go | 1 + hack/verify-flags/known-flags.txt | 5 + .../admission/initialresources/admission.go | 157 ++++++++++++++ .../initialresources/admission_test.go | 192 ++++++++++++++++++ .../admission/initialresources/data_source.go | 192 ++++++++++++++++++ 5 files changed, 547 insertions(+) create mode 100644 plugin/pkg/admission/initialresources/admission.go create mode 100644 plugin/pkg/admission/initialresources/admission_test.go create mode 100644 plugin/pkg/admission/initialresources/data_source.go diff --git a/cmd/kube-apiserver/app/plugins.go b/cmd/kube-apiserver/app/plugins.go index 3d8eba2d2d..20f690d416 100644 --- a/cmd/kube-apiserver/app/plugins.go +++ b/cmd/kube-apiserver/app/plugins.go @@ -27,6 +27,7 @@ import ( _ "k8s.io/kubernetes/plugin/pkg/admission/admit" _ "k8s.io/kubernetes/plugin/pkg/admission/deny" _ "k8s.io/kubernetes/plugin/pkg/admission/exec/denyprivileged" + _ "k8s.io/kubernetes/plugin/pkg/admission/initialresources" _ "k8s.io/kubernetes/plugin/pkg/admission/limitranger" _ "k8s.io/kubernetes/plugin/pkg/admission/namespace/autoprovision" _ "k8s.io/kubernetes/plugin/pkg/admission/namespace/exists" diff --git a/hack/verify-flags/known-flags.txt b/hack/verify-flags/known-flags.txt index a3a4be222d..71a9127a3c 100644 --- a/hack/verify-flags/known-flags.txt +++ b/hack/verify-flags/known-flags.txt @@ -110,6 +110,11 @@ insecure-bind-address insecure-port insecure-skip-tls-verify iptables-sync-period +ir-data-source +ir-dbname +ir-influxdb-host +ir-password +ir-user jenkins-host jenkins-jobs km-path diff --git a/plugin/pkg/admission/initialresources/admission.go b/plugin/pkg/admission/initialresources/admission.go new file mode 100644 index 0000000000..090092cefe --- /dev/null +++ b/plugin/pkg/admission/initialresources/admission.go @@ -0,0 +1,157 @@ +/* +Copyright 2015 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package initialresources + +import ( + "flag" + "io" + "strings" + "time" + + "github.com/golang/glog" + "k8s.io/kubernetes/pkg/admission" + "k8s.io/kubernetes/pkg/api" + apierrors "k8s.io/kubernetes/pkg/api/errors" + "k8s.io/kubernetes/pkg/api/resource" + client "k8s.io/kubernetes/pkg/client/unversioned" +) + +var ( + source = flag.String("ir-data-source", "influxdb", "Data source used by InitialResources. Supported options: influxdb.") + percentile = flag.Int64("ir-percentile", 90, "Which percentile of samples should InitialResources use when estimating resources. For experiment purposes.") +) + +const ( + samplesThreshold = 60 + week = 7 * 24 * time.Hour + month = 30 * 24 * time.Hour +) + +// WARNING: this feature is experimental and will definitely change. +func init() { + admission.RegisterPlugin("InitialResources", func(client client.Interface, config io.Reader) (admission.Interface, error) { + s, err := newDataSource(*source) + if err != nil { + return nil, err + } + return newInitialResources(s), nil + }) +} + +type initialResources struct { + *admission.Handler + source dataSource +} + +func newInitialResources(source dataSource) admission.Interface { + return &initialResources{ + Handler: admission.NewHandler(admission.Create), + source: source, + } +} + +func (ir initialResources) Admit(a admission.Attributes) (err error) { + // Ignore all calls to subresources or resources other than pods. + if a.GetSubresource() != "" || a.GetResource() != string(api.ResourcePods) { + return nil + } + pod, ok := a.GetObject().(*api.Pod) + if !ok { + return apierrors.NewBadRequest("Resource was marked with kind Pod but was unable to be converted") + } + + ir.estimateAndFillResourcesIfNotSet(pod) + return nil +} + +// The method veryfies whether resources should be set for the given pod and +// if there is estimation available the method fills Request field. +func (ir initialResources) estimateAndFillResourcesIfNotSet(pod *api.Pod) { + for i := range pod.Spec.Containers { + c := &pod.Spec.Containers[i] + req := c.Resources.Requests + lim := c.Resources.Limits + var cpu, mem *resource.Quantity + var err error + if _, ok := req[api.ResourceCPU]; !ok { + if _, ok2 := lim[api.ResourceCPU]; !ok2 { + cpu, err = ir.getEstimation(api.ResourceCPU, c) + if err != nil { + glog.Errorf("Error while trying to estimate resources: %v", err) + } + } + } + if _, ok := req[api.ResourceMemory]; !ok { + if _, ok2 := lim[api.ResourceMemory]; !ok2 { + mem, err = ir.getEstimation(api.ResourceMemory, c) + if err != nil { + glog.Errorf("Error while trying to estimate resources: %v", err) + } + } + } + + // If Requests doesn't exits and an estimation was made, create Requests. + if req == nil && (cpu != nil || mem != nil) { + c.Resources.Requests = api.ResourceList{} + req = c.Resources.Requests + } + if cpu != nil { + glog.Infof("CPU estimation for container %v in pod %v/%v is %v", c.Name, pod.ObjectMeta.Namespace, pod.ObjectMeta.Name, cpu.String()) + req[api.ResourceCPU] = *cpu + } + if mem != nil { + glog.Infof("Memory estimation for container %v in pod %v/%v is %v", c.Name, pod.ObjectMeta.Namespace, pod.ObjectMeta.Name, mem.String()) + req[api.ResourceMemory] = *mem + } + } + // TODO(piosz): verify the estimates fits in LimitRanger +} + +func (ir initialResources) getEstimation(kind api.ResourceName, c *api.Container) (*resource.Quantity, error) { + end := time.Now() + start := end.Add(-week) + var usage, samples int64 + var err error + + // Historical data from last 7 days for the same image:tag. + if usage, samples, err = ir.source.GetUsagePercentile(kind, *percentile, c.Image, true, start, end); err != nil { + return nil, err + } + if samples < samplesThreshold { + // Historical data from last 30 days for the same image:tag. + start := end.Add(-month) + if usage, samples, err = ir.source.GetUsagePercentile(kind, *percentile, c.Image, true, start, end); err != nil { + return nil, err + } + } + if samples < samplesThreshold { + // Historical data from last 30 days for the same image. + start := end.Add(-month) + image := strings.Split(c.Image, ":")[0] + if usage, samples, err = ir.source.GetUsagePercentile(kind, *percentile, image, false, start, end); err != nil { + return nil, err + } + } + + if samples > 0 && kind == api.ResourceCPU { + return resource.NewMilliQuantity(usage, resource.DecimalSI), nil + } + if samples > 0 && kind == api.ResourceMemory { + return resource.NewQuantity(usage, resource.DecimalSI), nil + } + return nil, nil +} diff --git a/plugin/pkg/admission/initialresources/admission_test.go b/plugin/pkg/admission/initialresources/admission_test.go new file mode 100644 index 0000000000..2bfaada487 --- /dev/null +++ b/plugin/pkg/admission/initialresources/admission_test.go @@ -0,0 +1,192 @@ +/* +Copyright 2015 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package initialresources + +import ( + "testing" + "time" + + "k8s.io/kubernetes/pkg/admission" + "k8s.io/kubernetes/pkg/api" + "k8s.io/kubernetes/pkg/api/resource" +) + +type fakeSource struct { + f func(kind api.ResourceName, perc int64, image string, exactMatch bool, start, end time.Time) (int64, int64, error) +} + +func (s *fakeSource) GetUsagePercentile(kind api.ResourceName, perc int64, image string, exactMatch bool, start, end time.Time) (usage int64, samples int64, err error) { + return s.f(kind, perc, image, exactMatch, start, end) +} + +func parseReq(cpu, mem string) api.ResourceList { + if cpu == "" && mem == "" { + return nil + } + req := api.ResourceList{} + if cpu != "" { + req[api.ResourceCPU] = resource.MustParse(cpu) + } + if mem != "" { + req[api.ResourceMemory] = resource.MustParse(mem) + } + return req +} + +func addContainer(pod *api.Pod, name, image string, request api.ResourceList) { + pod.Spec.Containers = append(pod.Spec.Containers, api.Container{ + Name: name, + Image: image, + Resources: api.ResourceRequirements{Requests: request}, + }) +} + +func createPod(name string, image string, request api.ResourceList) api.Pod { + pod := api.Pod{ + ObjectMeta: api.ObjectMeta{Name: name, Namespace: "test"}, + Spec: api.PodSpec{}, + } + pod.Spec.Containers = []api.Container{} + addContainer(&pod, "c0", image, request) + return pod +} + +func getPods() []api.Pod { + return []api.Pod{ + createPod("p0", "image:v0", parseReq("", "")), + createPod("p1", "image:v1", parseReq("", "300")), + createPod("p2", "image:v2", parseReq("300m", "")), + createPod("p3", "image:v3", parseReq("300m", "300")), + } +} + +func verifyContainer(t *testing.T, c *api.Container, cpu, mem int64) { + req := c.Resources.Requests + if req.Cpu().MilliValue() != cpu { + t.Errorf("Wrong CPU request for container %v. Expected %v, got %v.", c.Name, cpu, req.Cpu().MilliValue()) + } + if req.Memory().Value() != mem { + t.Errorf("Wrong memory request for container %v. Expected %v, got %v.", c.Name, mem, req.Memory().Value()) + } +} + +func verifyPod(t *testing.T, pod *api.Pod, cpu, mem int64) { + verifyContainer(t, &pod.Spec.Containers[0], cpu, mem) +} + +func admit(t *testing.T, ir admission.Interface, pods []api.Pod) { + for i := range pods { + p := &pods[i] + if err := ir.Admit(admission.NewAttributesRecord(p, "Pod", "test", p.ObjectMeta.Name, "pods", "", admission.Create, nil)); err != nil { + t.Error(err) + } + } +} + +func TestEstimationBasedOnTheSameImage7d(t *testing.T) { + f := func(_ api.ResourceName, _ int64, _ string, exactMatch bool, start, end time.Time) (int64, int64, error) { + if exactMatch && end.Sub(start) == week { + return 100, 120, nil + } + return 200, 120, nil + } + ir := newInitialResources(&fakeSource{f: f}) + + pods := getPods() + admit(t, ir, pods) + + verifyPod(t, &pods[0], 100, 100) + verifyPod(t, &pods[1], 100, 300) + verifyPod(t, &pods[2], 300, 100) + verifyPod(t, &pods[3], 300, 300) +} + +func TestEstimationBasedOnTheSameImage30d(t *testing.T) { + f := func(_ api.ResourceName, _ int64, _ string, exactMatch bool, start, end time.Time) (int64, int64, error) { + if exactMatch && end.Sub(start) == week { + return 200, 20, nil + } + if exactMatch && end.Sub(start) == month { + return 100, 120, nil + } + return 200, 120, nil + } + ir := newInitialResources(&fakeSource{f: f}) + + pods := getPods() + admit(t, ir, pods) + + verifyPod(t, &pods[0], 100, 100) + verifyPod(t, &pods[1], 100, 300) + verifyPod(t, &pods[2], 300, 100) + verifyPod(t, &pods[3], 300, 300) +} + +func TestEstimationBasedOnOtherImages(t *testing.T) { + f := func(_ api.ResourceName, _ int64, image string, exactMatch bool, _, _ time.Time) (int64, int64, error) { + if image == "image" && !exactMatch { + return 100, 5, nil + } + return 200, 20, nil + } + ir := newInitialResources(&fakeSource{f: f}) + + pods := getPods() + admit(t, ir, pods) + + verifyPod(t, &pods[0], 100, 100) + verifyPod(t, &pods[1], 100, 300) + verifyPod(t, &pods[2], 300, 100) + verifyPod(t, &pods[3], 300, 300) +} + +func TestNoData(t *testing.T) { + f := func(_ api.ResourceName, _ int64, _ string, _ bool, _, _ time.Time) (int64, int64, error) { + return 200, 0, nil + } + ir := newInitialResources(&fakeSource{f: f}) + + pods := []api.Pod{ + createPod("p0", "image:v0", parseReq("", "")), + } + admit(t, ir, pods) + + if pods[0].Spec.Containers[0].Resources.Requests != nil { + t.Errorf("Unexpected resource estimation") + } +} + +func TestManyContainers(t *testing.T) { + f := func(_ api.ResourceName, _ int64, _ string, exactMatch bool, _, _ time.Time) (int64, int64, error) { + if exactMatch { + return 100, 120, nil + } + return 200, 30, nil + } + ir := newInitialResources(&fakeSource{f: f}) + + pod := createPod("p", "image:v0", parseReq("", "")) + addContainer(&pod, "c1", "image:v1", parseReq("", "300")) + addContainer(&pod, "c2", "image:v2", parseReq("300m", "")) + addContainer(&pod, "c3", "image:v3", parseReq("300m", "300")) + admit(t, ir, []api.Pod{pod}) + + verifyContainer(t, &pod.Spec.Containers[0], 100, 100) + verifyContainer(t, &pod.Spec.Containers[1], 100, 300) + verifyContainer(t, &pod.Spec.Containers[2], 300, 100) + verifyContainer(t, &pod.Spec.Containers[3], 300, 300) +} diff --git a/plugin/pkg/admission/initialresources/data_source.go b/plugin/pkg/admission/initialresources/data_source.go new file mode 100644 index 0000000000..b6bbbc15ea --- /dev/null +++ b/plugin/pkg/admission/initialresources/data_source.go @@ -0,0 +1,192 @@ +/* +Copyright 2015 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package initialresources + +import ( + "flag" + "fmt" + "strings" + "time" + + "github.com/golang/glog" + influxdb "github.com/influxdb/influxdb/client" + "k8s.io/kubernetes/pkg/api" +) + +const ( + cpuSeriesName = "autoscaling.cpu.usage.1m" + memSeriesName = "autoscaling.memory.usage.1m" + cpuContinuousQuery = "select derivative(value) as value from \"cpu/usage_ns_cumulative\" where pod_id <> '' group by pod_id, container_name, container_base_image, time(1m) into " + cpuSeriesName + memContinuousQuery = "select mean(value) as value from \"memory/usage_bytes_gauge\" where pod_id <> '' group by pod_id, container_name, container_base_image, time(1m) into " + memSeriesName + timeFormat = "2006-01-02 15:04:05" +) + +var ( + influxdbHost = flag.String("ir-influxdb-host", "localhost:8080/api/v1/proxy/namespaces/kube-system/services/monitoring-influxdb:api", "Address of InfluxDB which contains metrics requred by InitialResources") + user = flag.String("ir-user", "root", "User used for connecting to InfluxDB") + // TODO: figure out how to better pass password here + password = flag.String("ir-password", "root", "Password used for connecting to InfluxDB") + db = flag.String("ir-dbname", "k8s", "InfluxDB database name which contains metrics requred by InitialResources") +) + +// WARNING: If you are planning to add another implementation of dataSource interface please bear in mind, +// that dataSource will be moved to Heapster some time in the future and possibly rewritten. +type dataSource interface { + // Returns th of sample values which represent usage of for containers running , + // withing time range (start, end), number of samples considered and error if occured. + // If then take only samples that concern the same image (both name and take are the same), + // otherwise consider also samples with the same image a possibly different tag. + GetUsagePercentile(kind api.ResourceName, perc int64, image string, exactMatch bool, start, end time.Time) (usage int64, samples int64, err error) +} + +func newDataSource(kind string) (dataSource, error) { + if kind == "influxdb" { + return newInfluxdbSource(*influxdbHost, *user, *password, *db) + } + if kind == "gcm" { + return newGcmSource() + } + return nil, fmt.Errorf("Unknown data source %v", kind) +} + +// TODO(piosz): rewrite this once we will migrate into InfluxDB v0.9. +type influxdbSource struct { + conf *influxdb.ClientConfig +} + +func newInfluxdbSource(host, user, password, db string) (dataSource, error) { + conf := &influxdb.ClientConfig{ + Host: host, + Username: user, + Password: password, + Database: db, + } + source := &influxdbSource{ + conf: conf, + } + go source.ensureAutoscalingSeriesExist() + return source, nil +} + +func ensureSeriesExists(conn *influxdb.Client, existingQueries *influxdb.Series, seriesName, contQuery string) error { + queryExists := false + for _, p := range existingQueries.GetPoints() { + id := p[1].(float64) + query := p[2].(string) + if strings.Contains(query, "into "+seriesName) { + if query != contQuery { + if _, err := conn.Query(fmt.Sprintf("drop continuous query %v", id), influxdb.Second); err != nil { + return err + } + } else { + queryExists = true + } + } + } + if !queryExists { + if _, err := conn.Query("drop series "+seriesName, influxdb.Second); err != nil { + return err + } + if _, err := conn.Query(contQuery, influxdb.Second); err != nil { + return err + } + } + return nil +} + +func (s *influxdbSource) ensureAutoscalingSeriesExist() { + for { + time.Sleep(30 * time.Second) + client, err := influxdb.NewClient(s.conf) + if err != nil { + glog.Errorf("Error while trying to create InfluxDB client: %v", err) + continue + } + series, err := client.Query("list continuous queries", influxdb.Second) + if err != nil { + glog.Errorf("Error while trying to list continuous queries: %v", err) + continue + } + if err := ensureSeriesExists(client, series[0], cpuSeriesName, cpuContinuousQuery); err != nil { + glog.Errorf("Error while trying to create create autoscaling series: %v", err) + continue + } + if err := ensureSeriesExists(client, series[0], memSeriesName, memContinuousQuery); err != nil { + glog.Errorf("Error while trying to create create autoscaling series: %v", err) + continue + } + break + } +} + +func (s *influxdbSource) query(query string, precision ...influxdb.TimePrecision) ([]*influxdb.Series, error) { + client, err := influxdb.NewClient(s.conf) + if err != nil { + return nil, err + } + return client.Query(query, precision...) +} + +func (s *influxdbSource) GetUsagePercentile(kind api.ResourceName, perc int64, image string, exactMatch bool, start, end time.Time) (int64, int64, error) { + var series string + if kind == api.ResourceCPU { + series = cpuSeriesName + } else if kind == api.ResourceMemory { + series = memSeriesName + } + + var imgPattern string + if exactMatch { + imgPattern = "='" + image + "'" + } else { + imgPattern = "=~/^" + image + "/" + } + + query := fmt.Sprintf("select percentile(value, %v), count(pod_id) from %v where container_base_image%v and time > '%v' and time < '%v'", perc, series, imgPattern, start.UTC().Format(timeFormat), end.UTC().Format(timeFormat)) + var res []*influxdb.Series + var err error + if res, err = s.query(query, influxdb.Second); err != nil { + return 0, 0, fmt.Errorf("Error while trying to query InfluxDB: %v", err) + } + + // TODO(pszczesniak): fix issue with dropped data base + if len(res) == 0 { + return 0, 0, fmt.Errorf("Missing series %v in InfluxDB", series) + } + points := res[0].GetPoints() + if len(points) == 0 { + return 0, 0, fmt.Errorf("Missing data in series %v in InfluxDB", series) + } + p := points[0] + usage := p[1].(float64) + count := p[2].(float64) + if kind == api.ResourceCPU { + // convert from ns to millicores + usage = usage / 1000000 + } + return int64(usage), int64(count), nil +} + +type gcmSource struct{} + +func newGcmSource() (dataSource, error) { + return &gcmSource{}, fmt.Errorf("gcm source not implemented") +} + +func (s *gcmSource) GetUsagePercentile(kind api.ResourceName, perc int64, image string, exactMatch bool, start, end time.Time) (int64, int64, error) { + return 0, 0, fmt.Errorf("gcm source not implemented") +}