This is an automated email from the ASF dual-hosted git repository.
littlecui pushed a commit to branch master
in repository
https://gitbox.apache.org/repos/asf/incubator-servicecomb-service-center.git
The following commit(s) were added to refs/heads/master by this push:
new 0ebf06e SCB-393 Retry to request etcd too frequently when fail (#352)
0ebf06e is described below
commit 0ebf06e8f082ef9c0ac84cc2040eaa18fc899213
Author: little-cui <sure_0...@qq.com>
AuthorDate: Thu May 17 16:12:24 2018 +0800
SCB-393 Retry to request etcd too frequently when fail (#352)
* SCB-393 Retry to request etcd too frequently when fail
* SCB-393 Retry to request etcd too frequently when fail
* SCB-393 Restructure core package
* SCB-393 Restructure core package
* SCB-549 Do not check the instance endpoint whether belong to other service
* SCB-549 Fix UT failure.
* SCB-549 Fix UT failure.
* SCB-549 Fix UT failure.
---
integration/instances_test.go | 1 +
pkg/util/{util_test.go => backoff.go} | 50 +++++++++----------
pkg/util/{util_test.go => backoff_test.go} | 52 ++++++++++---------
pkg/util/util.go | 21 ++++----
pkg/util/util_test.go | 27 ++++++++++
server/broker/store.go | 58 +++++++++++-----------
server/core/backend/backend.go | 6 +--
server/core/backend/{store => }/cache_kv.go | 21 ++++----
server/core/backend/{store => }/cache_null.go | 2 +-
server/core/backend/{store => }/cacher.go | 2 +-
server/core/backend/{store => }/cacher_test.go | 2 +-
server/core/backend/{store => }/common.go | 4 +-
server/core/backend/{store => }/config.go | 2 +-
server/core/backend/{store => }/defer.go | 2 +-
server/core/backend/{store => }/defer_instance.go | 2 +-
server/core/backend/{store => }/defer_test.go | 2 +-
server/core/backend/{store => }/event.go | 2 +-
server/core/backend/{store => }/event_proxy.go | 2 +-
server/core/backend/{store => }/extend.go | 2 +-
server/core/backend/{store => }/extend_test.go | 2 +-
server/core/backend/{store => }/indexer.go | 49 ++++++++++--------
server/core/backend/{store => }/lease.go | 5 +-
server/core/backend/{store => }/listwatch.go | 4 +-
server/core/backend/{store => }/metric.go | 7 ++-
server/core/backend/{store => }/store.go | 2 +-
server/core/info.go | 39 ---------------
server/core/key_generator.go | 2 +-
server/govern/service.go | 14 +++---
server/handler/cache/cache.go | 4 +-
server/plugin/infra/quota/buildin/buildin.go | 12 ++---
.../infra/registry/embededetcd/embededetcd.go | 4 +-
server/plugin/infra/registry/etcd/logger.go | 9 +---
server/plugin/infra/tracing/buildin/common.go | 2 +-
server/server.go | 44 ++++++++++++++--
server/service/event/dependency_event_handler.go | 28 +++++------
server/service/event/event.go | 12 ++---
server/service/event/instance_event_handler.go | 8 +--
server/service/event/rule_event_handler.go | 8 +--
server/service/event/service_event_handler.go | 8 +--
server/service/event/tag_event_handler.go | 8 +--
server/service/instances.go | 3 --
server/service/instances_test.go | 9 ----
server/service/microservices.go | 3 +-
server/service/schema.go | 13 +++--
server/service/util/dependency.go | 9 ++--
server/service/util/domain_util.go | 7 ++-
server/service/util/heartbeat_util.go | 4 +-
server/service/util/instance_util.go | 36 ++++----------
server/service/util/instance_util_test.go | 2 +-
server/service/util/microservice_util.go | 20 ++++----
server/service/util/rule_util.go | 10 ++--
server/service/util/schema_util.go | 4 +-
server/service/util/tag_util.go | 3 +-
53 files changed, 322 insertions(+), 332 deletions(-)
diff --git a/integration/instances_test.go b/integration/instances_test.go
index 948c9f6..5004321 100644
--- a/integration/instances_test.go
+++ b/integration/instances_test.go
@@ -168,6 +168,7 @@ var _ = Describe("MicroService Api Test", func() {
"times": 2,
}
instance := map[string]interface{}{
+ "instanceId": serviceInstanceID,
"endpoints": endpoints,
"hostName": "cse",
"status": "UP",
diff --git a/pkg/util/util_test.go b/pkg/util/backoff.go
similarity index 57%
copy from pkg/util/util_test.go
copy to pkg/util/backoff.go
index 16f342f..7f96cfd 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/backoff.go
@@ -17,37 +17,35 @@
package util
import (
- "testing"
+ "math"
+ "time"
)
-func TestBytesToInt32(t *testing.T) {
- bs := []byte{0, 0, 0, 1}
- i := BytesToInt32(bs)
- if i != 1 {
- t.FailNow()
- }
+var DefaultBackoff Backoff = &PowerBackoff{
+ MaxDelay: 30 * time.Second,
+ InitDelay: 1 * time.Second,
+ Factor: 1.6,
+}
- bs = []byte{1, 0, 0, 0}
- i = BytesToInt32(bs)
- if i != 1<<(3*8) {
- t.FailNow()
- }
+type Backoff interface {
+ Delay(retries int) time.Duration
+}
- bs = []byte{0, 0, 0, 0, 1}
- i = BytesToInt32(bs)
- if i != 0 {
- t.FailNow()
- }
+// delay = min(MaxDelay, InitDelay * power(Factor, retries))
+type PowerBackoff struct {
+ MaxDelay time.Duration
+ InitDelay time.Duration
+ Factor float64
+}
- bs = []byte{1}
- i = BytesToInt32(bs)
- if i != 1 {
- t.FailNow()
+func (pb *PowerBackoff) Delay(retries int) time.Duration {
+ if retries <= 0 {
+ return pb.InitDelay
}
- bs = []byte{1, 0}
- i = BytesToInt32(bs)
- if i != 1<<8 {
- t.FailNow()
- }
+ return time.Duration(math.Min(float64(pb.MaxDelay),
float64(pb.InitDelay)*math.Pow(pb.Factor, float64(retries))))
+}
+
+func GetBackoff() Backoff {
+ return DefaultBackoff
}
diff --git a/pkg/util/util_test.go b/pkg/util/backoff_test.go
similarity index 51%
copy from pkg/util/util_test.go
copy to pkg/util/backoff_test.go
index 16f342f..301a914 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/backoff_test.go
@@ -18,36 +18,40 @@ package util
import (
"testing"
+ "time"
)
-func TestBytesToInt32(t *testing.T) {
- bs := []byte{0, 0, 0, 1}
- i := BytesToInt32(bs)
- if i != 1 {
- t.FailNow()
+func TestPowerBackoff_Delay(t *testing.T) {
+ i, m := time.Second, 30*time.Second
+ b := &PowerBackoff{
+ MaxDelay: 30 * time.Second,
+ InitDelay: 1 * time.Second,
+ Factor: 1.6,
}
-
- bs = []byte{1, 0, 0, 0}
- i = BytesToInt32(bs)
- if i != 1<<(3*8) {
- t.FailNow()
+ r := b.Delay(-1)
+ if r != i {
+ t.Fatalf("TestPowerBackoff_Delay -1 failed, result is %s", r)
}
-
- bs = []byte{0, 0, 0, 0, 1}
- i = BytesToInt32(bs)
- if i != 0 {
- t.FailNow()
+ r = b.Delay(0)
+ if r != i {
+ t.Fatalf("TestPowerBackoff_Delay 0 failed, result is %s", r)
}
-
- bs = []byte{1}
- i = BytesToInt32(bs)
- if i != 1 {
- t.FailNow()
+ r = b.Delay(1)
+ if r != 1600*time.Millisecond {
+ t.Fatalf("TestPowerBackoff_Delay 1 failed, result is %s", r)
}
+ r = b.Delay(4)
+ if r != 6553600*time.Microsecond {
+ t.Fatalf("TestPowerBackoff_Delay 4 failed, result is %s", r)
+ }
+ r = b.Delay(8)
+ if r != m {
+ t.Fatalf("TestPowerBackoff_Delay 8 failed, result is %s", r)
+ }
+}
- bs = []byte{1, 0}
- i = BytesToInt32(bs)
- if i != 1<<8 {
- t.FailNow()
+func TestGetBackoff(t *testing.T) {
+ if GetBackoff() == nil {
+ t.Fatalf("TestGetBackoff failed")
}
}
diff --git a/pkg/util/util.go b/pkg/util/util.go
index d63191a..a8ca183 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -142,25 +142,26 @@ func LogPanic(args ...interface{}) {
}
if strings.Index(file, "service-center") > 0 ||
strings.Index(file, "servicecenter") > 0 {
- idx := strings.LastIndex(file, "/")
- if idx >= 0 {
- file = file[idx+1:]
- }
- Logger().Errorf(nil, "recover from %s %s():%d! %s",
file, method, line, fmt.Sprint(args...))
+ Logger().Errorf(nil, "recover from %s %s():%d! %s",
FileLastName(file), method, line, fmt.Sprint(args...))
return
}
}
file, method, line, _ := GetCaller(0)
- idx := strings.LastIndex(file, "/")
- if idx >= 0 {
- file = file[idx+1:]
- }
fmt.Fprintln(os.Stderr,
time.Now().Format("2006-01-02T15:04:05.000Z07:00"), "FATAL", "system",
os.Getpid(),
- fmt.Sprintf("%s %s():%d", file, method, line),
fmt.Sprint(args...))
+ fmt.Sprintf("%s %s():%d", FileLastName(file), method, line),
fmt.Sprint(args...))
fmt.Fprintln(os.Stderr, BytesToStringWithNoCopy(debug.Stack()))
}
+func FileLastName(file string) string {
+ if sp1 := strings.LastIndex(file, "/"); sp1 >= 0 {
+ if sp2 := strings.LastIndex(file[:sp1], "/"); sp2 >= 0 {
+ file = file[sp2+1:]
+ }
+ }
+ return file
+}
+
func GetCaller(skip int) (string, string, int, bool) {
pc, file, line, ok := runtime.Caller(skip + 1)
method := FormatFuncName(runtime.FuncForPC(pc).Name())
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index 16f342f..94e1aa3 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -51,3 +51,30 @@ func TestBytesToInt32(t *testing.T) {
t.FailNow()
}
}
+
+func TestFileLastName(t *testing.T) {
+ n := FileLastName("")
+ if n != "" {
+ t.Fatal("TestFileLastName '' failed", n)
+ }
+ n = FileLastName("a")
+ if n != "a" {
+ t.Fatal("TestFileLastName 'a' failed", n)
+ }
+ n = FileLastName("a/b")
+ if n != "a/b" {
+ t.Fatal("TestFileLastName 'a/b' failed", n)
+ }
+ n = FileLastName("a/b/c")
+ if n != "b/c" {
+ t.Fatal("TestFileLastName 'b/c' failed", n)
+ }
+ n = FileLastName("b/")
+ if n != "b/" {
+ t.Fatal("TestFileLastName 'b' failed", n)
+ }
+ n = FileLastName("/")
+ if n != "/" {
+ t.Fatal("TestFileLastName 'b' failed", n)
+ }
+}
diff --git a/server/broker/store.go b/server/broker/store.go
index 3bfe982..51acaca 100644
--- a/server/broker/store.go
+++ b/server/broker/store.go
@@ -17,61 +17,61 @@
package broker
import (
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
)
var (
- PARTICIPANT store.StoreType
- VERSION store.StoreType
- PACT store.StoreType
- PACT_VERSION store.StoreType
- PACT_TAG store.StoreType
- VERIFICATION store.StoreType
- PACT_LATEST store.StoreType
+ PARTICIPANT backend.StoreType
+ VERSION backend.StoreType
+ PACT backend.StoreType
+ PACT_VERSION backend.StoreType
+ PACT_TAG backend.StoreType
+ VERIFICATION backend.StoreType
+ PACT_LATEST backend.StoreType
)
var brokerKvStore = &BKvStore{}
func init() {
- PARTICIPANT = store.Store().MustInstall(store.NewEntity("PARTICIPANT",
GetBrokerParticipantKey("")))
- VERSION = store.Store().MustInstall(store.NewEntity("VERSION",
GetBrokerVersionKey("")))
- PACT = store.Store().MustInstall(store.NewEntity("PACT",
GetBrokerPactKey("")))
- PACT_VERSION =
store.Store().MustInstall(store.NewEntity("PACT_VERSION",
GetBrokerPactVersionKey("")))
- PACT_TAG = store.Store().MustInstall(store.NewEntity("PACT_TAG",
GetBrokerTagKey("")))
- VERIFICATION =
store.Store().MustInstall(store.NewEntity("VERIFICATION",
GetBrokerVerificationKey("")))
- PACT_LATEST = store.Store().MustInstall(store.NewEntity("PACT_LATEST",
GetBrokerLatestKey("")))
+ PARTICIPANT =
backend.Store().MustInstall(backend.NewEntity("PARTICIPANT",
GetBrokerParticipantKey("")))
+ VERSION = backend.Store().MustInstall(backend.NewEntity("VERSION",
GetBrokerVersionKey("")))
+ PACT = backend.Store().MustInstall(backend.NewEntity("PACT",
GetBrokerPactKey("")))
+ PACT_VERSION =
backend.Store().MustInstall(backend.NewEntity("PACT_VERSION",
GetBrokerPactVersionKey("")))
+ PACT_TAG = backend.Store().MustInstall(backend.NewEntity("PACT_TAG",
GetBrokerTagKey("")))
+ VERIFICATION =
backend.Store().MustInstall(backend.NewEntity("VERIFICATION",
GetBrokerVerificationKey("")))
+ PACT_LATEST =
backend.Store().MustInstall(backend.NewEntity("PACT_LATEST",
GetBrokerLatestKey("")))
}
type BKvStore struct {
}
-func (s *BKvStore) Participant() *store.Indexer {
- return store.Store().Entity(PARTICIPANT)
+func (s *BKvStore) Participant() *backend.Indexer {
+ return backend.Store().Entity(PARTICIPANT)
}
-func (s *BKvStore) Version() *store.Indexer {
- return store.Store().Entity(VERSION)
+func (s *BKvStore) Version() *backend.Indexer {
+ return backend.Store().Entity(VERSION)
}
-func (s *BKvStore) Pact() *store.Indexer {
- return store.Store().Entity(PACT)
+func (s *BKvStore) Pact() *backend.Indexer {
+ return backend.Store().Entity(PACT)
}
-func (s *BKvStore) PactVersion() *store.Indexer {
- return store.Store().Entity(PACT_VERSION)
+func (s *BKvStore) PactVersion() *backend.Indexer {
+ return backend.Store().Entity(PACT_VERSION)
}
-func (s *BKvStore) PactTag() *store.Indexer {
- return store.Store().Entity(PACT_TAG)
+func (s *BKvStore) PactTag() *backend.Indexer {
+ return backend.Store().Entity(PACT_TAG)
}
-func (s *BKvStore) Verification() *store.Indexer {
- return store.Store().Entity(VERIFICATION)
+func (s *BKvStore) Verification() *backend.Indexer {
+ return backend.Store().Entity(VERIFICATION)
}
-func (s *BKvStore) PactLatest() *store.Indexer {
- return store.Store().Entity(PACT_LATEST)
+func (s *BKvStore) PactLatest() *backend.Indexer {
+ return backend.Store().Entity(PACT_LATEST)
}
func Store() *BKvStore {
diff --git a/server/core/backend/backend.go b/server/core/backend/backend.go
index 9711e44..8fe096d 100644
--- a/server/core/backend/backend.go
+++ b/server/core/backend/backend.go
@@ -29,7 +29,6 @@ import (
var (
registryInstance registry.Registry
singletonLock sync.Mutex
- wait_delay = []int{1, 1, 5, 10, 20, 30, 60}
)
const (
@@ -65,10 +64,7 @@ func Registry() registry.Registry {
return registryInstance
}
- if i >= len(wait_delay) {
- i = len(wait_delay) - 1
- }
- t := time.Duration(wait_delay[i]) * time.Second
+ t := util.GetBackoff().Delay(i)
util.Logger().Errorf(nil, "initialize service center
failed, retry after %s", t)
<-time.After(t)
}
diff --git a/server/core/backend/store/cache_kv.go
b/server/core/backend/cache_kv.go
similarity index 97%
rename from server/core/backend/store/cache_kv.go
rename to server/core/backend/cache_kv.go
index 9fb347e..2b46fe6 100644
--- a/server/core/backend/store/cache_kv.go
+++ b/server/core/backend/cache_kv.go
@@ -14,11 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
"github.com/coreos/etcd/mvcc/mvccpb"
"golang.org/x/net/context"
@@ -212,21 +211,23 @@ func (c *KvCacher) needDeferHandle(evts []KvEvent) bool {
func (c *KvCacher) refresh(ctx context.Context) {
util.Logger().Debugf("start to list and watch %s", c.Cfg)
+ retries := 0
for {
- start := time.Now()
- c.ListAndWatch(ctx)
- watchDuration := time.Since(start)
nextPeriod := minWaitInterval
- if watchDuration > 0 && c.Cfg.Period > watchDuration {
- nextPeriod = c.Cfg.Period - watchDuration
+ if err := c.ListAndWatch(ctx); err != nil {
+ nextPeriod = util.GetBackoff().Delay(retries)
+ retries++
+ } else {
+ retries = 0
+
+ ReportCacheMetrics(c.Name(), "raw", c.cache.RLock())
+ c.cache.RUnlock()
}
select {
case <-ctx.Done():
util.Logger().Debugf("stop to list and watch %s", c.Cfg)
return
case <-time.After(nextPeriod):
- ReportCacheMetrics(c.Name(), "raw", c.cache.RLock())
- c.cache.RUnlock()
}
}
}
@@ -503,7 +504,7 @@ func NewKvCacher(name string, opts ...ConfigOption)
*KvCacher {
Cfg: cfg,
ready: make(chan struct{}),
lw: ListWatcher{
- Client: backend.Registry(),
+ Client: Registry(),
Prefix: cfg.Prefix,
},
goroutine: util.NewGo(context.Background()),
diff --git a/server/core/backend/store/cache_null.go
b/server/core/backend/cache_null.go
similarity index 98%
rename from server/core/backend/store/cache_null.go
rename to server/core/backend/cache_null.go
index 2025bff..d6681aa 100644
--- a/server/core/backend/store/cache_null.go
+++ b/server/core/backend/cache_null.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
var (
NullCache = &nullCache{}
diff --git a/server/core/backend/store/cacher.go b/server/core/backend/cacher.go
similarity index 98%
rename from server/core/backend/store/cacher.go
rename to server/core/backend/cacher.go
index 898964c..410ef50 100644
--- a/server/core/backend/store/cacher.go
+++ b/server/core/backend/cacher.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
type Cache interface {
Version() int64
diff --git a/server/core/backend/store/cacher_test.go
b/server/core/backend/cacher_test.go
similarity index 99%
rename from server/core/backend/store/cacher_test.go
rename to server/core/backend/cacher_test.go
index 3225757..9e06f6d 100644
--- a/server/core/backend/store/cacher_test.go
+++ b/server/core/backend/cacher_test.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"encoding/json"
diff --git a/server/core/backend/store/common.go b/server/core/backend/common.go
similarity index 98%
rename from server/core/backend/store/common.go
rename to server/core/backend/common.go
index c9d9272..ed0727f 100644
--- a/server/core/backend/store/common.go
+++ b/server/core/backend/common.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
@@ -124,7 +124,7 @@ const (
const (
DEFAULT_COMPACT_TIMES = 3
DEFAULT_COMPACT_TIMEOUT = 5 * time.Minute
- minWaitInterval = 100 * time.Millisecond
+ minWaitInterval = 1 * time.Second
eventBlockSize = 1000
)
diff --git a/server/core/backend/store/config.go b/server/core/backend/config.go
similarity index 99%
rename from server/core/backend/store/config.go
rename to server/core/backend/config.go
index 9f0b2dd..13e162d 100644
--- a/server/core/backend/store/config.go
+++ b/server/core/backend/config.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"fmt"
diff --git a/server/core/backend/store/defer.go b/server/core/backend/defer.go
similarity index 98%
rename from server/core/backend/store/defer.go
rename to server/core/backend/defer.go
index da75b21..03edbe6 100644
--- a/server/core/backend/store/defer.go
+++ b/server/core/backend/defer.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
type DeferHandler interface {
OnCondition(Cache, []KvEvent) bool
diff --git a/server/core/backend/store/defer_instance.go
b/server/core/backend/defer_instance.go
similarity index 99%
rename from server/core/backend/store/defer_instance.go
rename to server/core/backend/defer_instance.go
index 177d8aa..d62afae 100644
--- a/server/core/backend/store/defer_instance.go
+++ b/server/core/backend/defer_instance.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"encoding/json"
diff --git a/server/core/backend/store/defer_test.go
b/server/core/backend/defer_test.go
similarity index 99%
rename from server/core/backend/store/defer_test.go
rename to server/core/backend/defer_test.go
index 4a034a6..93acd97 100644
--- a/server/core/backend/store/defer_test.go
+++ b/server/core/backend/defer_test.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"encoding/json"
diff --git a/server/core/backend/store/event.go b/server/core/backend/event.go
similarity index 98%
rename from server/core/backend/store/event.go
rename to server/core/backend/event.go
index 67714a0..839dc7a 100644
--- a/server/core/backend/store/event.go
+++ b/server/core/backend/event.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
diff --git a/server/core/backend/store/event_proxy.go
b/server/core/backend/event_proxy.go
similarity index 98%
rename from server/core/backend/store/event_proxy.go
rename to server/core/backend/event_proxy.go
index e66583d..592c70a 100644
--- a/server/core/backend/store/event_proxy.go
+++ b/server/core/backend/event_proxy.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import "sync"
diff --git a/server/core/backend/store/extend.go b/server/core/backend/extend.go
similarity index 99%
rename from server/core/backend/store/extend.go
rename to server/core/backend/extend.go
index 90b053f..200d45e 100644
--- a/server/core/backend/store/extend.go
+++ b/server/core/backend/extend.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"errors"
diff --git a/server/core/backend/store/extend_test.go
b/server/core/backend/extend_test.go
similarity index 98%
rename from server/core/backend/store/extend_test.go
rename to server/core/backend/extend_test.go
index 8359dd2..6a37ce5 100644
--- a/server/core/backend/store/extend_test.go
+++ b/server/core/backend/extend_test.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import "testing"
diff --git a/server/core/backend/store/indexer.go
b/server/core/backend/indexer.go
similarity index 90%
rename from server/core/backend/store/indexer.go
rename to server/core/backend/indexer.go
index 2b35a97..48889e8 100644
--- a/server/core/backend/store/indexer.go
+++ b/server/core/backend/indexer.go
@@ -14,11 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
+ "github.com/apache/incubator-servicecomb-service-center/server/core"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"github.com/coreos/etcd/mvcc/mvccpb"
@@ -35,6 +35,7 @@ type Indexer struct {
cacher Cacher
goroutine *util.GoRoutine
ready chan struct{}
+ lastMaxSize int
prefixIndex map[string]map[string]struct{}
prefixBuildQueue chan KvEvent
prefixLock sync.RWMutex
@@ -46,12 +47,13 @@ func (i *Indexer) Search(ctx context.Context, opts
...registry.PluginOpOption) (
key := util.BytesToStringWithNoCopy(op.Key)
- if op.Mode == registry.MODE_NO_CACHE ||
+ if !core.ServerInfo.Config.EnableCache ||
+ op.Mode == registry.MODE_NO_CACHE ||
op.Revision > 0 ||
(op.Offset >= 0 && op.Limit > 0) {
util.Logger().Debugf("search %s match special options, request
etcd server, opts: %s",
i.cacher.Name(), op)
- return backend.Registry().Do(ctx, opts...)
+ return Registry().Do(ctx, opts...)
}
if op.Prefix {
@@ -66,7 +68,7 @@ func (i *Indexer) Search(ctx context.Context, opts
...registry.PluginOpOption) (
util.Logger().Debugf("can not find any key from %s cache with
prefix, request etcd server, key: %s",
i.cacher.Name(), key)
- return backend.Registry().Do(ctx, opts...)
+ return Registry().Do(ctx, opts...)
}
resp := ®istry.PluginResponse{
@@ -86,7 +88,7 @@ func (i *Indexer) Search(ctx context.Context, opts
...registry.PluginOpOption) (
}
util.Logger().Debugf("%s cache does not store this key, request
etcd server, key: %s", i.cacher.Name(), key)
- return backend.Registry().Do(ctx, opts...)
+ return Registry().Do(ctx, opts...)
}
cacheData := i.Cache().Data(key)
@@ -97,7 +99,7 @@ func (i *Indexer) Search(ctx context.Context, opts
...registry.PluginOpOption) (
util.Logger().Debugf("do not match any key in %s cache store,
request etcd server, key: %s",
i.cacher.Name(), key)
- return backend.Registry().Do(ctx, opts...)
+ return Registry().Do(ctx, opts...)
}
resp.Count = 1
@@ -195,18 +197,27 @@ func (i *Indexer) buildIndex() {
default:
i.addPrefixKey(prefix, key)
}
- i.prefixLock.Unlock()
- util.LogNilOrWarnf(t, "too long to
rebuild(action: %s) index[%d], key is %s",
- evt.Type, key, len(i.prefixIndex))
- case <-time.After(10 * time.Second):
- i.prefixLock.Lock()
- if time.Now().Sub(lastCompactTime) >=
DEFAULT_COMPACT_TIMEOUT {
+ // compact
+ initSize, l := DEFAULT_CACHE_INIT_SIZE,
len(i.prefixIndex)
+ if i.lastMaxSize < l {
+ i.lastMaxSize = l
+ }
+ if initSize >= l &&
+ i.lastMaxSize >=
initSize*DEFAULT_COMPACT_TIMES &&
+ time.Now().Sub(lastCompactTime) >=
DEFAULT_COMPACT_TIMEOUT {
i.compact()
+ i.lastMaxSize = l
lastCompactTime = time.Now()
}
+
+ // report metrics
ReportCacheMetrics(i.cacher.Name(), "index",
i.prefixIndex)
+
i.prefixLock.Unlock()
+
+ util.LogNilOrWarnf(t, "too long to
rebuild(action: %s) index[%d], key is %s",
+ evt.Type, key, len(i.prefixIndex))
}
}
util.Logger().Debugf("the goroutine building index %s is
stopped", i.cacher.Name())
@@ -215,11 +226,7 @@ func (i *Indexer) buildIndex() {
}
func (i *Indexer) compact() {
- l := len(i.prefixIndex)
- if l < DEFAULT_CACHE_INIT_SIZE {
- l = DEFAULT_CACHE_INIT_SIZE
- }
- n := make(map[string]map[string]struct{}, l)
+ n := make(map[string]map[string]struct{}, DEFAULT_CACHE_INIT_SIZE)
for k, v := range i.prefixIndex {
c, ok := n[k]
if !ok {
@@ -232,8 +239,8 @@ func (i *Indexer) compact() {
}
i.prefixIndex = n
- util.Logger().Infof("index %s(%s): compact root capacity to size %d",
- i.cacher.Name(), DEFAULT_COMPACT_TIMEOUT, l)
+ util.Logger().Infof("index %s: compact root capacity to size %d",
+ i.cacher.Name(), DEFAULT_CACHE_INIT_SIZE)
}
func (i *Indexer) getPrefixKey(arr *[]string, prefix string) (count int) {
@@ -301,7 +308,7 @@ func (i *Indexer) Run() {
i.isClose = false
i.prefixLock.Unlock()
- if _, ok := i.cacher.(*nullCacher); ok {
+ if !core.ServerInfo.Config.EnableCache {
util.SafeCloseChan(i.ready)
return
}
diff --git a/server/core/backend/store/lease.go b/server/core/backend/lease.go
similarity index 93%
rename from server/core/backend/store/lease.go
rename to server/core/backend/lease.go
index d6ea702..d52511c 100644
--- a/server/core/backend/store/lease.go
+++ b/server/core/backend/lease.go
@@ -14,12 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
errorsEx
"github.com/apache/incubator-servicecomb-service-center/pkg/errors"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"golang.org/x/net/context"
"time"
@@ -41,7 +40,7 @@ func (lat *LeaseTask) Key() string {
func (lat *LeaseTask) Do(ctx context.Context) (err error) {
lat.StartTime = time.Now()
- lat.TTL, err = backend.Registry().LeaseRenew(ctx, lat.LeaseID)
+ lat.TTL, err = Registry().LeaseRenew(ctx, lat.LeaseID)
lat.EndTime = time.Now()
if err != nil {
util.Logger().Errorf(err, "[%s]renew lease %d failed(rev: %s,
run: %s), key %s",
diff --git a/server/core/backend/store/listwatch.go
b/server/core/backend/listwatch.go
similarity index 97%
rename from server/core/backend/store/listwatch.go
rename to server/core/backend/listwatch.go
index 42fbde5..43f31e4 100644
--- a/server/core/backend/store/listwatch.go
+++ b/server/core/backend/listwatch.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"fmt"
@@ -71,7 +71,7 @@ func (lw *ListWatcher) doWatch(ctx context.Context, f
func(evt []KvEvent)) error
return fmt.Errorf("unknown event %s",
resp)
}
- util.Logger().Infof("watch prefix %s, start rev
%d+1, event: %s", lw.Prefix, rev, resp)
+ util.Logger().Infof("caught event %s, watch
prefix %s, start rev %d+1,", resp, lw.Prefix, rev)
lw.setRevision(resp.Revision)
diff --git a/server/core/backend/store/metric.go b/server/core/backend/metric.go
similarity index 93%
rename from server/core/backend/store/metric.go
rename to server/core/backend/metric.go
index 889e4dc..79958cc 100644
--- a/server/core/backend/store/metric.go
+++ b/server/core/backend/metric.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
@@ -43,6 +43,11 @@ func init() {
}
func ReportCacheMetrics(resource, t string, obj interface{}) {
+ if len(core.Instance.Endpoints) == 0 {
+ // endpoints list will be empty when initializing
+ return
+ }
+
once.Do(func() {
instance, _ = util.ParseEndpoint(core.Instance.Endpoints[0])
})
diff --git a/server/core/backend/store/store.go b/server/core/backend/store.go
similarity index 99%
rename from server/core/backend/store/store.go
rename to server/core/backend/store.go
index 6684946..cecc1d1 100644
--- a/server/core/backend/store/store.go
+++ b/server/core/backend/store.go
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package store
+package backend
import (
"github.com/apache/incubator-servicecomb-service-center/pkg/async"
diff --git a/server/core/info.go b/server/core/info.go
index 4a20dcc..1d52697 100644
--- a/server/core/info.go
+++ b/server/core/info.go
@@ -17,14 +17,8 @@
package core
import (
- "encoding/json"
- "github.com/apache/incubator-servicecomb-service-center/pkg/util"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
-
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
- "github.com/apache/incubator-servicecomb-service-center/version"
"github.com/astaxie/beego"
- "golang.org/x/net/context"
)
var ServerInfo *pb.ServerInformation = newInfo()
@@ -78,36 +72,3 @@ func newInfo() *pb.ServerInformation {
},
}
}
-
-func LoadServerInformation() error {
- resp, err := backend.Registry().Do(context.Background(),
- registry.GET, registry.WithStrKey(GetSystemKey()))
- if err != nil {
- return err
- }
- if len(resp.Kvs) == 0 {
- return nil
- }
-
- err = json.Unmarshal(resp.Kvs[0].Value, ServerInfo)
- if err != nil {
- util.Logger().Errorf(err, "load system config failed, maybe
incompatible")
- return nil
- }
- return nil
-}
-
-func UpgradeServerVersion() error {
- ServerInfo.Version = version.Ver().Version
-
- bytes, err := json.Marshal(ServerInfo)
- if err != nil {
- return err
- }
- _, err = backend.Registry().Do(context.Background(),
- registry.PUT, registry.WithStrKey(GetSystemKey()),
registry.WithValue(bytes))
- if err != nil {
- return err
- }
- return nil
-}
diff --git a/server/core/key_generator.go b/server/core/key_generator.go
index 67119f1..8f2f6d7 100644
--- a/server/core/key_generator.go
+++ b/server/core/key_generator.go
@@ -328,7 +328,7 @@ func GenerateDomainKey(domain string) string {
}, "/")
}
-func GetSystemKey() string {
+func GetServerInfoKey() string {
return util.StringJoin([]string{
GetRootKey(),
REGISTRY_SYS_KEY,
diff --git a/server/govern/service.go b/server/govern/service.go
index 7209948..bdaabc8 100644
--- a/server/govern/service.go
+++ b/server/govern/service.go
@@ -19,7 +19,7 @@ package govern
import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
scerr
"github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
@@ -187,7 +187,7 @@ func (governService *GovernService) GetApplications(ctx
context.Context, in *pb.
registry.WithPrefix(),
registry.WithKeyOnly())
- resp, err := store.Store().ServiceIndex().Search(ctx, opts...)
+ resp, err := backend.Store().ServiceIndex().Search(ctx, opts...)
if err != nil {
return nil, err
}
@@ -223,7 +223,7 @@ func getServiceAllVersions(ctx context.Context, serviceKey
*pb.MicroServiceKey)
registry.WithStrKey(key),
registry.WithPrefix())
- resp, err := store.Store().ServiceIndex().Search(ctx, opts...)
+ resp, err := backend.Store().ServiceIndex().Search(ctx, opts...)
if err != nil {
return nil, err
}
@@ -244,7 +244,7 @@ func getSchemaInfoUtil(ctx context.Context, domainProject
string, serviceId stri
registry.WithStrKey(key),
registry.WithPrefix())
- resp, err := store.Store().Schema().Search(ctx, opts...)
+ resp, err := backend.Store().Schema().Search(ctx, opts...)
if err != nil {
util.Logger().Errorf(err, "Get schema failed")
return make([]*pb.Schema, 0), err
@@ -355,7 +355,7 @@ func statistics(ctx context.Context) (*pb.Statistics,
error) {
svcOpts := append(opts,
registry.WithStrKey(key),
registry.WithPrefix())
- respSvc, err := store.Store().ServiceIndex().Search(ctx, svcOpts...)
+ respSvc, err := backend.Store().ServiceIndex().Search(ctx, svcOpts...)
if err != nil {
return nil, err
}
@@ -391,7 +391,7 @@ func statistics(ctx context.Context) (*pb.Statistics,
error) {
registry.WithStrKey(key),
registry.WithPrefix(),
registry.WithKeyOnly())
- respIns, err := store.Store().Instance().Search(ctx, instOpts...)
+ respIns, err := backend.Store().Instance().Search(ctx, instOpts...)
if err != nil {
return nil, err
}
@@ -432,7 +432,7 @@ func getInstanceCountByDomain(ctx context.Context, resp
chan GetInstanceCountByD
registry.WithPrefix(),
registry.WithKeyOnly(),
registry.WithCountOnly())
- respIns, err := store.Store().Instance().Search(ctx, instOpts...)
+ respIns, err := backend.Store().Instance().Search(ctx, instOpts...)
if err != nil {
util.Logger().Errorf(err, "get instance count under same
domainId %s", domainId)
}
diff --git a/server/handler/cache/cache.go b/server/handler/cache/cache.go
index f6674f3..8b11992 100644
--- a/server/handler/cache/cache.go
+++ b/server/handler/cache/cache.go
@@ -20,7 +20,7 @@ import (
"fmt"
"github.com/apache/incubator-servicecomb-service-center/pkg/chain"
"github.com/apache/incubator-servicecomb-service-center/pkg/rest"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
"net/http"
"strconv"
)
@@ -32,7 +32,7 @@ func (l *CacheResponse) Handle(i *chain.Invocation) {
r := i.Context().Value(rest.CTX_REQUEST).(*http.Request)
w := i.Context().Value(rest.CTX_RESPONSE).(http.ResponseWriter)
- scRev := store.Revision()
+ scRev := backend.Revision()
w.Header().Set("X-Resource-Revision", fmt.Sprint(scRev))
rev, _ := strconv.ParseInt(r.URL.Query().Get("rev"), 10, 64)
diff --git a/server/plugin/infra/quota/buildin/buildin.go
b/server/plugin/infra/quota/buildin/buildin.go
index d708291..0f947a1 100644
--- a/server/plugin/infra/quota/buildin/buildin.go
+++ b/server/plugin/infra/quota/buildin/buildin.go
@@ -20,7 +20,7 @@ import (
"fmt"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
"github.com/apache/incubator-servicecomb-service-center/server/core"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
scerr
"github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/quota"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
@@ -85,7 +85,7 @@ func (q *BuildInQuota) RemandQuotas(ctx context.Context,
quotaType quota.Resourc
func ResourceLimitHandler(ctx context.Context, res *quota.ApplyQuotaResource)
*quota.ApplyQuotaResult {
var key string
var max int64 = 0
- var indexer *store.Indexer
+ var indexer *backend.Indexer
domainProject := res.DomainProject
serviceId := res.ServiceId
@@ -93,11 +93,11 @@ func ResourceLimitHandler(ctx context.Context, res
*quota.ApplyQuotaResource) *q
case quota.RuleQuotaType:
key = core.GenerateServiceRuleKey(domainProject, serviceId, "")
max = RULE_NUM_MAX_LIMIT_PER_SERVICE
- indexer = store.Store().Rule()
+ indexer = backend.Store().Rule()
case quota.SchemaQuotaType:
key = core.GenerateServiceSchemaKey(domainProject, serviceId,
"")
max = SCHEMA_NUM_MAX_LIMIT_PER_SERVICE
- indexer = store.Store().Schema()
+ indexer = backend.Store().Schema()
case quota.TagQuotaType:
applyNum := res.QuotaSize
max = TAG_NUM_MAX_LIMIT_PER_SERVICE
@@ -193,7 +193,7 @@ func getInstanceMaxLimit() int64 {
}
func getInstancesNum(ctx context.Context, key string) (int64, error) {
- resp, err := store.Store().Instance().Search(ctx,
+ resp, err := backend.Store().Instance().Search(ctx,
registry.WithStrKey(key),
registry.WithPrefix(),
registry.WithCountOnly())
@@ -229,7 +229,7 @@ func getServiceMaxLimit() int64 {
}
func getServicesNum(ctx context.Context, key string) (int64, error) {
- resp, err := store.Store().Service().Search(ctx,
+ resp, err := backend.Store().Service().Search(ctx,
registry.WithStrKey(key),
registry.WithPrefix(),
registry.WithCountOnly())
diff --git a/server/plugin/infra/registry/embededetcd/embededetcd.go
b/server/plugin/infra/registry/embededetcd/embededetcd.go
index 3bb7140..e9be1f3 100644
--- a/server/plugin/infra/registry/embededetcd/embededetcd.go
+++ b/server/plugin/infra/registry/embededetcd/embededetcd.go
@@ -41,7 +41,7 @@ import (
var embedTLSConfig *tls.Config
-const START_MANAGER_SERVER_TIMEOUT = 10
+const START_MANAGER_SERVER_TIMEOUT = 10 * time.Second
func init() {
mgr.RegisterPlugin(mgr.Plugin{mgr.REGISTRY, "embeded_etcd",
getEmbedInstance})
@@ -458,7 +458,7 @@ func (s *EtcdEmbed) Watch(ctx context.Context, opts
...registry.PluginOpOption)
}
func (s *EtcdEmbed) ReadyNotify() {
- timeout := START_MANAGER_SERVER_TIMEOUT * time.Second
+ timeout := START_MANAGER_SERVER_TIMEOUT
select {
case <-s.Embed.Server.ReadyNotify():
close(s.ready)
diff --git a/server/plugin/infra/registry/etcd/logger.go
b/server/plugin/infra/registry/etcd/logger.go
index 952834b..c0e35ba 100644
--- a/server/plugin/infra/registry/etcd/logger.go
+++ b/server/plugin/infra/registry/etcd/logger.go
@@ -21,7 +21,6 @@ import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
"github.com/coreos/pkg/capnslog"
"runtime"
- "strings"
)
// clientLogger implement from grcplog.LoggerV2s and capnslog.Formatter
@@ -45,13 +44,7 @@ func (l *clientLogger) getCaller(depth int) string {
if !ok {
return "???"
}
-
- if sp1 := strings.LastIndex(file, "/"); sp1 >= 0 {
- if sp2 := strings.LastIndex(file[:sp1], "/"); sp2 >= 0 {
- file = file[sp2+1:]
- }
- }
- return fmt.Sprintf("%s:%d", file, line)
+ return fmt.Sprintf("%s:%d", util.FileLastName(file), line)
}
func (l *clientLogger) Flush() {
diff --git a/server/plugin/infra/tracing/buildin/common.go
b/server/plugin/infra/tracing/buildin/common.go
index 1653537..ad9fff7 100644
--- a/server/plugin/infra/tracing/buildin/common.go
+++ b/server/plugin/infra/tracing/buildin/common.go
@@ -30,7 +30,7 @@ import (
func initTracer() {
collector, err := newCollector()
if err != nil {
- util.Logger().Errorf(err, "new tracing collector failed, use
the noop tracer")
+ util.Logger().Warnf(err, "new tracing collector failed, use the
noop tracer")
return
}
ipPort, _ := util.ParseEndpoint(core.Instance.Endpoints[0])
diff --git a/server/server.go b/server/server.go
index 611b83c..7e98b6a 100644
--- a/server/server.go
+++ b/server/server.go
@@ -18,11 +18,12 @@ package server
import _
"github.com/apache/incubator-servicecomb-service-center/server/service/event"
import (
+ "encoding/json"
"fmt"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
"github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
- st
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"github.com/apache/incubator-servicecomb-service-center/server/mux"
nf
"github.com/apache/incubator-servicecomb-service-center/server/service/notification"
serviceUtil
"github.com/apache/incubator-servicecomb-service-center/server/service/util"
@@ -42,7 +43,7 @@ var (
func init() {
server = &ServiceCenterServer{
- store: st.Store(),
+ store: backend.Store(),
notifyService: nf.GetNotifyService(),
apiServer: GetAPIServer(),
goroutine: util.NewGo(context.Background()),
@@ -52,7 +53,7 @@ func init() {
type ServiceCenterServer struct {
apiServer *APIServer
notifyService *nf.NotifyService
- store *st.KvStore
+ store *backend.KvStore
goroutine *util.GoRoutine
}
@@ -81,9 +82,42 @@ func (s *ServiceCenterServer) waitForQuit() {
util.Logger().Debugf("service center stopped")
}
+func LoadServerInformation() error {
+ resp, err := backend.Registry().Do(context.Background(),
+ registry.GET, registry.WithStrKey(core.GetServerInfoKey()))
+ if err != nil {
+ return err
+ }
+ if len(resp.Kvs) == 0 {
+ return nil
+ }
+
+ err = json.Unmarshal(resp.Kvs[0].Value, core.ServerInfo)
+ if err != nil {
+ util.Logger().Errorf(err, "load system config failed, maybe
incompatible")
+ return nil
+ }
+ return nil
+}
+
+func UpgradeServerVersion() error {
+ core.ServerInfo.Version = version.Ver().Version
+
+ bytes, err := json.Marshal(core.ServerInfo)
+ if err != nil {
+ return err
+ }
+ _, err = backend.Registry().Do(context.Background(),
+ registry.PUT, registry.WithStrKey(core.GetServerInfoKey()),
registry.WithValue(bytes))
+ if err != nil {
+ return err
+ }
+ return nil
+}
+
func (s *ServiceCenterServer) needUpgrade() bool {
if core.ServerInfo.Version == "0" {
- err := core.LoadServerInformation()
+ err := LoadServerInformation()
if err != nil {
util.Logger().Errorf(err, "check version failed, can
not load the system config")
return false
@@ -102,7 +136,7 @@ func (s *ServiceCenterServer) initialize() {
os.Exit(1)
}
if s.needUpgrade() {
- core.UpgradeServerVersion()
+ UpgradeServerVersion()
}
// cache mechanism
diff --git a/server/service/event/dependency_event_handler.go
b/server/service/event/dependency_event_handler.go
index 836008b..6667720 100644
--- a/server/service/event/dependency_event_handler.go
+++ b/server/service/event/dependency_event_handler.go
@@ -22,7 +22,6 @@ import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
"github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"github.com/apache/incubator-servicecomb-service-center/server/mux"
@@ -36,11 +35,11 @@ type DependencyEventHandler struct {
signals *util.UniQueue
}
-func (h *DependencyEventHandler) Type() store.StoreType {
- return store.DEPENDENCY_QUEUE
+func (h *DependencyEventHandler) Type() backend.StoreType {
+ return backend.DEPENDENCY_QUEUE
}
-func (h *DependencyEventHandler) OnEvent(evt store.KvEvent) {
+func (h *DependencyEventHandler) OnEvent(evt backend.KvEvent) {
action := evt.Type
if action != pb.EVT_CREATE && action != pb.EVT_UPDATE && action !=
pb.EVT_INIT {
return
@@ -51,14 +50,10 @@ func (h *DependencyEventHandler) OnEvent(evt store.KvEvent)
{
func (h *DependencyEventHandler) loop() {
util.Go(func(ctx context.Context) {
- waitDelayIndex := 0
- waitDelay := []int{1, 1, 5, 10, 20, 30, 60}
- retry := func() {
- if waitDelayIndex >= len(waitDelay) {
- waitDelayIndex = 0
- }
- <-time.After(time.Duration(waitDelay[waitDelayIndex]) *
time.Second)
- waitDelayIndex++
+ retries := 0
+ delay := func() {
+ <-time.After(util.GetBackoff().Delay(retries))
+ retries++
h.signals.Put(context.Background(), struct{}{})
}
@@ -70,11 +65,12 @@ func (h *DependencyEventHandler) loop() {
lock, err := mux.Try(mux.DEP_QUEUE_LOCK)
if err != nil {
util.Logger().Errorf(err, "try to lock
%s failed", mux.DEP_QUEUE_LOCK)
- retry()
+ delay()
continue
}
if lock == nil {
+ retries = 0
continue
}
@@ -82,9 +78,11 @@ func (h *DependencyEventHandler) loop() {
lock.Unlock()
if err != nil {
util.Logger().Errorf(err, "handle
dependency event failed")
- retry()
+ delay()
continue
}
+
+ retries = 0
}
}
})
@@ -115,7 +113,7 @@ func isAddToLeft(centerNode *util.Node, addRes interface{})
bool {
func (h *DependencyEventHandler) Handle() error {
key := core.GetServiceDependencyQueueRootKey("")
- resp, err :=
store.Store().DependencyQueue().Search(context.Background(),
+ resp, err :=
backend.Store().DependencyQueue().Search(context.Background(),
registry.WithStrKey(key),
registry.WithPrefix())
if err != nil {
diff --git a/server/service/event/event.go b/server/service/event/event.go
index 53fe882..271d50b 100644
--- a/server/service/event/event.go
+++ b/server/service/event/event.go
@@ -17,13 +17,13 @@
package event
import (
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
)
func init() {
- store.AddEventHandler(NewServiceEventHandler())
- store.AddEventHandler(NewInstanceEventHandler())
- store.AddEventHandler(NewRuleEventHandler())
- store.AddEventHandler(NewTagEventHandler())
- store.AddEventHandler(NewDependencyEventHandler())
+ backend.AddEventHandler(NewServiceEventHandler())
+ backend.AddEventHandler(NewInstanceEventHandler())
+ backend.AddEventHandler(NewRuleEventHandler())
+ backend.AddEventHandler(NewTagEventHandler())
+ backend.AddEventHandler(NewDependencyEventHandler())
}
diff --git a/server/service/event/instance_event_handler.go
b/server/service/event/instance_event_handler.go
index e9ee8b7..e52e457 100644
--- a/server/service/event/instance_event_handler.go
+++ b/server/service/event/instance_event_handler.go
@@ -20,7 +20,7 @@ import (
"encoding/json"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
nf
"github.com/apache/incubator-servicecomb-service-center/server/service/notification"
serviceUtil
"github.com/apache/incubator-servicecomb-service-center/server/service/util"
@@ -32,11 +32,11 @@ import (
type InstanceEventHandler struct {
}
-func (h *InstanceEventHandler) Type() store.StoreType {
- return store.INSTANCE
+func (h *InstanceEventHandler) Type() backend.StoreType {
+ return backend.INSTANCE
}
-func (h *InstanceEventHandler) OnEvent(evt store.KvEvent) {
+func (h *InstanceEventHandler) OnEvent(evt backend.KvEvent) {
action := evt.Type
if action == pb.EVT_INIT {
return
diff --git a/server/service/event/rule_event_handler.go
b/server/service/event/rule_event_handler.go
index 07cc316..42e131f 100644
--- a/server/service/event/rule_event_handler.go
+++ b/server/service/event/rule_event_handler.go
@@ -21,7 +21,7 @@ import (
"fmt"
"github.com/apache/incubator-servicecomb-service-center/pkg/async"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
nf
"github.com/apache/incubator-servicecomb-service-center/server/service/notification"
serviceUtil
"github.com/apache/incubator-servicecomb-service-center/server/service/util"
@@ -77,11 +77,11 @@ func (apt *RulesChangedTask) publish(ctx context.Context,
domainProject, provide
type RuleEventHandler struct {
}
-func (h *RuleEventHandler) Type() store.StoreType {
- return store.RULE
+func (h *RuleEventHandler) Type() backend.StoreType {
+ return backend.RULE
}
-func (h *RuleEventHandler) OnEvent(evt store.KvEvent) {
+func (h *RuleEventHandler) OnEvent(evt backend.KvEvent) {
action := evt.Type
if action == pb.EVT_INIT {
return
diff --git a/server/service/event/service_event_handler.go
b/server/service/event/service_event_handler.go
index 8ffd805..45392f1 100644
--- a/server/service/event/service_event_handler.go
+++ b/server/service/event/service_event_handler.go
@@ -18,7 +18,7 @@ package event
import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
serviceUtil
"github.com/apache/incubator-servicecomb-service-center/server/service/util"
"github.com/coreos/etcd/mvcc/mvccpb"
@@ -29,11 +29,11 @@ import (
type ServiceEventHandler struct {
}
-func (h *ServiceEventHandler) Type() store.StoreType {
- return store.SERVICE
+func (h *ServiceEventHandler) Type() backend.StoreType {
+ return backend.SERVICE
}
-func (h *ServiceEventHandler) OnEvent(evt store.KvEvent) {
+func (h *ServiceEventHandler) OnEvent(evt backend.KvEvent) {
action := evt.Type
if action != pb.EVT_CREATE && action != pb.EVT_INIT {
return
diff --git a/server/service/event/tag_event_handler.go
b/server/service/event/tag_event_handler.go
index cf2afa7..ef2968a 100644
--- a/server/service/event/tag_event_handler.go
+++ b/server/service/event/tag_event_handler.go
@@ -21,7 +21,7 @@ import (
"fmt"
"github.com/apache/incubator-servicecomb-service-center/pkg/async"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
nf
"github.com/apache/incubator-servicecomb-service-center/server/service/notification"
serviceUtil
"github.com/apache/incubator-servicecomb-service-center/server/service/util"
@@ -88,11 +88,11 @@ func (apt *TagsChangedTask) publish(ctx context.Context,
domainProject, consumer
type TagEventHandler struct {
}
-func (h *TagEventHandler) Type() store.StoreType {
- return store.SERVICE_TAG
+func (h *TagEventHandler) Type() backend.StoreType {
+ return backend.SERVICE_TAG
}
-func (h *TagEventHandler) OnEvent(evt store.KvEvent) {
+func (h *TagEventHandler) OnEvent(evt backend.KvEvent) {
action := evt.Type
if action == pb.EVT_INIT {
return
diff --git a/server/service/instances.go b/server/service/instances.go
index ffb4076..33d64e2 100644
--- a/server/service/instances.go
+++ b/server/service/instances.go
@@ -180,15 +180,12 @@ func (s *InstanceService) Register(ctx context.Context,
in *pb.RegisterInstanceR
// build the request options
key := apt.GenerateInstanceKey(domainProject, instance.ServiceId,
instanceId)
hbKey := apt.GenerateInstanceLeaseKey(domainProject,
instance.ServiceId, instanceId)
- epKey :=
util.StringToBytesWithNoCopy(apt.GenerateEndpointsIndexKey(domainProject,
instance))
opts := []registry.PluginOp{
registry.OpPut(registry.WithStrKey(key),
registry.WithValue(data),
registry.WithLease(leaseID)),
registry.OpPut(registry.WithStrKey(hbKey),
registry.WithStrValue(fmt.Sprintf("%d", leaseID)),
registry.WithLease(leaseID)),
- registry.OpPut(registry.WithKey(epKey),
registry.WithStrValue(instance.ServiceId+"/"+instanceId),
- registry.WithLease(leaseID)),
}
_, err = backend.Registry().Txn(ctx, opts)
diff --git a/server/service/instances_test.go b/server/service/instances_test.go
index 9c91841..e4723d7 100644
--- a/server/service/instances_test.go
+++ b/server/service/instances_test.go
@@ -115,15 +115,6 @@ var _ = Describe("'Instance' service", func() {
})
Expect(err).To(BeNil())
Expect(resp.Response.Code).To(Equal(pb.Response_SUCCESS))
-
- instance.InstanceId = ""
- instance.ServiceId = serviceId2
- resp, err =
instanceResource.Register(getContext(), &pb.RegisterInstanceRequest{
- Instance: instance,
- })
- Expect(err).To(BeNil())
-
Expect(resp.Response.Code).To(Equal(scerr.ErrEndpointAlreadyExists))
-
})
})
diff --git a/server/service/microservices.go b/server/service/microservices.go
index 6663ab9..2b83b7b 100644
--- a/server/service/microservices.go
+++ b/server/service/microservices.go
@@ -24,7 +24,6 @@ import (
"github.com/apache/incubator-servicecomb-service-center/server/core"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
scerr
"github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/quota"
@@ -237,7 +236,7 @@ func (s *MicroServiceService) DeleteServicePri(ctx
context.Context, serviceId st
}
instancesKey := apt.GenerateInstanceKey(domainProject,
serviceId, "")
- rsp, err := store.Store().Instance().Search(ctx,
+ rsp, err := backend.Store().Instance().Search(ctx,
registry.WithStrKey(instancesKey),
registry.WithPrefix(),
registry.WithCountOnly())
diff --git a/server/service/schema.go b/server/service/schema.go
index dd9c4d9..df66d15 100644
--- a/server/service/schema.go
+++ b/server/service/schema.go
@@ -21,7 +21,6 @@ import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
scerr
"github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/quota"
@@ -59,7 +58,7 @@ func (s *MicroServiceService) GetSchemaInfo(ctx
context.Context, in *pb.GetSchem
key := apt.GenerateServiceSchemaKey(domainProject, in.ServiceId,
in.SchemaId)
opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(key))
- resp, errDo := store.Store().Schema().Search(ctx, opts...)
+ resp, errDo := backend.Store().Schema().Search(ctx, opts...)
if errDo != nil {
util.Logger().Errorf(errDo, "get schema failed, serviceId %s,
schemaId %s: get schema info failed.", in.ServiceId, in.SchemaId)
return &pb.GetSchemaResponse{
@@ -130,7 +129,7 @@ func (s *MicroServiceService) GetAllSchemaInfo(ctx
context.Context, in *pb.GetAl
key := apt.GenerateServiceSchemaSummaryKey(domainProject, in.ServiceId,
"")
opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(key),
registry.WithPrefix())
- resp, errDo := store.Store().SchemaSummary().Search(ctx, opts...)
+ resp, errDo := backend.Store().SchemaSummary().Search(ctx, opts...)
if errDo != nil {
util.Logger().Errorf(errDo, "get schema failed, serviceId %s:
get schema info failed.", in.ServiceId)
return &pb.GetAllSchemaResponse{
@@ -142,7 +141,7 @@ func (s *MicroServiceService) GetAllSchemaInfo(ctx
context.Context, in *pb.GetAl
if in.WithSchema {
key := apt.GenerateServiceSchemaKey(domainProject,
in.ServiceId, "")
opts := append(serviceUtil.FromContext(ctx),
registry.WithStrKey(key), registry.WithPrefix())
- respWithSchema, errDo = store.Store().Schema().Search(ctx,
opts...)
+ respWithSchema, errDo = backend.Store().Schema().Search(ctx,
opts...)
if errDo != nil {
util.Logger().Errorf(errDo, "get schema failed,
serviceId %s: get schema info failed.", in.ServiceId)
return &pb.GetAllSchemaResponse{
@@ -565,7 +564,7 @@ func (s *MicroServiceService) modifySchema(ctx
context.Context, serviceId string
}
key := apt.GenerateServiceSchemaKey(domainProject, serviceId,
schemaId)
- respSchema, err := store.Store().Schema().Search(ctx,
registry.WithStrKey(key), registry.WithCountOnly())
+ respSchema, err := backend.Store().Schema().Search(ctx,
registry.WithStrKey(key), registry.WithCountOnly())
if err != nil {
util.Logger().Errorf(err, "modify schema failed, get
schema summary failed, %s %s", serviceId, schemaId)
return scerr.NewError(scerr.ErrInternal, "get schema
summary failed")
@@ -622,7 +621,7 @@ func (s *MicroServiceService) modifySchema(ctx
context.Context, serviceId string
func isExistSchemaSummary(ctx context.Context, domainProject, serviceId,
schemaId string) (bool, error) {
key := apt.GenerateServiceSchemaSummaryKey(domainProject, serviceId,
schemaId)
- resp, err := store.Store().SchemaSummary().Search(ctx,
registry.WithStrKey(key), registry.WithCountOnly())
+ resp, err := backend.Store().SchemaSummary().Search(ctx,
registry.WithStrKey(key), registry.WithCountOnly())
if err != nil {
return true, err
}
@@ -656,7 +655,7 @@ func containsValueInSlice(in []string, value string) bool {
func getSchemaSummary(ctx context.Context, domainProject string, serviceId
string, schemaId string) (string, error) {
key := apt.GenerateServiceSchemaSummaryKey(domainProject, serviceId,
schemaId)
- resp, err := store.Store().SchemaSummary().Search(ctx,
+ resp, err := backend.Store().SchemaSummary().Search(ctx,
registry.WithStrKey(key),
)
if err != nil {
diff --git a/server/service/util/dependency.go
b/server/service/util/dependency.go
index 0803901..330473a 100644
--- a/server/service/util/dependency.go
+++ b/server/service/util/dependency.go
@@ -23,7 +23,6 @@ import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
scerr
"github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
@@ -219,7 +218,7 @@ func TransferToMicroServiceDependency(ctx context.Context,
key string) (*pb.Micr
}
opts := append(FromContext(ctx), registry.WithStrKey(key))
- res, err := store.Store().DependencyRule().Search(ctx, opts...)
+ res, err := backend.Store().DependencyRule().Search(ctx, opts...)
if err != nil {
util.Logger().Errorf(nil, "Get dependency rule failed.")
return nil, err
@@ -802,7 +801,7 @@ func (dr *DependencyRelation)
parseDependencyRule(dependencyRule *pb.MicroServic
sopts := append(opts,
registry.WithStrKey(allServiceKey),
registry.WithPrefix())
- resp, err := store.Store().Service().Search(dr.ctx, sopts...)
+ resp, err := backend.Store().Service().Search(dr.ctx, sopts...)
if err != nil {
return nil, err
}
@@ -908,7 +907,7 @@ func (dr *DependencyRelation)
getConsumerOfDependAllServices() ([]*pb.MicroServi
providerService.ServiceName = "*"
relyAllKey := apt.GenerateProviderDependencyRuleKey(dr.domainProject,
providerService)
opts := append(FromContext(dr.ctx), registry.WithStrKey(relyAllKey))
- rsp, err := store.Store().DependencyRule().Search(dr.ctx, opts...)
+ rsp, err := backend.Store().DependencyRule().Search(dr.ctx, opts...)
if err != nil {
util.Logger().Errorf(err, "get consumer that rely all service
failed.")
return nil, err
@@ -934,7 +933,7 @@ func (dr *DependencyRelation)
getConsumerOfSameServiceNameAndAppId(provider *pb.
opts := append(FromContext(dr.ctx),
registry.WithStrKey(prefix),
registry.WithPrefix())
- rsp, err := store.Store().DependencyRule().Search(dr.ctx, opts...)
+ rsp, err := backend.Store().DependencyRule().Search(dr.ctx, opts...)
if err != nil {
util.Logger().Errorf(err, "get all dependency rule failed:
provider rule key %v.", provider)
return nil, err
diff --git a/server/service/util/domain_util.go
b/server/service/util/domain_util.go
index caadc78..acd49a5 100644
--- a/server/service/util/domain_util.go
+++ b/server/service/util/domain_util.go
@@ -20,7 +20,6 @@ import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"github.com/coreos/etcd/mvcc/mvccpb"
"golang.org/x/net/context"
@@ -31,7 +30,7 @@ func GetAllDomainRawData(ctx context.Context)
([]*mvccpb.KeyValue, error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateDomainKey("")),
registry.WithPrefix())
- rsp, err := store.Store().Domain().Search(ctx, opts...)
+ rsp, err := backend.Store().Domain().Search(ctx, opts...)
if err != nil {
return nil, err
}
@@ -66,7 +65,7 @@ func DomainExist(ctx context.Context, domain string) (bool,
error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateDomainKey(domain)),
registry.WithCountOnly())
- rsp, err := store.Store().Domain().Search(ctx, opts...)
+ rsp, err := backend.Store().Domain().Search(ctx, opts...)
if err != nil {
return false, err
}
@@ -77,7 +76,7 @@ func ProjectExist(ctx context.Context, domain, project
string) (bool, error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateProjectKey(domain, project)),
registry.WithCountOnly())
- rsp, err := store.Store().Project().Search(ctx, opts...)
+ rsp, err := backend.Store().Project().Search(ctx, opts...)
if err != nil {
return false, err
}
diff --git a/server/service/util/heartbeat_util.go
b/server/service/util/heartbeat_util.go
index 6ea8328..7f8a675 100644
--- a/server/service/util/heartbeat_util.go
+++ b/server/service/util/heartbeat_util.go
@@ -19,7 +19,7 @@ package util
import (
"errors"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"golang.org/x/net/context"
)
@@ -37,7 +37,7 @@ func KeepAliveLease(ctx context.Context, domainProject,
serviceId, instanceId st
if leaseID == -1 {
return ttl, errors.New("leaseId not exist, instance not exist.")
}
- ttl, err = store.Store().KeepAlive(ctx,
+ ttl, err = backend.Store().KeepAlive(ctx,
registry.WithStrKey(apt.GenerateInstanceLeaseKey(domainProject,
serviceId, instanceId)),
registry.WithLease(leaseID))
if err != nil {
diff --git a/server/service/util/instance_util.go
b/server/service/util/instance_util.go
index a6a0b8b..4e7b8cf 100644
--- a/server/service/util/instance_util.go
+++ b/server/service/util/instance_util.go
@@ -18,11 +18,9 @@ package util
import (
"encoding/json"
- "fmt"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
scerr
"github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
@@ -36,7 +34,7 @@ import (
func GetLeaseId(ctx context.Context, domainProject string, serviceId string,
instanceId string) (int64, error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateInstanceLeaseKey(domainProject,
serviceId, instanceId)))
- resp, err := store.Store().Lease().Search(ctx, opts...)
+ resp, err := backend.Store().Lease().Search(ctx, opts...)
if err != nil {
return -1, err
}
@@ -51,7 +49,7 @@ func GetInstance(ctx context.Context, domainProject string,
serviceId string, in
key := apt.GenerateInstanceKey(domainProject, serviceId, instanceId)
opts := append(FromContext(ctx), registry.WithStrKey(key))
- resp, err := store.Store().Instance().Search(ctx, opts...)
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
if err != nil {
return nil, err
}
@@ -70,7 +68,7 @@ func GetInstance(ctx context.Context, domainProject string,
serviceId string, in
func GetAllInstancesOfOneService(ctx context.Context, domainProject string,
serviceId string) ([]*pb.MicroServiceInstance, error) {
key := apt.GenerateInstanceKey(domainProject, serviceId, "")
opts := append(FromContext(ctx), registry.WithStrKey(key),
registry.WithPrefix())
- resp, err := store.Store().Instance().Search(ctx, opts...)
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
if err != nil {
util.Logger().Errorf(err, "Get instance of service %s from etcd
failed.", serviceId)
return nil, err
@@ -95,7 +93,7 @@ func GetInstanceCountOfOneService(ctx context.Context,
domainProject string, ser
registry.WithStrKey(key),
registry.WithPrefix(),
registry.WithCountOnly())
- resp, err := store.Store().Instance().Search(ctx, opts...)
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
if err != nil {
util.Logger().Errorf(err, "Get instance count of service %s
from etcd failed.", serviceId)
return 0, err
@@ -107,7 +105,7 @@ func InstanceExistById(ctx context.Context, domainProject
string, serviceId stri
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateInstanceKey(domainProject,
serviceId, instanceId)),
registry.WithCountOnly())
- resp, err := store.Store().Instance().Search(ctx, opts...)
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
if err != nil {
return false, err
}
@@ -129,23 +127,7 @@ func InstanceExist(ctx context.Context, instance
*pb.MicroServiceInstance) (stri
return instance.InstanceId, nil
}
}
-
- // check endpoint index
- resp, err := store.Store().Endpoints().Search(ctx,
-
registry.WithStrKey(apt.GenerateEndpointsIndexKey(domainProject, instance)))
- if err != nil {
- return "", scerr.NewError(scerr.ErrInternal, err.Error())
- }
- if resp.Count == 0 {
- return "", nil
- }
- endpointValue := ParseEndpointIndexValue(resp.Kvs[0].Value)
- if instance.ServiceId != endpointValue.serviceId {
- return endpointValue.instanceId,
- scerr.NewError(scerr.ErrEndpointAlreadyExists,
- fmt.Sprintf("Find the same endpoints in service
%s", endpointValue.serviceId))
- }
- return endpointValue.instanceId, nil
+ return "", nil
}
type EndpointIndexValue struct {
@@ -166,7 +148,7 @@ func DeleteServiceAllInstances(ctx context.Context,
serviceId string) error {
domainProject := util.ParseDomainProject(ctx)
instanceLeaseKey := apt.GenerateInstanceLeaseKey(domainProject,
serviceId, "")
- resp, err := store.Store().Lease().Search(ctx,
+ resp, err := backend.Store().Lease().Search(ctx,
registry.WithStrKey(instanceLeaseKey),
registry.WithPrefix(),
registry.WithNoCache())
@@ -205,7 +187,7 @@ func QueryAllProvidersInstances(ctx context.Context,
selfServiceId string) (resu
return
}
- rev = store.Revision()
+ rev = backend.Revision()
for _, providerId := range providerIds {
service, err := GetServiceWithRev(ctx, domainProject,
providerId, rev)
@@ -254,7 +236,7 @@ func QueryAllProvidersInstances(ctx context.Context,
selfServiceId string) (resu
func queryServiceInstancesKvs(ctx context.Context, serviceId string, rev
int64) ([]*mvccpb.KeyValue, error) {
domainProject := util.ParseDomainProject(ctx)
key := apt.GenerateInstanceKey(domainProject, serviceId, "")
- resp, err := store.Store().Instance().Search(ctx,
+ resp, err := backend.Store().Instance().Search(ctx,
registry.WithStrKey(key),
registry.WithPrefix(),
registry.WithRev(rev))
diff --git a/server/service/util/instance_util_test.go
b/server/service/util/instance_util_test.go
index 3e30ea6..fa27550 100644
--- a/server/service/util/instance_util_test.go
+++ b/server/service/util/instance_util_test.go
@@ -81,7 +81,7 @@ func TestInstanceExist(t *testing.T) {
_, err := InstanceExist(context.Background(),
&proto.MicroServiceInstance{
ServiceId: "a",
})
- if err == nil {
+ if err != nil {
t.Fatalf(`InstanceExist endpoint failed`)
}
_, err = InstanceExist(context.Background(),
&proto.MicroServiceInstance{
diff --git a/server/service/util/microservice_util.go
b/server/service/util/microservice_util.go
index c740668..e8f8e7d 100644
--- a/server/service/util/microservice_util.go
+++ b/server/service/util/microservice_util.go
@@ -20,7 +20,7 @@ import (
"encoding/json"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
"github.com/apache/incubator-servicecomb-service-center/server/infra/quota"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
@@ -34,7 +34,7 @@ import (
*/
func GetServiceWithRev(ctx context.Context, domain string, id string, rev
int64) (*pb.MicroService, error) {
key := apt.GenerateServiceKey(domain, id)
- serviceResp, err := store.Store().Service().Search(ctx,
+ serviceResp, err := backend.Store().Service().Search(ctx,
registry.WithStrKey(key),
registry.WithRev(rev))
if err != nil {
@@ -58,7 +58,7 @@ func GetServiceInCache(ctx context.Context, domain string, id
string) (*pb.Micro
func GetService(ctx context.Context, domainProject string, serviceId string)
(*pb.MicroService, error) {
key := apt.GenerateServiceKey(domainProject, serviceId)
opts := append(FromContext(ctx), registry.WithStrKey(key))
- serviceResp, err := store.Store().Service().Search(ctx, opts...)
+ serviceResp, err := backend.Store().Service().Search(ctx, opts...)
if err != nil {
return nil, err
}
@@ -78,7 +78,7 @@ func GetServicesRawData(ctx context.Context, domainProject
string) ([]*mvccpb.Ke
opts := append(FromContext(ctx),
registry.WithStrKey(key),
registry.WithPrefix())
- resp, err := store.Store().Service().Search(ctx, opts...)
+ resp, err := backend.Store().Service().Search(ctx, opts...)
if err != nil {
return nil, err
}
@@ -118,7 +118,7 @@ func GetServiceId(ctx context.Context, key
*pb.MicroServiceKey) (serviceId strin
func searchServiceId(ctx context.Context, key *pb.MicroServiceKey) (string,
error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateServiceIndexKey(key)))
- resp, err := store.Store().ServiceIndex().Search(ctx, opts...)
+ resp, err := backend.Store().ServiceIndex().Search(ctx, opts...)
if err != nil {
return "", err
}
@@ -130,7 +130,7 @@ func searchServiceId(ctx context.Context, key
*pb.MicroServiceKey) (string, erro
func searchServiceIdFromAlias(ctx context.Context, key *pb.MicroServiceKey)
(string, error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateServiceAliasKey(key)))
- resp, err := store.Store().ServiceAlias().Search(ctx, opts...)
+ resp, err := backend.Store().ServiceAlias().Search(ctx, opts...)
if err != nil {
return "", err
}
@@ -152,7 +152,7 @@ func GetServiceAllVersions(ctx context.Context, key
*pb.MicroServiceKey, alias b
registry.WithStrKey(prefix),
registry.WithPrefix(),
registry.WithDescendOrder())
- resp, err := store.Store().ServiceIndex().Search(ctx, opts...)
+ resp, err := backend.Store().ServiceIndex().Search(ctx, opts...)
return resp, err
}
@@ -195,7 +195,7 @@ func ServiceExist(ctx context.Context, domainProject
string, serviceId string) b
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateServiceKey(domainProject,
serviceId)),
registry.WithCountOnly())
- resp, err := store.Store().Service().Search(ctx, opts...)
+ resp, err := backend.Store().Service().Search(ctx, opts...)
if err != nil || resp.Count == 0 {
return false
}
@@ -237,7 +237,7 @@ func GetOneDomainProjectServiceCount(ctx context.Context,
domainProject string)
registry.WithStrKey(key),
registry.WithCountOnly(),
registry.WithPrefix())
- resp, err := store.Store().Service().Search(ctx, opts...)
+ resp, err := backend.Store().Service().Search(ctx, opts...)
if err != nil {
return 0, err
}
@@ -250,7 +250,7 @@ func GetOneDomainProjectInstanceCount(ctx context.Context,
domainProject string)
registry.WithStrKey(key),
registry.WithCountOnly(),
registry.WithPrefix())
- resp, err := store.Store().Instance().Search(ctx, opts...)
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
if err != nil {
return 0, err
}
diff --git a/server/service/util/rule_util.go b/server/service/util/rule_util.go
index c9bb01c..660c2e8 100644
--- a/server/service/util/rule_util.go
+++ b/server/service/util/rule_util.go
@@ -21,7 +21,7 @@ import (
"fmt"
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
pb
"github.com/apache/incubator-servicecomb-service-center/server/core/proto"
scerr
"github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
@@ -66,7 +66,7 @@ func GetRulesUtil(ctx context.Context, domainProject string,
serviceId string) (
}, "/")
opts := append(FromContext(ctx), registry.WithStrKey(key),
registry.WithPrefix())
- resp, err := store.Store().Rule().Search(ctx, opts...)
+ resp, err := backend.Store().Rule().Search(ctx, opts...)
if err != nil {
return nil, err
}
@@ -87,7 +87,7 @@ func RuleExist(ctx context.Context, domainProject string,
serviceId string, attr
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateRuleIndexKey(domainProject,
serviceId, attr, pattern)),
registry.WithCountOnly())
- resp, err := store.Store().RuleIndex().Search(ctx, opts...)
+ resp, err := backend.Store().RuleIndex().Search(ctx, opts...)
if err != nil || resp.Count == 0 {
return false
}
@@ -99,7 +99,7 @@ func GetServiceRuleType(ctx context.Context, domainProject
string, serviceId str
opts := append(FromContext(ctx),
registry.WithStrKey(key),
registry.WithPrefix())
- resp, err := store.Store().Rule().Search(ctx, opts...)
+ resp, err := backend.Store().Rule().Search(ctx, opts...)
if err != nil {
util.Logger().Errorf(err, "Get rule failed.%s", err.Error())
return "", 0, err
@@ -118,7 +118,7 @@ func GetServiceRuleType(ctx context.Context, domainProject
string, serviceId str
func GetOneRule(ctx context.Context, domainProject, serviceId, ruleId string)
(*pb.ServiceRule, error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateServiceRuleKey(domainProject,
serviceId, ruleId)))
- resp, err := store.Store().Rule().Search(ctx, opts...)
+ resp, err := backend.Store().Rule().Search(ctx, opts...)
if err != nil {
util.Logger().Errorf(nil, "Get rule for service failed for
%s.", err.Error())
return nil, err
diff --git a/server/service/util/schema_util.go
b/server/service/util/schema_util.go
index daa4dce..5a0bd0f 100644
--- a/server/service/util/schema_util.go
+++ b/server/service/util/schema_util.go
@@ -17,14 +17,14 @@
package util
import (
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
+
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"golang.org/x/net/context"
)
func CheckSchemaInfoExist(ctx context.Context, key string) (bool, error) {
opts := append(FromContext(ctx), registry.WithStrKey(key),
registry.WithCountOnly())
- resp, errDo := store.Store().Schema().Search(ctx, opts...)
+ resp, errDo := backend.Store().Schema().Search(ctx, opts...)
if errDo != nil {
return false, errDo
}
diff --git a/server/service/util/tag_util.go b/server/service/util/tag_util.go
index dc7fe6b..b416b76 100644
--- a/server/service/util/tag_util.go
+++ b/server/service/util/tag_util.go
@@ -21,7 +21,6 @@ import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
apt "github.com/apache/incubator-servicecomb-service-center/server/core"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
-
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"golang.org/x/net/context"
)
@@ -48,7 +47,7 @@ func AddTagIntoETCD(ctx context.Context, domainProject
string, serviceId string,
func GetTagsUtils(ctx context.Context, domainProject, serviceId string) (tags
map[string]string, err error) {
key := apt.GenerateServiceTagKey(domainProject, serviceId)
opts := append(FromContext(ctx), registry.WithStrKey(key))
- resp, err := store.Store().ServiceTag().Search(ctx, opts...)
+ resp, err := backend.Store().ServiceTag().Search(ctx, opts...)
if err != nil {
util.Logger().Errorf(err, "get service %s tags file failed",
key)
return tags, err
--
To stop receiving notification emails like this one, please contact
little...@apache.org.
