interceptor: support for http header routing

Signed-off-by: Jan Wozniak <wozniak.jan@gmail.com>

Author: wozniakjan

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

@@ -60,6 +60,27 @@ spec:
           spec:
             description: HTTPScaledObjectSpec defines the desired state of HTTPScaledObject
             properties:
+              headers:
+                description: |-
+                  The custom headers used to route. Once Hosts and PathPrefixes have been matched,
+                  if at least one header in the http request matches at least one header
+                  in .spec.headers, it will be routed to the Service and Port specified in
+                  the scaleTargetRef. First header it matches with, it will be routed to.
+                  If the headers can't be matched, then use first one without .spec.headers supplied
+                  If that doesn't exist then routing will fail.
+                items:
+                  description: Header contains the definition for matching on header
+                    name and/or value
+                  properties:
+                    name:
+                      type: string
+                    value:
+                      type: string
+                  required:
+                  - name
+                  - value
+                  type: object
+                type: array
               hosts:
                 description: |-
                   The hosts to route. All requests which the "Host" header

operator/apis/http/v1alpha1/httpscaledobject_types.go

@@ -76,6 +76,12 @@ type RateMetricSpec struct {
 	Granularity metav1.Duration `json:"granularity" description:"Time granularity for rate calculation"`
 }
 
+// Header contains the definition for matching on header name and/or value
+type Header struct {
+	Name  string `json:"name"`
+	Value string `json:"value"`
+}
+
 // HTTPScaledObjectSpec defines the desired state of HTTPScaledObject
 type HTTPScaledObjectSpec struct {
 	// The hosts to route. All requests which the "Host" header
@@ -89,6 +95,14 @@ type HTTPScaledObjectSpec struct {
 	// the scaleTargetRef.
 	// +optional
 	PathPrefixes []string `json:"pathPrefixes,omitempty"`
+	// The custom headers used to route. Once Hosts and PathPrefixes have been matched,
+	// if at least one header in the http request matches at least one header
+	// in .spec.headers, it will be routed to the Service and Port specified in
+	// the scaleTargetRef. First header it matches with, it will be routed to.
+	// If the headers can't be matched, then use first one without .spec.headers supplied
+	// If that doesn't exist then routing will fail.
+	// +optional
+	Headers []Header `json:"headers,omitempty"`
 	// The name of the deployment to route HTTP requests to (and to autoscale).
 	// 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'"

operator/apis/http/v1alpha1/zz_generated.deepcopy.go

@@ -0,0 +1,35 @@
+package routing
+
+import (
+	iradix "github.com/hashicorp/go-immutable-radix/v2"
+
+	httpv1alpha1 "github.com/kedacore/http-add-on/operator/apis/http/v1alpha1"
+)
+
+type httpSOIndex struct {
+	radix *iradix.Tree[*httpv1alpha1.HTTPScaledObject]
+}
+
+func newHTTPSOIndex() *httpSOIndex {
+	return &httpSOIndex{radix: iradix.New[*httpv1alpha1.HTTPScaledObject]()}
+}
+
+func (hi *httpSOIndex) insert(key tableMemoryIndexKey, httpso *httpv1alpha1.HTTPScaledObject) (*httpSOIndex, *httpv1alpha1.HTTPScaledObject, bool) {
+	newRadix, oldVal, oldSet := hi.radix.Insert(key, httpso)
+	newHTTPSOIndex := &httpSOIndex{
+		radix: newRadix,
+	}
+	return newHTTPSOIndex, oldVal, oldSet
+}
+
+func (hi *httpSOIndex) get(key tableMemoryIndexKey) (*httpv1alpha1.HTTPScaledObject, bool) {
+	return hi.radix.Get(key)
+}
+
+func (hi *httpSOIndex) delete(key tableMemoryIndexKey) (*httpSOIndex, *httpv1alpha1.HTTPScaledObject, bool) {
+	newRadix, oldVal, oldSet := hi.radix.Delete(key)
+	newHTTPSOIndex := &httpSOIndex{
+		radix: newRadix,
+	}
+	return newHTTPSOIndex, oldVal, oldSet
+}

pkg/routing/httpso_index.go

@@ -0,0 +1,157 @@
+package routing
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	httpv1alpha1 "github.com/kedacore/http-add-on/operator/apis/http/v1alpha1"
+	"github.com/kedacore/http-add-on/pkg/k8s"
+)
+
+var _ = Describe("httpSOIndex", func() {
+	var (
+		httpso0 = &httpv1alpha1.HTTPScaledObject{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "keda-sh",
+			},
+			Spec: httpv1alpha1.HTTPScaledObjectSpec{
+				Hosts: []string{
+					"keda.sh",
+				},
+			},
+		}
+
+		httpso0NamespacedName = k8s.NamespacedNameFromObject(httpso0)
+		httpso0IndexKey       = newTableMemoryIndexKey(httpso0NamespacedName)
+
+		httpso1 = &httpv1alpha1.HTTPScaledObject{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "one-one-one-one",
+			},
+			Spec: httpv1alpha1.HTTPScaledObjectSpec{
+				Hosts: []string{
+					"1.1.1.1",
+				},
+			},
+		}
+		httpso1NamespacedName = k8s.NamespacedNameFromObject(httpso1)
+		httpso1IndexKey       = newTableMemoryIndexKey(httpso1NamespacedName)
+	)
+	Context("New", func() {
+		It("returns a httpSOIndex with initialized tree", func() {
+			index := newHTTPSOIndex()
+			Expect(index.radix).NotTo(BeNil())
+		})
+	})
+
+	Context("Get / Insert", func() {
+		It("Get on empty httpSOIndex returns nil", func() {
+			index := newHTTPSOIndex()
+			_, ok := index.get(httpso0IndexKey)
+			Expect(ok).To(BeFalse())
+		})
+		It("httpSOIndex insert will return previous object if set", func() {
+			index := newHTTPSOIndex()
+			index, prevVal, prevSet := index.insert(httpso0IndexKey, httpso0)
+			Expect(prevSet).To(BeFalse())
+			Expect(prevVal).To(BeNil())
+			httpso0Copy := httpso0.DeepCopy()
+			httpso0Copy.Name = "httpso0Copy"
+			index, prevVal, prevSet = index.insert(httpso0IndexKey, httpso0Copy)
+			Expect(prevSet).To(BeTrue())
+			Expect(prevVal).To(Equal(httpso0))
+			Expect(prevVal).ToNot(Equal(httpso0Copy))
+			httpso, ok := index.get(httpso0IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).ToNot(Equal(httpso0))
+			Expect(httpso).To(Equal(httpso0Copy))
+		})
+
+		It("httpSOIndex with new object inserted returns object", func() {
+			index := newHTTPSOIndex()
+			index, httpso, prevSet := index.insert(httpso0IndexKey, httpso0)
+			Expect(prevSet).To(BeFalse())
+			Expect(httpso).To(BeNil())
+			httpso, ok := index.get(httpso0IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).To(Equal(httpso0))
+		})
+
+		It("httpSOIndex with new object inserted retains other object", func() {
+			index := newHTTPSOIndex()
+
+			index, _, _ = index.insert(httpso0IndexKey, httpso0)
+			httpso, ok := index.get(httpso0IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).To(Equal(httpso0))
+
+			_, ok = index.get(httpso1IndexKey)
+			Expect(ok).To(BeFalse())
+
+			index, _, _ = index.insert(httpso1IndexKey, httpso1)
+			httpso, ok = index.get(httpso1IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).To(Equal(httpso1))
+
+			// httpso0 still there
+			httpso, ok = index.get(httpso0IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).To(Equal(httpso0))
+		})
+	})
+
+	Context("Get / Delete", func() {
+		It("delete on empty httpSOIndex returns nil", func() {
+			index := newHTTPSOIndex()
+			_, httpso, oldSet := index.delete(httpso0IndexKey)
+			Expect(httpso).To(BeNil())
+			Expect(oldSet).To(BeFalse())
+		})
+
+		It("double delete returns nil the second time", func() {
+			index := newHTTPSOIndex()
+			index, _, _ = index.insert(httpso0IndexKey, httpso0)
+			index, _, _ = index.insert(httpso1IndexKey, httpso1)
+			index, deletedVal, oldSet := index.delete(httpso0IndexKey)
+			Expect(deletedVal).To(Equal(httpso0))
+			Expect(oldSet).To(BeTrue())
+			_, deletedVal, oldSet = index.delete(httpso0IndexKey)
+			Expect(deletedVal).To(BeNil())
+			Expect(oldSet).To(BeFalse())
+		})
+
+		It("delete on httpSOIndex removes object ", func() {
+			index := newHTTPSOIndex()
+			index, _, _ = index.insert(httpso0IndexKey, httpso0)
+			httpso, ok := index.get(httpso0IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).To(Equal(httpso0))
+			index, deletedVal, oldSet := index.delete(httpso0IndexKey)
+			Expect(deletedVal).To(Equal(httpso0))
+			Expect(oldSet).To(BeTrue())
+			httpso, ok = index.get(httpso0IndexKey)
+			Expect(httpso).To(BeNil())
+			Expect(ok).To(BeFalse())
+		})
+
+		It("httpSOIndex delete on one object does not affect other", func() {
+			index := newHTTPSOIndex()
+
+			index, _, _ = index.insert(httpso0IndexKey, httpso0)
+			index, _, _ = index.insert(httpso1IndexKey, httpso1)
+			httpso, ok := index.get(httpso0IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).To(Equal(httpso0))
+			index, deletedVal, oldSet := index.delete(httpso1IndexKey)
+			Expect(deletedVal).To(Equal(httpso1))
+			Expect(oldSet).To(BeTrue())
+			httpso, ok = index.get(httpso0IndexKey)
+			Expect(ok).To(BeTrue())
+			Expect(httpso).To(Equal(httpso0))
+			httpso, ok = index.get(httpso1IndexKey)
+			Expect(ok).To(BeFalse())
+			Expect(httpso).To(BeNil())
+		})
+	})
+})

pkg/routing/httpso_index_test.go

@@ -0,0 +1,105 @@
+package routing
+
+import (
+	iradix "github.com/hashicorp/go-immutable-radix/v2"
+
+	httpv1alpha1 "github.com/kedacore/http-add-on/operator/apis/http/v1alpha1"
+	"github.com/kedacore/http-add-on/pkg/k8s"
+)
+
+// light wrapper around radix tree containing HTTPScaledObjectList
+// with convenience functions to manage CRUD for individual HTTPScaledObject.
+// created as an abstraction to manage complexity for tablememory implementation
+// the store is meant to map host + path keys to one or more HTTPScaledObject
+// and return one arbitrarily or route based on headers
+type httpSOStore struct {
+	radix *iradix.Tree[httpv1alpha1.HTTPScaledObjectList]
+}
+
+func newHTTPSOStore() *httpSOStore {
+	return &httpSOStore{radix: iradix.New[httpv1alpha1.HTTPScaledObjectList]()}
+}
+
+// Insert key value into httpSOStore
+// Gets old list of HTTPScaledObjectList
+// if exists appends to list and returns new httpSOStore
+// with new radix tree
+func (hs *httpSOStore) append(key Key, httpso *httpv1alpha1.HTTPScaledObject) *httpSOStore {
+	httpsoList, found := hs.radix.Get(key)
+	var newHTTPSOStore *httpSOStore
+	if !found {
+		newList := httpv1alpha1.HTTPScaledObjectList{Items: []httpv1alpha1.HTTPScaledObject{*httpso}}
+		newRadix, _, _ := hs.radix.Insert(key, newList)
+		newHTTPSOStore = &httpSOStore{
+			radix: newRadix,
+		}
+	} else {
+		found = false
+		var newList httpv1alpha1.HTTPScaledObjectList
+		for i, httpsoItem := range httpsoList.Items {
+			if httpsoItem.Name == httpso.Name && httpsoItem.Namespace == httpso.Namespace {
+				httpsoList.Items[i] = *httpso
+				found = true
+				newList = httpsoList
+				break
+			}
+		}
+		if !found {
+			newList = httpv1alpha1.HTTPScaledObjectList{Items: append(httpsoList.Items, *httpso)}
+		}
+		newRadix, _, _ := hs.radix.Insert(key, newList)
+		newHTTPSOStore = &httpSOStore{
+			radix: newRadix,
+		}
+	}
+	return newHTTPSOStore
+}
+
+func (hs *httpSOStore) insert(key Key, httpsoList httpv1alpha1.HTTPScaledObjectList) (*httpSOStore, httpv1alpha1.HTTPScaledObjectList, bool) {
+	newRadix, oldVal, ok := hs.radix.Insert(key, httpsoList)
+	newHTTPSOStore := &httpSOStore{
+		radix: newRadix,
+	}
+	return newHTTPSOStore, oldVal, ok
+}
+
+func (hs *httpSOStore) get(key Key) (httpv1alpha1.HTTPScaledObjectList, bool) {
+	return hs.radix.Get(key)
+}
+
+func (hs *httpSOStore) delete(key Key) (*httpSOStore, httpv1alpha1.HTTPScaledObjectList, bool) {
+	newRadix, oldVal, oldSet := hs.radix.Delete(key)
+	newHTTPSOStore := &httpSOStore{
+		radix: newRadix,
+	}
+	return newHTTPSOStore, oldVal, oldSet
+}
+
+// convenience function
+// retrieves all keys associated with HTTPScaledObject
+// and deletes it from every list in the store
+func (hs *httpSOStore) DeleteAllInstancesOfHTTPSO(httpso *httpv1alpha1.HTTPScaledObject) *httpSOStore {
+	httpsoNamespacedName := k8s.NamespacedNameFromObject(httpso)
+	newHTTPSOStore := &httpSOStore{radix: hs.radix}
+	keys := NewKeysFromHTTPSO(httpso)
+	for _, key := range keys {
+		httpsoList, _ := newHTTPSOStore.radix.Get(key)
+		for i, httpso := range httpsoList.Items {
+			// delete only if namespaced names match
+			if currHttpsoNamespacedName := k8s.NamespacedNameFromObject(&httpso); *httpsoNamespacedName == *currHttpsoNamespacedName {
+				httpsoList.Items = append(httpsoList.Items[:i], httpsoList.Items[i+1:]...)
+				break
+			}
+		}
+		if len(httpsoList.Items) == 0 {
+			newHTTPSOStore.radix, _, _ = newHTTPSOStore.radix.Delete(key)
+		} else {
+			newHTTPSOStore.radix, _, _ = newHTTPSOStore.radix.Insert(key, httpsoList)
+		}
+	}
+	return newHTTPSOStore
+}
+
+func (hs *httpSOStore) GetLongestPrefix(key Key) ([]byte, httpv1alpha1.HTTPScaledObjectList, bool) {
+	return hs.radix.Root().LongestPrefix(key)
+}