服务注册发现中的watch机制如何与kitex 进行更好的对接？

[liu-song]

目前 kitex-contrib https://github.com/kitex-contrib中已经实现了多种类型的服务注册和发现，但目前相关的实现应该都没有将服务发现场景下的watch 机制集成到kitex里面去，包括register-polaris 也还只是把watch到的实例进行了一个转换，https://github.com/kitex-contrib/registry-polaris/blob/main/resolver.go#L83，转换成 kitex 框架里面的discovery.Change，没有将这个discovery.Change 进一步对接到kitex中。kitex 目前是通过event, watch 然后dispatch的机制来实现的discoveryEventHandler。但如何将kitex-contrib 中相关的服务注册发现的实现中的watch到的discovery.Change对接到这个newResolveMWBuilder 当中，感觉有些困惑，有什么比较好的方案来实现这个对接吗？

[YangruiEmma]

@liu-song 抱歉，没有注意到你发起的disussion。watch需要保持长连接，每个服务实例都与和注册中心保持长连接感知变更并不是一个建议的实践方式，在微服务规模变大后该方式并不可取，除非是通过proxy获取服务实例可以采用watch方式，否则建议的方式是短连接请求注册中心获取实例（触发请求由Kitex缓存策略决定）。

update: 也可以通过在 resolver 里实现一个 push -> pull 的 adaptor 来接入 kitex，详见后面的伪代码示例。

[someview]

watch怎么就变成长连接了呢，watch机制是应用层自己去实现的，如何保证watch的性能问题，这是应用自己需要去评估的，评估自己的服务注册中心是否能满足watch的需求和应用规模. 举个例子， 使用长轮询来实现watch机制.
服务规模变大以后，watch不可取，你是怎么证明不可取的呢，是不是你们实现watch的方式有问题呢.
用cache来决定如何更新，这不是很荒谬吗

[li-jin-gou]

上面已经说的是“建议”了，肯定是经过多重考量和踩坑才做出的回答，如果如果非要对 watch 有强诉求，可以把 kitex 的 cache 关掉，自己实现一个 watch 的服务注册发现扩展。

另外举个例子所说的 长轮训 不也是需要底层连接长时间持有才能实现的吧。

[someview]

有没有入口是框架层面考虑的问题，实不实现是应用层面考虑的问题.

[li-jin-gou]

我的评价是解决用户问题优先，不会为了对标能力而去实现什么功能，如果有真实场景的用户诉求可以飞书联系，详细聊聊怎么搞&支持。目前这么多的内外部用户都没有这样的对 watch 的强烈反馈。

[someview]

上面已经说的是“建议”了，肯定是经过多重考量和踩坑才做出的回答，如果如果非要对 watch 有强诉求，可以把 kitex 的 cache 关掉，自己实现一个 watch 的服务注册发现扩展。

另外举个例子所说的 长轮训 不也是需要底层连接长时间持有才能实现的吧。

上面已经说的是“建议”了，肯定是经过多重考量和踩坑才做出的回答，如果如果非要对 watch 有强诉求，可以把 kitex 的 cache 关掉，自己实现一个 watch 的服务注册发现扩展。

另外举个例子所说的 长轮训 不也是需要底层连接长时间持有才能实现的吧。
觉得很不可思议，一个pod一个长连接，一个服务注册中心，支持不了百万级别的tcp连接吗。字节肯定是用的服务网格。如果是采用胖客户端来做服务发现，负载均衡，那访问的服务扩容了， 字节的应用服务可以容忍cache在更新之前的失效？？？

[li-jin-gou]

早期用的胖客户端，目前是用服务网格搞的，如果也想用服务网格可以看 https://cloudwego-git-fork-ozline-dev-cloudwego.vercel.app/zh/docs/kitex/tutorials/advanced-feature/xds/

地址发成debug 地址了，看下面的吧

[li-jin-gou]

https://cloudwego.cn/zh/docs/kitex/tutorials/advanced-feature/xds/

[someview]

通常来讲，服务发现机制是watch + get 组成. 根据实际的考虑，框架还是得有这个东西，只是用不用，这个是实践的考量。 当然，也许我这个想法是错误的

[someview]

我的评价是解决用户问题优先，不会为了对标能力而去实现什么功能，如果有真实场景的用户诉求可以飞书联系，详细聊聊怎么搞&支持。目前这么多的内外部用户都没有这样的对 watch 的强烈反馈。

这和对标能力没有关系啊，只是一个接口设计的问题而已。不清楚字节内部以前使用胖客户端时的情况，如果长期使用cache，也不一定会会引起问题.只是对于应用来说，扩容什么的，都是要缓慢进行的。不会说一下扩容多少.如果扩容一个pod10s 算， ticker 30s，也就是说， endpoint的变更延迟20s. 如果是根据指标来扩缩容，那么，本身这个确认时间也是差不多10s级别这个数量级别.

func (bl *Balancer) Refresh() {
	res, err := bl.b.resolver.Resolve(context.Background(), bl.target)
	if err != nil {
		klog.Warnf("KITEX: resolver refresh failed, key=%s error=%s", bl.target, err.Error())
		return
	}
	renameResultCacheKey(&res, bl.b.resolver.Name())
	prev := bl.res.Load().(discovery.Result)
	if bl.b.rebalancer != nil {
		if ch, ok := bl.b.resolver.Diff(res.CacheKey, prev, res); ok {
			bl.b.rebalancer.Rebalance(ch)
		}
	}
	// replace previous result
	bl.res.Store(res)
}

只需要把这个当成callback暴露给resolver，在new resolver的时候，自定义的服务发现启动一个异步的watcher，watcher中调用这个内部的cb即可.

[li-jin-gou]

cc @YangruiEmma 还是让专业的来回答吧🥺

[GuangmingLuo]

@someview hello 同学，首先感谢你的关注和建议。Kitex 诞生于字节内部，诞生于超大规模场景，有内部的诸多可用性考量。如果你在企业内部实际的生产环境中有不同的需求，欢迎通过飞书联系我们，或者在 Issue 中详细描述下，我们再来评估下如何支持。

[li-jin-gou]

可以，从真实用户需求出发 👍🏻

[someview]

这个是grpc-go1.56里面resolver的定义:

// Resolver is also a resolver builder.
// It's build() function always returns itself.
type Resolver struct {
	// BuildCallback is called when the Build method is called.  Must not be
	// nil.  Must not be changed after the resolver may be built.
	BuildCallback func(resolver.Target, resolver.ClientConn, resolver.BuildOptions)
	// ResolveNowCallback is called when the ResolveNow method is called on the
	// resolver.  Must not be nil.  Must not be changed after the resolver may
	// be built.
	ResolveNowCallback func(resolver.ResolveNowOptions)
	// CloseCallback is called when the Close method is called.  Must not be
	// nil.  Must not be changed after the resolver may be built.
	CloseCallback func()
	scheme        string

	// Fields actually belong to the resolver.
	mu             sync.Mutex // Guards access to CC.
	CC             resolver.ClientConn
	bootstrapState *resolver.State
}

也是包含了一个resolveNowCb这样的回调

[GuangmingLuo]

同学，我们就不重复回复了，Kitex 是实际用户需求驱动的，不是其它项目有的逻辑我们就一定要有，如果有用户在生产环境有实际需求，我们会考虑支持。

[someview]

@someview hello 同学，首先感谢你的关注和建议。Kitex 诞生于字节内部，诞生于超大规模场景，有内部的诸多可用性考量。如果你在企业内部实际的生产环境中有不同的需求，欢迎通过飞书联系我们，或者在 Issue 中详细描述下，我们再来评估下如何支持。

我们现在是用的胖客户端，公司规模有限，服务网格有一些问题我们自己没法解决。现在的客户端负载均衡，通过访问代理的形式，感知到最新的dns记录。 我这里简单贴一下.net里面的grpc框架关于服务发现接口定义:

public abstract class ResolverFactory
{
    internal static readonly ResolverFactory[] KnownLoadResolverFactories = new ResolverFactory[1]
    {
        new DnsResolverFactory(Timeout.InfiniteTimeSpan)
    };

    //
    // 摘要:
    //     Gets the resolver factory name. A factory is used when the target System.Uri
    //     scheme matches the factory name.
    public abstract string Name { get; }

    //
    // 摘要:
    //     Creates a new Grpc.Net.Client.Balancer.Resolver with the specified options.
    //
    // 参数:
    //   options:
    //     Options for creating a Grpc.Net.Client.Balancer.Resolver.
    //
    // 返回结果:
    //     A new Grpc.Net.Client.Balancer.Resolver.
    public abstract Resolver Create(ResolverOptions options);
}

public abstract class Resolver : IDisposable
{
    /// <summary>
    /// Starts listening to resolver for results with the specified callback. Can only be called once.
    /// <para>
    /// The <see cref="ResolverResult"/> passed to the callback has addresses when successful,
    /// otherwise a <see cref="Status"/> details the resolution error.
    /// </para>
    /// </summary>
    /// <param name="listener">The callback used to receive updates on the target.</param>
    public abstract void Start(Action<ResolverResult> listener);

    /// <summary>
    /// Refresh resolution. Can only be called after <see cref="Start(Action{ResolverResult})"/>.
    /// The default implementation is no-op.
    /// <para>
    /// This is only a hint. Implementation takes it as a signal but may not start resolution.
    /// </para>
    /// </summary>
    public virtual void Refresh()
    {
        // no-op
    }

    /// <summary>
    /// Releases the unmanaged resources used by the <see cref="LoadBalancer"/> and optionally releases
    /// the managed resources.
    /// </summary>
    /// <param name="disposing">
    /// <c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.
    /// </param>
    protected virtual void Dispose(bool disposing)
    {
    }

    /// <summary>
    /// Disposes the <see cref="Resolver"/>. Stops resolution.
    /// </summary>
    public void Dispose()
    {
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
}

Action这个地址变更时的cb，本身是框架内部提供的，refresh里面调用这个cb，就可以主动刷新内部的dns。grpc-go也是类似的实现方式，resolver本身提供了一个refresh方法。当然了，这些框架里面把kitex现在resolver解析完地址的部分也抽象成接口了，从地址到可用连接这一层. 这个在kitex的改动上是不大，并且也很容易就能实现了.

我本来想表达的是，也许一开始就应该由watch, 逻辑上是完备的，即使用不到。遇到过类似案例：在使用quic-go的时候， 希望能够设置不加密，但是quic-go这个库本身做不到这件事情，因为quic协议本身是和加密绑定在一起的。然而其他语言的某些quic实现是能做到这件事情，因为实现quic协议的时候，是按照分层的角度去实现的，所以之需要把那一层设置为空就行了, 即使quic固定必须tls.

[li-jin-gou]

我想表达的是如果是需要使用 kitex 并且需要 watch支持的话直接说需求就可以，有实际需求并且合理，就会支持的。🤓

按照上面那么加，首先改动不大，带来了 break change，解决痛点还不一定多，需要 kitex 开发者和 kitex 用户去适配，另外如果真的想搞把缓存一关，自己实现 watch 就可以 https://www.cloudwego.io/zh/docs/kitex/tutorials/framework-exten/service_discovery/

[someview]

缓存一关也不行啊，是我理解错了吗，没有从resolver到subchannel的路径啊。除非直接discovery，把kitex当成单个端点开处理,在外面嵌套一层。去浏览issue证明肯定有这个需求...

[li-jin-gou]

是你理解错了

[li-jin-gou]

缓存关了，相当于每次服务发现都去调用 Resolve 接口了，在服务发现扩展里面实现 watch 逻辑🤨

[felix021]

以下伪代码展示如何实现一个基于 watch 机制的 resolver，供参考：

type MyResolver struct {
    endpoints sync.Map
}

func NewMyResolver() {
    r := &MyResolver{}
    go func() {
        watch(func(ep endpoint) {
            // add or remove an endpoint from the resolver's cache
            r.endpoints.Store(ep.Key(), ep)
        })
    }
    return r
}

func (r *MyResolver) Resolve(...) ... {
    return r.endpoints.LoadAll()
}

// omitting other methods here

Kitex 默认每隔 5s 会调用一次 Resolve 方法，因此 watch 得到的实例变化，平均 2.5s、最长 5s 内会生效。

[someview]

额，是这个意思. 好吧，我一直想的都是怎么主动刷新过去。

[felix021]

那 issue 我就关了，如果还有问题的话可以再沟通。