Istio/XDS support

CHANGELOG.md

@@ -345,7 +345,7 @@ Milestone: [https://github.com/apache/dubbo-go/milestone/5](https://github.com/a
 - [Fix go client quit abnormally when it connects java server](https://github.com/apache/dubbo-go/pull/820) [@wenxuwan](https://github.com/wenxuwan)
 - [Fix sentinel windows issue](https://github.com/apache/dubbo-go/pull/821) [@louyuting](https://github.com/louyuting)
 - [Fix metadata default port](https://github.com/apache/dubbo-go/pull/821) [@sanxun0325](https://github.com/sanxun0325)
-- [Fix consul can not destory](https://github.com/apache/dubbo-go/pull/788) [@LaurenceLiZhixin](https://github.com/LaurenceLiZhixin)
+- [Fix consul can not destroy](https://github.com/apache/dubbo-go/pull/788) [@LaurenceLiZhixin](https://github.com/LaurenceLiZhixin)
 
 Milestone: [https://github.com/apache/dubbo-go/milestone/6](https://github.com/apache/dubbo-go/milestone/6?closed=1)
 
@@ -390,7 +390,7 @@ Project: [https://github.com/apache/dubbo-go/projects/10](https://github.com/apa
 - [Fix go client quit abnormally when it connects java server](https://github.com/apache/dubbo-go/pull/820) [@wenxuwan](https://github.com/wenxuwan)      
 - [Fix sentinel windows issue](https://github.com/apache/dubbo-go/pull/821) [@louyuting](https://github.com/louyuting)      
 - [Fix metadata default port](https://github.com/apache/dubbo-go/pull/821) [@sanxun0325](https://github.com/sanxun0325)      
-- [Fix consul can not destory](https://github.com/apache/dubbo-go/pull/788) [@LaurenceLiZhixin](https://github.com/LaurenceLiZhixin)      
+- [Fix consul can not destroy](https://github.com/apache/dubbo-go/pull/788) [@LaurenceLiZhixin](https://github.com/LaurenceLiZhixin)      
 
 Milestone: [https://github.com/apache/dubbo-go/milestone/6](https://github.com/apache/dubbo-go/milestone/6?closed=1)
 

cluster/router/meshrouter/factory.go

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 meshrouter
+
+import (
+	"dubbo.apache.org/dubbo-go/v3/cluster/router"
+	"dubbo.apache.org/dubbo-go/v3/common/constant"
+	"dubbo.apache.org/dubbo-go/v3/common/extension"
+)
+
+func init() {
+	extension.SetRouterFactory(constant.MeshRouterFactoryKey, NewMeshRouterFactory)
+}
+
+// UniformRouteFactory is uniform router's factory
+type MeshRouterFactory struct{}
+
+// NewMeshRouterFactory constructs a new PriorityRouterFactory
+func NewMeshRouterFactory() router.PriorityRouterFactory {
+	return &MeshRouterFactory{}
+}
+
+// NewPriorityRouter construct a new UniformRouteFactory as PriorityRouter
+func (f *MeshRouterFactory) NewPriorityRouter() (router.PriorityRouter, error) {
+	return NewMeshRouter()
+}

cluster/router/meshrouter/meshrouter.go

@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 meshrouter
+
+import (
+	"bytes"
+	"math/rand"
+	"strings"
+)
+
+import (
+	"dubbo.apache.org/dubbo-go/v3/cluster/router"
+	"dubbo.apache.org/dubbo-go/v3/common"
+	"dubbo.apache.org/dubbo-go/v3/common/constant"
+	"dubbo.apache.org/dubbo-go/v3/common/logger"
+	"dubbo.apache.org/dubbo-go/v3/config_center"
+	"dubbo.apache.org/dubbo-go/v3/protocol"
+	"dubbo.apache.org/dubbo-go/v3/remoting/xds"
+	"dubbo.apache.org/dubbo-go/v3/xds/client/resource"
+	"dubbo.apache.org/dubbo-go/v3/xds/utils/resolver"
+)
+
+const (
+	name = "mesh-router"
+)
+
+// MeshRouter have
+type MeshRouter struct {
+	client *xds.WrappedClientImpl
+}
+
+// NewMeshRouter construct an NewConnCheckRouter via url
+func NewMeshRouter() (router.PriorityRouter, error) {
+	xdsWrappedClient := xds.GetXDSWrappedClient()
+	if xdsWrappedClient == nil {
+		logger.Debugf("[Mesh Router] xds wrapped client is not created.")
+	}
+	return &MeshRouter{
+		client: xdsWrappedClient,
+	}, nil
+}
+
+// Route gets a list of routed invoker
+func (r *MeshRouter) Route(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+	if r.client == nil {
+		return invokers
+	}
+	hostAddr, err := r.client.GetHostAddrByServiceUniqueKey(getSubscribeName(url))
+	if err != nil {
+		// todo deal with error
+		return nil
+	}
+	rconf := r.client.GetRouterConfig(hostAddr)
+
+	clusterInvokerMap := make(map[string][]protocol.Invoker)
+	for _, v := range invokers {
+		meshClusterID := v.GetURL().GetParam(constant.MeshClusterIDKey, "")
+		if _, ok := clusterInvokerMap[meshClusterID]; !ok {
+			clusterInvokerMap[meshClusterID] = make([]protocol.Invoker, 0)
+		}
+		clusterInvokerMap[meshClusterID] = append(clusterInvokerMap[meshClusterID], v)
+	}
+
+	if len(rconf.VirtualHosts) != 0 {
+		// try to route to sub virtual host
+		for _, vh := range rconf.VirtualHosts {
+			// 1. match domain
+			//vh.Domains == ["*"]
+
+			// 2. match http route
+			for _, r := range vh.Routes {
+				//route.
+				ctx := invocation.GetAttachmentAsContext()
+				matcher, err := resource.RouteToMatcher(r)
+				if err != nil {
+					logger.Errorf("[Mesh Router] router to matcher failed with error %s", err)
+					return invokers
+				}
+				if matcher.Match(resolver.RPCInfo{
+					Context: ctx,
+					Method:  "/" + invocation.MethodName(),
+				}) {
+					// Loop through routes in order and select first match.
+					if r == nil || r.WeightedClusters == nil {
+						logger.Errorf("[Mesh Router] route's WeightedClusters is empty, route: %+v", r)
+						return invokers
+					}
+					invokersWeightPairs := make(invokerWeightPairs, 0)
+
+					for clusterID, weight := range r.WeightedClusters {
+						// cluster -> invokers
+						targetInvokers := clusterInvokerMap[clusterID]
+						invokersWeightPairs = append(invokersWeightPairs, invokerWeightPair{
+							invokers: targetInvokers,
+							weight:   weight.Weight,
+						})
+					}
+					return invokersWeightPairs.GetInvokers()
+				}
+			}
+		}
+	}
+	return invokers
+}
+
+// Process there is no process needs for uniform Router, as it upper struct RouterChain has done it
+func (r *MeshRouter) Process(event *config_center.ConfigChangeEvent) {
+}
+
+// Name get name of ConnCheckerRouter
+func (r *MeshRouter) Name() string {
+	return name
+}
+
+// Priority get Router priority level
+func (r *MeshRouter) Priority() int64 {
+	return 0
+}
+
+// URL Return URL in router
+func (r *MeshRouter) URL() *common.URL {
+	return nil
+}
+
+// Notify the router the invoker list
+func (r *MeshRouter) Notify(invokers []protocol.Invoker) {
+}
+
+type invokerWeightPair struct {
+	invokers []protocol.Invoker
+	weight   uint32
+}
+
+type invokerWeightPairs []invokerWeightPair
+
+func (i *invokerWeightPairs) GetInvokers() []protocol.Invoker {
+	if len(*i) == 0 {
+		return nil
+	}
+	totalWeight := uint32(0)
+	tempWeight := uint32(0)
+	for _, v := range *i {
+		totalWeight += v.weight
+	}
+	randFloat := rand.Float64()
+	for _, v := range *i {
+		tempWeight += v.weight
+		tempPercent := float64(tempWeight) / float64(totalWeight)
+		if tempPercent >= randFloat {
+			return v.invokers
+		}
+	}
+	return (*i)[0].invokers
+}
+
+func getSubscribeName(url *common.URL) string {
+	var buffer bytes.Buffer
+
+	buffer.Write([]byte(common.DubboNodes[common.PROVIDER]))
+	appendParam(&buffer, url, constant.InterfaceKey)
+	appendParam(&buffer, url, constant.VersionKey)
+	appendParam(&buffer, url, constant.GroupKey)
+	return buffer.String()
+}
+
+func appendParam(target *bytes.Buffer, url *common.URL, key string) {
+	value := url.GetParam(key, "")
+	target.Write([]byte(constant.NacosServiceNameSeparator))
+	if strings.TrimSpace(value) != "" {
+		target.Write([]byte(value))
+	}
+}

common/constant/env.go

@@ -19,6 +19,8 @@ package constant
 
 // nolint
 const (
-	ConfigFileEnvKey = "DUBBO_GO_CONFIG_PATH" // key of environment variable dubbogo configure file path
-	AppLogConfFile   = "AppLogConfFile"
+	ConfigFileEnvKey   = "DUBBO_GO_CONFIG_PATH" // key of environment variable dubbogo configure file path
+	AppLogConfFile     = "AppLogConfFile"
+	PodNameEnvKey      = "POD_NAME"
+	PodNamespaceEnvKey = "POD_NAMESPACE"
 )

common/constant/key.go

@@ -92,6 +92,7 @@ const (
 	TokenFilterKey                       = "token"
 	TpsLimitFilterKey                    = "tps"
 	TracingFilterKey                     = "tracing"
+	XdsCircuitBreakerKey                 = "xds_circuit_reaker"
 )
 
 const (
@@ -274,6 +275,10 @@ const (
 	ZookeeperKey = "zookeeper"
 )
 
+const (
+	XDSRegistryKey = "xds"
+)
+
 const (
 	EtcdV3Key = "etcdv3"
 )
@@ -290,14 +295,15 @@ const (
 
 // Use for router module
 const (
-	TagRouterRuleSuffix       = ".tag-router"             // Specify tag router suffix
+	TagRouterRuleSuffix       = ".tag-router"
 	ConditionRouterRuleSuffix = ".condition-router"       // Specify condition router suffix
 	MeshRouteSuffix           = ".MESHAPPRULE"            // Specify mesh router suffix
 	ForceUseTag               = "dubbo.force.tag"         // the tag in attachment
 	Tagkey                    = "dubbo.tag"               // key of tag
 	AttachmentKey             = DubboCtxKey("attachment") // key in context in invoker
 	TagRouterFactoryKey       = "tag"
-	V3RouterFactoryKey        = "mesh"
+	V3RouterFactoryKey        = "v3router"
+	MeshRouterFactoryKey      = "mesh"
 )
 
 // Auth filter

common/constant/xds.go

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 constant
+
+const (
+	MeshClusterIDKey = "meshClusterID"
+	MeshHostAddrKey  = "meshHostAddr"
+	MeshSubsetKey    = "meshSubset"
+
+	MeshDeleteClusterPrefix = "-"
+	MeshAnyAddrMatcher      = "*"
+)
+
+const (
+	XDSMetadataClusterIDKey = "CLUSTER_ID"
+	XDSMetadataLabelsKey    = "LABELS"
+
+	XDSMetadataDefaultDomainName = "Kubernetes"
+
+	XDSMetadataDubboGoMapperKey = "DUBBO_GO"
+)

common/logger/logger.go

@@ -41,6 +41,7 @@ type DubboLogger struct {
 type Config struct {
 	LumberjackConfig *lumberjack.Logger `yaml:"lumberjack-config"`
 	ZapConfig        *zap.Config        `yaml:"zap-config"`
+	CallerSkip       int
 }
 
 // Logger is the interface for Logger types
@@ -89,8 +90,16 @@ func InitLogger(conf *Config) {
 		config.ZapConfig = conf.ZapConfig
 	}
 
+	if conf != nil {
+		config.CallerSkip = conf.CallerSkip
+	}
+
+	if config.CallerSkip == 0 {
+		config.CallerSkip = 1
+	}
+
 	if conf == nil || conf.LumberjackConfig == nil {
-		zapLogger, _ = config.ZapConfig.Build(zap.AddCaller(), zap.AddCallerSkip(1))
+		zapLogger, _ = config.ZapConfig.Build(zap.AddCaller(), zap.AddCallerSkip(config.CallerSkip))
 	} else {
 		config.LumberjackConfig = conf.LumberjackConfig
 		zapLogger = initZapLoggerWithSyncer(config)
@@ -144,7 +153,7 @@ func initZapLoggerWithSyncer(conf *Config) *zap.Logger {
 		zap.NewAtomicLevelAt(conf.ZapConfig.Level.Level()),
 	)
 
-	return zap.New(core, zap.AddCaller(), zap.AddCallerSkip(1))
+	return zap.New(core, zap.AddCaller(), zap.AddCallerSkip(conf.CallerSkip))
 }
 
 // getEncoder get encoder by config, zapcore support json and console encoder