Merge pull request kubernetes-retired#413 from mvdan/uris-type

Uris type

api/v1/api.go

@@ -201,16 +201,23 @@ func (a *Api) setSinks(req *restful.Request, resp *restful.Response) {
 		resp.WriteError(http.StatusBadRequest, err)
 		return
 	}
-	if err := a.manager.SetSinkUris(*sinkUris); err != nil {
+	var uris manager.Uris
+	for _, s := range *sinkUris {
+		if err := uris.Set(s); err != nil {
+			resp.WriteError(http.StatusBadRequest, err)
+			return
+		}
+	}
+	if err := a.manager.SetSinkUris(uris); err != nil {
 		resp.WriteError(http.StatusInternalServerError, err)
 		return
 	}
 }
 
 func (a *Api) getSinks(req *restful.Request, resp *restful.Response) {
-	sinks := a.manager.SinkUris()
-	if sinks == nil {
-		sinks = make([]string, 0)
+	sinkUris := a.manager.SinkUris()
+	if sinkUris == nil {
+		sinkUris = make(manager.Uris, 0)
 	}
-	resp.WriteEntity(sinks)
+	resp.WriteEntity(sinkUris)
 }

extpoints/interfaces.go

@@ -15,10 +15,12 @@
 package extpoints
 
 import (
+	"net/url"
+
 	sinksApi "github.com/GoogleCloudPlatform/heapster/sinks/api/v1"
 	sourceApi "github.com/GoogleCloudPlatform/heapster/sources/api"
 )
 
-type SourceFactory func(string, map[string][]string) ([]sourceApi.Source, error)
+type SourceFactory func(*url.URL) ([]sourceApi.Source, error)
 
-type SinkFactory func(string, map[string][]string) ([]sinksApi.ExternalSink, error)
+type SinkFactory func(*url.URL) ([]sinksApi.ExternalSink, error)

heapster.go

@@ -26,7 +26,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/heapster/manager"
 	"github.com/GoogleCloudPlatform/heapster/sinks"
-	"github.com/GoogleCloudPlatform/heapster/sources/api"
+	source_api "github.com/GoogleCloudPlatform/heapster/sources/api"
 	"github.com/GoogleCloudPlatform/heapster/version"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/golang/glog"
@@ -40,8 +40,8 @@ var (
 	argIp              = flag.String("listen_ip", "", "IP to listen on, defaults to all IPs")
 	argMaxProcs        = flag.Int("max_procs", 0, "max number of CPUs that can be used simultaneously. Less than 1 for default (number of cores).")
 	argCacheDuration   = flag.Duration("cache_duration", 10*time.Minute, "The total duration of the historical data that will be cached by heapster.")
-	argSources         Uris
-	argSinks           Uris
+	argSources         manager.Uris
+	argSinks           manager.Uris
 )
 
 func main() {
@@ -79,7 +79,7 @@ func validateFlags() error {
 	return nil
 }
 
-func doWork() ([]api.Source, sinks.ExternalSinkManager, manager.Manager, error) {
+func doWork() ([]source_api.Source, sinks.ExternalSinkManager, manager.Manager, error) {
 	sources, err := newSources()
 	if err != nil {
 		return nil, nil, nil, err

manager/flags.go

@@ -0,0 +1,77 @@
+// Copyright 2015 Google Inc. 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 manager
+
+import (
+	"bytes"
+	"fmt"
+	"net/url"
+	"os"
+	"strings"
+)
+
+type Uri struct {
+	Key string
+	Val url.URL
+}
+
+func (u *Uri) String() string {
+	val := u.Val.String()
+	if val == "" {
+		return fmt.Sprintf("%s", u.Key)
+	}
+	return fmt.Sprintf("%s:%s", u.Key, val)
+}
+
+func (u *Uri) Set(value string) error {
+	s := strings.SplitN(value, ":", 2)
+	if s[0] == "" {
+		return fmt.Errorf("missing uri key in '%s'", value)
+	}
+	u.Key = s[0]
+	if len(s) > 1 && s[1] != "" {
+		e := os.ExpandEnv(s[1])
+		uri, err := url.Parse(e)
+		if err != nil {
+			return err
+		}
+		u.Val = *uri
+	}
+	return nil
+}
+
+type Uris []Uri
+
+func (us *Uris) String() string {
+	var b bytes.Buffer
+	b.WriteString("[")
+	for i, u := range *us {
+		if i > 0 {
+			b.WriteString(" ")
+		}
+		b.WriteString(u.String())
+	}
+	b.WriteString("]")
+	return b.String()
+}
+
+func (us *Uris) Set(value string) error {
+	var u Uri
+	if err := u.Set(value); err != nil {
+		return err
+	}
+	*us = append(*us, u)
+	return nil
+}

manager/flags_test.go

@@ -0,0 +1,166 @@
+// Copyright 2015 Google Inc. 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 manager
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUriString(t *testing.T) {
+	tests := [...]struct {
+		in   Uri
+		want string
+	}{
+		{
+			Uri{
+				Key: "gcm",
+			},
+			"gcm",
+		},
+		{
+			Uri{
+				Key: "influxdb",
+				Val: url.URL{
+					Scheme:   "http",
+					Host:     "monitoring-influxdb:8086",
+					RawQuery: "key=val&key2=val2",
+				},
+			},
+			"influxdb:http://monitoring-influxdb:8086?key=val&key2=val2",
+		},
+	}
+	for _, c := range tests {
+		assert.Equal(t, c.want, c.in.String())
+	}
+}
+
+func TestUriSet(t *testing.T) {
+	tests := [...]struct {
+		in      string
+		want    Uri
+		wantErr bool
+	}{
+		{"", Uri{}, true},
+		{":", Uri{}, true},
+		{":foo", Uri{}, true},
+		{"key:incorrecturl/%gh&%ij", Uri{}, true},
+		{
+			"gcm",
+			Uri{Key: "gcm"},
+			false,
+		},
+		{
+			"gcm:",
+			Uri{Key: "gcm"},
+			false,
+		},
+		{
+			"influxdb:http://monitoring-influxdb:8086?key=val&key2=val2",
+			Uri{
+				Key: "influxdb",
+				Val: url.URL{
+					Scheme:   "http",
+					Host:     "monitoring-influxdb:8086",
+					RawQuery: "key=val&key2=val2",
+				},
+			},
+			false,
+		},
+	}
+	for _, c := range tests {
+		var uri Uri
+		err := uri.Set(c.in)
+		if c.wantErr {
+			assert.NotNil(t, err)
+		} else {
+			assert.Nil(t, err)
+			assert.Equal(t, c.want, uri)
+		}
+	}
+}
+
+func TestUrisString(t *testing.T) {
+	tests := [...]struct {
+		in   Uris
+		want string
+	}{
+		{
+			Uris{
+				Uri{Key: "gcm"},
+			},
+			"[gcm]",
+		},
+		{
+			Uris{
+				Uri{Key: "gcm"},
+				Uri{
+					Key: "influxdb",
+					Val: url.URL{Path: "foo"},
+				},
+			},
+			"[gcm influxdb:foo]",
+		},
+	}
+	for _, c := range tests {
+		assert.Equal(t, c.want, c.in.String())
+	}
+}
+
+func TestUrisSet(t *testing.T) {
+	tests := [...]struct {
+		in      []string
+		want    Uris
+		wantErr bool
+	}{
+		{[]string{""}, Uris{}, true},
+		{[]string{":foo"}, Uris{}, true},
+		{
+			[]string{"gcm"},
+			Uris{
+				Uri{Key: "gcm"},
+			},
+			false,
+		},
+		{
+			[]string{"gcm", "influxdb:foo"},
+			Uris{
+				Uri{Key: "gcm"},
+				Uri{
+					Key: "influxdb",
+					Val: url.URL{Path: "foo"},
+				},
+			},
+			false,
+		},
+	}
+	for _, c := range tests {
+		var uris Uris
+		var err error
+		for _, s := range c.in {
+			if err = uris.Set(s); err != nil {
+				break
+			}
+		}
+		if c.wantErr {
+			assert.NotNil(t, err)
+		} else {
+			assert.Nil(t, err)
+			assert.Equal(t, c.want, uris)
+		}
+	}
+}

manager/manager.go

@@ -37,17 +37,17 @@ type Manager interface {
 	ExportMetrics() ([]*sink_api.Point, error)
 
 	// Set the sinks to use
-	SetSinkUris([]string) error
+	SetSinkUris(Uris) error
 
 	// Get the sinks currently in use
-	SinkUris() []string
+	SinkUris() Uris
 }
 
 type realManager struct {
 	sources     []source_api.Source
 	cache       cache.Cache
 	sinkManager sinks.ExternalSinkManager
-	sinkUris    []string
+	sinkUris    Uris
 	lastSync    time.Time
 	resolution  time.Duration
 	align       bool
@@ -181,18 +181,18 @@ func onlyKeepLatestStat(cont *cache.ContainerElement) {
 	}
 }
 
-func (rm *realManager) SetSinkUris(sinkUris []string) error {
-	externalSinks, err := newSinks(sinkUris)
+func (rm *realManager) SetSinkUris(sinkUris Uris) error {
+	sinks, err := newSinks(sinkUris)
 	if err != nil {
 		return err
 	}
-	if err := rm.sinkManager.SetSinks(externalSinks); err != nil {
+	if err := rm.sinkManager.SetSinks(sinks); err != nil {
 		return err
 	}
 	rm.sinkUris = sinkUris
 	return nil
 }
 
-func (rm *realManager) SinkUris() []string {
+func (rm *realManager) SinkUris() Uris {
 	return rm.sinkUris
 }