Merge pull request #165 from flant/feat_metrics_from_hooks

feat: allow to send metrics from hooks

Author: diafour

METRICS.md

@@ -8,3 +8,28 @@ Shell-operator exports Prometheus metrics to the `/metrics` path. The default po
 * `shell_operator_hook_allowed_errors{hook="hook-name"}` – this is the counter of hooks’ execution errors. It only tracks errors of hooks that are allowed to exit with an error (the parameter `allowFailure: true` is set in the configuration). The metric has a “hook” label with the name of a failed hook.
 * `shell_operator_tasks_queue_length` – a gauge showing the length of the working queue. This metric can be used to warn about stuck hooks. It has no labels.
 * `shell_operator_live_ticks` – a counter that increases every 10 seconds. This metric can be used for alerting about an unhealthy Shell-operator. It has no labels.
+
+## Custom metrics
+
+Hooks can export metrics by writing a set of operation on JSON format into $METRICS_PATH file.
+
+Operation to increase a counter:
+
+```json
+{"name":"metric_name","add":1,"labels":{"label1":"value1"}}
+```
+
+Operation to set a value for a gauge:
+
+```json
+{"name":"metric_name","set":33,"labels":{"label1":"value1"}}
+```
+
+Labels are not required, but Shell-operator adds a `hook` label.
+
+Several metrics can be expored at once. For example, this script will create 2 metrics:
+
+```
+echo '{"name":"hook_metric_count","add":1,"labels":{"label1":"value1"}}' >> $METRICS_PATH
+echo '{"name":"hook_metrics_items","add":1,"labels":{"label1":"value1"}}' >> $METRICS_PATH
+```

go.sum

@@ -99,6 +99,7 @@ github.com/flant/go-openapi-validate v0.19.4-0.20200313141509-0c0fba4d39e1/go.mo
 github.com/flant/libjq-go v1.0.1-0.20200205115921-27e93c18c17f h1:3tmztWJjf61sHfHYLSMi5TDdz5jtmcVqe43PSwsxNvE=
 github.com/flant/libjq-go v1.0.1-0.20200205115921-27e93c18c17f/go.mod h1:+SYqi5wsNjtQVlkPg0Ep5IOuN+ydg79Jo/gk4/PuS8c=
 github.com/flant/libjq-go v1.6.1-0.20200331115542-04a1a2e80daa/go.mod h1:+SYqi5wsNjtQVlkPg0Ep5IOuN+ydg79Jo/gk4/PuS8c=
+github.com/flant/libjq-go v1.6.1-0.20200401092614-198670408da1 h1:pOPBJDB7PZz/SKa13mlR3bvkRJ0KWKRe6v+KZOObkKw=
 github.com/flant/libjq-go v1.6.1-0.20200401092614-198670408da1/go.mod h1:+SYqi5wsNjtQVlkPg0Ep5IOuN+ydg79Jo/gk4/PuS8c=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=

pkg/hook/hook.go

@@ -17,6 +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"
 )
 
 type CommonHook interface {
@@ -66,38 +67,50 @@ func (h *Hook) GetHookController() controller.HookController {
 	return h.HookController
 }
 
-func (h *Hook) Run(bindingType BindingType, context []BindingContext, logLabels map[string]string) error {
+func (h *Hook) Run(bindingType BindingType, context []BindingContext, logLabels map[string]string) ([]metrics_storage.MetricOperation, error) {
 	// Refresh snapshots
 	freshBindingContext := h.HookController.UpdateSnapshots(context)
 
 	versionedContextList := ConvertBindingContextList(h.Config.Version, freshBindingContext)
 
 	contextPath, err := h.prepareBindingContextJsonFile(versionedContextList)
 	if err != nil {
-		return err
+		return nil, err
 	}
+
+	metricsPath, err := h.prepareMetricsFile()
+	if err != nil {
+		return nil, err
+	}
+
 	// remove tmp file on hook exit
 	defer func() {
-		if app.DebugKeepTmpFiles == "yes" {
-			return
+		if app.DebugKeepTmpFiles != "yes" {
+			os.Remove(contextPath)
+			os.Remove(metricsPath)
 		}
-		os.Remove(contextPath)
 	}()
 
 	envs := []string{}
 	envs = append(envs, os.Environ()...)
 	if contextPath != "" {
 		envs = append(envs, fmt.Sprintf("BINDING_CONTEXT_PATH=%s", contextPath))
+		envs = append(envs, fmt.Sprintf("METRICS_PATH=%s", metricsPath))
 	}
 
 	hookCmd := executor.MakeCommand(path.Dir(h.Path), h.Path, []string{}, envs)
 
 	err = executor.RunAndLogLines(hookCmd, logLabels)
 	if err != nil {
-		return fmt.Errorf("%s FAILED: %s", h.Name, err)
+		return nil, fmt.Errorf("%s FAILED: %s", h.Name, err)
+	}
+
+	metrics, err := metrics_storage.MetricOperationsFromFile(metricsPath)
+	if err != nil {
+		return nil, fmt.Errorf("got bad metrics: %s", err)
 	}
 
-	return nil
+	return metrics, nil
 }
 
 func (h *Hook) SafeName() string {
@@ -150,3 +163,14 @@ func (h *Hook) prepareBindingContextJsonFile(context BindingContextList) (string
 
 	return bindingContextPath, nil
 }
+
+func (h *Hook) prepareMetricsFile() (string, error) {
+	metricsPath := filepath.Join(h.TmpDir, fmt.Sprintf("hook-%s-metrics-%s.json", h.SafeName(), uuid.NewV4().String()))
+
+	err := ioutil.WriteFile(metricsPath, []byte{}, 0644)
+	if err != nil {
+		return "", err
+	}
+
+	return metricsPath, nil
+}

pkg/metrics_storage/metric_operation.go

@@ -0,0 +1,50 @@
+package metrics_storage
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+)
+
+type MetricOperation struct {
+	Name   string            `json:"name"`
+	Add    *float64          `json:"add,omitempty"`
+	Set    *float64          `json:"set,omitempty"`
+	Labels map[string]string `json:"labels"`
+}
+
+func MetricOperationsFromReader(r io.Reader) ([]MetricOperation, error) {
+	var operations = make([]MetricOperation, 0)
+
+	dec := json.NewDecoder(r)
+	for {
+		var metricOperation MetricOperation
+		if err := dec.Decode(&metricOperation); err == io.EOF {
+			break
+		} else if err != nil {
+			return nil, err
+		}
+
+		operations = append(operations, metricOperation)
+	}
+
+	return operations, nil
+}
+
+func MetricOperationsFromBytes(data []byte) ([]MetricOperation, error) {
+	return MetricOperationsFromReader(bytes.NewReader(data))
+}
+
+func MetricOperationsFromFile(filePath string) ([]MetricOperation, error) {
+	data, err := ioutil.ReadFile(filePath)
+	if err != nil {
+		return nil, fmt.Errorf("cannot read %s: %s", filePath, err)
+	}
+
+	if len(data) == 0 {
+		return nil, nil
+	}
+	return MetricOperationsFromBytes(data)
+}

pkg/metrics_storage/metrics_storage.go

@@ -2,7 +2,9 @@ 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"
 )
@@ -38,31 +40,49 @@ func (m *MetricStorage) Stop() {
 	}
 }
 
-func (storage *MetricStorage) Start() {
+func (m *MetricStorage) Start() {
 	go func() {
 		for {
 			select {
-			case metric := <-storage.MetricChan:
-				metric.store(storage)
-			case <-storage.ctx.Done():
+			case metric := <-m.MetricChan:
+				metric.store(m)
+			case <-m.ctx.Done():
 				return
 			}
 		}
 	}()
 }
 
-func (storage *MetricStorage) SendGauge(metric string, value float64, labels map[string]string) {
-	storage.MetricChan <- NewGaugeMetric(storage.Prefix+metric, value, labels)
+func (m *MetricStorage) SendGauge(metric string, value float64, labels map[string]string) {
+	m.MetricChan <- NewGaugeMetric(m.Prefix+metric, value, labels)
 }
-func (storage *MetricStorage) SendCounter(metric string, value float64, labels map[string]string) {
-	storage.MetricChan <- NewCounterMetric(storage.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 (storage *MetricStorage) SendGaugeNoPrefix(metric string, value float64, labels map[string]string) {
-	storage.MetricChan <- NewGaugeMetric(metric, value, labels)
+func (m *MetricStorage) SendGaugeNoPrefix(metric string, value float64, labels map[string]string) {
+	m.MetricChan <- NewGaugeMetric(metric, value, labels)
 }
-func (storage *MetricStorage) SendCounterNoPrefix(metric string, value float64, labels map[string]string) {
-	storage.MetricChan <- NewCounterMetric(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 {

pkg/shell-operator/operator.go

@@ -294,7 +294,14 @@ func (op *ShellOperator) TaskHandler(t task.Task) queue.TaskResult {
 			}
 		}
 
-		err := taskHook.Run(hookMeta.BindingType, hookMeta.BindingContext, hookLogLabels)
+		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()