Merge pull request #13132 from piosz/initial

Implemented first version of InitialResources plugin.

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"

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

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
+}

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)
+}

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 <perc>th of sample values which represent usage of <kind> for containers running <image>,
+	// withing time range (start, end), number of samples considered and error if occured.
+	// If <exactMatch> 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")
+}