This is an automated email from the ASF dual-hosted git repository.
alexstocks pushed a commit to branch 1.5
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/1.5 by this push:
new 5a9a038 cluster interceptor support (#1206)
5a9a038 is described below
commit 5a9a0388450ff82125fa2de7dbfb6dc502687c4a
Author: Ian Luo <[email protected]>
AuthorDate: Thu May 20 18:31:12 2021 +0800
cluster interceptor support (#1206)
* cluster interceptor support
* fix go fmt issue
* add license header
* add license header
* move interceptors into init method
* make extension go-routine safe
---
cluster/cluster_impl/available_cluster.go | 2 +-
cluster/cluster_impl/base_cluster_invoker.go | 19 ------
cluster/cluster_impl/broadcast_cluster.go | 2 +-
cluster/cluster_impl/failback_cluster.go | 2 +-
cluster/cluster_impl/failfast_cluster.go | 2 +-
cluster/cluster_impl/failover_cluster.go | 2 +-
cluster/cluster_impl/failsafe_cluster.go | 2 +-
cluster/cluster_impl/forking_cluster.go | 2 +-
cluster/cluster_impl/interceptor_invoker.go | 76 ++++++++++++++++++++++
cluster/cluster_impl/mock_cluster.go | 2 +-
cluster/cluster_impl/zone_aware_cluster.go | 2 +-
...luster.go => zone_aware_cluster_interceptor.go} | 42 +++++++-----
cluster/cluster_impl/zone_aware_cluster_invoker.go | 32 +--------
cluster/cluster_interceptor.go | 16 ++---
common/extension/cluster_interceptor.go | 60 +++++++++++++++++
15 files changed, 180 insertions(+), 83 deletions(-)
diff --git a/cluster/cluster_impl/available_cluster.go
b/cluster/cluster_impl/available_cluster.go
index 1f41890..1b0a9da 100644
--- a/cluster/cluster_impl/available_cluster.go
+++ b/cluster/cluster_impl/available_cluster.go
@@ -40,5 +40,5 @@ func NewAvailableCluster() cluster.Cluster {
// Join returns a baseClusterInvoker instance
func (cluster *availableCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return NewAvailableClusterInvoker(directory)
+ return buildInterceptorChain(NewAvailableClusterInvoker(directory))
}
diff --git a/cluster/cluster_impl/base_cluster_invoker.go
b/cluster/cluster_impl/base_cluster_invoker.go
index 96b59a5..03800fd 100644
--- a/cluster/cluster_impl/base_cluster_invoker.go
+++ b/cluster/cluster_impl/base_cluster_invoker.go
@@ -18,10 +18,6 @@
package cluster_impl
import (
- "context"
-)
-
-import (
perrors "github.com/pkg/errors"
"go.uber.org/atomic"
)
@@ -40,7 +36,6 @@ type baseClusterInvoker struct {
availablecheck bool
destroyed *atomic.Bool
stickyInvoker protocol.Invoker
- interceptor cluster.ClusterInterceptor
}
func newBaseClusterInvoker(directory cluster.Directory) baseClusterInvoker {
@@ -166,20 +161,6 @@ func (invoker *baseClusterInvoker) doSelectInvoker(lb
cluster.LoadBalance, invoc
return nil
}
-func (invoker *baseClusterInvoker) Invoke(ctx context.Context, invocation
protocol.Invocation) protocol.Result {
- if invoker.interceptor != nil {
- invoker.interceptor.BeforeInvoker(ctx, invocation)
-
- result := invoker.interceptor.DoInvoke(ctx, invocation)
-
- invoker.interceptor.AfterInvoker(ctx, invocation)
-
- return result
- }
-
- return nil
-}
-
func isInvoked(selectedInvoker protocol.Invoker, invoked []protocol.Invoker)
bool {
for _, i := range invoked {
if i == selectedInvoker {
diff --git a/cluster/cluster_impl/broadcast_cluster.go
b/cluster/cluster_impl/broadcast_cluster.go
index ea3dee9..815f4ea 100644
--- a/cluster/cluster_impl/broadcast_cluster.go
+++ b/cluster/cluster_impl/broadcast_cluster.go
@@ -41,5 +41,5 @@ func NewBroadcastCluster() cluster.Cluster {
// Join returns a baseClusterInvoker instance
func (cluster *broadcastCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newBroadcastClusterInvoker(directory)
+ return buildInterceptorChain(newBroadcastClusterInvoker(directory))
}
diff --git a/cluster/cluster_impl/failback_cluster.go
b/cluster/cluster_impl/failback_cluster.go
index 278ac54..c7bbaf5 100644
--- a/cluster/cluster_impl/failback_cluster.go
+++ b/cluster/cluster_impl/failback_cluster.go
@@ -41,5 +41,5 @@ func NewFailbackCluster() cluster.Cluster {
// Join returns a baseClusterInvoker instance
func (cluster *failbackCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newFailbackClusterInvoker(directory)
+ return buildInterceptorChain(newFailbackClusterInvoker(directory))
}
diff --git a/cluster/cluster_impl/failfast_cluster.go
b/cluster/cluster_impl/failfast_cluster.go
index a5ea7a0..d5f63c1 100644
--- a/cluster/cluster_impl/failfast_cluster.go
+++ b/cluster/cluster_impl/failfast_cluster.go
@@ -41,5 +41,5 @@ func NewFailFastCluster() cluster.Cluster {
// Join returns a baseClusterInvoker instance
func (cluster *failfastCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newFailFastClusterInvoker(directory)
+ return buildInterceptorChain(newFailFastClusterInvoker(directory))
}
diff --git a/cluster/cluster_impl/failover_cluster.go
b/cluster/cluster_impl/failover_cluster.go
index 4c09fd1..6984aa4 100644
--- a/cluster/cluster_impl/failover_cluster.go
+++ b/cluster/cluster_impl/failover_cluster.go
@@ -41,5 +41,5 @@ func NewFailoverCluster() cluster.Cluster {
// Join returns a baseClusterInvoker instance
func (cluster *failoverCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newFailoverClusterInvoker(directory)
+ return buildInterceptorChain(newFailoverClusterInvoker(directory))
}
diff --git a/cluster/cluster_impl/failsafe_cluster.go
b/cluster/cluster_impl/failsafe_cluster.go
index d9465c0..d1e22ce 100644
--- a/cluster/cluster_impl/failsafe_cluster.go
+++ b/cluster/cluster_impl/failsafe_cluster.go
@@ -41,5 +41,5 @@ func NewFailsafeCluster() cluster.Cluster {
// Join returns a baseClusterInvoker instance
func (cluster *failsafeCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newFailsafeClusterInvoker(directory)
+ return buildInterceptorChain(newFailsafeClusterInvoker(directory))
}
diff --git a/cluster/cluster_impl/forking_cluster.go
b/cluster/cluster_impl/forking_cluster.go
index 8c99113..7181b9a 100644
--- a/cluster/cluster_impl/forking_cluster.go
+++ b/cluster/cluster_impl/forking_cluster.go
@@ -41,5 +41,5 @@ func NewForkingCluster() cluster.Cluster {
// Join returns a baseClusterInvoker instance
func (cluster *forkingCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newForkingClusterInvoker(directory)
+ return buildInterceptorChain(newForkingClusterInvoker(directory))
}
diff --git a/cluster/cluster_impl/interceptor_invoker.go
b/cluster/cluster_impl/interceptor_invoker.go
new file mode 100644
index 0000000..4318177
--- /dev/null
+++ b/cluster/cluster_impl/interceptor_invoker.go
@@ -0,0 +1,76 @@
+/*
+ * 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 cluster_impl
+
+import (
+ "context"
+)
+
+import (
+ "github.com/apache/dubbo-go/cluster"
+ "github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/extension"
+ "github.com/apache/dubbo-go/protocol"
+)
+
+// InterceptorInvoker mocks cluster interceptor as an invoker
+type InterceptorInvoker struct {
+ next protocol.Invoker
+ interceptor cluster.Interceptor
+}
+
+// GetURL is used to get url from InterceptorInvoker
+func (i *InterceptorInvoker) GetURL() *common.URL {
+ return i.next.GetURL()
+}
+
+// IsAvailable is used to get available status
+func (i *InterceptorInvoker) IsAvailable() bool {
+ return i.next.IsAvailable()
+}
+
+// Invoke is used to call service method by invocation
+func (i *InterceptorInvoker) Invoke(ctx context.Context, invocation
protocol.Invocation) protocol.Result {
+ return i.interceptor.Invoke(ctx, i.next, invocation)
+}
+
+// Destroy will destroy invoker
+func (i *InterceptorInvoker) Destroy() {
+ i.next.Destroy()
+}
+
+func buildInterceptorChain(invoker protocol.Invoker, builtins
...cluster.Interceptor) protocol.Invoker {
+ // The order of interceptors is from left to right, so loading from
right to left
+ next := invoker
+ interceptors := extension.GetClusterInterceptors()
+ if len(interceptors) != 0 {
+ for i := len(interceptors) - 1; i >= 0; i-- {
+ v := &InterceptorInvoker{next: next, interceptor:
interceptors[i]}
+ next = v
+ }
+ }
+
+ if builtins != nil && len(builtins) > 0 {
+ for i := len(builtins) - 1; i >= 0; i-- {
+ v := &InterceptorInvoker{next: next, interceptor:
builtins[i]}
+ next = v
+ }
+ }
+
+ return next
+}
diff --git a/cluster/cluster_impl/mock_cluster.go
b/cluster/cluster_impl/mock_cluster.go
index a643417..eb52461 100644
--- a/cluster/cluster_impl/mock_cluster.go
+++ b/cluster/cluster_impl/mock_cluster.go
@@ -35,5 +35,5 @@ func NewMockCluster() cluster.Cluster {
// nolint
func (cluster *mockCluster) Join(directory cluster.Directory) protocol.Invoker
{
- return protocol.NewBaseInvoker(directory.GetURL())
+ return
buildInterceptorChain(protocol.NewBaseInvoker(directory.GetURL()))
}
diff --git a/cluster/cluster_impl/zone_aware_cluster.go
b/cluster/cluster_impl/zone_aware_cluster.go
index 7439db2..3a47a95 100644
--- a/cluster/cluster_impl/zone_aware_cluster.go
+++ b/cluster/cluster_impl/zone_aware_cluster.go
@@ -40,5 +40,5 @@ func NewZoneAwareCluster() cluster.Cluster {
// Join returns a zoneAwareClusterInvoker instance
func (cluster *zoneAwareCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newZoneAwareClusterInvoker(directory)
+ return buildInterceptorChain(newZoneAwareClusterInvoker(directory),
getZoneAwareInterceptor())
}
diff --git a/cluster/cluster_impl/failover_cluster.go
b/cluster/cluster_impl/zone_aware_cluster_interceptor.go
similarity index 59%
copy from cluster/cluster_impl/failover_cluster.go
copy to cluster/cluster_impl/zone_aware_cluster_interceptor.go
index 4c09fd1..d23d254 100644
--- a/cluster/cluster_impl/failover_cluster.go
+++ b/cluster/cluster_impl/zone_aware_cluster_interceptor.go
@@ -18,28 +18,40 @@
package cluster_impl
import (
+ "context"
+)
+
+import (
"github.com/apache/dubbo-go/cluster"
"github.com/apache/dubbo-go/common/constant"
- "github.com/apache/dubbo-go/common/extension"
"github.com/apache/dubbo-go/protocol"
)
-type failoverCluster struct{}
-
-func init() {
- extension.SetCluster(constant.FAILOVER_CLUSTER_NAME, NewFailoverCluster)
+type zoneAwareInterceptor struct {
}
-// NewFailoverCluster returns a failover cluster instance
-//
-// Failure automatically switch, when there is a failure,
-// retry the other server (default). Usually used for read operations,
-// but retries can result in longer delays.
-func NewFailoverCluster() cluster.Cluster {
- return &failoverCluster{}
+func (z *zoneAwareInterceptor) Invoke(ctx context.Context, invoker
protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+ key := constant.REGISTRY_KEY + "." + constant.ZONE_FORCE_KEY
+ force := ctx.Value(key)
+
+ if force != nil {
+ switch value := force.(type) {
+ case bool:
+ if value {
+ invocation.SetAttachments(key, "true")
+ }
+ case string:
+ if "true" == value {
+ invocation.SetAttachments(key, "true")
+ }
+ default:
+ // ignore
+ }
+ }
+
+ return invoker.Invoke(ctx, invocation)
}
-// Join returns a baseClusterInvoker instance
-func (cluster *failoverCluster) Join(directory cluster.Directory)
protocol.Invoker {
- return newFailoverClusterInvoker(directory)
+func getZoneAwareInterceptor() cluster.Interceptor {
+ return &zoneAwareInterceptor{}
}
diff --git a/cluster/cluster_impl/zone_aware_cluster_invoker.go
b/cluster/cluster_impl/zone_aware_cluster_invoker.go
index 4891b9d..8673d1e 100644
--- a/cluster/cluster_impl/zone_aware_cluster_invoker.go
+++ b/cluster/cluster_impl/zone_aware_cluster_invoker.go
@@ -40,16 +40,14 @@ type zoneAwareClusterInvoker struct {
}
func newZoneAwareClusterInvoker(directory cluster.Directory) protocol.Invoker {
- invoke := &zoneAwareClusterInvoker{
+ invoker := &zoneAwareClusterInvoker{
baseClusterInvoker: newBaseClusterInvoker(directory),
}
- // add local to interceptor
- invoke.interceptor = invoke
- return invoke
+ return invoker
}
// nolint
-func (invoker *zoneAwareClusterInvoker) DoInvoke(ctx context.Context,
invocation protocol.Invocation) protocol.Result {
+func (invoker *zoneAwareClusterInvoker) Invoke(ctx context.Context, invocation
protocol.Invocation) protocol.Result {
invokers := invoker.directory.List(invocation)
err := invoker.checkInvokers(invokers, invocation)
@@ -104,30 +102,6 @@ func (invoker *zoneAwareClusterInvoker) DoInvoke(ctx
context.Context, invocation
}
}
-func (invoker *zoneAwareClusterInvoker) BeforeInvoker(ctx context.Context,
invocation protocol.Invocation) {
- key := constant.REGISTRY_KEY + "." + constant.ZONE_FORCE_KEY
- force := ctx.Value(key)
-
- if force != nil {
- switch value := force.(type) {
- case bool:
- if value {
- invocation.SetAttachments(key, "true")
- }
- case string:
- if "true" == value {
- invocation.SetAttachments(key, "true")
- }
- default:
- // ignore
- }
- }
-}
-
-func (invoker *zoneAwareClusterInvoker) AfterInvoker(ctx context.Context,
invocation protocol.Invocation) {
-
-}
-
func matchParam(target, key, def string, invoker protocol.Invoker) bool {
return target == invoker.GetURL().GetParam(key, def)
}
diff --git a/cluster/cluster_interceptor.go b/cluster/cluster_interceptor.go
index a627e81..dd28198 100644
--- a/cluster/cluster_interceptor.go
+++ b/cluster/cluster_interceptor.go
@@ -25,15 +25,9 @@ import (
"github.com/apache/dubbo-go/protocol"
)
-// ClusterInterceptor
-// Extension - ClusterInterceptor
-type ClusterInterceptor interface {
- // Before DoInvoke method
- BeforeInvoker(ctx context.Context, invocation protocol.Invocation)
-
- // After DoInvoke method
- AfterInvoker(ctx context.Context, invocation protocol.Invocation)
-
- // Corresponding cluster invoke
- DoInvoke(ctx context.Context, invocation protocol.Invocation)
protocol.Result
+// Interceptor
+// Extension - Interceptor
+type Interceptor interface {
+ // Invoke is the core function of a cluster interceptor, it determines
the process of the interceptor
+ Invoke(context.Context, protocol.Invoker, protocol.Invocation)
protocol.Result
}
diff --git a/common/extension/cluster_interceptor.go
b/common/extension/cluster_interceptor.go
new file mode 100644
index 0000000..47bc2b1
--- /dev/null
+++ b/common/extension/cluster_interceptor.go
@@ -0,0 +1,60 @@
+/*
+ * 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 extension
+
+import (
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go/cluster"
+)
+
+var (
+ lock sync.RWMutex
+ interceptors = make(map[string]func() cluster.Interceptor)
+)
+
+// SetClusterInterceptor sets cluster interceptor so that user has chance to
inject extra logics before and after
+// cluster invoker
+func SetClusterInterceptor(name string, fun func() cluster.Interceptor) {
+ lock.Lock()
+ defer lock.Unlock()
+ interceptors[name] = fun
+}
+
+// GetClusterInterceptor returns the cluster interceptor instance with the
given name
+func GetClusterInterceptor(name string) cluster.Interceptor {
+ lock.RLock()
+ defer lock.RUnlock()
+ if interceptors[name] == nil {
+ panic("cluster_interceptor for " + name + " doesn't exist, make
sure the corresponding package is imported")
+ }
+ return interceptors[name]()
+}
+
+// GetClusterInterceptors returns all instances of registered cluster
interceptors
+func GetClusterInterceptors() []cluster.Interceptor {
+ lock.RLock()
+ defer lock.RUnlock()
+ ret := make([]cluster.Interceptor, 0, len(interceptors))
+ for _, f := range interceptors {
+ ret = append(ret, f())
+ }
+ return ret
+}
