Merge pull request #276 from luksa/kubelet_https

Added support for connecting to kubelets through https (using the same certificates as for kubernetes master)

docs/source-configuration.md

@@ -33,6 +33,7 @@ The following options are available:
 * `apiVersion` - API version to use to talk to Kubernetes (default: `v1beta1`)
 * `insecure` - whether to trust kubernetes certificates (default: `false`)
 * `kubeletPort` - kubelet port to use (default: `10255`)
+* `kubeletHttps` - whether to use https to connect to kubelets (default: `false`)
 * `auth` - client auth file to use (default: /etc/kubernetes/kubeConfig/kubeConfig)
 
 ### Cadvisor

sources/datasource/kubelet.go

@@ -26,14 +26,18 @@ import (
 	"time"
 
 	"github.com/GoogleCloudPlatform/heapster/sources/api"
+	kube_client "github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/golang/glog"
 	cadvisor "github.com/google/cadvisor/info/v1"
 )
 
 // TODO(vmarmol): Use Kubernetes' if we export it as an API.
 const kubernetesPodNameLabel = "io.kubernetes.pod.name"
 
-type kubeletSource struct{}
+type kubeletSource struct {
+	config *kube_client.KubeletConfig
+	client *http.Client
+}
 
 func (self *kubeletSource) postRequestAndGetValue(client *http.Client, req *http.Request, value interface{}) error {
 	response, err := client.Do(req)
@@ -83,7 +87,11 @@ func (self *kubeletSource) getContainer(url string, start, end time.Time, resolu
 	}
 	req.Header.Set("Content-Type", "application/json")
 	var containerInfo cadvisor.ContainerInfo
-	err = self.postRequestAndGetValue(http.DefaultClient, req, &containerInfo)
+	client := self.client
+	if client == nil {
+		client = http.DefaultClient
+	}
+	err = self.postRequestAndGetValue(client, req, &containerInfo)
 	if err != nil {
 		glog.V(2).Infof("failed to get stats from kubelet url: %s - %s\n", url, err)
 		return nil, err
@@ -93,7 +101,14 @@ func (self *kubeletSource) getContainer(url string, start, end time.Time, resolu
 }
 
 func (self *kubeletSource) GetContainer(host Host, start, end time.Time, resolution time.Duration) (container *api.Container, err error) {
-	url := fmt.Sprintf("http://%s:%d/%s", host.IP, host.Port, host.Resource)
+	var schema string
+	if self.config != nil && self.config.EnableHttps {
+		schema = "https"
+	} else {
+		schema = "http"
+	}
+
+	url := fmt.Sprintf("%s://%s:%d/%s", schema, host.IP, host.Port, host.Resource)
 	glog.V(3).Infof("about to query kubelet using url: %q", url)
 
 	return self.getContainer(url, start, end, resolution)

sources/datasource/types.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/GoogleCloudPlatform/heapster/sources/api"
+	kube_client "github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 )
 
 type Host struct {
@@ -45,6 +46,14 @@ type Kubelet interface {
 	GetAllRawContainers(host Host, start, end time.Time, resolution time.Duration) ([]api.Container, error)
 }
 
-func NewKubelet() Kubelet {
-	return &kubeletSource{}
+func NewKubelet(kubeletConfig *kube_client.KubeletConfig) (Kubelet, error) {
+	kubeletHttpClient, err := kube_client.NewKubeletHttpClient(kubeletConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return &kubeletSource{
+		config: kubeletConfig,
+		client: kubeletHttpClient,
+	}, nil
 }

sources/kube_factory.go

@@ -35,6 +35,7 @@ const (
 	defaultInsecure       = false
 	defaultKubeletPort    = 10255
 	defaultKubeConfigFile = "/etc/kubernetes/kubeconfig/kubeconfig"
+	defaultKubeletHttps   = false
 )
 
 func init() {
@@ -113,10 +114,29 @@ func CreateKubeSources(uri string, options map[string][]string) ([]api.Source, e
 			return nil, err
 		}
 	}
+
+	kubeletHttps := defaultKubeletHttps
+	if len(options["kubeletHttps"]) >= 1 {
+		kubeletHttps, err = strconv.ParseBool(options["kubeletHttps"][0])
+		if err != nil {
+			return nil, err
+		}
+	}
 	glog.Infof("Using Kubernetes client with master %q and version %q\n", kubeConfig.Host, kubeConfig.Version)
 	glog.Infof("Using kubelet port %d", kubeletPort)
-	kubeletApi := datasource.NewKubelet()
-	kubePodsSource := NewKubePodMetrics(kubeletPort, nodesApi, newPodsApi(kubeClient), kubeletApi)
+
+	kubeletConfig := &kube_client.KubeletConfig{
+		Port:            uint(kubeletPort),
+		EnableHttps:     kubeletHttps,
+		TLSClientConfig: kubeConfig.TLSClientConfig,
+	}
+
+	kubeletApi, err := datasource.NewKubelet(kubeletConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	kubePodsSource := NewKubePodMetrics(kubeletPort, kubeletApi, nodesApi, newPodsApi(kubeClient))
 	kubeNodeSource := NewKubeNodeMetrics(kubeletPort, kubeletApi, nodesApi)
 	kubeEventsSource := NewKubeEvents(kubeClient)
 

sources/kube_pods.go

@@ -37,7 +37,7 @@ type kubePodsSource struct {
 
 const name = "kube-pods-source"
 
-func NewKubePodMetrics(kubeletPort int, nodesApi nodes.NodesApi, podsApi podsApi, kubeletApi datasource.Kubelet) api.Source {
+func NewKubePodMetrics(kubeletPort int, kubeletApi datasource.Kubelet, nodesApi nodes.NodesApi, podsApi podsApi) api.Source {
 	return &kubePodsSource{
 		kubeletPort: kubeletPort,
 		kubeletApi:  kubeletApi,

sources/kube_pods_test.go

@@ -38,7 +38,7 @@ func (self *fakePodsApi) DebugInfo() string {
 func TestKubePodMetricsBasic(t *testing.T) {
 	nodesApi := &fakeNodesApi{nodes.NodeList{}}
 	podsApi := &fakePodsApi{[]api.Pod{}}
-	source := NewKubePodMetrics(10250, nodesApi, podsApi, &fakeKubeletApi{})
+	source := NewKubePodMetrics(10250, &fakeKubeletApi{}, nodesApi, podsApi)
 	_, err := source.GetInfo(time.Now(), time.Now().Add(time.Minute), time.Second)
 	require.NoError(t, err)
 	require.NotEmpty(t, source.DebugInfo())
@@ -65,7 +65,7 @@ func TestKubePodMetricsFull(t *testing.T) {
 	nodesApi := &fakeNodesApi{nodeList}
 	podsApi := &fakePodsApi{podList}
 	kubeletApi := &fakeKubeletApi{container: container}
-	source := NewKubePodMetrics(10250, nodesApi, podsApi, kubeletApi)
+	source := NewKubePodMetrics(10250, kubeletApi, nodesApi, podsApi)
 	data, err := source.GetInfo(time.Now(), time.Now().Add(time.Minute), time.Second)
 	require.NoError(t, err)
 	require.NotEmpty(t, data)