[kitex source code interpretation] service discovery

Kitex

If you don't already know  Kitex, You can see 《

[ CloudWeGo  Microservice practice  - 01 ]  The opening

》.

If you want to participate in  CloudWeGo  Community , to  Kitex  And other components contribute code , You can see 《

How to give  CloudWeGo  Contribute

》.

Service discovery

Service registration and service discovery are  RPC  The most important components in the framework , Second only to the implementation of encryption and decryption and communication . stay  Kitex  in , The registry already supports mainstream components , Such as  

registry-etcd

、

registry-nacos

、

registry-zookeeper

、

registry-polaris

、

registry-eureka

、

registry-consul

  etc. .

There are generally two approaches to service discovery ：

• Clients do , The client gets one or a group of from the registry （ Then choose one according to the algorithm ）;

• Do it on the server side , The client directly requests an address , Then do load balancing and forward to one of the service nodes , for example  k8s  Of  service .

stay  RPC  In the frame , Service discovery is generally done on the client , Select service node , This is more common .Kitex  That's it .

usage

We use  nacos  For example , The code is as follows ：

https://github.com/kitex-contrib/registry-nacos/blob/main/example/basic/client/main.go#L28


func main() {
  //  Get an instance of the service discovery component through the default configuration
  //  The method has a default configuration to create a  nacos  client , Convenient for follow-up and  nacos  Interaction
r, err := resolver.NewDefaultNacosResolver()
if err != nil {
panic(err)
}
newClient := hello.MustNewClient(
"Hello",
 
  //  With  WithOption  Service discovery instances
client.WithResolver(r),
 
client.WithRPCTimeout(time.Second*3),
)
for {
resp, err := newClient.Echo(context.Background(), &api.Request{Message: "Hello"})
if err != nil {
log.Fatal(err)
}
log.Println(resp)
time.Sleep(time.Second)
}
}


The exploration process

experimental condition 、 The environmental preparation is the same as that in the previous section 《

[ Kitex  Source code interpretation  ]  Service registration

》 bring into correspondence with .

Start server

The case we run is  custom-config  Of  demo. there  nacos  Client parameters are configurable , Convenient debugging .

 go run example/custom-config/server/main.go

Start client

 go run example/custom-config/client/main.go

There are logs to see , We successfully started the client , And can correctly request the target service .

Source code analysis

client.WithResolver

Our starting point is still from a  WithOption  The configuration starts ,client.WithResolver, We can see that the last service discovery instance is placed in  client  Of  options  In the structure ：

Go down again , We see that the parser of service discovery is  189  It's useful to （ Let's not watch first  165  That's ok  WithProxy  The branch of ）.

So we came to  kClient.initLBCache  The place of , This is called when initializing the client of a service . For your convenience , Next, I will analyze it in two parts ：

• NewClient  when , On initialization ;

• When called remotely ,middleware  Service discovery processing in ; 

NewClient  initialization

NewClient  when , Received  WithResolver  Set up , Next, do initialization related work ：

initLBCache

type kClient struct {
svcInfo *serviceinfo.ServiceInfo
mws []endpoint.Middleware
eps endpoint.Endpoint
sEps endpoint.Endpoint
opt *client.Options
lbf *lbcache.BalancerFactory

inited bool
closed bool
}
type BalancerFactory struct {
Hookable
opts Options
cache sync.Map // key -> LoadBalancer
resolver discovery.Resolver
balancer loadbalance.Loadbalancer
rebalancer loadbalance.Rebalancer
sfg singleflight.Group
}
unc (kc *kClient) initLBCache() error {
resolver := kc.opt.Resolver
if resolver == nil {
//  Generate a carry  kerrors.ErrNoResolver  FALSE  resolver
}

balancer := kc.opt.Balancer

//  Weight based load balancing is used by default
//  see  https://www.cloudwego.io/zh/docs/kitex/tutorials/basic-feature/loadbalance/
if balancer == nil {
balancer = loadbalance.NewWeightedBalancer()
}

cacheOpts := lbcache.Options{DiagnosisService: kc.opt.DebugService}
if kc.opt.BalancerCacheOpt != nil {
cacheOpts = *kc.opt.BalancerCacheOpt
}

// 1、 Generate a  BalancerFactory  object ;
// 2、 Check  cacheOpts  Refresh interval settings , If not, it defaults to  5s
// 3、 Check  cacheOpts  Cache expiration time setting , If not, it defaults to  15s
// 4、 The asynchronous collaboration periodically marks that the information discovered by the service is out of date , The details are  BalancerFactory.cache  Field
// 5、 The final will be  resolver + balancer + cacheOpts.RefreshInterval + cacheOpts.ExpireInterval  by  key,
//  The cache to  balancerFactories  Of  sync.Map  On
kc.lbf = lbcache.NewBalancerFactory(resolver, balancer, cacheOpts)

// ...
return nil
}

initMiddlewares

// Middleware deal with input Endpoint and output Endpoint.
type Middleware func(Endpoint) Endpoint

//  ad locum , We just need to know  newResolveMWBuilder  Back to  endpoint.Middleware  The function of
//  We will analyze the specific logic later
func (kc *kClient) initMiddlewares(ctx context.Context) {
// ...
if kc.opt.Proxy == nil {
 // ...
kc.mws = append(kc.mws, newResolveMWBuilder(kc.lbf)(ctx))
// ...
} else {
if kc.opt.Resolver != nil { //  Custom service discovery
kc.mws = append(kc.mws, newResolveMWBuilder(kc.lbf)(ctx))
}
kc.mws = append(kc.mws, newProxyMW(kc.opt.Proxy))
}
// ...
}

//  After the above initialization process , It is the assembly of a series of things that need to be used  Middleware  To  kClient.mws  In this slice

buildInvokeChain

func (kc *kClient) buildInvokeChain() error {
 //  Build a that sends requests and receives responses  Endpoint, Mainly interact with the remote
innerHandlerEp, err := kc.invokeHandleEndpoint()
if err != nil {
return err
}
 
 //  Match the call processing logic with all the processed in the previous step  Middleware  In series , Put it in  kClient.eps  in
kc.eps = endpoint.Chain(kc.mws...)(innerHandlerEp)

// ...
return nil
}

Pay attention to the key points

middleware  Service discovery

kClient.eps

Let's continue with the content above , After reinitialization , We see that all the processing is put in  

kClient.eps

  in . We can start from here , Look at this here  

endpoint.Endpoint

  Where is the function called ：

In the screenshot  311  Row sum  334  That's ok , Let's go in separately , It's all in one function , It's a different branch ：

Come here  kClient.Call, In fact, if you catch up , We can also find , It's being  kitex_gen  The generated client code actually initiates  RPC  Where to call ：

newResolveMWBuilder

Just now we analyzed  

kClient.initMiddlewares

  When , We see there are calls  newResolveMWBuilder  The function of generates a  

endpoint.Middleware

  Function of , In fact, it is also placed in  

kClient.Call

  Of  

kc.ops

  Li implemented. . Here we will look at what has been implemented , Let's take out the main code ：

type RPCInfo interface {
From() EndpointInfo
To() EndpointInfo
Invocation() Invocation
Config() RPCConfig
Stats() RPCStats
}

type Balancer struct {
b *BalancerFactory
target string // a description returned from the resolver's Target method
res atomic.Value // newest and previous discovery result
expire int32 // 0 = normal, 1 = expire and collect next ticker
sharedTicker *sharedTicker
}

return func(ctx context.Context, request, response interface{}) error {
 
  //  Get  RPC  Related information
rpcInfo := rpcinfo.GetRPCInfo(ctx)
dest := rpcInfo.To()
if dest == nil {
return kerrors.ErrNoDestService
}

remote := remoteinfo.AsRemoteInfo(dest)
if remote == nil {
err := fmt.Errorf("unsupported target EndpointInfo type: %T", dest)
return kerrors.ErrInternalException.WithCause(err)
}
if remote.GetInstance() != nil {
return next(ctx, request, response)
}
 
  // 1、 from  BalancerFactory  Get the equalizer corresponding to the target service  Balancer  Etc
  // 2、 There is also a cache here , The refresh mechanism is set according to  cacheOpts.RefreshInterval  The time is refreshed regularly
 //  reference  Balancer.refresh  Code for (https://github.com/cloudwego/kitex/blob/develop/pkg/loadbalance/lbcache/cache.go#L184)
    //  3、 According to  dest  You can get different  Balancer
lb, err := lbf.Get(ctx, dest)
if err != nil {
return kerrors.ErrServiceDiscovery.WithCause(err)
}

  // picker  yes  BalancerFactory.balancer  Corresponding to the equalizer set
  // picker  Of   The main logic is to select one from a set of service instance information , As default  WeightedRandom  Weight based randomization
picker := lb.GetPicker()
if r, ok := picker.(internal.Reusable); ok {
defer r.Recycle()
}
var lastErr error
for {
select {
case <-ctx.Done():
return kerrors.ErrRPCTimeout
default:
}
//  Specifically, select one of the nodes , For example, you can refer to ： randomPicker.Next
ins := picker.Next(ctx, request)
if ins == nil {
return kerrors.ErrNoMoreInstance.WithCause(fmt.Errorf(&quot;last error: %w&quot;, lastErr))
}
remote.SetInstance(ins)

// TODO: generalize retry strategy
if err = next(ctx, request, response); err != nil && retryable(err) {
lastErr = err
continue
}
return err
}
}