Add trailing newline to stats response

This was just an annoyance when working with the API from a terminal.

Signed-off-by: Brian Goff <cpuguy83@gmail.com>

cmd/virtual-kubelet/internal/commands/root/http.go

@@ -16,31 +16,21 @@ package root
 
 import (
 	"fmt"
-	"os"
 	"time"
 )
 
 type apiServerConfig struct {
-	CertPath              string
-	KeyPath               string
-	CACertPath            string
 	Addr                  string
 	MetricsAddr           string
 	StreamIdleTimeout     time.Duration
 	StreamCreationTimeout time.Duration
 }
 
-func getAPIConfig(c Opts) (*apiServerConfig, error) {
-	config := apiServerConfig{
-		CertPath:   os.Getenv("APISERVER_CERT_LOCATION"),
-		KeyPath:    os.Getenv("APISERVER_KEY_LOCATION"),
-		CACertPath: os.Getenv("APISERVER_CA_CERT_LOCATION"),
+func getAPIConfig(c Opts) apiServerConfig {
+	return apiServerConfig{
+		Addr:                  fmt.Sprintf(":%d", c.ListenPort),
+		MetricsAddr:           c.MetricsAddr,
+		StreamIdleTimeout:     c.StreamIdleTimeout,
+		StreamCreationTimeout: c.StreamCreationTimeout,
 	}
-
-	config.Addr = fmt.Sprintf(":%d", c.ListenPort)
-	config.MetricsAddr = c.MetricsAddr
-	config.StreamIdleTimeout = c.StreamIdleTimeout
-	config.StreamCreationTimeout = c.StreamCreationTimeout
-
-	return &config, nil
 }

cmd/virtual-kubelet/internal/commands/root/root.go

@@ -16,8 +16,6 @@ package root
 
 import (
 	"context"
-	"crypto/tls"
-	"net/http"
 	"os"
 	"runtime"
 
@@ -28,10 +26,8 @@ import (
 	"github.com/virtual-kubelet/virtual-kubelet/internal/manager"
 	"github.com/virtual-kubelet/virtual-kubelet/log"
 	"github.com/virtual-kubelet/virtual-kubelet/node"
-	"github.com/virtual-kubelet/virtual-kubelet/node/api"
 	"github.com/virtual-kubelet/virtual-kubelet/node/nodeutil"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apiserver/pkg/server/dynamiccertificates"
 )
 
 // NewCommand creates a new top-level command.
@@ -73,14 +69,6 @@ func runRootCommand(ctx context.Context, s *provider.Store, c Opts) error {
 		}
 	}
 
-	// Ensure API client.
-	clientSet, err := nodeutil.ClientsetFromEnv(c.KubeConfigPath)
-	if err != nil {
-		return err
-	}
-
-	// Set-up the node provider.
-	mux := http.NewServeMux()
 	newProvider := func(cfg nodeutil.ProviderConfig) (nodeutil.Provider, node.NodeProvider, error) {
 		rm, err := manager.NewResourceManager(cfg.Pods, cfg.Secrets, cfg.ConfigMaps, cfg.Services)
 		if err != nil {
@@ -109,14 +97,9 @@ func runRootCommand(ctx context.Context, s *provider.Store, c Opts) error {
 		return p, nil, nil
 	}
 
-	apiConfig, err := getAPIConfig(c)
-	if err != nil {
-		return err
-	}
-
+	apiConfig := getAPIConfig(c)
 	cm, err := nodeutil.NewNode(c.NodeName, newProvider, func(cfg *nodeutil.NodeConfig) error {
 		cfg.KubeconfigPath = c.KubeConfigPath
-		cfg.Handler = mux
 		cfg.InformerResyncPeriod = c.InformerResyncPeriod
 
 		if taint != nil {
@@ -134,13 +117,7 @@ func runRootCommand(ctx context.Context, s *provider.Store, c Opts) error {
 
 		return nil
 	},
-		nodeutil.WithClient(clientSet),
-		setAuth(c.NodeName, apiConfig),
-		nodeutil.WithTLSConfig(
-			nodeutil.WithKeyPairFromPath(apiConfig.CertPath, apiConfig.KeyPath),
-			maybeCA(apiConfig.CACertPath),
-		),
-		nodeutil.AttachProviderRoutes(mux),
+		nodeutil.WithBootstrapFromRestConfig(),
 	)
 	if err != nil {
 		return err
@@ -179,32 +156,3 @@ func runRootCommand(ctx context.Context, s *provider.Store, c Opts) error {
 	}
 	return nil
 }
-
-func setAuth(node string, apiCfg *apiServerConfig) nodeutil.NodeOpt {
-	if apiCfg.CACertPath == "" {
-		return func(cfg *nodeutil.NodeConfig) error {
-			cfg.Handler = api.InstrumentHandler(nodeutil.WithAuth(nodeutil.NoAuth(), cfg.Handler))
-			return nil
-		}
-	}
-
-	return func(cfg *nodeutil.NodeConfig) error {
-		auth, err := nodeutil.WebhookAuth(cfg.Client, node, func(cfg *nodeutil.WebhookAuthConfig) error {
-			var err error
-			cfg.AuthnConfig.ClientCertificateCAContentProvider, err = dynamiccertificates.NewDynamicCAContentFromFile("ca-cert-bundle", apiCfg.CACertPath)
-			return err
-		})
-		if err != nil {
-			return err
-		}
-		cfg.Handler = api.InstrumentHandler(nodeutil.WithAuth(auth, cfg.Handler))
-		return nil
-	}
-}
-
-func maybeCA(p string) func(*tls.Config) error {
-	if p == "" {
-		return func(*tls.Config) error { return nil }
-	}
-	return nodeutil.WithCAFromPath(p)
-}

cmd/virtual-kubelet/internal/provider/mock/mock.go

@@ -102,7 +102,7 @@ func NewMockProvider(providerConfig, nodeName, operatingSystem string, internalI
 func loadConfig(providerConfig, nodeName string) (config MockConfig, err error) {
 	data, err := os.ReadFile(providerConfig)
 	if err != nil {
-		return config, err
+		return config, fmt.Errorf("error reaeding provider config: %w", err)
 	}
 	configMap := map[string]MockConfig{}
 	err = json.Unmarshal(data, &configMap)

errdefs/auth.go

@@ -0,0 +1,33 @@
+package errdefs
+
+import (
+	"errors"
+	"fmt"
+)
+
+var (
+	// ErrForbidden is returned when the user is not authorized to perform the operation.
+	ErrForbidden = errors.New("forbidden")
+	// ErrUnauthorized is returned when the user is not authenticated.
+	ErrUnauthorized = errors.New("unauthorized")
+)
+
+// Unauthorized wraps ErrUnauthorized with a message.
+func Unauthorized(msg string) error {
+	return fmt.Errorf("%w: %s", ErrUnauthorized, msg)
+}
+
+// Forbidden wraps ErrForbidden with a message.
+func Forbidden(msg string) error {
+	return fmt.Errorf("%w: %s", ErrForbidden, msg)
+}
+
+// IsForbidden returns true if the error has ErrForbidden in the error chain.
+func IsForbidden(err error) bool {
+	return errors.Is(err, ErrForbidden)
+}
+
+// IsUnauthorized returns true if the error has ErrUnauthorized in the error chain.
+func IsUnauthorized(err error) bool {
+	return errors.Is(err, ErrUnauthorized)
+}

node/api/stats.go

@@ -48,6 +48,7 @@ func HandlePodStatsSummary(h PodStatsSummaryHandlerFunc) http.HandlerFunc {
 		if _, err := w.Write(b); err != nil {
 			return errors.Wrap(err, "could not write to client")
 		}
+		_, _ = w.Write([]byte("\n"))
 		return nil
 	})
 }

node/nodeutil/auth.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/virtual-kubelet/virtual-kubelet/errdefs"
 	"github.com/virtual-kubelet/virtual-kubelet/log"
 	"github.com/virtual-kubelet/virtual-kubelet/trace"
 	"k8s.io/apiserver/pkg/authentication/authenticator"
@@ -69,14 +70,24 @@ func handleAuth(auth Auth, w http.ResponseWriter, r *http.Request, next http.Han
 	defer span.End()
 	r = r.WithContext(ctx)
 
+	logger := log.G(r.Context())
 	info, ok, err := auth.AuthenticateRequest(r)
-	if err != nil || !ok {
-		log.G(r.Context()).WithError(err).Error("Authorization error")
+	if err != nil {
+		logger.WithError(err).Error("Error authenticating request")
 		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		span.SetStatus(err)
 		return
 	}
 
-	logger := log.G(ctx).WithFields(log.Fields{
+	if !ok {
+		logger.Error("Request not authenticated")
+		log.G(r.Context()).Infof("Unauthorized: RequestURI: %s", r.RequestURI)
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		span.SetStatus(errdefs.ErrUnauthorized)
+		return
+	}
+
+	logger = logger.WithFields(log.Fields{
 		"user-name": info.User.GetName(),
 		"user-id":   info.User.GetUID(),
 	})
@@ -90,11 +101,13 @@ func handleAuth(auth Auth, w http.ResponseWriter, r *http.Request, next http.Han
 	if err != nil {
 		log.G(r.Context()).WithError(err).Error("Authorization error")
 		http.Error(w, err.Error(), http.StatusInternalServerError)
+		span.SetStatus(err)
 		return
 	}
 
 	if decision != authorizer.DecisionAllow {
 		http.Error(w, "Forbidden", http.StatusForbidden)
+		span.SetStatus(errdefs.ErrForbidden)
 		return
 	}
 

node/nodeutil/client.go

@@ -24,7 +24,7 @@ func ClientsetFromEnv(kubeConfigPath string) (*kubernetes.Clientset, error) {
 	)
 
 	if kubeConfigPath != "" {
-		config, err = clientsetFromEnvKubeConfigPath(kubeConfigPath)
+		config, err = RestConfigFromEnv(kubeConfigPath)
 	} else {
 		config, err = rest.InClusterConfig()
 	}
@@ -36,7 +36,8 @@ func ClientsetFromEnv(kubeConfigPath string) (*kubernetes.Clientset, error) {
 	return kubernetes.NewForConfig(config)
 }
 
-func clientsetFromEnvKubeConfigPath(kubeConfigPath string) (*rest.Config, error) {
+// RestConfigFromEnv is like ClientsetFromEnv except it returns a rest config instead of a full client.
+func RestConfigFromEnv(kubeConfigPath string) (*rest.Config, error) {
 	_, err := os.Stat(kubeConfigPath)
 	if os.IsNotExist(err) {
 		return rest.InClusterConfig()

node/nodeutil/controller.go

@@ -48,7 +48,8 @@ type Node struct {
 
 	workers int
 
-	eb record.EventBroadcaster
+	eb           record.EventBroadcaster
+	caController caController
 }
 
 // NodeController returns the configured node controller.
@@ -107,6 +108,10 @@ func (n *Node) Run(ctx context.Context) (retErr error) {
 		log.G(ctx).Debug("Started event broadcaster")
 	}
 
+	if n.caController != nil {
+		go n.caController.Run(ctx, 1)
+	}
+
 	cancelHTTP, err := n.runHTTP(ctx)
 	if err != nil {
 		return err
@@ -212,6 +217,8 @@ func (n *Node) Err() error {
 // NodeOpt is used as functional options when configuring a new node in NewNodeFromClient
 type NodeOpt func(c *NodeConfig) error
 
+type caController interface{ Run(context.Context, int) }
+
 // NodeConfig is used to hold configuration items for a Node.
 // It gets used in conjection with NodeOpt in NewNodeFromClient
 type NodeConfig struct {
@@ -260,22 +267,8 @@ type NodeConfig struct {
 	SkipDownwardAPIResolution bool
 
 	routeAttacher func(Provider, NodeConfig, corev1listers.PodLister)
-}
 
-// WithNodeConfig returns a NodeOpt which replaces the NodeConfig with the passed in value.
-func WithNodeConfig(c NodeConfig) NodeOpt {
-	return func(orig *NodeConfig) error {
-		*orig = c
-		return nil
-	}
-}
-
-// WithClient return a NodeOpt that sets the client that will be used to create/manage the node.
-func WithClient(c kubernetes.Interface) NodeOpt {
-	return func(cfg *NodeConfig) error {
-		cfg.Client = c
-		return nil
-	}
+	caController caController
 }
 
 // NewNode creates a new node using the provided client and name.
@@ -326,10 +319,6 @@ func NewNode(name string, newProvider NewProviderFunc, opts ...NodeOpt) (*Node,
 		return nil, errors.Wrap(err, "error parsing http listen address")
 	}
 
-	if cfg.Client == nil {
-		return nil, errors.New("no client provided")
-	}
-
 	podInformerFactory := informers.NewSharedInformerFactoryWithOptions(
 		cfg.Client,
 		cfg.InformerResyncPeriod,
@@ -425,6 +414,7 @@ func NewNode(name string, newProvider NewProviderFunc, opts ...NodeOpt) (*Node,
 		h:                  cfg.Handler,
 		listenAddr:         cfg.HTTPListenAddr,
 		workers:            cfg.NumWorkers,
+		caController:       cfg.caController,
 	}, nil
 }
 

node/nodeutil/opts.go

@@ -0,0 +1,150 @@
+package nodeutil
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/virtual-kubelet/virtual-kubelet/errdefs"
+	"github.com/virtual-kubelet/virtual-kubelet/node/api"
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+// WithNodeConfig returns a NodeOpt which replaces the NodeConfig with the passed in value.
+func WithNodeConfig(c NodeConfig) NodeOpt {
+	return func(orig *NodeConfig) error {
+		*orig = c
+		return nil
+	}
+}
+
+// WithClient return a NodeOpt that sets the client that will be used to create/manage the node.
+func WithClient(c kubernetes.Interface) NodeOpt {
+	return func(cfg *NodeConfig) error {
+		cfg.Client = c
+		return nil
+	}
+}
+
+// BootstrapConfig is the configuration for bootstrapping a node.
+type BootstrapConfig struct {
+	// ClientCAConfigMapSpec is the config map information for getting the CA for authenticating clients
+	// If not set, the CA in the rest config will be used.
+	ClientCAConfigMapSpec *ConfigMapCASpec
+	// A set of options to pass when creating the webhook auth.
+	WebhookAuthOpts []WebhookAuthOption
+	// RestConfig is the rest config to use for the client
+	// If it is not provided a default one from the environment will be created.
+	RestConfig *rest.Config
+}
+
+// ConfigMapCASpec is the spec for a config map that contains a CA.
+type ConfigMapCASpec struct {
+	Namespace string
+	Name      string
+	Key       string
+}
+
+// BootstrapOpt is a functional option used to configure node bootstrapping
+type BootstrapOpt func(*BootstrapConfig) error
+
+// WithBootstrapFromRestConfig takes a reset config (or a default one from the environment) and returns a NodeOpt that will:
+// 1. Create a client from the rest config (if no client is set)
+// 2. Create webhook authn/authz from the rest config
+// 3. Configure the TLS config for the HTTP server from the certs in the rest config.
+func WithBootstrapFromRestConfig(opts ...BootstrapOpt) NodeOpt {
+	return func(cfg *NodeConfig) error {
+		var bOpts BootstrapConfig
+		if rCfg, err := RestConfigFromEnv(cfg.KubeconfigPath); err == nil {
+			bOpts.RestConfig = rCfg
+		}
+
+		for _, o := range opts {
+			if err := o(&bOpts); err != nil {
+				return err
+			}
+		}
+
+		if bOpts.RestConfig == nil {
+			return fmt.Errorf("no rest config provided")
+		}
+
+		if cfg.Client == nil {
+			client, err := kubernetes.NewForConfig(bOpts.RestConfig)
+			if err != nil {
+				return err
+			}
+			cfg.Client = client
+		}
+
+		if err := WithTLSConfig(tlsFromRestConfig(bOpts.RestConfig))(cfg); err != nil {
+			return err
+		}
+
+		if err := configureWebhookCA(cfg, &bOpts); err != nil {
+			return fmt.Errorf("error configure webhook auth: %w", err)
+		}
+
+		var mux *http.ServeMux
+		if cfg.Handler == nil {
+			mux = http.NewServeMux()
+			cfg.Handler = mux
+		}
+		if cfg.routeAttacher == nil && mux != nil {
+			if err := AttachProviderRoutes(mux)(cfg); err != nil {
+				return err
+			}
+		}
+
+		auth, err := WebhookAuth(cfg.Client, cfg.NodeSpec.Name, bOpts.WebhookAuthOpts...)
+		if err != nil {
+			return err
+		}
+
+		cfg.Handler = api.InstrumentHandler(WithAuth(auth, cfg.Handler))
+		return nil
+	}
+}
+
+func configureWebhookCA(cfg *NodeConfig, bCfg *BootstrapConfig) error {
+	if bCfg.ClientCAConfigMapSpec != nil {
+		bCfg.WebhookAuthOpts = append(bCfg.WebhookAuthOpts, func(auth *WebhookAuthConfig) error {
+			cmCA, err := dynamiccertificates.NewDynamicCAFromConfigMapController("client-ca", bCfg.ClientCAConfigMapSpec.Namespace, bCfg.ClientCAConfigMapSpec.Name, bCfg.ClientCAConfigMapSpec.Key, cfg.Client)
+			if err != nil {
+				return fmt.Errorf("error loading dynamic CA from config map: %w", err)
+			}
+			auth.AuthnConfig.ClientCertificateCAContentProvider = cmCA
+			cfg.caController = cmCA
+			return nil
+		})
+		return nil
+	}
+
+	if bCfg.RestConfig.CAFile != "" {
+		bCfg.WebhookAuthOpts = append(bCfg.WebhookAuthOpts, func(auth *WebhookAuthConfig) error {
+			caFile, err := dynamiccertificates.NewDynamicCAContentFromFile("ca-file", bCfg.RestConfig.CAFile)
+			if err != nil {
+				return fmt.Errorf("error loading dynamic CA file from rest config: %w", err)
+			}
+			auth.AuthnConfig.ClientCertificateCAContentProvider = caFile
+			cfg.caController = caFile
+			return nil
+		})
+		return nil
+	}
+
+	if bCfg.RestConfig.CAData != nil {
+		bCfg.WebhookAuthOpts = append(bCfg.WebhookAuthOpts, func(auth *WebhookAuthConfig) error {
+			caData, err := dynamiccertificates.NewStaticCAContent("ca-data", bCfg.RestConfig.CAData)
+			if err != nil {
+				return fmt.Errorf("error loading static ca from rest config: %w", err)
+			}
+			auth.AuthnConfig.ClientCertificateCAContentProvider = caData
+			return nil
+		})
+		return nil
+	}
+
+	return errdefs.InvalidInput("no client CA found")
+}

node/nodeutil/tls.go

@@ -5,25 +5,28 @@ import (
 	"crypto/x509"
 	"fmt"
 	"os"
+
+	"k8s.io/client-go/rest"
 )
 
 // WithTLSConfig returns a NodeOpt which creates a base TLSConfig with the default cipher suites and tls min verions.
 // The tls config can be modified through functional options.
 func WithTLSConfig(opts ...func(*tls.Config) error) NodeOpt {
 	return func(cfg *NodeConfig) error {
-		tlsCfg := &tls.Config{
-			MinVersion:               tls.VersionTLS12,
-			PreferServerCipherSuites: true,
-			CipherSuites:             DefaultServerCiphers(),
-			ClientAuth:               tls.RequestClientCert,
+		if cfg.TLSConfig == nil {
+			cfg.TLSConfig = &tls.Config{
+				MinVersion:               tls.VersionTLS12,
+				PreferServerCipherSuites: true,
+				CipherSuites:             DefaultServerCiphers(),
+				ClientAuth:               tls.RequestClientCert,
+			}
 		}
 		for _, o := range opts {
-			if err := o(tlsCfg); err != nil {
+			if err := o(cfg.TLSConfig); err != nil {
 				return err
 			}
 		}
 
-		cfg.TLSConfig = tlsCfg
 		return nil
 	}
 }
@@ -45,7 +48,7 @@ func WithKeyPairFromPath(cert, key string) func(*tls.Config) error {
 	return func(cfg *tls.Config) error {
 		cert, err := tls.LoadX509KeyPair(cert, key)
 		if err != nil {
-			return err
+			return fmt.Errorf("error loading x509 key pair: %w", err)
 		}
 		cfg.Certificates = append(cfg.Certificates, cert)
 		return nil
@@ -56,7 +59,7 @@ func WithKeyPairFromPath(cert, key string) func(*tls.Config) error {
 // If a cert pool is not defined on the tls config an empty one will be created.
 func WithCACert(pem []byte) func(*tls.Config) error {
 	return func(cfg *tls.Config) error {
-		if cfg.ClientCAs == nil {
+		if cfg.RootCAs == nil {
 			cfg.ClientCAs = x509.NewCertPool()
 		}
 		if !cfg.ClientCAs.AppendCertsFromPEM(pem) {
@@ -81,3 +84,48 @@ func DefaultServerCiphers() []uint16 {
 		tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
 	}
 }
+
+func tlsFromRestConfig(r *rest.Config) func(*tls.Config) error {
+	return func(cfg *tls.Config) error {
+		var err error
+
+		cfg.ClientAuth = tls.RequestClientCert
+
+		certData := r.CertData
+		if certData == nil && r.CertFile != "" {
+			certData, err = os.ReadFile(r.CertFile)
+			if err != nil {
+				return fmt.Errorf("error reading cert file from clientset: %w", err)
+			}
+		}
+
+		keyData := r.KeyData
+		if keyData == nil && r.KeyFile != "" {
+			keyData, err = os.ReadFile(r.KeyFile)
+			if err != nil {
+				return fmt.Errorf("error reading key file from clientset: %w", err)
+			}
+		}
+		if keyData != nil && certData != nil {
+			pem, err := tls.X509KeyPair(certData, keyData)
+			if err != nil {
+				return fmt.Errorf("error creating key pair from clientset: %w", err)
+			}
+			cfg.Certificates = append(cfg.Certificates, pem)
+			cfg.ClientAuth = tls.RequestClientCert
+		}
+
+		caData := r.CAData
+		if certData == nil && r.CAFile != "" {
+			caData, err = os.ReadFile(r.CAFile)
+			if err != nil {
+				return fmt.Errorf("error reading ca file from clientset: %w", err)
+			}
+		}
+		if caData != nil {
+			return WithCACert(caData)(cfg)
+		}
+
+		return nil
+	}
+}

trace/opencensus/opencensus.go

@@ -79,8 +79,12 @@ func (s *span) SetStatus(err error) {
 		status.Code = octrace.StatusCodeNotFound
 	case errdefs.IsInvalidInput(err):
 		status.Code = octrace.StatusCodeInvalidArgument
-		// TODO: other error types
+	case errdefs.IsForbidden(err):
+		status.Code = octrace.StatusCodePermissionDenied
+	case errdefs.IsUnauthorized(err):
+		status.Code = octrace.StatusCodeUnauthenticated
 	default:
+		// TODO: other error types
 		status.Code = octrace.StatusCodeUnknown
 	}