feat: reconciler integration test

shaxbee · shaxbee · commit 462e533080ac · 2025-05-24T17:43:21.000+09:00
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
@@ -56,15 +56,15 @@ jobs:
         run: make lint
       - name: Integration test
         run: make integration-test coverage
-      - name: Archive coverage report
-        uses: actions/upload-artifact@v4
+      - name: Upload coverage reports to Codecov
+        uses: codecov/codecov-action@v5
         with:
-          name: coverage
-          path: ${{ env.CODECOV_FILE }}
-
+          token: ${{ secrets.CODECOV_TOKEN }}
+          files: ${{ env.CODECOV_FILE }}
   e2e:
     name: E2E Test
-    if: github.actor!= 'dependabot-preview[bot]'
+    if: false
+    # if: github.actor!= 'dependabot-preview[bot]'
     needs: 
       - verify
       - image-build
@@ -111,7 +111,8 @@ jobs:
 
   coverage:
     name: Coverage Report
-    if: github.actor!= 'dependabot-preview[bot]'
+    if: false
+    # if: github.actor!= 'dependabot-preview[bot]'
     needs:
       - verify
       - e2e
diff --git a/e2e/cluster_test.go b/e2e/cluster_test.go
@@ -26,13 +26,13 @@ func TestCluster(t *testing.T) {
 	})
 
 	t.Run("scale", func(t *testing.T) {
-		scale(t, kcl, cluster, 5)
-		poll(t, kcl, cluster, 2*time.Minute, available)
+		Scale(t, kcl, cluster, 5)
+		Poll(t, kcl, cluster, 2*time.Minute, Available)
 
 		t.Logf("cluster %q scaled up to 5 members", cluster.Name)
 
-		scale(t, kcl, cluster, 3)
-		poll(t, kcl, cluster, time.Minute, available)
+		Scale(t, kcl, cluster, 3)
+		Poll(t, kcl, cluster, time.Minute, Available)
 
 		t.Logf("cluster %q scaled down to 3 members", cluster.Name)
 	})
@@ -60,8 +60,8 @@ func TestCluster(t *testing.T) {
 		}
 		t.Log("evict pod", key)
 
-		poll(t, kcl, cluster, 2*time.Minute, func(cluster *apiv1.EtcdCluster) bool {
-			if !available(cluster) {
+		Poll(t, kcl, cluster, 2*time.Minute, func(cluster *apiv1.EtcdCluster) bool {
+			if !Available(cluster) {
 				return false
 			}
 
@@ -85,7 +85,7 @@ func TestCluster(t *testing.T) {
 		}
 
 		version = strings.TrimPrefix(version, "v")
-		poll(t, kcl, cluster, 5*time.Minute, func(cluster *apiv1.EtcdCluster) bool {
+		Poll(t, kcl, cluster, 5*time.Minute, func(cluster *apiv1.EtcdCluster) bool {
 			if !conditions.StatusTrue(cluster.Status.Conditions, apiv1.ClusterAvailable) {
 				return false
 			}
diff --git a/e2e/kube_test.go b/e2e/kube_test.go
@@ -16,6 +16,7 @@ import (
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -147,14 +148,14 @@ func createCluster(t testing.TB, kcl client.Client, timeout time.Duration, spec
 	key := client.ObjectKeyFromObject(cluster)
 	t.Logf("cluster %q created", key)
 
-	poll(t, kcl, cluster, timeout, available)
+	Poll(t, kcl, cluster, timeout, Available)
 
 	t.Logf("cluster %q available", key)
 
 	return cluster
 }
 
-func poll[T client.Object](t testing.TB, kcl client.Client, obj T, timeout time.Duration, f func(obj T) bool) {
+func Poll[T client.Object](t testing.TB, kcl client.Client, obj T, timeout time.Duration, f func(obj T) bool) {
 	t.Helper()
 
 	ctx, cancel := context.WithTimeout(t.Context(), timeout)
@@ -173,11 +174,11 @@ func poll[T client.Object](t testing.TB, kcl client.Client, obj T, timeout time.
 	}
 }
 
-func available(cluster *apiv1.EtcdCluster) bool {
+func Available(cluster *apiv1.EtcdCluster) bool {
 	return cluster.Status.AvailableReplicas == cluster.Spec.Replicas
 }
 
-func scale(t testing.TB, kcl client.Client, obj client.Object, replicas int32) {
+func Scale(t testing.TB, kcl client.Client, obj client.Object, replicas int32) {
 	scale := &autoscalingv1.Scale{}
 	err := kcl.SubResource("scale").Get(t.Context(), obj, scale)
 	if err != nil {
@@ -226,7 +227,11 @@ func triggerCronJob(t testing.TB, kcl client.Client, key client.ObjectKey, timeo
 	t.Logf("created job %q", client.ObjectKeyFromObject(job))
 
 	err = wait.PollUntilContextCancel(ctx, 500*time.Millisecond, true, func(ctx context.Context) (done bool, err error) {
-		if err := kcl.Get(ctx, client.ObjectKeyFromObject(job), job); err != nil {
+		err = kcl.Get(ctx, client.ObjectKeyFromObject(job), job)
+		switch {
+		case apierrors.IsTooManyRequests(err):
+			return false, err
+		case err != nil:
 			return true, fmt.Errorf("get backup job: %w", err)
 		}
 
diff --git a/pkg/metrics/observer.go b/pkg/metrics/observer.go
@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"sync/atomic"
 
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 
@@ -20,16 +22,14 @@ type Observer struct {
 	backupLastScheduleTime, backupLastSuccessfulTime                                              metric.Int64ObservableGauge
 }
 
-func Register(meter metric.Meter, cluster *apiv1.EtcdCluster) (*Observer, error) {
-	observer := &Observer{
+func Register(meter metric.Meter, key client.ObjectKey) (*Observer, error) {
+	o := &Observer{
 		attributes: attribute.NewSet(
-			attribute.String("fleet.etcd.cluster.namespace", cluster.Namespace),
-			attribute.String("fleet.etcd.cluster.name", cluster.Name),
+			attribute.String("fleet.etcd.cluster.namespace", key.Namespace),
+			attribute.String("fleet.etcd.cluster.name", key.Name),
 		),
 	}
 
-	observer.cluster.Store(cluster)
-
 	gauges := []struct {
 		name        string
 		description string
@@ -38,42 +38,42 @@ func Register(meter metric.Meter, cluster *apiv1.EtcdCluster) (*Observer, error)
 		{
 			name:        "fleet.etcd.cluster.desired_replicas",
 			description: "Number of desired replicas",
-			dest:        &observer.desiredReplicas,
+			dest:        &o.desiredReplicas,
 		},
 		{
 			name:        "fleet.etcd.cluster.replicas",
 			description: "Number of replicas",
-			dest:        &observer.replicas,
+			dest:        &o.replicas,
 		},
 		{
 			name:        "fleet.etcd.cluster.ready_replicas",
 			description: "Number of ready replicas",
-			dest:        &observer.readyReplicas,
+			dest:        &o.readyReplicas,
 		},
 		{
 			name:        "fleet.etcd.cluster.updated_replicas",
 			description: "Number of updated replicas",
-			dest:        &observer.updatedReplicas,
+			dest:        &o.updatedReplicas,
 		},
 		{
 			name:        "fleet.etcd.cluster.available_replicas",
 			description: "Number of available replicas",
-			dest:        &observer.availableReplicas,
+			dest:        &o.availableReplicas,
 		},
 		{
 			name:        "fleet.etcd.cluster.learner_replicas",
 			description: "Number of learner replicas",
-			dest:        &observer.learnerReplicas,
+			dest:        &o.learnerReplicas,
 		},
 		{
 			name:        "fleet.etcd.cluster.backup.last_schedule_time",
 			description: "Last backup schedule time",
-			dest:        &observer.backupLastScheduleTime,
+			dest:        &o.backupLastScheduleTime,
 		},
 		{
 			name:        "fleet.etcd.cluster.backup.last_successful_time",
 			description: "Last backup successful time",
-			dest:        &observer.backupLastSuccessfulTime,
+			dest:        &o.backupLastSuccessfulTime,
 		},
 	}
 
@@ -88,40 +88,42 @@ func Register(meter metric.Meter, cluster *apiv1.EtcdCluster) (*Observer, error)
 		observables = append(observables, observable)
 	}
 
-	registration, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
-		cluster := observer.cluster.Load()
-		if cluster == nil {
-			return nil
-		}
+	registration, err := meter.RegisterCallback(o.Observe, observables...)
+	if err != nil {
+		return nil, fmt.Errorf("register callback: %w", err)
+	}
 
-		opts := []metric.ObserveOption{
-			metric.WithAttributeSet(observer.attributes),
-		}
+	o.registration = registration
 
-		o.ObserveInt64(observer.desiredReplicas, int64(cluster.Spec.Replicas), opts...)
-		o.ObserveInt64(observer.replicas, int64(cluster.Status.Replicas), opts...)
-		o.ObserveInt64(observer.readyReplicas, int64(cluster.Status.ReadyReplicas), opts...)
-		o.ObserveInt64(observer.updatedReplicas, int64(cluster.Status.UpdatedReplicas), opts...)
-		o.ObserveInt64(observer.availableReplicas, int64(cluster.Status.AvailableReplicas), opts...)
-		o.ObserveInt64(observer.learnerReplicas, int64(cluster.Status.LearnerReplicas), opts...)
+	return o, nil
+}
 
-		if cluster.Status.Backup != nil && cluster.Status.Backup.LastScheduleTime != nil {
-			o.ObserveInt64(observer.backupLastScheduleTime, cluster.Status.Backup.LastScheduleTime.Unix(), opts...)
-		}
+func (o *Observer) Observe(ctx context.Context, observer metric.Observer) error {
+	cluster := o.cluster.Load()
+	if cluster == nil {
+		return nil
+	}
 
-		if cluster.Status.Backup != nil && cluster.Status.Backup.LastSuccessfulTime != nil {
-			o.ObserveInt64(observer.backupLastSuccessfulTime, cluster.Status.Backup.LastSuccessfulTime.Unix(), opts...)
-		}
+	opts := []metric.ObserveOption{
+		metric.WithAttributeSet(o.attributes),
+	}
 
-		return nil
-	}, observables...)
-	if err != nil {
-		return nil, fmt.Errorf("register callback: %w", err)
+	observer.ObserveInt64(o.desiredReplicas, int64(cluster.Spec.Replicas), opts...)
+	observer.ObserveInt64(o.replicas, int64(cluster.Status.Replicas), opts...)
+	observer.ObserveInt64(o.readyReplicas, int64(cluster.Status.ReadyReplicas), opts...)
+	observer.ObserveInt64(o.updatedReplicas, int64(cluster.Status.UpdatedReplicas), opts...)
+	observer.ObserveInt64(o.availableReplicas, int64(cluster.Status.AvailableReplicas), opts...)
+	observer.ObserveInt64(o.learnerReplicas, int64(cluster.Status.LearnerReplicas), opts...)
+
+	if cluster.Status.Backup != nil && cluster.Status.Backup.LastScheduleTime != nil {
+		observer.ObserveInt64(o.backupLastScheduleTime, cluster.Status.Backup.LastScheduleTime.Unix(), opts...)
 	}
 
-	observer.registration = registration
+	if cluster.Status.Backup != nil && cluster.Status.Backup.LastSuccessfulTime != nil {
+		observer.ObserveInt64(o.backupLastSuccessfulTime, cluster.Status.Backup.LastSuccessfulTime.Unix(), opts...)
+	}
 
-	return observer, nil
+	return nil
 }
 
 func (o *Observer) Update(cluster *apiv1.EtcdCluster) {
diff --git a/pkg/metrics/observer_test.go b/pkg/metrics/observer_test.go
@@ -10,6 +10,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
 
 	"go.opentelemetry.io/otel/sdk/metric"
@@ -60,7 +61,9 @@ func TestObserver(t *testing.T) {
 			},
 		},
 	}
-	observer, err := Register(meter, cluster)
+
+	key := client.ObjectKeyFromObject(cluster)
+	observer, err := Register(meter, key)
 	if err != nil {
 		t.Fatal("register:", err)
 	}
@@ -72,6 +75,7 @@ func TestObserver(t *testing.T) {
 	})
 
 	metrics := &metricdata.ResourceMetrics{}
+	observer.Update(cluster)
 	err = reader.Collect(t.Context(), metrics)
 	if err != nil {
 		t.Fatal("collect:", err)
diff --git a/pkg/metrics/reconciler.go b/pkg/metrics/reconciler.go
@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"context"
+	"sync"
 
 	"go.opentelemetry.io/otel/metric"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -17,8 +18,7 @@ type Reconciler struct {
 	kcl   client.Client
 	meter metric.Meter
 
-	// no lock required - workqueue is stingy preventing concurrent reconciles of same resource
-	// https://pkg.go.dev/k8s.io/client-go/util/workqueue
+	mtx       sync.Mutex
 	observers map[client.ObjectKey]*Observer
 }
 
@@ -31,6 +31,7 @@ func SetupWithManager(mgr manager.Manager, meterProvider metric.MeterProvider) e
 	}
 
 	return builder.ControllerManagedBy(mgr).
+		Named("metrics").
 		For(&apiv1.EtcdCluster{}).
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: 4,
@@ -39,28 +40,58 @@ func SetupWithManager(mgr manager.Manager, meterProvider metric.MeterProvider) e
 }
 
 func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
-	observer := r.observers[req.NamespacedName]
-
+	key := req.NamespacedName
 	cluster := &apiv1.EtcdCluster{}
-	err := r.kcl.Get(ctx, req.NamespacedName, cluster)
+	err := r.kcl.Get(ctx, key, cluster)
 	switch {
 	// passthrough error
 	case client.IgnoreNotFound(err) != nil:
 		return reconcile.Result{}, err
-	// unregister if cluster not found or deleted
-	case (err != nil || !cluster.DeletionTimestamp.IsZero()) && observer != nil:
-		err = observer.Unregister()
+	// delete observer if cluster not found or deleted
+	case err != nil || !cluster.DeletionTimestamp.IsZero():
+		err = r.Delete(key)
+		return reconcile.Result{}, err
+	}
+
+	observer, err := r.GetOrCreate(key)
+	if err != nil {
 		return reconcile.Result{}, err
-	// register observer
-	case observer == nil:
-		observer, err = Register(r.meter, cluster)
-		if err != nil {
-			return reconcile.Result{}, err
-		}
-		r.observers[req.NamespacedName] = observer
 	}
 
 	observer.Update(cluster)
 
 	return reconcile.Result{}, nil
 }
+
+func (r *Reconciler) GetOrCreate(key client.ObjectKey) (*Observer, error) {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+
+	observer, ok := r.observers[key]
+	if ok {
+		return observer, nil
+	}
+
+	observer, err := Register(r.meter, key)
+	if err != nil {
+		return nil, err
+	}
+
+	r.observers[key] = observer
+
+	return observer, nil
+}
+
+func (r *Reconciler) Delete(key client.ObjectKey) error {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+
+	observer, ok := r.observers[key]
+	if !ok {
+		return nil
+	}
+
+	delete(r.observers, key)
+
+	return observer.Unregister()
+}
diff --git a/pkg/metrics/reconciler_test.go b/pkg/metrics/reconciler_test.go
diff --git a/pkg/metrics/testdata/TestReconciler/cluster_with_empty_status/metrics.golden.yaml b/pkg/metrics/testdata/TestReconciler/cluster_with_empty_status/metrics.golden.yaml
diff --git a/pkg/metrics/testdata/TestReconciler/cluster_with_status/metrics.golden.yaml b/pkg/metrics/testdata/TestReconciler/cluster_with_status/metrics.golden.yaml
