Repository: flant/shell-operator
Pull request #352 (2022-02-10 08:41:36)

feat: runtime config for "live" debugging options
<!--
Thank you for sending a pull request! Here some tips for contributors:

1. Fill the description template below.
2. Include appropriate tests (if necessary). Make sure that all CI checks passed.
3. If the Pull Request is a work in progress, make use of GitHub's "Draft PR" feature and mark it as such.
-->

#### Overview

- Add runtime configuration parameter "log.level".
- Add a `config` command to list and set runtime parameters.
- A runtime parameter can be set for a period of time.

<img width="660" src="https://user-images.githubusercontent.com/1055474/153444508-3c3a8dd4-5cca-4fd8-8749-d43752d9fd3a.png">


#### What this PR does / why we need it

Change "log.level" on the fly enable debugging without restart.

Also, config with runtime parameters is the one more step to implement https://github.com/flant/shell-operator/issues/101

<!--
- Please state in detail why we need this PR and what it solves.
- If your PR closes some of the existing issues, please add links to them here.
  Mentioned issues will be automatically closed.
  Usage: "Closes #<issue number>", or "Closes (paste link of issue)"
-->

#### Special notes for your reviewer

#### Does this PR introduce a user-facing change?

<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->

```release-note
```

diafour: More changes:

- Add forced duration for log.level=debug

![Снимок экрана 2022-02-21 в 16 40 01](https://user-images.githubusercontent.com/1055474/154966349-6f8243f5-5c0c-4d21-be50-3cc4e2760025.png)


Patch:
diff --git a/.github/workflows/publish-dev.yaml b/.github/workflows/publish-dev.yaml
index 497003b..807d687 100644
--- a/.github/workflows/publish-dev.yaml
+++ b/.github/workflows/publish-dev.yaml
@@ -18,7 +18,7 @@ jobs:
       run_publish: ${{ steps.check.outputs.run_publish }}
       build_multi_arch: ${{ steps.check.outputs.build_multi_arch }}
     steps:
-      - uses: actions/github-script@v5
+      - uses: actions/github-script@v6
         id: check
         with:
           script: |
@@ -105,7 +105,7 @@ jobs:
           version: latest
 
       - name: Login to DockerHub
-        uses: docker/login-action@v1.12.0
+        uses: docker/login-action@v1.13.0
         with:
           username: ${{ secrets.DOCKER_USER }}
           password: ${{ secrets.DOCKER_PASS }}
diff --git a/.github/workflows/publish-release.yaml b/.github/workflows/publish-release.yaml
index aab9963..2478f30 100644
--- a/.github/workflows/publish-release.yaml
+++ b/.github/workflows/publish-release.yaml
@@ -26,7 +26,7 @@ jobs:
       image_tag: ${{ steps.check.outputs.image_tag }}
       additional_tag: ${{ steps.check.outputs.additional_tag }}
     steps:
-      - uses: actions/github-script@v5
+      - uses: actions/github-script@v6
         id: check
         with:
           script: |
@@ -129,7 +129,7 @@ jobs:
           version: latest
 
       - name: Login to DockerHub
-        uses: docker/login-action@v1.12.0
+        uses: docker/login-action@v1.13.0
         with:
           username: ${{ secrets.DOCKER_USER }}
           password: ${{ secrets.DOCKER_PASS }}
diff --git a/cmd/shell-operator/main.go b/cmd/shell-operator/main.go
index b01ef76..c4c9259 100644
--- a/cmd/shell-operator/main.go
+++ b/cmd/shell-operator/main.go
@@ -6,13 +6,13 @@ import (
 	"os"
 	"time"
 
+	"github.com/flant/kube-client/klogtologrus"
 	log "github.com/sirupsen/logrus"
 	"gopkg.in/alecthomas/kingpin.v2"
 
-	"github.com/flant/kube-client/klogtologrus"
-	"github.com/flant/shell-operator/pkg/debug"
-
 	"github.com/flant/shell-operator/pkg/app"
+	"github.com/flant/shell-operator/pkg/config"
+	"github.com/flant/shell-operator/pkg/debug"
 	shell_operator "github.com/flant/shell-operator/pkg/shell-operator"
 	utils_signal "github.com/flant/shell-operator/pkg/utils/signal"
 )
@@ -39,13 +39,15 @@ func main() {
 	startCmd := kpApp.Command("start", "Start shell-operator.").
 		Default().
 		Action(func(c *kingpin.ParseContext) error {
+			runtimeConfig := config.NewConfig()
 			// Init logging subsystem.
-			app.SetupLogging()
+			app.SetupLogging(runtimeConfig)
 			log.Infof("%s %s", app.AppName, app.Version)
 			// Init rand generator.
 			rand.Seed(time.Now().UnixNano())
 
 			defaultOperator := shell_operator.DefaultOperator()
+			defaultOperator.WithRuntimeConfig(runtimeConfig)
 			err := shell_operator.InitAndStart(defaultOperator)
 			if err != nil {
 				os.Exit(1)
diff --git a/pkg/app/log.go b/pkg/app/log.go
index 6ba1f24..50be16c 100644
--- a/pkg/app/log.go
+++ b/pkg/app/log.go
@@ -1,10 +1,14 @@
 package app
 
 import (
+	"fmt"
 	"strings"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 	"gopkg.in/alecthomas/kingpin.v2"
+
+	"github.com/flant/shell-operator/pkg/config"
 )
 
 // Use info level with timestamps and a text output by default
@@ -12,7 +16,10 @@ var LogLevel = "info"
 var LogNoTime = false
 var LogType = "text"
 
-// SetupLoggingSettings init global flags for logging
+// ForcedDurationForDebugLevel - force expiration for debug level.
+const ForcedDurationForDebugLevel = 30 * time.Minute
+
+// DefineLoggingFlags defines flags for logger settings.
 func DefineLoggingFlags(cmd *kingpin.CmdClause) {
 	cmd.Flag("log-level", "Logging level: debug, info, error. Default is info. Can be set with $LOG_LEVEL.").
 		Envar("LOG_LEVEL").
@@ -27,9 +34,9 @@ func DefineLoggingFlags(cmd *kingpin.CmdClause) {
 		BoolVar(&LogNoTime)
 }
 
-// SetupLogging sets logging output
-func SetupLogging() {
-	switch LogType {
+// SetupLogging sets logger formatter and level.
+func SetupLogging(runtimeConfig *config.Config) {
+	switch strings.ToLower(LogType) {
 	case "json":
 		log.SetFormatter(&log.JSONFormatter{DisableTimestamp: LogNoTime})
 	case "text":
@@ -40,7 +47,25 @@ func SetupLogging() {
 		log.SetFormatter(&log.JSONFormatter{DisableTimestamp: LogNoTime})
 	}
 
-	switch strings.ToLower(LogLevel) {
+	setLogLevel(LogLevel)
+
+	runtimeConfig.Register("log.level",
+		fmt.Sprintf("Global log level. Default duration for debug level is %s", ForcedDurationForDebugLevel),
+		strings.ToLower(LogLevel),
+		func(oldValue string, newValue string) error {
+			log.Infof("Set log level to '%s'", newValue)
+			setLogLevel(newValue)
+			return nil
+		}, func(oldValue string, newValue string) time.Duration {
+			if strings.ToLower(newValue) == "debug" {
+				return ForcedDurationForDebugLevel
+			}
+			return 0
+		})
+}
+
+func setLogLevel(logLevel string) {
+	switch strings.ToLower(logLevel) {
 	case "debug":
 		log.SetLevel(log.DebugLevel)
 	case "error":
diff --git a/pkg/config/config.go b/pkg/config/config.go
new file mode 100644
index 0000000..89c311d
--- /dev/null
+++ b/pkg/config/config.go
@@ -0,0 +1,274 @@
+package config
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+/**
+ * Runtime configuration parameters. This is a simple flat KV storage.
+ */
+
+// Parameter is a runtime configuration parameter.
+type Parameter struct {
+	name          string
+	description   string
+	defaultValue  string
+	onChange      func(oldValue string, newValue string) error
+	forceDuration func(oldValue string, newValue string) time.Duration
+}
+
+// TemporalValue is a value for parameter that will expire in the future.
+type TemporalValue struct {
+	value  string
+	expire time.Time
+}
+
+const CheckExpiredTemporalValuesPeriod = 15 * time.Second
+
+// Config is a storage for all runtime parameters.
+type Config struct {
+	m      sync.Mutex
+	params map[string]*Parameter
+	values map[string]string
+	// Errors from onChange handlers.
+	errors map[string]error
+	// Temporal values with expiration.
+	temporalValues map[string]*TemporalValue
+	expireTicker   *time.Ticker
+
+	logEntry *log.Entry
+}
+
+func NewConfig() *Config {
+	return &Config{
+		params:         make(map[string]*Parameter),
+		values:         make(map[string]string),
+		temporalValues: make(map[string]*TemporalValue),
+		errors:         make(map[string]error),
+		logEntry:       log.WithField("component", "runtimeConfig"),
+	}
+}
+
+func (c *Config) Register(name string, description string, defaultValue string, onChange func(oldValue string, newValue string) error, forceDuration func(oldValue string, newValue string) time.Duration) {
+	if c == nil {
+		return
+	}
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	if _, exists := c.params[name]; exists {
+		return
+	}
+	c.params[name] = &Parameter{
+		name:          name,
+		defaultValue:  defaultValue,
+		description:   description,
+		onChange:      onChange,
+		forceDuration: forceDuration,
+	}
+}
+
+func (c *Config) List() []map[string]string {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	res := make([]map[string]string, 0)
+	for paramName, param := range c.params {
+		paramValue := c.value(paramName)
+		paramInfo := map[string]string{
+			"name":        paramName,
+			"description": param.description,
+			"default":     param.defaultValue,
+			"value":       paramValue,
+		}
+		if tempValue, ok := c.temporalValues[paramName]; ok {
+			paramInfo["expireAt"] = tempValue.expire.Format(time.RFC3339)
+		}
+		lastError := c.errors[paramName]
+		if lastError != nil {
+			paramInfo["lastError"] = lastError.Error()
+		}
+		res = append(res, paramInfo)
+	}
+	sort.SliceStable(res, func(i, j int) bool {
+		return res[i]["name"] < res[j]["name"]
+	})
+	return res
+}
+
+func (c *Config) String() string {
+	b := new(strings.Builder)
+	// Header
+	b.WriteString(fmt.Sprintf("%-30s %-20s %-20s %-40s\n", "NAME", "VALUE", "EXPIRE AT", "DESCRIPTION"))
+
+	params := c.List()
+	for _, param := range params {
+		description := param["description"]
+		if param["lastError"] != "" {
+			description = "Error: " + param["lastError"]
+		}
+		b.WriteString(fmt.Sprintf("%-30s %-20s %-20s %-40s\n", param["name"], param["value"], param["expireAt"], description))
+	}
+
+	return b.String()
+}
+
+func (c *Config) Has(name string) bool {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	return c.has(name)
+}
+
+func (c *Config) has(name string) bool {
+	_, registered := c.params[name]
+	return registered
+}
+
+func (c *Config) LastError(name string) error {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	err := c.errors[name]
+	return err
+}
+
+// Set updates a value of the parameter by its name.
+// Deletes a temporal value if set.
+func (c *Config) Set(name string, value string) {
+	forceDuration := c.callForceDuration(name, value)
+	if forceDuration > 0 {
+		c.SetTemporarily(name, value, forceDuration)
+		return
+	}
+
+	c.m.Lock()
+	oldValue := c.value(name)
+	delete(c.temporalValues, name)
+	c.values[name] = value
+	c.m.Unlock()
+
+	c.callOnChange(name, oldValue, value)
+}
+
+// Unset removes temporal value and value for parameter by its name.
+func (c *Config) Unset(name string) {
+	c.m.Lock()
+	oldValue := c.value(name)
+	delete(c.values, name)
+	delete(c.temporalValues, name)
+	newValue := c.value(name)
+	c.m.Unlock()
+
+	c.callOnChange(name, oldValue, newValue)
+}
+
+func (c *Config) Value(name string) string {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	return c.value(name)
+}
+
+func (c *Config) value(name string) string {
+	if !c.has(name) {
+		return ""
+	}
+
+	if v, ok := c.temporalValues[name]; ok {
+		return v.value
+	}
+	if v, ok := c.values[name]; ok {
+		return v
+	}
+	return c.params[name].defaultValue
+}
+
+func (c *Config) SetTemporarily(name string, value string, duration time.Duration) {
+	c.m.Lock()
+	oldValue := c.value(name)
+	delete(c.temporalValues, name)
+	c.temporalValues[name] = &TemporalValue{
+		value:  value,
+		expire: time.Now().Add(duration),
+	}
+	newValue := c.value(name)
+	// Start go routine to expire temporal values.
+	if c.expireTicker == nil {
+		c.expireTicker = time.NewTicker(CheckExpiredTemporalValuesPeriod)
+		go func() {
+			for {
+				<-c.expireTicker.C
+				c.expireOverrides()
+			}
+		}()
+	}
+
+	c.m.Unlock()
+
+	c.callOnChange(name, oldValue, newValue)
+}
+
+// TODO accumulate changes and call onChange for all expired params (outside of locking, because callback can be long-lasted).
+func (c *Config) expireOverrides() {
+	if len(c.temporalValues) == 0 {
+		return
+	}
+
+	now := time.Now()
+	expires := make([][]string, 0)
+
+	c.m.Lock()
+	for name, temporalValue := range c.temporalValues {
+		if temporalValue.expire.Before(now) {
+			oldValue := c.value(name)
+			delete(c.temporalValues, name)
+			newValue := c.value(name)
+			expires = append(expires, []string{name, oldValue, newValue})
+		}
+	}
+	c.m.Unlock()
+
+	for _, expire := range expires {
+		name, oldValue, newValue := expire[0], expire[1], expire[2]
+		c.logEntry.Debugf("Parameter '%s' expired", name)
+		c.callOnChange(name, oldValue, newValue)
+	}
+}
+
+// callOnChange executes onChange handler if defined for the parameter.
+func (c *Config) callOnChange(name string, oldValue string, newValue string) {
+	if c.params[name].onChange == nil {
+		return
+	}
+	err := c.params[name].onChange(oldValue, newValue)
+	if err != nil {
+		c.logEntry.Errorf("OnChange handler failed for '%s' during value change from '%s' to '%s': %v",
+			name, oldValue, newValue, err)
+	}
+	c.m.Lock()
+	delete(c.errors, name)
+	c.errors[name] = err
+	c.m.Unlock()
+}
+
+func (c *Config) callForceDuration(name string, newValue string) time.Duration {
+	c.m.Lock()
+	defer c.m.Unlock()
+	param, has := c.params[name]
+	if !has {
+		return 0
+	}
+	if param.forceDuration == nil {
+		return 0
+	}
+
+	oldValue := c.value(name)
+	return param.forceDuration(oldValue, newValue)
+}
diff --git a/pkg/debug/client.go b/pkg/debug/client.go
index 244040e..b928536 100644
--- a/pkg/debug/client.go
+++ b/pkg/debug/client.go
@@ -1,9 +1,10 @@
 package debug
 
 import (
+	"bytes"
 	"context"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net"
 	"net/http"
 
@@ -59,5 +60,30 @@ func (c *Client) Get(url string) ([]byte, error) {
 	}
 	defer resp.Body.Close()
 
-	return ioutil.ReadAll(resp.Body)
+	bodyBuf := new(bytes.Buffer)
+	_, err = io.Copy(bodyBuf, resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	return bodyBuf.Bytes(), nil
+}
+
+func (c *Client) Post(targetUrl string, data map[string][]string) ([]byte, error) {
+	httpc, err := c.newHttpClient()
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := httpc.PostForm(targetUrl, data)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	bodyBuf := new(bytes.Buffer)
+	_, err = io.Copy(bodyBuf, resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	return bodyBuf.Bytes(), nil
 }
diff --git a/pkg/debug/debug-cmd.go b/pkg/debug/debug-cmd.go
index 16721b8..438a237 100644
--- a/pkg/debug/debug-cmd.go
+++ b/pkg/debug/debug-cmd.go
@@ -2,6 +2,7 @@ package debug
 
 import (
 	"fmt"
+	"time"
 
 	"gopkg.in/alecthomas/kingpin.v2"
 
@@ -38,6 +39,38 @@ func DefineDebugCommands(kpApp *kingpin.Application) {
 	AddOutputJsonYamlTextFlag(queueMainCmd)
 	app.DefineDebugUnixSocketFlag(queueMainCmd)
 
+	// Runtime config command.
+	configCmd := app.CommandWithDefaultUsageTemplate(kpApp, "config", "Manage runtime parameters.")
+
+	configListCmd := configCmd.Command("list", "List available runtime parameters.").
+		Action(func(c *kingpin.ParseContext) error {
+			out, err := Config(DefaultClient()).List(OutputFormat)
+			if err != nil {
+				return err
+			}
+			fmt.Println(string(out))
+			return nil
+		})
+	AddOutputJsonYamlTextFlag(configListCmd)
+	app.DefineDebugUnixSocketFlag(configListCmd)
+
+	var paramName string
+	var paramValue string
+	var paramDuration time.Duration
+	configSetCmd := configCmd.Command("set", "Set runtime parameter.").
+		Action(func(c *kingpin.ParseContext) error {
+			out, err := Config(DefaultClient()).Set(paramName, paramValue, paramDuration)
+			if err != nil {
+				return err
+			}
+			fmt.Println(string(out))
+			return nil
+		})
+	configSetCmd.Arg("name", "A name of runtime parameter").Required().StringVar(&paramName)
+	configSetCmd.Arg("value", "A new value for the runtime parameter").Required().StringVar(&paramValue)
+	configSetCmd.Arg("duration", "Set value for a period of time, then return a previous value. Use Go notation: 10s, 15m30s, etc.").DurationVar(&paramDuration)
+	app.DefineDebugUnixSocketFlag(configSetCmd)
+
 	// Raw request command
 	var rawUrl string
 	rawCommand := app.CommandWithDefaultUsageTemplate(kpApp, "raw", "Make a raw request to debug endpoint.").
@@ -134,3 +167,29 @@ func (r *HookRequest) Snapshots(format string) ([]byte, error) {
 	url := fmt.Sprintf("http://unix/hook/%s/snapshots.%s", r.name, format)
 	return r.client.Get(url)
 }
+
+type ConfigRequest struct {
+	client *Client
+}
+
+func Config(client *Client) *ConfigRequest {
+	return &ConfigRequest{
+		client: client,
+	}
+}
+
+func (cr *ConfigRequest) List(format string) ([]byte, error) {
+	url := fmt.Sprintf("http://unix/config/list.%s", format)
+	return cr.client.Get(url)
+}
+
+func (cr *ConfigRequest) Set(name string, value string, duration time.Duration) ([]byte, error) {
+	data := map[string][]string{
+		"name":  {name},
+		"value": {value},
+	}
+	if duration != 0 {
+		data["duration"] = []string{duration.String()}
+	}
+	return cr.client.Post("http://unix/config/set", data)
+}
diff --git a/pkg/debug/server.go b/pkg/debug/server.go
index 317e204..9c3affd 100644
--- a/pkg/debug/server.go
+++ b/pkg/debug/server.go
@@ -88,30 +88,49 @@ func (s *Server) Route(pattern string, handler func(request *http.Request) (inte
 		return
 	}
 	s.Router.Get(pattern, func(writer http.ResponseWriter, request *http.Request) {
-		out, err := handler(request)
+		HandleFormattedOutput(writer, request, handler)
+	})
+}
 
+func (s *Server) RoutePOST(pattern string, handler func(request *http.Request) (interface{}, error)) {
+	// Should not happen.
+	if handler == nil {
+		return
+	}
+	s.Router.Post(pattern, func(writer http.ResponseWriter, request *http.Request) {
+		//
+		err := request.ParseForm()
 		if err != nil {
 			writer.WriteHeader(http.StatusInternalServerError)
 			_, _ = fmt.Fprintf(writer, "Error: %s", err)
 			return
 		}
 
-		format := chi.URLParam(request, "format")
-		if format == "" {
-			format = "text"
-		}
-		structured_logger.GetLogEntry(request).Debugf("use format '%s'", format)
+		HandleFormattedOutput(writer, request, handler)
+	})
+}
 
-		outBytes, err := TransformUsingFormat(out, format)
+func HandleFormattedOutput(writer http.ResponseWriter, request *http.Request, handler func(request *http.Request) (interface{}, error)) {
+	out, err := handler(request)
 
-		if err != nil {
-			writer.WriteHeader(http.StatusInternalServerError)
-			_, _ = fmt.Fprintf(writer, "Error '%s' transform: %s", format, err)
-			return
-		}
+	if err != nil {
+		writer.WriteHeader(http.StatusInternalServerError)
+		_, _ = fmt.Fprintf(writer, "Error: %s", err)
+		return
+	}
 
-		_, _ = writer.Write(outBytes)
-	})
+	format := FormatFromRequest(request)
+	structured_logger.GetLogEntry(request).Debugf("use format '%s'", format)
+
+	outBytes, err := TransformUsingFormat(out, format)
+
+	if err != nil {
+		writer.WriteHeader(http.StatusInternalServerError)
+		_, _ = fmt.Fprintf(writer, "Error '%s' transform: %s", format, err)
+		return
+	}
+
+	_, _ = writer.Write(outBytes)
 }
 
 func TransformUsingFormat(val interface{}, format string) ([]byte, error) {
@@ -142,3 +161,11 @@ func TransformUsingFormat(val interface{}, format string) ([]byte, error) {
 
 	return outBytes, err
 }
+
+func FormatFromRequest(request *http.Request) string {
+	format := chi.URLParam(request, "format")
+	if format == "" {
+		format = "text"
+	}
+	return format
+}
diff --git a/pkg/kube_events_manager/kube_events_manager.go b/pkg/kube_events_manager/kube_events_manager.go
index 6b11f44..45727aa 100644
--- a/pkg/kube_events_manager/kube_events_manager.go
+++ b/pkg/kube_events_manager/kube_events_manager.go
@@ -3,6 +3,7 @@ package kube_events_manager
 import (
 	"context"
 	"runtime/trace"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 
@@ -15,7 +16,7 @@ type KubeEventsManager interface {
 	WithContext(ctx context.Context)
 	WithMetricStorage(mstor *metric_storage.MetricStorage)
 	WithKubeClient(client klient.Client)
-
+	WithSyncPeriod(time.Duration)
 	AddMonitor(monitorConfig *MonitorConfig) error
 	HasMonitor(monitorID string) bool
 	GetMonitor(monitorID string) Monitor
@@ -31,7 +32,8 @@ type kubeEventsManager struct {
 	// Array of monitors
 	Monitors map[string]Monitor
 	// channel to emit KubeEvent objects
-	KubeEventCh chan KubeEvent
+	KubeEventCh      chan KubeEvent
+	informerSyncTime time.Duration
 
 	KubeClient klient.Client
 
@@ -46,8 +48,9 @@ var _ KubeEventsManager = &kubeEventsManager{}
 // NewKubeEventsManager returns an implementation of KubeEventsManager.
 var NewKubeEventsManager = func() *kubeEventsManager {
 	em := &kubeEventsManager{
-		Monitors:    make(map[string]Monitor),
-		KubeEventCh: make(chan KubeEvent, 1),
+		Monitors:         make(map[string]Monitor),
+		KubeEventCh:      make(chan KubeEvent, 1),
+		informerSyncTime: 100 * time.Millisecond,
 	}
 	return em
 }
@@ -64,6 +67,10 @@ func (mgr *kubeEventsManager) WithKubeClient(client klient.Client) {
 	mgr.KubeClient = client
 }
 
+func (mgr *kubeEventsManager) WithSyncPeriod(period time.Duration) {
+	mgr.informerSyncTime = period
+}
+
 // AddMonitor creates a monitor with informers and return a KubeEvent with existing objects.
 // TODO cleanup informers in case of error
 // TODO use Context to stop informers
@@ -74,6 +81,7 @@ func (mgr *kubeEventsManager) AddMonitor(monitorConfig *MonitorConfig) error {
 	monitor.WithKubeClient(mgr.KubeClient)
 	monitor.WithMetricStorage(mgr.metricStorage)
 	monitor.WithConfig(monitorConfig)
+	monitor.WithSyncPeriod(mgr.informerSyncTime)
 	monitor.WithKubeEventCb(func(ev KubeEvent) {
 		defer trace.StartRegion(context.Background(), "EmitKubeEvent").End()
 		mgr.KubeEventCh <- ev
diff --git a/pkg/kube_events_manager/monitor.go b/pkg/kube_events_manager/monitor.go
index b5f5d35..19f6ad7 100644
--- a/pkg/kube_events_manager/monitor.go
+++ b/pkg/kube_events_manager/monitor.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 
@@ -19,6 +20,7 @@ type Monitor interface {
 	WithMetricStorage(mstor *metric_storage.MetricStorage)
 	WithConfig(config *MonitorConfig)
 	WithKubeEventCb(eventCb func(KubeEvent))
+	WithSyncPeriod(time.Duration)
 	CreateInformers() error
 	Start(context.Context)
 	Stop()
@@ -41,6 +43,8 @@ type monitor struct {
 	// map of dynamically starting informers
 	VaryingInformers map[string][]ResourceInformer
 
+	informerSyncTime time.Duration
+
 	eventCb func(KubeEvent)
 	// Index of namespaces statically defined in monitor configuration
 	staticNamespaces map[string]bool
@@ -56,6 +60,7 @@ var NewMonitor = func() Monitor {
 	return &monitor{
 		ResourceInformers: make([]ResourceInformer, 0),
 		VaryingInformers:  make(map[string][]ResourceInformer),
+		informerSyncTime:  100 * time.Millisecond,
 		cancelForNs:       make(map[string]context.CancelFunc),
 		staticNamespaces:  make(map[string]bool),
 	}
@@ -85,6 +90,10 @@ func (m *monitor) WithKubeEventCb(eventCb func(KubeEvent)) {
 	m.eventCb = eventCb
 }
 
+func (m *monitor) WithSyncPeriod(period time.Duration) {
+	m.informerSyncTime = period
+}
+
 // CreateInformers creates all informers and
 // a namespace informer if namespace.labelSelector is defined.
 // If MonitorConfig.NamespaceSelector.MatchNames is defined, then
@@ -252,6 +261,7 @@ func (m *monitor) CreateInformersForNamespace(namespace string) (informers []Res
 		informer.WithNamespace(namespace)
 		informer.WithName(objName)
 		informer.WithKubeEventCb(m.eventCb)
+		informer.WithSyncPeriod(m.informerSyncTime)
 
 		err := informer.CreateSharedInformer()
 		if err != nil {
@@ -283,6 +293,7 @@ func (m *monitor) Start(parentCtx context.Context) {
 
 	if m.NamespaceInformer != nil {
 		m.NamespaceInformer.WithContext(m.ctx)
+		m.NamespaceInformer.WithSyncPeriod(m.informerSyncTime)
 		m.NamespaceInformer.Start()
 	}
 }
diff --git a/pkg/kube_events_manager/namespace_informer.go b/pkg/kube_events_manager/namespace_informer.go
index 90d2f2d..1b56be0 100644
--- a/pkg/kube_events_manager/namespace_informer.go
+++ b/pkg/kube_events_manager/namespace_informer.go
@@ -5,8 +5,10 @@ package kube_events_manager
 import (
 	"context"
 	"fmt"
+	"time"
 
 	log "github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -20,6 +22,7 @@ type NamespaceInformer interface {
 	WithContext(ctx context.Context)
 	WithKubeClient(client klient.Client)
 	CreateSharedInformer(addFn func(string), delFn func(string)) error
+	WithSyncPeriod(time.Duration)
 	GetExistedObjects() map[string]bool
 	Start()
 	Stop()
@@ -31,9 +34,10 @@ type namespaceInformer struct {
 	cancel  context.CancelFunc
 	stopped bool
 
-	KubeClient     klient.Client
-	Monitor        *MonitorConfig
-	SharedInformer cache.SharedInformer
+	KubeClient       klient.Client
+	Monitor          *MonitorConfig
+	SharedInformer   cache.SharedInformer
+	informerSyncTime time.Duration
 
 	ExistedObjects map[string]bool
 
@@ -60,6 +64,10 @@ func (ni *namespaceInformer) WithKubeClient(client klient.Client) {
 	ni.KubeClient = client
 }
 
+func (ni *namespaceInformer) WithSyncPeriod(period time.Duration) {
+	ni.informerSyncTime = period
+}
+
 func (ni *namespaceInformer) CreateSharedInformer(addFn func(string), delFn func(string)) error {
 	// define resyncPeriod for informer
 	resyncPeriod := RandomizedResyncPeriod()
@@ -144,7 +152,9 @@ func (ni *namespaceInformer) Start() {
 	}()
 
 	go ni.SharedInformer.Run(stopCh)
-	if ok := cache.WaitForCacheSync(stopCh, ni.SharedInformer.HasSynced); !ok {
+	if err := wait.PollImmediateUntil(ni.informerSyncTime, func() (bool, error) {
+		return ni.SharedInformer.HasSynced(), nil
+	}, stopCh); err != nil {
 		ni.Monitor.LogEntry.Errorf("%s: cache is not synced for informer", ni.Monitor.Metadata.DebugName)
 	}
 
diff --git a/pkg/kube_events_manager/resource_informer.go b/pkg/kube_events_manager/resource_informer.go
index fb13e99..2621cf0 100644
--- a/pkg/kube_events_manager/resource_informer.go
+++ b/pkg/kube_events_manager/resource_informer.go
@@ -11,6 +11,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/tools/cache"
 
@@ -28,6 +29,7 @@ type ResourceInformer interface {
 	WithNamespace(string)
 	WithName(string)
 	WithKubeEventCb(eventCb func(KubeEvent))
+	WithSyncPeriod(time.Duration)
 	CreateSharedInformer() error
 	CachedObjects() []ObjectAndFilterResult
 	CachedObjectsBytes() int64
@@ -51,6 +53,7 @@ type resourceInformer struct {
 	SharedInformer       cache.SharedInformer
 	GroupVersionResource schema.GroupVersionResource
 	ListOptions          metav1.ListOptions
+	informerSyncTime     time.Duration
 
 	// A cache of objects and filterResults. It is a part of the Monitor's snapshot.
 	cachedObjects map[string]*ObjectAndFilterResult
@@ -91,6 +94,7 @@ var NewResourceInformer = func(monitor *MonitorConfig) ResourceInformer {
 		eventBufLock:           sync.Mutex{},
 		cachedObjectsInfo:      &CachedObjectsInfo{},
 		cachedObjectsIncrement: &CachedObjectsInfo{},
+		informerSyncTime:       100 * time.Millisecond,
 	}
 	return informer
 }
@@ -115,6 +119,10 @@ func (ei *resourceInformer) WithName(name string) {
 	ei.Name = name
 }
 
+func (ei *resourceInformer) WithSyncPeriod(period time.Duration) {
+	ei.informerSyncTime = period
+}
+
 func (ei *resourceInformer) WithKubeEventCb(eventCb func(KubeEvent)) {
 	ei.eventCb = eventCb
 }
@@ -480,7 +488,9 @@ func (ei *resourceInformer) Start() {
 
 	go ei.SharedInformer.Run(stopCh)
 
-	if ok := cache.WaitForCacheSync(stopCh, ei.SharedInformer.HasSynced); !ok {
+	if err := wait.PollImmediateUntil(ei.informerSyncTime, func() (bool, error) {
+		return ei.SharedInformer.HasSynced(), nil
+	}, stopCh); err != nil {
 		ei.Monitor.LogEntry.Errorf("%s: cache is not synced for informer", ei.Monitor.Metadata.DebugName)
 	}
 
diff --git a/pkg/shell-operator/operator.go b/pkg/shell-operator/operator.go
index 262467d..55fa560 100644
--- a/pkg/shell-operator/operator.go
+++ b/pkg/shell-operator/operator.go
@@ -9,23 +9,22 @@ import (
 	"path/filepath"
 	"time"
 
+	klient "github.com/flant/kube-client/client"
 	"github.com/go-chi/chi"
 	log "github.com/sirupsen/logrus"
 	uuid "gopkg.in/satori/go.uuid.v1"
 
-	. "github.com/flant/shell-operator/pkg/hook/binding_context"
-	. "github.com/flant/shell-operator/pkg/hook/task_metadata"
-	. "github.com/flant/shell-operator/pkg/hook/types"
-	"github.com/flant/shell-operator/pkg/kube/object_patch"
-	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
-	. "github.com/flant/shell-operator/pkg/webhook/validating/types"
-
-	klient "github.com/flant/kube-client/client"
 	"github.com/flant/shell-operator/pkg/app"
+	"github.com/flant/shell-operator/pkg/config"
 	"github.com/flant/shell-operator/pkg/debug"
 	"github.com/flant/shell-operator/pkg/hook"
+	. "github.com/flant/shell-operator/pkg/hook/binding_context"
 	"github.com/flant/shell-operator/pkg/hook/controller"
+	. "github.com/flant/shell-operator/pkg/hook/task_metadata"
+	. "github.com/flant/shell-operator/pkg/hook/types"
+	"github.com/flant/shell-operator/pkg/kube/object_patch"
 	"github.com/flant/shell-operator/pkg/kube_events_manager"
+	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
 	"github.com/flant/shell-operator/pkg/metric_storage"
 	"github.com/flant/shell-operator/pkg/schedule_manager"
 	"github.com/flant/shell-operator/pkg/task"
@@ -36,6 +35,7 @@ import (
 	"github.com/flant/shell-operator/pkg/utils/measure"
 	"github.com/flant/shell-operator/pkg/webhook/conversion"
 	"github.com/flant/shell-operator/pkg/webhook/validating"
+	. "github.com/flant/shell-operator/pkg/webhook/validating/types"
 )
 
 var WaitQueuesTimeout = time.Second * 10
@@ -70,6 +70,8 @@ type ShellOperator struct {
 	ConversionWebhookManager *conversion.WebhookManager
 
 	DebugServer *debug.Server
+
+	RuntimeConfig *config.Config
 }
 
 func NewShellOperator() *ShellOperator {
@@ -103,6 +105,10 @@ func (op *ShellOperator) WithMetricStorage(metricStorage *metric_storage.MetricS
 	op.MetricStorage = metricStorage
 }
 
+func (op *ShellOperator) WithRuntimeConfig(runtimeConfig *config.Config) {
+	op.RuntimeConfig = runtimeConfig
+}
+
 // InitMetricStorage creates default MetricStorage object if not set earlier.
 func (op *ShellOperator) InitMetricStorage() {
 	if op.MetricStorage != nil {
@@ -141,6 +147,7 @@ func (op *ShellOperator) InitHookMetricStorage() {
 //   - hook manager
 //   - kubernetes events manager
 //   - schedule manager
+//   - runtime config
 func (op *ShellOperator) Init() (err error) {
 	log.Debug("MAIN Init")
 
@@ -1037,6 +1044,45 @@ func (op *ShellOperator) SetupDebugServerHandles() {
 		h := op.HookManager.GetHook(hookName)
 		return h.HookController.SnapshotsDump(), nil
 	})
+
+	op.DebugServer.Route("/config/list.{format:(json|yaml|text)}", func(r *http.Request) (interface{}, error) {
+		format := debug.FormatFromRequest(r)
+		if format == "text" {
+			return op.RuntimeConfig.String(), nil
+		}
+		return op.RuntimeConfig.List(), nil
+	})
+
+	op.DebugServer.RoutePOST("/config/set", func(r *http.Request) (interface{}, error) {
+		name := r.PostForm.Get("name")
+		if name == "" {
+			return nil, fmt.Errorf("'name' parameter is required")
+		}
+		if !op.RuntimeConfig.Has(name) {
+			return nil, fmt.Errorf("unknown runtime parameter '%s'", name)
+		}
+
+		value := r.PostForm.Get("value")
+		if name == "" {
+			return nil, fmt.Errorf("'value' parameter is required")
+		}
+
+		var duration time.Duration
+		var err error
+		durationStr := r.PostForm.Get("duration")
+		if durationStr != "" {
+			duration, err = time.ParseDuration(durationStr)
+			if err != nil {
+				return nil, fmt.Errorf("parse duration: %v", err)
+			}
+		}
+		if duration == 0 {
+			op.RuntimeConfig.Set(name, value)
+		} else {
+			op.RuntimeConfig.SetTemporarily(name, value, duration)
+		}
+		return nil, op.RuntimeConfig.LastError(name)
+	})
 }
 
 func (op *ShellOperator) SetupHttpServerHandles() {


Test patch:
diff --git a/.github/workflows/tests-labeled.yaml b/.github/workflows/tests-labeled.yaml
index b24e972..a96aa79 100644
--- a/.github/workflows/tests-labeled.yaml
+++ b/.github/workflows/tests-labeled.yaml
@@ -12,7 +12,7 @@ jobs:
       run_tests: ${{ steps.check.outputs.run_tests }}
       build_multi_arch: ${{ steps.check.outputs.build_multi_arch }}
     steps:
-      - uses: actions/github-script@v5
+      - uses: actions/github-script@v6
         id: check
         with:
           script: |
diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go
new file mode 100644
index 0000000..32904a2
--- /dev/null
+++ b/pkg/config/config_test.go
@@ -0,0 +1,86 @@
+package config
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConfig_Register(t *testing.T) {
+	c := NewConfig()
+
+	c.Register("log.level", "", "info", nil, nil)
+
+	logLevel := c.Value("log.level")
+	assert.Equal(t, "info", logLevel)
+
+	c.Set("log.level", "debug")
+
+	logLevel = c.Value("log.level")
+	assert.Equal(t, "debug", logLevel)
+
+	c.Unset("log.level")
+
+	logLevel = c.Value("log.level")
+	assert.Equal(t, "info", logLevel)
+
+	params := c.List()
+
+	assert.Len(t, params, 1)
+}
+
+func TestConfig_OnChange(t *testing.T) {
+	c := NewConfig()
+
+	newValue := ""
+	c.Register("log.level", "", "info", func(oldValue string, n string) error {
+		newValue = n
+		return nil
+	}, nil)
+
+	c.Set("log.level", "debug")
+	assert.Equal(t, "debug", newValue, "onChange not called for Set")
+
+	c.Unset("log.level")
+	assert.Equal(t, "info", newValue, "onChange not called for Unset after Set")
+
+	// Temporal value
+	c.SetTemporarily("log.level", "debug", time.Duration(10*time.Second))
+	assert.Equal(t, "debug", newValue, "onChange not called for SetTemporarily")
+
+	c.Unset("log.level")
+	assert.Equal(t, "info", newValue, "onChange not called for Unset after SetTemporarily")
+
+	// Set after SetTemporarily
+	c.SetTemporarily("log.level", "debug", time.Duration(10*time.Second))
+	assert.Equal(t, "debug", newValue, "onChange not called for SetTemporarily")
+
+	c.Set("log.level", "error")
+	assert.Equal(t, "error", newValue, "onChange not called for Set after SetTemporarily")
+
+	c.Unset("log.level")
+	assert.Equal(t, "info", newValue, "onChange not called for Unset after SetTemporarily+Set")
+
+}
+
+func TestConfig_Errors(t *testing.T) {
+	var err error
+	c := NewConfig()
+
+	c.Register("log.level", "", "info", func(oldValue string, n string) error {
+		if n == "debug" {
+			return nil
+		}
+		return fmt.Errorf("unknown value")
+	}, nil)
+
+	c.Set("log.level", "bad-value")
+	err = c.LastError("log.level")
+	assert.Error(t, err, "Set should save error about unknown value")
+
+	c.Set("log.level", "debug")
+	err = c.LastError("log.level")
+	assert.NoError(t, err, "Set should clean error after success in onChange handler")
+}
diff --git a/test/hook/context/README.md b/test/hook/context/README.md
index e4831bf..dd08a65 100644
--- a/test/hook/context/README.md
+++ b/test/hook/context/README.md
@@ -1,9 +1,9 @@
 Binding Context Generator
 =========================
-Binding Context Generator allows to generate binding context values for hook testing purpose.
+Binding Context Generator provides the ability to generate binding contexts for hooks testing purposes.
 
 Usage example:
-1. Declare the hook config
+1. Hook Config
 ```go
 config := `
 configVersion: v1
@@ -25,7 +25,7 @@ kubernetes:
       matchNames:
       - default`
 ```
-2. Declare initial state of kubernetes objects
+2. Initial objects state
 ```go
 initialState := `
 ---
@@ -40,7 +40,7 @@ metadata:
   name: pod-1
 `
 ```
-3. Declare new state (one of pods are deleted)
+3. New state
 ```go
 newState := `
 ---
@@ -52,12 +52,9 @@ metadata:
 ```
 4. Create new binding context controller
 ```go
-c, err := context.NewBindingContextController(config, initialState)
-if err != nil {
-  return err
-}
+c := context.NewBindingContextController(config, initialState)
 ```
-5. Register CRD (if you use some in tests) in format [Group, Version, Kind, isNamespaced]:
+5. Register CRDs (if it is necessary) in format [Group, Version, Kind, isNamespaced]:
 ```go
 c.RegisterCRD("example.com", "v1", "Example", true)
 c.RegisterCRD("example.com", "v1", "ClusterExample", false)
@@ -70,7 +67,7 @@ if err != nil {
 }
 testContexts(contexts)
 ```
-7. Change state to new get binding contexts
+7. Change state to get new binding contexts
 ```go
 contexts, err = c.ChangeState(newState)
 if err != nil {
@@ -78,11 +75,11 @@ if err != nil {
 }
 testNewContexts(contexts)
 ```
-8. Run schedule to get binding contexts
+8. Run schedule to get binding contexts for a cron task
 ```go
 contexts, err = c.RunSchedule("* * * * *")
 if err != nil {
   return err
 }
 testScheduleContexts(contexts)
-```
\ No newline at end of file
+```
diff --git a/test/hook/context/generator.go b/test/hook/context/generator.go
index 3a43375..27a1e8b 100644
--- a/test/hook/context/generator.go
+++ b/test/hook/context/generator.go
@@ -3,6 +3,7 @@ package context
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/flant/kube-client/fake"
@@ -31,28 +32,40 @@ type BindingContextController struct {
 	KubeEventsManager kubeeventsmanager.KubeEventsManager
 	ScheduleManager   schedulemanager.ScheduleManager
 
-	UpdateTimeout time.Duration
-
 	fakeCluster *fake.Cluster
+
+	mu      sync.Mutex
+	started bool
 }
 
-func NewBindingContextController(config string, version ...fake.ClusterVersion) (*BindingContextController, error) {
+func NewBindingContextController(config string, version ...fake.ClusterVersion) *BindingContextController {
 	logrus.SetLevel(logrus.ErrorLevel)
 
 	k8sVersion := fake.ClusterVersionV119
 	if len(version) > 0 {
 		k8sVersion = version[0]
 	}
-	fc := fake.NewFakeCluster(k8sVersion)
-	return &BindingContextController{
-		HookMap:    make(map[string]string),
-		HookConfig: config,
 
-		Controller:    NewStateController(fc),
-		UpdateTimeout: 1500 * time.Millisecond,
+	fc := fake.NewFakeCluster(k8sVersion)
+	ctx := context.Background()
 
+	b := &BindingContextController{
+		HookMap:     make(map[string]string),
+		HookConfig:  config,
 		fakeCluster: fc,
-	}, nil
+	}
+
+	b.KubeEventsManager = kubeeventsmanager.NewKubeEventsManager()
+	b.KubeEventsManager.WithContext(ctx)
+	b.KubeEventsManager.WithKubeClient(b.fakeCluster.Client)
+	b.KubeEventsManager.WithSyncPeriod(time.Microsecond)
+
+	b.ScheduleManager = schedulemanager.NewScheduleManager()
+	b.ScheduleManager.WithContext(ctx)
+
+	b.Controller = NewStateController(fc, b.KubeEventsManager)
+
+	return b
 }
 
 func (b *BindingContextController) WithHook(h *hook.Hook) {
@@ -69,18 +82,18 @@ func (b *BindingContextController) RegisterCRD(group, version, kind string, name
 	b.fakeCluster.RegisterCRD(group, version, kind, namespaced)
 }
 
-// BindingContextsGenerator generates binding contexts for hook tests
+// Run generates binding contexts for hook tests
 func (b *BindingContextController) Run(initialState string) (GeneratedBindingContexts, error) {
-	ctx := context.Background()
+	// fmt.Println("Run start")
+	// defer func() { fmt.Println("Run end") }()
 
-	b.KubeEventsManager = kubeeventsmanager.NewKubeEventsManager()
-	b.KubeEventsManager.WithContext(ctx)
-	b.KubeEventsManager.WithKubeClient(b.fakeCluster.Client)
+	b.mu.Lock()
+	defer b.mu.Unlock()
 
-	b.ScheduleManager = schedulemanager.NewScheduleManager()
-	b.ScheduleManager.WithContext(ctx)
+	if b.started {
+		return GeneratedBindingContexts{}, fmt.Errorf("attempt to runner started runner, it cannot be started twice")
+	}
 
-	// Use StateController to apply changes
 	err := b.Controller.SetInitialState(initialState)
 	if err != nil {
 		return GeneratedBindingContexts{}, err
@@ -113,65 +126,50 @@ func (b *BindingContextController) Run(initialState string) (GeneratedBindingCon
 	}
 
 	b.HookCtrl.UnlockKubernetesEvents()
+	b.started = true
 
 	return cc.CombinedAndUpdated(b.HookCtrl)
 }
 
-func (b *BindingContextController) ChangeState(newState ...string) (GeneratedBindingContexts, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), b.UpdateTimeout)
-	defer cancel()
-
-	cc := NewContextCombiner()
+func (b *BindingContextController) ChangeState(newState string) (GeneratedBindingContexts, error) {
+	// fmt.Println("Change state start")
+	// defer func() { fmt.Println("Change state end") }()
 
-	for _, state := range newState {
-		_, err := b.Controller.ChangeState(state)
-		if err != nil {
-			return GeneratedBindingContexts{}, fmt.Errorf("error while changing BindingContextGenerator state: %v", err)
-		}
+	b.mu.Lock()
+	defer b.mu.Unlock()
 
-	outer:
-		for {
-			select {
-			case ev := <-b.KubeEventsManager.Ch():
-				b.HookCtrl.HandleKubeEvent(ev, func(info controller.BindingExecutionInfo) {
-					cc.AddBindingContext(types.OnKubernetesEvent, info)
-				})
-				continue
-			case <-ctx.Done():
-				break outer
-			}
-		}
-	}
-	return cc.CombinedAndUpdated(b.HookCtrl)
-}
-
-func (b *BindingContextController) ChangeStateAndWaitForBindingContexts(desiredQuantity int, newState string) (GeneratedBindingContexts, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
 
 	cc := NewContextCombiner()
 
-	_, err := b.Controller.ChangeState(newState)
-	if err != nil {
+	if err := b.Controller.ChangeState(newState); err != nil {
 		return GeneratedBindingContexts{}, fmt.Errorf("error while changing BindingContextGenerator state: %v", err)
 	}
 
-	for cc.QueueLen() != desiredQuantity {
+outer:
+	for {
 		select {
 		case ev := <-b.KubeEventsManager.Ch():
-			b.HookCtrl.HandleKubeEvent(ev, func(info controller.BindingExecutionInfo) {
-				cc.AddBindingContext(types.OnKubernetesEvent, info)
-			})
-			continue
+			switch ev.MonitorId {
+			case "STOP_EVENTS":
+				break outer
+			default:
+				b.HookCtrl.HandleKubeEvent(ev, func(info controller.BindingExecutionInfo) {
+					cc.AddBindingContext(types.OnKubernetesEvent, info)
+				})
+			}
 		case <-ctx.Done():
 			return GeneratedBindingContexts{}, fmt.Errorf("timeout occurred while waiting for binding contexts")
 		}
 	}
-
 	return cc.CombinedAndUpdated(b.HookCtrl)
 }
 
 func (b *BindingContextController) RunSchedule(crontab string) (GeneratedBindingContexts, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	cc := NewContextCombiner()
 	b.HookCtrl.HandleScheduleEvent(crontab, func(info controller.BindingExecutionInfo) {
 		cc.AddBindingContext(types.Schedule, info)
@@ -180,6 +178,9 @@ func (b *BindingContextController) RunSchedule(crontab string) (GeneratedBinding
 }
 
 func (b *BindingContextController) RunBindingWithAllSnapshots(binding types.BindingType) (GeneratedBindingContexts, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	bc := BindingContext{
 		Binding:   string(binding),
 		Snapshots: b.HookCtrl.KubernetesSnapshots(),
diff --git a/test/hook/context/generator_test.go b/test/hook/context/generator_test.go
index b57ea86..c3d53e0 100644
--- a/test/hook/context/generator_test.go
+++ b/test/hook/context/generator_test.go
@@ -18,7 +18,7 @@ func parseContexts(contexts string) []BindingContext {
 func Test_BindingContextGenerator(t *testing.T) {
 	g := NewWithT(t)
 
-	c, err := NewBindingContextController(`
+	c := NewBindingContextController(`
 configVersion: v1
 kubernetes:
 - apiVersion: v1
@@ -32,7 +32,6 @@ schedule:
   includeSnapshotsFrom:
   - selected_pods
 `)
-	g.Expect(err).ShouldNot(HaveOccurred())
 
 	// Synchronization contexts
 	contexts, err := c.Run(`
@@ -56,7 +55,7 @@ metadata:
 	g.Expect(parsedBindingContexts[0].Snapshots["selected_pods"]).To(HaveLen(2))
 
 	// Object added
-	contexts, err = c.ChangeStateAndWaitForBindingContexts(1, `
+	contexts, err = c.ChangeState(`
 ---
 apiVersion: v1
 kind: Pod
@@ -82,7 +81,7 @@ spec:
 	g.Expect(parsedBindingContexts[0].Snapshots["selected_pods"]).To(HaveLen(3))
 
 	// Object modified
-	contexts, err = c.ChangeStateAndWaitForBindingContexts(1, `
+	contexts, err = c.ChangeState(`
 ---
 apiVersion: v1
 kind: Pod
@@ -109,7 +108,7 @@ spec:
 	g.Expect(parsedBindingContexts[0].Snapshots["selected_pods"]).To(HaveLen(3))
 
 	// Object deleted
-	contexts, err = c.ChangeStateAndWaitForBindingContexts(1, `
+	contexts, err = c.ChangeState(`
 ---
 apiVersion: v1
 kind: Pod
@@ -137,7 +136,7 @@ metadata:
 func Test_RegisterCRD(t *testing.T) {
 	g := NewWithT(t)
 
-	c, err := NewBindingContextController(`configVersion: v1
+	c := NewBindingContextController(`configVersion: v1
 kubernetes:
 - apiVersion: my.crd.io/v1alpha1
   includeSnapshotsFrom:
@@ -145,8 +144,6 @@ kubernetes:
   kind: MyResource
   name: selected_crds
 `)
-	g.Expect(err).ShouldNot(HaveOccurred())
-
 	c.RegisterCRD("my.crd.io", "v1alpha1", "MyResource", true)
 
 	gvr, err := c.fakeCluster.FindGVR("my.crd.io/v1alpha1", "MyResource")
@@ -162,7 +159,7 @@ kubernetes:
 	g.Expect(bindingContexts.Rendered).To(ContainSubstring("Synchronization"))
 
 	// Event phase
-	bindingContexts, err = c.ChangeStateAndWaitForBindingContexts(2, `
+	bindingContexts, err = c.ChangeState(`
 apiVersion: my.crd.io/v1alpha1
 kind: MyResource
 metadata:
@@ -185,7 +182,7 @@ spec:
 func Test_PreferredGVR(t *testing.T) {
 	g := NewWithT(t)
 
-	c, err := NewBindingContextController(`configVersion: v1
+	c := NewBindingContextController(`configVersion: v1
 kubernetes:
 - apiVersion: apps/v1
   includeSnapshotsFrom:
@@ -193,8 +190,6 @@ kubernetes:
   kind: Deployment
   name: deployment
 `)
-	g.Expect(err).ShouldNot(HaveOccurred())
-
 	contexts, err := c.Run(`
 ---
 apiVersion: apps/v1
@@ -213,7 +208,7 @@ metadata:
 func Test_Synchronization(t *testing.T) {
 	g := NewWithT(t)
 
-	c, err := NewBindingContextController(`
+	c := NewBindingContextController(`
 configVersion: v1
 kubernetes:
 - apiVersion: apps/v1
@@ -244,8 +239,6 @@ kubernetes:
   name: pods-grouped
   group: group1
 `)
-	g.Expect(err).ShouldNot(HaveOccurred())
-
 	contexts, err := c.Run(`
 ---
 apiVersion: apps/v1
@@ -278,7 +271,7 @@ metadata:
 func Test_Groups(t *testing.T) {
 	g := NewWithT(t)
 
-	c, err := NewBindingContextController(`configVersion: v1
+	c := NewBindingContextController(`configVersion: v1
 kubernetes:
 - apiVersion: apps/v1
   group: main
@@ -289,8 +282,6 @@ kubernetes:
   kind: Secret
   name: secret
 `)
-	g.Expect(err).ShouldNot(HaveOccurred())
-
 	contexts, err := c.Run(`
 ---
 apiVersion: apps/v1
@@ -316,7 +307,7 @@ metadata:
 func Test_ExecuteOnSynchronization_false(t *testing.T) {
 	g := NewWithT(t)
 
-	c, err := NewBindingContextController(`
+	c := NewBindingContextController(`
 configVersion: v1
 kubernetes:
 - apiVersion: v1
@@ -331,8 +322,6 @@ kubernetes:
   name: selected_pods_nosync
   executeHookOnSynchronization: false
 `)
-	g.Expect(err).ShouldNot(HaveOccurred())
-
 	// Synchronization contexts
 	contexts, err := c.Run(`
 ---
@@ -361,7 +350,7 @@ metadata:
 func Test_RunSchedule(t *testing.T) {
 	g := NewWithT(t)
 
-	c, err := NewBindingContextController(`
+	c := NewBindingContextController(`
 configVersion: v1
 kubernetes:
 - apiVersion: v1
@@ -375,8 +364,6 @@ schedule:
   includeSnapshotsFrom:
   - selected_pods
 `)
-	g.Expect(err).ShouldNot(HaveOccurred())
-
 	// Synchronization contexts
 	contexts, err := c.Run(`
 ---
@@ -399,7 +386,7 @@ metadata:
 	g.Expect(parsedBindingContexts[0].Snapshots["selected_pods"]).To(HaveLen(2))
 
 	// Object added
-	contexts, err = c.ChangeStateAndWaitForBindingContexts(1, `
+	contexts, err = c.ChangeState(`
 ---
 apiVersion: v1
 kind: Pod
@@ -425,7 +412,7 @@ spec:
 	g.Expect(parsedBindingContexts[0].Snapshots["selected_pods"]).To(HaveLen(3))
 
 	// Object modified
-	contexts, err = c.ChangeStateAndWaitForBindingContexts(1, `
+	contexts, err = c.ChangeState(`
 ---
 apiVersion: v1
 kind: Pod
@@ -452,7 +439,7 @@ spec:
 	g.Expect(parsedBindingContexts[0].Snapshots["selected_pods"]).To(HaveLen(3))
 
 	// Object deleted
-	contexts, err = c.ChangeStateAndWaitForBindingContexts(1, `
+	contexts, err = c.ChangeState(`
 ---
 apiVersion: v1
 kind: Pod
diff --git a/test/hook/context/state.go b/test/hook/context/state.go
index dbce73b..61b7c70 100644
--- a/test/hook/context/state.go
+++ b/test/hook/context/state.go
@@ -3,28 +3,35 @@ package context
 import (
 	"fmt"
 	"reflect"
+	"time"
 
 	"github.com/flant/kube-client/fake"
 	"github.com/flant/kube-client/manifest"
+	kubeeventsmanager "github.com/flant/shell-operator/pkg/kube_events_manager"
+	"github.com/flant/shell-operator/pkg/kube_events_manager/types"
 )
 
 // if we use default, then we are not able to emulate global resources due to fake cluster limitations
 // that's why an empty namespace is used here, specify namespace in a resource by your own if needed
-var defaultNamespace = ""
+var defaultNamespace = "" //nolint:gochecknoglobals
 
 // StateController holds objects state for FakeCluster
 type StateController struct {
 	CurrentState map[string]manifest.Manifest
+	fakeCluster  *fake.Cluster
 
-	fakeCluster *fake.Cluster
+	stopCh chan<- types.KubeEvent
 }
 
 // NewStateController creates controller to apply state changes
-func NewStateController(fc *fake.Cluster) *StateController {
-	return &StateController{
+func NewStateController(fc *fake.Cluster, ev kubeeventsmanager.KubeEventsManager) *StateController {
+	c := &StateController{
 		CurrentState: make(map[string]manifest.Manifest),
 		fakeCluster:  fc,
+		stopCh:       ev.Ch(),
 	}
+
+	return c
 }
 
 func (c *StateController) SetInitialState(initialState string) error {
@@ -44,20 +51,18 @@ func (c *StateController) SetInitialState(initialState string) error {
 	}
 
 	c.CurrentState = newState
-
 	return nil
 }
 
 // ChangeState apply changes to current objects state
-func (c *StateController) ChangeState(newRawState string) (int, error) {
+func (c *StateController) ChangeState(newRawState string) error {
 	newManifests, err := manifest.ListFromYamlDocs(newRawState)
 	if err != nil {
-		return 0, fmt.Errorf("error while changing state: %v", err)
+		return fmt.Errorf("error while changing state: %v", err)
 	}
 
 	var newState = make(map[string]manifest.Manifest)
 
-	generatedEvents := 0
 	// Create new objects in FakeCluster
 	for _, m := range newManifests {
 		// save manifest to new State
@@ -66,36 +71,33 @@ func (c *StateController) ChangeState(newRawState string) (int, error) {
 		currM, ok := c.CurrentState[m.Id()]
 		if !ok {
 			// Create object if not exist
-			err = c.fakeCluster.Create(m.Namespace(defaultNamespace), m)
-			//err := createObject(newStateObject, newUnstructuredObject)
-			if err != nil {
-				return generatedEvents, err
+			if err := c.fakeCluster.Create(m.Namespace(defaultNamespace), m); err != nil {
+				return err
 			}
 			c.CurrentState[m.Id()] = m
-			generatedEvents++
-		} else {
-			// Update object if changed
-			if !reflect.DeepEqual(currM, m) {
-				err := c.fakeCluster.Update(m.Namespace(defaultNamespace), m)
-				if err != nil {
-					return generatedEvents, err
-				}
-				c.CurrentState[m.Id()] = m
-				generatedEvents++
+		} else if !reflect.DeepEqual(currM, m) {
+			// Update object
+			if err := c.fakeCluster.Update(m.Namespace(defaultNamespace), m); err != nil {
+				return err
 			}
+			c.CurrentState[m.Id()] = m
 		}
 	}
 	// Remove obsolete objects
-	for currId, currM := range c.CurrentState {
-		if _, ok := newState[currId]; !ok {
+	for currID, currM := range c.CurrentState {
+		if _, ok := newState[currID]; !ok {
 			// Delete object
-			err := c.fakeCluster.Delete(currM.Namespace(defaultNamespace), currM)
-			if err != nil {
-				return generatedEvents, err
+			if err := c.fakeCluster.Delete(currM.Namespace(defaultNamespace), currM); err != nil {
+				return err
 			}
-			delete(c.CurrentState, currId)
-			generatedEvents++
+			delete(c.CurrentState, currID)
 		}
 	}
-	return generatedEvents, nil
+
+	go func() {
+		time.Sleep(20 * time.Millisecond)
+		c.stopCh <- types.KubeEvent{MonitorId: "STOP_EVENTS"}
+	}()
+
+	return nil
 }
diff --git a/test/integration/kube_event_manager/kube_event_manager_test.go b/test/integration/kube_event_manager/kube_event_manager_test.go
index 06db1fd..b76030b 100644
--- a/test/integration/kube_event_manager/kube_event_manager_test.go
+++ b/test/integration/kube_event_manager/kube_event_manager_test.go
@@ -68,7 +68,7 @@ var _ = Describe("Binding 'kubernetes' with kind 'Pod' should emit KubeEvent obj
 
 		When("Pod is Added", func() {
 			JustBeforeEach(func() {
-				app.SetupLogging()
+				app.SetupLogging(nil)
 
 				// Unlock KubeEvent emitting.
 				m := KubeEventsManager.GetMonitor(monitorConfig.Metadata.MonitorId)