Merge pull request #114609 from pohly/log-runtime-verbosity-level

runtime log verbosity level changes

staging/src/k8s.io/component-base/logs/api/v1/options.go

@@ -19,7 +19,9 @@ package v1
 import (
 	"flag"
 	"fmt"
+	"io"
 	"math"
+	"os"
 	"strings"
 	"time"
 
@@ -31,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	cliflag "k8s.io/component-base/cli/flag"
 	"k8s.io/component-base/featuregate"
+	"k8s.io/component-base/logs/internal/setverbositylevel"
 	"k8s.io/component-base/logs/klogflags"
 )
 
@@ -62,18 +65,41 @@ func NewLoggingConfiguration() *LoggingConfiguration {
 // The optional FeatureGate controls logging features. If nil, the default for
 // these features is used.
 func ValidateAndApply(c *LoggingConfiguration, featureGate featuregate.FeatureGate) error {
-	return ValidateAndApplyAsField(c, featureGate, nil)
+	return validateAndApply(c, nil, featureGate, nil)
+}
+
+// ValidateAndApplyWithOptions is a variant of ValidateAndApply which accepts
+// additional options beyond those that can be configured through the API. This
+// is meant for testing.
+func ValidateAndApplyWithOptions(c *LoggingConfiguration, options *LoggingOptions, featureGate featuregate.FeatureGate) error {
+	return validateAndApply(c, options, featureGate, nil)
+}
+
+// +k8s:deepcopy-gen=false
+
+// LoggingOptions can be used with ValidateAndApplyWithOptions to override
+// certain global defaults.
+type LoggingOptions struct {
+	// ErrorStream can be used to override the os.Stderr default.
+	ErrorStream io.Writer
+
+	// InfoStream can be used to override the os.Stdout default.
+	InfoStream io.Writer
 }
 
 // ValidateAndApplyAsField is a variant of ValidateAndApply that should be used
 // when the LoggingConfiguration is embedded in some larger configuration
 // structure.
 func ValidateAndApplyAsField(c *LoggingConfiguration, featureGate featuregate.FeatureGate, fldPath *field.Path) error {
+	return validateAndApply(c, nil, featureGate, fldPath)
+}
+
+func validateAndApply(c *LoggingConfiguration, options *LoggingOptions, featureGate featuregate.FeatureGate, fldPath *field.Path) error {
 	errs := Validate(c, featureGate, fldPath)
 	if len(errs) > 0 {
 		return errs.ToAggregate()
 	}
-	return apply(c, featureGate)
+	return apply(c, options, featureGate)
 }
 
 // Validate can be used to check for invalid settings without applying them.
@@ -156,7 +182,7 @@ func featureEnabled(featureGate featuregate.FeatureGate, feature featuregate.Fea
 	return enabled
 }
 
-func apply(c *LoggingConfiguration, featureGate featuregate.FeatureGate) error {
+func apply(c *LoggingConfiguration, options *LoggingOptions, featureGate featuregate.FeatureGate) error {
 	contextualLoggingEnabled := contextualLoggingDefault
 	if featureGate != nil {
 		contextualLoggingEnabled = featureGate.Enabled(ContextualLogging)
@@ -167,8 +193,19 @@ func apply(c *LoggingConfiguration, featureGate featuregate.FeatureGate) error {
 	if format.factory == nil {
 		klog.ClearLogger()
 	} else {
-		log, flush := format.factory.Create(*c)
-		klog.SetLoggerWithOptions(log, klog.ContextualLogger(contextualLoggingEnabled), klog.FlushLogger(flush))
+		if options == nil {
+			options = &LoggingOptions{
+				ErrorStream: os.Stderr,
+				InfoStream:  os.Stdout,
+			}
+		}
+		log, control := format.factory.Create(*c, *options)
+		if control.SetVerbosityLevel != nil {
+			setverbositylevel.Mutex.Lock()
+			defer setverbositylevel.Mutex.Unlock()
+			setverbositylevel.Callbacks = append(setverbositylevel.Callbacks, control.SetVerbosityLevel)
+		}
+		klog.SetLoggerWithOptions(log, klog.ContextualLogger(contextualLoggingEnabled), klog.FlushLogger(control.Flush))
 	}
 	if err := loggingFlags.Lookup("v").Value.Set(VerbosityLevelPflag(&c.Verbosity).String()); err != nil {
 		return fmt.Errorf("internal error while setting klog verbosity: %v", err)

staging/src/k8s.io/component-base/logs/api/v1/registry.go

@@ -39,14 +39,29 @@ type logFormat struct {
 	feature featuregate.Feature
 }
 
+// +k8s:deepcopy-gen=false
+
+// RuntimeControl provides operations that aren't available through the normal
+// Logger or LogSink API.
+type RuntimeControl struct {
+	// Flush ensures that all in-memory data is written.
+	// May be nil.
+	Flush func()
+
+	// SetVerbosityLevel changes the level for all Logger instances
+	// derived from the initial one. May be nil.
+	//
+	// The parameter is intentionally a plain uint32 instead of
+	// VerbosityLevel to enable implementations that don't need to import
+	// the API (helps avoid circular dependencies).
+	SetVerbosityLevel func(v uint32) error
+}
+
 // LogFormatFactory provides support for a certain additional,
 // non-default log format.
 type LogFormatFactory interface {
 	// Create returns a logger with the requested configuration.
-	// Returning a flush function for the logger is optional.
-	// If provided, the caller must ensure that it is called
-	// periodically (if desired) and at program exit.
-	Create(c LoggingConfiguration) (log logr.Logger, flush func())
+	Create(c LoggingConfiguration, o LoggingOptions) (logr.Logger, RuntimeControl)
 }
 
 // RegisterLogFormat registers support for a new logging format. This must be called

staging/src/k8s.io/component-base/logs/internal/setverbositylevel/setverbositylevel.go

@@ -0,0 +1,34 @@
+/*
+Copyright 2022 The Kubernetes 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 setverbositylevel stores callbacks that will be invoked by logs.GlogLevel.
+//
+// This is a separate package to avoid a dependency from
+// k8s.io/component-base/logs (uses the callbacks) to
+// k8s.io/component-base/logs/api/v1 (adds them). Not all users of the logs
+// package also use the API.
+package setverbositylevel
+
+import (
+	"sync"
+)
+
+var (
+	// Mutex controls access to the callbacks.
+	Mutex sync.Mutex
+
+	Callbacks []func(v uint32) error
+)

staging/src/k8s.io/component-base/logs/json/json.go

@@ -14,11 +14,11 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package logs
+package json
 
 import (
 	"io"
-	"os"
+	"sync/atomic"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -35,12 +35,32 @@ var (
 	timeNow = time.Now
 )
 
+type runtime struct {
+	v uint32
+}
+
+func (r *runtime) ZapV() zapcore.Level {
+	// zap levels are inverted: everything with a verbosity >= threshold gets logged.
+	return -zapcore.Level(atomic.LoadUint32(&r.v))
+}
+
+// Enabled implements the zapcore.LevelEnabler interface.
+func (r *runtime) Enabled(level zapcore.Level) bool {
+	return level >= r.ZapV()
+}
+
+func (r *runtime) SetVerbosityLevel(v uint32) error {
+	atomic.StoreUint32(&r.v, v)
+	return nil
+}
+
+var _ zapcore.LevelEnabler = &runtime{}
+
 // NewJSONLogger creates a new json logr.Logger and its associated
-// flush function. The separate error stream is optional and may be nil.
+// control interface. The separate error stream is optional and may be nil.
 // The encoder config is also optional.
-func NewJSONLogger(v logsapi.VerbosityLevel, infoStream, errorStream zapcore.WriteSyncer, encoderConfig *zapcore.EncoderConfig) (logr.Logger, func()) {
-	// zap levels are inverted: everything with a verbosity >= threshold gets logged.
-	zapV := -zapcore.Level(v)
+func NewJSONLogger(v logsapi.VerbosityLevel, infoStream, errorStream zapcore.WriteSyncer, encoderConfig *zapcore.EncoderConfig) (logr.Logger, logsapi.RuntimeControl) {
+	r := &runtime{v: uint32(v)}
 
 	if encoderConfig == nil {
 		encoderConfig = &zapcore.EncoderConfig{
@@ -57,23 +77,27 @@ func NewJSONLogger(v logsapi.VerbosityLevel, infoStream, errorStream zapcore.Wri
 	encoder := zapcore.NewJSONEncoder(*encoderConfig)
 	var core zapcore.Core
 	if errorStream == nil {
-		core = zapcore.NewCore(encoder, infoStream, zapV)
+		core = zapcore.NewCore(encoder, infoStream, r)
 	} else {
 		highPriority := zap.LevelEnablerFunc(func(lvl zapcore.Level) bool {
-			return lvl >= zapcore.ErrorLevel && lvl >= zapV
+			return lvl >= zapcore.ErrorLevel && r.Enabled(lvl)
 		})
 		lowPriority := zap.LevelEnablerFunc(func(lvl zapcore.Level) bool {
-			return lvl < zapcore.ErrorLevel && lvl >= zapV
+			return lvl < zapcore.ErrorLevel && r.Enabled(lvl)
 		})
 		core = zapcore.NewTee(
 			zapcore.NewCore(encoder, errorStream, highPriority),
 			zapcore.NewCore(encoder, infoStream, lowPriority),
 		)
 	}
 	l := zap.New(core, zap.WithCaller(true))
-	return zapr.NewLoggerWithOptions(l, zapr.LogInfoLevel("v"), zapr.ErrorKey("err")), func() {
-		l.Sync()
-	}
+	return zapr.NewLoggerWithOptions(l, zapr.LogInfoLevel("v"), zapr.ErrorKey("err")),
+		logsapi.RuntimeControl{
+			SetVerbosityLevel: r.SetVerbosityLevel,
+			Flush: func() {
+				_ = l.Sync()
+			},
+		}
 }
 
 func epochMillisTimeEncoder(_ time.Time, enc zapcore.PrimitiveArrayEncoder) {
@@ -91,7 +115,7 @@ func (f Factory) Feature() featuregate.Feature {
 	return logsapi.LoggingBetaOptions
 }
 
-func (f Factory) Create(c logsapi.LoggingConfiguration) (logr.Logger, func()) {
+func (f Factory) Create(c logsapi.LoggingConfiguration, o logsapi.LoggingOptions) (logr.Logger, logsapi.RuntimeControl) {
 	// We intentionally avoid all os.File.Sync calls. Output is unbuffered,
 	// therefore we don't need to flush, and calling the underlying fsync
 	// would just slow down writing.
@@ -100,9 +124,9 @@ func (f Factory) Create(c logsapi.LoggingConfiguration) (logr.Logger, func()) {
 	// written to the output stream before the process terminates, but
 	// doesn't need to worry about data not being written because of a
 	// system crash or powerloss.
-	stderr := zapcore.Lock(AddNopSync(os.Stderr))
+	stderr := zapcore.Lock(AddNopSync(o.ErrorStream))
 	if c.Options.JSON.SplitStream {
-		stdout := zapcore.Lock(AddNopSync(os.Stdout))
+		stdout := zapcore.Lock(AddNopSync(o.InfoStream))
 		size := c.Options.JSON.InfoBufferSize.Value()
 		if size > 0 {
 			// Prevent integer overflow.

staging/src/k8s.io/component-base/logs/json/json_benchmark_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package logs
+package json
 
 import (
 	"fmt"

staging/src/k8s.io/component-base/logs/json/json_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package logs
+package json
 
 import (
 	"bytes"

staging/src/k8s.io/component-base/logs/json/klog_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package logs
+package json
 
 import (
 	"bytes"

staging/src/k8s.io/component-base/logs/logs.go

@@ -23,10 +23,12 @@ import (
 	"flag"
 	"fmt"
 	"log"
+	"strconv"
 	"time"
 
 	"github.com/spf13/pflag"
 	logsapi "k8s.io/component-base/logs/api/v1"
+	"k8s.io/component-base/logs/internal/setverbositylevel"
 	"k8s.io/component-base/logs/klogflags"
 	"k8s.io/klog/v2"
 )
@@ -182,11 +184,26 @@ func NewLogger(prefix string) *log.Logger {
 	return log.New(KlogWriter{}, prefix, 0)
 }
 
-// GlogSetter is a setter to set glog level.
+// GlogSetter modifies the verbosity threshold for the entire program.
+// Some components have HTTP-based APIs for invoking this at runtime.
 func GlogSetter(val string) (string, error) {
+	v, err := strconv.ParseUint(val, 10, 32)
+	if err != nil {
+		return "", err
+	}
+
 	var level klog.Level
 	if err := level.Set(val); err != nil {
 		return "", fmt.Errorf("failed set klog.logging.verbosity %s: %v", val, err)
 	}
+
+	setverbositylevel.Mutex.Lock()
+	defer setverbositylevel.Mutex.Unlock()
+	for _, cb := range setverbositylevel.Callbacks {
+		if err := cb(uint32(v)); err != nil {
+			return "", err
+		}
+	}
+
 	return fmt.Sprintf("successfully set klog.logging.verbosity to %s", val), nil
 }