jiazhizhong/higress
1
0
Fork 0
mirror of https://github.com/alibaba/higress.git synced 2026-03-07 10:00:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

Supports MCP service configuration protocol and SNI, along with various other fixes. (#1369)

Browse Source
This commit is contained in:
澄潭
2024-10-09 15:54:19 +08:00
committed by GitHub
parent 3ed28f2a66
commit ecf52aecfc
23 changed files with 282 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
api/kubernetes/customresourcedefinitions.gen.yaml
View File

@@ -284,6 +284,10 @@ spec:
type: string
port:
type: integer
protocol:
type: string
sni:
type: string
type:
type: string
zkServicesPath:

29
api/networking/v1/mcp_bridge.pb.go
View File

@@ -126,6 +126,8 @@ type RegistryConfig struct {
ConsulServiceTag string `protobuf:"bytes,15,opt,name=consulServiceTag,proto3" json:"consulServiceTag,omitempty"`
ConsulRefreshInterval int64 `protobuf:"varint,16,opt,name=consulRefreshInterval,proto3" json:"consulRefreshInterval,omitempty"`
AuthSecretName string `protobuf:"bytes,17,opt,name=authSecretName,proto3" json:"authSecretName,omitempty"`
Protocol string `protobuf:"bytes,18,opt,name=protocol,proto3" json:"protocol,omitempty"`
Sni string `protobuf:"bytes,19,opt,name=sni,proto3" json:"sni,omitempty"`
}
func (x *RegistryConfig) Reset() {
@@ -279,6 +281,20 @@ func (x *RegistryConfig) GetAuthSecretName() string {
return ""
}
func (x *RegistryConfig) GetProtocol() string {
if x != nil {
return x.Protocol
}
return ""
}
func (x *RegistryConfig) GetSni() string {
if x != nil {
return x.Sni
}
return ""
}
var File_networking_v1_mcp_bridge_proto protoreflect.FileDescriptor
var file_networking_v1_mcp_bridge_proto_rawDesc = []byte{
@@ -292,7 +308,7 @@ var file_networking_v1_mcp_bridge_proto_rawDesc = []byte{
0x69, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x25, 0x2e, 0x68, 0x69, 0x67, 0x72,
0x65, 0x73, 0x73, 0x2e, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x69, 0x6e, 0x67, 0x2e, 0x76,
0x31, 0x2e, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72, 0x79, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67,
0x52, 0x0a, 0x72, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72, 0x69, 0x65, 0x73, 0x22, 0xa5, 0x05, 0x0a,
0x52, 0x0a, 0x72, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72, 0x69, 0x65, 0x73, 0x22, 0xd3, 0x05, 0x0a,
0x0e, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72, 0x79, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12,
0x17, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x42, 0x03, 0xe0,
0x41, 0x02, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65,
@@ -335,10 +351,13 @@ var file_networking_v1_mcp_bridge_proto_rawDesc = []byte{
0x72, 0x65, 0x73, 0x68, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x76, 0x61, 0x6c, 0x12, 0x26, 0x0a, 0x0e,
0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x11,
0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74,
0x4e, 0x61, 0x6d, 0x65, 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
0x6f, 0x6d, 0x2f, 0x61, 0x6c, 0x69, 0x62, 0x61, 0x62, 0x61, 0x2f, 0x68, 0x69, 0x67, 0x72, 0x65,
0x73, 0x73, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x69, 0x6e,
0x67, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c,
0x18, 0x12, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c,
0x12, 0x10, 0x0a, 0x03, 0x73, 0x6e, 0x69, 0x18, 0x13, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x73,
0x6e, 0x69, 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d,
0x2f, 0x61, 0x6c, 0x69, 0x62, 0x61, 0x62, 0x61, 0x2f, 0x68, 0x69, 0x67, 0x72, 0x65, 0x73, 0x73,
0x2f, 0x61, 0x70, 0x69, 0x2f, 0x6e, 0x65, 0x74, 0x77, 0x6f, 0x72, 0x6b, 0x69, 0x6e, 0x67, 0x2f,
0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (

2
api/networking/v1/mcp_bridge.proto
View File

@@ -64,4 +64,6 @@ message RegistryConfig {
string consulServiceTag = 15;
int64 consulRefreshInterval = 16;
string authSecretName = 17;
string protocol = 18;
string sni = 19;
}

2
envoy/envoy

Submodule envoy/envoy updated: 9c9c3b717c...b3541845c1

5
helm/core/crds/customresourcedefinitions.gen.yaml
View File

@@ -284,6 +284,10 @@ spec:
type: string
port:
type: integer
protocol:
type: string
sni:
type: string
type:
type: string
zkServicesPath:
@@ -302,3 +306,4 @@ spec:
subresources:
status: {}
---

4
helm/core/templates/_pod.tpl
View File

@@ -180,7 +180,7 @@ template:
{{- end }}
- name: config
mountPath: /etc/istio/config
- name: istio-ca-root-cert
- name: higress-ca-root-cert
mountPath: /var/run/secrets/istio
- name: istio-data
mountPath: /var/lib/istio/data
@@ -266,7 +266,7 @@ template:
expirationSeconds: 43200
path: istio-token
{{- end }}
- name: istio-ca-root-cert
- name: higress-ca-root-cert
configMap:
{{- if .Values.global.enableHigressIstio }}
name: istio-ca-root-cert

2
istio/istio

Submodule istio/istio updated: 8918eb802a...dae7ac29f4

22
pkg/common/protocol.go
View File

@@ -21,7 +21,10 @@ type Protocol string
const (
TCP Protocol = "TCP"
HTTP Protocol = "HTTP"
HTTP2 Protocol = "HTTP2"
HTTPS Protocol = "HTTPS"
GRPC Protocol = "GRPC"
GRPCS Protocol = "GRPCS"
Dubbo Protocol = "Dubbo"
Unsupported Protocol = "UnsupportedProtocol"
)
@@ -32,8 +35,14 @@ func ParseProtocol(s string) Protocol {
return TCP
case "http":
return HTTP
case "https":
return HTTPS
case "http2":
return HTTP2
case "grpc", "triple", "tri":
return GRPC
case "grpcs":
return GRPCS
case "dubbo":
return Dubbo
}
@@ -51,7 +60,7 @@ func (p Protocol) IsTCP() bool {
func (p Protocol) IsHTTP() bool {
switch p {
case HTTP, GRPC:
case HTTP, GRPC, GRPCS, HTTP2, HTTPS:
return true
default:
return false
@@ -60,7 +69,16 @@ func (p Protocol) IsHTTP() bool {
func (p Protocol) IsGRPC() bool {
switch p {
case GRPC:
case GRPC, GRPCS:
return true
default:
return false
}
}
func (i Protocol) IsHTTPS() bool {
switch i {
case HTTPS, GRPCS:
return true
default:
return false

4
pkg/config/constants/constants.go
View File

@@ -23,3 +23,7 @@ const KnativeIngressCRDName = "ingresses.networking.internal.knative.dev"
const KnativeServicesCRDName = "services.serving.knative.dev"
const ManagedGatewayController = "higress.io/gateway-controller"
const RegistryTypeLabelKey = "higress-registry-type"
const RegistryNameLabelKey = "higress-registry-name"

55
pkg/ingress/config/ingress_config.go
View File

@@ -53,6 +53,7 @@ import (
extlisterv1 "github.com/alibaba/higress/client/pkg/listers/extensions/v1alpha1"
netlisterv1 "github.com/alibaba/higress/client/pkg/listers/networking/v1"
"github.com/alibaba/higress/pkg/cert"
higressconst "github.com/alibaba/higress/pkg/config/constants"
"github.com/alibaba/higress/pkg/ingress/kube/annotations"
"github.com/alibaba/higress/pkg/ingress/kube/common"
"github.com/alibaba/higress/pkg/ingress/kube/configmap"
@@ -628,8 +629,8 @@ func (m *IngressConfig) convertServiceEntry([]common.WrapperConfig) []config.Con
if m.RegistryReconciler == nil {
return nil
}
serviceEntries := m.RegistryReconciler.GetAllServiceEntryWrapper()
IngressLog.Infof("Found http2rpc serviceEntries %s", serviceEntries)
serviceEntries := m.RegistryReconciler.GetAllServiceWrapper()
IngressLog.Infof("Found mcp serviceEntries %v", serviceEntries)
out := make([]config.Config, 0, len(serviceEntries))
for _, se := range serviceEntries {
out = append(out, config.Config{
@@ -638,6 +639,10 @@ func (m *IngressConfig) convertServiceEntry([]common.WrapperConfig) []config.Con
Name: se.ServiceEntry.Hosts[0],
Namespace: "mcp",
CreationTimestamp: se.GetCreateTime(),
Labels: map[string]string{
higressconst.RegistryTypeLabelKey: se.RegistryType,
higressconst.RegistryNameLabelKey: se.RegistryName,
},
},
Spec: se.ServiceEntry,
})
@@ -703,6 +708,32 @@ func (m *IngressConfig) convertDestinationRule(configs []common.WrapperConfig) [
destinationRules[serviceName] = dr
}
if m.RegistryReconciler != nil {
drws := m.RegistryReconciler.GetAllDestinationRuleWrapper()
IngressLog.Infof("Found mcp destinationRules: %v", drws)
for _, destinationRuleWrapper := range drws {
serviceName := destinationRuleWrapper.ServiceKey.ServiceFQDN
dr, exist := destinationRules[serviceName]
if !exist {
destinationRules[serviceName] = destinationRuleWrapper
} else if dr.DestinationRule.TrafficPolicy != nil {
portTrafficPolicy := destinationRuleWrapper.DestinationRule.TrafficPolicy.PortLevelSettings[0]
portUpdated := false
for _, portTrafficPolicy := range dr.DestinationRule.TrafficPolicy.PortLevelSettings {
if portTrafficPolicy.Port.Number == portTrafficPolicy.Port.Number {
portTrafficPolicy.Tls = portTrafficPolicy.Tls
portUpdated = true
break
}
}
if portUpdated {
continue
}
dr.DestinationRule.TrafficPolicy.PortLevelSettings = append(dr.DestinationRule.TrafficPolicy.PortLevelSettings, portTrafficPolicy)
}
}
}
out := make([]config.Config, 0, len(destinationRules))
for _, dr := range destinationRules {
sort.SliceStable(dr.DestinationRule.TrafficPolicy.PortLevelSettings, func(i, j int) bool {
@@ -727,6 +758,7 @@ func (m *IngressConfig) convertDestinationRule(configs []common.WrapperConfig) [
Spec: dr.DestinationRule,
})
}
return out
}
@@ -1034,16 +1066,27 @@ func (m *IngressConfig) AddOrUpdateMcpBridge(clusterNamespacedName util.ClusterN
}
if m.RegistryReconciler == nil {
m.RegistryReconciler = reconcile.NewReconciler(func() {
metadata := config.Meta{
seMetadata := config.Meta{
Name: "mcpbridge-serviceentry",
Namespace: m.namespace,
GroupVersionKind: gvk.ServiceEntry,
// Set this label so that we do not compare configs and just push.
Labels: map[string]string{constants.AlwaysPushLabel: "true"},
}
drMetadata := config.Meta{
Name: "mcpbridge-destinationrule",
Namespace: m.namespace,
GroupVersionKind: gvk.DestinationRule,
// Set this label so that we do not compare configs and just push.
Labels: map[string]string{constants.AlwaysPushLabel: "true"},
}
for _, f := range m.serviceEntryHandlers {
IngressLog.Debug("McpBridge triggerd serviceEntry update")
f(config.Config{Meta: metadata}, config.Config{Meta: metadata}, istiomodel.EventUpdate)
f(config.Config{Meta: seMetadata}, config.Config{Meta: seMetadata}, istiomodel.EventUpdate)
}
for _, f := range m.destinationRuleHandlers {
IngressLog.Debug("McpBridge triggerd destinationRule update")
f(config.Config{Meta: drMetadata}, config.Config{Meta: drMetadata}, istiomodel.EventUpdate)
}
}, m.localKubeClient, m.namespace)
}
@@ -1489,7 +1532,7 @@ func constructBasicAuthEnvoyFilter(rules *common.BasicAuthRules, namespace strin
}, nil
}
func QueryByName(serviceEntries []*memory.ServiceEntryWrapper, serviceName string) (*memory.ServiceEntryWrapper, error) {
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 {
@@ -1499,7 +1542,7 @@ func QueryByName(serviceEntries []*memory.ServiceEntryWrapper, serviceName strin
return nil, fmt.Errorf("can't find ServiceEntry by serviceName:%v", serviceName)
}
func QueryRpcServiceVersion(serviceEntry *memory.ServiceEntryWrapper, serviceName string) (string, error) {
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)

9
pkg/ingress/kube/common/controller.go
View File

@@ -52,6 +52,15 @@ type WrapperGateway struct {
Host string
}
func CreateMcpServiceKey(host string, portNumber int32) ServiceKey {
return ServiceKey{
Namespace: "mcp",
Name: host,
ServiceFQDN: host,
Port: portNumber,
}
}
func (w *WrapperGateway) IsHTTPS() bool {
if w.Gateway == nil || len(w.Gateway.Servers) == 0 {
return false

7
pkg/ingress/kube/ingress/controller.go
View File

@@ -920,12 +920,7 @@ func (c *controller) storeBackendTrafficPolicy(wrapper *common.WrapperConfig, ba
if common.ValidateBackendResource(backend.Resource) && wrapper.AnnotationsConfig.Destination != nil {
for _, dest := range wrapper.AnnotationsConfig.Destination.McpDestination {
portNumber := dest.Destination.GetPort().GetNumber()
serviceKey := common.ServiceKey{
Namespace: "mcp",
Name: dest.Destination.Host,
Port: int32(portNumber),
ServiceFQDN: dest.Destination.Host,
}
serviceKey := common.CreateMcpServiceKey(dest.Destination.Host, int32(portNumber))
if _, exist := store[serviceKey]; !exist {
if serviceKey.Port != 0 {
store[serviceKey] = &common.WrapperTrafficPolicy{

7
pkg/ingress/kube/ingressv1/controller.go
View File

@@ -900,12 +900,7 @@ func (c *controller) storeBackendTrafficPolicy(wrapper *common.WrapperConfig, ba
if common.ValidateBackendResource(backend.Resource) && wrapper.AnnotationsConfig.Destination != nil {
for _, dest := range wrapper.AnnotationsConfig.Destination.McpDestination {
portNumber := dest.Destination.GetPort().GetNumber()
serviceKey := common.ServiceKey{
Namespace: "mcp",
Name: dest.Destination.Host,
Port: int32(portNumber),
ServiceFQDN: dest.Destination.Host,
}
serviceKey := common.CreateMcpServiceKey(dest.Destination.Host, int32(portNumber))
if _, exist := store[serviceKey]; !exist {
if serviceKey.Port != 0 {
store[serviceKey] = &common.WrapperTrafficPolicy{

10
pkg/ingress/mcp/generator.go
View File

@@ -64,7 +64,7 @@ func (c ServiceEntryGenerator) Generate(proxy *model.Proxy, w *model.WatchedReso
return serviceEntries[i].CreationTimestamp.Before(serviceEntries[j].CreationTimestamp)
})
}
return generate(proxy, serviceEntries, w, updates, false, false)
return generate(proxy, serviceEntries, w, updates, c.GeneratorOptions.KeepConfigLabels, c.GeneratorOptions.KeepConfigAnnotations)
}
func (c ServiceEntryGenerator) GenerateDeltas(proxy *model.Proxy, updates *model.PushRequest,
@@ -82,7 +82,7 @@ type VirtualServiceGenerator struct {
func (c VirtualServiceGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource,
updates *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
virtualServices := c.Environment.List(gvk.VirtualService, model.NamespaceAll)
return generate(proxy, virtualServices, w, updates, false, false)
return generate(proxy, virtualServices, w, updates, c.GeneratorOptions.KeepConfigLabels, c.GeneratorOptions.KeepConfigAnnotations)
}
func (c VirtualServiceGenerator) GenerateDeltas(proxy *model.Proxy, updates *model.PushRequest,
@@ -100,7 +100,7 @@ type DestinationRuleGenerator struct {
func (c DestinationRuleGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource,
updates *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
rules := c.Environment.List(gvk.DestinationRule, model.NamespaceAll)
return generate(proxy, rules, w, updates, false, false)
return generate(proxy, rules, w, updates, c.GeneratorOptions.KeepConfigLabels, c.GeneratorOptions.KeepConfigAnnotations)
}
func (c DestinationRuleGenerator) GenerateDeltas(proxy *model.Proxy, updates *model.PushRequest,
@@ -118,7 +118,7 @@ type EnvoyFilterGenerator struct {
func (c EnvoyFilterGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource,
updates *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
filters := c.Environment.List(gvk.EnvoyFilter, model.NamespaceAll)
return generate(proxy, filters, w, updates, false, false)
return generate(proxy, filters, w, updates, c.GeneratorOptions.KeepConfigLabels, c.GeneratorOptions.KeepConfigAnnotations)
}
func (c EnvoyFilterGenerator) GenerateDeltas(proxy *model.Proxy, updates *model.PushRequest,
@@ -154,7 +154,7 @@ type WasmPluginGenerator struct {
func (c WasmPluginGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource,
updates *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
wasmPlugins := c.Environment.List(gvk.WasmPlugin, model.NamespaceAll)
return generate(proxy, wasmPlugins, w, updates, false, false)
return generate(proxy, wasmPlugins, w, updates, c.GeneratorOptions.KeepConfigLabels, c.GeneratorOptions.KeepConfigAnnotations)
}
func (c WasmPluginGenerator) GenerateDeltas(proxy *model.Proxy, push *model.PushContext, updates *model.PushRequest,

7
registry/consul/watcher.go
View File

@@ -237,7 +237,7 @@ func (w *watcher) Stop() {
// clean the cache
suffix := strings.Join([]string{serviceName, w.ConsulDatacenter, w.Type}, common.DotSeparator)
host := strings.ReplaceAll(suffix, common.Underscore, common.Hyphen)
w.cache.DeleteServiceEntryWrapper(host)
w.cache.DeleteServiceWrapper(host)
}
w.isStop = true
close(w.stop)
@@ -295,15 +295,16 @@ func (w *watcher) getSubscribeCallback(serviceName string) func(idx uint64, data
serviceEntry := w.generateServiceEntry(host, services)
if serviceEntry != nil {
log.Infof("consul update serviceEntry %s cache", host)
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceEntry: serviceEntry,
ServiceName: serviceName,
Suffix: suffix,
RegistryType: w.Type,
RegistryName: w.Name,
})
} else {
log.Infof("consul serviceEntry %s is nil", host)
//w.cache.DeleteServiceEntryWrapper(host)
//w.cache.DeleteServiceWrapper(host)
}
}
}

75
registry/direct/watcher.go
View File

@@ -22,14 +22,15 @@ import (
"sync"
"istio.io/api/networking/v1alpha3"
"istio.io/istio/pkg/config/protocol"
"istio.io/pkg/log"
apiv1 "github.com/alibaba/higress/api/networking/v1"
"github.com/alibaba/higress/pkg/common"
ingress "github.com/alibaba/higress/pkg/ingress/kube/common"
"github.com/alibaba/higress/registry"
provider "github.com/alibaba/higress/registry"
"github.com/alibaba/higress/registry/memory"
"github.com/go-errors/errors"
)
type watcher struct {
@@ -48,6 +49,9 @@ func NewWatcher(cache memory.Cache, opts ...WatcherOption) (provider.Watcher, er
for _, opt := range opts {
opt(w)
}
if common.ParseProtocol(w.Protocol) == common.Unsupported {
return nil, errors.Errorf("invalid protocol:%s", w.Protocol)
}
return w, nil
}
@@ -75,17 +79,42 @@ func WithPort(port uint32) WatcherOption {
}
}
func WithProtocol(protocol string) WatcherOption {
return func(w *watcher) {
w.Protocol = protocol
if w.Protocol == "" {
w.Protocol = string(common.HTTP)
}
}
}
func WithSNI(sni string) WatcherOption {
return func(w *watcher) {
w.Sni = sni
}
}
func (w *watcher) Run() {
w.mutex.Lock()
defer w.mutex.Unlock()
host := strings.Join([]string{w.Name, w.Type}, common.DotSeparator)
serviceEntry := w.generateServiceEntry(host)
if serviceEntry != nil {
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
ServiceName: w.Name,
ServiceEntry: serviceEntry,
Suffix: w.Type,
RegistryType: w.Type,
var destinationRuleWrapper *ingress.WrapperDestinationRule
destinationRule := w.generateDestinationRule(serviceEntry)
if destinationRule != nil {
destinationRuleWrapper = &ingress.WrapperDestinationRule{
DestinationRule: destinationRule,
ServiceKey: ingress.CreateMcpServiceKey(host, int32(w.Port)),
}
}
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceName: w.Name,
ServiceEntry: serviceEntry,
Suffix: w.Type,
RegistryType: w.Type,
RegistryName: w.Name,
DestinationRuleWrapper: destinationRuleWrapper,
})
w.UpdateService()
}
@@ -96,7 +125,7 @@ func (w *watcher) Stop() {
w.mutex.Lock()
defer w.mutex.Unlock()
host := strings.Join([]string{w.Name, w.Type}, common.DotSeparator)
w.cache.DeleteServiceEntryWrapper(host)
w.cache.DeleteServiceWrapper(host)
w.Ready(false)
}
@@ -146,8 +175,8 @@ func (w *watcher) generateServiceEntry(host string) *v1alpha3.ServiceEntry {
var ports []*v1alpha3.ServicePort
ports = append(ports, &v1alpha3.ServicePort{
Number: w.Port,
Name: "http",
Protocol: string(protocol.HTTP),
Name: w.Protocol,
Protocol: string(common.ParseProtocol(w.Protocol)),
})
se := &v1alpha3.ServiceEntry{
Hosts: []string{host},
@@ -163,6 +192,34 @@ func (w *watcher) generateServiceEntry(host string) *v1alpha3.ServiceEntry {
return se
}
func (w *watcher) generateDestinationRule(se *v1alpha3.ServiceEntry) *v1alpha3.DestinationRule {
if !common.Protocol(se.Ports[0].Protocol).IsHTTPS() {
return nil
}
sni := w.Sni
// DNS type, automatically sets SNI based on domain name.
if sni == "" && w.Type == string(registry.DNS) && len(se.Endpoints) == 1 {
sni = w.Domain
}
return &v1alpha3.DestinationRule{
Host: se.Hosts[0],
TrafficPolicy: &v1alpha3.TrafficPolicy{
PortLevelSettings: []*v1alpha3.TrafficPolicy_PortTrafficPolicy{
&v1alpha3.TrafficPolicy_PortTrafficPolicy{
Port: &v1alpha3.PortSelector{
Number: se.Ports[0].Number,
},
Tls: &v1alpha3.ClientTLSSettings{
Mode: v1alpha3.ClientTLSSettings_SIMPLE,
Sni: sni,
},
},
},
},
}
}
func (w *watcher) GetRegistryType() string {
return w.RegistryConfig.Type
}

7
registry/eureka/watcher.go
View File

@@ -147,7 +147,7 @@ func (w *watcher) Stop() {
log.Errorf("Failed to unsubscribe service : %v", serviceName)
continue
}
w.cache.DeleteServiceEntryWrapper(makeHost(serviceName))
w.cache.DeleteServiceWrapper(makeHost(serviceName))
}
w.UpdateService()
}
@@ -203,17 +203,18 @@ func (w *watcher) subscribe(service *fargo.Application) error {
if err != nil {
return err
}
w.cache.UpdateServiceEntryWrapper(makeHost(service.Name), &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(makeHost(service.Name), &memory.ServiceWrapper{
ServiceName: service.Name,
ServiceEntry: se,
Suffix: suffix,
RegistryType: w.Type,
RegistryName: w.Name,
})
return nil
}
if w.updateCacheWhenEmpty {
w.cache.DeleteServiceEntryWrapper(makeHost(service.Name))
w.cache.DeleteServiceWrapper(makeHost(service.Name))
}
return nil

57
registry/memory/cache.go
View File

@@ -24,26 +24,28 @@ import (
"istio.io/pkg/log"
"github.com/alibaba/higress/pkg/common"
ingress "github.com/alibaba/higress/pkg/ingress/kube/common"
)
type Cache interface {
UpdateServiceEntryWrapper(service string, data *ServiceEntryWrapper)
DeleteServiceEntryWrapper(service string)
UpdateServiceWrapper(service string, data *ServiceWrapper)
DeleteServiceWrapper(service string)
PurgeStaleService()
UpdateServiceEntryEndpointWrapper(service, ip, regionId, zoneId, protocol string, labels map[string]string)
GetServiceByEndpoints(requestVersions, endpoints map[string]bool, versionKey string, protocol common.Protocol) map[string][]string
GetAllServiceEntry() []*v1alpha3.ServiceEntry
GetAllServiceEntryWrapper() []*ServiceEntryWrapper
GetIncrementalServiceEntryWrapper() (updatedList []*ServiceEntryWrapper, deletedList []*ServiceEntryWrapper)
GetAllServiceWrapper() []*ServiceWrapper
GetAllDestinationRuleWrapper() []*ingress.WrapperDestinationRule
GetIncrementalServiceWrapper() (updatedList []*ServiceWrapper, deletedList []*ServiceWrapper)
RemoveEndpointByIp(ip string)
}
func NewCache() Cache {
return &store{
mux: &sync.RWMutex{},
sew: make(map[string]*ServiceEntryWrapper),
toBeUpdated: make([]*ServiceEntryWrapper, 0),
toBeDeleted: make([]*ServiceEntryWrapper, 0),
sew: make(map[string]*ServiceWrapper),
toBeUpdated: make([]*ServiceWrapper, 0),
toBeDeleted: make([]*ServiceWrapper, 0),
ip2services: make(map[string]map[string]bool),
deferedDelete: make(map[string]struct{}),
}
@@ -51,9 +53,9 @@ func NewCache() Cache {
type store struct {
mux *sync.RWMutex
sew map[string]*ServiceEntryWrapper
toBeUpdated []*ServiceEntryWrapper
toBeDeleted []*ServiceEntryWrapper
sew map[string]*ServiceWrapper
toBeUpdated []*ServiceWrapper
toBeDeleted []*ServiceWrapper
ip2services map[string]map[string]bool
deferedDelete map[string]struct{}
}
@@ -94,7 +96,7 @@ func (s *store) UpdateServiceEntryEndpointWrapper(service, ip, regionId, zoneId,
return
}
func (s *store) UpdateServiceEntryWrapper(service string, data *ServiceEntryWrapper) {
func (s *store) UpdateServiceWrapper(service string, data *ServiceWrapper) {
s.mux.Lock()
defer s.mux.Unlock()
@@ -116,7 +118,7 @@ func (s *store) UpdateServiceEntryWrapper(service string, data *ServiceEntryWrap
log.Infof("ServiceEntry updated, host:%s", service)
}
func (s *store) DeleteServiceEntryWrapper(service string) {
func (s *store) DeleteServiceWrapper(service string) {
s.mux.Lock()
defer s.mux.Unlock()
@@ -199,31 +201,46 @@ func (s *store) GetAllServiceEntry() []*v1alpha3.ServiceEntry {
return seList
}
// GetAllServiceEntryWrapper get all ServiceEntryWrapper in the store for xds push
func (s *store) GetAllServiceEntryWrapper() []*ServiceEntryWrapper {
// GetAllServiceWrapper get all ServiceWrapper in the store for xds push
func (s *store) GetAllServiceWrapper() []*ServiceWrapper {
s.mux.RLock()
defer s.mux.RUnlock()
defer s.cleanUpdateAndDeleteArray()
sewList := make([]*ServiceEntryWrapper, 0)
sewList := make([]*ServiceWrapper, 0)
for _, serviceEntryWrapper := range s.sew {
sewList = append(sewList, serviceEntryWrapper.DeepCopy())
}
return sewList
}
// GetIncrementalServiceEntryWrapper get incremental ServiceEntryWrapper in the store for xds push
func (s *store) GetIncrementalServiceEntryWrapper() ([]*ServiceEntryWrapper, []*ServiceEntryWrapper) {
// GetAllDestinationRuleWrapper get all DestinationRuleWrapper in the store for xds push
func (s *store) GetAllDestinationRuleWrapper() []*ingress.WrapperDestinationRule {
s.mux.RLock()
defer s.mux.RUnlock()
defer s.cleanUpdateAndDeleteArray()
updatedList := make([]*ServiceEntryWrapper, 0)
drwList := make([]*ingress.WrapperDestinationRule, 0)
for _, serviceEntryWrapper := range s.sew {
if serviceEntryWrapper.DestinationRuleWrapper != nil {
drwList = append(drwList, serviceEntryWrapper.DeepCopy().DestinationRuleWrapper)
}
}
return drwList
}
// GetIncrementalServiceWrapper get incremental ServiceWrapper in the store for xds push
func (s *store) GetIncrementalServiceWrapper() ([]*ServiceWrapper, []*ServiceWrapper) {
s.mux.RLock()
defer s.mux.RUnlock()
defer s.cleanUpdateAndDeleteArray()
updatedList := make([]*ServiceWrapper, 0)
for _, serviceEntryWrapper := range s.toBeUpdated {
updatedList = append(updatedList, serviceEntryWrapper.DeepCopy())
}
deletedList := make([]*ServiceEntryWrapper, 0)
deletedList := make([]*ServiceWrapper, 0)
for _, serviceEntryWrapper := range s.toBeDeleted {
deletedList = append(deletedList, serviceEntryWrapper.DeepCopy())
}
@@ -236,7 +253,7 @@ func (s *store) cleanUpdateAndDeleteArray() {
s.toBeDeleted = nil
}
func (s *store) updateIpMap(service string, data *ServiceEntryWrapper) {
func (s *store) updateIpMap(service string, data *ServiceWrapper) {
for _, ep := range data.ServiceEntry.Endpoints {
if s.ip2services[ep.Address] == nil {
s.ip2services[ep.Address] = make(map[string]bool)

34
registry/memory/model.go
View File

@@ -18,27 +18,37 @@ import (
"time"
"istio.io/api/networking/v1alpha3"
"github.com/alibaba/higress/pkg/ingress/kube/common"
)
type ServiceEntryWrapper struct {
ServiceName string
ServiceEntry *v1alpha3.ServiceEntry
Suffix string
RegistryType string
createTime time.Time
type ServiceWrapper struct {
ServiceName string
ServiceEntry *v1alpha3.ServiceEntry
DestinationRuleWrapper *common.WrapperDestinationRule
Suffix string
RegistryType string
RegistryName string
createTime time.Time
}
func (sew *ServiceEntryWrapper) DeepCopy() *ServiceEntryWrapper {
return &ServiceEntryWrapper{
ServiceEntry: sew.ServiceEntry.DeepCopy(),
createTime: sew.GetCreateTime(),
func (sew *ServiceWrapper) DeepCopy() *ServiceWrapper {
res := &ServiceWrapper{}
res = sew
res.ServiceEntry = sew.ServiceEntry.DeepCopy()
res.createTime = sew.GetCreateTime()
if sew.DestinationRuleWrapper != nil {
res.DestinationRuleWrapper = sew.DestinationRuleWrapper
res.DestinationRuleWrapper.DestinationRule = sew.DestinationRuleWrapper.DestinationRule.DeepCopy()
}
return res
}
func (sew *ServiceEntryWrapper) SetCreateTime(createTime time.Time) {
func (sew *ServiceWrapper) SetCreateTime(createTime time.Time) {
sew.createTime = createTime
}
func (sew *ServiceEntryWrapper) GetCreateTime() time.Time {
func (sew *ServiceWrapper) GetCreateTime() time.Time {
return sew.createTime
}

9
registry/nacos/v2/watcher.go
View File

@@ -66,7 +66,7 @@ type watcher struct {
isStop bool
addrProvider *address.NacosAddressProvider
updateCacheWhenEmpty bool
nacosClientConfig *constant.ClientConfig
nacosClientConfig *constant.ClientConfig
authOption provider.AuthOption
}
@@ -413,7 +413,7 @@ func (w *watcher) getSubscribeCallback(groupName string, serviceName string) fun
if err != nil {
if strings.Contains(err.Error(), "hosts is empty") {
if w.updateCacheWhenEmpty {
w.cache.DeleteServiceEntryWrapper(host)
w.cache.DeleteServiceWrapper(host)
}
} else {
log.Errorf("callback error:%v", err)
@@ -425,11 +425,12 @@ func (w *watcher) getSubscribeCallback(groupName string, serviceName string) fun
return
}
serviceEntry := w.generateServiceEntry(host, services)
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceName: serviceName,
ServiceEntry: serviceEntry,
Suffix: suffix,
RegistryType: w.Type,
RegistryName: w.Name,
})
}
}
@@ -487,7 +488,7 @@ func (w *watcher) Stop() {
suffix := strings.Join([]string{s[0], w.NacosNamespace, "nacos"}, common.DotSeparator)
suffix = strings.ReplaceAll(suffix, common.Underscore, common.Hyphen)
host := strings.Join([]string{s[1], suffix}, common.DotSeparator)
w.cache.DeleteServiceEntryWrapper(host)
w.cache.DeleteServiceWrapper(host)
}
w.isStop = true

7
registry/nacos/watcher.go
View File

@@ -301,7 +301,7 @@ func (w *watcher) getSubscribeCallback(groupName string, serviceName string) fun
if err != nil {
if strings.Contains(err.Error(), "hosts is empty") {
if w.updateCacheWhenEmpty {
w.cache.DeleteServiceEntryWrapper(host)
w.cache.DeleteServiceWrapper(host)
}
} else {
log.Errorf("callback error:%v", err)
@@ -312,11 +312,12 @@ func (w *watcher) getSubscribeCallback(groupName string, serviceName string) fun
return
}
serviceEntry := w.generateServiceEntry(host, services)
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceName: serviceName,
ServiceEntry: serviceEntry,
Suffix: suffix,
RegistryType: w.Type,
RegistryName: w.Name,
})
}
}
@@ -374,7 +375,7 @@ func (w *watcher) Stop() {
suffix := strings.Join([]string{s[0], w.NacosNamespace, w.Type}, common.DotSeparator)
suffix = strings.ReplaceAll(suffix, common.Underscore, common.Hyphen)
host := strings.Join([]string{s[1], suffix}, common.DotSeparator)
w.cache.DeleteServiceEntryWrapper(host)
w.cache.DeleteServiceWrapper(host)
}
w.isStop = true
close(w.stop)

2
registry/reconcile/reconcile.go
View File

@@ -211,6 +211,8 @@ func (r *Reconciler) generateWatcherFromRegistryConfig(registry *apiv1.RegistryC
direct.WithName(registry.Name),
direct.WithDomain(registry.Domain),
direct.WithPort(registry.Port),
direct.WithProtocol(registry.Protocol),
direct.WithSNI(registry.Sni),
)
case string(Eureka):
watcher, err = eureka.NewWatcher(

16
registry/zookeeper/watcher.go
View File

@@ -331,11 +331,12 @@ func (w *watcher) DataChange(eventType Event) bool {
se := w.generateServiceEntry(w.serviceEntry[host])
w.seMux.Unlock()
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceName: host,
ServiceEntry: se,
Suffix: "zookeeper",
RegistryType: w.Type,
RegistryName: w.Name,
})
w.UpdateService()
} else if eventType.Action == EventTypeDel {
@@ -358,14 +359,15 @@ func (w *watcher) DataChange(eventType Event) bool {
//todo update
if len(se.Endpoints) == 0 {
if !w.keepStaleWhenEmpty {
w.cache.DeleteServiceEntryWrapper(host)
w.cache.DeleteServiceWrapper(host)
}
} else {
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceName: host,
ServiceEntry: se,
Suffix: "zookeeper",
RegistryType: w.Type,
RegistryName: w.Name,
})
}
w.UpdateService()
@@ -560,20 +562,22 @@ func (w *watcher) ChildToServiceEntry(children []string, interfaceName, zkPath s
if !reflect.DeepEqual(value, config) {
w.serviceEntry[host] = config
//todo update or create serviceentry
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceName: host,
ServiceEntry: se,
Suffix: "zookeeper",
RegistryType: w.Type,
RegistryName: w.Name,
})
}
} else {
w.serviceEntry[host] = config
w.cache.UpdateServiceEntryWrapper(host, &memory.ServiceEntryWrapper{
w.cache.UpdateServiceWrapper(host, &memory.ServiceWrapper{
ServiceName: host,
ServiceEntry: se,
Suffix: "zookeeper",
RegistryType: w.Type,
RegistryName: w.Name,
})
}
}
@@ -708,7 +712,7 @@ func (w *watcher) Stop() {
w.seMux.Lock()
for key := range w.serviceEntry {
w.cache.DeleteServiceEntryWrapper(key)
w.cache.DeleteServiceWrapper(key)
}
w.UpdateService()
w.seMux.Unlock()