Add proper status handling and conditions

api/v1alpha1/targetsource_types.go

@@ -401,6 +401,13 @@ type TargetSourceStatus struct {
 	ObservedGeneration int64       `json:"observedGeneration"`
 	TargetsCount       int32       `json:"targetsCount,omitempty"`
 	LastSync           metav1.Time `json:"lastSync,omitempty"`
+
+	// Conditions represent the latest available observations of the
+	// TargetSource's state.
+	// +listType=map
+	// +listMapKey=type
+	// +optional
+	Conditions []metav1.Condition `json:"conditions,omitempty"`
 }
 
 //+kubebuilder:object:root=true

config/crd/bases/operator.gnmic.dev_targetsources.yaml

@@ -451,6 +451,68 @@ spec:
           status:
             description: TargetSourceStatus defines the observed state of TargetSource
             properties:
+              conditions:
+                description: |-
+                  Conditions represent the latest available observations of the
+                  TargetSource's state.
+                items:
+                  description: Condition contains details for one aspect of the current
+                    state of this API Resource.
+                  properties:
+                    lastTransitionTime:
+                      description: |-
+                        lastTransitionTime is the last time the condition transitioned from one status to another.
+                        This should be when the underlying condition changed.  If that is not known, then using the time when the API field changed is acceptable.
+                      format: date-time
+                      type: string
+                    message:
+                      description: |-
+                        message is a human readable message indicating details about the transition.
+                        This may be an empty string.
+                      maxLength: 32768
+                      type: string
+                    observedGeneration:
+                      description: |-
+                        observedGeneration represents the .metadata.generation that the condition was set based upon.
+                        For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date
+                        with respect to the current state of the instance.
+                      format: int64
+                      minimum: 0
+                      type: integer
+                    reason:
+                      description: |-
+                        reason contains a programmatic identifier indicating the reason for the condition's last transition.
+                        Producers of specific condition types may define expected values and meanings for this field,
+                        and whether the values are considered a guaranteed API.
+                        The value should be a CamelCase string.
+                        This field may not be empty.
+                      maxLength: 1024
+                      minLength: 1
+                      pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
+                      type: string
+                    status:
+                      description: status of the condition, one of True, False, Unknown.
+                      enum:
+                      - "True"
+                      - "False"
+                      - Unknown
+                      type: string
+                    type:
+                      description: type of condition in CamelCase or in foo.example.com/CamelCase.
+                      maxLength: 316
+                      pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$
+                      type: string
+                  required:
+                  - lastTransitionTime
+                  - message
+                  - reason
+                  - status
+                  - type
+                  type: object
+                type: array
+                x-kubernetes-list-map-keys:
+                - type
+                x-kubernetes-list-type: map
               lastSync:
                 format: date-time
                 type: string

internal/controller/discovery/client.go

@@ -9,7 +9,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
@@ -78,23 +77,6 @@ func deleteTarget(ctx context.Context, c client.Client, name string, namespace s
 	return err
 }
 
-// updateTargetSourceStatus updates the status of the TargetSource Object ts. The only fields updated are targetCount and LastSync, which takes the current timestamp.
-func updateTargetSourceStatus(ctx context.Context, c client.Client, ts *gnmicv1alpha1.TargetSource, targetCount int32) error {
-	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		latest := &gnmicv1alpha1.TargetSource{}
-		if err := c.Get(ctx, client.ObjectKeyFromObject(ts), latest); err != nil {
-			return err
-		}
-
-		latest.Status.TargetsCount = targetCount
-		latest.Status.LastSync = metav1.Now()
-
-		return c.Status().Update(ctx, latest)
-	})
-
-	return err
-}
-
 // Helper: GetSecretValues returns values from a secret
 // If keys are provided -> returns only those keys
 // If keys is empty -> returns entire secret data

internal/controller/discovery/core/status_updater_interface.go

@@ -0,0 +1,33 @@
+package core
+
+import (
+	"context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	ConditionTypeReady       = "Ready"
+	ConditionTypeReconciling = "Reconciling"
+	ConditionTypeDegraded    = "Degraded"
+	ConditionTypeStalled     = "Stalled"
+
+	ReasonWaitingForSync Reason = "WaitingForSync"
+	ReasonSyncStarted    Reason = "SyncStarted"
+	ReasonSyncSucceeded  Reason = "SyncSucceeded"
+	ReasonSyncCompleted  Reason = "SyncCompleted"
+	ReasonSyncWithErrors Reason = "SyncSucceededWithErrors"
+	ReasonSyncFailed     Reason = "SyncFailed"
+)
+
+type Reason string
+
+type StatusUpdate struct {
+	Conditions   []metav1.Condition
+	TargetsCount *int32
+}
+
+// StatusUpdater defines the interface for TargetLoaders and MessageProcessor to update the status of the TargetSource
+type StatusUpdater interface {
+	UpdateStatus(context.Context, StatusUpdate) error
+}

internal/controller/discovery/core/types.go

@@ -26,6 +26,7 @@ type CommonLoaderConfig struct {
 	PushConfig      *v1alpha1.PushSpec
 	Router          *gin.Engine
 	ResourceFetcher ResourceFetcher
+	Updater         StatusUpdater
 }
 
 // EventAction represents the type of a discovery event

internal/controller/discovery/loaders/http/loader.go

@@ -13,6 +13,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/google/uuid"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
@@ -39,6 +40,17 @@ func (l *Loader) Name() string {
 	return "http"
 }
 
+// reportStatus emits a status update through the configured StatusUpdater,
+// if one is set. It is a no-op when no updater is configured (e.g. in tests).
+func (l *Loader) reportStatus(ctx context.Context, update core.StatusUpdate) {
+	if l.loaderCfg.Updater == nil {
+		return
+	}
+	if err := l.loaderCfg.Updater.UpdateStatus(ctx, update); err != nil {
+		log.FromContext(ctx).Error(err, "failed to update TargetSource status")
+	}
+}
+
 // Run starts the HTTP discovery loop
 // It performs an immediate fetch and then continues polling at a fixed interval
 func (l *Loader) Run(ctx context.Context, out chan<- []core.DiscoveryMessage) error {
@@ -79,6 +91,17 @@ func (l *Loader) Run(ctx context.Context, out chan<- []core.DiscoveryMessage) er
 
 	// helper function to fetch targets and emit discovery messages
 	fetchAndEmit := func() {
+		l.reportStatus(ctx, core.StatusUpdate{
+			Conditions: []metav1.Condition{
+				{
+					Type:    core.ConditionTypeReconciling,
+					Status:  metav1.ConditionTrue,
+					Reason:  string(core.ReasonSyncStarted),
+					Message: "Fetching targets from HTTP endpoint",
+				},
+			},
+		})
+
 		// Fetch targets from HTTP endpoint
 		targets, err := l.fetchTargetsFromHTTPEndpoint(ctx, client, logger)
 		if err != nil {
@@ -87,6 +110,16 @@ func (l *Loader) Run(ctx context.Context, out chan<- []core.DiscoveryMessage) er
 				"Failed to fetch targets from HTTP endpoint",
 				"url", l.spec.URL,
 			)
+			l.reportStatus(ctx, core.StatusUpdate{
+				Conditions: []metav1.Condition{
+					{
+						Type:    core.ConditionTypeStalled,
+						Status:  metav1.ConditionTrue,
+						Reason:  string(core.ReasonSyncFailed),
+						Message: err.Error(),
+					},
+				},
+			})
 			return
 		}
 

internal/controller/discovery/message_processor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -31,15 +32,17 @@ type MessageProcessor struct {
 	// Events are deferred while snapshot is in progress
 	deferredEvents []core.DiscoveryEvent
 	targetCount    int32
+	updater        core.StatusUpdater
 }
 
 // NewMessageProcessor wires a MessageProcessor instance
-func NewMessageProcessor(c client.Client, s *runtime.Scheme, ts *gnmicv1alpha1.TargetSource, in <-chan []core.DiscoveryMessage) *MessageProcessor {
+func NewMessageProcessor(c client.Client, s *runtime.Scheme, ts *gnmicv1alpha1.TargetSource, in <-chan []core.DiscoveryMessage, u core.StatusUpdater) *MessageProcessor {
 	return &MessageProcessor{
 		client:       c,
 		scheme:       s,
 		targetSource: ts,
 		in:           in,
+		updater:      u,
 	}
 }
 
@@ -366,7 +369,22 @@ func (m *MessageProcessor) applyEvent(ctx context.Context, event core.DiscoveryE
 }
 
 func (m *MessageProcessor) updateStatus(ctx context.Context, logger logr.Logger) {
-	if err := updateTargetSourceStatus(ctx, m.client, m.targetSource, m.targetCount); err != nil {
+	if m.updater == nil {
+		return
+	}
+	count := m.targetCount
+	update := core.StatusUpdate{
+		Conditions: []metav1.Condition{
+			{
+				Type:    core.ConditionTypeReady,
+				Status:  metav1.ConditionTrue,
+				Reason:  string(core.ReasonSyncSucceeded),
+				Message: "Successfully synced targets",
+			},
+		},
+		TargetsCount: &count,
+	}
+	if err := m.updater.UpdateStatus(ctx, update); err != nil {
 		logger.Error(err, "error updating TargetSource status")
 	} else {
 		logger.Info("updated target source status",

internal/controller/discovery/message_processor_test.go

@@ -39,6 +39,7 @@ func mockMessageProcessor(opts ...func(*MessageProcessor)) *MessageProcessor {
 		scheme,
 		&targetSource,
 		targetChannel,
+		nil,
 	)
 
 	for _, opt := range opts {

internal/controller/discovery/status_updater_client.go

@@ -0,0 +1,69 @@
+package discovery
+
+import (
+	"context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/util/retry"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
+	"github.com/gnmic/operator/internal/controller/discovery/core"
+)
+
+// k8sStatusUpdater is a client which fulfills the StatusUpdater interface
+type k8sStatusUpdater struct {
+	client       client.Client
+	scheme       *runtime.Scheme
+	targetSource *gnmicv1alpha1.TargetSource
+}
+
+// Returns an instance of k8sStatusUpdater
+func NewK8sStatusUpdater(c client.Client, s *runtime.Scheme, ts *gnmicv1alpha1.TargetSource) *k8sStatusUpdater {
+	return &k8sStatusUpdater{
+		client:       c,
+		scheme:       s,
+		targetSource: ts,
+	}
+}
+
+// UpdateStatus takes a StatusUpdate holding Conditions and a pointer referencing the TargetsCount.
+// If TargetsCount is set, the LastSync time gets set to metav1.Now().
+// Replaces LastTransitionTime of each Condition with metav1.Now().
+func (c *k8sStatusUpdater) UpdateStatus(ctx context.Context, update core.StatusUpdate) error {
+
+	return c.patchStatus(ctx, func(
+		ts *gnmicv1alpha1.TargetSource,
+	) {
+		now := metav1.Now()
+
+		// Update status fields: Replace all Conditions and set TargetsCount and LastSync if pointer != nil
+		for i := range update.Conditions {
+			update.Conditions[i].LastTransitionTime = now
+		}
+		ts.Status.Conditions = update.Conditions
+
+		if update.TargetsCount != nil {
+			ts.Status.TargetsCount = *update.TargetsCount
+			ts.Status.LastSync = now
+		}
+	})
+}
+
+// patchStatus is an internal function to update the Kubernetes object
+func (c *k8sStatusUpdater) patchStatus(ctx context.Context, mutate func(*gnmicv1alpha1.TargetSource)) error {
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		latest := &gnmicv1alpha1.TargetSource{}
+		if err := c.client.Get(ctx, client.ObjectKeyFromObject(c.targetSource), latest); err != nil {
+			return err
+		}
+
+		patch := client.MergeFrom(latest.DeepCopy())
+		mutate(latest)
+
+		return c.client.Status().Patch(ctx, latest, patch)
+	})
+
+	return err
+}

internal/controller/targetsource_controller.go

@@ -192,9 +192,11 @@ func (r *TargetSourceReconciler) startDiscovery(
 ) error {
 	targetChannel := make(chan []discoveryTypes.DiscoveryMessage, r.BufferSize)
 	ctx, cancel := context.WithCancel(context.Background())
+	statusUpdater := discovery.NewK8sStatusUpdater(r.Client, r.Scheme, targetSource)
 	loaderConfig := discoveryTypes.CommonLoaderConfig{
 		TargetsourceNN: key,
 		ChunkSize:      r.ChunkSize,
+		Updater:        statusUpdater,
 	}
 
 	// Cleanup function to cleanup discovery runtime of targetsource
@@ -208,6 +210,7 @@ func (r *TargetSourceReconciler) startDiscovery(
 		r.Scheme,
 		targetSource,
 		targetChannel,
+		statusUpdater,
 	)
 	loader, err := discovery.NewLoader(reconcileCtx, r.Client, &loaderConfig, targetSource.Spec)
 	if err != nil {