[api] Use TunnelPeerOffer.Status.Phase as aggregate Condition

Author: dilyevsky

api/core/v1alpha/tunnel_types.go

@@ -182,12 +182,22 @@ type TunnelPeerOfferSpec struct {
 	Offer *ICEOffer `json:"iceOffer,omitempty"`
 }
 
+type TunnelPeerOfferPhase string
+
+const (
+	TunnelPeerOfferPhaseConnecting TunnelPeerOfferPhase = "Connecting"
+	TunnelPeerOfferPhaseConnected  TunnelPeerOfferPhase = "Connected"
+	TunnelPeerOfferPhaseFailed     TunnelPeerOfferPhase = "Failed"
+)
+
 type TunnelPeerOfferStatus struct {
-	Conditions []metav1.Condition `json:"conditions,omitempty"`
+	// Phase is the current aggregate phase of the tunnel peer offer.
+	// It may be represented by one or more conditions.
+	Phase TunnelPeerOfferPhase `json:"phase,omitempty"`
 
-	// PeerOffer is the offer from the remote peer.
+	// Conditions is a list of conditions that apply to the tunnel peer offer.
 	// +optional
-	PeerOffer *ICEOffer `json:"peerOffer,omitempty"`
+	Conditions []metav1.Condition `json:"conditions,omitempty"`
 }
 
 var _ resource.StatusSubResource = &TunnelPeerOfferStatus{}

api/core/v1alpha/zz_generated.deepcopy.go

@@ -379,12 +379,7 @@ func isConnected(offer *corev1alpha.TunnelPeerOffer) bool {
 	if offer == nil {
 		return false
 	}
-	for _, condition := range offer.Status.Conditions {
-		if condition.Type == "Connected" && condition.Status == metav1.ConditionTrue {
-			return true
-		}
-	}
-	return false
+	return offer.Status.Phase == corev1alpha.TunnelPeerOfferPhaseConnected
 }
 
 func (t *tunnelNodeReconciler) offerPeer(

api/generated/zz_generated.openapi.go

@@ -2,20 +2,20 @@ package tunnel
 
 import (
 	"context"
-	"errors"
+	goerrors "errors"
 	"fmt"
 	"slices"
 	"sync"
 
 	"github.com/pion/ice/v4"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	clog "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
@@ -24,10 +24,6 @@ import (
 	corev1alpha "github.com/apoxy-dev/apoxy-cli/api/core/v1alpha"
 )
 
-const (
-	tunnelPeerOfferFinalizer = "tunnelpeeroffer.apoxy.dev/finalizer"
-)
-
 type tunnelPeerOfferReconciler struct {
 	client.Client
 
@@ -53,73 +49,81 @@ func (r *tunnelPeerOfferReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	log.Info("Reconciling TunnelPeerOffer")
 
 	tunnelPeerOffer := &corev1alpha.TunnelPeerOffer{}
-	if err := r.Get(ctx, req.NamespacedName, tunnelPeerOffer); err != nil {
-		return ctrl.Result{}, client.IgnoreNotFound(err)
-	}
-
-	if tunnelPeerOffer.Spec.RemoteTunnelNodeName == r.localTunnelNodeName { // Remote offer.
-		remoteName, err := getOfferOwner(ctx, r.Client, tunnelPeerOffer)
-		if err != nil {
-			return ctrl.Result{}, err
-		}
-		log.Info("Offer controlled by remote node, starting ICE negotiation", "RemotePeer", remoteName)
-
+	if err := r.Get(ctx, req.NamespacedName, tunnelPeerOffer); client.IgnoreNotFound(err) != nil {
+		return ctrl.Result{}, err
+	} else if errors.IsNotFound(err) {
+		remoteName := req.Name
 		r.mu.Lock()
-		defer r.mu.Unlock()
-		peer, ok := r.peers[remoteName]
-		if !ok { // Haven't started our end of the ICE negotiation yet.
-			return ctrl.Result{Requeue: true}, nil
-		}
-
-		remoteOffer := tunnelPeerOffer.Spec.Offer
-		if remoteOffer == nil {
-			log.Info("ICE offer not yet created")
-			return ctrl.Result{}, nil // Will re-trigger when offer is created.
+		if peer, ok := r.peers[remoteName]; ok {
+			peer.Close()
+			delete(r.peers, remoteName)
+			log.Info("Deleted TunnelPeerOffer")
 		}
+		r.mu.Unlock()
 
-		log.Info("Connecting to remote peer", "RemotePeer", remoteName)
-
-		return r.connect(ctx, remoteName, req, peer, remoteOffer)
+		return ctrl.Result{}, nil
 	}
 
-	remoteName := tunnelPeerOffer.Spec.RemoteTunnelNodeName
-
-	if tunnelPeerOffer.ObjectMeta.DeletionTimestamp.IsZero() {
-		if !controllerutil.ContainsFinalizer(tunnelPeerOffer, tunnelPeerOfferFinalizer) {
-			controllerutil.AddFinalizer(tunnelPeerOffer, tunnelPeerOfferFinalizer)
-			if err := r.Update(ctx, tunnelPeerOffer); err != nil {
+	if tunnelPeerOffer.Spec.RemoteTunnelNodeName == r.localTunnelNodeName { // Remote offer.
+		switch tunnelPeerOffer.Status.Phase {
+		case corev1alpha.TunnelPeerOfferPhaseConnected:
+			log.Info("Already connected, ignoring")
+			return ctrl.Result{}, nil
+		case corev1alpha.TunnelPeerOfferPhaseFailed:
+			log.Info("Remote peer offer is in failed, ignoring")
+			return ctrl.Result{}, nil
+		case corev1alpha.TunnelPeerOfferPhaseConnecting:
+			remoteName, err := getOfferOwner(ctx, r.Client, tunnelPeerOffer)
+			if err != nil {
 				return ctrl.Result{}, err
 			}
-		}
-	} else {
-		if controllerutil.ContainsFinalizer(tunnelPeerOffer, tunnelPeerOfferFinalizer) {
+
+			log.Info("Offer controlled by remote node, starting ICE negotiation", "RemotePeer", remoteName)
+
 			r.mu.Lock()
-			if peer, ok := r.peers[remoteName]; ok {
-				peer.Close()
-				delete(r.peers, remoteName)
+			defer r.mu.Unlock()
+			peer, ok := r.peers[remoteName]
+			if !ok { // Haven't started our end of the ICE negotiation yet.
+				return ctrl.Result{Requeue: true}, nil
 			}
-			r.mu.Unlock()
-			controllerutil.RemoveFinalizer(tunnelPeerOffer, tunnelPeerOfferFinalizer)
-			if err := r.Update(ctx, tunnelPeerOffer); err != nil {
-				return ctrl.Result{}, err
+
+			remoteOffer := tunnelPeerOffer.Spec.Offer
+			if remoteOffer == nil {
+				log.Info("ICE offer not yet created")
+				return ctrl.Result{}, nil // Will re-trigger when offer is created.
 			}
-			log.Info("Deleted TunnelPeerOffer")
-		}
 
-		log.V(1).Info("TunnelPeerOffer is being deleted")
+			log.Info("Connecting to remote peer", "RemotePeer", remoteName)
 
-		return ctrl.Result{}, nil // Already deleted, nothing to do.
+			return r.connect(ctx, remoteName, req, peer, remoteOffer)
+		default:
+			log.Info("Remote offer is in unknown state, ignoring")
+			return ctrl.Result{}, nil
+		}
 	}
 
-	log.Info("Offer controlled by local node, starting ICE negotiation", "RemotePeer", remoteName)
+	remoteName := tunnelPeerOffer.Spec.RemoteTunnelNodeName
+	log.Info("Offer controlled by local node", "RemotePeer", remoteName)
 
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	peer, ok := r.peers[remoteName]
-	if ok { // Already connected, just return.
-		return ctrl.Result{}, nil
+	if ok {
+		switch tunnelPeerOffer.Status.Phase {
+		case corev1alpha.TunnelPeerOfferPhaseConnected, corev1alpha.TunnelPeerOfferPhaseConnecting:
+			return ctrl.Result{}, nil // Already connected, or connecting, just return.
+		case corev1alpha.TunnelPeerOfferPhaseFailed:
+			log.Info("Failed state, cleaning up peer")
+			peer.Close()
+			delete(r.peers, remoteName)
+			return ctrl.Result{}, nil
+		default:
+			return ctrl.Result{}, nil
+		}
 	}
 
+	log.Info("Starting ICE negotiation", "RemotePeer", remoteName)
+
 	var err error
 	isControlling := r.localTunnelNodeName > remoteName
 	peer, err = r.bind.NewPeer(ctx, isControlling)
@@ -149,6 +153,45 @@ func (r *tunnelPeerOfferReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			log.Error(err, "Failed to update tunnel peer offer status")
 		}
 	}
+	peer.OnConnected = func() {
+		log.Info("ICE connection established")
+		if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+			var tn corev1alpha.TunnelPeerOffer
+			if err := r.Get(ctx, req.NamespacedName, &tn); err != nil {
+				return err
+			}
+			tn.Status.Conditions = append(tn.Status.Conditions, metav1.Condition{
+				Type:               "Connected",
+				Status:             metav1.ConditionTrue,
+				Reason:             "IceConnected",
+				LastTransitionTime: metav1.Now(),
+			})
+			tn.Status.Phase = corev1alpha.TunnelPeerOfferPhaseConnected
+			return r.Status().Update(ctx, &tn)
+		}); err != nil {
+			log.Error(err, "Failed to update tunnel peer offer status")
+		}
+	}
+	peer.OnDisconnected = func(msg string) {
+		log.Info("ICE connection disconnected", "Reason", msg)
+		if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+			var tn corev1alpha.TunnelPeerOffer
+			if err := r.Get(ctx, req.NamespacedName, &tn); err != nil {
+				return err
+			}
+			tn.Status.Phase = corev1alpha.TunnelPeerOfferPhaseFailed
+			tn.Status.Conditions = append(tn.Status.Conditions, metav1.Condition{
+				Type:               "ICEConnected",
+				Status:             metav1.ConditionFalse,
+				Reason:             "Failed",
+				Message:            fmt.Sprintf("Peer %s failed to connect: %v", remoteName, msg),
+				LastTransitionTime: metav1.Now(),
+			})
+			return r.Status().Update(ctx, &tn)
+		}); err != nil {
+			log.Error(err, "Failed to update tunnel peer offer status")
+		}
+	}
 	if err := peer.Init(ctx); err != nil {
 		log.Error(err, "Failed to initialize ICE peer")
 		peer.Close()
@@ -157,6 +200,17 @@ func (r *tunnelPeerOfferReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 
 	r.peers[remoteName] = peer
 
+	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		var tn corev1alpha.TunnelPeerOffer
+		if err := r.Get(ctx, req.NamespacedName, &tn); err != nil {
+			return err
+		}
+		tn.Status.Phase = corev1alpha.TunnelPeerOfferPhaseConnecting
+		return r.Status().Update(ctx, &tn)
+	}); err != nil {
+		log.Error(err, "Failed to update tunnel peer offer status")
+	}
+
 	return ctrl.Result{}, nil // Wait until the remote offer is created.
 }
 
@@ -180,19 +234,21 @@ func (r *tunnelPeerOfferReconciler) connect(
 	}
 
 	go func() {
-		if err := peer.Connect(ctx, remoteName); err != nil && !errors.Is(err, ice.ErrMultipleStart) {
+		if err := peer.Connect(ctx, remoteName); err != nil && !goerrors.Is(err, ice.ErrMultipleStart) {
 			log.Error(err, "Failed to connect to ICE peer")
 
 			if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
 				var tunnelPeerOffer corev1alpha.TunnelPeerOffer
 				if err := r.Get(ctx, req.NamespacedName, &tunnelPeerOffer); err != nil {
 					return err
 				}
+				tunnelPeerOffer.Status.Phase = corev1alpha.TunnelPeerOfferPhaseFailed
 				tunnelPeerOffer.Status.Conditions = append(tunnelPeerOffer.Status.Conditions, metav1.Condition{
-					Type:    "Connected",
-					Status:  metav1.ConditionFalse,
-					Reason:  "Failed",
-					Message: fmt.Sprintf("Peer %s failed to connect: %v", r.localTunnelNodeName, err),
+					Type:               "Connected",
+					Status:             metav1.ConditionFalse,
+					Reason:             "Failed",
+					Message:            fmt.Sprintf("Peer %s failed to connect: %v", r.localTunnelNodeName, err),
+					LastTransitionTime: metav1.Now(),
 				})
 				return r.Status().Update(ctx, &tunnelPeerOffer)
 			}); err != nil {
@@ -208,26 +264,10 @@ func (r *tunnelPeerOfferReconciler) connect(
 			peer.Close()
 
 			return
-		} else if errors.Is(err, ice.ErrMultipleStart) {
+		} else if goerrors.Is(err, ice.ErrMultipleStart) {
 			log.Info("ICE connection already established, ignoring")
 			return
 		}
-
-		if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-			var tunnelPeerOffer corev1alpha.TunnelPeerOffer
-			if err := r.Get(ctx, req.NamespacedName, &tunnelPeerOffer); err != nil {
-				return err
-			}
-			tunnelPeerOffer.Status.Conditions = append(tunnelPeerOffer.Status.Conditions, metav1.Condition{
-				Type:    "Connected",
-				Status:  metav1.ConditionTrue,
-				Reason:  "Success",
-				Message: fmt.Sprintf("Peer %s successfully connected", r.localTunnelNodeName),
-			})
-			return r.Status().Update(ctx, &tunnelPeerOffer)
-		}); err != nil {
-			log.Error(err, "Failed to update tunnel peer offer status")
-		}
 	}()
 
 	return ctrl.Result{}, nil