feat: refactor etcd registry protocol (#485)

pkg/ecode/code.go

@@ -15,11 +15,8 @@
 package ecode
 
 import (
- "encoding/json"
- "net/http"
  "sync"
 
- "github.com/douyu/jupiter/pkg/governor"
  "github.com/douyu/jupiter/pkg/xlog"
  spb "google.golang.org/genproto/googleapis/rpc/status"
  "google.golang.org/grpc/codes"
@@ -38,20 +35,6 @@ var (
  OK = add(int(codes.OK), "OK")
 )
 
-func init() {
- // status code list
- governor.HandleFunc("/status/code/list", func(w http.ResponseWriter, r *http.Request) {
- var rets = make(map[int]*spbStatus)
- _codes.Range(func(key, val interface{}) bool {
- code := key.(int)
- status := val.(*spbStatus)
- rets[code] = status
- return true
- })
- _ = json.NewEncoder(w).Encode(rets)
- })
-}
-
 // Add ...
 func Add(code int, message string) *spbStatus {
  if code > maxCustomizeCode {

pkg/env.go

@@ -5,6 +5,7 @@ import (
  "fmt"
  "os"
 
+ "github.com/douyu/jupiter/pkg/conf"
  "github.com/douyu/jupiter/pkg/constant"
 )
 
@@ -42,7 +43,17 @@ func SetAppLogDir(logDir string) {
  appLogDir = logDir
 }
 
+// AppMode returns the current application mode.
 func AppMode() string {
+ confMode := conf.GetString(constant.ConfigKey("mode"))
+ if appMode == "" {
+ if confMode == "" {
+ return "unkown-mode"
+ }
+
+ return confMode
+ }
+
  return appMode
 }
 

pkg/registry/etcdv3/config.go

@@ -24,6 +24,7 @@ import (
  "github.com/douyu/jupiter/pkg/registry"
  "github.com/douyu/jupiter/pkg/singleton"
  "github.com/douyu/jupiter/pkg/xlog"
+ "github.com/spf13/cast"
  "go.uber.org/zap"
 )
 
@@ -53,7 +54,7 @@ func DefaultConfig() *Config {
  ReadTimeout: time.Second * 3,
  Prefix: "wsd-reg",
  logger: xlog.Jupiter(),
- ServiceTTL: 0,
+ ServiceTTL: cast.ToDuration("20s"),
  }
 }
 

pkg/registry/etcdv3/registry.go

@@ -19,13 +19,13 @@ import (
  "encoding/json"
  "fmt"
  "net"
- "net/url"
  "strings"
  "sync"
  "time"
 
  "github.com/douyu/jupiter/pkg"
  "github.com/douyu/jupiter/pkg/client/etcdv3"
+ "github.com/douyu/jupiter/pkg/conf"
  "github.com/douyu/jupiter/pkg/constant"
  "github.com/douyu/jupiter/pkg/ecode"
  "github.com/douyu/jupiter/pkg/registry"
@@ -46,6 +46,12 @@ type etcdv3Registry struct {
  sessions map[string]*concurrency.Session
 }
 
+const (
+ servicePrefix = "%s:%s:%s:%s/"
+ // schema:appname:version:mode/host:port
+ registerService = "%s:%s:%s:%s/%s"
+)
+
 func newETCDRegistry(config *Config) (*etcdv3Registry, error) {
  if config.logger == nil {
  config.logger = xlog.Jupiter()
@@ -84,7 +90,7 @@ func (reg *etcdv3Registry) UnregisterService(ctx context.Context, info *server.S
 
 // ListServices list service registered in registry with name `name`
 func (reg *etcdv3Registry) ListServices(ctx context.Context, name string, scheme string) (services []*server.ServiceInfo, err error) {
- target := fmt.Sprintf("/%s/%s/providers/%s://", reg.Prefix, name, scheme)
+ target := fmt.Sprintf(servicePrefix, scheme, name, "v1", conf.GetString("app.mode"))
  getResp, getErr := reg.client.Get(ctx, target, clientv3.WithPrefix())
  if getErr != nil {
  reg.logger.Error(ecode.MsgWatchRequestErr, xlog.FieldErrKind(ecode.ErrKindRequestErr), xlog.FieldErr(getErr), xlog.FieldAddr(target))
@@ -105,7 +111,7 @@ func (reg *etcdv3Registry) ListServices(ctx context.Context, name string, scheme
 
 // WatchServices watch service change event, then return address list
 func (reg *etcdv3Registry) WatchServices(ctx context.Context, name string, scheme string) (chan registry.Endpoints, error) {
- prefix := fmt.Sprintf("/%s/%s/", reg.Prefix, name)
+ prefix := fmt.Sprintf(servicePrefix, scheme, name, "v1", pkg.AppMode())
  watch, err := reg.client.WatchPrefix(context.Background(), prefix)
  if err != nil {
  return nil, err
@@ -213,8 +219,8 @@ func (reg *etcdv3Registry) registerMetric(ctx context.Context, info *server.Serv
  opOptions := make([]clientv3.OpOption, 0)
  // opOptions = append(opOptions, clientv3.WithSerializable())
  if ttl := reg.Config.ServiceTTL.Seconds(); ttl > 0 {
- //todo ctx without timeout for same as service life?
- sess, err := reg.getSession(key, concurrency.WithTTL(int(ttl)))
+ // 这里基于应用名为key做缓存，每个服务实例应该只需要创建一个lease，降低etcd的压力
+ sess, err := reg.getSession(info.Name, concurrency.WithTTL(int(ttl)))
  if err != nil {
  return err
  }
@@ -244,8 +250,8 @@ func (reg *etcdv3Registry) registerBiz(ctx context.Context, info *server.Service
  opOptions := make([]clientv3.OpOption, 0)
  // opOptions = append(opOptions, clientv3.WithSerializable())
  if ttl := reg.Config.ServiceTTL.Seconds(); ttl > 0 {
- //todo ctx without timeout for same as service life?
- sess, err := reg.getSession(key, concurrency.WithTTL(int(ttl)))
+ // 这里基于应用名为key做缓存，每个服务实例应该只需要创建一个lease，降低etcd的压力
+ sess, err := reg.getSession(info.Name, concurrency.WithTTL(int(ttl)))
  if err != nil {
  return err
  }
@@ -262,19 +268,21 @@ func (reg *etcdv3Registry) registerBiz(ctx context.Context, info *server.Service
 }
 
 func (reg *etcdv3Registry) getSession(k string, opts ...concurrency.SessionOption) (*concurrency.Session, error) {
- reg.rmu.RLock()
+ // 需要对整个方法加锁，防止并发创建session
+ reg.rmu.Lock()
+ defer reg.rmu.Unlock()
  sess, ok := reg.sessions[k]
- reg.rmu.RUnlock()
  if ok {
  return sess, nil
  }
+
  sess, err := concurrency.NewSession(reg.client.Client, opts...)
  if err != nil {
+ xlog.Jupiter().Error("create session failed", xlog.FieldKeyAny(k))
  return sess, err
  }
- reg.rmu.Lock()
  reg.sessions[k] = sess
- reg.rmu.Unlock()
+ xlog.Jupiter().Info("create session", xlog.FieldKeyAny(k), xlog.FieldValueAny(sess))
  return sess, nil
 }
 
@@ -296,43 +304,23 @@ func (reg *etcdv3Registry) delSession(k string) error {
 }
 
 func (reg *etcdv3Registry) registerKey(info *server.ServiceInfo) string {
- return registry.GetServiceKey(reg.Prefix, info)
+ return fmt.Sprintf(registerService, info.Scheme, info.Name, "v1", pkg.AppMode(), info.Address)
 }
 
 func (reg *etcdv3Registry) registerValue(info *server.ServiceInfo) string {
- return registry.GetServiceValue(info)
+ update := Update{
+ Op: Add,
+ Addr: info.Address,
+ }
+
+ val, _ := json.Marshal(update)
+
+ return string(val)
 }
 
 func deleteAddrList(al *registry.Endpoints, prefix, scheme string, kvs ...*mvccpb.KeyValue) {
  for _, kv := range kvs {
  var addr = strings.TrimPrefix(string(kv.Key), prefix)
- if strings.HasPrefix(addr, "providers/"+scheme) {
- // 解析服务注册键
- addr = strings.TrimPrefix(addr, "providers/")
- if addr == "" {
- continue
- }
- uri, err := url.Parse(addr)
- if err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
- continue
- }
- delete(al.Nodes, uri.String())
- }
-
- if strings.HasPrefix(addr, "configurators/"+scheme) {
- // 解析服务配置键
- addr = strings.TrimPrefix(addr, "configurators/")
- if addr == "" {
- continue
- }
- uri, err := url.Parse(addr)
- if err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
- continue
- }
- delete(al.RouteConfigs, uri.String())
- }
 
  if isIPPort(addr) {
  // 直接删除addr 因为Delete操作的value值为空
@@ -345,117 +333,28 @@ func deleteAddrList(al *registry.Endpoints, prefix, scheme string, kvs ...*mvccp
 func updateAddrList(al *registry.Endpoints, prefix, scheme string, kvs ...*mvccpb.KeyValue) {
  for _, kv := range kvs {
  var addr = strings.TrimPrefix(string(kv.Key), prefix)
- switch {
- // 解析服务注册键
- case strings.HasPrefix(addr, "providers/"+scheme):
- addr = strings.TrimPrefix(addr, "providers/")
- uri, err := url.Parse(addr)
- if err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
- continue
- }
- var serviceInfo server.ServiceInfo
- if err := json.Unmarshal(kv.Value, &serviceInfo); err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
- continue
- }
- if serviceInfo.Enable {
- al.Nodes[uri.String()] = serviceInfo
- } else {
- delete(al.Nodes, uri.String())
- }
-
- case strings.HasPrefix(addr, "configurators/"+scheme):
- addr = strings.TrimPrefix(addr, "configurators/")
-
- uri, err := url.Parse(addr)
- if err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
+ if isIPPort(addr) {
+ var meta Update
+ if err := json.Unmarshal(kv.Value, &meta); err != nil {
+ xlog.Jupiter().Error("unmarshal meta", xlog.FieldErr(err),
+ xlog.FieldExtMessage("value", string(kv.Value), "key", string(kv.Key)))
  continue
  }
 
- if strings.HasPrefix(uri.Path, "/routes/") { // 路由配置
- var routeConfig registry.RouteConfig
- if err := json.Unmarshal(kv.Value, &routeConfig); err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
- continue
- }
- routeConfig.ID = strings.TrimPrefix(uri.Path, "/routes/")
- routeConfig.Scheme = uri.Scheme
- routeConfig.Host = uri.Host
- al.RouteConfigs[uri.String()] = routeConfig
- }
-
- if strings.HasPrefix(uri.Path, "/providers/") {
- var providerConfig registry.ProviderConfig
- if err := json.Unmarshal(kv.Value, &providerConfig); err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
- continue
- }
- providerConfig.ID = strings.TrimPrefix(uri.Path, "/providers/")
- providerConfig.Scheme = uri.Scheme
- providerConfig.Host = uri.Host
- al.ProviderConfigs[uri.String()] = providerConfig
- }
-
- if strings.HasPrefix(uri.Path, "/consumers/") {
- var consumerConfig registry.ConsumerConfig
- if err := json.Unmarshal(kv.Value, &consumerConfig); err != nil {
- xlog.Jupiter().Error("parse uri", xlog.FieldErrKind(ecode.ErrKindUriErr), xlog.FieldErr(err), xlog.FieldKey(string(kv.Key)))
- continue
+ switch meta.Op {
+ case Add:
+ al.Nodes[addr] = server.ServiceInfo{
+ Address: addr,
  }
- consumerConfig.ID = strings.TrimPrefix(uri.Path, "/consumers/")
- consumerConfig.Scheme = uri.Scheme
- consumerConfig.Host = uri.Host
- al.ConsumerConfigs[uri.String()] = consumerConfig
+ case Delete:
+ delete(al.Nodes, addr)
  }
  }
  }
 }
 
+
 func isIPPort(addr string) bool {
  _, _, err := net.SplitHostPort(addr)
  return err == nil
 }
-
-/*
-key: /jupiter/main/configurator/grpc:///routes/1
-val:
-{
- "upstream": { // 客户端配置
- "nodes": { // 按照node负载均衡
- "127.0.0.1:1980": 1,
- "127.0.0.1:1981": 4
- },
- "group": { // 按照group负载均衡
- "red": 2,
- "green": 1
- }
- },
- "uri": "/hello",
- "deployment": "open_api"
-}
-
-key: /jupiter/main/configurator/grpc://127.0.0.1/routes/2
-val:
-{
- "upstream": { // 客户端配置
- "nodes": { // 按照node负载均衡
- "127.0.0.1:1980": 1,
- "127.0.0.1:1981": 1
- },
- "group": { // 按照group负载均衡
- "red": 1,
- "green": 2
- }
- },
- "uri": "/hello",
- "deployment": "core_api" // 部署组
-}
-
-key: /jupiter/main/configurator/grpc:///consumers/client-demo
-val:
-{
-
-}
-*/

pkg/registry/etcdv3/update.go

@@ -0,0 +1,36 @@
+// Copyright 2022 Douyu
+//
+// Licensed 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 etcdv3
+
+type Operation uint8
+
+const (
+ // Add indicates an Endpoint is added.
+ Add Operation = iota
+ // Delete indicates an existing address is deleted.
+ Delete
+)
+
+// Update defines a name resolution update. Notice that it is not valid having both
+// empty string Addr and nil Metadata in an Update.
+type Update struct {
+ // Op indicates the operation of the update.
+ Op Operation
+ // Addr is the updated address. It is empty string if there is no address update.
+ Addr string
+ // Metadata is the updated metadata. It is nil if there is no metadata update.
+ // Metadata is not required for a custom naming implementation.
+ Metadata interface{}
+}