Merge pull request #15476 from burmanm/hawkular_save

Godeps/Godeps.json

@@ -695,6 +695,10 @@
 			"Rev": "d1e82c1ec3f15ee991f7cc7ffd5b67ff6f5bbaee"
 		},
 		{
+			"ImportPath": "github.com/hawkular/hawkular-client-go/metrics",
+			"Rev": "1d46ce7e1eca635f372357a8ccbf1fa7cc28b7d2"
+		},
+		{
 			"ImportPath": "github.com/imdario/mergo",
 			"Comment": "0.1.3-8-g6633656",
 			"Rev": "6633656539c1639d9d78127b7d47c622b5d7b6dc"

Godeps/LICENSES.md

@@ -54,6 +54,7 @@ github.com/hashicorp/go-msgpack | spdxBSD3
 github.com/hashicorp/golang-lru | IntelPart08
 github.com/hashicorp/raft | IntelPart08
 github.com/hashicorp/raft-boltdb | IntelPart08
+github.com/hawkular/hawkular-client-go | Apache-2
 github.com/imdario/mergo | spdxBSD3
 github.com/inconshreveable/mousetrap | Apache-2
 github.com/influxdb/influxdb | MITname

Godeps/_workspace/src/github.com/hawkular/hawkular-client-go/metrics/helpers.go

+package metrics
+
+import (
+	"fmt"
+	"math"
+	"strconv"
+	"time"
+)
+
+func ConvertToFloat64(v interface{}) (float64, error) {
+	switch i := v.(type) {
+	case float64:
+		return float64(i), nil
+	case float32:
+		return float64(i), nil
+	case int64:
+		return float64(i), nil
+	case int32:
+		return float64(i), nil
+	case int16:
+		return float64(i), nil
+	case int8:
+		return float64(i), nil
+	case uint64:
+		return float64(i), nil
+	case uint32:
+		return float64(i), nil
+	case uint16:
+		return float64(i), nil
+	case uint8:
+		return float64(i), nil
+	case int:
+		return float64(i), nil
+	case uint:
+		return float64(i), nil
+	case string:
+		f, err := strconv.ParseFloat(i, 64)
+		if err != nil {
+			return math.NaN(), err
+		}
+		return f, err
+	default:
+		return math.NaN(), fmt.Errorf("Cannot convert %s to float64", i)
+	}
+}
+
+// Returns milliseconds since epoch
+func UnixMilli(t time.Time) int64 {
+	return t.UnixNano() / 1e6
+}

Godeps/_workspace/src/github.com/hawkular/hawkular-client-go/metrics/types.go

+package metrics
+
+import (
+	"encoding/json"
+	"fmt"
+	// "time"
+)
+
+// MetricType restrictions
+type MetricType int
+
+const (
+	Gauge = iota
+	Availability
+	Counter
+	Generic
+)
+
+var longForm = []string{
+	"gauges",
+	"availability",
+	"counters",
+	"metrics",
+}
+
+var shortForm = []string{
+	"gauge",
+	"availability",
+	"counter",
+	"metrics",
+}
+
+func (self MetricType) validate() error {
+	if int(self) > len(longForm) && int(self) > len(shortForm) {
+		return fmt.Errorf("Given MetricType value %d is not valid", self)
+	}
+	return nil
+}
+
+func (self MetricType) String() string {
+	if err := self.validate(); err != nil {
+		return "unknown"
+	}
+	return longForm[self]
+}
+
+func (self MetricType) shortForm() string {
+	if err := self.validate(); err != nil {
+		return "unknown"
+	}
+	return shortForm[self]
+}
+
+// Custom unmarshaller
+func (self *MetricType) UnmarshalJSON(b []byte) error {
+	var f interface{}
+	err := json.Unmarshal(b, &f)
+	if err != nil {
+		return err
+	}
+
+	if str, ok := f.(string); ok {
+		for i, v := range shortForm {
+			if str == v {
+				*self = MetricType(i)
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
+func (self MetricType) MarshalJSON() ([]byte, error) {
+	return json.Marshal(self.String())
+}
+
+type SortKey struct {
+	Tenant string
+	Type   MetricType
+}
+
+// Hawkular-Metrics external structs
+// Do I need external.. hmph.
+
+type MetricHeader struct {
+	Tenant string      `json:"-"`
+	Type   MetricType  `json:"-"`
+	Id     string      `json:"id"`
+	Data   []Datapoint `json:"data"`
+}
+
+// Value should be convertible to float64 for numeric values
+// Timestamp is milliseconds since epoch
+type Datapoint struct {
+	Timestamp int64             `json:"timestamp"`
+	Value     interface{}       `json:"value"`
+	Tags      map[string]string `json:"tags,omitempty"`
+}
+
+type HawkularError struct {
+	ErrorMsg string `json:"errorMsg"`
+}
+
+type MetricDefinition struct {
+	Tenant        string            `json:"-"`
+	Type          MetricType        `json:"type,omitempty"`
+	Id            string            `json:"id"`
+	Tags          map[string]string `json:"tags,omitempty"`
+	RetentionTime int               `json:"dataRetention,omitempty"`
+}
+
+// TODO Fix the Start & End to return a time.Time
+type Bucketpoint struct {
+	Start       int64        `json:"start"`
+	End         int64        `json:"end"`
+	Min         float64      `json:"min"`
+	Max         float64      `json:"max"`
+	Avg         float64      `json:"avg"`
+	Median      float64      `json:"median"`
+	Empty       bool         `json:"empty"`
+	Samples     int64        `json:"samples"`
+	Percentiles []Percentile `json:"percentiles"`
+}
+
+type Percentile struct {
+	Quantile float64 `json:"quantile"`
+	Value    float64 `json:"value"`
+}

hack/verify-flags/known-flags.txt

@@ -158,6 +158,7 @@ ir-influxdb-host
 ir-namespace-only
 ir-password
 ir-user
+ir-hawkular
 jenkins-host
 jenkins-jobs
 k8s-build-output

plugin/pkg/admission/initialresources/data_source.go

@@ -28,8 +28,9 @@ 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")
+	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")
+	hawkularConfig = flag.String("ir-hawkular", "", "Hawkular configuration URL")
 )
 
 // WARNING: If you are planning to add another implementation of dataSource interface please bear in mind,
@@ -50,7 +51,7 @@ func newDataSource(kind string) (dataSource, error) {
 		return newGcmSource()
 	}
 	if kind == "hawkular" {
-		return newHawkularSource()
+		return newHawkularSource(*hawkularConfig)
 	}
 	return nil, fmt.Errorf("Unknown data source %v", kind)
 }

plugin/pkg/admission/initialresources/hawkular.go

@@ -17,18 +17,209 @@ limitations under the License.
 package initialresources
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
+	"github.com/golang/glog"
+	"github.com/hawkular/hawkular-client-go/metrics"
+	"io/ioutil"
+	"k8s.io/kubernetes/pkg/api"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
 	"time"
 
-	"k8s.io/kubernetes/pkg/api"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
+	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
+)
+
+type hawkularSource struct {
+	client       *metrics.Client
+	uri          *url.URL
+	useNamespace bool
+	modifiers    []metrics.Modifier
+}
+
+const (
+	containerImageTag string = "container_base_image"
+	descriptorTag     string = "descriptor_name"
+	separator         string = "/"
+
+	defaultServiceAccountFile = "/var/run/secrets/kubernetes.io/serviceaccount/token"
 )
 
-type hawkularSource struct{}
+// heapsterName gets the equivalent MetricDescriptor.Name used in the Heapster
+func heapsterName(kind api.ResourceName) string {
+	switch kind {
+	case api.ResourceCPU:
+		return "cpu/usage"
+	case api.ResourceMemory:
+		return "memory/usage"
+	default:
+		return ""
+	}
+}
+
+// tagQuery creates tagFilter query for Hawkular
+func tagQuery(kind api.ResourceName, image string, exactMatch bool) map[string]string {
+	q := make(map[string]string)
+
+	// Add here the descriptor_tag..
+	q[descriptorTag] = heapsterName(kind)
+
+	if exactMatch {
+		q[containerImageTag] = image
+	} else {
+		split := strings.Index(image, "@")
+		if split < 0 {
+			split = strings.Index(image, ":")
+		}
+		q[containerImageTag] = fmt.Sprintf("%s:*", image[:split])
+	}
+
+	return q
+}
+
+// dataSource API
+
+func (hs *hawkularSource) GetUsagePercentile(kind api.ResourceName, perc int64, image, namespace string, exactMatch bool, start, end time.Time) (int64, int64, error) {
+	q := tagQuery(kind, image, exactMatch)
+
+	m := make([]metrics.Modifier, len(hs.modifiers), 2+len(hs.modifiers))
+	copy(m, hs.modifiers)
+
+	if namespace != api.NamespaceAll {
+		m = append(m, metrics.Tenant(namespace))
+	}
+
+	p, err := metrics.ConvertToFloat64(perc)
+	if err != nil {
+		return 0, 0, err
+	}
+
+	m = append(m, metrics.Filters(metrics.TagsFilter(q), metrics.BucketsFilter(1), metrics.StartTimeFilter(start), metrics.EndTimeFilter(end), metrics.PercentilesFilter([]float64{p})))
 
-func newHawkularSource() (dataSource, error) {
-	return nil, fmt.Errorf("hawkular source not implemented")
+	bp, err := hs.client.ReadBuckets(metrics.Counter, m...)
+	if err != nil {
+		return 0, 0, err
+	}
+
+	if len(bp) > 0 && len(bp[0].Percentiles) > 0 {
+		return int64(bp[0].Percentiles[0].Value), int64(bp[0].Samples), nil
+	}
+	return 0, 0, nil
 }
 
-func (s *hawkularSource) GetUsagePercentile(kind api.ResourceName, perc int64, image, namespace string, exactMatch bool, start, end time.Time) (int64, int64, error) {
-	return 0, 0, fmt.Errorf("gcm source not implemented")
+// newHawkularSource creates a new Hawkular Source. The uri follows the scheme from Heapster
+func newHawkularSource(uri string) (dataSource, error) {
+	u, err := url.Parse(uri)
+	if err != nil {
+		return nil, err
+	}
+
+	d := &hawkularSource{
+		uri: u,
+	}
+	if err = d.init(); err != nil {
+		return nil, err
+	}
+	return d, nil
+}
+
+// init initializes the Hawkular dataSource. Almost equal to the Heapster initialization
+func (hs *hawkularSource) init() error {
+	hs.modifiers = make([]metrics.Modifier, 0)
+	p := metrics.Parameters{
+		Tenant: "heapster", // This data is stored by the heapster - for no-namespace hits
+		Url:    hs.uri.String(),
+	}
+
+	opts := hs.uri.Query()
+
+	if v, found := opts["tenant"]; found {
+		p.Tenant = v[0]
+	}
+
+	if v, found := opts["useServiceAccount"]; found {
+		if b, _ := strconv.ParseBool(v[0]); b {
+			accountFile := defaultServiceAccountFile
+			if file, f := opts["serviceAccountFile"]; f {
+				accountFile = file[0]
+			}
+
+			// If a readable service account token exists, then use it
+			if contents, err := ioutil.ReadFile(accountFile); err == nil {
+				p.Token = string(contents)
+			} else {
+				glog.Errorf("Could not read contents of %s, no token authentication is used\n", defaultServiceAccountFile)
+			}
+		}
+	}
+
+	// Authentication / Authorization parameters
+	tC := &tls.Config{}
+
+	if v, found := opts["auth"]; found {
+		if _, f := opts["caCert"]; f {
+			return fmt.Errorf("Both auth and caCert files provided, combination is not supported")
+		}
+		if len(v[0]) > 0 {
+			// Authfile
+			kubeConfig, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&clientcmd.ClientConfigLoadingRules{
+				ExplicitPath: v[0]},
+				&clientcmd.ConfigOverrides{}).ClientConfig()
+			if err != nil {
+				return err
+			}
+			tC, err = client.TLSConfigFor(kubeConfig)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	if u, found := opts["user"]; found {
+		if _, wrong := opts["useServiceAccount"]; wrong {
+			return fmt.Errorf("If user and password are used, serviceAccount cannot be used")
+		}
+		if p, f := opts["pass"]; f {
+			hs.modifiers = append(hs.modifiers, func(req *http.Request) error {
+				req.SetBasicAuth(u[0], p[0])
+				return nil
+			})
+		}
+	}
+
+	if v, found := opts["caCert"]; found {
+		caCert, err := ioutil.ReadFile(v[0])
+		if err != nil {
+			return err
+		}
+
+		caCertPool := x509.NewCertPool()
+		caCertPool.AppendCertsFromPEM(caCert)
+
+		tC.RootCAs = caCertPool
+	}
+
+	if v, found := opts["insecure"]; found {
+		insecure, err := strconv.ParseBool(v[0])
+		if err != nil {
+			return err
+		}
+		tC.InsecureSkipVerify = insecure
+	}
+
+	p.TLSConfig = tC
+
+	c, err := metrics.NewHawkularClient(p)
+	if err != nil {
+		return err
+	}
+
+	hs.client = c
+
+	glog.Infof("Initialised Hawkular Source with parameters %v", p)
+	return nil
 }

plugin/pkg/admission/initialresources/hawkular_test.go

+/*
+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 (
+	"fmt"
+	"k8s.io/kubernetes/pkg/api"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	assert "github.com/stretchr/testify/require"
+)
+
+const (
+	testImageName    string = "hawkular/hawkular-metrics"
+	testImageVersion string = "latest"
+	testImageSHA     string = "b727ece3780cdd30e9a86226e520f26bcc396071ed7a86b7ef6684bb93a9f717"
+	testPartialMatch string = "hawkular/hawkular-metrics:*"
+)
+
+func testImageWithVersion() string {
+	return fmt.Sprintf("%s:%s", testImageName, testImageVersion)
+}
+
+func testImageWithReference() string {
+	return fmt.Sprintf("%s@sha256:%s", testImageName, testImageSHA)
+}
+
+func TestTaqQuery(t *testing.T) {
+	kind := api.ResourceCPU
+	tQ := tagQuery(kind, testImageWithVersion(), false)
+
+	assert.Equal(t, 2, len(tQ))
+	assert.Equal(t, testPartialMatch, tQ[containerImageTag])
+	assert.Equal(t, "cpu/usage", tQ[descriptorTag])
+
+	tQe := tagQuery(kind, testImageWithVersion(), true)
+	assert.Equal(t, 2, len(tQe))
+	assert.Equal(t, testImageWithVersion(), tQe[containerImageTag])
+	assert.Equal(t, "cpu/usage", tQe[descriptorTag])
+
+	tQr := tagQuery(kind, testImageWithReference(), false)
+	assert.Equal(t, 2, len(tQe))
+	assert.Equal(t, testPartialMatch, tQr[containerImageTag])
+	assert.Equal(t, "cpu/usage", tQr[descriptorTag])
+
+	tQre := tagQuery(kind, testImageWithReference(), true)
+	assert.Equal(t, 2, len(tQe))
+	assert.Equal(t, testImageWithReference(), tQre[containerImageTag])
+	assert.Equal(t, "cpu/usage", tQre[descriptorTag])
+}
+
+func TestGetUsagePercentile(t *testing.T) {
+	tenant := "16a8884e4c155457ee38a8901df6b536"
+	reqs := make(map[string]string)
+
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, tenant, r.Header.Get("Hawkular-Tenant"))
+		assert.Equal(t, "Basic", r.Header.Get("Authorization")[:5])
+
+		if strings.Contains(r.RequestURI, "counters/data") {
+			assert.True(t, strings.Contains(r.RequestURI, url.QueryEscape(testImageWithVersion())))
+			assert.True(t, strings.Contains(r.RequestURI, "cpu%2Fusage"))
+			assert.True(t, strings.Contains(r.RequestURI, "percentiles=90"))
+
+			reqs["counters/data"] = r.RequestURI
+			fmt.Fprintf(w, ` [{"start":1444620095882,"end":1444648895882,"min":1.45,"avg":1.45,"median":1.45,"max":1.45,"percentile95th":1.45,"samples":123456,"percentiles":[{"value":7896.54,"quantile":0.9},{"value":1.45,"quantile":0.99}],"empty":false}]`)
+		} else {
+			reqs["unknown"] = r.RequestURI
+		}
+	}))
+
+	paramUri := fmt.Sprintf("%s?user=test&pass=yep", s.URL)
+
+	hSource, err := newHawkularSource(paramUri)
+	assert.NoError(t, err)
+
+	usage, samples, err := hSource.GetUsagePercentile(api.ResourceCPU, 90, testImageWithVersion(), "16a8884e4c155457ee38a8901df6b536", true, time.Now(), time.Now())
+	assert.NoError(t, err)
+
+	assert.Equal(t, 1, len(reqs))
+	assert.Equal(t, "", reqs["unknown"])
+
+	assert.Equal(t, int64(123456), int64(samples))
+	assert.Equal(t, int64(7896), usage) // float64 -> int64
+}