MT-Broker: return appropriate status code based on the state to lever…

…age retries The ingress or filter deployments were returning 400 even in the case where a given resource (like trigger, broker, subscription) wasn't found, however, this is a common case where the lister cache hasn't caught up with the latest state. Signed-off-by: Pierangelo Di Pilato <[email protected]>
knative · Dec 3, 2024 · a4bbf94 · a4bbf94
1 parent 795e4a3
commit a4bbf94
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 8 deletions.
diff --git a/pkg/broker/filter/filter_handler.go b/pkg/broker/filter/filter_handler.go
@@ -26,6 +26,8 @@ import (
 	"net/http"
 	"time"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
 	messaginginformers "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
 	"knative.dev/eventing/pkg/reconciler/broker/resources"
@@ -178,16 +180,14 @@ func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	}
 
 	trigger, err := h.getTrigger(triggerRef)
-	if err != nil {
-		h.logger.Info("Unable to get the Trigger", zap.Error(err), zap.Any("triggerRef", triggerRef))
-		writer.WriteHeader(http.StatusBadRequest)
+	if apierrors.IsNotFound(err) {
+		h.logger.Info("Unable to find the Trigger", zap.Error(err), zap.Any("triggerRef", triggerRef))
+		writer.WriteHeader(http.StatusNotFound)
 		return
 	}
-
-	subscription, err := h.getSubscription(features, trigger)
 	if err != nil {
-		h.logger.Info("Unable to get the Subscription of the Trigger", zap.Error(err), zap.Any("triggerRef", triggerRef))
-		writer.WriteHeader(http.StatusInternalServerError)
+		h.logger.Info("Unable to get the Trigger", zap.Error(err), zap.Any("triggerRef", triggerRef))
+		writer.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
@@ -216,6 +216,18 @@ func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	if features.IsOIDCAuthentication() {
 		h.logger.Debug("OIDC authentication is enabled")
 
+		subscription, err := h.getSubscription(features, trigger)
+		if apierrors.IsNotFound(err) {
+			h.logger.Info("Unable to find the Subscription for trigger", zap.Error(err), zap.Any("triggerRef", triggerRef))
+			writer.WriteHeader(http.StatusNotFound)
+			return
+		}
+		if err != nil {
+			h.logger.Info("Unable to get the Subscription of the Trigger", zap.Error(err), zap.Any("triggerRef", triggerRef))
+			writer.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
 		audience := FilterAudience
 
 		if subscription.Status.Auth == nil || subscription.Status.Auth.ServiceAccountName == nil {
@@ -266,6 +278,11 @@ func (h *Handler) handleDispatchToReplyRequest(ctx context.Context, trigger *eve
 	}
 
 	broker, err := h.brokerLister.Brokers(brokerNamespace).Get(brokerName)
+	if apierrors.IsNotFound(err) {
+		h.logger.Info("Unable to get the Broker", zap.Error(err))
+		writer.WriteHeader(http.StatusNotFound)
+		return
+	}
 	if err != nil {
 		h.logger.Info("Unable to get the Broker", zap.Error(err))
 		writer.WriteHeader(http.StatusBadRequest)
@@ -311,6 +328,11 @@ func (h *Handler) handleDispatchToDLSRequest(ctx context.Context, trigger *event
 		brokerNamespace = trigger.Namespace
 	}
 	broker, err := h.brokerLister.Brokers(brokerNamespace).Get(brokerName)
+	if apierrors.IsNotFound(err) {
+		h.logger.Info("Unable to get the Broker", zap.Error(err))
+		writer.WriteHeader(http.StatusNotFound)
+		return
+	}
 	if err != nil {
 		h.logger.Info("Unable to get the Broker", zap.Error(err))
 		writer.WriteHeader(http.StatusBadRequest)
@@ -331,6 +353,9 @@ func (h *Handler) handleDispatchToDLSRequest(ctx context.Context, trigger *event
 			Audience: broker.Status.DeadLetterSinkAudience,
 		}
 	}
+	if target == nil {
+		return
+	}
 
 	reportArgs := &ReportArgs{
 		ns:          trigger.Namespace,

diff --git a/pkg/broker/ingress/ingress_handler.go b/pkg/broker/ingress/ingress_handler.go
@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/utils/ptr"
 
 	opencensusclient "github.com/cloudevents/sdk-go/observability/opencensus/v2/client"
@@ -226,6 +227,11 @@ func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	}
 
 	broker, err := h.getBroker(brokerName, brokerNamespace)
+	if apierrors.IsNotFound(err) {
+		h.Logger.Warn("Failed to retrieve broker", zap.Error(err))
+		writer.WriteHeader(http.StatusNotFound)
+		return
+	}
 	if err != nil {
 		h.Logger.Warn("Failed to retrieve broker", zap.Error(err))
 		writer.WriteHeader(http.StatusBadRequest)
@@ -315,7 +321,7 @@ func (h *Handler) receive(ctx context.Context, headers http.Header, event *cloud
 	channelAddress, err := h.getChannelAddress(brokerObj)
 	if err != nil {
 		h.Logger.Warn("could not get channel address from broker", zap.Error(err))
-		return http.StatusBadRequest, kncloudevents.NoDuration
+		return http.StatusInternalServerError, kncloudevents.NoDuration
 	}
 
 	opts := []kncloudevents.SendOption{