fix: etcd scheme on grpc resolver (#4121)

This commit is contained in:
Kevin Wan 2024-08-27 23:13:37 +08:00 committed by GitHub
parent 896e1a2abb
commit 47d13e5ef8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 36 additions and 9 deletions

View File

@ -9,10 +9,41 @@ import (
"google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver"
) )
type discovBuilder struct{} type discovBuilder struct {
cc resolver.ClientConn
update func()
}
func (b *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) ( func (b *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) (
resolver.Resolver, error) { resolver.Resolver, error) {
b.cc = cc
if err := b.updateState(target); err != nil {
return nil, err
}
return &nopResolver{cc: cc}, nil
}
func (b *discovBuilder) Scheme() string {
return DiscovScheme
}
func (b *discovBuilder) updateState(target resolver.Target) error {
if b.update == nil {
update, err := b.buildEndpointsUpdater(target)
if err != nil {
return err
}
b.update = update
}
b.update()
return nil
}
func (b *discovBuilder) buildEndpointsUpdater(target resolver.Target) (func(), error) {
hosts := strings.FieldsFunc(targets.GetAuthority(target), func(r rune) bool { hosts := strings.FieldsFunc(targets.GetAuthority(target), func(r rune) bool {
return r == EndpointSepChar return r == EndpointSepChar
}) })
@ -29,18 +60,13 @@ func (b *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, _
Addr: val, Addr: val,
}) })
} }
if err := cc.UpdateState(resolver.State{ if err := b.cc.UpdateState(resolver.State{
Addresses: addrs, Addresses: addrs,
}); err != nil { }); err != nil {
logx.Error(err) logx.Error(err)
} }
} }
sub.AddListener(update) sub.AddListener(update)
update()
return &nopResolver{cc: cc}, nil return update, nil
}
func (b *discovBuilder) Scheme() string {
return DiscovScheme
} }

View File

@ -28,7 +28,8 @@ type kubeResolver struct {
stopCh chan struct{} stopCh chan struct{}
} }
func (r *kubeResolver) ResolveNow(_ resolver.ResolveNowOptions) {} func (r *kubeResolver) ResolveNow(_ resolver.ResolveNowOptions) {
}
func (r *kubeResolver) start() { func (r *kubeResolver) start() {
threading.GoSafe(func() { threading.GoSafe(func() {