add Target Status using gNMIc SSE endpoint

api/v1alpha1/target_types.go

@@ -28,19 +28,46 @@ type TargetSpec struct {
 	Profile string `json:"profile"`
 }
 
-// TargetStatus defines the observed state of Target
+// TargetStatus defines the observed state of Target.
+// A single Target may be collected by multiple Clusters (via different Pipelines),
+// so the status is reported per-cluster.
 type TargetStatus struct {
-	// The connection state of the target
-	ConnectionState  string      `json:"connectionState"`
-	LastConnected    metav1.Time `json:"lastConnected"`
-	LastDisconnected metav1.Time `json:"lastDisconnected"`
-	LastError        string      `json:"lastError"`
+	// Number of clusters currently collecting this target.
+	Clusters int32 `json:"clusters"`
+	// Aggregate connection state across all clusters.
+	// READY if all clusters report READY, DEGRADED if any do not.
+	// Empty when no clusters are collecting this target.
+	ConnectionState string `json:"connectionState,omitempty"`
+	// Per-cluster target state, keyed by Cluster CR name.
+	// A target may be collected by multiple clusters (via different pipelines).
+	// +optional
+	ClusterStates map[string]ClusterTargetState `json:"clusterStates,omitempty"`
+}
+
+// ClusterTargetState represents the state of a target on a specific gNMIc cluster pod.
+type ClusterTargetState struct {
+	// The pod within the cluster that currently owns this target.
+	Pod string `json:"pod"`
+	// The target's operational state (starting, running, stopping, stopped, failed).
+	State string `json:"state,omitempty"`
+	// The reason for failure when state is "failed".
+	// +optional
+	FailedReason string `json:"failedReason,omitempty"`
+	// The gNMI connection state (CONNECTING, READY, TRANSIENT_FAILURE, etc.).
+	ConnectionState string `json:"connectionState,omitempty"`
+	// Per-subscription state (subscription name -> running/stopped).
+	// +optional
+	Subscriptions map[string]string `json:"subscriptions,omitempty"`
+	// When this state was last updated by the gNMIc pod.
+	LastUpdated metav1.Time `json:"lastUpdated,omitempty"`
 }
 
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
 // +kubebuilder:printcolumn:name="Address",type=string,JSONPath=`.spec.address`
 // +kubebuilder:printcolumn:name="Profile",type=string,JSONPath=`.spec.profile`
+// +kubebuilder:printcolumn:name="Clusters",type=integer,JSONPath=`.status.clusters`
+// +kubebuilder:printcolumn:name="State",type=string,JSONPath=`.status.connectionState`
 
 // Target is the Schema for the targets API
 type Target struct {

cmd/main.go

@@ -178,6 +178,13 @@ func main() {
 			os.Exit(1)
 		}
 	}
+	if err = (&controller.TargetStateReconciler{
+		Client: mgr.GetClient(),
+		Scheme: mgr.GetScheme(),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "TargetState")
+		os.Exit(1)
+	}
 	if err := (&controller.TunnelTargetPolicyReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),

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

@@ -21,6 +21,12 @@ spec:
     - jsonPath: .spec.profile
       name: Profile
       type: string
+    - jsonPath: .status.clusters
+      name: Clusters
+      type: integer
+    - jsonPath: .status.connectionState
+      name: State
+      type: string
     name: v1alpha1
     schema:
       openAPIV3Schema:
@@ -57,24 +63,59 @@ spec:
             - profile
             type: object
           status:
-            description: TargetStatus defines the observed state of Target
+            description: |-
+              TargetStatus defines the observed state of Target.
+              A single Target may be collected by multiple Clusters (via different Pipelines),
+              so the status is reported per-cluster.
             properties:
+              clusterStates:
+                additionalProperties:
+                  description: ClusterTargetState represents the state of a target
+                    on a specific gNMIc cluster pod.
+                  properties:
+                    connectionState:
+                      description: The gNMI connection state (CONNECTING, READY, TRANSIENT_FAILURE,
+                        etc.).
+                      type: string
+                    failedReason:
+                      description: The reason for failure when state is "failed".
+                      type: string
+                    lastUpdated:
+                      description: When this state was last updated by the gNMIc pod.
+                      format: date-time
+                      type: string
+                    pod:
+                      description: The pod within the cluster that currently owns
+                        this target.
+                      type: string
+                    state:
+                      description: The target's operational state (starting, running,
+                        stopping, stopped, failed).
+                      type: string
+                    subscriptions:
+                      additionalProperties:
+                        type: string
+                      description: Per-subscription state (subscription name -> running/stopped).
+                      type: object
+                  required:
+                  - pod
+                  type: object
+                description: |-
+                  Per-cluster target state, keyed by Cluster CR name.
+                  A target may be collected by multiple clusters (via different pipelines).
+                type: object
+              clusters:
+                description: Number of clusters currently collecting this target.
+                format: int32
+                type: integer
               connectionState:
-                description: The connection state of the target
-                type: string
-              lastConnected:
-                format: date-time
-                type: string
-              lastDisconnected:
-                format: date-time
-                type: string
-              lastError:
+                description: |-
+                  Aggregate connection state across all clusters.
+                  READY if all clusters report READY, DEGRADED if any do not.
+                  Empty when no clusters are collecting this target.
                 type: string
             required:
-            - connectionState
-            - lastConnected
-            - lastDisconnected
-            - lastError
+            - clusters
             type: object
         type: object
     served: true

config/rbac/role.yaml

@@ -87,6 +87,7 @@ rules:
   resources:
   - clusters/status
   - pipelines/status
+  - targets/status
   - targetsources/status
   - tunneltargetpolicies/status
   verbs:

internal/controller/cluster_controller.go

@@ -1978,13 +1978,24 @@ func (r *ClusterReconciler) updatePipelineStatus(ctx context.Context, pipeline *
 		}
 	}
 
-	// update status if changed
+	// update status if changed, with retry on conflict
 	if !pipelineStatusEqual(pipeline.Status, newStatus) {
-		pipeline.Status = newStatus
-		if err := r.Status().Update(ctx, pipeline); err != nil {
-			return fmt.Errorf("failed to update pipeline status: %w", err)
+		pipelineNN := types.NamespacedName{Name: pipeline.Name, Namespace: pipeline.Namespace}
+		for attempt := 0; attempt < 5; attempt++ {
+			// re-fetch to get the latest resourceVersion
+			if err := r.Get(ctx, pipelineNN, pipeline); err != nil {
+				return fmt.Errorf("failed to re-fetch pipeline: %w", err)
+			}
+			pipeline.Status = newStatus
+			if err := r.Status().Update(ctx, pipeline); err != nil {
+				if errors.IsConflict(err) {
+					continue
+				}
+				return fmt.Errorf("failed to update pipeline status: %w", err)
+			}
+			logger.Info("updated pipeline status", "pipeline", pipeline.Name, "targets", newStatus.TargetsCount)
+			break
 		}
-		logger.Info("updated pipeline status", "pipeline", pipeline.Name, "targets", newStatus.TargetsCount)
 	}
 
 	return nil
@@ -2032,8 +2043,21 @@ func (r *ClusterReconciler) updatePipelineStatusWithError(ctx context.Context, p
 		},
 	}
 
-	pipeline.Status = newStatus
-	return r.Status().Update(ctx, pipeline)
+	pipelineNN := types.NamespacedName{Name: pipeline.Name, Namespace: pipeline.Namespace}
+	for attempt := 0; attempt < 5; attempt++ {
+		if err := r.Get(ctx, pipelineNN, pipeline); err != nil {
+			return fmt.Errorf("failed to re-fetch pipeline: %w", err)
+		}
+		pipeline.Status = newStatus
+		if err := r.Status().Update(ctx, pipeline); err != nil {
+			if errors.IsConflict(err) {
+				continue
+			}
+			return fmt.Errorf("failed to update pipeline status: %w", err)
+		}
+		return nil
+	}
+	return fmt.Errorf("failed to update pipeline status after retries: conflict")
 }
 
 // listPipelinesForCluster returns all enabled Pipelines that reference this Cluster