feat: support multi backends for httproute (#48)

Author: AlinsRan

Makefile

@@ -12,7 +12,8 @@ ENVTEST_K8S_VERSION = 1.30.0
 KIND_NAME ?= api7-ingress-cluster
 GATEAY_API_VERSION ?= v1.1.0
 
-DASHBOARD_VERSION ?= v3.2.14.2
+DASHBOARD_VERSION ?= v3.2.14.6
+TEST_TIMEOUT ?= 30m
 
 # go 
 VERSYM="github.com/api7/api7-ingress-controller/internal/version._buildVersion"
@@ -86,11 +87,11 @@ kind-e2e-test: kind-up build-image kind-load-images e2e-test
 # Utilize Kind or modify the e2e tests to load the image locally, enabling compatibility with other vendors.
 .PHONY: e2e-test
 e2e-test:
-	DASHBOARD_VERSION=$(DASHBOARD_VERSION) go test ./test/e2e/ -v -ginkgo.v
+	DASHBOARD_VERSION=$(DASHBOARD_VERSION) go test ./test/e2e/ -test.timeout=$(TEST_TIMEOUT) -v -ginkgo.v
 
 .PHONY: conformance-test
 conformance-test:
-	DASHBOARD_VERSION=v3.2.14.2 go test -v ./test/conformance -tags=conformance
+	DASHBOARD_VERSION=$(DASHBOARD_VERSION) go test -v ./test/conformance -tags=conformance
 
 .PHONY: lint
 lint: golangci-lint ## Run golangci-lint linter

api/dashboard/v1/plugin_types.go

@@ -41,8 +41,9 @@ type TrafficSplitConfigRule struct {
 // the traffic split plugin rule.
 // +k8s:deepcopy-gen=true
 type TrafficSplitConfigRuleWeightedUpstream struct {
-	UpstreamID string `json:"upstream_id,omitempty"`
-	Weight     int    `json:"weight"`
+	UpstreamID string    `json:"upstream_id,omitempty"`
+	Upstream   *Upstream `json:"upstream,omitempty"`
+	Weight     int       `json:"weight"`
 }
 
 // IPRestrictConfig is the rule config for ip-restriction plugin.

api/dashboard/v1/types.go

@@ -557,6 +557,7 @@ func NewDefaultService() *Service {
 				"managed-by": "api7-ingress-controller",
 			},
 		},
+		Plugins: make(Plugins),
 	}
 }
 
@@ -665,6 +666,23 @@ func ComposeServiceNameWithRule(namespace, name string, rule string) string {
 	return buf.String()
 }
 
+func ComposeUpstreamNameWithRule(namespace, name string, rule string) string {
+	// FIXME Use sync.Pool to reuse this buffer if the upstream
+	// name composing code path is hot.
+	var p []byte
+	plen := len(namespace) + len(name) + 2
+
+	p = make([]byte, 0, plen)
+	buf := bytes.NewBuffer(p)
+	buf.WriteString(namespace)
+	buf.WriteByte('_')
+	buf.WriteString(name)
+	buf.WriteByte('_')
+	buf.WriteString(rule)
+
+	return buf.String()
+}
+
 // ComposeExternalUpstreamName uses ApisixUpstream namespace, name to compose the upstream name.
 func ComposeExternalUpstreamName(namespace, name string) string {
 	return namespace + "_" + name

api/dashboard/v1/zz_generated.deepcopy.go

@@ -108,7 +108,7 @@ func (c *Config) validateGatewayConfig(gc *GatewayConfig) error {
 var gatewayNameMap map[string]*GatewayConfig
 var gatewayNameList []string
 
-func init_gatewayNameMap() {
+func initGatewayNameMap() {
 	if gatewayNameMap == nil {
 		gatewayNameMap = make(map[string]*GatewayConfig)
 		for _, gc := range ControllerConfig.GatewayConfigs {
@@ -118,15 +118,15 @@ func init_gatewayNameMap() {
 }
 
 func GetControlPlaneConfigByGatewatName(gatewatName string) *ControlPlaneConfig {
-	init_gatewayNameMap()
+	initGatewayNameMap()
 	if gc, ok := gatewayNameMap[gatewatName]; ok {
 		return gc.ControlPlane
 	}
 	return nil
 }
 
 func GetGatewayConfig(gatewayName string) *GatewayConfig {
-	init_gatewayNameMap()
+	initGatewayNameMap()
 	if gc, ok := gatewayNameMap[gatewayName]; ok {
 		return gc
 	}
@@ -141,7 +141,7 @@ func GetFirstGatewayConfig() *GatewayConfig {
 }
 
 func GetGatewayAddresses(gatewayName string) []string {
-	init_gatewayNameMap()
+	initGatewayNameMap()
 	if gc, ok := gatewayNameMap[gatewayName]; ok {
 		return gc.Addresses
 	}

internal/controller/config/config.go

@@ -18,8 +18,7 @@ const (
 	RequestMirror = "proxy-mirror"
 )
 
-func (t *Translator) generatePluginsFromHTTPRouteFilters(namespace string, filters []gatewayv1.HTTPRouteFilter) v1.Plugins {
-	plugins := v1.Plugins{}
+func (t *Translator) fillPluginsFromHTTPRouteFilters(plugins v1.Plugins, namespace string, filters []gatewayv1.HTTPRouteFilter) {
 	for _, filter := range filters {
 		switch filter.Type {
 		case gatewayv1.HTTPRouteFilterRequestHeaderModifier:
@@ -34,7 +33,6 @@ func (t *Translator) generatePluginsFromHTTPRouteFilters(namespace string, filte
 			t.fillPluginFromHTTPResponseHeaderFilter(plugins, filter.ResponseHeaderModifier)
 		}
 	}
-	return plugins
 }
 
 func (t *Translator) fillPluginFromHTTPRequestHeaderFilter(plugins v1.Plugins, reqHeaderModifier *gatewayv1.HTTPHeaderFilter) {
@@ -158,7 +156,7 @@ func (t *Translator) fillPluginFromHTTPRequestRedirectFilter(plugins v1.Plugins,
 	plugin.URI = uri
 }
 
-func (t *Translator) translateEndpointSlice(endpointSlices []discoveryv1.EndpointSlice, weight int) v1.UpstreamNodes {
+func (t *Translator) translateEndpointSlice(endpointSlices []discoveryv1.EndpointSlice) v1.UpstreamNodes {
 	var nodes v1.UpstreamNodes
 	if len(endpointSlices) == 0 {
 		return nodes
@@ -170,7 +168,7 @@ func (t *Translator) translateEndpointSlice(endpointSlices []discoveryv1.Endpoin
 					node := v1.UpstreamNode{
 						Host:   addr,
 						Port:   int(*port.Port),
-						Weight: weight,
+						Weight: 1,
 					}
 					nodes = append(nodes, node)
 				}
@@ -188,11 +186,7 @@ func (t *Translator) translateBackendRef(tctx *TranslateContext, ref gatewayv1.B
 		Name:      string(ref.Name),
 	}]
 
-	weight := 100
-	if ref.Weight != nil {
-		weight = int(*ref.Weight)
-	}
-	upstream.Nodes = t.translateEndpointSlice(endpointSlices, weight)
+	upstream.Nodes = t.translateEndpointSlice(endpointSlices)
 	return upstream
 }
 
@@ -207,6 +201,8 @@ func (t *Translator) TranslateGatewayHTTPRoute(tctx *TranslateContext, httpRoute
 	rules := httpRoute.Spec.Rules
 
 	for i, rule := range rules {
+
+		var weightedUpstreams []v1.TrafficSplitConfigRuleWeightedUpstream
 		upstreams := []*v1.Upstream{}
 		for _, backend := range rule.BackendRefs {
 			if backend.Namespace == nil {
@@ -226,27 +222,47 @@ func (t *Translator) TranslateGatewayHTTPRoute(tctx *TranslateContext, httpRoute
 					},
 				}
 			}
+
+			weight := 100
+			if backend.Weight != nil {
+				weight = int(*backend.Weight)
+			}
+			weightedUpstreams = append(weightedUpstreams, v1.TrafficSplitConfigRuleWeightedUpstream{
+				Upstream: upstream,
+				Weight:   weight,
+			})
 		}
 
-		service := v1.NewDefaultService()
-		if len(upstreams) > 0 {
-			service.Upstream = upstreams[0]
-			// TODO: support multiple upstreams
-		} else if len(upstreams) == 0 {
-			service.Upstream = v1.NewDefaultUpstream()
-			service.Upstream.Nodes = v1.UpstreamNodes{
+		if len(upstreams) == 0 {
+			upstream := v1.NewDefaultUpstream()
+			upstream.Nodes = v1.UpstreamNodes{
 				{
 					Host:   "0.0.0.0",
 					Port:   80,
 					Weight: 100,
 				},
 			}
+			upstreams = append(upstreams, upstream)
 		}
+
+		service := v1.NewDefaultService()
+		service.Upstream = upstreams[0]
+		if len(weightedUpstreams) > 1 {
+			weightedUpstreams[0].Upstream = nil
+			service.Plugins["traffic-split"] = &v1.TrafficSplitConfig{
+				Rules: []v1.TrafficSplitConfigRule{
+					{
+						WeightedUpstreams: weightedUpstreams,
+					},
+				},
+			}
+		}
+
 		service.Name = v1.ComposeServiceNameWithRule(httpRoute.Namespace, httpRoute.Name, fmt.Sprintf("%d", i))
 		service.ID = id.GenID(service.Name)
 		service.Labels = label.GenLabel(httpRoute)
 		service.Hosts = hosts
-		service.Plugins = t.generatePluginsFromHTTPRouteFilters(httpRoute.GetNamespace(), rule.Filters)
+		t.fillPluginsFromHTTPRouteFilters(service.Plugins, httpRoute.GetNamespace(), rule.Filters)
 
 		result.Services = append(result.Services, service)
 

internal/controlplane/translator/httproute.go

@@ -165,6 +165,7 @@ func (u *serviceClient) Delete(ctx context.Context, obj *v1.Service) error {
 
 func (u *serviceClient) Update(ctx context.Context, obj *v1.Service) (*v1.Service, error) {
 	url := u.url + "/" + obj.ID
+	log.Debugw("try to update service", zap.Any("service", obj), zap.String("url", url))
 	return updateResource(
 		ctx,
 		obj,

pkg/dashboard/service.go

@@ -19,42 +19,8 @@ data:
           return 200 'Hello, World!';
         }
       }
-    
-      server {
-        listen 8080 default_server;
-
-        location / {
-          return 500 'Internal Server Error!';
-        }
-
-      }
     }
 
-    
-    stream {
-      server {
-        listen 443 reuseport ssl;
-        ssl_certificate /etc/nginx/ssl/tls.crt;
-        ssl_certificate_key /etc/nginx/ssl/tls.key;
-        ssl_protocols  TLSv1 TLSv1.1 TLSv1.2;
-        ssl_ciphers    HIGH:!aNULL:!MD5;
-
-        proxy_pass 127.0.0.1:80;
-      }
-
-      server {
-        listen 8443 reuseport ssl;
-        ssl_certificate /etc/nginx/ssl/tls.crt;
-        ssl_certificate_key /etc/nginx/ssl/tls.key;
-        ssl_trusted_certificate /etc/nginx/ssl/ca.crt;
-        ssl_verify_client on;
-        ssl_client_certificate /etc/nginx/ssl/tls.crt;
-        ssl_protocols  TLSv1 TLSv1.1 TLSv1.2;
-        ssl_ciphers    HIGH:!aNULL:!MD5;
-
-        proxy_pass 127.0.0.1:80;
-      }
-    }
 ---
 apiVersion: apps/v1
 kind: Deployment
@@ -71,9 +37,6 @@ spec:
         app: nginx
     spec:
       volumes:
-        - name: nginx-ssl
-          secret:
-            secretName: nginx-ssl
         - name: nginx-config
           configMap:
             name: nginx-config
@@ -91,21 +54,10 @@ spec:
           imagePullPolicy: IfNotPresent
           name: nginx
           ports:
-            - containerPort: 443
-              name: "https"
-              protocol: "TCP"
-            - containerPort: 8443
-              name: "mtls"
-              protocol: "TCP"
-            - containerPort: 8080
-              name: "http1"
-              protocol: "TCP"
             - containerPort: 80
               name: "http"
               protocol: "TCP"
           volumeMounts:
-            - mountPath: /etc/nginx/ssl
-              name: nginx-ssl
             - mountPath: /etc/nginx/nginx.conf
               name: nginx-config
               subPath: nginx.conf
@@ -122,16 +74,4 @@ spec:
       port: 80
       protocol: TCP
       targetPort: 80
-    - name: http1
-      port: 8080
-      protocol: TCP
-      targetPort: 8080
-    - name: https
-      port: 443
-      protocol: TCP
-      targetPort: 443
-    - name: mtls
-      port: 8443
-      protocol: TCP
-      targetPort: 8443
   type: ClusterIP

test/e2e/framework/manifests/nginx.yaml

@@ -0,0 +1,49 @@
+package framework
+
+import (
+	"bytes"
+	_ "embed"
+	"text/template"
+
+	"github.com/Masterminds/sprig/v3"
+	"github.com/gruntwork-io/terratest/modules/k8s"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var (
+	//go:embed manifests/nginx.yaml
+	_ngxSpec   string
+	ngxSpecTpl *template.Template
+)
+
+type NginxOptions struct {
+	Namespace string
+}
+
+func init() {
+	tpl, err := template.New("ngx").Funcs(sprig.TxtFuncMap()).Parse(_ngxSpec)
+	if err != nil {
+		panic(err)
+	}
+	ngxSpecTpl = tpl
+}
+
+func (f *Framework) DeployNginx(opts NginxOptions) *corev1.Service {
+	buf := bytes.NewBuffer(nil)
+
+	err := ngxSpecTpl.Execute(buf, opts)
+	f.GomegaT.Expect(err).ToNot(HaveOccurred(), "rendering nginx spec")
+
+	kubectlOpts := k8s.NewKubectlOptions("", "", opts.Namespace)
+
+	k8s.KubectlApplyFromString(f.GinkgoT, kubectlOpts, buf.String())
+
+	err = WaitPodsAvailable(f.GinkgoT, kubectlOpts, metav1.ListOptions{
+		LabelSelector: "app=nginx",
+	})
+	Expect(err).ToNot(HaveOccurred(), "waiting for nginx pod ready")
+
+	return k8s.GetService(f.GinkgoT, kubectlOpts, "nginx")
+}