feat: Support adding a proxy server in between when forwarding requests to upstream (#2710)

Co-authored-by: 澄潭 <zty98751@alibaba-inc.com>

pkg/common/protocol.go

@@ -93,6 +93,15 @@ func (p Protocol) IsUnsupported() bool {
 	return p == Unsupported
 }
 
+func (p Protocol) IsSupportedByProxy() bool {
+	switch p {
+	case HTTPS:
+		return true
+	default:
+		return false
+	}
+}
+
 func (p Protocol) String() string {
 	return string(p)
 }

pkg/common/proxy.go

@@ -0,0 +1,59 @@
+// Copyright (c) 2022 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	"strings"
+)
+
+type ProxyType string
+
+const (
+	ProxyType_Unknown ProxyType = "Unknown"
+	ProxyType_HTTP    ProxyType = "HTTP"
+	ProxyType_HTTPS   ProxyType = "HTTPS"
+	ProxyType_SOCKS4  ProxyType = "SOCKS4"
+	ProxyType_SOCKS5  ProxyType = "SOCKS5"
+)
+
+func ParseProxyType(s string) ProxyType {
+	switch strings.ToLower(s) {
+	case "http":
+		return ProxyType_HTTP
+	case "https":
+		return ProxyType_HTTPS
+	case "socks4":
+		return ProxyType_SOCKS4
+	case "socks5":
+		return ProxyType_SOCKS5
+	}
+	return ProxyType_Unknown
+}
+
+func (p ProxyType) GetTransportProtocol() Protocol {
+	switch p {
+	case ProxyType_HTTP:
+		return HTTP
+	case ProxyType_HTTPS:
+		return HTTPS
+	case ProxyType_SOCKS4, ProxyType_SOCKS5:
+		return TCP
+	}
+	return Unsupported
+}
+
+func (p ProxyType) String() string {
+	return string(p)
+}

pkg/ingress/config/ingress_config.go

@@ -69,7 +69,7 @@ import (
 	"github.com/alibaba/higress/pkg/ingress/kube/wasmplugin"
 	. "github.com/alibaba/higress/pkg/ingress/log"
 	"github.com/alibaba/higress/pkg/kube"
-	"github.com/alibaba/higress/registry/memory"
+	"github.com/alibaba/higress/registry"
 	"github.com/alibaba/higress/registry/reconcile"
 )
 
@@ -340,10 +340,6 @@ func (m *IngressConfig) listFromIngressControllers(typ config.GroupVersionKind,
 			}
 			IngressLog.Infof("Append %d configmap EnvoyFilters", len(configmapEnvoyFilters))
 		}
-		if len(envoyFilters) == 0 {
-			IngressLog.Infof("resource type %s, configs number %d", typ, len(m.cachedEnvoyFilters))
-			return m.cachedEnvoyFilters
-		}
 		envoyFilters = append(envoyFilters, m.cachedEnvoyFilters...)
 		IngressLog.Infof("resource type %s, configs number %d", typ, len(envoyFilters))
 		return envoyFilters
@@ -490,6 +486,22 @@ func (m *IngressConfig) convertVirtualService(configs []common.WrapperConfig) []
 		VirtualServices:   map[string]*common.WrapperVirtualService{},
 		HTTPRoutes:        map[string][]*common.WrapperHTTPRoute{},
 		Route2Ingress:     map[string]*common.WrapperConfigWithRuleKey{},
+		ServiceWrappers:   make(map[string]*common.ServiceWrapper),
+		ProxyWrappers:     make(map[string]*common.ProxyWrapper),
+	}
+	if m.RegistryReconciler != nil {
+		for _, sew := range m.RegistryReconciler.GetAllServiceWrapper() {
+			hosts := sew.ServiceEntry.Hosts
+			if len(hosts) == 0 {
+				continue
+			}
+			for _, host := range hosts {
+				convertOptions.ServiceWrappers[host] = sew
+			}
+		}
+		for _, pw := range m.RegistryReconciler.GetAllProxyWrapper() {
+			convertOptions.ProxyWrappers[pw.ProxyName] = pw
+		}
 	}
 
 	// convert http route
@@ -669,6 +681,12 @@ func (m *IngressConfig) convertEnvoyFilter(convertOptions *common.ConvertOptions
 		}
 	}
 
+	if proxyEnvoyFilters := constructProxyEnvoyFilters(convertOptions.ProxyWrappers, convertOptions.ServiceWrappers, m.namespace); len(proxyEnvoyFilters) != 0 {
+		for _, ef := range proxyEnvoyFilters {
+			envoyFilters = append(envoyFilters, *ef)
+		}
+	}
+
 	// TODO Support other envoy filters
 
 	IngressLog.Infof("Found %d number of envoyFilters", len(envoyFilters))
@@ -1493,7 +1511,7 @@ func (m *IngressConfig) constructHttp2RpcEnvoyFilter(http2rpcConfig *annotations
 	return &config.Config{
 		Meta: config.Meta{
 			GroupVersionKind: gvk.EnvoyFilter,
-			Name:             common.CreateConvertedName(constants.IstioIngressGatewayName, http2rpcConfig.Name),
+			Name:             common.CreateConvertedName(constants.IstioIngressGatewayName, "http2rpc", http2rpcConfig.Name, "route", httpRoute.Name),
 			Namespace:        namespace,
 		},
 		Spec: &networking.EnvoyFilter{
@@ -1675,28 +1693,150 @@ func constructBasicAuthEnvoyFilter(rules *common.BasicAuthRules, namespace strin
 	}, nil
 }
 
-func QueryByName(serviceEntries []*memory.ServiceWrapper, serviceName string) (*memory.ServiceWrapper, error) {
-	IngressLog.Infof("Found http2rpc serviceEntries %s", serviceEntries)
-	for _, se := range serviceEntries {
-		if se.ServiceName == serviceName {
-			return se, nil
+func constructProxyEnvoyFilters(proxyWrappers map[string]*common.ProxyWrapper, serviceWrappers map[string]*common.ServiceWrapper, namespace string) []*config.Config {
+	var envoyFilters []*config.Config
+	for _, proxyWrapper := range proxyWrappers {
+		envoyFilters = append(envoyFilters, &config.Config{
+			Meta: config.Meta{
+				GroupVersionKind: gvk.EnvoyFilter,
+				Name:             common.CreateConvertedName(constants.IstioIngressGatewayName, "proxy", proxyWrapper.ProxyName),
+				Namespace:        namespace,
+			},
+			Spec: proxyWrapper.EnvoyFilter,
+		})
+	}
+
+	// Create a cluster for each service that uses a proxy.
+	var serviceProxyPatches []*networking.EnvoyFilter_EnvoyConfigObjectPatch
+	for _, serviceWrapper := range serviceWrappers {
+		proxyConfig := serviceWrapper.ProxyConfig
+		if proxyConfig == nil || proxyConfig.ProxyName == "" {
+			continue
+		}
+		IngressLog.Debugf("Found service %s using proxy %s", serviceWrapper.ServiceName, proxyConfig.ProxyName)
+		if err := validateServiceWrapperForProxy(serviceWrapper); err != nil {
+			IngressLog.Warnf("Service wrapper validation failed for proxy: %v", err)
+			continue
+		}
+		proxyWrapper := proxyWrappers[proxyConfig.ProxyName]
+		if proxyWrapper == nil {
+			IngressLog.Warnf("Service %s has proxy config %s, but no corresponding proxy wrapper found", serviceWrapper.ServiceName, proxyConfig.ProxyName)
+			continue
+		}
+		if !proxyConfig.UpstreamProtocol.IsSupportedByProxy() {
+			IngressLog.Warnf("Proxy %s does not support upstream protocol %s, skipping EnvoyFilter construction for service %s")
+			continue
+		}
+		if proxyWrapper.EnvoyFilter == nil {
+			IngressLog.Warnf("Proxy %s has no EnvoyFilter generated, meaning not ready for use.", proxyConfig.ProxyName)
+			continue
+		}
+		se := serviceWrapper.ServiceEntry
+		if se == nil || len(se.Hosts) == 0 || len(se.Ports) == 0 {
+			continue
+		}
+		for _, host := range se.Hosts {
+			IngressLog.Debugf("Constructing EnvoyFilter for service %s using proxy %s", host, proxyConfig.ProxyName)
+			for _, port := range se.Ports {
+				if port == nil || port.Number <= 0 {
+					continue
+				}
+				clusterName := fmt.Sprintf("outbound|%d||%s", port.Number, host)
+
+				// We need to delete the original cluster and add a new one pointing to the local proxy listener.
+				serviceProxyPatches = append(serviceProxyPatches, &networking.EnvoyFilter_EnvoyConfigObjectPatch{
+					ApplyTo: networking.EnvoyFilter_CLUSTER,
+					Match: &networking.EnvoyFilter_EnvoyConfigObjectMatch{
+						Context: networking.EnvoyFilter_GATEWAY,
+						ObjectTypes: &networking.EnvoyFilter_EnvoyConfigObjectMatch_Cluster{
+							Cluster: &networking.EnvoyFilter_ClusterMatch{
+								Name: clusterName,
+							},
+						},
+					},
+					Patch: &networking.EnvoyFilter_Patch{
+						Operation: networking.EnvoyFilter_Patch_REMOVE,
+					},
+				})
+
+				patchObj := map[string]interface{}{
+					"name":            clusterName,
+					"type":            "STATIC",
+					"connect_timeout": "10s",
+					"load_assignment": map[string]interface{}{
+						"cluster_name": clusterName,
+						"endpoints": []map[string]interface{}{
+							{
+								"lb_endpoints": []map[string]interface{}{
+									{
+										"endpoint": map[string]interface{}{
+											"address": map[string]interface{}{
+												"socket_address": map[string]interface{}{
+													"address":    "127.0.0.1",
+													"port_value": proxyWrapper.ListenerPort,
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				}
+				if proxyConfig.UpstreamProtocol.IsHTTPS() {
+					tlsTypedConfig := map[string]interface{}{
+						"@type": "type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext",
+					}
+					if proxyConfig.UpstreamSni != "" {
+						tlsTypedConfig["sni"] = proxyConfig.UpstreamSni
+					}
+					patchObj["transport_socket"] = map[string]interface{}{
+						"name":         "envoy.transport_sockets.tls",
+						"typed_config": tlsTypedConfig,
+					}
+				}
+				patchJson, _ := json.Marshal(patchObj)
+				serviceProxyPatches = append(serviceProxyPatches, &networking.EnvoyFilter_EnvoyConfigObjectPatch{
+					ApplyTo: networking.EnvoyFilter_CLUSTER,
+					Match: &networking.EnvoyFilter_EnvoyConfigObjectMatch{
+						Context: networking.EnvoyFilter_GATEWAY,
+					},
+					Patch: &networking.EnvoyFilter_Patch{
+						Operation: networking.EnvoyFilter_Patch_ADD,
+						Value:     util.BuildPatchStruct(string(patchJson)),
+					},
+				})
+			}
 		}
 	}
-	return nil, fmt.Errorf("can't find ServiceEntry by serviceName:%v", serviceName)
+	if len(serviceProxyPatches) != 0 {
+		envoyFilters = append(envoyFilters, &config.Config{
+			Meta: config.Meta{
+				GroupVersionKind: gvk.EnvoyFilter,
+				Name:             common.CreateConvertedName(constants.IstioIngressGatewayName, "service-proxy"),
+				Namespace:        namespace,
+			},
+			Spec: &networking.EnvoyFilter{
+				ConfigPatches: serviceProxyPatches,
+			},
+		})
+	}
+
+	return envoyFilters
 }
 
-func QueryRpcServiceVersion(serviceEntry *memory.ServiceWrapper, serviceName string) (string, error) {
-	IngressLog.Infof("Found http2rpc serviceEntry %s", serviceEntry)
-	IngressLog.Infof("Found http2rpc ServiceEntry %s", serviceEntry.ServiceEntry)
-	IngressLog.Infof("Found http2rpc WorkloadSelector %s", serviceEntry.ServiceEntry.WorkloadSelector)
-	IngressLog.Infof("Found http2rpc Labels %s", serviceEntry.ServiceEntry.WorkloadSelector.Labels)
-	labels := (*serviceEntry).ServiceEntry.WorkloadSelector.Labels
-	for key, value := range labels {
-		if key == "version" {
-			return value, nil
-		}
+func validateServiceWrapperForProxy(serviceWrapper *common.ServiceWrapper) error {
+	registryType := registry.ServiceRegistryType(serviceWrapper.RegistryType)
+	switch registryType {
+	case registry.DNS:
+		break
+	default:
+		return fmt.Errorf("service %s has proxy config %s, but registry type %s is not supported for proxying", serviceWrapper.ServiceName, serviceWrapper.ProxyConfig.ProxyName, registryType)
 	}
-	return "", fmt.Errorf("can't get RpcServiceVersion for serviceName:%v", serviceName)
+	if len(serviceWrapper.ServiceEntry.Endpoints) > 1 {
+		return fmt.Errorf("service %s has multiple endpoints, which is not supported for proxying with EnvoyFilter. Skipping EnvoyFilter construction", serviceWrapper.ServiceName)
+	}
+	return nil
 }
 
 func (m *IngressConfig) Run(stop <-chan struct{}) {

pkg/ingress/kube/common/controller.go

@@ -16,9 +16,8 @@ package common
 
 import (
 	"strings"
+	"time"
 
-	"github.com/alibaba/higress/pkg/cert"
-	"github.com/alibaba/higress/pkg/ingress/kube/annotations"
 	networking "istio.io/api/networking/v1alpha3"
 	"istio.io/istio/pilot/pkg/model"
 	"istio.io/istio/pkg/cluster"
@@ -26,6 +25,10 @@ import (
 	gatewaytool "istio.io/istio/pkg/config/gateway"
 	listerv1 "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
+
+	"github.com/alibaba/higress/pkg/cert"
+	"github.com/alibaba/higress/pkg/common"
+	"github.com/alibaba/higress/pkg/ingress/kube/annotations"
 )
 
 type ServiceKey struct {
@@ -120,6 +123,68 @@ type WrapperDestinationRule struct {
 	ServiceKey      ServiceKey
 }
 
+type ServiceProxyConfig struct {
+	ProxyName        string
+	UpstreamProtocol common.Protocol
+	UpstreamSni      string
+}
+
+type ServiceWrapper struct {
+	ServiceName            string
+	ServiceEntry           *networking.ServiceEntry
+	DestinationRuleWrapper *WrapperDestinationRule
+	Suffix                 string
+	RegistryType           string
+	RegistryName           string
+	ProxyConfig            *ServiceProxyConfig
+	createTime             time.Time
+}
+
+func (sew *ServiceWrapper) DeepCopy() *ServiceWrapper {
+	res := &ServiceWrapper{}
+	*res = *sew
+	res.ServiceEntry = sew.ServiceEntry.DeepCopy()
+
+	if sew.DestinationRuleWrapper != nil {
+		res.DestinationRuleWrapper = sew.DestinationRuleWrapper
+		res.DestinationRuleWrapper.DestinationRule = sew.DestinationRuleWrapper.DestinationRule.DeepCopy()
+	}
+	return res
+}
+
+func (sew *ServiceWrapper) SetCreateTime(createTime time.Time) {
+	sew.createTime = createTime
+}
+
+func (sew *ServiceWrapper) GetCreateTime() time.Time {
+	return sew.createTime
+}
+
+type ProxyWrapper struct {
+	ProxyName    string
+	ListenerPort uint32
+	EnvoyFilter  *networking.EnvoyFilter
+	createTime   time.Time
+}
+
+func (pw *ProxyWrapper) DeepCopy() *ProxyWrapper {
+	res := &ProxyWrapper{}
+	*res = *pw
+
+	if pw.EnvoyFilter != nil {
+		res.EnvoyFilter = pw.EnvoyFilter.DeepCopy()
+	}
+	return res
+}
+
+func (pw *ProxyWrapper) SetCreateTime(createTime time.Time) {
+	pw.createTime = createTime
+}
+
+func (pw *ProxyWrapper) GetCreateTime() time.Time {
+	return pw.createTime
+}
+
 type IngressController interface {
 	// RegisterEventHandler adds a handler to receive config update events for a
 	// configuration type

pkg/ingress/kube/common/model.go

@@ -169,6 +169,10 @@ type ConvertOptions struct {
 
 	Service2TrafficPolicy map[ServiceKey]*WrapperTrafficPolicy
 
+	ServiceWrappers map[string]*ServiceWrapper
+
+	ProxyWrappers map[string]*ProxyWrapper
+
 	HasDefaultBackend bool
 }