Improve startup time by using first alive zk

.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

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/...

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,15 +57,18 @@ 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)
 		if err != nil {
 			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