This is an automated email from the ASF dual-hosted git repository.
ccondit pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new 292e26fd [YUNIKORN-1797] K8Shim: Implement scoped logging (#611)
292e26fd is described below
commit 292e26fd9d910be39bb98f3bc01b7445179dff9e
Author: Craig Condit <[email protected]>
AuthorDate: Tue Jun 20 09:12:51 2023 -0500
[YUNIKORN-1797] K8Shim: Implement scoped logging (#611)
Add API support for scoped logging to the Kubernetes shim and admission
controller. This will allow for fine-grained logging control per-subsystem.
Closes: #611
Signed-off-by: Craig Condit <[email protected]>
---
Makefile | 6 +
go.mod | 2 +-
go.sum | 4 +-
pkg/admission/conf/am_conf.go | 8 +-
pkg/admission/conf/am_conf_test.go | 8 -
pkg/client/apifactory_mock.go | 1 -
pkg/cmd/admissioncontroller/main.go | 1 +
pkg/cmd/shim/main.go | 3 +-
pkg/conf/schedulerconf.go | 21 +--
pkg/conf/schedulerconf_test.go | 3 -
pkg/log/filtered_core.go | 54 +++++++
pkg/log/logger.go | 223 +++++++++++++++++++++++++-
pkg/log/logger_test.go | 272 ++++++++++++++++++++++++++++++++
pkg/schedulerplugin/scheduler_plugin.go | 3 +-
14 files changed, 570 insertions(+), 39 deletions(-)
diff --git a/Makefile b/Makefile
index 8b8458a1..c710ec23 100644
--- a/Makefile
+++ b/Makefile
@@ -403,6 +403,12 @@ test: clean
go test ./pkg/... -cover -race -tags deadlock
-coverprofile=coverage.txt -covermode=atomic
go vet $(REPO)...
+# Run benchmarks
+.PHONY: bench
+bench:
+ @echo "running benchmarks"
+ go test -v -run '^Benchmark' -bench . ./pkg/...
+
# Generate FSM graphs (dot/png)
.PHONY: fsm_graph
fsm_graph: clean
diff --git a/go.mod b/go.mod
index 8076ff04..dd210270 100644
--- a/go.mod
+++ b/go.mod
@@ -22,7 +22,7 @@ go 1.20
require (
github.com/GoogleCloudPlatform/spark-on-k8s-operator
v0.0.0-20201215015655-2e8b733f5ad0
- github.com/apache/yunikorn-core v0.0.0-20230614110833-fecbbc81241c
+ github.com/apache/yunikorn-core v0.0.0-20230620141112-5d04f9c0dd32
github.com/apache/yunikorn-scheduler-interface
v0.0.0-20230614104003-12c630245d86
github.com/google/uuid v1.2.0
github.com/gorilla/mux v1.8.0
diff --git a/go.sum b/go.sum
index 499c882a..b3f133b2 100644
--- a/go.sum
+++ b/go.sum
@@ -104,8 +104,8 @@ github.com/alecthomas/units
v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod
h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
github.com/antihax/optional v1.0.0/go.mod
h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
github.com/antlr/antlr4/runtime/Go/antlr
v0.0.0-20210826220005-b48c857c3a0e/go.mod
h1:F7bn7fEU90QkQ3tnmaTx3LTKLEDqnwWODIYppRQ5hnY=
-github.com/apache/yunikorn-core v0.0.0-20230614110833-fecbbc81241c
h1:coSVYzpPFsqVGvU9WPLQr0y1l7ZGpIio/S8xuU38qdw=
-github.com/apache/yunikorn-core v0.0.0-20230614110833-fecbbc81241c/go.mod
h1:1K9P/AYVE5RcErlgzyzn/rvHAH4/+fQP7bFRkxvZEW4=
+github.com/apache/yunikorn-core v0.0.0-20230620141112-5d04f9c0dd32
h1:eKGNJ/4SM6r34wHZsfc9JUPqV0lslMVGkaPiZUIcoVo=
+github.com/apache/yunikorn-core v0.0.0-20230620141112-5d04f9c0dd32/go.mod
h1:1K9P/AYVE5RcErlgzyzn/rvHAH4/+fQP7bFRkxvZEW4=
github.com/apache/yunikorn-scheduler-interface
v0.0.0-20230614104003-12c630245d86
h1:GLBGSuJYNjcCdMbKwId2u1BZyhKNzQaocnLqQkRhBLE=
github.com/apache/yunikorn-scheduler-interface
v0.0.0-20230614104003-12c630245d86/go.mod
h1:r4amgZ4uYT1K0KpfYoNV4Ao2GVIYOUU0Rxru89g0WsQ=
github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod
h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
diff --git a/pkg/admission/conf/am_conf.go b/pkg/admission/conf/am_conf.go
index c049bd2e..5cf0a1a9 100644
--- a/pkg/admission/conf/am_conf.go
+++ b/pkg/admission/conf/am_conf.go
@@ -26,7 +26,6 @@ import (
"sync"
"go.uber.org/zap"
- "go.uber.org/zap/zapcore"
v1 "k8s.io/api/core/v1"
informersv1 "k8s.io/client-go/informers/core/v1"
"k8s.io/client-go/tools/cache"
@@ -306,10 +305,6 @@ func (acc *AdmissionControllerConf)
updateConfigMaps(configMaps []*v1.ConfigMap,
// hot refresh
acc.enableConfigHotRefresh = parseConfigBool(configs,
schedulerconf.CMSvcEnableConfigHotRefresh,
schedulerconf.DefaultEnableConfigHotRefresh)
- // logging
- logLevel := parseConfigInt(configs, schedulerconf.CMLogLevel,
schedulerconf.DefaultLoggingLevel)
- log.GetZapConfigs().Level.SetLevel(zapcore.Level(logLevel))
-
// scheduler
acc.policyGroup = parseConfigString(configs,
schedulerconf.CMSvcPolicyGroup, schedulerconf.DefaultPolicyGroup)
@@ -334,6 +329,9 @@ func (acc *AdmissionControllerConf)
updateConfigMaps(configMaps []*v1.ConfigMap,
// labeling
acc.defaultQueueName = parseConfigString(configs,
AMFilteringDefaultQueueName, DefaultFilteringQueueName)
+ // logging
+ log.UpdateLoggingConfig(configs)
+
acc.dumpConfigurationInternal()
}
diff --git a/pkg/admission/conf/am_conf_test.go
b/pkg/admission/conf/am_conf_test.go
index 11fdc6cc..600656b6 100644
--- a/pkg/admission/conf/am_conf_test.go
+++ b/pkg/admission/conf/am_conf_test.go
@@ -21,12 +21,10 @@ package conf
import (
"testing"
- "go.uber.org/zap/zapcore"
"gotest.tools/v3/assert"
v1 "k8s.io/api/core/v1"
schedulerconf "github.com/apache/yunikorn-k8shim/pkg/conf"
- "github.com/apache/yunikorn-k8shim/pkg/log"
)
func TestConfigMapVars(t *testing.T) {
@@ -89,12 +87,6 @@ func TestConfigMapVars(t *testing.T) {
assert.Equal(t, conf.GetTrustControllers(),
DefaultAccessControlTrustControllers)
assert.Equal(t, conf.GetGenerateUniqueAppIds(),
DefaultFilteringGenerateUniqueAppIds)
- // test faulty settings for int values
- NewAdmissionControllerConf([]*v1.ConfigMap{nil, {Data:
map[string]string{
- schedulerconf.CMLogLevel: "not int",
- }}})
- assert.Equal(t, log.GetZapConfigs().Level.Level(),
zapcore.Level(schedulerconf.DefaultLoggingLevel))
-
// test faulty settings for regexp values
conf = NewAdmissionControllerConf([]*v1.ConfigMap{nil, {Data:
map[string]string{
AMFilteringProcessNamespaces: "?",
diff --git a/pkg/client/apifactory_mock.go b/pkg/client/apifactory_mock.go
index 7c60a898..7c2ad65b 100644
--- a/pkg/client/apifactory_mock.go
+++ b/pkg/client/apifactory_mock.go
@@ -72,7 +72,6 @@ func NewMockedAPIProvider(showError bool) *MockedAPIProvider {
PolicyGroup: "queues",
Interval: 0,
KubeConfig: "",
- LoggingLevel: 0,
VolumeBindTimeout: 0,
TestMode: true,
EventChannelCapacity: 0,
diff --git a/pkg/cmd/admissioncontroller/main.go
b/pkg/cmd/admissioncontroller/main.go
index 09e7b221..273d61bb 100644
--- a/pkg/cmd/admissioncontroller/main.go
+++ b/pkg/cmd/admissioncontroller/main.go
@@ -52,6 +52,7 @@ type WebHook struct {
}
func main() {
+ log.SetDefaultLogger(log.Admission)
configMaps, err :=
client.LoadBootstrapConfigMaps(schedulerconf.GetSchedulerNamespace())
if err != nil {
log.Logger().Fatal("Failed to load initial configmaps",
zap.Error(err))
diff --git a/pkg/cmd/shim/main.go b/pkg/cmd/shim/main.go
index 851595cf..cd50f5a2 100644
--- a/pkg/cmd/shim/main.go
+++ b/pkg/cmd/shim/main.go
@@ -55,6 +55,7 @@ func main() {
conf.SiSHA = siSHA
conf.ShimSHA = shimSHA
+ log.SetDefaultLogger(log.K8Shim)
log.Logger().Info(fmt.Sprintf("Build info: version=%s date=%s
isPluginVersion=%t goVersion=%s arch=%s coreSHA=%s siSHA=%s shimSHA=%s",
version, date, false, goVersion, arch, coreSHA, siSHA, shimSHA))
configMaps, err :=
client.LoadBootstrapConfigMaps(conf.GetSchedulerNamespace())
@@ -68,7 +69,7 @@ func main() {
}
log.Logger().Info("Starting scheduler", zap.String("name",
constants.SchedulerName))
- serviceContext := entrypoint.StartAllServicesWithLogger(log.Logger(),
log.GetZapConfigs())
+ serviceContext :=
entrypoint.StartAllServicesWithLogger(log.RootLogger(), log.GetZapConfigs())
if sa, ok := serviceContext.RMProxy.(api.SchedulerAPI); ok {
ss := shim.NewShimScheduler(sa, conf.GetSchedulerConf(),
configMaps)
diff --git a/pkg/conf/schedulerconf.go b/pkg/conf/schedulerconf.go
index e5d1f623..759d8c86 100644
--- a/pkg/conf/schedulerconf.go
+++ b/pkg/conf/schedulerconf.go
@@ -65,9 +65,6 @@ const (
CMSvcPlaceholderImage = PrefixService + "placeholderImage"
CMSvcNodeInstanceTypeNodeLabelKey = PrefixService +
"nodeInstanceTypeNodeLabelKey"
- // log
- CMLogLevel = PrefixLog + "level"
-
// kubernetes
CMKubeQPS = PrefixKubernetes + "qps"
CMKubeBurst = PrefixKubernetes + "burst"
@@ -83,8 +80,6 @@ const (
DefaultOperatorPlugins = "general"
DefaultDisableGangScheduling = false
DefaultEnableConfigHotRefresh = true
- DefaultLoggingLevel = 0
- DefaultLogEncoding = "console"
DefaultKubeQPS = 1000
DefaultKubeBurst = 1000
)
@@ -112,7 +107,6 @@ type SchedulerConf struct {
PolicyGroup string `json:"policyGroup"`
Interval time.Duration `json:"schedulingIntervalSecond"`
KubeConfig string
`json:"absoluteKubeConfigFilePath"`
- LoggingLevel int `json:"loggingLevel"`
VolumeBindTimeout time.Duration `json:"volumeBindTimeout"`
TestMode bool `json:"testMode"`
EventChannelCapacity int `json:"eventChannelCapacity"`
@@ -140,7 +134,6 @@ func (conf *SchedulerConf) Clone() *SchedulerConf {
PolicyGroup: conf.PolicyGroup,
Interval: conf.Interval,
KubeConfig: conf.KubeConfig,
- LoggingLevel: conf.LoggingLevel,
VolumeBindTimeout: conf.VolumeBindTimeout,
TestMode: conf.TestMode,
EventChannelCapacity: conf.EventChannelCapacity,
@@ -182,13 +175,13 @@ func UpdateConfigMaps(configMaps []*v1.ConfigMap, initial
bool) error {
// update scheduler config with merged version
SetSchedulerConf(newConf)
- conf := GetSchedulerConf()
+ _ = GetSchedulerConf()
// update logger configuration
- log.GetZapConfigs().Level.SetLevel(zapcore.Level(conf.LoggingLevel))
+ log.UpdateLoggingConfig(config)
// update Kubernetes logger configuration
- updateKubeLogger(conf)
+ updateKubeLogger()
// dump new scheduler configuration
DumpConfiguration()
@@ -327,7 +320,6 @@ func CreateDefaultConfig() *SchedulerConf {
PolicyGroup: DefaultPolicyGroup,
Interval: DefaultSchedulingInterval,
KubeConfig: GetDefaultKubeConfigPath(),
- LoggingLevel: DefaultLoggingLevel,
VolumeBindTimeout: DefaultVolumeBindTimeout,
TestMode: false,
EventChannelCapacity: DefaultEventChannelCapacity,
@@ -366,9 +358,6 @@ func parseConfig(config map[string]string, prev
*SchedulerConf) (*SchedulerConf,
parser.stringVar(&conf.PlaceHolderImage, CMSvcPlaceholderImage)
parser.stringVar(&conf.InstanceTypeNodeLabelKey,
CMSvcNodeInstanceTypeNodeLabelKey)
- // log
- parser.intVar(&conf.LoggingLevel, CMLogLevel)
-
// kubernetes
parser.intVar(&conf.KubeQPS, CMKubeQPS)
parser.intVar(&conf.KubeBurst, CMKubeBurst)
@@ -434,14 +423,14 @@ func (cp *configParser) durationVar(p *time.Duration,
name string) {
}
}
-func updateKubeLogger(conf *SchedulerConf) {
+func updateKubeLogger() {
// if log level is debug, enable klog and set its log level verbosity
to 4 (represents debug level),
// For details refer to the Logging Conventions of klog at
//
https://github.com/kubernetes/community/blob/master/contributors/devel/sig-instrumentation/logging.md
// danger, this can only be called once!
kubeLoggerOnce.Do(func() {
- if zapcore.Level(conf.LoggingLevel).Enabled(zapcore.DebugLevel)
{
+ if log.Log(log.Kubernetes).Core().Enabled(zapcore.DebugLevel) {
klog.InitFlags(nil)
// cannot really handle the error here ignore it
//nolint:errcheck
diff --git a/pkg/conf/schedulerconf_test.go b/pkg/conf/schedulerconf_test.go
index 11c31b60..210d654b 100644
--- a/pkg/conf/schedulerconf_test.go
+++ b/pkg/conf/schedulerconf_test.go
@@ -65,7 +65,6 @@ func assertDefaults(t *testing.T, conf *SchedulerConf) {
assert.Equal(t, conf.ClusterID, DefaultClusterID)
assert.Equal(t, conf.PolicyGroup, DefaultPolicyGroup)
assert.Equal(t, conf.ClusterVersion, BuildVersion)
- assert.Equal(t, conf.LoggingLevel, DefaultLoggingLevel)
assert.Equal(t, conf.EventChannelCapacity, DefaultEventChannelCapacity)
assert.Equal(t, conf.DispatchTimeout, DefaultDispatchTimeout)
assert.Equal(t, conf.KubeQPS, DefaultKubeQPS)
@@ -90,7 +89,6 @@ func TestParseConfigMap(t *testing.T) {
{CMSvcEnableConfigHotRefresh, "EnableConfigHotRefresh", false},
{CMSvcPlaceholderImage, "PlaceHolderImage", "test-image"},
{CMSvcNodeInstanceTypeNodeLabelKey, "InstanceTypeNodeLabelKey",
"node.kubernetes.io/instance-type"},
- {CMLogLevel, "LoggingLevel", -1},
{CMKubeQPS, "KubeQPS", 2345},
{CMKubeBurst, "KubeBurst", 3456},
}
@@ -123,7 +121,6 @@ func TestUpdateConfigMapNonReloadable(t *testing.T) {
{CMSvcDisableGangScheduling, "DisableGangScheduling", true,
false},
{CMSvcPlaceholderImage, "PlaceHolderImage", "test-image",
false},
{CMSvcNodeInstanceTypeNodeLabelKey, "InstanceTypeNodeLabelKey",
"node.kubernetes.io/instance-type", false},
- {CMLogLevel, "LoggingLevel", -1, true},
{CMKubeQPS, "KubeQPS", 2345, false},
{CMKubeBurst, "KubeBurst", 3456, false},
}
diff --git a/pkg/log/filtered_core.go b/pkg/log/filtered_core.go
new file mode 100644
index 00000000..1e2dcf6a
--- /dev/null
+++ b/pkg/log/filtered_core.go
@@ -0,0 +1,54 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you 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 log
+
+import "go.uber.org/zap/zapcore"
+
+type filteredCore struct {
+ level zapcore.Level
+ inner zapcore.Core
+}
+
+var _ zapcore.Core = filteredCore{}
+
+func (f filteredCore) Enabled(level zapcore.Level) bool {
+ if level < f.level {
+ return false
+ }
+ return f.inner.Enabled(level)
+}
+
+func (f filteredCore) With(fields []zapcore.Field) zapcore.Core {
+ return f.inner.With(fields)
+}
+
+func (f filteredCore) Check(entry zapcore.Entry, ce *zapcore.CheckedEntry)
*zapcore.CheckedEntry {
+ if entry.Level < f.level {
+ return ce
+ }
+ return f.inner.Check(entry, ce)
+}
+
+func (f filteredCore) Write(entry zapcore.Entry, fields []zapcore.Field) error
{
+ return f.inner.Write(entry, fields)
+}
+
+func (f filteredCore) Sync() error {
+ return f.inner.Sync()
+}
diff --git a/pkg/log/logger.go b/pkg/log/logger.go
index 2ba4fb18..81c73a5c 100644
--- a/pkg/log/logger.go
+++ b/pkg/log/logger.go
@@ -20,7 +20,10 @@ package log
import (
"fmt"
+ "strconv"
+ "strings"
"sync"
+ "sync/atomic"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
@@ -30,12 +33,122 @@ var once sync.Once
var logger *zap.Logger
var zapConfigs *zap.Config
+// LoggerHandle is used to efficiently look up logger references
+type LoggerHandle struct {
+ id int
+ name string
+}
+
+func (h LoggerHandle) String() string {
+ return h.name
+}
+
+// Logger constants for configuration
+const (
+ nullLogger = ""
+ defaultLog = "log.level"
+ logPrefix = "log."
+ levelSuffix = ".level"
+)
+
+// Predefined loggers: when adding new loggers, ids must be sequential, and
all must be added to the loggers slice in the same order
+var (
+ K8Shim = &LoggerHandle{id: 1, name: "k8shim"}
+ Kubernetes = &LoggerHandle{id: 2, name: "kubernetes"}
+ Admission = &LoggerHandle{id: 3, name: "admission"}
+ Test = &LoggerHandle{id: 4, name: "test"}
+)
+
+// this tracks all the known logger handles, used to preallocate the real
logger instances when configuration changes
+var loggers = []*LoggerHandle{
+ K8Shim,
+ Kubernetes,
+ Admission,
+ Test,
+}
+
+// structure to hold all current logger configuration state
+type loggerConfig struct {
+ loggers []*zap.Logger
+}
+
+// tracks the currently used set of loggers; replaced completely whenever
configuration changes
+var currentLoggerConfig = atomic.Pointer[loggerConfig]{}
+
+// tracks the default logger handle, which is used for legacy log.Logger()
calls
+var defaultLogger = atomic.Pointer[LoggerHandle]{}
+
+// Logger retrieves the global logger. This is for compatibility with legacy
code and
+// should not be used for new log messages; use Log(loggerHandle) instead
func Logger() *zap.Logger {
+ once.Do(initLogger)
+ return Log(defaultLogger.Load())
+}
+
+// RootLogger retrieves the root logger, used to pass the configured logger to
the scheduler core on startup
+func RootLogger() *zap.Logger {
once.Do(initLogger)
return logger
}
+// Log retrieves a named logger
+func Log(handle *LoggerHandle) *zap.Logger {
+ once.Do(initLogger)
+ if handle == nil || handle.id == 0 {
+ handle = defaultLogger.Load()
+ }
+ conf := currentLoggerConfig.Load()
+ return conf.loggers[handle.id-1]
+}
+
+// createLogger creates a new, named logger that has log levels filtered based
on the given configuration.
+// levelMap contains a mapping of fully-qualified logger names to log levels
+func createLogger(levelMap map[string]zapcore.Level, name string) *zap.Logger {
+ level := loggerLevel(levelMap, name)
+ return logger.Named(name).WithOptions(zap.WrapCore(func(inner
zapcore.Core) zapcore.Core {
+ return filteredCore{inner: inner, level: level}
+ }))
+}
+
+// loggerLevel computes the log level that should be associated with an
arbitrarily named logger.
+// The levelMap is used to look up the level. If not found, the parent logger
is looked up until no further
+// parents can be tried, consulting levelMap for each possible match. For
example, given a logger named
+// k8shim.context.cache, the following keys will be looked up in levelMap:
+//
+// "k8shim.context.cache"
+// "k8shim.context"
+// "k8shim"
+// "" (default logger)
+//
+// The first key that returns a match determines the log level returned. This
allows a level of
+// "k8shim" to control all the child loggers as well if there is not a more
specific override.
+// If no configuration can be found, even for the default empty logger,
InfoLevel will be returned.
+func loggerLevel(levelMap map[string]zapcore.Level, name string) zapcore.Level
{
+ for ; name != nullLogger; name = parentLogger(name) {
+ if level, ok := levelMap[name]; ok {
+ return level
+ }
+ }
+ if level, ok := levelMap[nullLogger]; ok {
+ return level
+ }
+ return zapcore.InfoLevel
+}
+
+// parentLogger returns the name of the parent logger or the empty string ""
if no parent exists.
+// Loggers are named with periods (.) as separators; i.e. the parent logger of
"a.b.c" is "a.b", the parent of
+// "a.b" is "a", and the parent of "a" is "".
+func parentLogger(name string) string {
+ i := strings.LastIndex(name, ".")
+ if i < 0 {
+ // no remaining periods; parent is the null logger
+ return nullLogger
+ }
+ return name[0:i]
+}
+
func initLogger() {
+ defaultLogger.Store(K8Shim)
outputPaths := []string{"stdout"}
zapConfigs = &zap.Config{
@@ -49,7 +162,7 @@ func initLogger() {
MessageKey: "message",
LevelKey: "level",
TimeKey: "time",
- NameKey: "name",
+ NameKey: "logger",
CallerKey: "caller",
StacktraceKey: "stacktrace",
LineEnding: zapcore.DefaultLineEnding,
@@ -72,6 +185,9 @@ func initLogger() {
logger = zap.NewNop()
}
+ // initialize sub-loggers
+ initLoggingConfig(nil)
+
// make sure logs are flushed
//nolint:errcheck
defer logger.Sync()
@@ -82,3 +198,108 @@ func GetZapConfigs() *zap.Config {
_ = Logger()
return zapConfigs
}
+
+// SetDefaultLogger allows customization of the default logger
+func SetDefaultLogger(handle *LoggerHandle) {
+ once.Do(initLogger)
+ defaultLogger.Store(handle)
+}
+
+// UpdateLoggingConfig is used to reconfigure logging. This uses config keys
of the form log.{logger}.level={level}.
+// The default level is set by log.level={level}. The {level} value can be
either numeric (-1 through 5), or
+// textual (DEBUG, INFO, WARN, ERROR, DPANIC, PANIC, or ERROR). See zapcore
documentation for more details.
+func UpdateLoggingConfig(config map[string]string) {
+ once.Do(initLogger)
+ initLoggingConfig(config)
+}
+
+// initLoggingConfig replaces the existing set of loggers with new ones
configured according to the given
+// configuration. All keys of the form "log.{name}.level" will be parsed and a
map of logger name -> logger level
+// will be created. For each defined logger handle (see above), a new logger
instance is created using the
+// most specific configuration found. For example, a logger named "a.b.c" will
be configured by "log.a.b.c.level".
+// If this key does not exist, "a.b" will be consulted, then "a", and finally
"". If no configuration is found for a
+// given key, Info will be used. Finally, the finest log level specified in
any configuration will be used to set the
+// zap log level of the root logger. So if two loggers (one INFO, and one
DEBUG) are configured, the root logger will
+// be set to DEBUG level. If both loggers were at INFO level, the root logger
would be set to INFO.
+// Each configured logger will filter log messages at its own level by
wrapping the underlying zap.Core implementation
+// with one that checks the enabled log level first.
+func initLoggingConfig(config map[string]string) {
+ levelMap := make(map[string]zapcore.Level)
+ levelMap[nullLogger] = zapcore.InfoLevel
+ zapLoggers := make([]*zap.Logger, len(loggers))
+
+ // override default level if found (log.level key)
+ if defaultLevel, ok := config[defaultLog]; ok {
+ if levelRef := parseLevel(defaultLevel); levelRef != nil {
+ levelMap[nullLogger] = *levelRef
+ }
+ }
+
+ // parse out log entries and build level map
+ for k, v := range config {
+ // disallow spaces and periods
+ if strings.Contains(k, "..") || strings.Contains(k, " ") {
+ continue
+ }
+ // ensure config key starts with "log."
+ name, ok := strings.CutPrefix(k, logPrefix)
+ if !ok {
+ continue
+ }
+ // ensure config key ends with ".level"
+ name, ok = strings.CutSuffix(name, levelSuffix)
+ if !ok {
+ continue
+ }
+ // if value is a valid log level, store it in the level map
+ if levelRef := parseLevel(v); levelRef != nil {
+ levelMap[name] = *levelRef
+ }
+ }
+
+ // compute the finest log level necessary to allow all loggers to
succeed
+ minLevel := zapcore.InvalidLevel - 1
+ for _, v := range levelMap {
+ if minLevel > v {
+ minLevel = v
+ }
+ }
+
+ // create each configured logger and initialize the overall
configuration
+ for i := 0; i < len(loggers); i++ {
+ zapLoggers[i] = createLogger(levelMap, loggers[i].name)
+ }
+ newLoggerConfig := loggerConfig{loggers: zapLoggers}
+
+ // update the root zap logger level
+ zapConfigs.Level.SetLevel(minLevel)
+
+ // atomically update the set of loggers
+ currentLoggerConfig.Store(&newLoggerConfig)
+}
+
+// parseLevel parses a textual (or numeric) log level into a zapcore.Level
instance. Both numeric (-1 <= level <= 5)
+// and textual (DEBUG, INFO, WARN, ERROR, DPANIC, PANIC, FATAL) are supported.
+func parseLevel(level string) *zapcore.Level {
+ // parse text
+ zapLevel, err := zapcore.ParseLevel(level)
+ if err == nil {
+ return &zapLevel
+ }
+
+ // parse numeric
+ levelNum, err := strconv.ParseInt(level, 10, 31)
+ if err == nil {
+ zapLevel = zapcore.Level(levelNum)
+ if zapLevel < zapcore.DebugLevel {
+ zapLevel = zapcore.DebugLevel
+ }
+ if zapLevel >= zapcore.InvalidLevel {
+ zapLevel = zapcore.InvalidLevel - 1
+ }
+ return &zapLevel
+ }
+
+ // parse failed
+ return nil
+}
diff --git a/pkg/log/logger_test.go b/pkg/log/logger_test.go
new file mode 100644
index 00000000..e8e800ad
--- /dev/null
+++ b/pkg/log/logger_test.go
@@ -0,0 +1,272 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you 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 log
+
+import (
+ "fmt"
+ "os"
+ "testing"
+ "time"
+
+ "go.uber.org/zap"
+ "go.uber.org/zap/zapcore"
+ "gotest.tools/v3/assert"
+)
+
+var logDir string
+var logFile string
+
+var iterations = 100000
+
+func TestLoggerIds(t *testing.T) {
+ _ = Logger()
+
+ // validate logger count
+ assert.Equal(t, 4, len(loggers), "wrong logger count")
+
+ // validate that all loggers are populated and have sequential ids
+ for i := 0; i < len(loggers); i++ {
+ handle := loggers[i]
+ assert.Assert(t, handle != nil, "nil handle for index", i)
+ assert.Equal(t, handle.id, i+1, "wrong id", handle.name)
+ }
+}
+
+func BenchmarkLegacyLoggerDebug(b *testing.B) {
+ benchmarkLegacyLoggerDebug(b.N)
+}
+
+func TestLegacyLoggerDebug(t *testing.T) {
+ nsOp := benchmarkLegacyLoggerDebug(iterations)
+ RootLogger().Info("log.Logger() performance", zap.Int64("debug
(ns/op)", nsOp))
+}
+
+func benchmarkLegacyLoggerDebug(iterations int) int64 {
+ _ = Logger()
+ initTestLogger()
+ defer resetTestLogger()
+ start := time.Now()
+ for i := 0; i < iterations; i++ {
+ RootLogger().Debug("test", zap.String("foo", "bar"))
+ }
+ return (time.Since(start).Nanoseconds()) / int64(iterations)
+}
+
+func BenchmarkLegacyLoggerInfo(b *testing.B) {
+ benchmarkLegacyLoggerInfo(b.N)
+}
+
+func TestLegacyLoggerInfo(t *testing.T) {
+ nsOp := benchmarkLegacyLoggerInfo(iterations)
+ RootLogger().Info("log.Logger() performance", zap.Int64("info (ns/op)",
nsOp))
+}
+
+func benchmarkLegacyLoggerInfo(iterations int) int64 {
+ _ = Logger()
+ initTestLogger()
+ defer resetTestLogger()
+ start := time.Now()
+ for i := 0; i < iterations; i++ {
+ RootLogger().Info("test", zap.String("foo", "bar"))
+ }
+ return (time.Since(start).Nanoseconds()) / int64(iterations)
+}
+
+func BenchmarkScopedLoggerDebug(b *testing.B) {
+ benchmarkScopedLoggerDebug(b.N)
+}
+
+func TestScopedLoggerDebug(t *testing.T) {
+ nsOp := benchmarkScopedLoggerDebug(iterations)
+ Log(Test).Info("log.Log(...) performance (root=INFO)", zap.Int64("debug
(ns/op)", nsOp))
+}
+
+func benchmarkScopedLoggerDebug(iterations int) int64 {
+ _ = Logger()
+ initTestLogger()
+ defer resetTestLogger()
+ UpdateLoggingConfig(map[string]string{
+ "log.level": "INFO",
+ })
+ start := time.Now()
+ for i := 0; i < iterations; i++ {
+ Log(K8Shim).Debug("test", zap.String("foo", "bar"))
+ }
+ return (time.Since(start).Nanoseconds()) / int64(iterations)
+}
+
+func BenchmarkScopedLoggerInfo(b *testing.B) {
+ benchmarkScopedLoggerInfo(b.N)
+}
+
+func TestScopedLoggerInfo(t *testing.T) {
+ nsOp := benchmarkScopedLoggerInfo(iterations)
+ Log(Test).Info("log.Log(...) performance (root=INFO)", zap.Int64("info
(ns/op)", nsOp))
+}
+
+func benchmarkScopedLoggerInfo(iterations int) int64 {
+ _ = Logger()
+ initTestLogger()
+ defer resetTestLogger()
+ UpdateLoggingConfig(map[string]string{
+ "log.level": "INFO",
+ })
+ start := time.Now()
+ for i := 0; i < iterations; i++ {
+ Log(K8Shim).Info("test", zap.String("foo", "bar"))
+ }
+ return (time.Since(start).Nanoseconds()) / int64(iterations)
+}
+
+func BenchmarkScopedLoggerDebugEnabled(b *testing.B) {
+ benchmarkScopedLoggerDebugEnabled(b.N)
+}
+
+func TestScopedLoggerDebugEnabled(t *testing.T) {
+ nsOp := benchmarkScopedLoggerDebugEnabled(iterations)
+ Log(Test).Info("log.Log(...) performance (root=DEBUG)",
zap.Int64("debug (ns/op)", nsOp))
+}
+
+func benchmarkScopedLoggerDebugEnabled(iterations int) int64 {
+ _ = Logger()
+ initTestLogger()
+ defer resetTestLogger()
+ UpdateLoggingConfig(map[string]string{
+ "log.test.level": "DEBUG",
+ })
+ start := time.Now()
+ for i := 0; i < iterations; i++ {
+ Log(Test).Debug("test", zap.String("foo", "bar"))
+ }
+ return (time.Since(start).Nanoseconds()) / int64(iterations)
+}
+
+func BenchmarkScopedLoggerInfoFiltered(b *testing.B) {
+ benchmarkScopedLoggerInfoFiltered(b.N)
+}
+
+func TestScopedLoggerInfoFiltered(t *testing.T) {
+ nsOp := benchmarkScopedLoggerInfoFiltered(iterations)
+ Log(Test).Info("log.Log(...) performance (root=DEBUG)", zap.Int64("info
(ns/op)", nsOp))
+}
+
+func benchmarkScopedLoggerInfoFiltered(iterations int) int64 {
+ _ = Logger()
+ initTestLogger()
+ defer resetTestLogger()
+ UpdateLoggingConfig(map[string]string{
+ "log.test.level": "DEBUG",
+ })
+ start := time.Now()
+ for i := 0; i < iterations; i++ {
+ Log(K8Shim).Info("test", zap.String("foo", "bar"))
+ }
+ return (time.Since(start).Nanoseconds()) / int64(iterations)
+}
+
+func TestParseLevel(t *testing.T) {
+ assert.Equal(t, zapcore.DebugLevel, *parseLevel("-2"), "out of range
low")
+ assert.Equal(t, zapcore.DebugLevel, *parseLevel("-1"))
+ assert.Equal(t, zapcore.InfoLevel, *parseLevel("0"))
+ assert.Equal(t, zapcore.WarnLevel, *parseLevel("1"))
+ assert.Equal(t, zapcore.ErrorLevel, *parseLevel("2"))
+ assert.Equal(t, zapcore.DPanicLevel, *parseLevel("3"))
+ assert.Equal(t, zapcore.PanicLevel, *parseLevel("4"))
+ assert.Equal(t, zapcore.FatalLevel, *parseLevel("5"))
+ assert.Equal(t, zapcore.FatalLevel, *parseLevel("6"), "out of range
high")
+ assert.Assert(t, parseLevel("+2-3") == nil, "parse error")
+ assert.Equal(t, zapcore.DebugLevel, *parseLevel("Debug"))
+ assert.Equal(t, zapcore.InfoLevel, *parseLevel("iNFO"))
+ assert.Equal(t, zapcore.WarnLevel, *parseLevel("WaRn"))
+ assert.Equal(t, zapcore.ErrorLevel, *parseLevel("ERROR"))
+ assert.Equal(t, zapcore.DPanicLevel, *parseLevel("dpanic"))
+ assert.Equal(t, zapcore.PanicLevel, *parseLevel("PAnIC"))
+ assert.Equal(t, zapcore.FatalLevel, *parseLevel("faTal"))
+ assert.Assert(t, parseLevel("x") == nil, "parse error")
+}
+
+func TestParentLogger(t *testing.T) {
+ assert.Equal(t, "", parentLogger(""), "nullLogger")
+ assert.Equal(t, "", parentLogger("a"), "level 1")
+ assert.Equal(t, "a", parentLogger("a.b"), "level 2")
+ assert.Equal(t, "a.b", parentLogger("a.b.c"), "level 3")
+}
+
+func resetTestLogger() {
+ // flush log
+ logger.Sync() //nolint:errcheck
+
+ // init default logger
+ initLogger()
+
+ // update logger config to defaults
+ UpdateLoggingConfig(map[string]string{})
+
+ if logFile != "" {
+ logFile = ""
+ }
+ if logDir != "" {
+ if err := os.RemoveAll(logDir); err != nil {
+ fmt.Printf("Error removing log dir: %s", err.Error())
+ }
+ }
+}
+
+// initTestLogger is basically the same as the default initLogger() function
but uses a temporary file.
+// this ensures that the logging API is actually used, while allowing us to
avoid massive log spam to stdout
+func initTestLogger() {
+ path, err := os.MkdirTemp("", "log*")
+ if err != nil {
+ panic(err)
+ }
+ logDir = path
+ logFile = fmt.Sprintf("%s/log.stdout", logDir)
+ outputPaths := []string{logFile}
+ zapConfigs = &zap.Config{
+ Level: zap.NewAtomicLevelAt(zapcore.Level(0)),
+ Development: false,
+ DisableCaller: false,
+ DisableStacktrace: false,
+ Sampling: nil,
+ Encoding: "console",
+ EncoderConfig: zapcore.EncoderConfig{
+ MessageKey: "message",
+ LevelKey: "level",
+ TimeKey: "time",
+ NameKey: "logger",
+ CallerKey: "caller",
+ StacktraceKey: "stacktrace",
+ LineEnding: zapcore.DefaultLineEnding,
+ // note:
https://godoc.org/go.uber.org/zap/zapcore#EncoderConfig
+ // only EncodeName is optional all others must be set
+ EncodeLevel: zapcore.CapitalLevelEncoder,
+ EncodeTime: zapcore.ISO8601TimeEncoder,
+ EncodeDuration: zapcore.StringDurationEncoder,
+ EncodeCaller: zapcore.ShortCallerEncoder,
+ },
+ OutputPaths: outputPaths,
+ ErrorOutputPaths: []string{"stderr"},
+ }
+
+ logger, err = zapConfigs.Build()
+ if err != nil {
+ panic(err)
+ }
+ defer logger.Sync() //nolint:errcheck
+}
diff --git a/pkg/schedulerplugin/scheduler_plugin.go
b/pkg/schedulerplugin/scheduler_plugin.go
index 5d1c465a..3a906352 100644
--- a/pkg/schedulerplugin/scheduler_plugin.go
+++ b/pkg/schedulerplugin/scheduler_plugin.go
@@ -216,6 +216,7 @@ func (sp *YuniKornSchedulerPlugin) PostBind(_
context.Context, _ *framework.Cycl
// NewSchedulerPlugin initializes a new plugin and returns it
func NewSchedulerPlugin(_ runtime.Object, handle framework.Handle)
(framework.Plugin, error) {
+ log.SetDefaultLogger(log.K8Shim)
log.Logger().Info(fmt.Sprintf("Build info: version=%s date=%s
isPluginVersion=%t goVersion=%s arch=%s coreSHA=%s siSHA=%s shimSHA=%s",
conf.BuildVersion, conf.BuildDate, conf.IsPluginVersion, conf.GoVersion,
conf.Arch, conf.CoreSHA, conf.SiSHA, conf.ShimSHA))
configMaps, err :=
client.LoadBootstrapConfigMaps(conf.GetSchedulerNamespace())
@@ -229,7 +230,7 @@ func NewSchedulerPlugin(_ runtime.Object, handle
framework.Handle) (framework.Pl
}
// start the YK core scheduler
- serviceContext := entrypoint.StartAllServicesWithLogger(log.Logger(),
log.GetZapConfigs())
+ serviceContext :=
entrypoint.StartAllServicesWithLogger(log.RootLogger(), log.GetZapConfigs())
if sa, ok := serviceContext.RMProxy.(api.SchedulerAPI); ok {
// we need our own informer factory here because the informers
we get from the framework handle aren't yet initialized
informerFactory :=
informers.NewSharedInformerFactory(handle.ClientSet(), 0)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
