Merge branch 'main' into lk/fix-flipt-export-ids

Author: kodiakhq[bot]

DEVELOPMENT.md

@@ -28,7 +28,7 @@ If you run into errors such as:
 undefined: sqlite3.Error
 ```
 
-Then you need to enable CGO. 
+Then you need to enable CGO.
 
 ### Windows
 
@@ -43,10 +43,10 @@ Then you need to enable CGO.
 ## Setup
 
 1. Clone this repo: `git clone https://github.com/flipt-io/flipt`.
-1. Run `mage bootstrap` to install required development tools. See [#bootstrap](#bootstrap) below.
-1. Run `mage go:test` to execute the Go test suite. For more information on tests, see also [here](build/README.md)
-1. Run `mage` to build the binary with embedded assets.
-1. Run `mage -l` to see a full list of possible commands.
+2. Run `mage bootstrap` to install required development tools. See [#bootstrap](#bootstrap) below.
+3. Run `mage go:test` to execute the Go test suite. For more information on tests, see also [here](build/README.md)
+4. Run `mage` to build the binary with embedded assets.
+5. Run `mage -l` to see a full list of possible commands.
 
 ## Conventional Commits
 
@@ -108,7 +108,7 @@ These ports will be forwarded to your local machine automatically if you are dev
 
 ## Docker Compose
 
-If you want to develop Flipt using Docker Compose, you can use the `docker-compose.yml` file in the root of this repository. 
+If you want to develop Flipt using Docker Compose, you can use the `docker-compose.yml` file in the root of this repository.
 
 This will start two Docker containers:
 

go.work.sum

@@ -44,14 +44,15 @@ type MustInt64Meter interface {
 	// with options. The instrument is used to synchronously record increasing
 	// int64 measurements during a computational operation.
 	Counter(name string, options ...metric.Int64CounterOption) metric.Int64Counter
-	// UpDownCounter returns a new instrument identified by name and
-	// configured with options. The instrument is used to synchronously record
-	// int64 measurements during a computational operation.
-	UpDownCounter(name string, options ...metric.Int64UpDownCounterOption) metric.Int64UpDownCounter
 	// Histogram returns a new instrument identified by name and
 	// configured with options. The instrument is used to synchronously record
 	// the distribution of int64 measurements during a computational operation.
 	Histogram(name string, options ...metric.Int64HistogramOption) metric.Int64Histogram
+	// ObservableGauge returns a new instrument
+	// identified by name and configured with options. The instrument is used
+	// to asynchronously record instantaneous measurements once per a
+	// measurement collection cycle.
+	ObservableGauge(name string, v func() int64, options ...metric.Int64ObservableGaugeOption) metric.Int64ObservableGauge
 }
 
 type mustInt64Meter struct{}
@@ -66,14 +67,23 @@ func (m mustInt64Meter) Counter(name string, opts ...metric.Int64CounterOption)
 	return counter
 }
 
-// UpDownCounter creates an instrument for recording changes of a value.
-func (m mustInt64Meter) UpDownCounter(name string, opts ...metric.Int64UpDownCounterOption) metric.Int64UpDownCounter {
-	counter, err := meter().Int64UpDownCounter(name, opts...)
+// Gauge creates an instrument for recording changes of a value.
+func (m mustInt64Meter) ObservableGauge(name string, v func() int64, opts ...metric.Int64ObservableGaugeOption) metric.Int64ObservableGauge {
+	gauge, err := meter().Int64ObservableGauge(name, opts...)
 	if err != nil {
 		panic(err)
 	}
-
-	return counter
+	_, err = meter().RegisterCallback(
+		func(ctx context.Context, observer metric.Observer) error {
+			observer.ObserveInt64(gauge, v())
+			return nil
+		},
+		gauge,
+	)
+	if err != nil {
+		panic(err)
+	}
+	return gauge
 }
 
 // Histogram creates an instrument for recording a distribution of values.
@@ -100,10 +110,6 @@ type MustFloat64Meter interface {
 	// with options. The instrument is used to synchronously record increasing
 	// float64 measurements during a computational operation.
 	Counter(name string, options ...metric.Float64CounterOption) metric.Float64Counter
-	// UpDownCounter returns a new instrument identified by name and
-	// configured with options. The instrument is used to synchronously record
-	// float64 measurements during a computational operation.
-	UpDownCounter(name string, options ...metric.Float64UpDownCounterOption) metric.Float64UpDownCounter
 	// Histogram returns a new instrument identified by name and
 	// configured with options. The instrument is used to synchronously record
 	// the distribution of float64 measurements during a computational operation.

internal/metrics/metrics.go

@@ -0,0 +1,72 @@
+package git
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"go.flipt.io/flipt/internal/metrics"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const (
+	namespace = "flipt"
+	subsystem = "git_sync"
+)
+
+var (
+	// syncLatency is a histogram for git sync operation duration.
+	syncLatency = metrics.MustFloat64().
+			Histogram(
+			prometheus.BuildFQName(namespace, subsystem, "latency"),
+			metric.WithDescription("The duration of git sync operations in seconds"),
+			metric.WithUnit("s"),
+		)
+
+	// syncFlags is a counter for the number of flags during sync.
+	syncFlags = metrics.MustInt64().
+			Counter(
+			prometheus.BuildFQName(namespace, subsystem, "flags"),
+			metric.WithDescription("The number of flags fetched during git sync"),
+		)
+
+	// syncErrors is a counter for failed git sync operations.
+	syncErrors = metrics.MustInt64().
+			Counter(
+			prometheus.BuildFQName(namespace, subsystem, "errors"),
+			metric.WithDescription("The number of errors git sync operations"),
+		)
+
+	_ = metrics.MustInt64().
+		ObservableGauge(
+			prometheus.BuildFQName(namespace, subsystem, "last_time"),
+			getLastSyncTime,
+			metric.WithDescription("The unix timestamp of the last git sync operation"),
+		)
+
+	// internal storage for last sync time value
+	lastSyncTimeValue atomic.Int64
+)
+
+// observeSync records a complete git sync operation with all relevant metrics.
+func observeSync(ctx context.Context, duration time.Duration, flagsFetched int64, success bool) {
+	// Always record duration and update last sync time
+	syncLatency.Record(ctx, duration.Seconds())
+	setLastSyncTime(time.Now().UTC())
+
+	syncFlags.Add(ctx, flagsFetched)
+	if !success {
+		syncErrors.Add(ctx, 1)
+	}
+}
+
+// setLastSyncTime updates the last sync time value that will be reported by the observable gauge.
+func setLastSyncTime(ts time.Time) {
+	lastSyncTimeValue.Store(ts.Unix())
+}
+
+// getLastSyncTime returns the current last sync time value.
+func getLastSyncTime() int64 {
+	return lastSyncTimeValue.Load()
+}

internal/storage/fs/git/metrics.go

@@ -0,0 +1,39 @@
+package git
+
+import (
+	"testing"
+	"time"
+)
+
+// TestObserveSyncSuccess ensures success path sets last sync time and updates counters.
+func TestObserveSyncSuccess(t *testing.T) {
+	before := getLastSyncTime()
+	observeSync(t.Context(), 2*time.Second, 3, true)
+
+	after := getLastSyncTime()
+	if after <= before {
+		t.Errorf("expected last sync time to update, before=%d after=%d", before, after)
+	}
+}
+
+// TestObserveSyncFailure ensures failure path sets last sync time and updates counters.
+func TestObserveSyncFailure(t *testing.T) {
+	setLastSyncTime(time.Now().UTC().Add(-time.Minute))
+	before := getLastSyncTime()
+	observeSync(t.Context(), time.Second, 0, false)
+
+	after := getLastSyncTime()
+	if after <= before {
+		t.Errorf("expected last sync time to update, before=%d after=%d", before, after)
+	}
+}
+
+// TestSetAndGetLastSyncTime directly validates setter/getter.
+func TestSetAndGetLastSyncTime(t *testing.T) {
+	now := time.Now()
+	setLastSyncTime(now)
+	got := getLastSyncTime()
+	if got != now.Unix() {
+		t.Errorf("expected last sync time %d, got %d", now.Unix(), got)
+	}
+}

internal/storage/fs/git/metrics_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"slices"
 	"sync"
+	"time"
 
 	"github.com/go-git/go-billy/v5/osfs"
 	"github.com/go-git/go-git/v5"
@@ -18,6 +19,7 @@ import (
 	gitstorage "github.com/go-git/go-git/v5/storage"
 	"github.com/go-git/go-git/v5/storage/filesystem"
 	"github.com/go-git/go-git/v5/storage/memory"
+
 	"go.flipt.io/flipt/internal/containers"
 	"go.flipt.io/flipt/internal/gitfs"
 	"go.flipt.io/flipt/internal/storage"
@@ -139,6 +141,7 @@ func NewSnapshotStore(ctx context.Context, logger *zap.Logger, url string, opts
 		baseRef:           "main",
 		referenceResolver: staticResolver(),
 	}
+
 	containers.ApplyAll(store, opts...)
 
 	store.logger = store.logger.With(zap.String("ref", store.baseRef))
@@ -335,14 +338,20 @@ func (s *SnapshotStore) listRemoteRefs(ctx context.Context) (map[string]struct{}
 // HEAD updates to a new revision, it builds a snapshot and updates it
 // on the store.
 func (s *SnapshotStore) update(ctx context.Context) (bool, error) {
+	syncStart := time.Now()
 	updated, fetchErr := s.fetch(ctx, s.snaps.References())
 
+	flagsFetched := int64(0)
+	var errs []error
+
 	if !updated && fetchErr == nil {
+		// No update and no error: record metrics for a successful no-change sync
+		duration := time.Since(syncStart)
+		observeSync(ctx, duration, 0, true)
 		return false, nil
 	}
 
-	// If we can't fetch, we need to check if the remote refs have changed
-	// and remove any references that are no longer present
+	// If fetchErr exists, try cleaning up refs but do not declare full failure yet
 	if fetchErr != nil {
 		remoteRefs, listErr := s.listRemoteRefs(ctx)
 		if listErr != nil {
@@ -361,23 +370,37 @@ func (s *SnapshotStore) update(ctx context.Context) (bool, error) {
 				}
 			}
 		}
-	}
-
-	var errs []error
-	if fetchErr != nil {
 		errs = append(errs, fetchErr)
 	}
+
+	// Try to rebuild refs even if fetch failed
 	for _, ref := range s.snaps.References() {
 		hash, err := s.resolve(ref)
 		if err != nil {
 			errs = append(errs, err)
 			continue
 		}
-		if _, err := s.snaps.AddOrBuild(ctx, ref, hash, s.buildSnapshot); err != nil {
+
+		snap, err := s.snaps.AddOrBuild(ctx, ref, hash, s.buildSnapshot)
+		if err != nil {
 			errs = append(errs, err)
+			continue
+		}
+
+		if snap != nil {
+			flagsFetched += int64(snap.TotalFlagsCount())
 		}
 	}
-	return true, errors.Join(errs...)
+
+	duration := time.Since(syncStart)
+	observeSync(ctx, duration, flagsFetched, len(errs) == 0)
+
+	if len(errs) > 0 {
+		s.logger.Error("git backend flag sync failed", zap.Errors("errors", errs))
+		return false, errors.Join(errs...)
+	}
+
+	return true, nil
 }
 
 func (s *SnapshotStore) fetch(ctx context.Context, heads []string) (bool, error) {

internal/storage/fs/git/store.go

@@ -789,6 +789,14 @@ func (ss *Snapshot) CountFlags(ctx context.Context, p storage.NamespaceRequest)
 	return uint64(len(ns.flags)), nil
 }
 
+func (ss *Snapshot) TotalFlagsCount() int {
+	total := 0
+	for _, ns := range ss.ns {
+		total += len(ns.flags)
+	}
+	return total
+}
+
 func (ss *Snapshot) GetEvaluationRules(ctx context.Context, flag storage.ResourceRequest) ([]*storage.EvaluationRule, error) {
 	ns, ok := ss.ns[flag.Namespace()]
 	if !ok {

internal/storage/fs/snapshot.go

@@ -1840,3 +1840,58 @@ func TestEtagWithFewDocs(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, "dc26c96ddf6430ed603862ffe5d3ac9a", version)
 }
+
+func TestTotalFlagsCount(t *testing.T) {
+	testCases := []struct {
+		name          string
+		setupSnapshot func() *Snapshot
+		expectedCount int
+	}{
+		{
+			name: "EmptySnapshot",
+			setupSnapshot: func() *Snapshot {
+				return &Snapshot{ns: map[string]*namespace{}}
+			},
+			expectedCount: 0,
+		},
+		{
+			name: "SingleNamespaceWithTwoFlags",
+			setupSnapshot: func() *Snapshot {
+				ns := &namespace{
+					flags: map[string]*flipt.Flag{
+						"flag1": {Key: "flag1", NamespaceKey: "default"},
+						"flag2": {Key: "flag2", NamespaceKey: "default"},
+					},
+				}
+				return &Snapshot{ns: map[string]*namespace{"default": ns}}
+			},
+			expectedCount: 2,
+		},
+		{
+			name: "MultipleNamespaces",
+			setupSnapshot: func() *Snapshot {
+				ns1 := &namespace{
+					flags: map[string]*flipt.Flag{
+						"flag1": {Key: "flag1", NamespaceKey: "default"},
+					},
+				}
+				ns2 := &namespace{
+					flags: map[string]*flipt.Flag{
+						"flagA": {Key: "flagA", NamespaceKey: "other"},
+						"flagB": {Key: "flagB", NamespaceKey: "other"},
+					},
+				}
+				return &Snapshot{ns: map[string]*namespace{"default": ns1, "other": ns2}}
+			},
+			expectedCount: 3,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			snapshot := tc.setupSnapshot()
+			got := snapshot.TotalFlagsCount()
+			assert.Equal(t, tc.expectedCount, got, "unexpected flag count")
+		})
+	}
+}