feat: make logger more dynamic by allowing synchronous lumberjack

feat: make logger more dynamic by allowing synchronous lumberjack

feat: make logger more dynamic by allowing synchronous lumberjack

Author: JustinChengLZ

cmd/base/options/log.go

@@ -24,29 +24,24 @@ import (
 )
 
 type LogsOptions struct {
-	LogPackageLevel     general.LoggingPKG
-	LogFileMaxSizeInMB  uint64
-	SupportAsyncLogging bool
-	LogDir              string
-	LogBufferSize       int
-	LogFileMaxAge       int
-	LogFileMaxBackups   int
+	LogPackageLevel    general.LoggingPKG
+	LogFileMaxSizeInMB uint64
+	CustomLogDir       string
+	LogBufferSize      int
+	LogFileMaxAge      int
+	LogFileMaxBackups  int
 }
 
 func NewLogsOptions() *LogsOptions {
 	return &LogsOptions{
 		LogPackageLevel:    general.LoggingPKGFull,
 		LogFileMaxSizeInMB: 1800,
-		LogBufferSize:      10000,
-		LogFileMaxAge:      7,
-		LogFileMaxBackups:  10,
 	}
 }
 
 // AddFlags adds flags  to the specified FlagSet.
 func (o *LogsOptions) AddFlags(fs *pflag.FlagSet) {
-	fs.BoolVar(&o.SupportAsyncLogging, "support-async-logging", o.SupportAsyncLogging, "whether to support async logging")
-	fs.StringVar(&o.LogDir, "async_log_dir", o.LogDir, "directory to store logs")
+	fs.StringVar(&o.CustomLogDir, "custom_log_dir", o.CustomLogDir, "directory to store logs for custom logger")
 	fs.Var(&o.LogPackageLevel, "logs-package-level", "the default package level for logging")
 	fs.Uint64Var(&o.LogFileMaxSizeInMB, "log-file-max-size", o.LogFileMaxSizeInMB, "Max size of klog file in MB.")
 	fs.IntVar(&o.LogBufferSize, "log-buffer-size", o.LogBufferSize, "size of the ring buffer to store async logs")
@@ -57,8 +52,7 @@ func (o *LogsOptions) AddFlags(fs *pflag.FlagSet) {
 func (o *LogsOptions) ApplyTo(c *generic.LogConfiguration) error {
 	general.SetDefaultLoggingPackage(o.LogPackageLevel)
 	general.SetLogFileMaxSize(o.LogFileMaxSizeInMB)
-	c.SupportAsyncLogging = o.SupportAsyncLogging
-	c.LogDir = o.LogDir
+	c.CustomLogDir = o.CustomLogDir
 	c.LogBufferSize = o.LogBufferSize
 	c.LogFileMaxSize = int(o.LogFileMaxSizeInMB)
 	c.LogFileMaxAge = o.LogFileMaxAge

cmd/katalyst-agent/app/agent.go

@@ -63,10 +63,8 @@ func Run(
 		return err
 	}
 
-	if conf.SupportAsyncLogging {
-		asyncLogger := logging.NewAsyncLogger(genericCtx, conf.LogDir, conf.LogFileMaxSize, conf.LogFileMaxAge, conf.LogFileMaxBackups, conf.LogBufferSize)
-		defer asyncLogger.Shutdown()
-	}
+	asyncLogger := logging.NewCustomLogger(genericCtx, conf.CustomLogDir, conf.LogFileMaxSize, conf.LogFileMaxAge, conf.LogFileMaxBackups, conf.LogBufferSize)
+	defer asyncLogger.Shutdown()
 
 	for _, genericOption := range genericOptions {
 		genericOption(genericCtx)

pkg/config/generic/log.go

@@ -17,12 +17,11 @@ limitations under the License.
 package generic
 
 type LogConfiguration struct {
-	SupportAsyncLogging bool
-	LogDir              string
-	LogFileMaxSize      int
-	LogBufferSize       int
-	LogFileMaxAge       int
-	LogFileMaxBackups   int
+	CustomLogDir      string
+	LogFileMaxSize    int
+	LogBufferSize     int
+	LogFileMaxAge     int
+	LogFileMaxBackups int
 }
 
 func NewLogConfiguration() *LogConfiguration {

pkg/util/logging/logger.go

@@ -67,18 +67,22 @@ var logInfoMap = map[SeverityName]*logInfo{
 	FatalSeverity:   {fileName: defaultFatalLogFileName, metricsName: metricsNameNumDroppedFatalLogs},
 }
 
-type AsyncLogger struct {
+type CustomLogger struct {
 	diodeWriters []diode.Writer
 }
 
-// NewAsyncLogger creates an async logger that produces an async writer for each of the severity levels.
-// The async writer spins up a goroutine that periodically flushes the buffered logs to disk.
-func NewAsyncLogger(
+// NewCustomLogger creates a custom logger that can either be asynchronous or synchronous, depending on configuration.
+func NewCustomLogger(
 	agentCtx *agent.GenericContext, logDir string, maxSizeMB, maxAge, maxBackups, bufferSize int,
-) *AsyncLogger {
+) *CustomLogger {
 	wrappedEmitter := agentCtx.EmitterPool.GetDefaultMetricsEmitter()
 
-	asyncLogger := &AsyncLogger{}
+	// If logDir is not set, we are still using klog's native logger, so we just return an empty logger without calling SetOutput()
+	if logDir == "" {
+		return &CustomLogger{}
+	}
+
+	customLogger := &CustomLogger{}
 	for severity, logInfo := range logInfoMap {
 		// lumberjackLogger is a logger that rotates log files
 		lumberjackLogger := &lumberjack.Logger{
@@ -88,19 +92,26 @@ func NewAsyncLogger(
 			MaxBackups: maxBackups,
 		}
 
-		// diodeWriter is a writer that stores logs in a ring buffer and asynchronously flushes them
-		diodeWriter := diode.NewWriter(lumberjackLogger, bufferSize, 10*time.Millisecond, func(missed int) {
-			_ = wrappedEmitter.StoreInt64(logInfo.metricsName, int64(missed), metrics.MetricTypeNameRaw)
-		})
-		// Overrides the default synchronous writer with the diode writer
-		klog.SetOutputBySeverity(string(severity), diodeWriter)
-		asyncLogger.diodeWriters = append(asyncLogger.diodeWriters, diodeWriter)
+		// Enable async logger if buffer size is more than 0; otherwise, use synchronous lumberjack logger
+		if bufferSize > 0 {
+			// diodeWriter is a writer that stores logs in a ring buffer and asynchronously flushes them to disk
+			diodeWriter := diode.NewWriter(lumberjackLogger, bufferSize, 10*time.Millisecond, func(missed int) {
+				_ = wrappedEmitter.StoreInt64(logInfo.metricsName, int64(missed), metrics.MetricTypeNameRaw)
+			})
+			// Overrides the default synchronous writer with the diode writer
+			klog.SetOutputBySeverity(string(severity), diodeWriter)
+			customLogger.diodeWriters = append(customLogger.diodeWriters, diodeWriter)
+			klog.Infof("custom async logger is enabled for the severity %s", severity)
+		} else {
+			klog.SetOutputBySeverity(string(severity), lumberjackLogger)
+			klog.Infof("custom sync logger is enabled for the severity %s", severity)
+		}
 	}
 
-	return asyncLogger
+	return customLogger
 }
 
-func (a *AsyncLogger) Shutdown() {
+func (a *CustomLogger) Shutdown() {
 	klog.Info("[Shutdown] async writer is shutting down...")
 	klog.Flush()
 	for _, writer := range a.diodeWriters {

pkg/util/logging/logger_test.go

@@ -26,15 +26,63 @@ import (
 	metrics_pool "github.com/kubewharf/katalyst-core/pkg/metrics/metrics-pool"
 )
 
-func TestNewAsyncLogger(t *testing.T) {
+func TestNewCustomLogger(t *testing.T) {
 	t.Parallel()
-	agentCtx := &agent.GenericContext{
-		GenericContext: &katalyst_base.GenericContext{
-			EmitterPool: metrics_pool.DummyMetricsEmitterPool{},
+	testCases := []struct {
+		name                      string
+		logDir                    string
+		maxSizeMB                 int
+		maxAge                    int
+		maxBackups                int
+		bufferSize                int
+		expectedDiodeWritersCount int
+	}{
+		{
+			name:                      "Disabled logger (logDir is empty), logger is nil",
+			logDir:                    "",
+			maxSizeMB:                 10,
+			maxAge:                    0,
+			maxBackups:                10,
+			bufferSize:                1024,
+			expectedDiodeWritersCount: 0,
+		},
+		{
+			name:                      "Synchronous logger (bufferSize=0)",
+			logDir:                    t.TempDir(),
+			maxSizeMB:                 1,
+			maxAge:                    1,
+			maxBackups:                1,
+			bufferSize:                0,
+			expectedDiodeWritersCount: 0,
+		},
+		{
+			name:                      "Asynchronous logger",
+			logDir:                    t.TempDir(),
+			maxSizeMB:                 1,
+			maxAge:                    1,
+			maxBackups:                1,
+			bufferSize:                1024,
+			expectedDiodeWritersCount: 4,
 		},
 	}
-	asyncLogger := NewAsyncLogger(agentCtx, "testDir", 100, 100, 100, 100)
-	assert.NotNil(t, asyncLogger)
 
-	assert.Equal(t, len(asyncLogger.diodeWriters), 4)
+	for _, tc := range testCases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			agentCtx := &agent.GenericContext{
+				GenericContext: &katalyst_base.GenericContext{
+					EmitterPool: metrics_pool.DummyMetricsEmitterPool{},
+				},
+			}
+
+			logger := NewCustomLogger(agentCtx, tc.logDir, tc.maxSizeMB, tc.maxAge, tc.maxBackups, tc.bufferSize)
+
+			assert.NotNil(t, logger)
+			assert.Len(t, logger.diodeWriters, tc.expectedDiodeWritersCount)
+			assert.NotPanics(t, func() {
+				logger.Shutdown()
+			})
+		})
+	}
 }