add kube_config extender and scheduler

miklezzzz · miklezzzz · commit dd04cd5d0c3f · 2024-04-09T18:57:10.000+03:00
Signed-off-by: Mikhail Scherba &lt;mikhail.scherba@flant.com&gt;
diff --git a/pkg/kube_config_manager/kube_config_manager.go b/pkg/kube_config_manager/kube_config_manager.go
@@ -72,13 +72,13 @@ func NewKubeConfigManager(ctx context.Context, bk backend.ConfigHandler, runtime
 	}
 }
 
-func (kcm *KubeConfigManager) IsModuleEnabled(moduleName string) bool {
+func (kcm *KubeConfigManager) IsModuleEnabled(moduleName string) *bool {
 	moduleConfig, found := kcm.currentConfig.Modules[moduleName]
 	if !found {
-		return false
+		return nil
 	}
 
-	return moduleConfig.IsEnabled != nil && *moduleConfig.IsEnabled
+	return moduleConfig.IsEnabled
 }
 
 func (kcm *KubeConfigManager) Init() error {
diff --git a/pkg/module_manager/module_manager.go b/pkg/module_manager/module_manager.go
@@ -29,6 +29,7 @@ import (
 	"github.com/flant/addon-operator/pkg/module_manager/models/moduleset"
 	"github.com/flant/addon-operator/pkg/module_manager/scheduler"
 	"github.com/flant/addon-operator/pkg/module_manager/scheduler/extenders/dynamically_enabled"
+	"github.com/flant/addon-operator/pkg/module_manager/scheduler/extenders/kube_config"
 	"github.com/flant/addon-operator/pkg/module_manager/scheduler/extenders/static"
 	"github.com/flant/addon-operator/pkg/task"
 	"github.com/flant/addon-operator/pkg/utils"
@@ -69,7 +70,7 @@ type DirectoryConfig struct {
 
 type KubeConfigManager interface {
 	SaveConfigValues(key string, values utils.Values) error
-	IsModuleEnabled(moduleName string) bool
+	IsModuleEnabled(moduleName string) *bool
 	UpdateModuleConfig(moduleName string) error
 	SafeReadConfig(handler func(config *config.KubeConfig))
 }
@@ -503,6 +504,9 @@ func (mm *ModuleManager) Init() error {
 		mm.moduleScheduler.AddExtender(dynamicExtender)
 	}
 
+	kubeConfigExtender := kube_config.NewExtender(mm.dependencies.KubeConfigManager)
+	mm.moduleScheduler.AddExtender(kubeConfigExtender)
+
 	return mm.registerModules()
 }
 
@@ -1291,8 +1295,10 @@ func (mm *ModuleManager) RegisterModule(moduleSource, modulePath string) error {
 		if !mm.IsModuleEnabled(moduleName) {
 			// update(upsert) module config in moduleset
 			mm.modules.Add(basicModule)
-			// if module is disabled in the module kube config - exit
-			if !mm.dependencies.KubeConfigManager.IsModuleEnabled(moduleName) {
+			// get kube config for the module to check if it has enabled: true
+			moduleKubeConfigEnabled := mm.dependencies.KubeConfigManager.IsModuleEnabled(moduleName)
+			// if module isn't explicitly enabled in the module kube config - exit
+			if moduleKubeConfigEnabled == nil || (moduleKubeConfigEnabled != nil && !*moduleKubeConfigEnabled) {
 				return nil
 			}
 			mm.AddEnabledModuleByConfigName(moduleName)
@@ -1365,8 +1371,10 @@ func (mm *ModuleManager) RegisterModule(moduleSource, modulePath string) error {
 		EventType:  events.ModuleRegistered,
 	})
 
-	// if module is disabled in the module kube config - exit
-	if !mm.dependencies.KubeConfigManager.IsModuleEnabled(moduleName) {
+	// get kube config for the module to check if it has enabled: true
+	moduleKubeConfigEnabled := mm.dependencies.KubeConfigManager.IsModuleEnabled(moduleName)
+	// if module isn't explicitly enabled in the module kube config - exit
+	if moduleKubeConfigEnabled == nil || (moduleKubeConfigEnabled != nil && !*moduleKubeConfigEnabled) {
 		return nil
 	}
 	mm.AddEnabledModuleByConfigName(moduleName)
diff --git a/pkg/module_manager/module_manager_hooks.go b/pkg/module_manager/module_manager_hooks.go
@@ -84,8 +84,9 @@ func (mm *ModuleManager) registerGlobalModule(globalValues utils.Values) (*dynam
 	gm := modules.NewGlobalModule(mm.GlobalHooksDir, globalValues, mm.ValuesValidator, &dep)
 	mm.global = gm
 
-	// catch dynamin Enabled patches from global hooks
+	// applies a scheduler extender to follow which modules get enabled/disabled by dynamic patches
 	dynamicExtender := dynamically_enabled.NewExtender()
+	// catch dynamin Enabled patches from global hooks
 	go mm.runDynamicEnabledLoop(dynamicExtender)
 
 	return dynamicExtender, mm.registerGlobalHooks(gm)
diff --git a/pkg/module_manager/scheduler/extenders/dynamically_enabled/dynamic.go b/pkg/module_manager/scheduler/extenders/dynamically_enabled/dynamic.go
@@ -1,10 +1,13 @@
 package dynamically_enabled
 
 import (
+	"sync"
+
 	log "github.com/sirupsen/logrus"
 )
 
 type Extender struct {
+	l             sync.RWMutex
 	modulesStatus map[string]bool
 }
 
@@ -17,21 +20,26 @@ func NewExtender() *Extender {
 }
 
 func (e *Extender) UpdateStatus(moduleName, operation string) {
+	e.l.Lock()
 	switch operation {
 	case "add":
 		e.modulesStatus[moduleName] = true
 	case "remove":
 		delete(e.modulesStatus, moduleName)
 	default:
-		log.Warn("Unknown patch operation: %s", operation)
+		log.Warnf("Unknown patch operation: %s", operation)
 	}
+	e.l.Unlock()
 }
 
 func (e Extender) Name() string {
 	return "DynamicallyEnabled"
 }
 
 func (e Extender) Filter(moduleName string) *bool {
+	e.l.RLock()
+	defer e.l.RUnlock()
+
 	if val, found := e.modulesStatus[moduleName]; found {
 		return &val
 	}
diff --git a/pkg/module_manager/scheduler/extenders/kube_config/kube_config.go b/pkg/module_manager/scheduler/extenders/kube_config/kube_config.go
@@ -0,0 +1,29 @@
+package kube_config
+
+type kubeConfigManager interface {
+	IsModuleEnabled(moduleName string) *bool
+}
+
+type Extender struct {
+	kubeConfigManager kubeConfigManager
+}
+
+func NewExtender(kcm kubeConfigManager) *Extender {
+	e := &Extender{
+		kubeConfigManager: kcm,
+	}
+
+	return e
+}
+
+func (e Extender) Name() string {
+	return "KubeConfig"
+}
+
+func (e Extender) Filter(moduleName string) *bool {
+	return e.kubeConfigManager.IsModuleEnabled(moduleName)
+}
+
+func (e Extender) Order() error {
+	return nil
+}
diff --git a/pkg/module_manager/scheduler/extenders/static/static_test.go b/pkg/module_manager/scheduler/extenders/static/static_test.go
@@ -40,6 +40,7 @@ fooBarEnabled: false
 # FE Bundle "Default"
 flantIntegrationEnabled: true
 monitoringApplicationsEnabled: true
+l2LoadBalancerEnabled: false
 `
 
 	tmp, err := os.CreateTemp(t.TempDir(), "values-*.yaml")
@@ -52,13 +53,14 @@ monitoringApplicationsEnabled: true
 	assert.NoError(t, err)
 
 	expected := map[string]bool{
-		"admissionPolicyEngine":  true,
-		"certManager":            true,
-		"chrony":                 true,
-		"nodeLocalDns":           true,
-		"fooBar":                 false,
-		"flantIntegration":       true,
-		"monitoringApplications": true,
+		"admission-policy-engine": true,
+		"cert-manager":            true,
+		"chrony":                  true,
+		"node-local-dns":          true,
+		"foo-bar":                 false,
+		"flant-integration":       true,
+		"monitoring-applications": true,
+		"l2-load-balancer":        false,
 	}
 
 	assert.Equal(t, expected, e.modulesStatus)
diff --git a/pkg/module_manager/scheduler/scheduler.go b/pkg/module_manager/scheduler/scheduler.go
@@ -2,30 +2,42 @@ package scheduler
 
 import (
 	"sync"
+
+	log "github.com/sirupsen/logrus"
 )
 
 type Scheduler struct {
 	extenders []Extender
 
-	l sync.Mutex
+	l            sync.Mutex
+	moduleStatus map[string]*bool
 	// result *interface{}
 	// newResult *interface{}
 }
 
 func NewScheduler() *Scheduler {
-	return &Scheduler{extenders: make([]Extender, 0)}
+	return &Scheduler{
+		extenders:    make([]Extender, 0),
+		moduleStatus: make(map[string]*bool),
+	}
 }
 
 func (s *Scheduler) AddExtender(ext Extender) {
 	s.extenders = append(s.extenders, ext)
 }
 
 func (s *Scheduler) Run(moduleNames []string) {
+	s.l.Lock()
 	for _, ex := range s.extenders {
+		log.Infof("Applying %s extender", ex.Name())
 		for _, moduleName := range moduleNames {
-			ex.Filter(moduleName)
+			filterResult := ex.Filter(moduleName)
+			if _, ok := s.moduleStatus[moduleName]; !ok || filterResult != nil {
+				s.moduleStatus[moduleName] = filterResult
+			}
 		}
 	}
+	s.l.Unlock()
 }
 
 type Extender interface {
diff --git a/pkg/module_manager/scheduler/scheduler_test.go b/pkg/module_manager/scheduler/scheduler_test.go
@@ -0,0 +1,64 @@
+package scheduler
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/flant/addon-operator/pkg/module_manager/scheduler/extenders/static"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRun(t *testing.T) {
+
+	values := `
+# Default global values section
+# todo remove duplicate config values they should be in global-hooks/openapi/config-values.yaml only
+# now we have strange behaviour in template tests
+# probably, test helm render does not get defaults from global-hooks/openapi/config-values.yaml
+global:
+  modules:
+    ingressClass: nginx
+    placement: {}
+    https:
+      mode: CertManager
+      certManager:
+        clusterIssuerName: letsencrypt
+    resourcesRequests:
+      everyNode:
+        cpu: 300m
+        memory: 512Mi
+# CE Bundle "Default"
+admissionPolicyEngineEnabled: true
+certManagerEnabled: true
+chronyEnabled: true
+
+# BE Bundle "Default"
+nodeLocalDnsEnabled: true
+
+# EE Bundle "Default"
+fooBarEnabled: false
+
+# FE Bundle "Default"
+flantIntegrationEnabled: true
+monitoringApplicationsEnabled: true
+l2LoadBalancerEnabled: false
+`
+	modules := []string{"admission-policy-engine", "cert-manager", "chrony", "node-local-dns", "foo-bar", "flant-integration", "monitoring-applications", "ingress-nginx", "l2-load-balancer"}
+
+	tmp, err := os.CreateTemp(t.TempDir(), "values-*.yaml")
+	require.NoError(t, err)
+
+	err = os.WriteFile(tmp.Name(), []byte(values), 0x600)
+	require.NoError(t, err)
+
+	e, err := static.NewExtender(tmp.Name())
+	assert.NoError(t, err)
+
+	s := NewScheduler()
+	s.AddExtender(e)
+	s.Run(modules)
+
+	fmt.Println(s.moduleStatus)
+}

Original file line number	Diff line number	Diff line change
`@@ -72,13 +72,13 @@ func NewKubeConfigManager(ctx context.Context, bk backend.ConfigHandler, runtime`
`72`	`72`	`}`
`73`	`73`	`}`
`74`	`74`
`75`		`-func (kcm *KubeConfigManager) IsModuleEnabled(moduleName string) bool {`
	`75`	`+func (kcm KubeConfigManager) IsModuleEnabled(moduleName string) bool {`
`76`	`76`	`moduleConfig, found := kcm.currentConfig.Modules[moduleName]`
`77`	`77`	`if !found {`
`78`		`- return false`
	`78`	`+ return nil`
`79`	`79`	`}`
`80`	`80`
`81`		`- return moduleConfig.IsEnabled != nil && *moduleConfig.IsEnabled`
	`81`	`+ return moduleConfig.IsEnabled`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`func (kcm *KubeConfigManager) Init() error {`
Original file line number	Diff line number	Diff line change
`@@ -1,10 +1,13 @@`
`1`	`1`	`package dynamically_enabled`
`2`	`2`
`3`	`3`	`import (`
	`4`	`+ "sync"`
	`5`	`+`
`4`	`6`	`log "github.com/sirupsen/logrus"`
`5`	`7`	`)`
`6`	`8`
`7`	`9`	`type Extender struct {`
	`10`	`+ l sync.RWMutex`
`8`	`11`	`modulesStatus map[string]bool`
`9`	`12`	`}`
`10`	`13`
`@@ -17,21 +20,26 @@ func NewExtender() *Extender {`
`17`	`20`	`}`
`18`	`21`
`19`	`22`	`func (e *Extender) UpdateStatus(moduleName, operation string) {`
	`23`	`+ e.l.Lock()`
`20`	`24`	`switch operation {`
`21`	`25`	`case "add":`
`22`	`26`	`e.modulesStatus[moduleName] = true`
`23`	`27`	`case "remove":`
`24`	`28`	`delete(e.modulesStatus, moduleName)`
`25`	`29`	`default:`
`26`		`- log.Warn("Unknown patch operation: %s", operation)`
	`30`	`+ log.Warnf("Unknown patch operation: %s", operation)`
`27`	`31`	`}`
	`32`	`+ e.l.Unlock()`
`28`	`33`	`}`
`29`	`34`
`30`	`35`	`func (e Extender) Name() string {`
`31`	`36`	`return "DynamicallyEnabled"`
`32`	`37`	`}`
`33`	`38`
`34`	`39`	`func (e Extender) Filter(moduleName string) *bool {`
	`40`	`+ e.l.RLock()`
	`41`	`+ defer e.l.RUnlock()`
	`42`	`+`
`35`	`43`	`if val, found := e.modulesStatus[moduleName]; found {`
`36`	`44`	`return &val`
`37`	`45`	`}`
Original file line number	Diff line number	Diff line change
`@@ -2,30 +2,42 @@ package scheduler`
`2`	`2`
`3`	`3`	`import (`
`4`	`4`	`"sync"`
	`5`	`+`
	`6`	`+ log "github.com/sirupsen/logrus"`
`5`	`7`	`)`
`6`	`8`
`7`	`9`	`type Scheduler struct {`
`8`	`10`	`extenders []Extender`
`9`	`11`
`10`		`- l sync.Mutex`
	`12`	`+ l sync.Mutex`
	`13`	`+ moduleStatus map[string]*bool`
`11`	`14`	`// result *interface{}`
`12`	`15`	`// newResult *interface{}`
`13`	`16`	`}`
`14`	`17`
`15`	`18`	`func NewScheduler() *Scheduler {`
`16`		`- return &Scheduler{extenders: make([]Extender, 0)}`
	`19`	`+ return &Scheduler{`
	`20`	`+ extenders: make([]Extender, 0),`
	`21`	`+ moduleStatus: make(map[string]*bool),`
	`22`	`+ }`
`17`	`23`	`}`
`18`	`24`
`19`	`25`	`func (s *Scheduler) AddExtender(ext Extender) {`
`20`	`26`	`s.extenders = append(s.extenders, ext)`
`21`	`27`	`}`
`22`	`28`
`23`	`29`	`func (s *Scheduler) Run(moduleNames []string) {`
	`30`	`+ s.l.Lock()`
`24`	`31`	`for _, ex := range s.extenders {`
	`32`	`+ log.Infof("Applying %s extender", ex.Name())`
`25`	`33`	`for _, moduleName := range moduleNames {`
`26`		`- ex.Filter(moduleName)`
	`34`	`+ filterResult := ex.Filter(moduleName)`
	`35`	`+ if _, ok := s.moduleStatus[moduleName]; !ok \|\| filterResult != nil {`
	`36`	`+ s.moduleStatus[moduleName] = filterResult`
	`37`	`+ }`
`27`	`38`	`}`
`28`	`39`	`}`
	`40`	`+ s.l.Unlock()`
`29`	`41`	`}`
`30`	`42`
`31`	`43`	`type Extender interface {`