kzkzzzz/mycommon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
kzkzzzz
2025-03-23 20:29:29 +08:00
parent 7e0bf82418
commit 9fd0eaadb8
14 changed files with 661 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
myregistry/consul/builder.go
View File

@@ -5,6 +5,7 @@ import (
"fmt"
"github.com/jpillora/backoff"
"google.golang.org/grpc/grpclog"
"log"
"sort"
"time"
@@ -33,7 +34,7 @@ func (b *builder) Build(url resolver.Target, cc resolver.ClientConn, opts resolv
ctx, cancel := context.WithCancel(context.Background())
pipe := make(chan []string)
go watchConsulService(ctx, cli.Health(), tgt, pipe)
go populateEndpoints(ctx, cc, pipe)
go populateEndpoints(ctx, cc, pipe, tgt)
return &resolvr{cancelFunc: cancel}, nil
}
@@ -156,7 +157,7 @@ func watchConsulService(ctx context.Context, s servicer, tgt target, out chan<-
}
}
func populateEndpoints(ctx context.Context, clientConn resolver.ClientConn, input <-chan []string) {
func populateEndpoints(ctx context.Context, clientConn resolver.ClientConn, input <-chan []string, tgt target) {
for {
select {
case cc := <-input:
@@ -168,7 +169,11 @@ func populateEndpoints(ctx context.Context, clientConn resolver.ClientConn, inpu
for c := range connsSet {
conns = append(conns, resolver.Address{Addr: c})
}
sort.Sort(byAddressString(conns)) // Don't replace the same address list in the balancer
log.Printf("update conn: %s - %s", tgt.Service, conns)
err := clientConn.UpdateState(resolver.State{Addresses: conns})
if err != nil {
grpclog.Errorf("[Consul resolver] Couldn't update client connection. error={%v}", err)

126
myregistry/consul/consul.go
View File

@@ -4,11 +4,14 @@ import (
"fmt"
"git.makemake.in/kzkzzzz/mycommon/myconf"
"git.makemake.in/kzkzzzz/mycommon/myregistry"
"github.com/google/uuid"
api "github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/api/watch"
"github.com/hashicorp/go-hclog"
"github.com/rs/xid"
"log"
"net"
"net/url"
"os"
"time"
)
@@ -18,6 +21,15 @@ type Consul struct {
client *api.Client
serviceIds map[string][]string
serviceTags []string
services []*api.AgentServiceRegistration
}
type Opt func(*Consul)
func WithServiceTags(tags ...string) Opt {
return func(c *Consul) {
c.serviceTags = tags
}
}
func (c *Consul) Name() string {
@@ -26,7 +38,13 @@ func (c *Consul) Name() string {
func (c *Consul) Register(service *myregistry.ServiceInfo) error {
// 健康检查
serviceId := uuid.New().String()
serviceId := xid.New().String()
hostname, err := os.Hostname()
if err != nil {
log.Printf("get hostname err: %s", err)
} else {
serviceId = fmt.Sprintf("%s-%s", hostname, serviceId)
}
c.serviceIds[service.ServiceName] = append(c.serviceIds[service.ServiceName], serviceId)
@@ -34,13 +52,13 @@ func (c *Consul) Register(service *myregistry.ServiceInfo) error {
CheckID: serviceId,
TCP: fmt.Sprintf("%s:%d", service.Ip, service.Port),
Timeout: "5s", // 超时时间
Interval: "20s", // 运行检查的频率
Interval: "30s", // 运行检查的频率
// 指定时间后自动注销不健康的服务节点
// 最小超时时间为1分钟，收获不健康服务的进程每30秒运行一次，因此触发注销的时间可能略长于配置的超时时间。
DeregisterCriticalServiceAfter: "5m",
DeregisterCriticalServiceAfter: "6m",
Status: "passing",
}
srv := &api.AgentServiceRegistration{
svc := &api.AgentServiceRegistration{
ID: serviceId, // 服务唯一ID
Name: service.ServiceName, // 服务名称
Tags: c.serviceTags, // 为服务打标签
@@ -49,7 +67,9 @@ func (c *Consul) Register(service *myregistry.ServiceInfo) error {
Check: check,
}
return c.client.Agent().ServiceRegister(srv)
c.services = append(c.services, svc)
return c.client.Agent().ServiceRegister(svc)
}
func (c *Consul) Deregister(service *myregistry.ServiceInfo) error {
@@ -62,22 +82,23 @@ func (c *Consul) Deregister(service *myregistry.ServiceInfo) error {
return nil
}
type Conf struct {
Addr string
Token string
}
func MustNew(conf *myconf.Config) *Consul {
consul, err := New(conf)
func MustNew(conf *myconf.Config, opts ...Opt) *Consul {
consul, err := New(conf, opts...)
if err != nil {
panic(err)
}
return consul
}
func New(conf *myconf.Config) (*Consul, error) {
func New(conf *myconf.Config, opts ...Opt) (*Consul, error) {
cfg := api.DefaultConfig()
cfg.Address = conf.GetString("addr")
if cfg.Address == "" {
return nil, fmt.Errorf("consul address is empty")
}
cfg.Transport.DialContext = (&net.Dialer{
Timeout: 3 * time.Second,
KeepAlive: 20 * time.Second,
@@ -99,27 +120,64 @@ func New(conf *myconf.Config) (*Consul, error) {
if err != nil {
return nil, err
}
cl := &Consul{
client: client,
serviceIds: make(map[string][]string),
serviceTags: make([]string, 0),
serviceTags: conf.GetStringSlice("serviceTags"),
services: make([]*api.AgentServiceRegistration, 0),
}
if v := conf.GetStringSlice("serviceTags"); len(v) > 0 {
cl.serviceTags = v
} else {
cl.serviceTags = []string{}
for _, opt := range opts {
opt(cl)
}
go cl.healthCheck()
return cl, nil
}
func (c *Consul) healthCheck() {
wlog := newWatchLogger()
wp, err := watch.Parse(map[string]any{
"type": "services",
})
if err != nil {
panic(fmt.Sprintf("parse watch err: %s", err))
}
wp.Handler = func(u uint64, raw any) {
if wlog.isWatchErr == true {
for _, svc := range c.services {
//c.client.Agent().ServiceDeregister(svc.ID)
err := c.client.Agent().ServiceRegister(svc)
if err != nil {
log.Printf("retry register service err: %s: %s", svc.Name, err)
} else {
log.Printf("retry register service ok: %s", svc.Name)
}
}
wlog.isWatchErr = false
}
//fmt.Println("watch", u, raw)
}
err = wp.RunWithClientAndHclog(c.client, wlog)
if err != nil {
log.Printf("watch err: %s", err)
}
}
func (c *Consul) Client() *api.Client {
return c.client
}
func GrpcUrl(serviceName string, conf *myconf.Config) string {
return GrpcUrlWithTag("", serviceName, conf)
return GrpcUrlWithTag("", "grpc@"+serviceName, conf)
}
func GrpcUrlWithTag(tag string, serviceName string, conf *myconf.Config) string {
@@ -129,6 +187,10 @@ func GrpcUrlWithTag(tag string, serviceName string, conf *myconf.Config) string
Path: serviceName,
}
if u.Host == "" {
panic("consul address is empty")
}
query := u.Query()
query.Set("healthy", "true")
@@ -151,3 +213,25 @@ func GrpcUrlWithTag(tag string, serviceName string, conf *myconf.Config) string
return u.String()
}
type watchLogger struct {
hclog.Logger
isWatchErr bool
}
func newWatchLogger() *watchLogger {
return &watchLogger{Logger: hclog.New(&hclog.LoggerOptions{
Name: "watch",
Output: os.Stdout,
})}
}
func (l *watchLogger) Error(msg string, args ...interface{}) {
l.isWatchErr = true
log.Printf("is watch err: %s", msg)
l.Logger.Error(msg, args...)
}
func (l *watchLogger) Named(name string) hclog.Logger {
return l
}