Add support for ServiceAccount auth to kubeletstats

receiver/kubeletstatsreceiver/README.md

@@ -23,6 +23,13 @@ TLS tells this receiver to use TLS for auth and requires that the fields
 ServiceAccount tells this receiver to use the default service account token
 to authenticate to the kubelet API.
 
+Note: a missing or empty `endpoint` will cause the hostname on which the collector
+is running to be used as the endpoint. If the hostNetwork flag is set, and the
+collector is running in a pod, this hostname will resolve to the node's network
+namespace.
+
+### TLS Example
+
 ```yaml
 receivers:
   kubeletstats:
@@ -42,3 +49,37 @@ service:
       receivers: [kubeletstats]
       exporters: [file]
 ```
+
+### ServiceAccount Example
+
+Although it's possible to use kubernetes' hostNetwork feature to talk to the
+kubelet api from a pod, the preferred approach is to use the downard API.
+
+Make sure the pod spec sets the node name as follows:
+
+```yaml
+env:
+  - name: K8S_NODE_NAME
+    valueFrom:
+      fieldRef:
+        fieldPath: spec.nodeName
+```
+
+Then the otel config can reference the `K8S_NODE_NAME` environment variable:
+
+```yaml
+receivers:
+  kubeletstats:
+    collection_interval: 20s
+    auth_type: "serviceAccount"
+    endpoint: "${K8S_NODE_NAME}:10250"
+    insecure_skip_verify: true
+exporters:
+  file:
+    path: "fileexporter.txt"
+service:
+  pipelines:
+    metrics:
+      receivers: [kubeletstats]
+      exporters: [file]
+```    

receiver/kubeletstatsreceiver/config_test.go

@@ -0,0 +1,78 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// 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 kubeletstatsreceiver
+
+import (
+	"path"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/config/configtls"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/k8sconfig"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver/kubelet"
+)
+
+func TestLoadConfig(t *testing.T) {
+	factories, err := config.ExampleComponents()
+	require.NoError(t, err)
+	factory := &Factory{}
+	factories.Receivers[configmodels.Type(typeStr)] = factory
+	cfg, err := config.LoadConfigFile(
+		t, path.Join(".", "testdata", "config.yaml"), factories,
+	)
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	tlsCfg := cfg.Receivers["kubeletstats/tls"].(*Config)
+	duration := 10 * time.Second
+	require.Equal(t, &Config{
+		ReceiverSettings: configmodels.ReceiverSettings{
+			TypeVal:  "kubeletstats",
+			NameVal:  "kubeletstats/tls",
+			Endpoint: "1.2.3.4:5555",
+		},
+		ClientConfig: kubelet.ClientConfig{
+			APIConfig: k8sconfig.APIConfig{
+				AuthType: "tls",
+			},
+			TLSSetting: configtls.TLSSetting{
+				CAFile:   "/path/to/ca.crt",
+				CertFile: "/path/to/apiserver.crt",
+				KeyFile:  "/path/to/apiserver.key",
+			},
+			InsecureSkipVerify: true,
+		},
+		CollectionInterval: duration,
+	}, tlsCfg)
+
+	saCfg := cfg.Receivers["kubeletstats/sa"].(*Config)
+	require.Equal(t, &Config{
+		ReceiverSettings: configmodels.ReceiverSettings{
+			TypeVal: "kubeletstats",
+			NameVal: "kubeletstats/sa",
+		},
+		ClientConfig: kubelet.ClientConfig{
+			APIConfig: k8sconfig.APIConfig{
+				AuthType: "serviceAccount",
+			},
+			InsecureSkipVerify: true,
+		},
+		CollectionInterval: duration,
+	}, saCfg)
+}

receiver/kubeletstatsreceiver/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/golang/protobuf v1.3.5
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/redisreceiver v0.0.0-20200518175917-05cf2ea24e6c
+	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.5.1
 	go.opentelemetry.io/collector v0.3.1-0.20200612184320-01ce74db9c44
 	go.uber.org/zap v1.10.0

receiver/kubeletstatsreceiver/kubelet/cert.go

@@ -16,22 +16,23 @@ package kubelet
 
 import (
 	"crypto/x509"
-	"errors"
 	"io/ioutil"
+
+	"github.com/pkg/errors"
 )
 
 func systemCertPoolPlusPath(certPath string) (*x509.CertPool, error) {
 	sysCerts, err := x509.SystemCertPool()
 	if err != nil {
-		return nil, err
+		return nil, errors.WithMessage(err, "Could not load system x509 cert pool")
 	}
 	return certPoolPlusPath(sysCerts, certPath)
 }
 
 func certPoolPlusPath(certPool *x509.CertPool, certPath string) (*x509.CertPool, error) {
 	certBytes, err := ioutil.ReadFile(certPath)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrapf(err, "Cert path %s could not be read", certPath)
 	}
 	ok := certPool.AppendCertsFromPEM(certBytes)
 	if !ok {

receiver/kubeletstatsreceiver/kubelet/client.go

@@ -20,41 +20,32 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"os"
 
+	"github.com/pkg/errors"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/k8sconfig"
 )
 
-// Config for a kubelet Client. Mostly for talking to the kubelet HTTP endpoint.
-type ClientConfig struct {
-	k8sconfig.APIConfig `mapstructure:",squash"`
-	// Path to the CA cert. For a client this verifies the server certificate.
-	// For a server this verifies client certificates. If empty uses system root CA.
-	// (optional)
-	CAFile string `mapstructure:"ca_file"`
-	// Path to the TLS cert to use for TLS required connections. (optional)
-	CertFile string `mapstructure:"cert_file"`
-	// Path to the TLS key to use for TLS required connections. (optional)
-	// TODO replace with open-telemetry/opentelemetry-collector#933 when done
-	KeyFile string `mapstructure:"key_file"`
-	// InsecureSkipVerify controls whether the client verifies the server's
-	// certificate chain and host name.
-	InsecureSkipVerify bool `mapstructure:"insecure_skip_verify"`
-}
+const svcAcctCACertPath = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+const svcAcctTokenPath = "/var/run/secrets/kubernetes.io/serviceaccount/token"
 
 type Client interface {
 	Get(path string) ([]byte, error)
 }
 
 func NewClient(endpoint string, cfg *ClientConfig, logger *zap.Logger) (Client, error) {
-	if cfg.APIConfig.AuthType != k8sconfig.AuthTypeTLS {
+	switch cfg.APIConfig.AuthType {
+	case k8sconfig.AuthTypeTLS:
+		return newTLSClient(endpoint, cfg, logger)
+	case k8sconfig.AuthTypeServiceAccount:
+		return newServiceAccountClient(endpoint, svcAcctCACertPath, svcAcctTokenPath, logger)
+	default:
 		return nil, fmt.Errorf("AuthType [%s] not supported", cfg.APIConfig.AuthType)
 	}
-	return newTLSClient(endpoint, cfg, logger)
 }
 
-// not unit tested
 func newTLSClient(endpoint string, cfg *ClientConfig, logger *zap.Logger) (Client, error) {
 	rootCAs, err := systemCertPoolPlusPath(cfg.CAFile)
 	if err != nil {
@@ -64,44 +55,99 @@ func newTLSClient(endpoint string, cfg *ClientConfig, logger *zap.Logger) (Clien
 	if err != nil {
 		return nil, err
 	}
-	return defaultTLSClient(endpoint, cfg.InsecureSkipVerify, rootCAs, clientCert, logger), nil
+	return defaultTLSClient(
+		endpoint,
+		cfg.InsecureSkipVerify,
+		rootCAs,
+		[]tls.Certificate{clientCert},
+		nil,
+		logger,
+	)
 }
 
-func defaultTLSClient(endpoint string, insecureSkipVerify bool, rootCAs *x509.CertPool, clientCert tls.Certificate, logger *zap.Logger) *tlsClient {
+func newServiceAccountClient(
+	endpoint string,
+	caCertPath string,
+	tokenPath string,
+	logger *zap.Logger,
+) (*clientImpl, error) {
+	rootCAs, err := systemCertPoolPlusPath(caCertPath)
+	if err != nil {
+		return nil, err
+	}
+	tok, err := ioutil.ReadFile(tokenPath)
+	if err != nil {
+		return nil, err
+	}
+	tr := defaultTransport()
+	tr.TLSClientConfig = &tls.Config{
+		RootCAs: rootCAs,
+	}
+	return defaultTLSClient(endpoint, true, rootCAs, nil, tok, logger)
+}
+
+func defaultTLSClient(
+	endpoint string,
+	insecureSkipVerify bool,
+	rootCAs *x509.CertPool,
+	certificates []tls.Certificate,
+	tok []byte,
+	logger *zap.Logger,
+) (*clientImpl, error) {
 	tr := defaultTransport()
 	tr.TLSClientConfig = &tls.Config{
 		RootCAs:            rootCAs,
-		Certificates:       []tls.Certificate{clientCert},
+		Certificates:       certificates,
 		InsecureSkipVerify: insecureSkipVerify,
 	}
-	return &tlsClient{
+	if endpoint == "" {
+		var err error
+		endpoint, err = defaultEndpoint()
+		if err != nil {
+			return nil, err
+		}
+		logger.Warn("Kubelet endpoint not defined, using default endpoint " + endpoint)
+	}
+	return &clientImpl{
 		baseURL:    "https://" + endpoint,
 		httpClient: http.Client{Transport: tr},
+		tok:        tok,
 		logger:     logger,
+	}, nil
+}
+
+// This will work if hostNetwork is turned on, in which case the pod has access
+// to the node's loopback device.
+// https://kubernetes.io/docs/concepts/policy/pod-security-policy/#host-namespaces
+func defaultEndpoint() (string, error) {
+	hostname, err := os.Hostname()
+	if err != nil {
+		return "", errors.WithMessage(err, "Unable to get hostname for default endpoint")
 	}
+	const kubeletPort = "10250"
+	return hostname + ":" + kubeletPort, nil
 }
 
 func defaultTransport() *http.Transport {
 	return http.DefaultTransport.(*http.Transport).Clone()
 }
 
-// tlsClient
+// clientImpl
 
-var _ Client = (*tlsClient)(nil)
+var _ Client = (*clientImpl)(nil)
 
-type tlsClient struct {
+type clientImpl struct {
 	baseURL    string
 	httpClient http.Client
 	logger     *zap.Logger
+	tok        []byte
 }
 
-func (c *tlsClient) Get(path string) ([]byte, error) {
-	url := c.baseURL + path
-	req, err := http.NewRequest("GET", url, nil)
+func (c *clientImpl) Get(path string) ([]byte, error) {
+	req, err := c.buildReq(path)
 	if err != nil {
 		return nil, err
 	}
-	req.Header.Set("Content-Type", "application/json")
 	resp, err := c.httpClient.Do(req)
 	if err != nil {
 		return nil, err
@@ -118,3 +164,16 @@ func (c *tlsClient) Get(path string) ([]byte, error) {
 	}
 	return body, nil
 }
+
+func (c *clientImpl) buildReq(path string) (*http.Request, error) {
+	url := c.baseURL + path
+	req, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if c.tok != nil {
+		req.Header.Set("Authorization", fmt.Sprintf("bearer %s", c.tok))
+	}
+	return req, nil
+}

receiver/kubeletstatsreceiver/kubelet/client_config.go

@@ -0,0 +1,30 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// 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 kubelet
+
+import (
+	"go.opentelemetry.io/collector/config/configtls"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/k8sconfig"
+)
+
+// Config for a kubelet client for talking to a kubelet HTTP endpoint.
+type ClientConfig struct {
+	k8sconfig.APIConfig  `mapstructure:",squash"`
+	configtls.TLSSetting `mapstructure:",squash"`
+	// InsecureSkipVerify controls whether the client verifies the server's
+	// certificate chain and host name.
+	InsecureSkipVerify bool `mapstructure:"insecure_skip_verify"`
+}