Support structured logging

connection/connection.go

@@ -108,9 +108,10 @@ func ExitOnConnectionLoss() func() bool {
 	return func() bool {
 		terminationMsg := "Lost connection to CSI driver, exiting"
 		if err := ioutil.WriteFile(terminationLogPath, []byte(terminationMsg), 0644); err != nil {
-			klog.Errorf("%s: %s", terminationLogPath, err)
+			klog.ErrorS(err, "Failed to write a message to the termination logfile", "terminationLogPath", terminationLogPath)
 		}
-		klog.Exit(terminationMsg)
+		klog.ErrorS(nil, terminationMsg)
+		klog.FlushAndExit(klog.ExitFlushTimeout, 1)
 		// Not reached.
 		return false
 	}
@@ -188,7 +189,7 @@ func connect(
 			if haveConnected && !lostConnection {
 				// We have detected a loss of connection for the first time. Decide what to do...
 				// Record this once. TODO (?): log at regular time intervals.
-				klog.Errorf("Lost connection to %s.", address)
+				klog.ErrorS(nil, "Lost connection", "address", address)
 				// Inform caller and let it decide? Default is to reconnect.
 				if o.reconnect != nil {
 					reconnect = o.reconnect()
@@ -210,7 +211,7 @@ func connect(
 		return nil, errors.New("OnConnectionLoss callback only supported for unix:// addresses")
 	}
 
-	klog.V(5).Infof("Connecting to %s", address)
+	klog.V(5).InfoS("Connecting", "address", address)
 
 	// Connect in background.
 	var conn *grpc.ClientConn
@@ -229,7 +230,7 @@ func connect(
 	for {
 		select {
 		case <-ticker.C:
-			klog.Warningf("Still connecting to %s", address)
+			klog.InfoS("Still connecting", "address", address)
 
 		case <-ready:
 			return conn, err
@@ -239,15 +240,13 @@ func connect(
 
 // LogGRPC is gPRC unary interceptor for logging of CSI messages at level 5. It removes any secrets from the message.
 func LogGRPC(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
-	klog.V(5).Infof("GRPC call: %s", method)
-	klog.V(5).Infof("GRPC request: %s", protosanitizer.StripSecrets(req))
+	klog.V(5).InfoS("GRPC call", "method", method, "request", protosanitizer.StripSecrets(req))
 	err := invoker(ctx, method, req, reply, cc, opts...)
 	cappedStr := protosanitizer.StripSecrets(reply).String()
 	if maxLogChar > 0 && len(cappedStr) > maxLogChar {
 		cappedStr = cappedStr[:maxLogChar] + fmt.Sprintf(" [response body too large, log capped to %d chars]", maxLogChar)
 	}
-	klog.V(5).Infof("GRPC response: %s", cappedStr)
-	klog.V(5).Infof("GRPC error: %v", err)
+	klog.V(5).Infof("GRPC response", "response", cappedStr, "err", err)
 	return err
 }
 
@@ -284,14 +283,14 @@ func (cmm ExtendedCSIMetricsManager) RecordMetricsClientInterceptor(
 		if additionalInfo != nil {
 			additionalInfoVal, ok := additionalInfo.(AdditionalInfo)
 			if !ok {
-				klog.Errorf("Failed to record migrated status, cannot convert additional info %v", additionalInfo)
+				klog.ErrorS(nil, "Failed to record migrated status, cannot convert additional info", "additionalInfo", additionalInfo)
 				return err
 			}
 			migrated = additionalInfoVal.Migrated
 		}
 		cmmv, metricsErr := cmm.WithLabelValues(map[string]string{metrics.LabelMigrated: migrated})
 		if metricsErr != nil {
-			klog.Errorf("Failed to record migrated status, error: %v", metricsErr)
+			klog.ErrorS(metricsErr, "Failed to record migrated status")
 		} else {
 			cmmBase = cmmv
 		}

deprecatedflags/deprecatedflags.go

@@ -20,6 +20,7 @@ package deprecatedflags
 
 import (
 	"flag"
+	"fmt"
 
 	"k8s.io/klog/v2"
 )
@@ -50,7 +51,7 @@ var _ flag.Value = deprecated{}
 
 func (d deprecated) String() string { return "" }
 func (d deprecated) Set(value string) error {
-	klog.Warningf("Warning: option %s=%q is deprecated and has no effect", d.name, value)
+	klog.InfoS("Warning: this option is deprecated and has no effect", "option", fmt.Sprintf("%s=%q", d.name, value))
 	return nil
 }
 func (d deprecated) Type() string     { return "" }

leaderelection/leader_election.go

@@ -26,7 +26,7 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -175,14 +175,15 @@ func (l *leaderElection) Run() error {
 		RetryPeriod:   l.retryPeriod,
 		Callbacks: leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(ctx context.Context) {
-				klog.V(2).Info("became leader, starting")
+				klog.V(2).InfoS("became leader, starting")
 				l.runFunc(ctx)
 			},
 			OnStoppedLeading: func() {
-				klog.Fatal("stopped leading")
+				klog.ErrorS(nil, "Stopped leading")
+				klog.FlushAndExit(klog.ExitFlushTimeout, 1)
 			},
 			OnNewLeader: func(identity string) {
-				klog.V(3).Infof("new leader detected, current leader: %s", identity)
+				klog.V(3).InfoS("New leader detected", "leader", identity)
 			},
 		},
 		WatchDog: l.healthCheck,

rpc/common.go

@@ -135,7 +135,7 @@ func GetGroupControllerCapabilities(ctx context.Context, conn *grpc.ClientConn)
 // Any error other than timeout is returned.
 func ProbeForever(conn *grpc.ClientConn, singleProbeTimeout time.Duration) error {
 	for {
-		klog.Info("Probing CSI driver for readiness")
+		klog.InfoS("Probing CSI driver for readiness")
 		ready, err := probeOnce(conn, singleProbeTimeout)
 		if err != nil {
 			st, ok := status.FromError(err)
@@ -148,12 +148,12 @@ func ProbeForever(conn *grpc.ClientConn, singleProbeTimeout time.Duration) error
 				return fmt.Errorf("CSI driver probe failed: %s", err)
 			}
 			// Timeout -> driver is not ready. Fall through to sleep() below.
-			klog.Warning("CSI driver probe timed out")
+			klog.InfoS("CSI driver probe timed out")
 		} else {
 			if ready {
 				return nil
 			}
-			klog.Warning("CSI driver is not ready")
+			klog.InfoS("CSI driver is not ready")
 		}
 		// Timeout was returned or driver is not ready.
 		time.Sleep(probeInterval)