[gateway] Implement Gateway status error handling (APO-387)

Ensure validation errors in the xDS pipeline bubble up to users via
Gateway and Route status conditions.

Key fixes:
- Fix status updates always being bypassed due to in-place mutation
  comparison bug (deep copy obj before mutation)
- Add status runner infrastructure for processing Gateway API resource
  status updates
- Wire up status updater through gateway server and apiserver

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: dilyevsky

pkg/apiserver/manager.go

@@ -46,6 +46,7 @@ import (
 	"github.com/apoxy-dev/apoxy/pkg/apiserver/gateway"
 	"github.com/apoxy-dev/apoxy/pkg/cryptoutils"
 	"github.com/apoxy-dev/apoxy/pkg/gateway/message"
+	statusrunner "github.com/apoxy-dev/apoxy/pkg/gateway/status/runner"
 	"github.com/apoxy-dev/apoxy/pkg/log"
 	apoxynet "github.com/apoxy-dev/apoxy/pkg/tunnel/net"
 	tunnet "github.com/apoxy-dev/apoxy/pkg/tunnel/net"
@@ -523,6 +524,18 @@ func (m *Manager) Start(
 		return fmt.Errorf("failed to wait for APIService %s: %v", corev1alpha2.GroupVersion.Group, err)
 	}
 
+	// Start the Status Runner to write Gateway API resource statuses back to the API server.
+	// This subscribes to status updates from the gateway translation pipeline and writes
+	// them back to Kubernetes.
+	log.Infof("Starting Gateway API Status Runner")
+	statusRunner := statusrunner.New(&statusrunner.Config{
+		Client:            m.manager.GetClient(),
+		ProviderResources: gwResources,
+	})
+	if err := statusRunner.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start status runner: %v", err)
+	}
+
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	g, ctx := errgroup.WithContext(ctx)

pkg/backplane/controllers/gateway.go

@@ -6,8 +6,6 @@ import (
 	"strconv"
 
 	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -154,48 +152,15 @@ func (r *GatewayReconciler) extractEnvoyListeners(gw *gatewayv1.Gateway) []*envo
 	return listeners
 }
 
-// updateGatewayStatus updates the Gateway status with conditions based on the Proxy state.
+// updateGatewayStatus updates the Gateway status based on the Proxy state.
+// Note: Gateway status conditions (Accepted, Programmed) are now managed by the
+// apiserver's status runner which subscribes to translation pipeline status updates.
+// This function is kept for potential future health-check based status updates.
 func (r *GatewayReconciler) updateGatewayStatus(ctx context.Context, gw *gatewayv1.Gateway, proxy *corev1alpha2.Proxy) error {
-	acceptedCondition := metav1.Condition{
-		Type:               string(gwapiv1.GatewayConditionAccepted),
-		Status:             metav1.ConditionTrue,
-		ObservedGeneration: gw.Generation,
-		LastTransitionTime: metav1.Now(),
-		Reason:             string(gwapiv1.GatewayReasonAccepted),
-		Message:            "Gateway accepted and associated with Proxy",
-	}
-
-	programmedCondition := metav1.Condition{
-		Type:               string(gwapiv1.GatewayConditionProgrammed),
-		Status:             metav1.ConditionTrue,
-		ObservedGeneration: gw.Generation,
-		LastTransitionTime: metav1.Now(),
-		Reason:             string(gwapiv1.GatewayReasonProgrammed),
-		Message:            "Gateway listeners are configured in Envoy",
-	}
-
-	meta.SetStatusCondition(&gw.Status.Conditions, acceptedCondition)
-	meta.SetStatusCondition(&gw.Status.Conditions, programmedCondition)
-
-	for i, l := range gw.Spec.Listeners {
-		if i < len(gw.Status.Listeners) {
-			gw.Status.Listeners[i].Name = l.Name
-			gw.Status.Listeners[i].AttachedRoutes = 0 // This would need route counting logic
-
-			// Set listener conditions
-			listenerAccepted := metav1.Condition{
-				Type:               string(gwapiv1.ListenerConditionAccepted),
-				Status:             metav1.ConditionTrue,
-				ObservedGeneration: gw.Generation,
-				LastTransitionTime: metav1.Now(),
-				Reason:             string(gwapiv1.ListenerReasonAccepted),
-				Message:            "Listener accepted",
-			}
-			meta.SetStatusCondition(&gw.Status.Listeners[i].Conditions, listenerAccepted)
-		}
-	}
-
-	return r.Status().Update(ctx, gw)
+	// Status conditions are managed by the apiserver status runner.
+	// The backplane should only update status based on health checks or
+	// runtime information that isn't available at translation time.
+	return nil
 }
 
 // SetupWithManager sets up the controller with the Manager.

pkg/gateway/gatewayapi/contexts.go

@@ -25,6 +25,39 @@ type GatewayContext struct {
 	listeners []*ListenerContext
 }
 
+// SetCondition sets a condition on the Gateway status.
+func (g *GatewayContext) SetCondition(conditionType gwapiv1.GatewayConditionType, status metav1.ConditionStatus, reason gwapiv1.GatewayConditionReason, message string) {
+	cond := metav1.Condition{
+		Type:               string(conditionType),
+		Status:             status,
+		Reason:             string(reason),
+		Message:            message,
+		ObservedGeneration: g.Generation,
+		LastTransitionTime: metav1.NewTime(time.Now()),
+	}
+
+	idx := -1
+	for i, existing := range g.Status.Conditions {
+		if existing.Type == cond.Type {
+			// return early if the condition is unchanged
+			if existing.Status == cond.Status &&
+				existing.Reason == cond.Reason &&
+				existing.Message == cond.Message &&
+				existing.ObservedGeneration == cond.ObservedGeneration {
+				return
+			}
+			idx = i
+			break
+		}
+	}
+
+	if idx > -1 {
+		g.Status.Conditions[idx] = cond
+	} else {
+		g.Status.Conditions = append(g.Status.Conditions, cond)
+	}
+}
+
 // ResetListeners resets the listener statuses and re-generates the GatewayContext
 // ListenerContexts from the Gateway spec.
 func (g *GatewayContext) ResetListeners() {

pkg/gateway/gatewayapi/runner/runner.go

@@ -2,8 +2,11 @@ package runner
 
 import (
 	"context"
+	"fmt"
 
 	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -103,6 +106,22 @@ func (r *Runner) subscribeAndTranslate(ctx context.Context) {
 					if err := val.Validate(); err != nil {
 						log.Error("unable to validate xds ir, skipped sending it", "error", err)
 						errChan <- err
+
+						// Update Gateway status to reflect validation error
+						for _, gateway := range result.Gateways {
+							gwKey := utils.NamespacedName(gateway)
+							meta.SetStatusCondition(&gateway.Status.Conditions, metav1.Condition{
+								Type:               string(gwapiv1.GatewayConditionProgrammed),
+								Status:             metav1.ConditionFalse,
+								ObservedGeneration: gateway.Generation,
+								LastTransitionTime: metav1.Now(),
+								Reason:             string(gwapiv1.GatewayReasonInvalid),
+								Message:            fmt.Sprintf("IR validation failed: %v", err),
+							})
+							r.ProviderResources.GatewayStatuses.Store(gwKey, &gateway.Status)
+							delete(statusesToDelete.GatewayStatusKeys, gwKey)
+						}
+
 						continue
 					}
 

pkg/gateway/gatewayapi/status/status.go

@@ -28,6 +28,9 @@ import (
 	gwapiv1a2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
 
 	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
+
+	gatewayv1 "github.com/apoxy-dev/apoxy/api/gateway/v1"
+	gatewayv1alpha2 "github.com/apoxy-dev/apoxy/api/gateway/v1alpha2"
 )
 
 // Update contains an all the information needed to update an object's status.
@@ -84,16 +87,20 @@ func (u *UpdateHandler) apply(update Update) {
 			return err
 		}
 
+		// Deep copy before mutation to preserve original state for comparison.
+		// This is necessary because Mutate() modifies the object in-place.
+		oldObj := obj.DeepCopyObject().(client.Object)
+
 		newObj := update.Mutator.Mutate(obj)
 
-		if isStatusEqual(obj, newObj) {
+		if isStatusEqual(oldObj, newObj) {
 			u.log.WithName(update.NamespacedName.Name).
 				WithName(update.NamespacedName.Namespace).
 				Info("status unchanged, bypassing update")
 			return nil
 		}
 
-		newObj.SetUID(obj.GetUID())
+		newObj.SetUID(oldObj.GetUID())
 
 		return u.client.Status().Update(context.Background(), newObj)
 	}); err != nil {
@@ -171,6 +178,7 @@ func (u *UpdateWriter) Send(update Update) {
 //	ClientTrafficPolicy
 //	SecurityPolicy
 //	BackendTLSPolicy
+//	Apoxy Gateway types (gatewayv1.Gateway, gatewayv1.HTTPRoute, etc.)
 func isStatusEqual(objA, objB interface{}) bool {
 	opts := cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")
 	switch a := objA.(type) {
@@ -240,6 +248,43 @@ func isStatusEqual(objA, objB interface{}) bool {
 				return true
 			}
 		}
+	// Apoxy Gateway types
+	case *gatewayv1.Gateway:
+		if b, ok := objB.(*gatewayv1.Gateway); ok {
+			if cmp.Equal(a.Status, b.Status, opts) {
+				return true
+			}
+		}
+	case *gatewayv1.HTTPRoute:
+		if b, ok := objB.(*gatewayv1.HTTPRoute); ok {
+			if cmp.Equal(a.Status, b.Status, opts) {
+				return true
+			}
+		}
+	case *gatewayv1.GRPCRoute:
+		if b, ok := objB.(*gatewayv1.GRPCRoute); ok {
+			if cmp.Equal(a.Status, b.Status, opts) {
+				return true
+			}
+		}
+	case *gatewayv1alpha2.TLSRoute:
+		if b, ok := objB.(*gatewayv1alpha2.TLSRoute); ok {
+			if cmp.Equal(a.Status, b.Status, opts) {
+				return true
+			}
+		}
+	case *gatewayv1alpha2.TCPRoute:
+		if b, ok := objB.(*gatewayv1alpha2.TCPRoute); ok {
+			if cmp.Equal(a.Status, b.Status, opts) {
+				return true
+			}
+		}
+	case *gatewayv1alpha2.UDPRoute:
+		if b, ok := objB.(*gatewayv1alpha2.UDPRoute); ok {
+			if cmp.Equal(a.Status, b.Status, opts) {
+				return true
+			}
+		}
 	}
 
 	return false

pkg/gateway/gatewayapi/translator.go

@@ -6,6 +6,7 @@
 package gatewayapi
 
 import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 
@@ -202,6 +203,10 @@ func (t *Translator) GetRelevantGateways(gateways []*gwapiv1.Gateway) []*Gateway
 			}
 			gc.ResetListeners()
 
+			// Set Accepted condition - Gateway is accepted by this controller
+			gc.SetCondition(gwapiv1.GatewayConditionAccepted, metav1.ConditionTrue,
+				gwapiv1.GatewayReasonAccepted, "Gateway accepted by controller")
+
 			relevant = append(relevant, gc)
 		}
 	}

pkg/gateway/server.go

@@ -4,9 +4,11 @@ import (
 	"context"
 
 	"google.golang.org/grpc/credentials"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	gatewayapirunner "github.com/apoxy-dev/apoxy/pkg/gateway/gatewayapi/runner"
 	"github.com/apoxy-dev/apoxy/pkg/gateway/message"
+	statusrunner "github.com/apoxy-dev/apoxy/pkg/gateway/status/runner"
 	xdsserverrunner "github.com/apoxy-dev/apoxy/pkg/gateway/xds/server/runner"
 	xdstranslator "github.com/apoxy-dev/apoxy/pkg/gateway/xds/translator"
 	xdstranslatorrunner "github.com/apoxy-dev/apoxy/pkg/gateway/xds/translator/runner"
@@ -21,6 +23,16 @@ type serverOptions struct {
 	extensionServerAddr  string
 	extensionServerCreds credentials.TransportCredentials
 	extensionFailOpen    bool
+	client               client.Client
+}
+
+// WithClient sets the Kubernetes client for status updates.
+// If provided, a status runner will be started to write Gateway API
+// resource statuses back to the API server.
+func WithClient(c client.Client) ServerOption {
+	return func(o *serverOptions) {
+		o.client = c
+	}
 }
 
 // WithExtensionServer sets the extension server for the server options. If failOpen is true,
@@ -46,6 +58,19 @@ func RunServer(ctx context.Context, resources *message.ProviderResources, opts .
 		opt(options)
 	}
 
+	// Start the Status Runner if a client is provided.
+	// It subscribes to Gateway API resource status updates and writes them
+	// back to Kubernetes.
+	if options.client != nil {
+		statusRunner := statusrunner.New(&statusrunner.Config{
+			Client:            options.client,
+			ProviderResources: resources,
+		})
+		if err := statusRunner.Start(ctx); err != nil {
+			return err
+		}
+	}
+
 	xdsIR := new(message.XdsIR)
 	// Start the GatewayAPI Translator Runner.
 	// It subscribes to the provider resources, translates it to xDS IR