Merge pull request #117108 from pohly/test-integration-race-detection-component-base-logs

component-base/logs: improve handling of re-applying a configuration

cmd/kube-apiserver/app/testing/testserver.go

@@ -42,6 +42,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/util/cert"
+	logsapi "k8s.io/component-base/logs/api/v1"
 	"k8s.io/klog/v2"
 	"k8s.io/kube-aggregator/pkg/apiserver"
 
@@ -51,6 +52,13 @@ import (
 	testutil "k8s.io/kubernetes/test/utils"
 )
 
+func init() {
+	// If instantiated more than once or together with other servers, the
+	// servers would try to modify the global logging state. This must get
+	// ignored during testing.
+	logsapi.ReapplyHandling = logsapi.ReapplyHandlingIgnoreUnchanged
+}
+
 // This key is for testing purposes only and is not considered secure.
 const ecdsaPrivateKey = `-----BEGIN EC PRIVATE KEY-----
 MHcCAQEEIEZmTmUhuanLjPA2CLquXivuwBDHTt5XYwgIr/kA1LtRoAoGCCqGSM49

cmd/kube-controller-manager/app/testing/testserver.go

@@ -28,14 +28,21 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
+	logsapi "k8s.io/component-base/logs/api/v1"
 	"k8s.io/klog/v2"
-
 	"k8s.io/kubernetes/cmd/kube-controller-manager/app"
 	kubecontrollerconfig "k8s.io/kubernetes/cmd/kube-controller-manager/app/config"
 	"k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
 	"k8s.io/kubernetes/cmd/kube-controller-manager/names"
 )
 
+func init() {
+	// If instantiated more than once or together with other servers, the
+	// servers would try to modify the global logging state. This must get
+	// ignored during testing.
+	logsapi.ReapplyHandling = logsapi.ReapplyHandlingIgnoreUnchanged
+}
+
 // TearDownFunc is to be called to tear down a test server.
 type TearDownFunc func()
 

cmd/kube-scheduler/app/testing/testserver.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/component-base/configz"
+	logsapi "k8s.io/component-base/logs/api/v1"
 	"k8s.io/kubernetes/cmd/kube-scheduler/app"
 	kubeschedulerconfig "k8s.io/kubernetes/cmd/kube-scheduler/app/config"
 	"k8s.io/kubernetes/cmd/kube-scheduler/app/options"
@@ -36,6 +37,13 @@ import (
 	"k8s.io/klog/v2"
 )
 
+func init() {
+	// If instantiated more than once or together with other servers, the
+	// servers would try to modify the global logging state. This must get
+	// ignored during testing.
+	logsapi.ReapplyHandling = logsapi.ReapplyHandlingIgnoreUnchanged
+}
+
 // TearDownFunc is to be called to tear down a test server.
 type TearDownFunc func()
 

staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/testing/testserver.go

@@ -34,9 +34,17 @@ import (
 	"k8s.io/apiserver/pkg/storage/storagebackend"
 	"k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
+	logsapi "k8s.io/component-base/logs/api/v1"
 	"k8s.io/klog/v2"
 )
 
+func init() {
+	// If instantiated more than once or together with other servers, the
+	// servers would try to modify the global logging state. This must get
+	// ignored during testing.
+	logsapi.ReapplyHandling = logsapi.ReapplyHandlingIgnoreUnchanged
+}
+
 // TearDownFunc is to be called to tear down a test server.
 type TearDownFunc func()
 

staging/src/k8s.io/cloud-provider/app/testing/testserver.go

@@ -34,9 +34,26 @@ import (
 	"k8s.io/cloud-provider/names"
 	"k8s.io/cloud-provider/options"
 	cliflag "k8s.io/component-base/cli/flag"
+	logsapi "k8s.io/component-base/logs/api/v1"
 	"k8s.io/klog/v2"
 )
 
+func init() {
+	// If instantiated more than once or together with other servers, the
+	// servers would try to modify the global logging state. This must get
+	// ignored during testing.
+	logsapi.ReapplyHandling = logsapi.ReapplyHandlingIgnoreUnchanged
+
+	// Because the test server gets started after other goroutines are
+	// running already, we also have to initialize logging here when
+	// those goroutines are not running yet. This works because the
+	// test server uses the default config.
+	config := logsapi.NewLoggingConfiguration()
+	if err := logsapi.ValidateAndApply(config, nil); err != nil {
+		panic(err)
+	}
+}
+
 // TearDownFunc is to be called to tear down a test server.
 type TearDownFunc func()
 

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

@@ -17,14 +17,17 @@ limitations under the License.
 package v1
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"io"
 	"math"
 	"os"
 	"strings"
+	"sync/atomic"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/spf13/pflag"
 
 	"k8s.io/klog/v2"
@@ -57,6 +60,24 @@ func NewLoggingConfiguration() *LoggingConfiguration {
 	return &c
 }
 
+// Applying configurations multiple times is not safe unless it's guaranteed that there
+// are no goroutines which might call logging functions. The default for ValidateAndApply
+// and ValidateAndApplyWithOptions is to return an error when called more than once.
+// Binaries and unit tests can override that behavior.
+var ReapplyHandling = ReapplyHandlingError
+
+type ReapplyHandlingType int
+
+const (
+	// ReapplyHandlingError is the default: calling ValidateAndApply or
+	// ValidateAndApplyWithOptions again returns an error.
+	ReapplyHandlingError ReapplyHandlingType = iota
+	// ReapplyHandlingIgnoreUnchanged silently ignores any additional calls of
+	// ValidateAndApply or ValidateAndApplyWithOptions if the configuration
+	// is unchanged, otherwise they return an error.
+	ReapplyHandlingIgnoreUnchanged
+)
+
 // ValidateAndApply combines validation and application of the logging configuration.
 // This should be invoked as early as possible because then the rest of the program
 // startup (including validation of other options) will already run with the final
@@ -64,6 +85,10 @@ func NewLoggingConfiguration() *LoggingConfiguration {
 //
 // The optional FeatureGate controls logging features. If nil, the default for
 // these features is used.
+//
+// Logging options must be applied as early as possible during the program
+// startup. Some changes are global and cannot be done safely when there are
+// already goroutines running.
 func ValidateAndApply(c *LoggingConfiguration, featureGate featuregate.FeatureGate) error {
 	return validateAndApply(c, nil, featureGate, nil)
 }
@@ -71,6 +96,10 @@ func ValidateAndApply(c *LoggingConfiguration, featureGate featuregate.FeatureGa
 // ValidateAndApplyWithOptions is a variant of ValidateAndApply which accepts
 // additional options beyond those that can be configured through the API. This
 // is meant for testing.
+//
+// Logging options must be applied as early as possible during the program
+// startup. Some changes are global and cannot be done safely when there are
+// already goroutines running.
 func ValidateAndApplyWithOptions(c *LoggingConfiguration, options *LoggingOptions, featureGate featuregate.FeatureGate) error {
 	return validateAndApply(c, options, featureGate, nil)
 }
@@ -183,10 +212,30 @@ func featureEnabled(featureGate featuregate.FeatureGate, feature featuregate.Fea
 }
 
 func apply(c *LoggingConfiguration, options *LoggingOptions, featureGate featuregate.FeatureGate) error {
-	contextualLoggingEnabled := contextualLoggingDefault
-	if featureGate != nil {
-		contextualLoggingEnabled = featureGate.Enabled(ContextualLogging)
+	p := &parameters{
+		C:                        c,
+		Options:                  options,
+		ContextualLoggingEnabled: contextualLoggingDefault,
 	}
+	if featureGate != nil {
+		p.ContextualLoggingEnabled = featureGate.Enabled(ContextualLogging)
+	}
+
+	oldP := applyParameters.Load()
+	if oldP != nil {
+		switch ReapplyHandling {
+		case ReapplyHandlingError:
+			return errors.New("logging configuration was already applied earlier, changing it is not allowed")
+		case ReapplyHandlingIgnoreUnchanged:
+			if diff := cmp.Diff(oldP, p); diff != "" {
+				return fmt.Errorf("the logging configuration should not be changed after setting it once (- old setting, + new setting):\n%s", diff)
+			}
+			return nil
+		default:
+			return fmt.Errorf("invalid value %d for ReapplyHandling", ReapplyHandling)
+		}
+	}
+	applyParameters.Store(p)
 
 	// if log format not exists, use nil loggr
 	format, _ := logRegistry.get(c.Format)
@@ -205,7 +254,7 @@ func apply(c *LoggingConfiguration, options *LoggingOptions, featureGate feature
 			defer setverbositylevel.Mutex.Unlock()
 			setverbositylevel.Callbacks = append(setverbositylevel.Callbacks, control.SetVerbosityLevel)
 		}
-		klog.SetLoggerWithOptions(log, klog.ContextualLogger(contextualLoggingEnabled), klog.FlushLogger(control.Flush))
+		klog.SetLoggerWithOptions(log, klog.ContextualLogger(p.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)
@@ -214,7 +263,40 @@ func apply(c *LoggingConfiguration, options *LoggingOptions, featureGate feature
 		return fmt.Errorf("internal error while setting klog vmodule: %v", err)
 	}
 	klog.StartFlushDaemon(c.FlushFrequency)
-	klog.EnableContextualLogging(contextualLoggingEnabled)
+	klog.EnableContextualLogging(p.ContextualLoggingEnabled)
+	return nil
+}
+
+type parameters struct {
+	C                        *LoggingConfiguration
+	Options                  *LoggingOptions
+	ContextualLoggingEnabled bool
+}
+
+var applyParameters atomic.Pointer[parameters]
+
+// ResetForTest restores the default settings. This is not thread-safe and should only
+// be used when there are no goroutines running. The intended users are unit
+// tests in other packages.
+func ResetForTest(featureGate featuregate.FeatureGate) error {
+	oldP := applyParameters.Load()
+	if oldP == nil {
+		// Nothing to do.
+		return nil
+	}
+
+	// This makes it possible to call apply again without triggering errors.
+	applyParameters.Store(nil)
+
+	// Restore defaults. Shouldn't fail, but check anyway.
+	config := NewLoggingConfiguration()
+	if err := ValidateAndApply(config, featureGate); err != nil {
+		return fmt.Errorf("apply default configuration: %v", err)
+	}
+
+	// And again...
+	applyParameters.Store(nil)
+
 	return nil
 }
 

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

@@ -32,6 +32,34 @@ import (
 	"k8s.io/klog/v2"
 )
 
+func TestReapply(t *testing.T) {
+	oldReapplyHandling := ReapplyHandling
+	defer func() {
+		ReapplyHandling = oldReapplyHandling
+		if err := ResetForTest(nil /* feature gates */); err != nil {
+			t.Errorf("Unexpected error resetting the logging configuration: %v", err)
+		}
+	}()
+
+	newOptions := NewLoggingConfiguration()
+	if err := ValidateAndApply(newOptions, nil); err != nil {
+		t.Errorf("unexpected error for first ValidateAndApply: %v", err)
+	}
+	ReapplyHandling = ReapplyHandlingError
+	if err := ValidateAndApply(newOptions, nil); err == nil {
+		t.Error("did not get expected error for second ValidateAndApply")
+	}
+	ReapplyHandling = ReapplyHandlingIgnoreUnchanged
+	if err := ValidateAndApply(newOptions, nil); err != nil {
+		t.Errorf("unexpected error for third ValidateAndApply: %v", err)
+	}
+	modifiedOptions := newOptions.DeepCopy()
+	modifiedOptions.Verbosity = 100
+	if err := ValidateAndApply(modifiedOptions, nil); err == nil {
+		t.Errorf("unexpected success for forth ValidateAndApply, should have complained about modified config")
+	}
+}
+
 func TestOptions(t *testing.T) {
 	newOptions := NewLoggingConfiguration()
 	testcases := []struct {
@@ -75,6 +103,11 @@ func TestOptions(t *testing.T) {
 			if !assert.Equal(t, tc.want, c) {
 				t.Errorf("Wrong Validate() result for %q. expect %v, got %v", tc.name, tc.want, c)
 			}
+			defer func() {
+				if err := ResetForTest(nil /* feature gates */); err != nil {
+					t.Errorf("Unexpected error resetting the logging configuration: %v", err)
+				}
+			}()
 			errs := ValidateAndApply(c, nil /* We don't care about feature gates here. */)
 			defer klog.StopFlushDaemon()
 			if !assert.ElementsMatch(t, tc.errs, errs) {
@@ -182,6 +215,11 @@ func testContextualLogging(t *testing.T, enabled bool) {
 	AddFeatureGates(featureGate)
 	err = featureGate.SetFromMap(map[string]bool{string(ContextualLogging): enabled})
 	require.NoError(t, err)
+	defer func() {
+		if err := ResetForTest(nil /* feature gates */); err != nil {
+			t.Errorf("Unexpected error resetting the logging configuration: %v", err)
+		}
+	}()
 	err = ValidateAndApply(c, featureGate)
 	require.NoError(t, err)
 	defer klog.StopFlushDaemon()

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

@@ -155,8 +155,12 @@ func TestJSONFormatRegister(t *testing.T) {
 			err := mutable.SetFromMap(map[string]bool{string(logsapi.ContextualLogging): tc.contextualLogging})
 			require.NoError(t, err)
 			featureGate = mutable
+			defer func() {
+				if err := logsapi.ResetForTest(featureGate); err != nil {
+					t.Errorf("Unexpected error while resetting the logging configuration: %v", err)
+				}
+			}()
 			errs := logsapi.ValidateAndApply(c, featureGate)
-			defer klog.ClearLogger()
 			if !assert.ElementsMatch(t, tc.errs, errs) {
 				t.Errorf("Wrong Validate() result for %q.\n expect:\t%+v\n got:\t%+v", tc.name, tc.errs, errs)
 

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

@@ -81,6 +81,11 @@ func TestGlogSetter(t *testing.T) {
 			}
 			os.Stderr = newStderr
 
+			defer func() {
+				if err := logsapi.ResetForTest(nil /* feature gates */); err != nil {
+					t.Errorf("Unexpected error resetting the logging configuration: %v", err)
+				}
+			}()
 			tc.init(t)
 
 			// First write with default verbosity level of 0 -> no output.

test/integration/logs/benchmark/load_test.go

@@ -194,6 +194,11 @@ func TestData(t *testing.T) {
 					InfoStream:  &buffer,
 				}
 				klog.SetOutput(&buffer)
+				defer func() {
+					if err := logsapi.ResetForTest(nil); err != nil {
+						t.Errorf("Unexpected error resetting the logging configuration: %v", err)
+					}
+				}()
 				if err := logsapi.ValidateAndApplyWithOptions(c, &o, nil); err != nil {
 					t.Fatalf("Unexpected error configuring logging: %v", err)
 				}