feat: add fallbackTargetRef to CRD and required changes

Author: yyewolf

config/crd/bases/http.keda.sh_httpscaledobjects.yaml

@@ -60,6 +60,26 @@ spec:
           spec:
             description: HTTPScaledObjectSpec defines the desired state of HTTPScaledObject
             properties:
+              fallbackTargetRef:
+                description: (optional) The name of the fallback service to route
+                  HTTP requests to when the target is not available
+                properties:
+                  port:
+                    description: The port to route to
+                    format: int32
+                    type: integer
+                  portName:
+                    description: The port to route to referenced by name
+                    type: string
+                  service:
+                    description: The name of the service to route to
+                    type: string
+                required:
+                - service
+                type: object
+                x-kubernetes-validations:
+                - message: must define either the 'portName' or the 'port'
+                  rule: has(self.portName) != has(self.port)
               hosts:
                 description: |-
                   The hosts to route. All requests which the "Host" header

interceptor/handler/upstream.go

@@ -13,12 +13,14 @@ var (
 )
 
 type Upstream struct {
-	roundTripper http.RoundTripper
+	roundTripper   http.RoundTripper
+	shouldFallback bool
 }
 
-func NewUpstream(roundTripper http.RoundTripper) *Upstream {
+func NewUpstream(roundTripper http.RoundTripper, shouldFallback bool) *Upstream {
 	return &Upstream{
-		roundTripper: roundTripper,
+		roundTripper:   roundTripper,
+		shouldFallback: shouldFallback,
 	}
 }
 
@@ -29,6 +31,10 @@ func (uh *Upstream) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 
 	stream := util.StreamFromContext(ctx)
+	if uh.shouldFallback {
+		stream = util.FallbackStreamFromContext(ctx)
+	}
+
 	if stream == nil {
 		sh := NewStatic(http.StatusInternalServerError, errNilStream)
 		sh.ServeHTTP(w, r)

interceptor/handler/upstream_test.go

@@ -43,7 +43,7 @@ func TestForwarderSuccess(t *testing.T) {
 	timeouts := defaultTimeouts()
 	dialCtxFunc := retryDialContextFunc(timeouts, timeouts.DefaultBackoff())
 	rt := newRoundTripper(dialCtxFunc, timeouts.ResponseHeader)
-	uh := NewUpstream(rt)
+	uh := NewUpstream(rt, false)
 	uh.ServeHTTP(res, req)
 
 	r.True(
@@ -88,7 +88,7 @@ func TestForwarderHeaderTimeout(t *testing.T) {
 	r.NoError(err)
 	req = util.RequestWithStream(req, originURL)
 	rt := newRoundTripper(dialCtxFunc, timeouts.ResponseHeader)
-	uh := NewUpstream(rt)
+	uh := NewUpstream(rt, false)
 	uh.ServeHTTP(res, req)
 
 	forwardedRequests := hdl.IncomingRequests()
@@ -138,7 +138,7 @@ func TestForwarderWaitsForSlowOrigin(t *testing.T) {
 	r.NoError(err)
 	req = util.RequestWithStream(req, originURL)
 	rt := newRoundTripper(dialCtxFunc, timeouts.ResponseHeader)
-	uh := NewUpstream(rt)
+	uh := NewUpstream(rt, false)
 	uh.ServeHTTP(res, req)
 	// wait for the goroutine above to finish, with a little cusion
 	ensureSignalBeforeTimeout(originWaitCh, originDelay*2)
@@ -161,7 +161,7 @@ func TestForwarderConnectionRetryAndTimeout(t *testing.T) {
 	r.NoError(err)
 	req = util.RequestWithStream(req, noSuchURL)
 	rt := newRoundTripper(dialCtxFunc, timeouts.ResponseHeader)
-	uh := NewUpstream(rt)
+	uh := NewUpstream(rt, false)
 
 	start := time.Now()
 	uh.ServeHTTP(res, req)
@@ -217,7 +217,7 @@ func TestForwardRequestRedirectAndHeaders(t *testing.T) {
 	r.NoError(err)
 	req = util.RequestWithStream(req, srvURL)
 	rt := newRoundTripper(dialCtxFunc, timeouts.ResponseHeader)
-	uh := NewUpstream(rt)
+	uh := NewUpstream(rt, false)
 	uh.ServeHTTP(res, req)
 	r.Equal(301, res.Code)
 	r.Equal("abc123.com", res.Header().Get("Location"))

interceptor/middleware/routing.go

@@ -57,7 +57,7 @@ func (rm *Routing) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 	r = r.WithContext(util.ContextWithHTTPSO(r.Context(), httpso))
 
-	stream, err := rm.streamFromHTTPSO(r.Context(), httpso)
+	stream, err := rm.streamFromHTTPSO(r.Context(), httpso, httpso.Spec.ScaleTargetRef)
 	if err != nil {
 		sh := handler.NewStatic(http.StatusInternalServerError, err)
 		sh.ServeHTTP(w, r)
@@ -66,45 +66,61 @@ func (rm *Routing) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 	r = r.WithContext(util.ContextWithStream(r.Context(), stream))
 
+	if httpso.Spec.FallbackTargetRef != nil {
+		fallbackStream, err := rm.streamFromHTTPSO(r.Context(), httpso, httpso.Spec.FallbackTargetRef)
+		if err != nil {
+			sh := handler.NewStatic(http.StatusInternalServerError, err)
+			sh.ServeHTTP(w, r)
+			return
+		}
+		r = r.WithContext(util.ContextWithFallbackStream(r.Context(), fallbackStream))
+	}
+
 	rm.upstreamHandler.ServeHTTP(w, r)
 }
 
-func (rm *Routing) getPort(ctx context.Context, httpso *httpv1alpha1.HTTPScaledObject) (int32, error) {
-	if httpso.Spec.ScaleTargetRef.Port != 0 {
-		return httpso.Spec.ScaleTargetRef.Port, nil
+func (rm *Routing) getPort(ctx context.Context, httpso *httpv1alpha1.HTTPScaledObject, reference httpv1alpha1.Ref) (int32, error) {
+	var (
+		port        int32  = reference.GetPort()
+		portName    string = reference.GetPortName()
+		serviceName string = reference.GetServiceName()
+	)
+
+	if port != 0 {
+		return port, nil
 	}
-	if httpso.Spec.ScaleTargetRef.PortName == "" {
+	if portName == "" {
 		return 0, fmt.Errorf(`must specify either "port" or "portName"`)
 	}
-	svc, err := rm.svcCache.Get(ctx, httpso.GetNamespace(), httpso.Spec.ScaleTargetRef.Service)
+	svc, err := rm.svcCache.Get(ctx, httpso.GetNamespace(), serviceName)
 	if err != nil {
 		return 0, fmt.Errorf("failed to get Service: %w", err)
 	}
 	for _, port := range svc.Spec.Ports {
-		if port.Name == httpso.Spec.ScaleTargetRef.PortName {
+		if port.Name == portName {
 			return port.Port, nil
 		}
 	}
-	return 0, fmt.Errorf("portName %q not found in Service", httpso.Spec.ScaleTargetRef.PortName)
+	return 0, fmt.Errorf("portName %q not found in Service", portName)
 }
 
-func (rm *Routing) streamFromHTTPSO(ctx context.Context, httpso *httpv1alpha1.HTTPScaledObject) (*url.URL, error) {
-	port, err := rm.getPort(ctx, httpso)
+func (rm *Routing) streamFromHTTPSO(ctx context.Context, httpso *httpv1alpha1.HTTPScaledObject, reference httpv1alpha1.Ref) (*url.URL, error) {
+	port, err := rm.getPort(ctx, httpso, reference)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get port: %w", err)
 	}
 	if rm.tlsEnabled {
 		return url.Parse(fmt.Sprintf(
 			"https://%s.%s:%d",
-			httpso.Spec.ScaleTargetRef.Service,
+			reference.GetServiceName(),
 			httpso.GetNamespace(),
 			port,
 		))
 	}
 	//goland:noinspection HttpUrlsUsage
 	return url.Parse(fmt.Sprintf(
 		"http://%s.%s:%d",
-		httpso.Spec.ScaleTargetRef.Service,
+		reference.GetServiceName(),
 		httpso.GetNamespace(),
 		port,
 	))

interceptor/proxy_handlers.go

@@ -65,6 +65,7 @@ func newForwardingHandler(
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
 		httpso := util.HTTPSOFromContext(ctx)
+		hasFallback := httpso.Spec.FallbackTargetRef != nil
 
 		waitFuncCtx, done := context.WithTimeout(r.Context(), fwdCfg.waitTimeout)
 		defer done()
@@ -73,7 +74,7 @@ func newForwardingHandler(
 			httpso.GetNamespace(),
 			httpso.Spec.ScaleTargetRef.Service,
 		)
-		if err != nil {
+		if err != nil && !hasFallback {
 			lggr.Error(err, "wait function failed, not forwarding request")
 			w.WriteHeader(http.StatusBadGateway)
 			if _, err := w.Write([]byte(fmt.Sprintf("error on backend (%s)", err))); err != nil {
@@ -83,7 +84,8 @@ func newForwardingHandler(
 		}
 		w.Header().Add("X-KEDA-HTTP-Cold-Start", strconv.FormatBool(isColdStart))
 
-		uh := handler.NewUpstream(roundTripper)
+		shouldFallback := hasFallback && err != nil
+		uh := handler.NewUpstream(roundTripper, shouldFallback)
 		uh.ServeHTTP(w, r)
 	})
 }

interceptor/proxy_handlers_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"log"
 	"net/http"
@@ -149,6 +150,76 @@ func TestImmediatelySuccessfulProxyTLS(t *testing.T) {
 	r.Equal("test response", res.Body.String())
 }
 
+// the proxy should successfully forward a request to the fallback when the server is not reachable
+func TestImmediatelySuccessfulFallbackProxy(t *testing.T) {
+	host := fmt.Sprintf("%s.testing", t.Name())
+	r := require.New(t)
+
+	initialStream, err := url.Parse("http://0.0.0.0:0")
+	r.NoError(err)
+
+	fallbackHdl := kedanet.NewTestHTTPHandlerWrapper(
+		http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			w.WriteHeader(200)
+			_, err := w.Write([]byte("test response"))
+			r.NoError(err)
+		}),
+	)
+	srv, fallbackURL, err := kedanet.StartTestServer(fallbackHdl)
+	r.NoError(err)
+	defer srv.Close()
+	fallbackPort, err := strconv.Atoi(fallbackURL.Port())
+	r.NoError(err)
+
+	timeouts := defaultTimeouts()
+	dialCtxFunc := retryDialContextFunc(timeouts, timeouts.DefaultBackoff())
+	waitFunc := func(ctx context.Context, _ string, _ string) (bool, error) {
+		return false, errors.New("nothing")
+	}
+	hdl := newForwardingHandler(
+		logr.Discard(),
+		dialCtxFunc,
+		waitFunc,
+		forwardingConfig{
+			waitTimeout:       0,
+			respHeaderTimeout: timeouts.ResponseHeader,
+		},
+		&tls.Config{},
+	)
+	const path = "/testfwd"
+	res, req, err := reqAndRes(path)
+	r.NoError(err)
+	req = util.RequestWithHTTPSO(req,
+		&httpv1alpha1.HTTPScaledObject{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "@" + host,
+			},
+			Spec: httpv1alpha1.HTTPScaledObjectSpec{
+				ScaleTargetRef: httpv1alpha1.ScaleTargetRef{
+					Name:    "testdepl",
+					Service: "testsvc",
+					Port:    int32(456),
+				},
+				FallbackTargetRef: &httpv1alpha1.FallbackTargetRef{
+					Service: "testsvc",
+					Port:    int32(fallbackPort),
+				},
+				TargetPendingRequests: ptr.To[int32](123),
+			},
+		},
+	)
+	req = util.RequestWithStream(req, initialStream)
+	req = util.RequestWithFallbackStream(req, fallbackURL)
+	req.Host = host
+
+	hdl.ServeHTTP(res, req)
+
+	fmt.Println(res)
+
+	r.Equal(200, res.Code, "expected response code 200")
+	r.Equal("test response", res.Body.String())
+}
+
 // the proxy should wait for a timeout and fail if there is no
 // origin to which to connect
 func TestWaitFailedConnection(t *testing.T) {

operator/apis/http/v1alpha1/httpscaledobject_types.go

@@ -20,6 +20,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+type Ref interface {
+	GetServiceName() string
+	GetPort() int32
+	GetPortName() string
+}
+
 // ScaleTargetRef contains all the details about an HTTP application to scale and route to
 type ScaleTargetRef struct {
 	// +optional
@@ -36,6 +42,40 @@ type ScaleTargetRef struct {
 	PortName string `json:"portName,omitempty"`
 }
 
+func (s ScaleTargetRef) GetServiceName() string {
+	return s.Service
+}
+
+func (s ScaleTargetRef) GetPort() int32 {
+	return s.Port
+}
+
+func (s ScaleTargetRef) GetPortName() string {
+	return s.PortName
+}
+
+// FallbackTargetRef contains all the details about an HTTP application to scale and route to
+type FallbackTargetRef struct {
+	// The name of the service to route to
+	Service string `json:"service"`
+	// The port to route to
+	Port int32 `json:"port,omitempty"`
+	// The port to route to referenced by name
+	PortName string `json:"portName,omitempty"`
+}
+
+func (s *FallbackTargetRef) GetServiceName() string {
+	return s.Service
+}
+
+func (s *FallbackTargetRef) GetPort() int32 {
+	return s.Port
+}
+
+func (s *FallbackTargetRef) GetPortName() string {
+	return s.PortName
+}
+
 // ReplicaStruct contains the minimum and maximum amount of replicas to have in the deployment
 type ReplicaStruct struct {
 	// Minimum amount of replicas to have in the deployment (Default 0)
@@ -93,6 +133,10 @@ type HTTPScaledObjectSpec struct {
 	// Including validation as a requirement to define either the PortName or the Port
 	// +kubebuilder:validation:XValidation:rule="has(self.portName) != has(self.port)",message="must define either the 'portName' or the 'port'"
 	ScaleTargetRef ScaleTargetRef `json:"scaleTargetRef"`
+	// (optional) The name of the fallback service to route HTTP requests to when the target is not available
+	// +optional
+	// +kubebuilder:validation:XValidation:rule="has(self.portName) != has(self.port)",message="must define either the 'portName' or the 'port'"
+	FallbackTargetRef *FallbackTargetRef `json:"fallbackTargetRef,omitempty"`
 	// (optional) Replica information
 	// +optional
 	Replicas *ReplicaStruct `json:"replicas,omitempty"`

pkg/util/context.go

@@ -15,6 +15,7 @@ const (
 	ckLogger contextKey = iota
 	ckHTTPSO
 	ckStream
+	ckFallbackStream
 )
 
 func ContextWithLogger(ctx context.Context, logger logr.Logger) context.Context {
@@ -43,3 +44,12 @@ func StreamFromContext(ctx context.Context) *url.URL {
 	cv, _ := ctx.Value(ckStream).(*url.URL)
 	return cv
 }
+
+func ContextWithFallbackStream(ctx context.Context, url *url.URL) context.Context {
+	return context.WithValue(ctx, ckFallbackStream, url)
+}
+
+func FallbackStreamFromContext(ctx context.Context) *url.URL {
+	cv, _ := ctx.Value(ckFallbackStream).(*url.URL)
+	return cv
+}

pkg/util/contexthttp.go

@@ -36,3 +36,10 @@ func RequestWithStream(r *http.Request, stream *url.URL) *http.Request {
 
 	return r.WithContext(ctx)
 }
+
+func RequestWithFallbackStream(r *http.Request, stream *url.URL) *http.Request {
+	ctx := r.Context()
+	ctx = ContextWithFallbackStream(ctx, stream)
+
+	return r.WithContext(ctx)
+}