You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2021/11/22 13:44:43 UTC
[dubbo-go] branch 1.5 updated: remove black_list module (#1605)
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 60c07e0 remove black_list module (#1605)
60c07e0 is described below
commit 60c07e026b6d9a702848563b177ce552ff9abe51
Author: XX <45...@users.noreply.github.com>
AuthorDate: Mon Nov 22 21:44:38 2021 +0800
remove black_list module (#1605)
* remove black_list module
* remove black_list module
* remove black_list module
Co-authored-by: dongzhenxing <do...@xiaomi.com>
---
cluster/cluster_impl/base_cluster_invoker.go | 4 -
cluster/router/conn_checker.go | 28 ----
cluster/router/conncheck/conn_check_route.go | 138 ----------------
cluster/router/conncheck/conn_check_route_test.go | 178 ---------------------
cluster/router/conncheck/conn_health_check.go | 45 ------
cluster/router/conncheck/conn_health_check_test.go | 52 ------
cluster/router/conncheck/factory.go | 44 -----
cluster/router/conncheck/factory_test.go | 74 ---------
common/extension/conn_checker.go | 41 -----
protocol/rpc_status.go | 122 --------------
registry/directory/directory.go | 1 -
11 files changed, 727 deletions(-)
diff --git a/cluster/cluster_impl/base_cluster_invoker.go b/cluster/cluster_impl/base_cluster_invoker.go
index 9491716..27ba9b0 100644
--- a/cluster/cluster_impl/base_cluster_invoker.go
+++ b/cluster/cluster_impl/base_cluster_invoker.go
@@ -129,12 +129,10 @@ func (invoker *baseClusterInvoker) doSelectInvoker(lb cluster.LoadBalance, invoc
if len(invokers) == 0 {
return nil
}
- invoker.getServiceHealthState().TryRefreshBlackList()
if len(invokers) == 1 {
if invokers[0].IsAvailable() {
return invokers[0]
}
- invoker.getServiceHealthState().SetInvokerUnhealthyStatus(invokers[0])
logger.Errorf("the invokers of %s is nil. ", invokers[0].GetURL().ServiceKey())
return nil
}
@@ -143,7 +141,6 @@ func (invoker *baseClusterInvoker) doSelectInvoker(lb cluster.LoadBalance, invoc
//judge if the selected Invoker is invoked and available
if (!selectedInvoker.IsAvailable() && invoker.availablecheck) || isInvoked(selectedInvoker, invoked) {
- invoker.getServiceHealthState().SetInvokerUnhealthyStatus(selectedInvoker)
otherInvokers := getOtherInvokers(invokers, selectedInvoker)
// do reselect
for i := 0; i < 5; i++ {
@@ -159,7 +156,6 @@ func (invoker *baseClusterInvoker) doSelectInvoker(lb cluster.LoadBalance, invoc
if !reselectedInvoker.IsAvailable() {
logger.Infof("the invoker of %s is not available, maybe some network error happened or the server is shutdown.",
invoker.GetURL().Ip)
- invoker.getServiceHealthState().SetInvokerUnhealthyStatus(reselectedInvoker)
otherInvokers = getOtherInvokers(otherInvokers, reselectedInvoker)
continue
}
diff --git a/cluster/router/conn_checker.go b/cluster/router/conn_checker.go
deleted file mode 100644
index fda0ef8..0000000
--- a/cluster/router/conn_checker.go
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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 router
-
-import (
- "github.com/apache/dubbo-go/protocol"
-)
-
-// ConnChecker is used to determine whether the invoker is healthy or not
-type ConnChecker interface {
- // IsConnHealthy evaluates the healthy state on the given Invoker
- IsConnHealthy(invoker protocol.Invoker) bool
-}
diff --git a/cluster/router/conncheck/conn_check_route.go b/cluster/router/conncheck/conn_check_route.go
deleted file mode 100644
index 819d99d..0000000
--- a/cluster/router/conncheck/conn_check_route.go
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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 conncheck
-
-import (
- "github.com/RoaringBitmap/roaring"
- "strconv"
- "strings"
-)
-
-import (
- "github.com/apache/dubbo-go/cluster/router"
- "github.com/apache/dubbo-go/cluster/router/utils"
- "github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/common/constant"
- "github.com/apache/dubbo-go/common/extension"
- "github.com/apache/dubbo-go/common/logger"
- "github.com/apache/dubbo-go/protocol"
-)
-
-const (
- connUnHealthy = "conn-unhealthy"
- connHealthy = "conn-healthy"
- name = "conn-check-router"
-)
-
-// ConnCheckRouter provides a health-first routing mechanism through ConnChecker
-type ConnCheckRouter struct {
- url *common.URL
- checker router.ConnChecker
- notify chan struct{}
-}
-
-// NewConnCheckRouter construct an NewConnCheckRouter via url
-func NewConnCheckRouter(url *common.URL, notify chan struct{}) (router.PriorityRouter, error) {
- r := &ConnCheckRouter{
- url: url,
- notify: notify,
- }
- checkerName := url.GetParam(constant.HEALTH_CHECKER, constant.DEFAULT_HEALTH_CHECKER)
- r.checker = extension.GetConnChecker(checkerName, url)
- return r, nil
-}
-
-// Route gets a list of healthy invoker
-func (r *ConnCheckRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
- addrPool := cache.FindAddrPool(r)
- // Add healthy invoker to the list
- unhealthyBits := addrPool[connUnHealthy]
- healthyBits := addrPool[connHealthy]
- // check if empty, return
- if unhealthyBits == nil || unhealthyBits.IsEmpty() || healthyBits == nil || healthyBits.IsEmpty() {
- return invokers
- }
- healthyInvokers := utils.JoinIfNotEqual(healthyBits, invokers)
- // If all invokers are considered unhealthy, downgrade to all invoker
- if healthyInvokers.IsEmpty() {
- logger.Warnf(" Now all invokers are unhealthy, so downgraded to all! Service: [%s]", url.ServiceKey())
- return invokers
- }
- return healthyInvokers
-}
-
-func (r *ConnCheckRouter) RouteSnapshot(cache router.Cache) string {
- addrPool := cache.FindAddrPool(r)
- // Add healthy invoker to the list
- healthBit := addrPool[connHealthy]
- sb := strings.Builder{}
- sb.WriteString(r.Name())
- sb.WriteString(" -> ")
- if healthBit == nil {
- sb.WriteString(" No Unhealthy Invoker")
- return sb.String()
- }
- sb.WriteString("Count:")
- sb.WriteString(strconv.FormatUint(healthBit.GetCardinality(), 10))
- sb.WriteString(" ")
- sb.WriteString(healthBit.String())
- return sb.String()
-}
-
-// Pool separates healthy invokers from others.
-func (r *ConnCheckRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
- rb := make(router.AddrPool, 8)
- healthyInvokers := utils.ToBitmap(invokers)
- unhealthyInvokers := roaring.NewBitmap()
- for i, invoker := range invokers {
- if !r.checker.IsConnHealthy(invoker) {
- unhealthyInvokers.Add(uint32(i))
- healthyInvokers.Remove(uint32(i))
- }
- }
- if !unhealthyInvokers.IsEmpty() {
- rb[connHealthy] = healthyInvokers
- rb[connUnHealthy] = unhealthyInvokers
- }
- return rb, nil
-}
-
-// ShouldPool will always return true to make sure healthy check constantly.
-func (r *ConnCheckRouter) ShouldPool() bool {
- return true
-}
-
-// Name get name of ConnCheckerRouter
-func (r *ConnCheckRouter) Name() string {
- return name
-}
-
-// Priority get Router priority level
-func (r *ConnCheckRouter) Priority() int64 {
- return 0
-}
-
-// URL Return URL in router
-func (r *ConnCheckRouter) URL() *common.URL {
- return r.url
-}
-
-// ConnChecker returns the HealthChecker bound to this HealthCheckRouter
-func (r *ConnCheckRouter) ConnChecker() router.ConnChecker {
- return r.checker
-}
diff --git a/cluster/router/conncheck/conn_check_route_test.go b/cluster/router/conncheck/conn_check_route_test.go
deleted file mode 100644
index bcdee07..0000000
--- a/cluster/router/conncheck/conn_check_route_test.go
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * 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 conncheck
-
-import (
- "fmt"
- "testing"
- "time"
-)
-
-import (
- "github.com/golang/mock/gomock"
- "github.com/stretchr/testify/assert"
-)
-
-import (
- "github.com/apache/dubbo-go/cluster/router"
- "github.com/apache/dubbo-go/cluster/router/chain"
- "github.com/apache/dubbo-go/cluster/router/utils"
- "github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/protocol"
- "github.com/apache/dubbo-go/protocol/invocation"
- "github.com/apache/dubbo-go/protocol/mock"
-)
-
-const (
- connCheckRoute1010IP = "192.168.10.10"
- connCheckRoute1011IP = "192.168.10.11"
- connCheckRoute1012IP = "192.168.10.12"
- connCheckRouteMethodNameTest = "test"
- connCheck1001URL = "dubbo://192.168.10.1/com.ikurento.user.UserProvider"
- connCheckRouteUrlFormat = "dubbo://%s:20000/com.ikurento.user.UserProvider"
-)
-
-func TestConnCheckRouterRoute(t *testing.T) {
- defer protocol.CleanAllStatus()
- notify := make(chan struct{})
- go func() {
- for range notify {
- }
- }()
- consumerURL, _ := common.NewURL(connCheck1001URL)
- url1, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrlFormat, connCheckRoute1010IP))
- url2, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrlFormat, connCheckRoute1011IP))
- url3, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrlFormat, connCheckRoute1012IP))
- hcr, _ := NewConnCheckRouter(consumerURL, notify)
-
- var invokers []protocol.Invoker
- invoker1 := NewMockInvoker(url1)
- invoker2 := NewMockInvoker(url2)
- invoker3 := NewMockInvoker(url3)
- state1 := protocol.NewServiceState(consumerURL.ServiceKey())
-
- state1.SetInvokerUnhealthyStatus(invoker1)
- state1.SetInvokerUnhealthyStatus(invoker2)
-
- invokers = append(invokers, invoker1, invoker2, invoker3)
- inv := invocation.NewRPCInvocation(connCheckRouteMethodNameTest, nil, nil)
- res := hcr.Route(utils.ToBitmap(invokers), setUpAddrCache(hcr.(*ConnCheckRouter), invokers), consumerURL, inv)
-
- // now invoker3 is healthy
- assert.True(t, len(res.ToArray()) == 1)
-
- // check blacklist remove
- state1.RemoveInvokerUnhealthyStatus(invoker1)
- res = hcr.Route(utils.ToBitmap(invokers), setUpAddrCache(hcr.(*ConnCheckRouter), invokers), consumerURL, inv)
- // now invoker3 invoker1 is healthy
- assert.True(t, len(res.ToArray()) == 2)
-
-}
-
-func TestRecovery(t *testing.T) {
- // check recover
-
- ctrl := gomock.NewController(t)
- defer ctrl.Finish()
-
- invoker1 := mock.NewMockInvoker(ctrl)
- invoker2 := mock.NewMockInvoker(ctrl)
-
- invoker1.EXPECT().GetURL().Return(&common.URL{Path: "path1"}).AnyTimes()
- invoker2.EXPECT().GetURL().Return(&common.URL{Path: "path2"}).AnyTimes()
- invoker1.EXPECT().IsAvailable().Return(true).AnyTimes()
- invoker2.EXPECT().IsAvailable().Return(true).AnyTimes()
-
- state1 := protocol.NewServiceState(invoker1.GetURL().ServiceKey())
- state2 := protocol.NewServiceState(invoker2.GetURL().ServiceKey())
- assert.Equal(t, len(state1.GetBlackListInvokers(16)), 0)
- assert.Equal(t, len(state2.GetBlackListInvokers(16)), 0)
- state1.SetInvokerUnhealthyStatus(invoker1)
- state2.SetInvokerUnhealthyStatus(invoker2)
- assert.Equal(t, len(state1.GetBlackListInvokers(16)), 1)
- assert.Equal(t, len(state2.GetBlackListInvokers(16)), 1)
- state1.TryRefreshBlackList()
- time.Sleep(300 * time.Millisecond)
- assert.Equal(t, len(state1.GetBlackListInvokers(16)), 0)
- assert.Equal(t, len(state2.GetBlackListInvokers(16)), 1)
- state2.TryRefreshBlackList()
- time.Sleep(300 * time.Millisecond)
- assert.Equal(t, len(state1.GetBlackListInvokers(16)), 0)
- assert.Equal(t, len(state2.GetBlackListInvokers(16)), 0)
-
-}
-
-func TestPrintlnConnCheckRouterRoute(t *testing.T) {
-
- connCheck1001URL2 := "dubbo://192.168.10.1/com.ikurento.user.UserNoProvider"
- connCheckRouteUrl2Format := "dubbo://%s:20000/com.ikurento.user.UserNoProvider"
-
- defer protocol.CleanAllStatus()
- notify := make(chan struct{})
- go func() {
- for range notify {
- }
- }()
- consumerURL, _ := common.NewURL(connCheck1001URL2)
- url1, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrl2Format, connCheckRoute1010IP))
- url2, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrl2Format, connCheckRoute1011IP))
- url3, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrl2Format, connCheckRoute1012IP))
- hcr, _ := NewConnCheckRouter(consumerURL, notify)
-
- var invokers []protocol.Invoker
- invoker1 := NewMockInvoker(url1)
- invoker2 := NewMockInvoker(url2)
- invoker3 := NewMockInvoker(url3)
-
- srvState := protocol.NewServiceState(consumerURL.ServiceKey())
- srvState.SetInvokerUnhealthyStatus(invoker1)
- srvState.SetInvokerUnhealthyStatus(invoker2)
- srvState.SetInvokerUnhealthyStatus(invoker3)
-
- invokers = append(invokers, invoker1, invoker2, invoker3)
- inv := invocation.NewRPCInvocation(connCheckRouteMethodNameTest, nil, nil)
- cache := setUpAddrCache(hcr.(*ConnCheckRouter), invokers)
- res := hcr.Route(utils.ToBitmap(invokers), cache, consumerURL, inv)
-
- // now invoker3 is healthy
- assert.True(t, len(res.ToArray()) == 3)
- var (
- router *ConnCheckRouter
- ok bool
- )
- router, ok = hcr.(*ConnCheckRouter)
- assert.True(t, ok)
- assert.Equal(t, router.RouteSnapshot(cache), "conn-check-router -> Count:0 {}")
-
- // check blacklist remove
- srvState.RemoveInvokerUnhealthyStatus(invoker1)
- srvState.RemoveInvokerUnhealthyStatus(invoker3)
- cache = setUpAddrCache(hcr.(*ConnCheckRouter), invokers)
- res = hcr.Route(utils.ToBitmap(invokers), cache, consumerURL, inv)
- // now invoker3 invoker1 is healthy
- assert.True(t, len(res.ToArray()) == 2)
- assert.Equal(t, router.RouteSnapshot(cache), "conn-check-router -> Count:2 {0,2}")
-}
-
-func setUpAddrCache(r router.Poolable, addrs []protocol.Invoker) router.Cache {
- pool, info := r.Pool(addrs)
- cache := chain.BuildCache(addrs)
- cache.SetAddrMeta(r.Name(), info)
- cache.SetAddrPool(r.Name(), pool)
- return cache
-}
diff --git a/cluster/router/conncheck/conn_health_check.go b/cluster/router/conncheck/conn_health_check.go
deleted file mode 100644
index 9f05b06..0000000
--- a/cluster/router/conncheck/conn_health_check.go
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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 conncheck
-
-import (
- "github.com/apache/dubbo-go/cluster/router"
- "github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/common/constant"
- "github.com/apache/dubbo-go/common/extension"
- "github.com/apache/dubbo-go/protocol"
-)
-
-func init() {
- extension.SetConnChecker(constant.DEFAULT_CONN_CHECKER, NewDefaultConnChecker)
-}
-
-// DefaultConnChecker is the default implementation of ConnChecker, which determines the health status of invoker conn
-type DefaultConnChecker struct {
-}
-
-// IsConnHealthy evaluates the healthy state on the given Invoker based on the number of successive bad request
-// and the current active request
-func (c *DefaultConnChecker) IsConnHealthy(invoker protocol.Invoker) bool {
- return protocol.GetInvokerHealthyStatus(invoker)
-}
-
-// NewDefaultConnChecker constructs a new DefaultConnChecker based on the url
-func NewDefaultConnChecker(url *common.URL) router.ConnChecker {
- return &DefaultConnChecker{}
-}
diff --git a/cluster/router/conncheck/conn_health_check_test.go b/cluster/router/conncheck/conn_health_check_test.go
deleted file mode 100644
index a7b6fe3..0000000
--- a/cluster/router/conncheck/conn_health_check_test.go
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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 conncheck
-
-import (
- "fmt"
- "testing"
-)
-
-import (
- "github.com/stretchr/testify/assert"
-)
-
-import (
- "github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/protocol"
-)
-
-const (
- connCheckDubbo1010IP = "192.168.10.10"
- connCheckDubboUrlFormat = "dubbo://%s:20000/com.ikurento.user.UserProvider"
-)
-
-func TestDefaultConnCheckerIsHealthy(t *testing.T) {
- defer protocol.CleanAllStatus()
- url, _ := common.NewURL(fmt.Sprintf(connCheckDubboUrlFormat, connCheckDubbo1010IP))
- cc := NewDefaultConnChecker(url).(*DefaultConnChecker)
- invoker := NewMockInvoker(url)
- healthy := cc.IsConnHealthy(invoker)
- assert.True(t, healthy)
-
- invoker = NewMockInvoker(url)
- cc = NewDefaultConnChecker(url).(*DefaultConnChecker)
- // add to black list
- protocol.NewServiceState(url.ServiceKey()).SetInvokerUnhealthyStatus(invoker)
- assert.False(t, cc.IsConnHealthy(invoker))
-}
diff --git a/cluster/router/conncheck/factory.go b/cluster/router/conncheck/factory.go
deleted file mode 100644
index a7b19aa..0000000
--- a/cluster/router/conncheck/factory.go
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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 conncheck
-
-import (
- "github.com/apache/dubbo-go/cluster/router"
- "github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/common/constant"
- "github.com/apache/dubbo-go/common/extension"
-)
-
-func init() {
- extension.SetRouterFactory(constant.ConnCheckRouterName, newConnCheckRouteFactory)
-}
-
-// ConnCheckRouteFactory is the factory to create conn check router, it aims at filter ip with unhealthy status
-// the unhealthy status is storied in protocol/rpc_status.go with sync.Map
-type ConnCheckRouteFactory struct {
-}
-
-// newConnCheckRouteFactory construct a new ConnCheckRouteFactory
-func newConnCheckRouteFactory() router.PriorityRouterFactory {
- return &ConnCheckRouteFactory{}
-}
-
-// NewPriorityRouter construct a new NewConnCheckRouter via url
-func (f *ConnCheckRouteFactory) NewPriorityRouter(url *common.URL, notify chan struct{}) (router.PriorityRouter, error) {
- return NewConnCheckRouter(url, notify)
-}
diff --git a/cluster/router/conncheck/factory_test.go b/cluster/router/conncheck/factory_test.go
deleted file mode 100644
index 166aec3..0000000
--- a/cluster/router/conncheck/factory_test.go
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * 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 conncheck
-
-import (
- "context"
- "testing"
-)
-
-import (
- "github.com/stretchr/testify/assert"
-)
-
-import (
- "github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/protocol"
-)
-
-// nolint
-type MockInvoker struct {
- url *common.URL
-}
-
-// nolint
-func NewMockInvoker(url *common.URL) *MockInvoker {
- return &MockInvoker{
- url: url,
- }
-}
-
-// nolint
-func (bi *MockInvoker) GetURL() *common.URL {
- return bi.url
-}
-
-// nolint
-func (bi *MockInvoker) IsAvailable() bool {
- return true
-}
-
-// nolint
-func (bi *MockInvoker) IsDestroyed() bool {
- return true
-}
-
-// nolint
-func (bi *MockInvoker) Invoke(_ context.Context, _ protocol.Invocation) protocol.Result {
- return nil
-}
-
-// nolint
-func (bi *MockInvoker) Destroy() {
-}
-
-// nolint
-func TestHealthCheckRouteFactory(t *testing.T) {
- factory := newConnCheckRouteFactory()
- assert.NotNil(t, factory)
-}
diff --git a/common/extension/conn_checker.go b/common/extension/conn_checker.go
deleted file mode 100644
index fbd9e34..0000000
--- a/common/extension/conn_checker.go
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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 (
- "github.com/apache/dubbo-go/cluster/router"
- "github.com/apache/dubbo-go/common"
-)
-
-var (
- connCheckers = make(map[string]func(url *common.URL) router.ConnChecker)
-)
-
-// SetHealthChecker sets the HealthChecker with @name
-func SetConnChecker(name string, fcn func(_ *common.URL) router.ConnChecker) {
- connCheckers[name] = fcn
-}
-
-// GetHealthChecker gets the HealthChecker with @name
-func GetConnChecker(name string, url *common.URL) router.ConnChecker {
- f, ok := connCheckers[name]
- if !ok || f == nil {
- panic("connCheckers for " + name + " is not existing, make sure you have import the package.")
- }
- return connCheckers[name](url)
-}
diff --git a/protocol/rpc_status.go b/protocol/rpc_status.go
index a878d75..65ffc6e 100644
--- a/protocol/rpc_status.go
+++ b/protocol/rpc_status.go
@@ -25,8 +25,6 @@ import (
import (
"github.com/apache/dubbo-go/common"
- "github.com/apache/dubbo-go/common/constant"
- "github.com/apache/dubbo-go/common/logger"
)
var (
@@ -187,16 +185,6 @@ func CleanAllStatus() {
return true
}
serviceStatistic.Range(delete2)
- delete3 := func(_, value interface{}) bool {
- if v, ok := value.(*ServiceHealthState); ok {
- v.blackList.Range(func(key, value interface{}) bool {
- v.blackList.Delete(key)
- return true
- })
- }
- return true
- }
- serviceStateMap.Range(delete3)
}
// if the ip is changed in kubernetes, then the ip will not exist. So we should recycle the invoker.
@@ -218,17 +206,6 @@ func (s *invokerState) increateRetryTimes() {
s.retryTimes++
}
-// GetInvokerHealthyStatus get invoker's conn healthy status
-func GetInvokerHealthyStatus(invoker Invoker) bool {
- if v, ok := serviceStateMap.Load(invoker.GetURL().ServiceKey()); ok {
- if state, ok := v.(*ServiceHealthState); ok {
- _, found := state.blackList.Load(invoker.GetURL().Key())
- return !found
- }
- }
- return true
-}
-
// nolint
func GetAndRefreshState(url *common.URL) bool {
if v, ok := serviceStateMap.Load(url.ServiceKey()); ok {
@@ -245,7 +222,6 @@ type ServiceHealthState struct {
refreshState *int32
refresh atomic.Value
rebuildRoute *int32
- blackList sync.Map // store unhealthy url blackList
}
func NewServiceState(serviceKey string) *ServiceHealthState {
@@ -261,15 +237,6 @@ func NewServiceState(serviceKey string) *ServiceHealthState {
return serviceState
}
-func (s *ServiceHealthState) reset() {
- s.refresh.Store(false)
- atomic.StoreInt32(s.rebuildRoute, 0)
- s.blackList.Range(func(key, value interface{}) bool {
- s.blackList.Delete(key)
- return true
- })
-}
-
func (s *ServiceHealthState) configNeedRefresh(needRefresh bool) {
s.refresh.Store(needRefresh)
}
@@ -282,95 +249,6 @@ func (s *ServiceHealthState) needRefresh() bool {
return v.(bool)
}
-// SetInvokerUnhealthyStatus add target invoker to black list
-func (s *ServiceHealthState) SetInvokerUnhealthyStatus(invoker Invoker) {
- s.configNeedRefresh(true)
- s.blackList.Store(invoker.GetURL().Key(), newInvokeState(invoker))
- logger.Infof("Add invoker ip=%s to black list for service(%s)", invoker.GetURL().Location, invoker.GetURL().ServiceKey())
- s.activeBlackListCacheDirty()
-}
-
-// RemoveInvokerUnhealthyStatus remove unhealthy status of target invoker from blacklist
-func (s *ServiceHealthState) RemoveInvokerUnhealthyStatus(invoker Invoker) {
- s.blackList.Delete(invoker.GetURL().Key())
- logger.Infof("Remove invoker ip(%s) from black list for service(%s)", invoker.GetURL().Location, invoker.GetURL().ServiceKey())
- s.activeBlackListCacheDirty()
-}
-
-// GetBlackListInvokers get at most size of blockSize invokers from black list
-func (s *ServiceHealthState) GetBlackListInvokers(blockSize int) []*invokerState {
- resultIvks := make([]*invokerState, 0, blockSize)
- s.blackList.Range(func(k, v interface{}) bool {
- if v == nil {
- return true
- }
- resultIvks = append(resultIvks, v.(*invokerState))
- if len(resultIvks) == blockSize {
- return false
- }
- return true
- })
- return resultIvks
-}
-
-// RemoveUrlKeyUnhealthyStatus called when event of provider unregister, delete from black list
-func (s *ServiceHealthState) RemoveUrlKeyUnhealthyStatus(key string) {
- if _, ok := s.blackList.Load(key); ok {
- s.blackList.Delete(key)
- s.activeBlackListCacheDirty()
- }
- logger.Info("Remove invoker key = ", key, " from black list")
-}
-
func (s *ServiceHealthState) activeBlackListCacheDirty() {
atomic.StoreInt32(s.rebuildRoute, 1)
}
-
-// TryRefreshBlackList start 3 gr to check at most block=16 invokers in black list
-// if target invoker is available, then remove it from black list
-func (s *ServiceHealthState) TryRefreshBlackList() {
- if s.needRefresh() {
- go s.refreshBlackList()
- }
-}
-
-func (s *ServiceHealthState) refreshBlackList() {
- defer func() {
- if r := recover(); r != nil {
- logger.Errorf("try to refresh black list failed: %s, %+v", s.serviceKey, r)
- }
- }()
-
- if atomic.CompareAndSwapInt32(s.refreshState, 0, 1) {
- defer func() {
- atomic.CompareAndSwapInt32(s.refreshState, 1, 0)
- }()
-
- ivkStates := s.GetBlackListInvokers(constant.DEFAULT_BLACK_LIST_RECOVER_BLOCK)
- logger.Debug("blackList len = ", len(ivkStates))
- if len(ivkStates) == 0 {
- logger.Infof("there is no data in black list(%s), and will not refresh black list.", s.serviceKey)
- s.configNeedRefresh(false)
- return
- }
- wg := sync.WaitGroup{}
- for i := 0; i < 3; i++ {
- wg.Add(1)
- go func(ivks []*invokerState, i int) {
- defer wg.Done()
- for j := range ivks {
- if j%3-i == 0 && ivks[j].invoker.IsAvailable() {
- s.RemoveInvokerUnhealthyStatus(ivks[j].invoker)
- } else {
- ivks[j].increateRetryTimes()
- }
- // if the ip is changed in kubernetes, then the ip will not exist. So we should recycle the invoker.
- if ivks[j].retryTimes > constant.DEFAULT_BLACK_LIST_MAX_RETRY_TIMES {
- s.RemoveInvokerUnhealthyStatus(ivks[j].invoker)
- }
- }
- }(ivkStates, i)
- }
- wg.Wait()
- }
-}
diff --git a/registry/directory/directory.go b/registry/directory/directory.go
index 602ea57..7441aca 100644
--- a/registry/directory/directory.go
+++ b/registry/directory/directory.go
@@ -330,7 +330,6 @@ func (dir *RegistryDirectory) uncacheInvoker(event *registry.ServiceEvent) proto
func (dir *RegistryDirectory) uncacheInvokerWithKey(key string) protocol.Invoker {
logger.Debugf("service will be deleted in cache invokers: invokers key is %s!", key)
- dir.serviceHealthState.RemoveUrlKeyUnhealthyStatus(key)
if cacheInvoker, ok := dir.cacheInvokersMap.Load(key); ok {
dir.cacheInvokersMap.Delete(key)
return cacheInvoker.(protocol.Invoker)