Repository: fluxcd/flux
Pull request #2535 (2019-10-21 21:02:40)

Create metric for flux manifest errors
https://github.com/fluxcd/flux/issues/2199



stefanprodan: I think the metric should be named `sync_error` and it should act like a boolean, values can be 0 or 1. 

Labels:
- `fetch` (records git errors)
- `generate` (records .flux.yaml generators errors)
- `parse` (records manifestsStore errors)
- `apply` (records kubectl apply errors)

With this metric and labels one can compose alerts with `avg_over_time(flux_daemon_sync_error [5m]) > 0` and compose messages like `flux sync apply failed` or `flux sync parse falied`.
squaremo: It seems like the main thrust of this PR is to be able to know the number of manifests that failed to apply, so you can alert on manifests failing to apply.

There's a metric which record how long each sync takes, and whether it was successful or not -- `flux_demon_sync_duration_seconds` -- but that might be too broad for your purposes, since it will label a sample as unsuccessful if any error came up while syncing, e.g., it failed to post a webhook.

So I think there's room another metric to record specifically the application step. A gauge counting how many things were applied is appropriate -- I would record how many were attempted, and label according to whether they were successful applied or not.

Tangentially for your purpose, a metric timing (and labelling the success) of each phase of the syncing might be useful too. E.g., generating the manifests, parsing the manifests, doing the application, doing the garbage collection.
squaremo: > I think the metric should be named `sync_error` and it should act like a boolean

I'm not a fan of metrics that are there just so you can write a very specific alert as simply as possible, but provide no other information. What if you want to alert on the _proportion_ of apply errors to successes? Or an alert on whether the number of manifests being applied is not what you expected -- a mistake in configuration could mean you suddenly have no manifests, and no errors.

mpashka: Main idea of initial request - https://github.com/fluxcd/flux/issues/2199 - is to provide possibility to get information about current flux state - is everything is in sync and works as expected or something can't be synced and manual intervention is needed. flux_demon_sync_duration_seconds doesn't give flux current state information.
I suppose metric is proper way to get flux _state_.
squaremo: Sorry, this difference of opinion is going to hold up the PR. @stefanprodan Can we agree a minimal change that makes it mergeable?

I suggest: make the metric a gauge of attempted manifest applications, labelled by success or failure. In other words,

    flux_daemon_sync_manifests{success=true|false}

If you want to alert on _any_failures, you can use

    flux_daemon_sync_manifests{success=false} > 0

stefanprodan: I'm ok with `flux_daemon_sync_manifests{success=true|false}`. It's easy to reason about and create alerts.
mpashka: 
So if sync was successful we put flux_daemon_sync_manifests{success=true}=manifest_number, if not - flux_daemon_sync_manifests{success=false}=manifest_errors
If there was error reading manifests or parsing manifest we put flux_daemon_sync_manifests{success=true}=0 and flux_daemon_sync_manifests{success=false}=number_of_parse_errors? Or just flux_daemon_sync_manifests{success=false}=1?
squaremo: I'm not sure of the most convenient way to get these numbers, but

```
flux_daemon_sync_manifests{success=false}=manifest_errors
flux_daemon_sync_manifests{success=true}=total_manifest_number - manifest_errors
```

manifest_errors is the number of resource errors received back from the sync, as you have it now (sync.go L163).
mpashka: Done
mpashka: Done
mpashka: That's correct. Success or failure of the overall sync process can be obtained by running queries:
delta(flux_daemon_sync_duration_seconds_count{success='true'}[6m]) < 1
or
rate(flux_daemon_sync_duration_seconds_count{success='false'}[6m]) > 0

So we need only measure the case where manifests are correct but there were some problems applying manifests to kubernetes - e.g. configmap size is too big to fit in annotations or immutable field (like label selector) was changed.

Probably it is better to put glux alarming info into documentation as well. I suppose https://github.com/fluxcd/flux/blob/master/docs/references/monitoring.md is a good place for it.

Patch:
diff --git a/docs/references/monitoring.md b/docs/references/monitoring.md
index d061686d..ed2267e5 100644
--- a/docs/references/monitoring.md
+++ b/docs/references/monitoring.md
@@ -14,5 +14,15 @@ The following metrics are exposed:
 | `flux_daemon_queue_duration_seconds`     | Duration of time spent in the job queue before execution
 | `flux_daemon_queue_length_count`         | Count of jobs waiting in the queue to be run
 | `flux_daemon_sync_duration_seconds`      | Duration of git-to-cluster synchronisation
+| `flux_daemon_sync_manifests`             | Number of manifests being synced to cluster
 | `flux_registry_fetch_duration_seconds`   | Duration of image metadata requests (from cache)
 | `flux_fluxd_connection_duration_seconds` | Duration in seconds of the current connection to fluxsvc
+
+Flux sync state can be obtained by using the following PromQL expressions:
+* `delta(flux_daemon_sync_duration_seconds_count{success='true'}[6m]) < 1` - for general flux sync errors - usually if 
+that is true then there are some problems with infrastructure or there are manifests parse error or there are manifests 
+with duplicate ids.
+
+* `flux_daemon_sync_manifests{success='false'} > 0` - for git manifests errors - if true then there are either some 
+problems with applying git manifests to kubernetes - e.g. configmap size is too big to fit in annotations or 
+immutable field (like label selector) was changed. 
diff --git a/pkg/daemon/metrics.go b/pkg/daemon/metrics.go
index 7cad9a16..6c873e58 100644
--- a/pkg/daemon/metrics.go
+++ b/pkg/daemon/metrics.go
@@ -44,4 +44,11 @@ var (
 		Name:      "queue_length_count",
 		Help:      "Count of jobs waiting in the queue to be run.",
 	}, []string{})
+
+	syncManifestsMetric = prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+		Namespace: "flux",
+		Subsystem: "daemon",
+		Name:      "sync_manifests",
+		Help:      "Number of synchronized manifests",
+	}, []string{fluxmetrics.LabelSuccess})
 )
diff --git a/pkg/daemon/sync.go b/pkg/daemon/sync.go
index e0c59780..6d764236 100644
--- a/pkg/daemon/sync.go
+++ b/pkg/daemon/sync.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/base64"
+	"github.com/fluxcd/flux/pkg/metrics"
 	"github.com/go-kit/kit/log"
 	"github.com/pkg/errors"
 	"path/filepath"
@@ -157,6 +158,7 @@ func doSync(ctx context.Context, manifestsStore manifests.Store, clus cluster.Cl
 		switch syncerr := err.(type) {
 		case cluster.SyncError:
 			logger.Log("err", err)
+			updateSyncManifestsMetric(len(resources)-len(syncerr), len(syncerr))
 			for _, e := range syncerr {
 				resourceErrors = append(resourceErrors, event.ResourceError{
 					ID:    e.ResourceID,
@@ -167,10 +169,17 @@ func doSync(ctx context.Context, manifestsStore manifests.Store, clus cluster.Cl
 		default:
 			return nil, nil, err
 		}
+	} else {
+		updateSyncManifestsMetric(len(resources), 0)
 	}
 	return resources, resourceErrors, nil
 }
 
+func updateSyncManifestsMetric(success, failure int) {
+	syncManifestsMetric.With(metrics.LabelSuccess, "true").Set(float64(success))
+	syncManifestsMetric.With(metrics.LabelSuccess, "false").Set(float64(failure))
+}
+
 // getChangedResources calculates what resources are modified during
 // this sync.
 func (d *Daemon) getChangedResources(ctx context.Context, c changeSet, timeout time.Duration, working *git.Export,


Test patch:
diff --git a/pkg/daemon/sync_test.go b/pkg/daemon/sync_test.go
index d159c35f..e5e9f2cc 100644
--- a/pkg/daemon/sync_test.go
+++ b/pkg/daemon/sync_test.go
@@ -12,8 +12,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/go-kit/kit/log"
-
 	"github.com/fluxcd/flux/pkg/cluster"
 	"github.com/fluxcd/flux/pkg/cluster/kubernetes"
 	"github.com/fluxcd/flux/pkg/cluster/kubernetes/testfiles"
@@ -26,6 +24,9 @@ import (
 	registryMock "github.com/fluxcd/flux/pkg/registry/mock"
 	"github.com/fluxcd/flux/pkg/resource"
 	fluxsync "github.com/fluxcd/flux/pkg/sync"
+	"github.com/go-kit/kit/log"
+	"github.com/prometheus/client_golang/prometheus"
+	promdto "github.com/prometheus/client_model/go"
 )
 
 const (
@@ -86,6 +87,50 @@ func daemon(t *testing.T) (*Daemon, func()) {
 	}
 }
 
+func findMetric(name string, metricType promdto.MetricType, labels ...string) (*promdto.Metric, error) {
+	metricsRegistry := prometheus.DefaultRegisterer.(*prometheus.Registry)
+	if metrics, err := metricsRegistry.Gather(); err == nil {
+		for _, metricFamily := range metrics {
+			if *metricFamily.Name == name {
+				if *metricFamily.Type != metricType {
+					return nil, fmt.Errorf("Metric types for %v doesn't correpond: %v != %v", name, metricFamily.Type, metricType)
+				}
+				for _, metric := range metricFamily.Metric {
+					if len(labels) != len(metric.Label)*2 {
+						return nil, fmt.Errorf("Metric labels length for %v doesn't correpond: %v != %v", name, len(labels)*2, len(metric.Label))
+					}
+					for labelIdx, label := range metric.Label {
+						if labels[labelIdx*2] != *label.Name {
+							return nil, fmt.Errorf("Metric label for %v doesn't correpond: %v != %v", name, labels[labelIdx*2], *label.Name)
+						} else if labels[labelIdx*2+1] != *label.Value {
+							break
+						} else if labelIdx == len(metric.Label)-1 {
+							return metric, nil
+						}
+					}
+				}
+				return nil, fmt.Errorf("Can't find metric %v with appropriate labels in registry", name)
+			}
+		}
+		return nil, fmt.Errorf("Can't find metric %v in registry", name)
+	} else {
+		return nil, fmt.Errorf("Error reading metrics registry %v", err)
+	}
+}
+
+func checkSyncManifestsMetrics(t *testing.T, manifestSuccess, manifestFailures int) {
+	if metric, err := findMetric("flux_daemon_sync_manifests", promdto.MetricType_GAUGE, "success", "true"); err != nil {
+		t.Errorf("Error collecting flux_daemon_sync_manifests{success='true'} metric: %v", err)
+	} else if int(*metric.Gauge.Value) != manifestSuccess {
+		t.Errorf("flux_daemon_sync_manifests{success='true'} must be %v. Got %v", manifestSuccess, *metric.Gauge.Value)
+	}
+	if metric, err := findMetric("flux_daemon_sync_manifests", promdto.MetricType_GAUGE, "success", "false"); err != nil {
+		t.Errorf("Error collecting flux_daemon_sync_manifests{success='false'} metric: %v", err)
+	} else if int(*metric.Gauge.Value) != manifestFailures {
+		t.Errorf("flux_daemon_sync_manifests{success='false'} must be %v. Got %v", manifestFailures, *metric.Gauge.Value)
+	}
+}
+
 func TestPullAndSync_InitialSync(t *testing.T) {
 	d, cleanup := daemon(t)
 	defer cleanup()
@@ -148,6 +193,9 @@ func TestPullAndSync_InitialSync(t *testing.T) {
 	} else if len(revs) <= 0 {
 		t.Errorf("Found no revisions before the sync tag")
 	}
+
+	// Check 0 error stats
+	checkSyncManifestsMetrics(t, len(expectedResourceIDs), 0)
 }
 
 func TestDoSync_NoNewCommits(t *testing.T) {
@@ -357,3 +405,113 @@ func TestDoSync_WithNewCommit(t *testing.T) {
 		t.Errorf("Should have moved sync tag to HEAD (%s), but was moved to: %s", newRevision, revs[len(revs)-1].Revision)
 	}
 }
+
+func TestDoSync_WithErrors(t *testing.T) {
+	d, cleanup := daemon(t)
+	defer cleanup()
+
+	expectedResourceIDs := resource.IDs{}
+	for id, _ := range testfiles.ResourceMap {
+		expectedResourceIDs = append(expectedResourceIDs, id)
+	}
+
+	k8s.SyncFunc = func(def cluster.SyncSet) error {
+		return nil
+	}
+
+	ctx := context.Background()
+	head, err := d.Repo.BranchHead(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	syncTag := "sync"
+	gitSync, _ := fluxsync.NewGitTagSyncProvider(d.Repo, syncTag, "", false, d.GitConfig)
+	syncState := &lastKnownSyncState{logger: d.Logger, state: gitSync}
+
+	if err := d.Sync(ctx, time.Now().UTC(), head, syncState); err != nil {
+		t.Error(err)
+	}
+
+	// Check 0 error stats
+	checkSyncManifestsMetrics(t, len(expectedResourceIDs), 0)
+
+	// Now add wrong manifest
+	err = d.WithWorkingClone(ctx, func(checkout *git.Checkout) error {
+		ctx, cancel := context.WithTimeout(ctx, 5000*time.Second)
+		defer cancel()
+
+		absolutePath := path.Join(checkout.Dir(), "error_manifest.yaml")
+		if err := ioutil.WriteFile(absolutePath, []byte("Manifest that must produce errors"), 0600); err != nil {
+			return err
+		}
+		commitAction := git.CommitAction{Author: "", Message: "test error commit"}
+		err = checkout.CommitAndPush(ctx, commitAction, nil, true)
+		if err != nil {
+			return err
+		}
+		return err
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = d.Repo.Refresh(ctx)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if err := d.Sync(ctx, time.Now().UTC(), "HEAD", syncState); err != nil {
+		// Check error not nil, manifest counters remain the same
+		checkSyncManifestsMetrics(t, len(expectedResourceIDs), 0)
+	} else {
+		t.Error("Sync must fail because of invalid manifest")
+	}
+
+	// Fix manifest
+	err = d.WithWorkingClone(ctx, func(checkout *git.Checkout) error {
+		ctx, cancel := context.WithTimeout(ctx, 5000*time.Second)
+		defer cancel()
+
+		absolutePath := path.Join(checkout.Dir(), "error_manifest.yaml")
+		if err := ioutil.WriteFile(absolutePath, []byte("# Just comment"), 0600); err != nil {
+			return err
+		}
+		commitAction := git.CommitAction{Author: "", Message: "test fix commit"}
+		err = checkout.CommitAndPush(ctx, commitAction, nil, true)
+		if err != nil {
+			return err
+		}
+		return err
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = d.Repo.Refresh(ctx)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if err := d.Sync(ctx, time.Now().UTC(), "HEAD", syncState); err != nil {
+		t.Error(err)
+	}
+	// Check 0 manifest error stats
+	checkSyncManifestsMetrics(t, len(expectedResourceIDs), 0)
+
+	// Emulate sync errors
+	k8s.SyncFunc = func(def cluster.SyncSet) error {
+		return cluster.SyncError{
+			cluster.ResourceError{resource.MustParseID("mynamespace:deployment/depl1"), "src1", fmt.Errorf("Error1")},
+			cluster.ResourceError{resource.MustParseID("mynamespace:deployment/depl2"), "src2", fmt.Errorf("Error2")},
+		}
+	}
+
+	if err := d.Sync(ctx, time.Now().UTC(), "HEAD", syncState); err != nil {
+		t.Error(err)
+	}
+
+	// Check 2 sync error in stats
+	checkSyncManifestsMetrics(t, len(expectedResourceIDs)-2, 2)
+}