Enable structured logging for core Kubernetes components

cmd/containerd/main.go

@@ -7,5 +7,6 @@ import (
 
 func main() {
 	klog.InitFlags(nil)
+	klog.EnableContextualLogging(true)
 	containerd.Main()
 }

pkg/certmonitor/certmonitor.go

@@ -50,7 +50,7 @@ func Setup(ctx context.Context, nodeConfig *daemonconfig.Node, dataDir string) e
 		return err
 	}
 
-	recorder := util.BuildControllerEventRecorder(client, controllerName, metav1.NamespaceDefault)
+	recorder := util.BuildControllerEventRecorder(ctx, client, controllerName, metav1.NamespaceDefault)
 
 	// This is consistent with events attached to the node generated by the kubelet
 	// https://github.com/kubernetes/kubernetes/blob/612130dd2f4188db839ea5c2dea07a96b0ad8d1c/pkg/kubelet/kubelet.go#L479-L485

pkg/cli/agent/agent.go

@@ -20,12 +20,14 @@ import (
 	"github.com/k3s-io/k3s/pkg/signals"
 	"github.com/k3s-io/k3s/pkg/spegel"
 	"github.com/k3s-io/k3s/pkg/util"
+	"github.com/k3s-io/k3s/pkg/util/logger"
 	"github.com/k3s-io/k3s/pkg/util/permissions"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/k3s-io/k3s/pkg/vpn"
 	pkgerrors "github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
+	"k8s.io/klog/v2"
 )
 
 func Run(clx *cli.Context) (rerr error) {
@@ -47,7 +49,8 @@ func Run(clx *cli.Context) (rerr error) {
 		return err
 	}
 
-	ctx := signals.SetupSignalContext()
+	klog.EnableContextualLogging(true)
+	ctx := klog.NewContext(signals.SetupSignalContext(), logger.NewLogrusSink(nil).AsLogr())
 	wg := &sync.WaitGroup{}
 
 	// If exiting due to an error, ensure that contexts are cancelled so that the

pkg/cli/server/server.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	systemd "github.com/coreos/go-systemd/v22/daemon"
+	"github.com/go-logr/logr"
 	"github.com/gorilla/mux"
 	"github.com/k3s-io/k3s/pkg/agent"
 	"github.com/k3s-io/k3s/pkg/agent/https"
@@ -30,6 +31,7 @@ import (
 	"github.com/k3s-io/k3s/pkg/signals"
 	"github.com/k3s-io/k3s/pkg/spegel"
 	"github.com/k3s-io/k3s/pkg/util"
+	"github.com/k3s-io/k3s/pkg/util/logger"
 	"github.com/k3s-io/k3s/pkg/util/permissions"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/k3s-io/k3s/pkg/vpn"
@@ -40,6 +42,7 @@ import (
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/wait"
 	kubeapiserverflag "k8s.io/component-base/cli/flag"
+	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/controlplane/apiserver/options"
 	utilsnet "k8s.io/utils/net"
 )
@@ -76,7 +79,8 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
 		return err
 	}
 
-	ctx := signals.SetupSignalContext()
+	klog.EnableContextualLogging(true)
+	ctx := logr.NewContext(signals.SetupSignalContext(), logger.NewLogrusSink(nil).AsLogr())
 	wg := &sync.WaitGroup{}
 
 	// If exiting due to an error, ensure that contexts are cancelled so that the

pkg/cloudprovider/cloudprovider.go

@@ -88,7 +88,7 @@ func (k *k3s) Initialize(clientBuilder cloudprovider.ControllerClientBuilder, st
 
 	if k.LBEnabled {
 		// Wrangler controller and caches are only needed if the load balancer controller is enabled.
-		k.recorder = util.BuildControllerEventRecorder(k.client, controllerName, meta.NamespaceAll)
+		k.recorder = util.BuildControllerEventRecorder(ctx, k.client, controllerName, meta.NamespaceAll)
 		coreFactory := core.NewFactoryFromConfigOrDie(config)
 		k.nodeCache = coreFactory.Core().V1().Node().Cache()
 

pkg/deploy/controller.go

@@ -84,7 +84,7 @@ type watcher struct {
 
 // start calls listFiles at regular intervals to trigger application of manifests that have changed on disk.
 func (w *watcher) start(ctx context.Context, client kubernetes.Interface) {
-	w.recorder = pkgutil.BuildControllerEventRecorder(client, ControllerName, metav1.NamespaceSystem)
+	w.recorder = pkgutil.BuildControllerEventRecorder(ctx, client, ControllerName, metav1.NamespaceSystem)
 	force := true
 	for {
 		if err := w.listFiles(force); err == nil {

pkg/executor/embed/embed.go

@@ -27,6 +27,7 @@ import (
 	"github.com/k3s-io/k3s/pkg/executor/embed/etcd"
 	"github.com/k3s-io/k3s/pkg/signals"
 	"github.com/k3s-io/k3s/pkg/util"
+	"github.com/k3s-io/k3s/pkg/util/logger"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/k3s-io/k3s/pkg/vpn"
 	pkgerrors "github.com/pkg/errors"
@@ -80,6 +81,7 @@ func (e *Embedded) Bootstrap(ctx context.Context, nodeConfig *daemonconfig.Node,
 		defer cancel()
 
 		klog.InitFlags(nil)
+		klog.EnableContextualLogging(true)
 		for {
 			flag.Set("v", strconv.Itoa(cmds.LogConfig.VLevel))
 
@@ -170,7 +172,8 @@ func (e *Embedded) Bootstrap(ctx context.Context, nodeConfig *daemonconfig.Node,
 }
 
 func (e *Embedded) Kubelet(ctx context.Context, args []string) error {
-	command := kubelet.NewKubeletCommand(context.Background())
+	ctx = logger.NewContextWithName(ctx, "kubelet")
+	command := kubelet.NewKubeletCommand(logger.NewContextWithName(context.Background(), "kubelet"))
 	command.SetArgs(args)
 
 	go func() {
@@ -191,6 +194,7 @@ func (e *Embedded) Kubelet(ctx context.Context, args []string) error {
 }
 
 func (e *Embedded) KubeProxy(ctx context.Context, args []string) error {
+	ctx = logger.NewContextWithName(ctx, "kube-proxy")
 	command := proxy.NewProxyCommand()
 	command.SetArgs(util.GetArgs(platformKubeProxyArgs(e.nodeConfig), args))
 
@@ -217,6 +221,7 @@ func (*Embedded) APIServerHandlers(ctx context.Context) (authenticator.Request,
 }
 
 func (e *Embedded) APIServer(ctx context.Context, args []string) error {
+	ctx = logger.NewContextWithName(ctx, "kube-apiserver")
 	command := apiapp.NewAPIServerCommand(ctx.Done())
 	command.SetArgs(args)
 
@@ -238,6 +243,7 @@ func (e *Embedded) APIServer(ctx context.Context, args []string) error {
 }
 
 func (e *Embedded) Scheduler(ctx context.Context, nodeReady <-chan struct{}, args []string) error {
+	ctx = logger.NewContextWithName(ctx, "kube-scheduler")
 	command := sapp.NewSchedulerCommand(ctx.Done())
 	command.SetArgs(args)
 
@@ -260,6 +266,7 @@ func (e *Embedded) Scheduler(ctx context.Context, nodeReady <-chan struct{}, arg
 }
 
 func (e *Embedded) ControllerManager(ctx context.Context, args []string) error {
+	ctx = logger.NewContextWithName(ctx, "kube-controller-manager")
 	command := cmapp.NewControllerManagerCommand()
 	command.SetArgs(args)
 
@@ -281,6 +288,7 @@ func (e *Embedded) ControllerManager(ctx context.Context, args []string) error {
 }
 
 func (*Embedded) CloudControllerManager(ctx context.Context, ccmRBACReady <-chan struct{}, args []string) error {
+	ctx = logger.NewContextWithName(ctx, "cloud-controller-manager")
 	ccmOptions, err := ccmopt.NewCloudControllerManagerOptions()
 	if err != nil {
 		logrus.Fatalf("unable to initialize command options: %v", err)
@@ -330,6 +338,7 @@ func (e *Embedded) CurrentETCDOptions() (executor.InitialOptions, error) {
 }
 
 func (e *Embedded) ETCD(ctx context.Context, wg *sync.WaitGroup, args *executor.ETCDConfig, extraArgs []string, test executor.TestFunc) error {
+	ctx = logger.NewContextWithName(ctx, "etcd")
 	// Start a goroutine to call the provided test function until it returns true.
 	// The test function is reponsible for ensuring that the etcd server is up
 	// and ready to accept client requests.
@@ -356,10 +365,12 @@ func (e *Embedded) ETCD(ctx context.Context, wg *sync.WaitGroup, args *executor.
 }
 
 func (e *Embedded) Containerd(ctx context.Context, cfg *daemonconfig.Node) error {
+	ctx = logger.NewContextWithName(ctx, "containerd")
 	return executor.CloseIfNilErr(containerd.Run(ctx, cfg), e.criReady)
 }
 
 func (e *Embedded) Docker(ctx context.Context, cfg *daemonconfig.Node) error {
+	ctx = logger.NewContextWithName(ctx, "cri-dockerd")
 	return executor.CloseIfNilErr(cridockerd.Run(ctx, cfg), e.criReady)
 }
 
@@ -373,6 +384,7 @@ func (e *Embedded) CRI(ctx context.Context, cfg *daemonconfig.Node) error {
 
 func (e *Embedded) CNI(ctx context.Context, wg *sync.WaitGroup, cfg *daemonconfig.Node) error {
 	if cfg.Flannel.Backend != flannel.BackendNone {
+		ctx := logger.NewContextWithName(ctx, "flannel")
 		if (cfg.Flannel.ExternalIP) && (len(cfg.AgentConfig.NodeExternalIPs) == 0) {
 			logrus.Warnf("Server has flannel-external-ip flag set but this node does not set node-external-ip. Flannel will use internal address when connecting to this node.")
 		} else if (cfg.Flannel.ExternalIP) && (cfg.Flannel.Backend != flannel.BackendWireguardNative) {
@@ -388,6 +400,7 @@ func (e *Embedded) CNI(ctx context.Context, wg *sync.WaitGroup, cfg *daemonconfi
 	}
 
 	if !cfg.AgentConfig.DisableNPC {
+		ctx := logger.NewContextWithName(ctx, "kube-proxy")
 		if err := netpol.Run(ctx, wg, cfg); err != nil {
 			return err
 		}

pkg/server/context.go

@@ -10,6 +10,7 @@ import (
 	helmcrds "github.com/k3s-io/helm-controller/pkg/crds"
 	"github.com/k3s-io/helm-controller/pkg/generated/controllers/helm.cattle.io"
 	"github.com/k3s-io/k3s/pkg/util"
+	"github.com/k3s-io/k3s/pkg/util/logger"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/rancher/wrangler/v3/pkg/crd"
 	"github.com/rancher/wrangler/v3/pkg/generated/controllers/apps"
@@ -73,6 +74,7 @@ func NewContext(ctx context.Context, config *Config) (*Context, error) {
 		hf = helm.NewFactoryFromConfigOrDie(restConfig)
 	}
 
+	ctx = logger.NewContextWithName(ctx, version.Program+"-supervisor")
 	c := &Context{
 		K3s:       k3s.NewFactoryFromConfigOrDie(restConfig),
 		Auth:      rbac.NewFactoryFromConfigOrDie(restConfig),
@@ -82,7 +84,7 @@ func NewContext(ctx context.Context, config *Config) (*Context, error) {
 		Discovery: discovery.NewFactoryFromConfigOrDie(restConfig),
 		Helm:      hf,
 
-		Event: util.BuildControllerEventRecorder(k8s, version.Program+"-supervisor", metav1.NamespaceAll),
+		Event: util.BuildControllerEventRecorder(ctx, k8s, version.Program+"-supervisor", metav1.NamespaceAll),
 		K8s:   k8s,
 		Ext:   ext,
 	}

pkg/server/handlers/handlers_test.go

@@ -1747,7 +1747,7 @@ func getMockedControl(t *testing.T) (*config.Control, context.CancelFunc) {
 	control.Runtime.K8s = k8s
 
 	// create event recorder
-	control.Runtime.Event = util.BuildControllerEventRecorder(control.Runtime.K8s, version.Program+"-supervisor", metav1.NamespaceAll)
+	control.Runtime.Event = util.BuildControllerEventRecorder(ctx, control.Runtime.K8s, version.Program+"-supervisor", metav1.NamespaceAll)
 
 	// start the node password controller
 	err = nodepassword.Register(ctx, control.Runtime.K8s, coreFactory.Core().V1().Secret(), coreFactory.Core().V1().Node())

pkg/server/handlers/secrets-encrypt.go

@@ -447,7 +447,7 @@ func updateSecrets(ctx context.Context, control *config.Control, nodeName string
 	}
 
 	// For backwards compatibility with the old controller, we use an event recorder instead of logrus
-	recorder := util.BuildControllerEventRecorder(k8s, "secrets-reencrypt", metav1.NamespaceDefault)
+	recorder := util.BuildControllerEventRecorder(ctx, k8s, "secrets-reencrypt", metav1.NamespaceDefault)
 
 	secretPager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
 		return k8s.CoreV1().Secrets(metav1.NamespaceAll).List(ctx, opts)

pkg/server/server.go

@@ -27,6 +27,7 @@ import (
 	"github.com/k3s-io/k3s/pkg/server/handlers"
 	"github.com/k3s-io/k3s/pkg/static"
 	"github.com/k3s-io/k3s/pkg/util"
+	"github.com/k3s-io/k3s/pkg/util/logger"
 	"github.com/k3s-io/k3s/pkg/util/permissions"
 	"github.com/k3s-io/k3s/pkg/version"
 	pkgerrors "github.com/pkg/errors"
@@ -250,7 +251,7 @@ func coreControllers(ctx context.Context, sc *Context, config *Config) error {
 			strconv.Itoa(config.ControlConfig.HTTPSPort),
 			k8s,
 			apply,
-			util.BuildControllerEventRecorder(k8s, helmcommon.Name, metav1.NamespaceAll),
+			util.BuildControllerEventRecorder(logger.NewContextWithName(ctx, "helm-controller"), k8s, helmcommon.Name, metav1.NamespaceAll),
 			helm.V1().HelmChart(),
 			helm.V1().HelmChart().Cache(),
 			helm.V1().HelmChartConfig(),

pkg/util/api.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/k3s-io/k3s/pkg/signals"
+	"github.com/k3s-io/k3s/pkg/util/logger"
 	pkgerrors "github.com/pkg/errors"
 	"github.com/rancher/wrangler/v3/pkg/merr"
 	"github.com/rancher/wrangler/v3/pkg/schemes"
@@ -251,9 +252,10 @@ func subjectAccessReview(authClient *authorizationv1client.AuthorizationV1Client
 	}
 }
 
-func BuildControllerEventRecorder(k8s clientset.Interface, controllerName, namespace string) record.EventRecorder {
+func BuildControllerEventRecorder(ctx context.Context, k8s clientset.Interface, controllerName, namespace string) record.EventRecorder {
+	ctx = logger.NewContextWithName(ctx, controllerName)
 	logrus.Infof("Creating %s event broadcaster", controllerName)
-	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx))
 	eventBroadcaster.StartStructuredLogging(0)
 	eventBroadcaster.StartRecordingToSink(&coregetter.EventSinkImpl{Interface: k8s.CoreV1().Events(namespace)})
 	nodeName := os.Getenv("NODE_NAME")

pkg/util/context.go

@@ -3,6 +3,8 @@ package util
 import (
 	"context"
 	"time"
+
+	"k8s.io/klog/v2"
 )
 
 const DefaultContextDelay = 5 * time.Second
@@ -16,5 +18,5 @@ func DelayCancel(ctx context.Context, delay time.Duration) context.Context {
 		time.Sleep(delay)
 		dcancel()
 	}()
-	return dctx
+	return klog.NewContext(dctx, klog.FromContext(ctx))
 }

pkg/util/logger/logger.go

@@ -1,10 +1,12 @@
 package logger
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/go-logr/logr"
 	"github.com/sirupsen/logrus"
+	"k8s.io/klog/v2"
 )
 
 // implicit interface check
@@ -87,3 +89,7 @@ func (ls *LogrusSink) WithName(name string) logr.LogSink {
 	}
 	return ls.WithValues("logger", name)
 }
+
+func NewContextWithName(ctx context.Context, name string) context.Context {
+	return klog.NewContext(ctx, klog.FromContext(ctx).WithName(name))
+}