Repository: flant/shell-operator
Pull request #176 (2020-06-02 17:43:46)

Cumulative update with observability and metric cleaner
- feat: observability metrics for queue actions, task executions and Kubernetes events handling.
- feat: cleaner for metrics returned from hooks.
- fix: wait for queues on exit by SIGTERM/SIGINT signals.


Patch:
diff --git a/cmd/shell-operator/main.go b/cmd/shell-operator/main.go
index 8033e38..558d7e7 100644
--- a/cmd/shell-operator/main.go
+++ b/cmd/shell-operator/main.go
@@ -40,7 +40,8 @@ func main() {
 
 			// Block action by waiting signals from OS.
 			utils_signal.WaitForProcessInterruption(func() {
-				defaultOperator.Stop()
+				defaultOperator.Shutdown()
+				os.Exit(1)
 			})
 
 			return nil
diff --git a/pkg/hook/hook.go b/pkg/hook/hook.go
index f22eea7..a7fe4ae 100644
--- a/pkg/hook/hook.go
+++ b/pkg/hook/hook.go
@@ -17,7 +17,7 @@ import (
 	"github.com/flant/shell-operator/pkg/app"
 	"github.com/flant/shell-operator/pkg/executor"
 	"github.com/flant/shell-operator/pkg/hook/controller"
-	"github.com/flant/shell-operator/pkg/metrics_storage"
+	"github.com/flant/shell-operator/pkg/metrics_storage/operation"
 )
 
 type CommonHook interface {
@@ -67,7 +67,7 @@ func (h *Hook) GetHookController() controller.HookController {
 	return h.HookController
 }
 
-func (h *Hook) Run(bindingType BindingType, context []BindingContext, logLabels map[string]string) ([]metrics_storage.MetricOperation, error) {
+func (h *Hook) Run(bindingType BindingType, context []BindingContext, logLabels map[string]string) ([]operation.MetricOperation, error) {
 	// Refresh snapshots
 	freshBindingContext := h.HookController.UpdateSnapshots(context)
 
@@ -105,7 +105,7 @@ func (h *Hook) Run(bindingType BindingType, context []BindingContext, logLabels
 		return nil, fmt.Errorf("%s FAILED: %s", h.Name, err)
 	}
 
-	metrics, err := metrics_storage.MetricOperationsFromFile(metricsPath)
+	metrics, err := operation.MetricOperationsFromFile(metricsPath)
 	if err != nil {
 		return nil, fmt.Errorf("got bad metrics: %s", err)
 	}
diff --git a/pkg/hook/hook_config.go b/pkg/hook/hook_config.go
index ab4ce88..2becae8 100644
--- a/pkg/hook/hook_config.go
+++ b/pkg/hook/hook_config.go
@@ -202,6 +202,7 @@ func (c *HookConfig) ConvertAndCheckV0() (err error) {
 		monitor.Metadata.DebugName = c.MonitorDebugName(kubeCfg.Name, i)
 		monitor.Metadata.MonitorId = c.MonitorConfigId()
 		monitor.Metadata.LogLabels = map[string]string{}
+		monitor.Metadata.MetricLabels = map[string]string{}
 		monitor.WithMode(ModeV0)
 
 		// convert event names from legacy config.
@@ -270,6 +271,7 @@ func (c *HookConfig) ConvertAndCheckV1() (err error) {
 		monitor.Metadata.DebugName = c.MonitorDebugName(kubeCfg.Name, i)
 		monitor.Metadata.MonitorId = c.MonitorConfigId()
 		monitor.Metadata.LogLabels = map[string]string{}
+		monitor.Metadata.MetricLabels = map[string]string{}
 		monitor.WithMode(kubeCfg.Mode)
 		monitor.ApiVersion = kubeCfg.ApiVersion
 		monitor.Kind = kubeCfg.Kind
diff --git a/pkg/hook/hook_manager.go b/pkg/hook/hook_manager.go
index f013f21..df2db1a 100644
--- a/pkg/hook/hook_manager.go
+++ b/pkg/hook/hook_manager.go
@@ -140,9 +140,13 @@ func (hm *hookManager) loadHook(hookPath string) (hook *Hook, err error) {
 		return nil, fmt.Errorf("creating hook '%s': %s", hookName, err.Error())
 	}
 
-	// Add hook info as log labels
+	// Add hook info as log labels, update MetricLabels
 	for _, kubeCfg := range hook.GetConfig().OnKubernetesEvents {
 		kubeCfg.Monitor.Metadata.LogLabels["hook"] = hook.Name
+		kubeCfg.Monitor.Metadata.MetricLabels = map[string]string{
+			"hook":    hook.Name,
+			"binding": kubeCfg.BindingName,
+		}
 	}
 
 	hookCtrl := controller.NewHookController()
diff --git a/pkg/kube_events_manager/kube_events_manager.go b/pkg/kube_events_manager/kube_events_manager.go
index f3c3993..21aa2e4 100644
--- a/pkg/kube_events_manager/kube_events_manager.go
+++ b/pkg/kube_events_manager/kube_events_manager.go
@@ -2,15 +2,18 @@ package kube_events_manager
 
 import (
 	"context"
+	"runtime/trace"
 
-	"github.com/flant/shell-operator/pkg/kube"
 	log "github.com/sirupsen/logrus"
 
+	"github.com/flant/shell-operator/pkg/kube"
 	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
+	"github.com/flant/shell-operator/pkg/metrics_storage"
 )
 
 type KubeEventsManager interface {
 	WithContext(ctx context.Context)
+	WithMetricStorage(mstor *metrics_storage.MetricStorage)
 	WithKubeClient(client kube.KubernetesClient)
 	AddMonitor(monitorConfig *MonitorConfig) (*KubeEvent, error)
 	HasMonitor(monitorId string) bool
@@ -20,6 +23,7 @@ type KubeEventsManager interface {
 
 	StopMonitor(configId string) error
 	Ch() chan KubeEvent
+	PauseHandleEvents()
 }
 
 // kubeEventsManager is a main implementation of KubeEventsManager.
@@ -31,8 +35,9 @@ type kubeEventsManager struct {
 
 	KubeClient kube.KubernetesClient
 
-	ctx    context.Context
-	cancel context.CancelFunc
+	ctx           context.Context
+	cancel        context.CancelFunc
+	metricStorage *metrics_storage.MetricStorage
 }
 
 // kubeEventsManager should implement KubeEventsManager.
@@ -51,6 +56,10 @@ func (mgr *kubeEventsManager) WithContext(ctx context.Context) {
 	mgr.ctx, mgr.cancel = context.WithCancel(ctx)
 }
 
+func (mgr *kubeEventsManager) WithMetricStorage(mstor *metrics_storage.MetricStorage) {
+	mgr.metricStorage = mstor
+}
+
 func (mgr *kubeEventsManager) WithKubeClient(client kube.KubernetesClient) {
 	mgr.KubeClient = client
 }
@@ -61,9 +70,12 @@ func (mgr *kubeEventsManager) WithKubeClient(client kube.KubernetesClient) {
 func (mgr *kubeEventsManager) AddMonitor(monitorConfig *MonitorConfig) (*KubeEvent, error) {
 	log.Debugf("Add MONITOR %+v", monitorConfig)
 	monitor := NewMonitor()
+	monitor.WithContext(mgr.ctx)
 	monitor.WithKubeClient(mgr.KubeClient)
+	monitor.WithMetricStorage(mgr.metricStorage)
 	monitor.WithConfig(monitorConfig)
 	monitor.WithKubeEventCb(func(ev KubeEvent) {
+		defer trace.StartRegion(context.Background(), "EmitKubeEvent").End()
 		outEvent := mgr.MakeKubeEvent(monitor, ev)
 		if outEvent != nil {
 			mgr.KubeEventCh <- *outEvent
@@ -140,18 +152,11 @@ func (mgr *kubeEventsManager) Ch() chan KubeEvent {
 	return mgr.KubeEventCh
 }
 
-func (mgr *kubeEventsManager) StopAll() {
-	mgr.cancel()
-	// wait?
-	//
-	//informers, ok := mgr.InformersStore[configId]
-	//if ok {
-	//	for _, informer := range informers {
-	//		informer.Stop()
-	//	}
-	//	delete(mgr.InformersStore, configId)
-	//} else {
-	//	log.Errorf("configId '%s' has no informers to stop", configId)
-	//}
-	//return nil
+// PauseHandleEvents set flags for all informers to ignore incoming events.
+// Useful for shutdown without panicking.
+// Calling cancel() leads to a race and panicking, see https://github.com/kubernetes/kubernetes/issues/59822
+func (mgr *kubeEventsManager) PauseHandleEvents() {
+	for _, monitor := range mgr.Monitors {
+		monitor.PauseHandleEvents()
+	}
 }
diff --git a/pkg/kube_events_manager/monitor.go b/pkg/kube_events_manager/monitor.go
index 7f63328..586162e 100644
--- a/pkg/kube_events_manager/monitor.go
+++ b/pkg/kube_events_manager/monitor.go
@@ -5,20 +5,24 @@ import (
 	"fmt"
 	"sort"
 
-	"github.com/flant/shell-operator/pkg/kube"
-	utils "github.com/flant/shell-operator/pkg/utils/labels"
 	log "github.com/sirupsen/logrus"
 
+	"github.com/flant/shell-operator/pkg/kube"
 	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
+	"github.com/flant/shell-operator/pkg/metrics_storage"
+	utils "github.com/flant/shell-operator/pkg/utils/labels"
 )
 
 type Monitor interface {
+	WithContext(ctx context.Context)
 	WithKubeClient(client kube.KubernetesClient)
+	WithMetricStorage(mstor *metrics_storage.MetricStorage)
 	WithConfig(config *MonitorConfig)
 	WithKubeEventCb(eventCb func(KubeEvent))
 	CreateInformers() error
 	Start(context.Context)
 	Stop()
+	PauseHandleEvents()
 	GetExistedObjects() []ObjectAndFilterResult
 	GetConfig() *MonitorConfig
 }
@@ -42,8 +46,9 @@ type monitor struct {
 
 	cancelForNs map[string]context.CancelFunc
 
-	ctx    context.Context
-	cancel context.CancelFunc
+	ctx           context.Context
+	cancel        context.CancelFunc
+	metricStorage *metrics_storage.MetricStorage
 }
 
 var NewMonitor = func() Monitor {
@@ -55,10 +60,18 @@ var NewMonitor = func() Monitor {
 	}
 }
 
+func (m *monitor) WithContext(ctx context.Context) {
+	m.ctx, m.cancel = context.WithCancel(ctx)
+}
+
 func (m *monitor) WithKubeClient(client kube.KubernetesClient) {
 	m.KubeClient = client
 }
 
+func (m *monitor) WithMetricStorage(mstor *metrics_storage.MetricStorage) {
+	m.metricStorage = mstor
+}
+
 func (m *monitor) WithConfig(config *MonitorConfig) {
 	m.Config = config
 }
@@ -103,6 +116,7 @@ func (m *monitor) CreateInformers() error {
 	if m.Config.NamespaceSelector != nil && m.Config.NamespaceSelector.LabelSelector != nil {
 		logEntry.Debugf("Create NamespaceInformer for namespace.labelSelector")
 		m.NamespaceInformer = NewNamespaceInformer(m.Config)
+		m.NamespaceInformer.WithContext(m.ctx)
 		m.NamespaceInformer.WithKubeClient(m.KubeClient)
 		err := m.NamespaceInformer.CreateSharedInformer(
 			func(nsName string) {
@@ -129,7 +143,8 @@ func (m *monitor) CreateInformers() error {
 				ctx, m.cancelForNs[nsName] = context.WithCancel(m.ctx)
 
 				for _, informer := range m.VaryingInformers[nsName] {
-					go informer.Run(ctx.Done())
+					informer.WithContext(ctx)
+					go informer.Start()
 				}
 			},
 			func(nsName string) {
@@ -212,7 +227,9 @@ func (m *monitor) CreateInformersForNamespace(namespace string) (informers []Res
 
 	for _, objName := range objNames {
 		informer := NewResourceInformer(m.Config)
+		informer.WithContext(m.ctx)
 		informer.WithKubeClient(m.KubeClient)
+		informer.WithMetricStorage(m.metricStorage)
 		informer.WithNamespace(namespace)
 		informer.WithName(objName)
 		informer.WithKubeEventCb(m.eventCb)
@@ -232,19 +249,20 @@ func (m *monitor) Start(parentCtx context.Context) {
 	m.ctx, m.cancel = context.WithCancel(parentCtx)
 
 	for _, informer := range m.ResourceInformers {
-		go informer.Run(m.ctx.Done())
+		go informer.Start()
 	}
 
 	for nsName := range m.VaryingInformers {
 		var ctx context.Context
 		ctx, m.cancelForNs[nsName] = context.WithCancel(m.ctx)
 		for _, informer := range m.VaryingInformers[nsName] {
-			go informer.Run(ctx.Done())
+			informer.WithContext(ctx)
+			go informer.Start()
 		}
 	}
 
 	if m.NamespaceInformer != nil {
-		go m.NamespaceInformer.Run(m.ctx.Done())
+		go m.NamespaceInformer.Start()
 	}
 }
 
@@ -252,3 +270,23 @@ func (m *monitor) Start(parentCtx context.Context) {
 func (m *monitor) Stop() {
 	m.cancel()
 }
+
+// PauseHandleEvents set flags for all informers to ignore incoming events.
+// Useful for shutdown without panicking.
+// Calling cancel() leads to a race and panicking, see https://github.com/kubernetes/kubernetes/issues/59822
+func (m *monitor) PauseHandleEvents() {
+	for _, informer := range m.ResourceInformers {
+		informer.PauseHandleEvents()
+	}
+
+	for _, informers := range m.VaryingInformers {
+		for _, informer := range informers {
+			informer.PauseHandleEvents()
+		}
+	}
+
+	if m.NamespaceInformer != nil {
+		m.NamespaceInformer.PauseHandleEvents()
+	}
+
+}
diff --git a/pkg/kube_events_manager/monitor_config.go b/pkg/kube_events_manager/monitor_config.go
index 4623b1d..0936efd 100644
--- a/pkg/kube_events_manager/monitor_config.go
+++ b/pkg/kube_events_manager/monitor_config.go
@@ -12,9 +12,10 @@ import (
 // version of OnKubernetesEventConfig.
 type MonitorConfig struct {
 	Metadata struct {
-		MonitorId string
-		DebugName string
-		LogLabels map[string]string
+		MonitorId    string
+		DebugName    string
+		LogLabels    map[string]string
+		MetricLabels map[string]string
 	}
 	EventTypes        []WatchEventType
 	ApiVersion        string
diff --git a/pkg/kube_events_manager/namespace_informer.go b/pkg/kube_events_manager/namespace_informer.go
index 6c5f48f..2d6a971 100644
--- a/pkg/kube_events_manager/namespace_informer.go
+++ b/pkg/kube_events_manager/namespace_informer.go
@@ -3,6 +3,7 @@ package kube_events_manager
 // Namespace manager monitor namespaces for onKubernetesEvent config.
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -15,19 +16,28 @@ import (
 )
 
 type NamespaceInformer interface {
+	WithContext(ctx context.Context)
 	WithKubeClient(client kube.KubernetesClient)
 	CreateSharedInformer(addFn func(string), delFn func(string)) error
 	GetExistedObjects() map[string]bool
-	Run(stopCh <-chan struct{})
+	Start()
 	Stop()
+	PauseHandleEvents()
 }
 
 type namespaceInformer struct {
+	ctx     context.Context
+	cancel  context.CancelFunc
+	stopped bool
+
 	KubeClient     kube.KubernetesClient
 	Monitor        *MonitorConfig
 	SharedInformer cache.SharedInformer
 
 	ExistedObjects map[string]bool
+
+	addFn func(string)
+	delFn func(string)
 }
 
 // namespaceInformer implements NamespaceInformer interface
@@ -41,6 +51,10 @@ var NewNamespaceInformer = func(monitor *MonitorConfig) NamespaceInformer {
 	return informer
 }
 
+func (ni *namespaceInformer) WithContext(ctx context.Context) {
+	ni.ctx, ni.cancel = context.WithCancel(ctx)
+}
+
 func (ni *namespaceInformer) WithKubeClient(client kube.KubernetesClient) {
 	ni.KubeClient = client
 }
@@ -64,7 +78,9 @@ func (ni *namespaceInformer) CreateSharedInformer(addFn func(string), delFn func
 	}
 
 	ni.SharedInformer = corev1.NewFilteredNamespaceInformer(ni.KubeClient, resyncPeriod, indexers, tweakListOptions)
-	ni.SharedInformer.AddEventHandler(SharedNamespaceInformerEventHandler(ni, addFn, delFn))
+	ni.addFn = addFn
+	ni.delFn = delFn
+	ni.SharedInformer.AddEventHandler(ni) //SharedNamespaceInformerEventHandler(ni, addFn, delFn))
 
 	listOptions := metav1.ListOptions{}
 	tweakListOptions(&listOptions)
@@ -86,30 +102,56 @@ func (ni *namespaceInformer) GetExistedObjects() map[string]bool {
 	return ni.ExistedObjects
 }
 
-var SharedNamespaceInformerEventHandler = func(informer *namespaceInformer, addFn func(string), delFn func(string)) cache.ResourceEventHandlerFuncs {
-	return cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			nsObj := obj.(*v1.Namespace)
-			log.Debugf("NamespaceInformer: Added ns/%s", nsObj.Name)
-			addFn(nsObj.Name)
-		},
-		DeleteFunc: func(obj interface{}) {
-			if staleObj, stale := obj.(cache.DeletedFinalStateUnknown); stale {
-				obj = staleObj.Obj
-			}
-			nsObj := obj.(*v1.Namespace)
-			log.Debugf("NamespaceInformer: Deleted ns/%s", nsObj.Name)
-			delFn(nsObj.Name)
-		},
+func (ni *namespaceInformer) OnAdd(obj interface{}) {
+	if ni.stopped {
+		return
+	}
+	nsObj := obj.(*v1.Namespace)
+	log.Debugf("NamespaceInformer: Added ns/%s", nsObj.Name)
+	if ni.addFn != nil {
+		ni.addFn(nsObj.Name)
+	}
+}
+func (ni *namespaceInformer) OnUpdate(_ interface{}, _ interface{}) {
+	// Modified event for namespace is ignored
+}
+func (ni *namespaceInformer) OnDelete(obj interface{}) {
+	if ni.stopped {
+		return
+	}
+	if staleObj, stale := obj.(cache.DeletedFinalStateUnknown); stale {
+		obj = staleObj.Obj
+	}
+	nsObj := obj.(*v1.Namespace)
+	log.Debugf("NamespaceInformer: Deleted ns/%s", nsObj.Name)
+	if ni.delFn != nil {
+		ni.delFn(nsObj.Name)
 	}
 }
 
-func (ni *namespaceInformer) Run(stopCh <-chan struct{}) {
+func (ni *namespaceInformer) Start() {
 	log.Debugf("%s: Run namespace informer", ni.Monitor.Metadata.DebugName)
-	if ni.SharedInformer != nil {
-		go ni.SharedInformer.Run(stopCh)
+	if ni.SharedInformer == nil {
+		log.Errorf("%s: Possible BUG!!! Start called before CreateSharedInformer, ShredInformer is nil", ni.Monitor.Metadata.DebugName)
+		return
 	}
+	stopCh := make(chan struct{}, 1)
+	go func() {
+		<-ni.ctx.Done()
+		ni.stopped = true
+		close(stopCh)
+	}()
+
+	ni.SharedInformer.Run(stopCh)
 }
 
 func (ni *namespaceInformer) Stop() {
+	if ni.cancel != nil {
+		ni.cancel()
+	}
+	ni.stopped = true
+}
+
+func (ni *namespaceInformer) PauseHandleEvents() {
+	ni.stopped = true
 }
diff --git a/pkg/kube_events_manager/resource_informer.go b/pkg/kube_events_manager/resource_informer.go
index 50b153c..b9eb5db 100644
--- a/pkg/kube_events_manager/resource_informer.go
+++ b/pkg/kube_events_manager/resource_informer.go
@@ -1,7 +1,9 @@
 package kube_events_manager
 
 import (
+	"context"
 	"fmt"
+	"runtime/trace"
 	"sync"
 	"time"
 
@@ -12,21 +14,25 @@ import (
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/tools/cache"
 
-	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
-
 	"github.com/flant/shell-operator/pkg/kube"
+	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
+	"github.com/flant/shell-operator/pkg/metrics_storage"
+	. "github.com/flant/shell-operator/pkg/utils/measure"
 )
 
 // ResourceInformer is a kube informer for particular onKubernetesEvent
 type ResourceInformer interface {
+	WithContext(ctx context.Context)
 	WithKubeClient(client kube.KubernetesClient)
+	WithMetricStorage(mstor *metrics_storage.MetricStorage)
 	WithNamespace(string)
 	WithName(string)
 	WithKubeEventCb(eventCb func(KubeEvent))
 	CreateSharedInformer() error
 	GetExistedObjects() []ObjectAndFilterResult
-	Run(stopCh <-chan struct{})
+	Start()
 	Stop()
+	PauseHandleEvents()
 }
 
 type resourceInformer struct {
@@ -47,7 +53,13 @@ type resourceInformer struct {
 	eventCb func(KubeEvent)
 
 	// TODO resourceInformer should be stoppable (think of deleted namespaces and disabled modules in addon-operator)
-	//ctx context.Context
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	metricStorage *metrics_storage.MetricStorage
+
+	// a flag to stop handle events after Stop()
+	stopped bool
 }
 
 // resourceInformer should implement ResourceInformer
@@ -62,10 +74,18 @@ var NewResourceInformer = func(monitor *MonitorConfig) ResourceInformer {
 	return informer
 }
 
+func (ei *resourceInformer) WithContext(ctx context.Context) {
+	ei.ctx, ei.cancel = context.WithCancel(ctx)
+}
+
 func (ei *resourceInformer) WithKubeClient(client kube.KubernetesClient) {
 	ei.KubeClient = client
 }
 
+func (ei *resourceInformer) WithMetricStorage(mstor *metrics_storage.MetricStorage) {
+	ei.metricStorage = mstor
+}
+
 func (ei *resourceInformer) WithNamespace(ns string) {
 	ei.Namespace = ns
 }
@@ -151,6 +171,7 @@ func (ei *resourceInformer) GetExistedObjects() []ObjectAndFilterResult {
 // ListExistedObjects get a list of existed objects in namespace that match selectors and
 // fills Checksum map with checksums of existing objects.
 func (ei *resourceInformer) ListExistedObjects() error {
+	defer trace.StartRegion(context.Background(), "ListExistedObjects").End()
 	objList, err := ei.KubeClient.Dynamic().
 		Resource(ei.GroupVersionResource).
 		Namespace(ei.Namespace).
@@ -174,7 +195,17 @@ func (ei *resourceInformer) ListExistedObjects() error {
 	for _, item := range objList.Items {
 		// copy loop var to avoid duplication of pointer
 		obj := item
-		objFilterRes, err := ApplyJqFilter(ei.Monitor.JqFilter, &obj)
+		//objFilterRes, err := ApplyJqFilter(ei.Monitor.JqFilter, &obj)
+
+		var objFilterRes *ObjectAndFilterResult
+		var err error
+		func() {
+			defer MeasureTime(func(nanos Nanos) {
+				ei.metricStorage.ObserveHistogram("kube_jq_hist", nanos.Ms(), ei.Monitor.Metadata.MetricLabels)
+			})()
+			objFilterRes, err = ApplyJqFilter(ei.Monitor.JqFilter, &obj)
+		}()
+
 		if err != nil {
 			return err
 		}
@@ -214,6 +245,16 @@ func (ei *resourceInformer) OnDelete(obj interface{}) {
 // TODO add delay to merge Added and Modified events (node added and then labels applied — one hook run on Added+Modified is enough)
 //func (ei *resourceInformer) HandleKubeEvent(obj *unstructured.Unstructured, objectId string, filterResult string, newChecksum string, eventType WatchEventType) {
 func (ei *resourceInformer) HandleWatchEvent(object interface{}, eventType WatchEventType) {
+	// check if stop
+	if ei.stopped {
+		return
+	}
+
+	defer MeasureTime(func(nanos Nanos) {
+		ei.metricStorage.ObserveHistogram("kube_event_duration_hist", nanos.Ms(), ei.Monitor.Metadata.MetricLabels)
+	})()
+	defer trace.StartRegion(context.Background(), "HandleWatchEvent").End()
+
 	if staleObj, stale := object.(cache.DeletedFinalStateUnknown); stale {
 		object = staleObj.Obj
 	}
@@ -223,7 +264,14 @@ func (ei *resourceInformer) HandleWatchEvent(object interface{}, eventType Watch
 
 	// Always calculate checksum and update cache, because we need actual state in CachedObjects
 
-	objFilterRes, err := ApplyJqFilter(ei.Monitor.JqFilter, obj)
+	var objFilterRes *ObjectAndFilterResult
+	var err error
+	func() {
+		defer MeasureTime(func(nanos Nanos) {
+			ei.metricStorage.ObserveHistogram("kube_jq_hist", nanos.Ms(), ei.Monitor.Metadata.MetricLabels)
+		})()
+		objFilterRes, err = ApplyJqFilter(ei.Monitor.JqFilter, obj)
+	}()
 	if err != nil {
 		log.Errorf("%s: WATCH %s: %s",
 			ei.Monitor.Metadata.DebugName,
@@ -320,11 +368,27 @@ func (ei *resourceInformer) ShouldFireEvent(checkEvent WatchEventType) bool {
 	return false
 }
 
-func (ei *resourceInformer) Run(stopCh <-chan struct{}) {
+func (ei *resourceInformer) Start() {
 	log.Debugf("%s: RUN resource informer", ei.Monitor.Metadata.DebugName)
+	stopCh := make(chan struct{}, 1)
+	go func() {
+		<-ei.ctx.Done()
+		ei.stopped = true
+		close(stopCh)
+	}()
+
 	ei.SharedInformer.Run(stopCh)
 }
 
 func (ei *resourceInformer) Stop() {
 	log.Debugf("%s: STOP resource informer", ei.Monitor.Metadata.DebugName)
+	if ei.cancel != nil {
+		ei.cancel()
+	}
+	ei.stopped = true
+}
+
+func (ei *resourceInformer) PauseHandleEvents() {
+	log.Debugf("%s: PAUSE resource informer", ei.Monitor.Metadata.DebugName)
+	ei.stopped = true
 }
diff --git a/pkg/kube_events_manager/util.go b/pkg/kube_events_manager/util.go
index f6b4193..1364f32 100644
--- a/pkg/kube_events_manager/util.go
+++ b/pkg/kube_events_manager/util.go
@@ -2,10 +2,12 @@ package kube_events_manager
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
 	"os/exec"
+	"runtime/trace"
 	"strings"
 
 	. "github.com/flant/libjq-go"
@@ -24,6 +26,8 @@ import (
 // over result and return ObjectAndFilterResult. If jqFilter is empty, no filter
 // is required and checksum is calculated over full json representation of the object.
 func ApplyJqFilter(jqFilter string, obj *unstructured.Unstructured) (*ObjectAndFilterResult, error) {
+	defer trace.StartRegion(context.Background(), "ApplyJqFilter").End()
+
 	res := &ObjectAndFilterResult{
 		Object: obj,
 	}
diff --git a/pkg/metrics_storage/metric_storage.go b/pkg/metrics_storage/metric_storage.go
new file mode 100644
index 0000000..238c16f
--- /dev/null
+++ b/pkg/metrics_storage/metric_storage.go
@@ -0,0 +1,334 @@
+package metrics_storage
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/flant/shell-operator/pkg/metrics_storage/operation"
+	"github.com/flant/shell-operator/pkg/metrics_storage/vault"
+	. "github.com/flant/shell-operator/pkg/utils/labels"
+)
+
+const (
+	HistogramDefaultStart float64 = 0.0
+	HistogramDefaultWidth float64 = 2
+	HistogramDefaultCount int     = 20
+)
+
+// MetricStorage is used to register metric values.
+type MetricStorage struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	Prefix string
+
+	Counters         map[string]*prometheus.CounterVec
+	Gauges           map[string]*prometheus.GaugeVec
+	Histograms       map[string]*prometheus.HistogramVec
+	HistogramBuckets map[string][]float64
+
+	countersLock   sync.Mutex
+	gaugesLock     sync.Mutex
+	histogramsLock sync.Mutex
+
+	GroupedVault *vault.GroupedVault
+}
+
+func NewMetricStorage() *MetricStorage {
+	return &MetricStorage{
+		Gauges:           make(map[string]*prometheus.GaugeVec),
+		Counters:         make(map[string]*prometheus.CounterVec),
+		Histograms:       make(map[string]*prometheus.HistogramVec),
+		HistogramBuckets: make(map[string][]float64),
+		GroupedVault:     vault.NewGroupedVault(),
+	}
+}
+
+func (m *MetricStorage) WithContext(ctx context.Context) {
+	m.ctx, m.cancel = context.WithCancel(ctx)
+}
+
+func (m *MetricStorage) WithPrefix(prefix string) {
+	m.Prefix = prefix
+}
+
+func (m *MetricStorage) Stop() {
+	if m.cancel != nil {
+		m.cancel()
+	}
+}
+
+func (m *MetricStorage) Start() {
+	//go func() {
+	//	<-m.ctx.Done()
+	//	return
+	//}()
+}
+
+func (m *MetricStorage) SendGauge(metric string, value float64, labels map[string]string) {
+	if m == nil {
+		return
+	}
+	m.SendGaugeNoPrefix(m.Prefix+metric, value, labels)
+}
+func (m *MetricStorage) SendGaugeNoPrefix(metric string, value float64, labels map[string]string) {
+	if m == nil {
+		return
+	}
+	defer func() {
+		if r := recover(); r != nil {
+			log.WithField("operator.component", "metricsStorage").
+				Errorf("Metric gauge %s %v update with %v error: %v", metric, LabelNames(labels), labels, r)
+		}
+	}()
+
+	m.GetOrDefineGauge(metric, labels).With(labels).Set(value)
+}
+
+func (m *MetricStorage) GetOrDefineGauge(metric string, labels map[string]string) *prometheus.GaugeVec {
+	defer func() {
+		if r := recover(); r != nil {
+			log.WithField("operator.component", "metricsStorage").
+				Errorf("Metric gauge %s %v create with %v error: %v", metric, LabelNames(labels), labels, r)
+		}
+	}()
+
+	log.WithField("operator.component", "metricsStorage").
+		Debugf("Get metric gauge %s", metric)
+	m.gaugesLock.Lock()
+	defer m.gaugesLock.Unlock()
+
+	vec, ok := m.Gauges[metric]
+
+	// create and register CounterVec
+	if !ok {
+		vec = prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: metric,
+				Help: metric,
+			},
+			LabelNames(labels),
+		)
+		log.WithField("operator.component", "metricsStorage").
+			Infof("Create new metric gauge %s", metric)
+		prometheus.MustRegister(vec)
+		m.Gauges[metric] = vec
+	}
+
+	return vec
+}
+
+func (m *MetricStorage) SendCounter(metric string, value float64, labels map[string]string) {
+	if m == nil {
+		return
+	}
+	m.SendCounterNoPrefix(m.Prefix+metric, value, labels)
+}
+
+func (m *MetricStorage) SendCounterNoPrefix(metric string, value float64, labels map[string]string) {
+	if m == nil {
+		return
+	}
+	defer func() {
+		if r := recover(); r != nil {
+			log.WithField("operator.component", "metricsStorage").
+				Errorf("Metric %s %v update with %v error: %v", metric, LabelNames(labels), labels, r)
+		}
+	}()
+
+	m.GetOrDefineCounter(metric, labels).With(labels).Add(value)
+}
+
+func (m *MetricStorage) GetOrDefineCounter(metric string, labels map[string]string) *prometheus.CounterVec {
+	defer func() {
+		if r := recover(); r != nil {
+			log.WithField("operator.component", "metricsStorage").
+				Errorf("Metric counter %s %v create with %v error: %v", metric, LabelNames(labels), labels, r)
+		}
+	}()
+
+	log.WithField("operator.component", "metricsStorage").
+		Debugf("Get metric counter %s", metric)
+	m.countersLock.Lock()
+	defer m.countersLock.Unlock()
+	vec, ok := m.Counters[metric]
+
+	// create and register CounterVec
+	if !ok {
+		vec = prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: metric,
+				Help: metric,
+			},
+			LabelNames(labels),
+		)
+		log.WithField("operator.component", "metricsStorage").
+			Infof("Create new metric counter %s", metric)
+		prometheus.MustRegister(vec)
+		m.Counters[metric] = vec
+	}
+
+	return vec
+}
+
+func (m *MetricStorage) ObserveHistogramNoPrefix(metric string, value float64, labels map[string]string) {
+	if m == nil {
+		return
+	}
+	defer func() {
+		if r := recover(); r != nil {
+			log.WithField("operator.component", "metricsStorage").
+				Errorf("Metric histogram %s %v update with %v error: %v", metric, LabelNames(labels), labels, r)
+		}
+	}()
+	m.GetOrDefineHistogram(metric, labels).With(labels).Observe(value)
+}
+
+func (m *MetricStorage) ObserveHistogram(metric string, value float64, labels map[string]string) {
+	if m == nil {
+		return
+	}
+	m.ObserveHistogramNoPrefix(m.Prefix+metric, value, labels)
+}
+
+func (m *MetricStorage) DefineHistogramBuckets(metric string, buckets []float64) {
+	m.DefineHistogramBucketsNoPrefix(m.Prefix+metric, buckets)
+}
+
+func (m *MetricStorage) DefineHistogramBucketsNoPrefix(metric string, buckets []float64) {
+	if m == nil {
+		return
+	}
+	m.HistogramBuckets[metric] = buckets
+}
+
+func (m *MetricStorage) GetOrDefineHistogram(metric string, labels map[string]string) *prometheus.HistogramVec {
+	defer func() {
+		if r := recover(); r != nil {
+			log.WithField("operator.component", "metricsStorage").
+				Errorf("Metric histogram %s %v create with %v error: %v", metric, LabelNames(labels), labels, r)
+		}
+	}()
+
+	log.WithField("operator.component", "metricsStorage").
+		Debugf("Get metric histogram %s", metric)
+
+	m.histogramsLock.Lock()
+	defer m.histogramsLock.Unlock()
+	vec, ok := m.Histograms[metric]
+	if !ok {
+		buckets, has := m.HistogramBuckets[metric]
+		if !has {
+			buckets = prometheus.LinearBuckets(HistogramDefaultStart, HistogramDefaultWidth, HistogramDefaultCount)
+		}
+		vec = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    metric,
+			Help:    metric,
+			Buckets: buckets,
+		}, LabelNames(labels))
+
+		log.WithField("operator.component", "metricsStorage").
+			Infof("Create new metric histogram %s", metric)
+		prometheus.MustRegister(vec)
+		m.Histograms[metric] = vec
+	}
+	return vec
+}
+
+func (m *MetricStorage) SendBatchV0(ops []operation.MetricOperation, labels map[string]string) error {
+	if m == nil {
+		return nil
+	}
+	// Apply metric operations
+	for _, metricOp := range ops {
+		labels := MergeLabels(metricOp.Labels, labels)
+
+		if metricOp.Add != nil {
+			m.SendCounterNoPrefix(metricOp.Name, *metricOp.Add, labels)
+			continue
+		}
+		if metricOp.Set != nil {
+			m.SendGaugeNoPrefix(metricOp.Name, *metricOp.Set, labels)
+			continue
+		}
+		return fmt.Errorf("no operation in metric from module hook, name=%s", metricOp.Name)
+	}
+	return nil
+}
+
+func (m *MetricStorage) SendBatch(ops []operation.MetricOperation, labels map[string]string) error {
+	if m == nil {
+		return nil
+	}
+
+	// Group ops by 'Group'.
+	var groupedOps = make(map[string][]operation.MetricOperation)
+	var nonGroupedOps = make([]operation.MetricOperation, 0)
+
+	for _, op := range ops {
+		if op.Group == "" {
+			nonGroupedOps = append(nonGroupedOps, op)
+			continue
+		}
+		if _, ok := groupedOps[op.Group]; !ok {
+			groupedOps[op.Group] = make([]operation.MetricOperation, 0)
+		}
+		groupedOps[op.Group] = append(groupedOps[op.Group], op)
+	}
+
+	// Apply metric operations for each group
+	for group, ops := range groupedOps {
+		// clean group
+		m.ApplyGroupOperations(group, ops, labels)
+	}
+
+	// backward compatibility — send metrics without cleaning
+	err := m.SendBatchV0(nonGroupedOps, labels)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m *MetricStorage) ApplyOperation(op operation.MetricOperation, commonLabels map[string]string) {
+	labels := MergeLabels(op.Labels, commonLabels)
+
+	if op.Add != nil {
+		m.SendCounterNoPrefix(op.Name, *op.Add, labels)
+		return
+	}
+	if op.Set != nil {
+		m.SendGaugeNoPrefix(op.Name, *op.Set, labels)
+		return
+	}
+}
+
+func (m *MetricStorage) ApplyGroupOperations(group string, ops []operation.MetricOperation, commonLabels map[string]string) {
+	// Gather labels to clear absent metrics by label values
+	var metricLabels = make(map[string][]map[string]string)
+	for _, op := range ops {
+		if _, ok := metricLabels[op.Name]; !ok {
+			metricLabels[op.Name] = make([]map[string]string, 0)
+		}
+		metricLabels[op.Name] = append(metricLabels[op.Name], MergeLabels(op.Labels, commonLabels))
+	}
+	m.GroupedVault.ClearMissingMetrics(group, metricLabels)
+
+	// Apply metric operations
+	for _, op := range ops {
+		labels := MergeLabels(op.Labels, commonLabels)
+		if op.Add != nil {
+			m.GroupedVault.CounterAdd(group, op.Name, *op.Add, labels)
+			continue
+		}
+		if op.Set != nil {
+			m.GroupedVault.GaugeSet(group, op.Name, *op.Set, labels)
+			continue
+		}
+	}
+}
diff --git a/pkg/metrics_storage/metrics_storage.go b/pkg/metrics_storage/metrics_storage.go
deleted file mode 100644
index c853d6d..0000000
--- a/pkg/metrics_storage/metrics_storage.go
+++ /dev/null
@@ -1,222 +0,0 @@
-package metrics_storage
-
-import (
-	"context"
-	"fmt"
-
-	utils "github.com/flant/shell-operator/pkg/utils/labels"
-	"github.com/prometheus/client_golang/prometheus"
-	log "github.com/sirupsen/logrus"
-)
-
-// MetricStorage is used to synchronously register metric values.
-type MetricStorage struct {
-	ctx    context.Context
-	cancel context.CancelFunc
-
-	MetricChan chan Metric
-	MetricVecs map[string]MetricVec
-	Prefix     string
-}
-
-func NewMetricStorage() *MetricStorage {
-	return &MetricStorage{
-		MetricChan: make(chan Metric, 1000),
-		MetricVecs: make(map[string]MetricVec),
-	}
-}
-
-func (m *MetricStorage) WithContext(ctx context.Context) {
-	m.ctx, m.cancel = context.WithCancel(ctx)
-}
-
-func (m *MetricStorage) WithPrefix(prefix string) {
-	m.Prefix = prefix
-}
-
-func (m *MetricStorage) Stop() {
-	if m.cancel != nil {
-		m.cancel()
-	}
-}
-
-func (m *MetricStorage) Start() {
-	go func() {
-		for {
-			select {
-			case metric := <-m.MetricChan:
-				metric.store(m)
-			case <-m.ctx.Done():
-				return
-			}
-		}
-	}()
-}
-
-func (m *MetricStorage) SendGauge(metric string, value float64, labels map[string]string) {
-	m.MetricChan <- NewGaugeMetric(m.Prefix+metric, value, labels)
-}
-func (m *MetricStorage) SendCounter(metric string, value float64, labels map[string]string) {
-	m.MetricChan <- NewCounterMetric(m.Prefix+metric, value, labels)
-}
-
-func (m *MetricStorage) SendGaugeNoPrefix(metric string, value float64, labels map[string]string) {
-	m.MetricChan <- NewGaugeMetric(metric, value, labels)
-}
-func (m *MetricStorage) SendCounterNoPrefix(metric string, value float64, labels map[string]string) {
-	m.MetricChan <- NewCounterMetric(metric, value, labels)
-}
-
-func (m *MetricStorage) SendBatch(ops []MetricOperation, labels map[string]string) error {
-	// Apply metric operations
-	for _, metricOp := range ops {
-		labels := utils.MergeLabels(metricOp.Labels, labels)
-
-		if metricOp.Add != nil {
-			m.SendCounterNoPrefix(metricOp.Name, *metricOp.Add, labels)
-			continue
-		}
-		if metricOp.Set != nil {
-			m.SendGaugeNoPrefix(metricOp.Name, *metricOp.Set, labels)
-			continue
-		}
-		return fmt.Errorf("no operation in metric from module hook, name=%s", metricOp.Name)
-	}
-	return nil
-}
-
-type Metric interface {
-	store(*MetricStorage)
-}
-
-type BaseMetric struct {
-	Metric string
-	Value  float64
-	Labels map[string]string
-}
-
-func (metric *BaseMetric) LabelsNames() []string {
-	variableLabelNames := make([]string, 0)
-	for labelName := range metric.Labels {
-		variableLabelNames = append(variableLabelNames, labelName)
-	}
-	return variableLabelNames
-}
-
-func (metric *BaseMetric) getOrCreateMetricVec(storage *MetricStorage, createVecFunc func() (prometheus.Collector, MetricVec)) MetricVec {
-	var metricVec MetricVec
-	var prometheusCollector prometheus.Collector
-	var hasMetricVec bool
-
-	metricVec, hasMetricVec = storage.MetricVecs[metric.Metric]
-
-	if !hasMetricVec {
-		prometheusCollector, metricVec = createVecFunc()
-
-		log.WithField("operator.component", "metricsStorage").Infof("Create new metric %s", metric.Metric)
-
-		prometheus.MustRegister(prometheusCollector)
-		storage.MetricVecs[metric.Metric] = metricVec
-	}
-
-	return metricVec
-}
-
-type GaugeMetric struct {
-	BaseMetric
-}
-
-func NewGaugeMetric(metric string, value float64, labels map[string]string) *GaugeMetric {
-	return &GaugeMetric{BaseMetric{
-		Metric: metric,
-		Value:  value,
-		Labels: labels,
-	}}
-}
-
-type CounterMetric struct {
-	BaseMetric
-}
-
-func NewCounterMetric(metric string, value float64, labels map[string]string) *CounterMetric {
-	return &CounterMetric{BaseMetric{
-		Metric: metric,
-		Value:  value,
-		Labels: labels,
-	}}
-}
-
-func (metric *GaugeMetric) store(storage *MetricStorage) {
-	metricVec := metric.getOrCreateMetricVec(storage, func() (prometheus.Collector, MetricVec) {
-		prometheusVec := prometheus.NewGaugeVec(
-			prometheus.GaugeOpts{
-				Name: metric.Metric,
-				Help: metric.Metric,
-			},
-			metric.LabelsNames(),
-		)
-		return prometheusVec, NewMetricGaugeVec(prometheusVec, metric.Metric, metric.LabelsNames())
-	})
-	metricVec.UpdateValue(metric.Labels, metric.Value)
-}
-
-func (metric *CounterMetric) store(storage *MetricStorage) {
-	metricVec := metric.getOrCreateMetricVec(storage, func() (prometheus.Collector, MetricVec) {
-		prometheusVec := prometheus.NewCounterVec(
-			prometheus.CounterOpts{
-				Name: metric.Metric,
-				Help: metric.Metric,
-			},
-			metric.LabelsNames(),
-		)
-		return prometheusVec, NewMetricCounterVec(prometheusVec, metric.Metric, metric.LabelsNames())
-	})
-	metricVec.UpdateValue(metric.Labels, metric.Value)
-}
-
-type MetricGaugeVec struct {
-	*prometheus.GaugeVec
-	Name       string
-	LabelNames []string
-}
-
-func NewMetricGaugeVec(gauge *prometheus.GaugeVec, name string, labelNames []string) *MetricGaugeVec {
-	metricGaugeVec := &MetricGaugeVec{gauge, name, make([]string, 0)}
-	metricGaugeVec.LabelNames = append(metricGaugeVec.LabelNames, labelNames...)
-	return metricGaugeVec
-}
-
-type MetricCounterVec struct {
-	*prometheus.CounterVec
-	Name       string
-	LabelNames []string
-}
-
-func NewMetricCounterVec(counter *prometheus.CounterVec, name string, labelNames []string) *MetricCounterVec {
-	metricCounterVec := &MetricCounterVec{counter, name, make([]string, 0)}
-	metricCounterVec.LabelNames = append(metricCounterVec.LabelNames, labelNames...)
-	return metricCounterVec
-}
-
-type MetricVec interface {
-	UpdateValue(labels prometheus.Labels, value float64)
-}
-
-func (metricVec *MetricGaugeVec) UpdateValue(labels prometheus.Labels, value float64) {
-	defer func() {
-		if r := recover(); r != nil {
-			log.WithField("operator.component", "metricsStorage").
-				Errorf("Metric %s %v update with %v error: %v", metricVec.Name, metricVec.LabelNames, labels, r)
-		}
-	}()
-	metricVec.With(labels).Set(value)
-}
-func (metricVec *MetricCounterVec) UpdateValue(labels prometheus.Labels, value float64) {
-	defer func() {
-		if r := recover(); r != nil {
-			log.WithField("operator.component", "metricsStorage").
-				Errorf("Metric %s %v update with %v error: %v", metricVec.Name, metricVec.LabelNames, labels, r)
-		}
-	}()
-	metricVec.With(labels).Add(value)
-}
diff --git a/pkg/metrics_storage/metric_operation.go b/pkg/metrics_storage/operation/operation.go
similarity index 94%
rename from pkg/metrics_storage/metric_operation.go
rename to pkg/metrics_storage/operation/operation.go
index d21f7a7..ef0e052 100644
--- a/pkg/metrics_storage/metric_operation.go
+++ b/pkg/metrics_storage/operation/operation.go
@@ -1,4 +1,4 @@
-package metrics_storage
+package operation
 
 import (
 	"bytes"
@@ -13,6 +13,7 @@ type MetricOperation struct {
 	Add    *float64          `json:"add,omitempty"`
 	Set    *float64          `json:"set,omitempty"`
 	Labels map[string]string `json:"labels"`
+	Group  string            `json:"group,omitempty"`
 }
 
 func MetricOperationsFromReader(r io.Reader) ([]MetricOperation, error) {
diff --git a/pkg/metrics_storage/vault/collector.go b/pkg/metrics_storage/vault/collector.go
new file mode 100644
index 0000000..88b2bd5
--- /dev/null
+++ b/pkg/metrics_storage/vault/collector.go
@@ -0,0 +1,225 @@
+package vault
+
+import (
+	"hash/fnv"
+	"sync"
+	"sync/atomic"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	. "github.com/flant/shell-operator/pkg/utils/labels"
+)
+
+type ConstMetricCollector interface {
+	Describe(ch chan<- *prometheus.Desc)
+	Collect(ch chan<- prometheus.Metric)
+	Type() string
+	LabelNames() []string
+	Name() string
+	ClearMissingMetrics(group string, labelSet []map[string]string)
+}
+
+var (
+	_ ConstMetricCollector = (*ConstCounterCollector)(nil)
+	_ ConstMetricCollector = (*ConstGaugeCollector)(nil)
+)
+
+type GroupedCounterMetric struct {
+	Value       uint64
+	LabelValues []string
+	Group       string
+}
+
+type GroupedGaugeMetric struct {
+	Value       float64
+	LabelValues []string
+	Group       string
+}
+
+type ConstCounterCollector struct {
+	mtx sync.RWMutex
+
+	collection map[uint64]GroupedCounterMetric
+	desc       *prometheus.Desc
+	name       string
+	labelNames []string
+}
+
+func NewConstCounterCollector(name string, labelNames []string) *ConstCounterCollector {
+	desc := prometheus.NewDesc(name, name, labelNames, nil)
+	return &ConstCounterCollector{
+		name:       name,
+		labelNames: labelNames,
+		desc:       desc,
+		collection: make(map[uint64]GroupedCounterMetric),
+	}
+}
+
+// Add increases a counter metric by a value. Metric is identified by label values and a group.
+func (c *ConstCounterCollector) Add(group string, value float64, labels map[string]string) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	labelValues := LabelValues(labels, c.labelNames)
+	labelsHash := HashLabelValues(labelValues)
+
+	// TODO add group to hash
+	storedMetric, ok := c.collection[labelsHash]
+	if !ok {
+		storedMetric = GroupedCounterMetric{
+			Value:       uint64(value),
+			LabelValues: labelValues,
+			Group:       group,
+		}
+	} else {
+		atomic.AddUint64(&storedMetric.Value, uint64(value))
+	}
+
+	c.collection[labelsHash] = storedMetric
+}
+
+func (c *ConstCounterCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.desc
+}
+
+func (c *ConstCounterCollector) Collect(ch chan<- prometheus.Metric) {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+
+	for _, s := range c.collection {
+		ch <- prometheus.MustNewConstMetric(c.desc, prometheus.CounterValue, float64(s.Value), s.LabelValues...)
+	}
+}
+
+func (c *ConstCounterCollector) Type() string {
+	return "counter"
+}
+
+func (c *ConstCounterCollector) LabelNames() []string {
+	return c.labelNames
+}
+
+func (c *ConstCounterCollector) Name() string {
+	return c.name
+}
+
+// ClearMissingMetrics deletes metrics from collection with unmatched hashes of labelValues.
+func (c *ConstCounterCollector) ClearMissingMetrics(group string, labelSet []map[string]string) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	hashedOps := make(map[uint64]struct{})
+	for _, labels := range labelSet {
+		hashedOps[HashLabelValues(LabelValues(labels, c.labelNames))] = struct{}{}
+	}
+
+	for hash, m := range c.collection {
+		if m.Group != group {
+			continue
+		}
+		if _, ok := hashedOps[hash]; !ok {
+			delete(c.collection, hash)
+		}
+	}
+}
+
+type ConstGaugeCollector struct {
+	mtx sync.RWMutex
+
+	name       string
+	labelNames []string
+	desc       *prometheus.Desc
+	collection map[uint64]GroupedGaugeMetric
+}
+
+func NewConstGaugeCollector(name string, labelNames []string) *ConstGaugeCollector {
+	desc := prometheus.NewDesc(name, name, labelNames, nil)
+	return &ConstGaugeCollector{
+		name:       name,
+		labelNames: labelNames,
+		desc:       desc,
+		collection: make(map[uint64]GroupedGaugeMetric),
+	}
+}
+
+func (c *ConstGaugeCollector) Set(group string, value float64, labels map[string]string) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	labelValues := LabelValues(labels, c.labelNames)
+	labelsHash := HashLabelValues(labelValues)
+
+	storedMetric, ok := c.collection[labelsHash]
+	if !ok {
+		storedMetric = GroupedGaugeMetric{
+			Value:       value,
+			LabelValues: labelValues,
+			Group:       group,
+		}
+	}
+
+	storedMetric.Value = value
+	c.collection[labelsHash] = storedMetric
+}
+
+func (c *ConstGaugeCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.desc
+}
+
+func (c *ConstGaugeCollector) Collect(ch chan<- prometheus.Metric) {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+
+	for _, s := range c.collection {
+		ch <- prometheus.MustNewConstMetric(c.desc, prometheus.GaugeValue, s.Value, s.LabelValues...)
+	}
+}
+
+func (c *ConstGaugeCollector) Type() string {
+	return "gauge"
+}
+
+func (c *ConstGaugeCollector) LabelNames() []string {
+	return c.labelNames
+}
+
+func (c *ConstGaugeCollector) Name() string {
+	return c.name
+}
+
+// ClearGroup removes all metrics for a group
+// TODO ClearGroup
+
+// Clear removes all metrics from this collector
+//TODO
+
+// ClearMissingMetrics deletes metrics from collection with unmatched hashes of labelValues.
+func (c *ConstGaugeCollector) ClearMissingMetrics(group string, labelSet []map[string]string) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	hashedOps := make(map[uint64]struct{})
+	for _, labels := range labelSet {
+		hashedOps[HashLabelValues(LabelValues(labels, c.labelNames))] = struct{}{}
+	}
+
+	for hash, m := range c.collection {
+		if m.Group != group {
+			continue
+		}
+		if _, ok := hashedOps[hash]; !ok {
+			delete(c.collection, hash)
+		}
+	}
+}
+
+const labelsSeparator = byte(255)
+
+func HashLabelValues(labelValues []string) uint64 {
+	hasher := fnv.New64a()
+	for _, labelValue := range labelValues {
+		_, _ = hasher.Write([]byte(labelValue))
+		_, _ = hasher.Write([]byte{labelsSeparator})
+	}
+	return hasher.Sum64()
+}
diff --git a/pkg/metrics_storage/vault/vault.go b/pkg/metrics_storage/vault/vault.go
new file mode 100644
index 0000000..1dae2af
--- /dev/null
+++ b/pkg/metrics_storage/vault/vault.go
@@ -0,0 +1,85 @@
+package vault
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	log "github.com/sirupsen/logrus"
+
+	. "github.com/flant/shell-operator/pkg/utils/labels"
+)
+
+type GroupedVault struct {
+	collectors map[string]ConstMetricCollector
+	mtx        sync.Mutex
+}
+
+func NewGroupedVault() *GroupedVault {
+	return &GroupedVault{
+		collectors: make(map[string]ConstMetricCollector),
+	}
+}
+
+// ClearMissingMetrics takes each collector in collector by its name and clear missing metrics by group.
+func (v *GroupedVault) ClearMissingMetrics(group string, metricLabels map[string][]map[string]string) {
+	v.mtx.Lock()
+	defer v.mtx.Unlock()
+	for name, collector := range v.collectors {
+		if labelSet, ok := metricLabels[name]; ok {
+			collector.ClearMissingMetrics(group, labelSet)
+		}
+	}
+}
+
+func (v *GroupedVault) GetOrCreateCounterCollector(name string, labelNames []string) (*ConstCounterCollector, error) {
+	v.mtx.Lock()
+	defer v.mtx.Unlock()
+	collector, ok := v.collectors[name]
+	if !ok {
+		collector = NewConstCounterCollector(name, labelNames)
+		if err := prometheus.Register(collector); err != nil {
+			return nil, fmt.Errorf("counter '%s' %v registration: %v", name, labelNames, err)
+		}
+		v.collectors[name] = collector
+	}
+	if counter, ok := collector.(*ConstCounterCollector); ok {
+		return counter, nil
+	}
+	return nil, fmt.Errorf("counter %v collector requested, but %s %v collector exists", labelNames, collector.Type(), collector.LabelNames())
+}
+
+func (v *GroupedVault) GetOrCreateGaugeCollector(name string, labelNames []string) (*ConstGaugeCollector, error) {
+	v.mtx.Lock()
+	defer v.mtx.Unlock()
+	collector, ok := v.collectors[name]
+	if !ok {
+		collector = NewConstGaugeCollector(name, labelNames)
+		if err := prometheus.Register(collector); err != nil {
+			return nil, fmt.Errorf("gauge '%s' %v registration: %v", name, labelNames, err)
+		}
+		v.collectors[name] = collector
+	}
+	if gauge, ok := collector.(*ConstGaugeCollector); ok {
+		return gauge, nil
+	}
+	return nil, fmt.Errorf("gauge %v collector requested, but %s %v collector exists", labelNames, collector.Type(), collector.LabelNames())
+}
+
+func (v *GroupedVault) CounterAdd(group string, name string, value float64, labels map[string]string) {
+	c, err := v.GetOrCreateCounterCollector(name, LabelNames(labels))
+	if err != nil {
+		log.Errorf("CounterAdd: %v", err)
+		return
+	}
+	c.Add(group, value, labels)
+}
+
+func (v *GroupedVault) GaugeSet(group string, name string, value float64, labels map[string]string) {
+	c, err := v.GetOrCreateGaugeCollector(name, LabelNames(labels))
+	if err != nil {
+		log.Errorf("CounterAdd: %v", err)
+		return
+	}
+	c.Set(group, value, labels)
+}
diff --git a/pkg/shell-operator/operator.go b/pkg/shell-operator/operator.go
index e389e84..4b0c5d5 100644
--- a/pkg/shell-operator/operator.go
+++ b/pkg/shell-operator/operator.go
@@ -18,6 +18,7 @@ import (
 	. "github.com/flant/shell-operator/pkg/hook/task_metadata"
 	. "github.com/flant/shell-operator/pkg/hook/types"
 	. "github.com/flant/shell-operator/pkg/kube_events_manager/types"
+	. "github.com/flant/shell-operator/pkg/utils/measure"
 
 	"github.com/flant/shell-operator/pkg/app"
 	"github.com/flant/shell-operator/pkg/debug"
@@ -34,6 +35,8 @@ import (
 	utils "github.com/flant/shell-operator/pkg/utils/labels"
 )
 
+var WaitQueuesTimeout = time.Second * 10
+
 type ShellOperator struct {
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -143,6 +146,11 @@ func (op *ShellOperator) Init() (err error) {
 		// Metric storage and live metrics
 		op.MetricStorage.Start()
 	}
+	// define buckets
+	op.MetricStorage.DefineHistogramBuckets("queue_hist", []float64{0.0, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0})                                        // 1 microsend to 1 millisecond
+	op.MetricStorage.DefineHistogramBuckets("hook_run_hist", []float64{0.0, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0, 5000.0, 10000.0})            // 1ms to 10s
+	op.MetricStorage.DefineHistogramBuckets("kube_jq_hist", []float64{0.0, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0, 5000.0, 10000.0})             // 1ms to 10s
+	op.MetricStorage.DefineHistogramBuckets("kube_event_duration_hist", []float64{0.0, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0, 5000.0, 10000.0}) // 1ms to 10s
 
 	if op.KubeClient == nil {
 		op.KubeClient = kube.NewKubernetesClient()
@@ -160,6 +168,7 @@ func (op *ShellOperator) Init() (err error) {
 	// Initialize the task queues set with the "main" queue.
 	op.TaskQueues = queue.NewTaskQueueSet()
 	op.TaskQueues.WithContext(op.ctx)
+	op.TaskQueues.WithMetricStorage(op.MetricStorage)
 
 	// Initialize schedule manager.
 	op.ScheduleManager = schedule_manager.NewScheduleManager()
@@ -169,6 +178,7 @@ func (op *ShellOperator) Init() (err error) {
 	op.KubeEventsManager = kube_events_manager.NewKubeEventsManager()
 	op.KubeEventsManager.WithKubeClient(op.KubeClient)
 	op.KubeEventsManager.WithContext(op.ctx)
+	op.KubeEventsManager.WithMetricStorage(op.MetricStorage)
 
 	// Initialize events handler that emit tasks to run hooks
 	op.ManagerEventsHandler = NewManagerEventsHandler()
@@ -277,107 +287,145 @@ func (op *ShellOperator) TaskHandler(t task.Task) queue.TaskResult {
 
 	switch t.GetType() {
 	case HookRun:
+		res = op.TaskHandleHookRun(t)
+
+	case EnableKubernetesBindings:
+		res = op.TaskHandleEnableKubernetesBindings(t)
+
+	case EnableScheduleBindings:
 		hookLogLabels := map[string]string{}
 		hookLogLabels["hook"] = hookMeta.HookName
-		hookLogLabels["binding"] = string(hookMeta.BindingType)
-		hookLogLabels["task"] = "HookRun"
-		hookLogLabels["queue"] = t.GetQueueName()
+		hookLogLabels["binding"] = string(Schedule)
+		hookLogLabels["task"] = "EnableScheduleBindings"
+		hookLogLabels["queue"] = "main"
+
 		taskLogEntry := logEntry.WithFields(utils.LabelsToLogFields(hookLogLabels))
-		taskLogEntry.Info("Execute hook")
 
 		taskHook := op.HookManager.GetHook(hookMeta.HookName)
-		if taskHook.Config.Version == "v1" {
-			bcs := op.CombineBindingContextForHook(op.TaskQueues.GetByName(t.GetQueueName()), t, nil)
-			if bcs != nil {
-				hookMeta.BindingContext = bcs
-				t.UpdateMetadata(hookMeta)
-			}
-		}
+		taskHook.HookController.EnableScheduleBindings()
+		taskLogEntry.Infof("Schedule binding for hook enabled successfully")
+		res.Status = "Success"
+	}
 
-		metrics, err := taskHook.Run(hookMeta.BindingType, hookMeta.BindingContext, hookLogLabels)
+	return res
+}
 
-		if err == nil {
-			err = op.MetricStorage.SendBatch(metrics, map[string]string{
-				"hook": hookMeta.HookName,
-			})
-		}
+func (op *ShellOperator) TaskHandleEnableKubernetesBindings(t task.Task) queue.TaskResult {
+	var hookMeta = HookMetadataAccessor(t)
 
-		if err != nil {
-			hookLabel := taskHook.SafeName()
-
-			if hookMeta.AllowFailure {
-				taskLogEntry.Infof("Hook failed, but allowed to fail: %v", err)
-				op.MetricStorage.SendCounter("shell_operator_hook_allowed_errors", 1.0, map[string]string{"hook": hookLabel})
-				res.Status = "Success"
-			} else {
-				op.MetricStorage.SendCounter("shell_operator_hook_errors", 1.0, map[string]string{"hook": hookLabel})
-				t.UpdateFailureMessage(err.Error())
-				taskLogEntry.Errorf("Hook failed. Will retry after delay. Failed count is %d. Error: %s", t.GetFailureCount()+1, err)
-				res.Status = "Fail"
-			}
-		} else {
-			taskLogEntry.Infof("Hook executed successfully")
-			res.Status = "Success"
-		}
+	metricLabels := map[string]string{
+		"hook":    hookMeta.HookName,
+		"binding": hookMeta.Binding,
+	}
+	defer MeasureTimeMs(func(ms float64) {
+		op.MetricStorage.SendGauge("enable_kubernetes_bindings_ms", ms, metricLabels)
+	})()
 
-	case EnableKubernetesBindings:
-		hookLogLabels := map[string]string{}
-		hookLogLabels["hook"] = hookMeta.HookName
-		hookLogLabels["binding"] = string(OnKubernetesEvent)
-		hookLogLabels["task"] = "EnableKubernetesBindings"
-		hookLogLabels["queue"] = "main"
+	var res queue.TaskResult
+	hookLogLabels := map[string]string{}
+	hookLogLabels["hook"] = hookMeta.HookName
+	hookLogLabels["binding"] = string(OnKubernetesEvent)
+	hookLogLabels["task"] = "EnableKubernetesBindings"
+	hookLogLabels["queue"] = "main"
 
-		taskLogEntry := logEntry.WithFields(utils.LabelsToLogFields(hookLogLabels))
+	taskLogEntry := log.WithFields(utils.LabelsToLogFields(hookLogLabels))
 
-		taskLogEntry.Info("Enable kubernetes binding for hook")
+	taskLogEntry.Info("Enable kubernetes binding for hook")
 
-		taskHook := op.HookManager.GetHook(hookMeta.HookName)
+	taskHook := op.HookManager.GetHook(hookMeta.HookName)
 
-		hookRunTasks := []task.Task{}
+	hookRunTasks := []task.Task{}
 
-		// Run hook for each binding with Synchronization binding context. Ignore queue name here, execute in main queue.
-		err := taskHook.HookController.HandleEnableKubernetesBindings(func(info controller.BindingExecutionInfo) {
-			newTask := task.NewTask(HookRun).
-				WithMetadata(HookMetadata{
-					HookName:       taskHook.Name,
-					BindingType:    OnKubernetesEvent,
-					BindingContext: info.BindingContext,
-					AllowFailure:   info.AllowFailure,
-					Binding:        info.Binding,
-					Group:          info.Group,
-				}).
-				WithLogLabels(hookLogLabels).
-				WithQueueName("main")
-			hookRunTasks = append(hookRunTasks, newTask)
+	// Run hook for each binding with Synchronization binding context. Ignore queue name here, execute in main queue.
+	err := taskHook.HookController.HandleEnableKubernetesBindings(func(info controller.BindingExecutionInfo) {
+		newTask := task.NewTask(HookRun).
+			WithMetadata(HookMetadata{
+				HookName:       taskHook.Name,
+				BindingType:    OnKubernetesEvent,
+				BindingContext: info.BindingContext,
+				AllowFailure:   info.AllowFailure,
+				Binding:        info.Binding,
+				Group:          info.Group,
+			}).
+			WithLogLabels(hookLogLabels).
+			WithQueueName("main")
+		hookRunTasks = append(hookRunTasks, newTask)
+	})
+
+	if err != nil {
+		hookLabel := taskHook.SafeName()
+		op.MetricStorage.SendCounter("hook_errors", 1.0, map[string]string{"hook": hookLabel})
+		t.UpdateFailureMessage(err.Error())
+		taskLogEntry.Errorf("Enable Kubernetes binding for hook failed. Will retry after delay. Failed count is %d. Error: %s", t.GetFailureCount()+1, err)
+		res.Status = "Fail"
+	} else {
+		// return Synchronization tasks to add to the queue head.
+		// Informers can be started now — their events will
+		// be added to the tail of the main queue or to the other queues.
+		op.HookManager.GetHook(hookMeta.HookName).HookController.StartMonitors()
+		taskLogEntry.Infof("Kubernetes binding for hook enabled successfully")
+		res.Status = "Success"
+		res.HeadTasks = hookRunTasks
+	}
+
+	return res
+}
+
+func (op *ShellOperator) TaskHandleHookRun(t task.Task) queue.TaskResult {
+	var hookMeta = HookMetadataAccessor(t)
+
+	metricLabels := map[string]string{
+		"hook":    hookMeta.HookName,
+		"binding": string(hookMeta.BindingType),
+		"queue":   t.GetQueueName(),
+	}
+	defer MeasureTimeMs(func(ms float64) {
+		op.MetricStorage.ObserveHistogram("hook_run_hist", ms, metricLabels)
+	})()
+
+	var res queue.TaskResult
+
+	hookLogLabels := map[string]string{}
+	hookLogLabels["hook"] = hookMeta.HookName
+	hookLogLabels["binding"] = string(hookMeta.BindingType)
+	hookLogLabels["task"] = "HookRun"
+	hookLogLabels["queue"] = t.GetQueueName()
+
+	taskLogEntry := log.WithFields(utils.LabelsToLogFields(hookLogLabels))
+	taskLogEntry.Info("Execute hook")
+
+	taskHook := op.HookManager.GetHook(hookMeta.HookName)
+	if taskHook.Config.Version == "v1" {
+		bcs := op.CombineBindingContextForHook(op.TaskQueues.GetByName(t.GetQueueName()), t, nil)
+		if bcs != nil {
+			hookMeta.BindingContext = bcs
+			t.UpdateMetadata(hookMeta)
+		}
+	}
+
+	metrics, err := taskHook.Run(hookMeta.BindingType, hookMeta.BindingContext, hookLogLabels)
+
+	if err == nil {
+		err = op.MetricStorage.SendBatch(metrics, map[string]string{
+			"hook": hookMeta.HookName,
 		})
+	}
 
-		if err != nil {
-			hookLabel := taskHook.SafeName()
-			op.MetricStorage.SendCounter("shell_operator_hook_errors", 1.0, map[string]string{"hook": hookLabel})
+	if err != nil {
+		hookLabel := taskHook.SafeName()
+
+		if hookMeta.AllowFailure {
+			taskLogEntry.Infof("Hook failed, but allowed to fail: %v", err)
+			op.MetricStorage.SendCounter("hook_allowed_errors", 1.0, map[string]string{"hook": hookLabel})
+			res.Status = "Success"
+		} else {
+			op.MetricStorage.SendCounter("hook_errors", 1.0, map[string]string{"hook": hookLabel})
 			t.UpdateFailureMessage(err.Error())
-			taskLogEntry.Errorf("Enable Kubernetes binding for hook failed. Will retry after delay. Failed count is %d. Error: %s", t.GetFailureCount()+1, err)
+			taskLogEntry.Errorf("Hook failed. Will retry after delay. Failed count is %d. Error: %s", t.GetFailureCount()+1, err)
 			res.Status = "Fail"
-		} else {
-			// return Synchronization tasks to add to the queue head.
-			// Informers can be started now — their events will
-			// be added to the tail of the main queue or to the other queues.
-			op.HookManager.GetHook(hookMeta.HookName).HookController.StartMonitors()
-			taskLogEntry.Infof("Kubernetes binding for hook enabled successfully")
-			res.Status = "Success"
-			res.HeadTasks = hookRunTasks
 		}
-	case EnableScheduleBindings:
-		hookLogLabels := map[string]string{}
-		hookLogLabels["hook"] = hookMeta.HookName
-		hookLogLabels["binding"] = string(Schedule)
-		hookLogLabels["task"] = "EnableScheduleBindings"
-		hookLogLabels["queue"] = "main"
-
-		taskLogEntry := logEntry.WithFields(utils.LabelsToLogFields(hookLogLabels))
-
-		taskHook := op.HookManager.GetHook(hookMeta.HookName)
-		taskHook.HookController.EnableScheduleBindings()
-		taskLogEntry.Infof("Schedule binding for hook enabled successfully")
+	} else {
+		taskLogEntry.Infof("Hook executed successfully")
 		res.Status = "Success"
 	}
 
@@ -567,7 +615,7 @@ func (op *ShellOperator) RunMetrics() {
 	// live ticks.
 	go func() {
 		for {
-			op.MetricStorage.SendCounter("shell_operator_live_ticks", 1.0, map[string]string{})
+			op.MetricStorage.SendCounter("live_ticks", 1.0, map[string]string{})
 			time.Sleep(10 * time.Second)
 		}
 	}()
@@ -577,7 +625,7 @@ func (op *ShellOperator) RunMetrics() {
 		for {
 			op.TaskQueues.Iterate(func(queue *queue.TaskQueue) {
 				queueLen := float64(queue.Length())
-				op.MetricStorage.SendGauge("shell_operator_tasks_queue_length", queueLen, map[string]string{"queue": queue.Name})
+				op.MetricStorage.SendGauge("tasks_queue_length", queueLen, map[string]string{"queue": queue.Name})
 			})
 			time.Sleep(5 * time.Second)
 		}
@@ -665,3 +713,12 @@ func InitAndStart(operator *ShellOperator) error {
 
 	return nil
 }
+
+// Shutdown pause kubernetes events handling and stop queues. Wait for queues to stop.
+func (op *ShellOperator) Shutdown() {
+	op.ScheduleManager.Stop()
+	op.KubeEventsManager.PauseHandleEvents()
+	op.TaskQueues.Stop()
+	// Wait for queues to stop, but no more than 10 seconds
+	op.TaskQueues.WaitStopWithTimeout(WaitQueuesTimeout)
+}
diff --git a/pkg/task/queue/queue_set.go b/pkg/task/queue/queue_set.go
index 24e95f1..6bc37df 100644
--- a/pkg/task/queue/queue_set.go
+++ b/pkg/task/queue/queue_set.go
@@ -3,7 +3,9 @@ package queue
 import (
 	"context"
 	"sync"
+	"time"
 
+	"github.com/flant/shell-operator/pkg/metrics_storage"
 	"github.com/flant/shell-operator/pkg/task"
 )
 
@@ -12,6 +14,8 @@ type TaskQueueSet struct {
 	Queues   map[string]*TaskQueue
 	MainName string
 
+	metricStorage *metrics_storage.MetricStorage
+
 	m      sync.Mutex
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -32,6 +36,10 @@ func (tqs *TaskQueueSet) WithContext(ctx context.Context) {
 	tqs.ctx, tqs.cancel = context.WithCancel(ctx)
 }
 
+func (tqs *TaskQueueSet) WithMetricStorage(mstor *metrics_storage.MetricStorage) {
+	tqs.metricStorage = mstor
+}
+
 func (tqs *TaskQueueSet) Stop() {
 	if tqs.cancel != nil {
 		tqs.cancel()
@@ -57,6 +65,7 @@ func (tqs *TaskQueueSet) NewNamedQueue(name string, handler func(task.Task) Task
 	q.WithName(name)
 	q.WithHandler(handler)
 	q.WithContext(tqs.ctx)
+	q.WithMetricStorage(tqs.metricStorage)
 	tqs.Queues[name] = q
 }
 
@@ -119,3 +128,28 @@ func (tqs *TaskQueueSet) Remove(name string) {
 	defer tqs.m.Unlock()
 	delete(tqs.Queues, name)
 }
+
+func (tqs *TaskQueueSet) WaitStopWithTimeout(timeout time.Duration) {
+	checkTick := time.NewTicker(time.Millisecond * 100)
+	defer checkTick.Stop()
+	timeoutTick := time.NewTicker(timeout)
+	defer timeoutTick.Stop()
+
+	for {
+		select {
+		case <-checkTick.C:
+			stopped := true
+			for _, q := range tqs.Queues {
+				if q.Status != "stop" {
+					stopped = false
+					break
+				}
+			}
+			if stopped {
+				return
+			}
+		case <-timeoutTick.C:
+			return
+		}
+	}
+}
diff --git a/pkg/task/queue/task_queue.go b/pkg/task/queue/task_queue.go
index 96fadd5..16194a1 100644
--- a/pkg/task/queue/task_queue.go
+++ b/pkg/task/queue/task_queue.go
@@ -9,7 +9,9 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
+	"github.com/flant/shell-operator/pkg/metrics_storage"
 	"github.com/flant/shell-operator/pkg/task"
+	. "github.com/flant/shell-operator/pkg/utils/measure"
 )
 
 /*
@@ -29,10 +31,6 @@ var (
 	DelayOnRepeat       = 25 * time.Millisecond
 )
 
-type QueueWatcher interface {
-	QueueChangeCallback()
-}
-
 type TaskResult struct {
 	Status     string
 	HeadTasks  []task.Task
@@ -45,14 +43,13 @@ type TaskResult struct {
 }
 
 type TaskQueue struct {
-	m              sync.Mutex
-	items          []task.Task
-	changesEnabled bool           // turns on and off callbacks execution
-	changesCount   int            // counts changes when callbacks turned off
-	queueWatchers  []QueueWatcher // callbacks to be executed on items change
-	ctx            context.Context
-	cancel         context.CancelFunc
-	started        bool // indicator to ignore multiple starts
+	m             sync.Mutex
+	metricStorage *metrics_storage.MetricStorage
+	ctx           context.Context
+	cancel        context.CancelFunc
+
+	items   []task.Task
+	started bool // a flag to ignore multiple starts
 
 	Name     string
 	Handler  func(task.Task) TaskResult
@@ -62,17 +59,42 @@ type TaskQueue struct {
 
 func NewTasksQueue() *TaskQueue {
 	return &TaskQueue{
-		m:              sync.Mutex{},
-		items:          make([]task.Task, 0),
-		changesCount:   0,
-		changesEnabled: false,
-		queueWatchers:  make([]QueueWatcher, 0),
-
+		m:        sync.Mutex{},
+		items:    make([]task.Task, 0),
 		HeadLock: sync.Mutex{},
 	}
 }
 
+func (q *TaskQueue) WithContext(ctx context.Context) {
+	q.ctx, q.cancel = context.WithCancel(ctx)
+}
+
+func (q *TaskQueue) WithMetricStorage(mstor *metrics_storage.MetricStorage) {
+	q.metricStorage = mstor
+}
+
+func (tq *TaskQueue) WithName(name string) *TaskQueue {
+	tq.Name = name
+	return tq
+}
+
+func (tq *TaskQueue) WithHandler(fn func(task.Task) TaskResult) *TaskQueue {
+	tq.Handler = fn
+	return tq
+}
+
+// MeasureActionTime is a helper to measure execution time of queue's actions
+func (q *TaskQueue) MeasureActionTime(action string) func() {
+	if q.metricStorage == nil {
+		return func() {}
+	}
+	return MeasureTime(func(nanos Nanos) {
+		q.metricStorage.ObserveHistogram("queue_hist", nanos.Micro(), map[string]string{"queue_name": q.Name, "queue_action": action})
+	})
+}
+
 func (q *TaskQueue) IsEmpty() bool {
+	defer q.MeasureActionTime("IsEmpty")()
 	q.m.Lock()
 	defer q.m.Unlock()
 	return q.isEmpty()
@@ -83,6 +105,7 @@ func (q *TaskQueue) isEmpty() bool {
 }
 
 func (q *TaskQueue) Length() int {
+	defer q.MeasureActionTime("Length")()
 	q.m.Lock()
 	defer q.m.Unlock()
 	return len(q.items)
@@ -90,6 +113,7 @@ func (q *TaskQueue) Length() int {
 
 // AddFirst adds new head element.
 func (q *TaskQueue) AddFirst(t task.Task) {
+	defer q.MeasureActionTime("AddFirst")()
 	q.m.Lock()
 	q.items = append([]task.Task{t}, q.items...)
 	q.m.Unlock()
@@ -97,6 +121,8 @@ func (q *TaskQueue) AddFirst(t task.Task) {
 
 // RemoveFirst deletes a head element, so head is moved.
 func (q *TaskQueue) RemoveFirst() (t task.Task) {
+	defer q.MeasureActionTime("RemoveFirst")()
+
 	q.m.Lock()
 	if q.isEmpty() {
 		q.m.Unlock()
@@ -110,6 +136,7 @@ func (q *TaskQueue) RemoveFirst() (t task.Task) {
 
 // GetFirst returns a head element.
 func (q *TaskQueue) GetFirst() task.Task {
+	defer q.MeasureActionTime("GetFirst")()
 	q.m.Lock()
 	defer q.m.Unlock()
 	if q.isEmpty() {
@@ -120,6 +147,7 @@ func (q *TaskQueue) GetFirst() task.Task {
 
 // AddFirst adds new tail element.
 func (q *TaskQueue) AddLast(task task.Task) {
+	defer q.MeasureActionTime("AddLast")()
 	q.m.Lock()
 	q.items = append(q.items, task)
 	q.m.Unlock()
@@ -127,6 +155,7 @@ func (q *TaskQueue) AddLast(task task.Task) {
 
 // RemoveLast deletes a tail element, so tail is moved.
 func (q *TaskQueue) RemoveLast() (t task.Task) {
+	defer q.MeasureActionTime("RemoveLast")()
 	q.m.Lock()
 	if q.isEmpty() {
 		q.m.Unlock()
@@ -144,6 +173,7 @@ func (q *TaskQueue) RemoveLast() (t task.Task) {
 
 // GetLast returns a tail element.
 func (q *TaskQueue) GetLast() task.Task {
+	defer q.MeasureActionTime("GetLast")()
 	q.m.Lock()
 	defer q.m.Unlock()
 	if q.isEmpty() {
@@ -154,6 +184,7 @@ func (q *TaskQueue) GetLast() task.Task {
 
 // Get returns a task by id.
 func (q *TaskQueue) Get(id string) task.Task {
+	defer q.MeasureActionTime("Get")()
 	q.m.Lock()
 	defer q.m.Unlock()
 	for _, t := range q.items {
@@ -166,6 +197,8 @@ func (q *TaskQueue) Get(id string) task.Task {
 
 // AddAfter inserts a task after the task with specified id.
 func (q *TaskQueue) AddAfter(id string, newTask task.Task) {
+	defer q.MeasureActionTime("AddAfter")()
+
 	newItems := make([]task.Task, len(q.items)+1)
 
 	idFound := false
@@ -189,6 +222,8 @@ func (q *TaskQueue) AddAfter(id string, newTask task.Task) {
 
 // AddBefore inserts a task before the task with specified id.
 func (q *TaskQueue) AddBefore(id string, newTask task.Task) {
+	defer q.MeasureActionTime("AddBefore")()
+
 	newItems := make([]task.Task, len(q.items)+1)
 
 	idFound := false
@@ -215,6 +250,8 @@ func (q *TaskQueue) AddBefore(id string, newTask task.Task) {
 
 // Remove finds element by id and deletes it.
 func (q *TaskQueue) Remove(id string) (t task.Task) {
+	defer q.MeasureActionTime("Remove")()
+
 	q.m.Lock()
 	defer q.m.Unlock()
 	delId := -1
@@ -232,28 +269,15 @@ func (q *TaskQueue) Remove(id string) (t task.Task) {
 	return t
 }
 
-func (tq *TaskQueue) WithName(name string) *TaskQueue {
-	tq.Name = name
-	return tq
-}
-
-func (tq *TaskQueue) WithHandler(fn func(task.Task) TaskResult) *TaskQueue {
-	tq.Handler = fn
-	return tq
-}
-
-func (tq *TaskQueue) DoWithHeadLock(fn func(tasksQueue *TaskQueue)) {
-	tq.HeadLock.Lock()
-	defer tq.HeadLock.Unlock()
+func (q *TaskQueue) DoWithHeadLock(fn func(tasksQueue *TaskQueue)) {
+	defer q.MeasureActionTime("DoWithHeadLock")()
+	q.HeadLock.Lock()
+	defer q.HeadLock.Unlock()
 	if fn != nil {
-		fn(tq)
+		fn(q)
 	}
 }
 
-func (q *TaskQueue) WithContext(ctx context.Context) {
-	q.ctx, q.cancel = context.WithCancel(ctx)
-}
-
 func (q *TaskQueue) Stop() {
 	if q.cancel != nil {
 		q.cancel()
@@ -272,8 +296,7 @@ func (q *TaskQueue) Start() {
 			var t = q.waitForTask(sleepDelay)
 			if t == nil {
 				q.Status = "stop"
-				log.Debugf("queue %s: got nil task, stop queue", q.Name)
-				q.started = false
+				log.Infof("queue '%s' stopped", q.Name)
 				return
 			}
 
@@ -289,6 +312,15 @@ func (q *TaskQueue) Start() {
 			q.Status = "run first task"
 			taskRes := q.Handler(t)
 
+			// Check Done channel after long running operation.
+			select {
+			case <-q.ctx.Done():
+				log.Infof("queue '%s' stopped after task handling", q.Name)
+				q.Status = "stop"
+				return
+			default:
+			}
+
 			switch taskRes.Status {
 			case "Fail":
 				t.IncrementFailureCount()
@@ -340,6 +372,13 @@ func (q *TaskQueue) Start() {
 // sleepDelay is used to sleep before check a task, e.g. in case of failed previous task.
 // If queue is empty, than it will be checked every DelayOnQueueIsEmpty.
 func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task {
+	// Check Done channel to be able to stop queue
+	select {
+	case <-q.ctx.Done():
+		return nil
+	default:
+	}
+
 	// Wait for non empty queue or closed Done channel
 	origStatus := q.Status
 	waitBegin := time.Now()
@@ -389,6 +428,9 @@ func (q *TaskQueue) Iterate(doFn func(task.Task)) {
 	if doFn == nil {
 		return
 	}
+
+	defer q.MeasureActionTime("Iterate")()
+
 	q.m.Lock()
 	defer q.m.Unlock()
 	for _, t := range q.items {
@@ -401,6 +443,9 @@ func (q *TaskQueue) Filter(filterFn func(task.Task) bool) {
 	if filterFn == nil {
 		return
 	}
+
+	defer q.MeasureActionTime("Filter")()
+
 	q.m.Lock()
 	defer q.m.Unlock()
 	var newItems = make([]task.Task, 0)
diff --git a/pkg/utils/labels/merge_labels.go b/pkg/utils/labels/labels.go
similarity index 58%
rename from pkg/utils/labels/merge_labels.go
rename to pkg/utils/labels/labels.go
index e0a9172..6da4ade 100644
--- a/pkg/utils/labels/merge_labels.go
+++ b/pkg/utils/labels/labels.go
@@ -2,6 +2,7 @@ package utils
 
 import (
 	log "github.com/sirupsen/logrus"
+	"sort"
 )
 
 // MergeLabels merges several maps into one. Last map keys overrides keys from first maps.
@@ -26,3 +27,21 @@ func LabelsToLogFields(labelsMaps ...map[string]string) log.Fields {
 	}
 	return fields
 }
+
+// LabelNames returns sorted label keys
+func LabelNames(labels map[string]string) []string {
+	names := make([]string, 0)
+	for labelName := range labels {
+		names = append(names, labelName)
+	}
+	sort.Strings(names)
+	return names
+}
+
+func LabelValues(labels map[string]string, labelNames []string) []string {
+	var values = make([]string, 0)
+	for _, name := range labelNames {
+		values = append(values, labels[name])
+	}
+	return values
+}
diff --git a/pkg/utils/measure/measure.go b/pkg/utils/measure/measure.go
new file mode 100644
index 0000000..048e389
--- /dev/null
+++ b/pkg/utils/measure/measure.go
@@ -0,0 +1,49 @@
+package measure
+
+import "time"
+
+// MeasureTimeMs is a helper to measure execution time in whole number of milliseconds
+func MeasureTimeMs(resultCallback func(ms float64)) func() {
+	return MeasureTime(func(nanos Nanos) {
+		if resultCallback != nil {
+			resultCallback(nanos.Ms())
+		}
+	})
+}
+
+// MeasureTime returns a function that can be deferred to measure an execution time of a method call
+func MeasureTime(resultCallback func(nanos Nanos)) func() {
+	start := time.Now()
+	return func() {
+		if resultCallback != nil {
+			resultCallback(Nanos(time.Since(start)))
+		}
+	}
+}
+
+type Nanos int64
+
+// Micro is a whole number of microseconds
+func (n Nanos) Micro() float64 {
+	return float64(n / 1e3)
+}
+
+// Ms is a whole number of milliseconds
+func (n Nanos) Ms() float64 {
+	return float64(n / 1e6)
+}
+
+// Ms2 is a milliseconds with 2 fractional digits
+func (n Nanos) Ms2() float64 {
+	return float64(n/1e4) / 100
+}
+
+// Sec is a whole number of seconds
+func (n Nanos) Sec() float64 {
+	return float64(n / 1e9)
+}
+
+// Sec2 is a seconds with 2 fractional digits
+func (n Nanos) Sec2() float64 {
+	return float64(n/1e7) / 100
+}


Test patch:
diff --git a/pkg/metrics_storage/vault/vault_test.go b/pkg/metrics_storage/vault/vault_test.go
new file mode 100644
index 0000000..36861ff
--- /dev/null
+++ b/pkg/metrics_storage/vault/vault_test.go
@@ -0,0 +1,119 @@
+package vault
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+
+	. "github.com/onsi/gomega"
+
+	"github.com/prometheus/client_golang/prometheus"
+	promtest "github.com/prometheus/client_golang/prometheus/testutil"
+	log "github.com/sirupsen/logrus"
+)
+
+func Test_CounterAdd(t *testing.T) {
+	g := NewWithT(t)
+
+	buf := &bytes.Buffer{}
+	log.SetOutput(buf)
+
+	v := NewGroupedVault()
+
+	v.CounterAdd("group1", "metric_total", 1.0, map[string]string{"lbl": "val"})
+
+	g.Expect(buf.String()).ShouldNot(ContainSubstring("error"), "error occurred in log: %s", buf.String())
+
+	expect := `
+# HELP metric_total metric_total
+# TYPE metric_total counter
+metric_total{lbl="val"} 1
+`
+	err := promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expect), "metric_total")
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	v.ClearMissingMetrics("group1", map[string][]map[string]string{
+		"metric_total": {
+			{"lbl": "qwe"},
+		},
+	})
+
+	g.Expect(buf.String()).ShouldNot(ContainSubstring("error"), "error occurred in log: %s", buf.String())
+
+	// Expect no metric with lbl="val"
+	expect = ``
+	err = promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expect), "metric_total")
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	v.CounterAdd("group1", "metric_total", 1.0, map[string]string{"lbl": "val2"})
+
+	g.Expect(buf.String()).ShouldNot(ContainSubstring("error"), "error occurred in log: %s", buf.String())
+
+	// Expect metric_total with new label value
+	expect = `
+# HELP metric_total metric_total
+# TYPE metric_total counter
+metric_total{lbl="val2"} 1
+`
+	err = promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expect), "metric_total")
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	v.CounterAdd("group1", "metric_total", 1.0, map[string]string{"lbl": "val2"})
+	v.CounterAdd("group1", "metric_total", 1.0, map[string]string{"lbl": "val222"})
+
+	g.Expect(buf.String()).ShouldNot(ContainSubstring("error"), "error occurred in log: %s", buf.String())
+
+	// Expect metric_total with 2 label values
+	expect = `
+# HELP metric_total metric_total
+# TYPE metric_total counter
+metric_total{lbl="val2"} 2
+metric_total{lbl="val222"} 1
+`
+	err = promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expect), "metric_total")
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	v.ClearMissingMetrics("group1", map[string][]map[string]string{
+		"metric_total": {
+			{"lbl": "val"},
+		},
+		"metric2_total": {
+			{"lbl222": "val222"},
+		},
+	})
+	v.CounterAdd("group1", "metric_total", 1.0, map[string]string{"lbl": "val"})
+	v.CounterAdd("group2", "metric2_total", 1.0, map[string]string{"lbl": "val222"})
+
+	g.Expect(buf.String()).ShouldNot(ContainSubstring("error"), "error occurred in log: %s", buf.String())
+	// Expect metric_total is updated and metric2_total
+	expect = `
+# HELP metric_total metric_total
+# TYPE metric_total counter
+metric_total{lbl="val"} 1
+# HELP metric2_total metric2_total
+# TYPE metric2_total counter
+metric2_total{lbl="val222"} 1
+`
+	err = promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expect), "metric_total", "metric2_total")
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	v.ClearMissingMetrics("group1", map[string][]map[string]string{
+		"metric2_total": {
+			{"lbl222": "val222"},
+		},
+	})
+	v.CounterAdd("group2", "metric2_total", 1.0, map[string]string{"lbl": "val222"})
+	g.Expect(buf.String()).ShouldNot(ContainSubstring("error"), "error occurred in log: %s", buf.String())
+	// Expect metric_total is updated and metric2_total is updated and metric_total left as is
+	expect = `
+# HELP metric_total metric_total
+# TYPE metric_total counter
+metric_total{lbl="val"} 1
+# HELP metric2_total metric2_total
+# TYPE metric2_total counter
+metric2_total{lbl="val222"} 2
+`
+	err = promtest.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expect), "metric_total", "metric2_total")
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+}