func (mm *moduleManager) Start() {
	go mm.kubeConfigManager.Start()
	go func() {
		for {
			select {
			case <-mm.globalValuesChanged:
				log.Debugf("MODULE_MANAGER_RUN global values changed")
				mm.EventCh <- Event{Type: GlobalChanged}
			case moduleName := <-mm.moduleValuesChanged:
				log.Debugf("MODULE_MANAGER_RUN module '%s' values changed", moduleName)
				mm.EventCh <- Event{
					Type: ModulesChanged,
					ModulesChanges: []ModuleChange{
						{Name: moduleName, ChangeType: Changed},
					},
				}
			case newKubeConfig := <-kube_config_manager.ConfigUpdated:
				err := mm.validateKubeConfig(mm.kubeConfigManager.CurrentConfig())
				if err != nil {
					log.Errorf("MODULE_MANAGER_RUN ConfigMap changed and is not valid, no ReloadAllModules: %v", err)
					break
				}
				handleRes, err := mm.handleNewKubeConfig(newKubeConfig)
				if err != nil {
					log.Errorf("MODULE_MANAGER_RUN unable to handle kube config update: %s", err)
				}
				if handleRes != nil {
					err = mm.applyKubeUpdate(handleRes)
					if err != nil {
						log.Errorf("MODULE_MANAGER_RUN cannot apply kube config update: %s", err)
					}
				}
			case newModuleConfigs := <-kube_config_manager.ModuleConfigsUpdated:
				mm.moduleConfigsUpdateBeforeAmbiguos = kube_config_manager.ModuleConfigs{}
				err := mm.validateKubeConfig(mm.kubeConfigManager.CurrentConfig())
				if err != nil {
					log.Errorf("MODULE_MANAGER_RUN ConfigMap changed and is not valid, no module restart: %v", err)
					break
				}
				moduleUpdates, err := mm.handleNewKubeModuleConfigs(newModuleConfigs)
				if err != nil {
					mm.moduleConfigsUpdateBeforeAmbiguos = newModuleConfigs
					log.Errorf("Unable to handle update of ConfigMap for modules [%s]: %s", strings.Join(newModuleConfigs.Names(), ", "), err)
				}
				if moduleUpdates != nil {
					err = mm.applyKubeUpdate(moduleUpdates)
					if err != nil {
						log.Errorf("ConfigMap update cannot be applied to values for modules %s: %s", strings.Join(newModuleConfigs.Names(), ", "), err)
					}
				}
			case <-mm.retryOnAmbiguous:
				if len(mm.moduleConfigsUpdateBeforeAmbiguos) != 0 {
					log.Infof("MODULE_MANAGER_RUN Retry saved moduleConfigs: %v", mm.moduleConfigsUpdateBeforeAmbiguos)
					kube_config_manager.ModuleConfigsUpdated <- mm.moduleConfigsUpdateBeforeAmbiguos
				} else {
					log.Debugf("MODULE_MANAGER_RUN Retry IS NOT needed")
				}
			}
		}
	}()
}
Module manager loop
package module_manager
import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"sync"
	klient "github.com/flant/kube-client/client"
	"github.com/hashicorp/go-multierror"
	log "github.com/sirupsen/logrus"
	. "github.com/flant/shell-operator/pkg/hook/binding_context"
	. "github.com/flant/shell-operator/pkg/hook/types"
	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
	. "github.com/flant/addon-operator/pkg/hook/types"
	"github.com/flant/shell-operator/pkg/hook/controller"
	"github.com/flant/shell-operator/pkg/kube/object_patch"
	"github.com/flant/shell-operator/pkg/kube_events_manager"
	"github.com/flant/shell-operator/pkg/metric_storage"
	"github.com/flant/shell-operator/pkg/schedule_manager"
	utils_checksum "github.com/flant/shell-operator/pkg/utils/checksum"
	"github.com/flant/addon-operator/pkg/app"
	"github.com/flant/addon-operator/pkg/helm"
	"github.com/flant/addon-operator/pkg/helm_resources_manager"
	"github.com/flant/addon-operator/pkg/kube_config_manager"
	"github.com/flant/addon-operator/pkg/module_manager/go_hook"
	"github.com/flant/addon-operator/pkg/utils"
	"github.com/flant/addon-operator/pkg/values/validation"
)
type ModuleManager interface {
	Init() error
	Start()
	Ch() chan Event
	WithContext(ctx context.Context)
	WithDirectories(modulesDir string, globalHooksDir string, tempDir string) ModuleManager
	WithKubeEventManager(kube_events_manager.KubeEventsManager)
	WithKubeObjectPatcher(*object_patch.ObjectPatcher)
	WithScheduleManager(schedule_manager.ScheduleManager)
	WithKubeConfigManager(kubeConfigManager kube_config_manager.KubeConfigManager) ModuleManager
	WithHelmResourcesManager(manager helm_resources_manager.HelmResourcesManager)
	WithMetricStorage(storage *metric_storage.MetricStorage)
	WithHookMetricStorage(storage *metric_storage.MetricStorage)
	GetGlobalHooksInOrder(bindingType BindingType) []string
	GetGlobalHook(name string) *GlobalHook
	GetModuleNamesInOrder() []string
	GetModule(name string) *Module
	GetModuleHookNames(moduleName string) []string
	GetModuleHook(name string) *ModuleHook
	GetModuleHooksInOrder(moduleName string, bindingType BindingType) []string
	GlobalStaticAndConfigValues() utils.Values
	GlobalStaticAndNewValues(newValues utils.Values) utils.Values
	GlobalConfigValues() utils.Values
	GlobalValues() (utils.Values, error)
	GlobalValuesPatches() []utils.ValuesPatch
	DiscoverModulesState(logLabels map[string]string) (*ModulesState, error)
	DeleteModule(moduleName string, logLabels map[string]string) error
	RunModule(moduleName string, onStartup bool, logLabels map[string]string, afterStartupCb func() error) (bool, error)
	RunGlobalHook(hookName string, binding BindingType, bindingContext []BindingContext, logLabels map[string]string) (beforeChecksum string, afterChecksum string, err error)
	RunModuleHook(hookName string, binding BindingType, bindingContext []BindingContext, logLabels map[string]string) error
	Retry()
	RegisterModuleHooks(module *Module, logLabels map[string]string) error
	HandleKubeEvent(kubeEvent KubeEvent, createGlobalTaskFn func(*GlobalHook, controller.BindingExecutionInfo), createModuleTaskFn func(*Module, *ModuleHook, controller.BindingExecutionInfo))
	HandleGlobalEnableKubernetesBindings(hookName string, createTaskFn func(*GlobalHook, controller.BindingExecutionInfo)) error
	HandleModuleEnableKubernetesBindings(hookName string, createTaskFn func(*ModuleHook, controller.BindingExecutionInfo)) error
	StartModuleHooks(moduleName string)
	DisableModuleHooks(moduleName string)
	HandleScheduleEvent(crontab string, createGlobalTaskFn func(*GlobalHook, controller.BindingExecutionInfo), createModuleTaskFn func(*Module, *ModuleHook, controller.BindingExecutionInfo)) error
	DynamicEnabledChecksum() string
	ApplyEnabledPatch(enabledPatch utils.ValuesPatch) error
	GlobalSynchronizationNeeded() bool
	GlobalSynchronizationDone() bool
	SynchronizationQueued(id string)
	SynchronizationDone(id string)
	DumpState()
}
type ModulesState struct {
	EnabledModules []string
	ModulesToDisable []string
	ReleasedUnknownModules []string
	NewlyEnabledModules []string
}
type moduleManager struct {
	ctx    context.Context
	cancel context.CancelFunc
	ValuesLock sync.Mutex
	ModulesDir     string
	GlobalHooksDir string
	TempDir        string
	EventCh chan Event
	KubeClient           klient.Client
	KubeObjectPatcher    *object_patch.ObjectPatcher
	kubeEventsManager    kube_events_manager.KubeEventsManager
	scheduleManager      schedule_manager.ScheduleManager
	HelmResourcesManager helm_resources_manager.HelmResourcesManager
	metricStorage        *metric_storage.MetricStorage
	hookMetricStorage    *metric_storage.MetricStorage
	ValuesValidator      *validation.ValuesValidator
	allModulesByName map[string]*Module
	allModulesNamesInOrder []string
	dynamicEnabled map[string]*bool
	enabledModulesByConfig []string
	enabledModulesInOrder []string
	globalHooksByName map[string]*GlobalHook
	globalHooksOrder map[BindingType][]*GlobalHook
	modulesHooksOrderByName map[string]map[BindingType][]*ModuleHook
	kubernetesBindingSynchronizationState map[string]*KubernetesBindingSynchronizationState
	commonStaticValues utils.Values
	kubeGlobalConfigValues utils.Values
	kubeModulesConfigValues map[string]utils.Values
	globalDynamicValuesPatches []utils.ValuesPatch
	modulesDynamicValuesPatches map[string][]utils.ValuesPatch
	moduleValuesChanged chan string
	globalValuesChanged chan bool
	kubeConfigManager kube_config_manager.KubeConfigManager
	moduleConfigsUpdateBeforeAmbiguos kube_config_manager.ModuleConfigs
	retryOnAmbiguous chan bool
}
var _ ModuleManager = &moduleManager{}
type EventType string
const (
	ModulesChanged EventType = "MODULES_CHANGED"
	GlobalChanged EventType = "GLOBAL_CHANGED"
	AmbiguousState EventType = "AMBIGUOUS_STATE"
)
type ChangeType string
const (
	Changed ChangeType = "MODULE_CHANGED"
)
type ModuleChange struct {
	Name       string
	ChangeType ChangeType
}
type Event struct {
	ModulesChanges []ModuleChange
	Type           EventType
}
func NewMainModuleManager() *moduleManager {
	return &moduleManager{
		EventCh:    make(chan Event),
		ValuesLock: sync.Mutex{},
		ValuesValidator: validation.NewValuesValidator(),
		allModulesByName:            make(map[string]*Module),
		allModulesNamesInOrder:      make([]string, 0),
		enabledModulesByConfig:      make([]string, 0),
		enabledModulesInOrder:       make([]string, 0),
		dynamicEnabled:              make(map[string]*bool),
		globalHooksByName:           make(map[string]*GlobalHook),
		globalHooksOrder:            make(map[BindingType][]*GlobalHook),
		modulesHooksOrderByName:     make(map[string]map[BindingType][]*ModuleHook),
		commonStaticValues:          make(utils.Values),
		kubeGlobalConfigValues:      make(utils.Values),
		kubeModulesConfigValues:     make(map[string]utils.Values),
		globalDynamicValuesPatches:  make([]utils.ValuesPatch, 0),
		modulesDynamicValuesPatches: make(map[string][]utils.ValuesPatch),
		moduleValuesChanged: make(chan string, 1),
		globalValuesChanged: make(chan bool, 1),
		kubeConfigManager: nil,
		moduleConfigsUpdateBeforeAmbiguos: make(kube_config_manager.ModuleConfigs),
		retryOnAmbiguous:                  make(chan bool, 1),
		kubernetesBindingSynchronizationState: make(map[string]*KubernetesBindingSynchronizationState),
	}
}
func (mm *moduleManager) WithDirectories(modulesDir string, globalHooksDir string, tempDir string) ModuleManager {
	mm.ModulesDir = modulesDir
	mm.GlobalHooksDir = globalHooksDir
	mm.TempDir = tempDir
	return mm
}
func (mm *moduleManager) WithKubeConfigManager(kubeConfigManager kube_config_manager.KubeConfigManager) ModuleManager {
	mm.kubeConfigManager = kubeConfigManager
	return mm
}
func (mm *moduleManager) WithKubeEventManager(mgr kube_events_manager.KubeEventsManager) {
	mm.kubeEventsManager = mgr
}
func (mm *moduleManager) WithKubeObjectPatcher(patcher *object_patch.ObjectPatcher) {
	mm.KubeObjectPatcher = patcher
}
func (mm *moduleManager) WithScheduleManager(mgr schedule_manager.ScheduleManager) {
	mm.scheduleManager = mgr
}
func (mm *moduleManager) WithHelmResourcesManager(manager helm_resources_manager.HelmResourcesManager) {
	mm.HelmResourcesManager = manager
}
func (mm *moduleManager) WithMetricStorage(storage *metric_storage.MetricStorage) {
	mm.metricStorage = storage
}
func (mm *moduleManager) WithHookMetricStorage(storage *metric_storage.MetricStorage) {
	mm.hookMetricStorage = storage
}
func (mm *moduleManager) WithContext(ctx context.Context) {
	mm.ctx, mm.cancel = context.WithCancel(ctx)
}
func (mm *moduleManager) Stop() {
	if mm.cancel != nil {
		mm.cancel()
	}
}
func (mm *moduleManager) RunModulesEnabledScript(enabledByConfig []string, logLabels map[string]string) ([]string, error) {
	enabledModules := make([]string, 0)
	for _, name := range utils.SortByReference(enabledByConfig, mm.allModulesNamesInOrder) {
		moduleLogLabels := utils.MergeLabels(logLabels)
		moduleLogLabels["module"] = name
		module := mm.allModulesByName[name]
		moduleIsEnabled, err := module.checkIsEnabledByScript(enabledModules, moduleLogLabels)
		if err != nil {
			return nil, err
		}
		if moduleIsEnabled {
			enabledModules = append(enabledModules, name)
		}
	}
	return enabledModules, nil
}
type kubeUpdate struct {
	EnabledModulesByConfig  []string
	KubeGlobalConfigValues  utils.Values
	KubeModulesConfigValues map[string]utils.Values
	Events                  []Event
}
func (mm *moduleManager) applyKubeUpdate(kubeUpdate *kubeUpdate) error {
	log.Debugf("Apply kubeupdate %+v", kubeUpdate)
	mm.kubeGlobalConfigValues = kubeUpdate.KubeGlobalConfigValues
	mm.kubeModulesConfigValues = kubeUpdate.KubeModulesConfigValues
	mm.enabledModulesByConfig = kubeUpdate.EnabledModulesByConfig
	for _, event := range kubeUpdate.Events {
		mm.EventCh <- event
	}
	return nil
}
func (mm *moduleManager) handleNewKubeConfig(newConfig kube_config_manager.Config) (*kubeUpdate, error) {
	logEntry := log.WithField("operator.component", "ModuleManager").
		WithField("operator.action", "handleNewKubeConfig")
	logEntry.Debugf("new kube config received")
	res := &kubeUpdate{
		KubeGlobalConfigValues: newConfig.Values,
		Events:                 []Event{{Type: GlobalChanged}},
	}
	var unknown []utils.ModuleConfig
	res.EnabledModulesByConfig, res.KubeModulesConfigValues, unknown = mm.calculateEnabledModulesByConfig(newConfig.ModuleConfigs)
	for _, moduleConfig := range unknown {
		logEntry.Warnf("Ignore ConfigMap section '%s' for absent module : \n%s",
			moduleConfig.ModuleName,
			moduleConfig.String(),
		)
	}
	return res, nil
}
func (mm *moduleManager) handleNewKubeModuleConfigs(moduleConfigs kube_config_manager.ModuleConfigs) (*kubeUpdate, error) {
	logLabels := map[string]string{
		"operator.component": "HandleConfigMap",
	}
	logEntry := log.WithFields(utils.LabelsToLogFields(logLabels))
	logEntry.Debugf("handle changes in module sections")
	res := &kubeUpdate{
		Events:                 make([]Event, 0),
		KubeGlobalConfigValues: mm.kubeGlobalConfigValues,
	}
	var unknown []utils.ModuleConfig
	res.EnabledModulesByConfig, res.KubeModulesConfigValues, unknown = mm.calculateEnabledModulesByConfig(moduleConfigs)
	for _, moduleConfig := range unknown {
		logEntry.Warnf("ignore module section for unknown module '%s':\n%s",
			moduleConfig.ModuleName, moduleConfig.String())
	}
	updateOnSectionRemove := make(map[string]bool)
	for moduleName, module := range mm.allModulesByName {
		_, hasKubeConfig := moduleConfigs[moduleName]
		if !hasKubeConfig {
			isEnabled := mergeEnabled(
				module.CommonStaticConfig.IsEnabled,
				module.StaticConfig.IsEnabled,
				mm.dynamicEnabled[moduleName])
			_, hasValues := mm.kubeModulesConfigValues[moduleName]
			if isEnabled && hasValues {
				updateOnSectionRemove[moduleName] = true
			}
		}
	}
	res.EnabledModulesByConfig = utils.SortByReference(res.EnabledModulesByConfig, mm.allModulesNamesInOrder)
	logEntry.Debugf("Run enabled script for %+v", res.EnabledModulesByConfig)
	enabledModules, err := mm.RunModulesEnabledScript(res.EnabledModulesByConfig, logLabels)
	if err != nil {
		return nil, err
	}
	logEntry.Infof("Modules enabled by script: %+v", enabledModules)
	if !reflect.DeepEqual(mm.enabledModulesInOrder, enabledModules) {
		logEntry.Debugf("enabled modules set changed from %v to %v: generate GlobalChanged event", mm.enabledModulesInOrder, res.EnabledModulesByConfig)
		res.Events = append(res.Events, Event{Type: GlobalChanged})
	} else {
		logEntry.Debugf("generate ModulesChanged events...")
		moduleChanges := make([]ModuleChange, 0)
		for _, name := range enabledModules {
			isUpdated := false
			moduleConfig, hasKubeConfig := moduleConfigs[name]
			if hasKubeConfig {
				isUpdated = moduleConfig.IsUpdated
				if !isUpdated {
					logEntry.Debugf("ignore module '%s': kube config is not updated", name)
					continue
				}
			}
			_, shouldUpdateAfterRemoval := updateOnSectionRemove[name]
			if (hasKubeConfig && isUpdated) || shouldUpdateAfterRemoval {
				moduleChanges = append(moduleChanges, ModuleChange{Name: name, ChangeType: Changed})
			}
		}
		if len(moduleChanges) > 0 {
			logEntry.Infof("fire ModulesChanged event for %d modules", len(moduleChanges))
			logEntry.Debugf("event changes: %v", moduleChanges)
			res.Events = append(res.Events, Event{Type: ModulesChanged, ModulesChanges: moduleChanges})
		}
	}
	return res, nil
}
func (mm *moduleManager) calculateEnabledModulesByConfig(moduleConfigs kube_config_manager.ModuleConfigs) (enabled []string, values map[string]utils.Values, unknown []utils.ModuleConfig) {
	values = make(map[string]utils.Values)
	log.Debugf("calculateEnabled: dynamicEnabled is %s", mm.DumpDynamicEnabled())
	for moduleName, module := range mm.allModulesByName {
		kubeConfig, hasKubeConfig := moduleConfigs[moduleName]
		if hasKubeConfig {
			isEnabled := mergeEnabled(
				module.CommonStaticConfig.IsEnabled,
				module.StaticConfig.IsEnabled,
				kubeConfig.IsEnabled,
				mm.dynamicEnabled[moduleName])
			if isEnabled {
				enabled = append(enabled, moduleName)
				values[moduleName] = kubeConfig.Values
			}
			log.Debugf("calculateEnabled: module '%s': static enabled %v, kubeConfig: enabled %v, updated %v, dynamic enabled: %v",
				module.Name,
				module.StaticConfig.GetEnabled(),
				kubeConfig.IsEnabled,
				kubeConfig.IsUpdated,
				mm.dynamicEnabled[moduleName])
		} else {
			isEnabled := mergeEnabled(
				module.CommonStaticConfig.IsEnabled,
				module.StaticConfig.IsEnabled,
				mm.dynamicEnabled[moduleName])
			if isEnabled {
				enabled = append(enabled, moduleName)
			}
			log.Debugf("calculateEnabled: module '%s': static enabled %v, no kubeConfig, dynamic enabled: %v",
				module.Name,
				module.StaticConfig.GetEnabled(),
				mm.dynamicEnabled[moduleName])
		}
	}
	for _, kubeConfig := range moduleConfigs {
		if _, hasKey := mm.allModulesByName[kubeConfig.ModuleName]; !hasKey {
			unknown = append(unknown, kubeConfig)
		}
	}
	enabled = utils.SortByReference(enabled, mm.allModulesNamesInOrder)
	return
}
func (mm *moduleManager) Init() error {
	log.Debug("Init ModuleManager")
	if err := mm.RegisterGlobalHooks(); err != nil {
		return err
	}
	if err := mm.RegisterModules(); err != nil {
		return err
	}
	kubeConfig := mm.kubeConfigManager.InitialConfig()
	mm.kubeGlobalConfigValues = kubeConfig.Values
	var unknown []utils.ModuleConfig
	mm.enabledModulesByConfig, mm.kubeModulesConfigValues, unknown = mm.calculateEnabledModulesByConfig(kubeConfig.ModuleConfigs)
	unknownNames := []string{}
	for _, config := range unknown {
		unknownNames = append(unknownNames, config.ModuleName)
	}
	if len(unknownNames) > 0 {
		log.Warnf("ConfigMap/%s has values for absent modules: %+v", app.ConfigMapName, unknownNames)
	}
	return mm.validateKubeConfig(mm.kubeConfigManager.CurrentConfig())
}
func (mm *moduleManager) validateKubeConfig(kubeConfig *kube_config_manager.Config) error {
	var validationErr error
	globalErr := mm.ValuesValidator.ValidateGlobalConfigValues(mm.GlobalStaticAndNewValues(kubeConfig.Values))
	if globalErr != nil {
		validationErr = multierror.Append(
			validationErr,
			fmt.Errorf("'global' section in ConfigMap/%s is not valid", app.ConfigMapName),
			globalErr,
		)
	}
	for _, moduleName := range mm.enabledModulesByConfig {
		mod := mm.allModulesByName[moduleName]
		modCfg, has := kubeConfig.ModuleConfigs[moduleName]
		if !has {
			continue
		}
		moduleErr := mm.ValuesValidator.ValidateModuleConfigValues(mod.ValuesKey(), mod.StaticAndNewValues(modCfg.Values))
		if moduleErr != nil {
			validationErr = multierror.Append(
				validationErr,
				fmt.Errorf("'%s' module section in ConfigMap/%s is not valid", mod.ValuesKey(), app.ConfigMapName),
				moduleErr,
			)
		}
	}
	if validationErr != nil {
		mm.metricStorage.CounterAdd("{PREFIX}config_values_errors_total", 1.0, map[string]string{})
	}
	return validationErr
}