Add new status conditions to replication status

api/replication.storage/v1alpha1/volumereplication_types.go

@@ -30,25 +30,29 @@ const (
 // "ConditionDegraded" means the condition is not fulfilled.
 // "ConditionResyncing" means the condition is resyncing.
 const (
-	ConditionCompleted = "Completed"
-	ConditionDegraded  = "Degraded"
-	ConditionResyncing = "Resyncing"
-	ConditionValidated = "Validated"
+	ConditionCompleted   = "Completed"
+	ConditionDegraded    = "Degraded"
+	ConditionResyncing   = "Resyncing"
+	ConditionValidated   = "Validated"
+	ConditionReplicating = "Replicating"
 )
 
 // These are valid messages for various conditions and states of volume replication.
 const (
-	MessagePromoted           = "is promoted to primary and replicating to secondary"
-	MessageHealthy            = "is healthy"
-	MessageNotResyncing       = "is not resyncing"
-	MessageValidated          = "is validated and met all prerequisites"
-	MessageFailedPromoted     = "failed to promote"
-	MessageFailedDemoted      = "failed to demote"
-	MessageFailedPreCondition = "failed to meet prerequisite"
-	MessageDemoted            = "is demoted to secondary"
-	MessageDegraded           = "is degraded"
-	MessageResyncTriggered    = "is resyncing changes from primary to secondary"
-	MessageResyncFailed       = "failed to resync"
+	MessagePromoted                = "is promoted to primary and replicating to secondary"
+	MessageHealthy                 = "is healthy"
+	MessageNotResyncing            = "is not resyncing"
+	MessageValidated               = "is validated and met all prerequisites"
+	MessageFailedPromoted          = "failed to promote"
+	MessageFailedDemoted           = "failed to demote"
+	MessageFailedPreCondition      = "failed to meet prerequisite"
+	MessageDemoted                 = "is demoted to secondary"
+	MessageDegraded                = "is degraded"
+	MessageResyncTriggered         = "is resyncing changes from primary to secondary"
+	MessageResyncFailed            = "failed to resync"
+	MessageReplicating             = "is replicating"
+	MessageNotReplicating          = "is not replicating"
+	MessageUnknownReplicationState = "replication status is unknown"
 )
 
 type Source string
@@ -87,6 +91,10 @@ const (
 	PrerequisiteMet = "PrerequisiteMet"
 	// PrerequisiteNotMet condition represents that the prerequisite is not met.
 	PrerequisiteNotMet = "PrerequisiteNotMet"
+	// Replicating condition represents that the volume/group is replicating.
+	Replicating = "Replicating"
+	// NotReplicating condition represents that the volume/group is not replicating.
+	NotReplicating = "NotReplicating"
 )
 
 // ReplicationState represents the replication operations to be performed on the volume.

internal/controller/replication.storage/status.go

@@ -250,6 +250,42 @@ func setFailedResyncCondition(conditions *[]metav1.Condition, observedGeneration
 	})
 }
 
+// sets ConditionUnknown for ConditionReplicating condition in VR status.Conditions.
+func setUnknownReplicationCondition(conditions *[]metav1.Condition, observedGeneration int64, dataSource, statusMessage string) {
+	source := getSource(dataSource)
+	setStatusCondition(conditions, &metav1.Condition{
+		Message:            fmt.Sprintf("%s %s: %s", source, v1alpha1.MessageUnknownReplicationState, statusMessage),
+		Type:               v1alpha1.ConditionReplicating,
+		Reason:             v1alpha1.Replicating,
+		ObservedGeneration: observedGeneration,
+		Status:             metav1.ConditionUnknown,
+	})
+}
+
+// sets ConditionTrue for ConditionReplicating condition in VR status.Conditions
+func setHealthyReplicationCondition(conditions *[]metav1.Condition, observedGeneration int64, dataSource, statusMessage string) {
+	source := getSource(dataSource)
+	setStatusCondition(conditions, &metav1.Condition{
+		Message:            fmt.Sprintf("%s %s: %s", source, v1alpha1.MessageReplicating, statusMessage),
+		Type:               v1alpha1.ConditionReplicating,
+		Reason:             v1alpha1.Replicating,
+		ObservedGeneration: observedGeneration,
+		Status:             metav1.ConditionTrue,
+	})
+}
+
+// sets ConditionFalse for ConditionReplicating condition in VR status.Conditions
+func setDegradedReplicationCondition(conditions *[]metav1.Condition, observedGeneration int64, dataSource, statusMessage string) {
+	source := getSource(dataSource)
+	setStatusCondition(conditions, &metav1.Condition{
+		Message:            fmt.Sprintf("%s %s: %s", source, v1alpha1.MessageNotReplicating, statusMessage),
+		Type:               v1alpha1.ConditionReplicating,
+		Reason:             v1alpha1.NotReplicating,
+		ObservedGeneration: observedGeneration,
+		Status:             metav1.ConditionFalse,
+	})
+}
+
 func setStatusCondition(existingConditions *[]metav1.Condition, newCondition *metav1.Condition) {
 	if existingConditions == nil {
 		existingConditions = &[]metav1.Condition{}

internal/controller/replication.storage/volumereplication_controller.go

@@ -449,6 +449,11 @@ func (r *VolumeReplicationReconciler) Reconcile(ctx context.Context, req ctrl.Re
 					instance.Status.LastSyncBytes = &tb
 				}
 			}
+			// Check for nil, as info might be nil for knownGRPCErrors.
+			if info != nil {
+				// We set Replicating condition only for primary state and not for secondary state.
+				setReplicationCondition(vr.instance, vr.instance.Spec.DataSource.Kind, info.StatusMessage, info.Status)
+			}
 			requeueForInfo = true
 		} else if !util.IsUnimplementedError(err) {
 			logger.Error(err, "Failed to get volume replication info")
@@ -772,6 +777,18 @@ func setFailureCondition(instance *replicationv1alpha1.VolumeReplication, errMes
 	}
 }
 
+// setReplicationCondition sets the replication condition in VR's status.Conditions for the PVC/VolumeGroup dataSource based on the replicationStatus value.
+func setReplicationCondition(instance *replicationv1alpha1.VolumeReplication, dataSource, statusMessage string, replicationStatus proto.GetVolumeReplicationInfoResponse_Status) {
+	switch replicationStatus {
+	case proto.GetVolumeReplicationInfoResponse_UNKNOWN:
+		setUnknownReplicationCondition(&instance.Status.Conditions, instance.Generation, dataSource, statusMessage)
+	case proto.GetVolumeReplicationInfoResponse_HEALTHY:
+		setHealthyReplicationCondition(&instance.Status.Conditions, instance.Generation, dataSource, statusMessage)
+	case proto.GetVolumeReplicationInfoResponse_DEGRADED, proto.GetVolumeReplicationInfoResponse_ERROR:
+		setDegradedReplicationCondition(&instance.Status.Conditions, instance.Generation, dataSource, statusMessage)
+	}
+}
+
 func getCurrentTime() *metav1.Time {
 	metav1NowTime := metav1.NewTime(time.Now())
 

internal/sidecar/service/volumereplication.go

@@ -255,6 +255,8 @@ func (rs *ReplicationServer) GetVolumeReplicationInfo(
 		LastSyncTime:     lastsynctime,
 		LastSyncDuration: resp.GetLastSyncDuration(),
 		LastSyncBytes:    resp.GetLastSyncBytes(),
+		Status:           proto.GetVolumeReplicationInfoResponse_Status(resp.GetStatus()),
+		StatusMessage:    resp.StatusMessage,
 	}, nil
 }