diff --git a/.github/workflows/golangci-lint.yml b/.github/workflows/golangci-lint.yml index 6ade9d4..69cac2f 100644 --- a/.github/workflows/golangci-lint.yml +++ b/.github/workflows/golangci-lint.yml @@ -24,4 +24,4 @@ jobs: - name: golangci-lint uses: golangci/golangci-lint-action@v9.3.0 with: - version: v2.14 + version: v2.15 diff --git a/Makefile b/Makefile index e78a076..2dcf1c4 100644 --- a/Makefile +++ b/Makefile @@ -10,7 +10,7 @@ format: goimports -w `find . -name '*.go'` lint: - docker run --rm -v ${CURDIR}:/app -w /app golangci/golangci-lint:v2.14-alpine golangci-lint run -v + docker run --rm -v ${CURDIR}:/app -w /app golangci/golangci-lint:v2.15-alpine golangci-lint run -v unittests: go test ./cmd/... ./internal/... diff --git a/internal/dcs/zk_host_provider.go b/internal/dcs/zk_host_provider.go index d537083..bb81af9 100644 --- a/internal/dcs/zk_host_provider.go +++ b/internal/dcs/zk_host_provider.go @@ -16,12 +16,18 @@ type zkhost struct { resolved []string } +type zkaddr struct { + host string + addr string +} + type RandomHostProvider struct { ctx context.Context logger *zerolog.Logger resolver *net.Resolver tried map[string]struct{} hosts sync.Map + preferred zkaddr hostsKeys []string lookupTTL time.Duration lookupTimeout time.Duration @@ -51,7 +57,8 @@ func NewRandomHostProvider(ctx context.Context, config *RandomHostProviderConfig func (rhp *RandomHostProvider) Init(servers []string) error { rhp.hostsKeys = nil rhp.hosts = sync.Map{} - var allResolvedServers []string + rhp.preferred = zkaddr{} + var allResolvedServers []zkaddr for _, host := range servers { resolved, err := rhp.resolveHost(host) @@ -59,7 +66,9 @@ func (rhp *RandomHostProvider) Init(servers []string) error { rhp.logger.Error().Err(err).Msgf("host definition %s is invalid", host) continue } - allResolvedServers = append(allResolvedServers, resolved...) + for _, addr := range resolved { + allResolvedServers = append(allResolvedServers, zkaddr{host: host, addr: addr}) + } rhp.hosts.Store(host, zkhost{ resolved: resolved, lastLookup: time.Now(), @@ -71,32 +80,51 @@ func (rhp *RandomHostProvider) Init(servers []string) error { return fmt.Errorf("unable to resolve any host from %v", servers) } - if err := rhp.checkZKConnectivity(allResolvedServers); err != nil { + preferred, err := rhp.checkZKConnectivity(allResolvedServers) + if err != nil { return err } + rhp.preferred = preferred go rhp.resolveHosts() return nil } -func (rhp *RandomHostProvider) checkZKConnectivity(servers []string) error { +func (rhp *RandomHostProvider) checkZKConnectivity(servers []zkaddr) (zkaddr, error) { if len(servers) == 0 { - return fmt.Errorf("no servers available for connectivity check") + return zkaddr{}, fmt.Errorf("no servers available for connectivity check") } + ctx, cancel := context.WithTimeout(rhp.ctx, rhp.connectivityCheckTimeout) + defer cancel() + + type checkResult struct { + err error + server zkaddr + } + results := make(chan checkResult, len(servers)) for _, server := range servers { - dialer := net.Dialer{Timeout: rhp.connectivityCheckTimeout} - conn, err := dialer.DialContext(rhp.ctx, "tcp", server) - if err == nil { - conn.Close() - rhp.logger.Info().Str("server", server).Msg("zk connectivity check succeeded") - return nil + go func() { + var dialer net.Dialer + conn, err := dialer.DialContext(ctx, "tcp", server.addr) + if err == nil { + conn.Close() + } + results <- checkResult{server: server, err: err} + }() + } + + for range servers { + res := <-results + if res.err == nil { + rhp.logger.Info().Str("server", res.server.addr).Msg("zk connectivity check succeeded") + return res.server, nil } - rhp.logger.Error().Str("server", server).Err(err).Msg("connectivity check failed") + rhp.logger.Error().Str("server", res.server.addr).Err(res.err).Msg("connectivity check failed") } - return fmt.Errorf("failed to connect to any zk server: all attempts timed out or refused") + return zkaddr{}, fmt.Errorf("failed to connect to any zk server: all attempts timed out or refused") } func (rhp *RandomHostProvider) resolveHosts() { @@ -173,6 +201,16 @@ func (rhp *RandomHostProvider) Next() (server string, retryStart bool) { return rhp.hostsKeys[0], true } + if rhp.preferred.host != "" { + preferred := rhp.preferred + rhp.preferred = zkaddr{} + rhp.tried[preferred.host] = struct{}{} + if rhp.useAddrs { + return preferred.addr, false + } + return preferred.host, false + } + needRetry := false var ret string