Skip to content

Commit 5e4e554

Browse files
ldmonsterldmonster
authored
[addon-operator] chore: make labels constant (#751)
Signed-off-by: Pavel Okhlopkov <pavel.okhlopkov@flant.com>
1 parent 65b1d9a commit 5e4e554

File tree

49 files changed

+640
-506
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

49 files changed

+640
-506
lines changed

‎pkg/addon-operator/admission_http_server.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ import (
1010
"time"
1111

1212
"github.com/deckhouse/deckhouse/pkg/log"
13+
14+
"github.com/flant/addon-operator/pkg"
1315
)
1416

1517
type AdmissionServer struct {
@@ -30,7 +32,7 @@ func NewAdmissionServer(listenPort, certsDir string) *AdmissionServer {
3032
func (as *AdmissionServer) RegisterHandler(route string, handler http.Handler) {
3133
if _, ok := as.routes[route]; ok {
3234
log.Fatal("Route is already registered",
33-
slog.String("route", route))
35+
slog.String(pkg.LogKeyRoute, route))
3436
}
3537

3638
as.routes[route] = handler
@@ -45,7 +47,7 @@ func (as *AdmissionServer) start(ctx context.Context) {
4547
}
4648

4749
log.Debug("Registered admission routes",
48-
slog.String("routes", fmt.Sprintf("%v", as.routes)))
50+
slog.String(pkg.LogKeyRoutes, fmt.Sprintf("%v", as.routes)))
4951

5052
srv := &http.Server{
5153
Addr: fmt.Sprintf(":%s", as.listenPort),

‎pkg/addon-operator/bootstrap.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66

77
"github.com/deckhouse/deckhouse/pkg/log"
88

9+
"github.com/flant/addon-operator/pkg"
910
"github.com/flant/addon-operator/pkg/app"
1011
"github.com/flant/addon-operator/pkg/kube_config_manager"
1112
"github.com/flant/addon-operator/pkg/kube_config_manager/backend"
@@ -25,11 +26,11 @@ func (op *AddonOperator) bootstrap() error {
2526

2627
// Log the path where modules will be searched
2728
log.Info("Search modules",
28-
slog.String("path", app.ModulesDir))
29+
slog.String(pkg.LogKeyPath, app.ModulesDir))
2930

3031
// Log the namespace in which the operator will work
3132
log.Info("Addon-operator namespace",
32-
slog.String("namespace", op.DefaultNamespace))
33+
slog.String(pkg.LogKeyNamespace, op.DefaultNamespace))
3334

3435
// Initialize the debug server for troubleshooting and monitoring
3536
// TODO: rewrite shapp global variables to the addon-operator one

‎pkg/addon-operator/handler_manager_events.go‎

Lines changed: 17 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -21,12 +21,12 @@ func (op *AddonOperator) RegisterManagerEventsHandlers() {
2121
// Register handler for schedule events
2222
op.engine.ManagerEventsHandler.WithScheduleEventHandler(func(ctx context.Context, crontab string) []sh_task.Task {
2323
logLabels := map[string]string{
24-
"event.id": uuid.Must(uuid.NewV4()).String(),
24+
pkg.LogKeyEventID: uuid.Must(uuid.NewV4()).String(),
2525
pkg.LogKeyBinding: string(htypes.Schedule),
2626
}
2727
logEntry := utils.EnrichLoggerWithLabels(op.Logger, logLabels)
2828
logEntry.Debug("Create tasks for 'schedule' event",
29-
slog.String("event", crontab))
29+
slog.String(pkg.LogKeyEvent, crontab))
3030

3131
// Handle global hook schedule events
3232
return op.ModuleManager.HandleScheduleEvent(
@@ -40,12 +40,12 @@ func (op *AddonOperator) RegisterManagerEventsHandlers() {
4040
// Register handler for kubernetes events
4141
op.engine.ManagerEventsHandler.WithKubeEventHandler(func(ctx context.Context, kubeEvent types.KubeEvent) []sh_task.Task {
4242
logLabels := map[string]string{
43-
"event.id": uuid.Must(uuid.NewV4()).String(),
43+
pkg.LogKeyEventID: uuid.Must(uuid.NewV4()).String(),
4444
pkg.LogKeyBinding: string(htypes.OnKubernetesEvent),
4545
}
4646
logEntry := utils.EnrichLoggerWithLabels(op.Logger, logLabels)
4747
logEntry.Debug("Create tasks for 'kubernetes' event",
48-
slog.String("event", kubeEvent.String()))
48+
slog.String(pkg.LogKeyEvent, kubeEvent.String()))
4949

5050
// Handle kubernetes events for global and module hooks
5151
tailTasks := op.ModuleManager.HandleKubeEvent(
@@ -73,19 +73,19 @@ func (op *AddonOperator) createGlobalHookTaskFactory(
7373
}
7474

7575
hookLabels := utils.MergeLabels(logLabels, map[string]string{
76-
pkg.LogKeyHook: globalHook.GetName(),
77-
"hook.type": "global",
78-
"queue": info.QueueName,
76+
pkg.LogKeyHook: globalHook.GetName(),
77+
pkg.LogKeyHookType: "global",
78+
pkg.LogKeyQueue: info.QueueName,
7979
})
8080

8181
if len(info.BindingContext) > 0 {
82-
hookLabels["binding.name"] = info.BindingContext[0].Binding
82+
hookLabels[pkg.LogKeyBindingName] = info.BindingContext[0].Binding
8383
if bindingType == htypes.OnKubernetesEvent {
84-
hookLabels["watchEvent"] = string(info.BindingContext[0].WatchEvent)
84+
hookLabels[pkg.LogKeyWatchEvent] = string(info.BindingContext[0].WatchEvent)
8585
}
8686
}
8787

88-
delete(hookLabels, "task.id")
88+
delete(hookLabels, pkg.LogKeyTaskID)
8989

9090
newTask := sh_task.NewTask(task.GlobalHookRun).
9191
WithLogLabels(hookLabels).
@@ -118,20 +118,20 @@ func (op *AddonOperator) createModuleHookTaskFactory(
118118
}
119119

120120
hookLabels := utils.MergeLabels(logLabels, map[string]string{
121-
"module": module.GetName(),
122-
pkg.LogKeyHook: moduleHook.GetName(),
123-
"hook.type": "module",
124-
"queue": info.QueueName,
121+
pkg.LogKeyModule: module.GetName(),
122+
pkg.LogKeyHook: moduleHook.GetName(),
123+
pkg.LogKeyHookType: "module",
124+
pkg.LogKeyQueue: info.QueueName,
125125
})
126126

127127
if len(info.BindingContext) > 0 {
128-
hookLabels["binding.name"] = info.BindingContext[0].Binding
128+
hookLabels[pkg.LogKeyBindingName] = info.BindingContext[0].Binding
129129
if bindingType == htypes.OnKubernetesEvent {
130-
hookLabels["watchEvent"] = string(info.BindingContext[0].WatchEvent)
130+
hookLabels[pkg.LogKeyWatchEvent] = string(info.BindingContext[0].WatchEvent)
131131
}
132132
}
133133

134-
delete(hookLabels, "task.id")
134+
delete(hookLabels, pkg.LogKeyTaskID)
135135

136136
newTask := sh_task.NewTask(task.ModuleHookRun).
137137
WithLogLabels(hookLabels).

‎pkg/addon-operator/handler_module_manager.go‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -20,17 +20,17 @@ import (
2020

2121
func (op *AddonOperator) StartModuleManagerEventHandler() {
2222
go func() {
23-
logEntry := op.Logger.With("operator.component", "handleManagerEvents")
23+
logEntry := op.Logger.With(pkg.LogKeyOperatorComponent, "handleManagerEvents")
2424
for {
2525
select {
2626
case schedulerEvent := <-op.ModuleManager.SchedulerEventCh():
2727
switch event := schedulerEvent.EncapsulatedEvent.(type) {
2828
// dynamically_enabled_extender
2929
case dynamic_extender.DynamicExtenderEvent:
3030
logLabels := map[string]string{
31-
"event.id": uuid.Must(uuid.NewV4()).String(),
32-
"type": "ModuleScheduler event",
33-
"event_source": "DymicallyEnabledExtenderChanged",
31+
pkg.LogKeyEventID: uuid.Must(uuid.NewV4()).String(),
32+
pkg.LogKeyType: "ModuleScheduler event",
33+
pkg.LogKeyEventSource: "DymicallyEnabledExtenderChanged",
3434
}
3535
eventLogEntry := utils.EnrichLoggerWithLabels(logEntry, logLabels)
3636
// if global hooks haven't been run yet, script enabled extender fails due to missing global values
@@ -50,7 +50,7 @@ func (op *AddonOperator) StartModuleManagerEventHandler() {
5050
// if converge has already begun - restart it immediately
5151
if op.engine.TaskQueues.GetMain().Length() > 0 && RemoveCurrentConvergeTasks(op.getConvergeQueues(), logLabels, op.Logger) && op.ConvergeState.GetPhase() != converge.StandBy {
5252
logEntry.Info("ConvergeModules: global hook dynamic modification detected, restart current converge process",
53-
slog.String("phase", string(op.ConvergeState.GetPhase())))
53+
slog.String(pkg.LogKeyPhase, string(op.ConvergeState.GetPhase())))
5454
op.engine.TaskQueues.GetMain().AddFirst(convergeTask)
5555
op.logTaskAdd(eventLogEntry, "DynamicExtender is updated, put first", convergeTask)
5656
} else {
@@ -65,9 +65,9 @@ func (op *AddonOperator) StartModuleManagerEventHandler() {
6565
// kube_config_extender
6666
case config.KubeConfigEvent:
6767
logLabels := map[string]string{
68-
"event.id": uuid.Must(uuid.NewV4()).String(),
69-
"type": "ModuleScheduler event",
70-
"event_source": "KubeConfigExtenderChanged",
68+
pkg.LogKeyEventID: uuid.Must(uuid.NewV4()).String(),
69+
pkg.LogKeyType: "ModuleScheduler event",
70+
pkg.LogKeyEventSource: "KubeConfigExtenderChanged",
7171
}
7272
eventLogEntry := utils.EnrichLoggerWithLabels(logEntry, logLabels)
7373
switch event.Type {
@@ -77,10 +77,10 @@ func (op *AddonOperator) StartModuleManagerEventHandler() {
7777

7878
case config.KubeConfigChanged:
7979
eventLogEntry.Debug("ModuleManagerEventHandler-KubeConfigChanged",
80-
slog.Bool("globalSectionChanged", event.GlobalSectionChanged),
81-
slog.Any("moduleValuesChanged", event.ModuleValuesChanged),
82-
slog.Any("moduleEnabledStateChanged", event.ModuleEnabledStateChanged),
83-
slog.Any("ModuleMaintenanceChanged", event.ModuleMaintenanceChanged))
80+
slog.Bool(pkg.LogKeyGlobalSectionChanged, event.GlobalSectionChanged),
81+
slog.Any(pkg.LogKeyModuleValuesChanged, event.ModuleValuesChanged),
82+
slog.Any(pkg.LogKeyModuleEnabledStateChanged, event.ModuleEnabledStateChanged),
83+
slog.Any(pkg.LogKeyModuleMaintenanceChanged, event.ModuleMaintenanceChanged))
8484
if !op.ModuleManager.GetKubeConfigValid() {
8585
eventLogEntry.Info("KubeConfig become valid")
8686
}
@@ -129,7 +129,7 @@ func (op *AddonOperator) StartModuleManagerEventHandler() {
129129
// if main queue isn't empty and there was another convergeModules task:
130130
if op.engine.TaskQueues.GetMain().Length() > 0 && RemoveCurrentConvergeTasks(op.getConvergeQueues(), logLabels, op.Logger) {
131131
logEntry.Info("ConvergeModules: kube config modification detected, restart current converge process",
132-
slog.String("phase", string(op.ConvergeState.GetPhase())))
132+
slog.String(pkg.LogKeyPhase, string(op.ConvergeState.GetPhase())))
133133
// put ApplyKubeConfig->NewConvergeModulesTask sequence in the beginning of the main queue
134134
if kubeConfigTask != nil {
135135
op.engine.TaskQueues.GetMain().AddFirst(kubeConfigTask)
@@ -174,8 +174,8 @@ func (op *AddonOperator) StartModuleManagerEventHandler() {
174174
op.engine.TaskQueues.GetMain().AddAfter(kubeConfigTask.GetId(), reloadTasks[i])
175175
}
176176
logEntry.Info("ConvergeModules: kube config modification detected, append tasks to rerun modules",
177-
slog.Int("count", len(reloadTasks)),
178-
slog.Any("modules", modulesToRerun))
177+
slog.Int(pkg.LogKeyCount, len(reloadTasks)),
178+
slog.Any(pkg.LogKeyModules, modulesToRerun))
179179
op.logTaskAdd(logEntry, "tail", reloadTasks...)
180180
}
181181
}
@@ -187,8 +187,8 @@ func (op *AddonOperator) StartModuleManagerEventHandler() {
187187

188188
case HelmReleaseStatusEvent := <-op.HelmResourcesManager.Ch():
189189
logLabels := map[string]string{
190-
"event.id": uuid.Must(uuid.NewV4()).String(),
191-
"module": HelmReleaseStatusEvent.ModuleName,
190+
pkg.LogKeyEventID: uuid.Must(uuid.NewV4()).String(),
191+
pkg.LogKeyModule: HelmReleaseStatusEvent.ModuleName,
192192
}
193193
eventLogEntry := utils.EnrichLoggerWithLabels(logEntry, logLabels)
194194

@@ -219,8 +219,8 @@ func (op *AddonOperator) StartModuleManagerEventHandler() {
219219
op.engine.TaskQueues.GetMain().AddLast(newTask.WithQueuedAt(time.Now()))
220220
op.logTaskAdd(logEntry, fmt.Sprintf("detected %s, append", additionalDescription), newTask)
221221
} else {
222-
eventLogEntry.With("task.flow", "noop").Info("Detected event, ModuleRun task already queued",
223-
slog.String("description", additionalDescription))
222+
eventLogEntry.With(pkg.LogKeyTaskFlow, "noop").Info("Detected event, ModuleRun task already queued",
223+
slog.String(pkg.LogKeyDescription, additionalDescription))
224224
}
225225
}
226226
}

‎pkg/addon-operator/operator.go‎

Lines changed: 26 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -234,7 +234,7 @@ func (op *AddonOperator) Setup() error {
234234
return fmt.Errorf("global hooks directory: %s", err)
235235
}
236236
log.Info("global hooks directory",
237-
slog.String("dir", globalHooksDir))
237+
slog.String(pkg.LogKeyDir, globalHooksDir))
238238

239239
tempDir, err := ensureTempDirectory(shapp.TempDir)
240240
if err != nil {
@@ -443,7 +443,7 @@ func (op *AddonOperator) BootstrapMainQueue(tqs *queue.TaskQueueSet) {
443443
tqs.NewNamedQueue("main", op.TaskService.Handle,
444444
queue.WithCompactionCallback(queueutils.CompactionCallback(op.ModuleManager, op.Logger)),
445445
queue.WithCompactableTypes(queueutils.MergeTasks...),
446-
queue.WithLogger(op.Logger.With("operator.component", "mainQueue")),
446+
queue.WithLogger(op.Logger.With(pkg.LogKeyOperatorComponent, "mainQueue")),
447447
)
448448

449449
tasks := op.CreateBootstrapTasks(logLabels)
@@ -457,7 +457,7 @@ func (op *AddonOperator) BootstrapMainQueue(tqs *queue.TaskQueueSet) {
457457
// Add "DiscoverHelmReleases" task to detect unknown releases and purge them.
458458
// this task will run only after the first converge, to keep all modules
459459
discoverLabels := utils.MergeLabels(logLabels, map[string]string{
460-
"queue": "main",
460+
pkg.LogKeyQueue: "main",
461461
pkg.LogKeyBinding: string(task.DiscoverHelmReleases),
462462
})
463463
discoverTask := sh_task.NewTask(task.DiscoverHelmReleases).
@@ -482,10 +482,10 @@ func (op *AddonOperator) CreateBootstrapTasks(logLabels map[string]string) []sh_
482482

483483
for _, hookName := range onStartupHooks {
484484
hookLogLabels := utils.MergeLabels(logLabels, map[string]string{
485-
pkg.LogKeyHook: hookName,
486-
"hook.type": "global",
487-
"queue": "main",
488-
pkg.LogKeyBinding: string(htypes.OnStartup),
485+
pkg.LogKeyHook: hookName,
486+
pkg.LogKeyHookType: "global",
487+
pkg.LogKeyQueue: "main",
488+
pkg.LogKeyBinding: string(htypes.OnStartup),
489489
})
490490

491491
onStartupBindingContext := bc.BindingContext{Binding: string(htypes.OnStartup)}
@@ -509,10 +509,10 @@ func (op *AddonOperator) CreateBootstrapTasks(logLabels map[string]string) []sh_
509509
schedHooks := op.ModuleManager.GetGlobalHooksInOrder(htypes.Schedule)
510510
for _, hookName := range schedHooks {
511511
hookLogLabels := utils.MergeLabels(logLabels, map[string]string{
512-
pkg.LogKeyHook: hookName,
513-
"hook.type": "global",
514-
"queue": "main",
515-
pkg.LogKeyBinding: string(task.GlobalHookEnableScheduleBindings),
512+
pkg.LogKeyHook: hookName,
513+
pkg.LogKeyHookType: "global",
514+
pkg.LogKeyQueue: "main",
515+
pkg.LogKeyBinding: string(task.GlobalHookEnableScheduleBindings),
516516
})
517517

518518
newTask := sh_task.NewTask(task.GlobalHookEnableScheduleBindings).
@@ -529,10 +529,10 @@ func (op *AddonOperator) CreateBootstrapTasks(logLabels map[string]string) []sh_
529529
kubeHooks := op.ModuleManager.GetGlobalHooksInOrder(htypes.OnKubernetesEvent)
530530
for _, hookName := range kubeHooks {
531531
hookLogLabels := utils.MergeLabels(logLabels, map[string]string{
532-
pkg.LogKeyHook: hookName,
533-
"hook.type": "global",
534-
"queue": "main",
535-
pkg.LogKeyBinding: string(task.GlobalHookEnableKubernetesBindings),
532+
pkg.LogKeyHook: hookName,
533+
pkg.LogKeyHookType: "global",
534+
pkg.LogKeyQueue: "main",
535+
pkg.LogKeyBinding: string(task.GlobalHookEnableKubernetesBindings),
536536
})
537537

538538
newTask := sh_task.NewTask(task.GlobalHookEnableKubernetesBindings).
@@ -547,7 +547,7 @@ func (op *AddonOperator) CreateBootstrapTasks(logLabels map[string]string) []sh_
547547

548548
// Task to wait for kubernetes.Synchronization.
549549
waitLogLabels := utils.MergeLabels(logLabels, map[string]string{
550-
"queue": "main",
550+
pkg.LogKeyQueue: "main",
551551
pkg.LogKeyBinding: string(task.GlobalHookWaitKubernetesSynchronization),
552552
})
553553
waitTask := sh_task.NewTask(task.GlobalHookWaitKubernetesSynchronization).
@@ -560,7 +560,7 @@ func (op *AddonOperator) CreateBootstrapTasks(logLabels map[string]string) []sh_
560560

561561
// Add "ConvergeModules" task to run modules converge sequence for the first time.
562562
convergeLabels := utils.MergeLabels(logLabels, map[string]string{
563-
"queue": "main",
563+
pkg.LogKeyQueue: "main",
564564
pkg.LogKeyBinding: string(task.ConvergeModules),
565565
})
566566
convergeTask := converge.NewConvergeModulesTask(eventDescription, converge.OperatorStartup, convergeLabels)
@@ -574,13 +574,13 @@ func (op *AddonOperator) CreateAndStartParallelQueues() {
574574
for i := range app.NumberOfParallelQueues {
575575
queueName := fmt.Sprintf(app.ParallelQueueNamePattern, i)
576576
if op.IsQueueExists(queueName) {
577-
log.Warn("Parallel queue already exists", slog.String("queue", queueName))
577+
log.Warn("Parallel queue already exists", slog.String(pkg.LogKeyQueue, queueName))
578578
continue
579579
}
580580

581581
op.startQueue(queueName, op.TaskService.ParallelHandle)
582582
log.Debug("Parallel queue started",
583-
slog.String("queue", queueName))
583+
slog.String(pkg.LogKeyQueue, queueName))
584584
}
585585
}
586586

@@ -594,7 +594,7 @@ func (op *AddonOperator) startQueue(queueName string, handler func(ctx context.C
594594
op.engine.TaskQueues.NewNamedQueue(queueName, handler,
595595
queue.WithCompactionCallback(queueutils.CompactionCallback(op.ModuleManager, op.Logger)),
596596
queue.WithCompactableTypes(queueutils.MergeTasks...),
597-
queue.WithLogger(op.Logger.With("operator.component", "queue", "queue", queueName)),
597+
queue.WithLogger(op.Logger.With(pkg.LogKeyOperatorComponent, "queue", "queue", queueName)),
598598
)
599599
op.engine.TaskQueues.GetByName(queueName).Start(op.ctx)
600600
}
@@ -615,7 +615,7 @@ func (op *AddonOperator) CreateAndStartQueuesForGlobalHooks() {
615615
op.CreateAndStartQueue(hookBinding.Queue)
616616

617617
log.Debug("Queue started for global 'schedule' hook",
618-
slog.String("queue", hookBinding.Queue),
618+
slog.String(pkg.LogKeyQueue, hookBinding.Queue),
619619
slog.String(pkg.LogKeyHook, hookName))
620620
}
621621
}
@@ -624,7 +624,7 @@ func (op *AddonOperator) CreateAndStartQueuesForGlobalHooks() {
624624
op.CreateAndStartQueue(hookBinding.Queue)
625625

626626
log.Debug("Queue started for global 'kubernetes' hook",
627-
slog.String("queue", hookBinding.Queue),
627+
slog.String(pkg.LogKeyQueue, hookBinding.Queue),
628628
slog.String(pkg.LogKeyHook, hookName))
629629
}
630630
}
@@ -647,7 +647,7 @@ func (op *AddonOperator) CreateAndStartQueuesForModuleHooks(moduleName string) {
647647
op.CreateAndStartQueue(hookBinding.Queue)
648648

649649
log.Debug("Queue started for module 'schedule'",
650-
slog.String("queue", hookBinding.Queue),
650+
slog.String(pkg.LogKeyQueue, hookBinding.Queue),
651651
slog.String(pkg.LogKeyHook, hook.GetName()))
652652
}
653653
}
@@ -660,7 +660,7 @@ func (op *AddonOperator) CreateAndStartQueuesForModuleHooks(moduleName string) {
660660
op.CreateAndStartQueue(hookBinding.Queue)
661661

662662
log.Debug("Queue started for module 'kubernetes'",
663-
slog.String("queue", hookBinding.Queue),
663+
slog.String(pkg.LogKeyQueue, hookBinding.Queue),
664664
slog.String(pkg.LogKeyHook, hook.GetName()))
665665
}
666666
}
@@ -697,8 +697,8 @@ func (op *AddonOperator) CreateReloadModulesTasks(moduleNames []string, logLabel
697697
}
698698

699699
newLogLabels := utils.MergeLabels(logLabels)
700-
newLogLabels["module"] = moduleName
701-
delete(newLogLabels, "task.id")
700+
newLogLabels[pkg.LogKeyModule] = moduleName
701+
delete(newLogLabels, pkg.LogKeyTaskID)
702702

703703
newTask := sh_task.NewTask(task.ModuleRun).
704704
WithLogLabels(newLogLabels).

0 commit comments

Comments
 (0)