fix: zrpc kube resolver builder (#4119)

Signed-off-by: soasurs <soasurs@gmail.com>
This commit is contained in:
soasurs 2024-05-06 14:50:35 +08:00 committed by GitHub
parent 527de1c50e
commit 62ba01120e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -3,10 +3,10 @@ package internal
import ( import (
"context" "context"
"fmt" "fmt"
"runtime/debug"
"time" "time"
"github.com/zeromicro/go-zero/core/logx" "github.com/zeromicro/go-zero/core/logx"
"github.com/zeromicro/go-zero/core/proc"
"github.com/zeromicro/go-zero/core/threading" "github.com/zeromicro/go-zero/core/threading"
"github.com/zeromicro/go-zero/zrpc/resolver/internal/kube" "github.com/zeromicro/go-zero/zrpc/resolver/internal/kube"
"google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver"
@ -21,10 +21,30 @@ const (
nameSelector = "metadata.name=" nameSelector = "metadata.name="
) )
type kubeResolver struct {
cc resolver.ClientConn
stopCh chan struct{}
inf informers.SharedInformerFactory
}
func (r *kubeResolver) start() {
threading.GoSafe(func() {
r.inf.Start(r.stopCh)
})
}
func (r *kubeResolver) ResolveNow(_ resolver.ResolveNowOptions) {}
func (r *kubeResolver) Close() {
close(r.stopCh)
}
type kubeBuilder struct{} type kubeBuilder struct{}
func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn, func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn,
_ resolver.BuildOptions) (resolver.Resolver, error) { _ resolver.BuildOptions) (resolver.Resolver, error) {
logx.Debugf("target: %s, callstack: %s, cc ptr: %p", target, string(debug.Stack()), cc)
svc, err := kube.ParseTarget(target) svc, err := kube.ParseTarget(target)
if err != nil { if err != nil {
return nil, err return nil, err
@ -73,9 +93,6 @@ func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn,
return nil, err return nil, err
} }
threading.GoSafe(func() {
inf.Start(proc.Done())
})
endpoints, err := cs.CoreV1().Endpoints(svc.Namespace).Get(context.Background(), svc.Name, v1.GetOptions{}) endpoints, err := cs.CoreV1().Endpoints(svc.Namespace).Get(context.Background(), svc.Name, v1.GetOptions{})
if err != nil { if err != nil {
return nil, err return nil, err
@ -83,7 +100,15 @@ func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn,
handler.Update(endpoints) handler.Update(endpoints)
return &nopResolver{cc: cc}, nil resolver := &kubeResolver{
cc: cc,
stopCh: make(chan struct{}),
inf: inf,
}
resolver.start()
return resolver, nil
} }
func (b *kubeBuilder) Scheme() string { func (b *kubeBuilder) Scheme() string {