Add status to KubernetesOperator (#467)

<!-- Thank you for contributing! Please make sure that your code changes
are covered with tests. In case of new features or big changes remember
to adjust the documentation.

In case of an existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

## What
*Describe what the change is solving*

Update status fields

Add basic registration status handling to KubernetesOperator


## How
*Describe the solution*

Use similar pattern to #464

## Breaking Changes
*Are there any breaking changes in this PR?*
No.

Author: hjkatz

api/bindings/v1alpha1/endpointbinding_types.go

@@ -125,6 +125,7 @@ type BindingEndpoint struct {
 	v6.Ref `json:",inline"`
 
 	// +kubebuilder:validation:Required
+	// +kube:validation:Enum=provisioning;bound;error;unknown
 	// +kubebuilder:default="unknown"
 	Status BindingEndpointStatus `json:"status"`
 
@@ -159,6 +160,7 @@ const (
 // +kubebuilder:printcolumn:name="URI",type="string",JSONPath=".spec.endpointURI"
 // +kubebuilder:printcolumn:name="Port",type="string",JSONPath=".spec.port"
 // +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.endpoints[0].status"
+// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`,description="Age"
 type EndpointBinding struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`

api/ngrok/v1alpha1/kubernetesoperator_types.go

@@ -80,8 +80,30 @@ type KubernetesOperatorStatus struct {
 
 	// URI is the URI for this Kubernetes Operator
 	URI string `json:"uri,omitempty"`
+
+	// RegistrationStatus is the status of the registration of this Kubernetes Operator with the ngrok API
+	// +kubebuilder:validation:Required
+	// +kube:validation:Enum=registered;error;pending
+	// +kubebuilder:default="pending"
+	RegistrationStatus string `json:"registrationStatus,omitempty"`
+
+	// RegistrationErrorCode is the returned ngrok error code
+	// +kube:validation:Optional
+	// +kubebuilder:validation:Pattern=`^NGROK_ERR_\d+$`
+	RegistrationErrorCode string `json:"registrationErrorCode,omitempty"`
+
+	// RegistrationErrorMessage is a free-form error message if the status is error
+	// +kubebuilder:validation:Optional
+	// +kubebuilder:validation:MaxLength=4096
+	RegistrationErrorMessage string `json:"errorMessage,omitempty"`
 }
 
+const (
+	KubernetesOperatorRegistrationStatusSuccess = "registered"
+	KubernetesOperatorRegistrationStatusError   = "error"
+	KubernetesOperatorRegistrationStatusPending = "pending"
+)
+
 const (
 	KubernetesOperatorFeatureIngress  = "ingress"
 	KubernetesOperatorFeatureGateway  = "gateway"
@@ -115,6 +137,7 @@ type KubernetesOperatorSpec struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
 // +kubebuilder:printcolumn:name="ID",type=string,JSONPath=`.status.id`,description="Kubernetes Operator ID"
+// +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.registrationStatus"
 // +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`,description="Age"
 
 // KubernetesOperator is the Schema for the ngrok kubernetesoperators API

helm/ngrok-operator/templates/crds/bindings.k8s.ngrok.com_endpointbindings.yaml

@@ -136,6 +136,22 @@ func (self *BaseController[T]) handleErr(op BaseControllerOp, obj T, err error)
 	return CtrlResultForErr(err)
 }
 
+// ReconcileStatus is a helper function to reconcile the status of an object and requeue on update errors
+// Note: obj must be the latest resource version from k8s api
+func (self *BaseController[T]) ReconcileStatus(ctx context.Context, obj T, origErr error) error {
+	log := ctrl.LoggerFrom(ctx)
+
+	if err := self.Kube.Status().Update(ctx, obj); err != nil {
+		self.Recorder.Event(obj, v1.EventTypeWarning, "StatusError", fmt.Sprintf("Failed to reconcile status: %s", err.Error()))
+		log.V(1).Error(err, "Failed to update status")
+		return fmt.Errorf("Error occured during status update: %w: %w", StatusError{origErr}, err)
+	}
+
+	self.Recorder.Event(obj, v1.EventTypeNormal, "Status", "Successfully reconciled status")
+	log.V(1).Info("Successfully updated status")
+	return origErr
+}
+
 // CtrlResultForErr is a helper function to convert an error into a ctrl.Result passing through ngrok error mappings
 func CtrlResultForErr(err error) (ctrl.Result, error) {
 	var nerr *ngrok.Error
@@ -151,5 +167,20 @@ func CtrlResultForErr(err error) (ctrl.Result, error) {
 		}
 	}
 
+	// if error was because of status update, requeue for 10 seconds
+	var serr *StatusError
+	if errors.As(err, &serr) {
+		return ctrl.Result{RequeueAfter: 10 * time.Second}, serr
+	}
+
 	return ctrl.Result{}, err
 }
+
+// StatusError wraps .Status().*() errors returned from k8s client
+type StatusError struct {
+	err error
+}
+
+func (e StatusError) Error() string {
+	return e.err.Error()
+}

helm/ngrok-operator/templates/crds/ngrok.k8s.ngrok.com_kubernetesoperators.yaml

@@ -160,59 +160,38 @@ func (r *EndpointBindingReconciler) SetupWithManager(mgr ctrl.Manager) error {
 // - ExternalName Target Service in the Target Namespace/Service name pointed at the Upstream Service
 // - Upstream Service in the ngrok-op namespace pointed at the Pod Forwarders
 func (r *EndpointBindingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	return r.controller.Reconcile(ctx, req, new(bindingsv1alpha1.EndpointBinding))
+	cr := &bindingsv1alpha1.EndpointBinding{}
+	if ctrlErr, err := r.controller.Reconcile(ctx, req, cr); err != nil {
+		return ctrlErr, err
+	}
+
+	// update ngrok api resource status on upsert
+	if controller.IsUpsert(cr) {
+		if err := postEndpointBindingUpdateToNgrokAPI(ctx, cr); err != nil {
+			return controller.CtrlResultForErr(r.controller.ReconcileStatus(ctx, cr, err))
+		}
+	}
+
+	// success
+	return ctrl.Result{}, nil
 }
 
 func (r *EndpointBindingReconciler) create(ctx context.Context, cr *bindingsv1alpha1.EndpointBinding) error {
 	targetService, upstreamService := r.convertEndpointBindingToServices(cr)
 
-	// defer updating statuses
-	defer func() {
-		// best effort
-		_ = r.updateEndpointBindingStatus(ctx, cr)
-
-		// best effort for now TODO(hkatz) retries?
-		_ = postEndpointBindingUpdateToNgrokAPI(ctx, cr)
-	}()
-
 	if err := r.createUpstreamService(ctx, cr, upstreamService); err != nil {
-		return err
+		return r.controller.ReconcileStatus(ctx, cr, err)
 	}
 
 	if err := r.createTargetService(ctx, cr, targetService); err != nil {
-		return err
+		return r.controller.ReconcileStatus(ctx, cr, err)
 	}
 
-	go r.tryToBindEndpointBinding(ctx, cr)
-
-	return nil
-}
-
-// updateEndpointBindingStatus updates the status of the EndpointBinding to the desired state
-func (r *EndpointBindingReconciler) updateEndpointBindingStatus(ctx context.Context, desired *bindingsv1alpha1.EndpointBinding) error {
-	log := ctrl.LoggerFrom(ctx).WithValues("name", desired.Name)
-
-	existing := desired.DeepCopy()
-	if err := r.Client.Get(ctx, client.ObjectKeyFromObject(desired), existing); err != nil {
-		log.Error(err, "Failed to get EndpointBinding for status update")
-		r.Recorder.Event(desired, v1.EventTypeWarning, "UpdateFailed", "Failed to get EndpointBinding for status update")
-		return err
+	if err := r.tryToBindEndpointBinding(ctx, cr); err != nil {
+		return r.controller.ReconcileStatus(ctx, cr, err)
 	}
 
-	// use existing
-
-	// set status to desired
-	existing.Status = desired.Status
-
-	if err := r.Client.Status().Update(ctx, existing); err != nil {
-		log.Error(err, "Failed to update EndpointBinding status")
-		r.Recorder.Event(existing, v1.EventTypeWarning, "UpdateFailed", "Failed to update EndpointBinding status")
-		return err
-	}
-
-	log.Info("Updated EndpointBinding status")
-	r.Recorder.Event(existing, v1.EventTypeNormal, "Updated", "Updated EndpointBinding status")
-	return nil
+	return r.controller.ReconcileStatus(ctx, cr, nil)
 }
 
 // setEndpointsStatus sets the status of every endpoint on endpointBinding to the desired status
@@ -251,7 +230,6 @@ func (r *EndpointBindingReconciler) createTargetService(ctx context.Context, own
 	r.Recorder.Event(service, v1.EventTypeNormal, "Created", "Created Target Service")
 	r.Recorder.Event(owner, v1.EventTypeNormal, "Created", "Created Target Service")
 	log.Info("Created Upstream Service", "service", service.Name)
-
 	return nil
 }
 
@@ -270,6 +248,7 @@ func (r *EndpointBindingReconciler) createUpstreamService(ctx context.Context, o
 
 		return err
 	}
+
 	r.Recorder.Event(service, v1.EventTypeNormal, "Created", "Created Upstream Service")
 	r.Recorder.Event(owner, v1.EventTypeNormal, "Created", "Created Upstream Service")
 	log.Info("Created Upstream Service", "service", service.Name)
@@ -285,28 +264,19 @@ func (r *EndpointBindingReconciler) update(ctx context.Context, cr *bindingsv1al
 	var existingTargetService v1.Service
 	var existingUpstreamService v1.Service
 
-	// defer updating statuses
-	defer func() {
-		// best effort
-		_ = r.updateEndpointBindingStatus(ctx, cr)
-
-		// best effort for now TODO(hkatz) retries?
-		_ = postEndpointBindingUpdateToNgrokAPI(ctx, cr)
-	}()
-
 	// upstream service
 	err := r.Get(ctx, client.ObjectKey{Namespace: desiredUpstreamService.Namespace, Name: desiredUpstreamService.Name}, &existingUpstreamService)
 	if err != nil {
 		if client.IgnoreNotFound(err) == nil {
 			// Upstream Service doesn't exist, create it
 			log.Info("Unable to find existing Upstream Service, creating...", "name", desiredUpstreamService.Name)
 			if err := r.createUpstreamService(ctx, cr, desiredUpstreamService); err != nil {
-				return err
+				return r.controller.ReconcileStatus(ctx, cr, err)
 			}
 		} else {
 			// real error
 			log.Error(err, "Failed to find existing Upstream Service", "name", cr.Name, "uri", cr.Spec.EndpointURI)
-			return err
+			return r.controller.ReconcileStatus(ctx, cr, err)
 		}
 	} else {
 		// update upstream service
@@ -319,7 +289,7 @@ func (r *EndpointBindingReconciler) update(ctx context.Context, cr *bindingsv1al
 			r.Recorder.Event(&existingUpstreamService, v1.EventTypeWarning, "UpdateFailed", "Failed to update Upstream Service")
 			r.Recorder.Event(cr, v1.EventTypeWarning, "UpdateFailed", "Failed to update Upstream Service")
 			log.Error(err, "Failed to update Upstream Service")
-			return err
+			return r.controller.ReconcileStatus(ctx, cr, err)
 		}
 		r.Recorder.Event(&existingUpstreamService, v1.EventTypeNormal, "Updated", "Updated Upstream Service")
 	}
@@ -331,12 +301,12 @@ func (r *EndpointBindingReconciler) update(ctx context.Context, cr *bindingsv1al
 			// Target Service doesn't exist, create it
 			log.Info("Unable to find existing Target Service, creating...", "name", desiredTargetService.Name)
 			if err := r.createTargetService(ctx, cr, desiredTargetService); err != nil {
-				return err
+				return r.controller.ReconcileStatus(ctx, cr, err)
 			}
 		} else {
 			// real error
 			log.Error(err, "Failed to find existing Target Service", "name", cr.Name, "uri", cr.Spec.EndpointURI)
-			return err
+			return r.controller.ReconcileStatus(ctx, cr, err)
 		}
 	} else {
 		// update target service
@@ -349,15 +319,17 @@ func (r *EndpointBindingReconciler) update(ctx context.Context, cr *bindingsv1al
 			r.Recorder.Event(&existingTargetService, v1.EventTypeWarning, "UpdateFailed", "Failed to update Target Service")
 			r.Recorder.Event(cr, v1.EventTypeWarning, "UpdateFailed", "Failed to update Target Service")
 			log.Error(err, "Failed to update Target Service")
-			return err
+			return r.controller.ReconcileStatus(ctx, cr, err)
 		}
 		r.Recorder.Event(&existingTargetService, v1.EventTypeNormal, "Updated", "Updated Target Service")
 	}
 
-	go r.tryToBindEndpointBinding(ctx, cr)
+	if err := r.tryToBindEndpointBinding(ctx, cr); err != nil {
+		return r.controller.ReconcileStatus(ctx, cr, err)
+	}
 
 	r.Recorder.Event(cr, v1.EventTypeNormal, "Updated", "Updated Services")
-	return nil
+	return r.controller.ReconcileStatus(ctx, cr, nil)
 }
 
 func (r *EndpointBindingReconciler) delete(ctx context.Context, cr *bindingsv1alpha1.EndpointBinding) error {
@@ -549,7 +521,7 @@ func (r *EndpointBindingReconciler) findEndpointBindingsForService(ctx context.C
 }
 
 // tryToBindEndpointBinding attempts a TCP connection through the provisioned services for the EndpointBinding
-func (r *EndpointBindingReconciler) tryToBindEndpointBinding(ctx context.Context, endpointBinding *bindingsv1alpha1.EndpointBinding) {
+func (r *EndpointBindingReconciler) tryToBindEndpointBinding(ctx context.Context, endpointBinding *bindingsv1alpha1.EndpointBinding) error {
 	log := ctrl.LoggerFrom(ctx).WithValues("uri", endpointBinding.Spec.EndpointURI)
 
 	retries := 5
@@ -613,10 +585,5 @@ func (r *EndpointBindingReconciler) tryToBindEndpointBinding(ctx context.Context
 
 	// set status
 	setEndpointsStatus(endpointBinding, desired)
-
-	// best effort
-	_ = r.updateEndpointBindingStatus(ctx, endpointBinding)
-
-	// best effort for now TODO(hkatz) retries?
-	_ = postEndpointBindingUpdateToNgrokAPI(ctx, endpointBinding)
+	return bindErr
 }