feat: add domain re-resolution support for Go client

go-client/admin/client_test.go

@@ -51,7 +51,7 @@ func defaultReplicaServerPorts() []int {
 
 func timeoutConfig() Config {
 	return Config{
-		MetaServers: []string{"0.0.0.0:123456"},
+		MetaServers: []string{"0.0.0.0:12345"},
 		Timeout:     500 * time.Millisecond,
 	}
 }

go-client/pegasus/client.go

@@ -64,7 +64,7 @@ func NewClient(cfg config.Config) Client {
 
 func newClientWithError(cfg config.Config) (Client, error) {
 	var err error
-	cfg.MetaServers, err = session.ResolveMetaAddr(cfg.MetaServers)
+	_, err = session.ResolveMetaAddr(cfg.MetaServers)
 	if err != nil {
 		return nil, err
 	}

go-client/session/addr.go

@@ -24,9 +24,8 @@ import (
 	"net"
 )
 
-// ResolveMetaAddr into a list of TCP4 addresses. Error is returned if the given `addrs` are not either
-// a list of valid TCP4 addresses, or a resolvable hostname.
-func ResolveMetaAddr(addrs []string) ([]string, error) {
+// resolveMetaAddrImpl is the actual implementation of ResolveMetaAddr
+func resolveMetaAddrImpl(addrs []string) ([]string, error) {
 	if len(addrs) == 0 {
 		return nil, fmt.Errorf("meta server list should not be empty")
 	}
@@ -54,3 +53,8 @@ func ResolveMetaAddr(addrs []string) ([]string, error) {
 
 	return nil, fmt.Errorf("illegal meta addresses: %s", addrs)
 }
+
+// ResolveMetaAddr into a list of TCP4 addresses. Error is returned if the given `addrs` are not either
+// a list of valid TCP4 addresses, or a resolvable hostname.
+// This is a function variable so it can be mocked in tests.
+var ResolveMetaAddr = resolveMetaAddrImpl

go-client/session/meta_call.go

@@ -27,7 +27,6 @@ import (
 
 	"github.com/apache/incubator-pegasus/go-client/idl/base"
 	"github.com/apache/incubator-pegasus/go-client/idl/replication"
-	"github.com/apache/incubator-pegasus/go-client/pegalog"
 )
 
 type metaCallFunc func(context.Context, *metaSession) (metaResponse, error)
@@ -127,28 +126,15 @@ func (c *metaCall) issueSingleMeta(ctx context.Context, curLeader int) bool {
 		}
 
 		addr := forwardAddr.GetAddress()
-		found := false
 		c.lock.Lock()
-		for i := range c.metaIPAddrs {
-			if addr == c.metaIPAddrs[i] {
-				found = true
-				break
-			}
-		}
-		c.lock.Unlock()
-
-		if !found {
-			c.lock.Lock()
-			c.metaIPAddrs = append(c.metaIPAddrs, addr)
-			c.metas = append(c.metas, &metaSession{
-				NodeSession: newNodeSession(addr, NodeTypeMeta, DisableMetrics),
-				logger:      pegalog.GetLogger(),
-			})
+		found := addMetaSession(&c.metaIPAddrs, &c.metas, addr)
+		if found {
 			curLeader = len(c.metas) - 1
 			meta = c.metas[curLeader]
 			meta.logger.Printf("add forward address %s as meta server", addr)
-			c.lock.Unlock()
-
+		}
+		c.lock.Unlock()
+		if found {
 			resp, err = c.callFunc(ctx, meta)
 		}
 	}

go-client/session/meta_session.go

@@ -21,6 +21,8 @@ package session
 
 import (
 	"context"
+	"net"
+	"strings"
 	"sync"
 
 	"github.com/apache/incubator-pegasus/go-client/idl/base"
@@ -65,6 +67,7 @@ type MetaManager struct {
 	logger pegalog.Logger
 
 	metaIPAddrs   []string
+	metaDomain    string
 	metas         []*metaSession
 	currentLeader int // current leader of meta servers
 
@@ -73,6 +76,16 @@ type MetaManager struct {
 }
 
 func NewMetaManager(addrs []string, creator NodeSessionCreator) *MetaManager {
+	metaHostAddr := ""
+	if len(addrs) == 1 && isDomain(addrs[0]) {
+		metaHostAddr = addrs[0]
+	}
+
+	addrs, err := ResolveMetaAddr(addrs)
+	if err != nil {
+		return nil
+	}
+
 	metas := make([]*metaSession, len(addrs))
 	metaIPAddrs := make([]string, len(addrs))
 	for i, addr := range addrs {
@@ -87,11 +100,30 @@ func NewMetaManager(addrs []string, creator NodeSessionCreator) *MetaManager {
 		currentLeader: 0,
 		metas:         metas,
 		metaIPAddrs:   metaIPAddrs,
+		metaDomain:    metaHostAddr,
 		logger:        pegalog.GetLogger(),
 	}
 	return mm
 }
 
+func isDomain(s string) bool {
+	host := s
+	if idx := strings.LastIndex(host, ":"); idx != -1 {
+		host = host[:idx]
+	}
+
+	if ip := net.ParseIP(host); ip != nil {
+		return false
+	}
+
+	ips, err := net.LookupIP(host)
+	if err != nil || len(ips) == 0 {
+		return false
+	}
+
+	return true
+}
+
 func (m *MetaManager) call(ctx context.Context, callFunc metaCallFunc) (metaResponse, error) {
 	call := newMetaCall(m.getCurrentLeader(), m.metas, callFunc, m.metaIPAddrs)
 	resp, err := call.Run(ctx)
@@ -101,10 +133,47 @@ func (m *MetaManager) call(ctx context.Context, callFunc metaCallFunc) (metaResp
 		m.setNewMetas(call.metas)
 		m.setMetaIPAddrs(call.metaIPAddrs)
 		call.lock.RUnlock()
+	} else if m.metaDomain != "" {
+		if newAddrs, err := ResolveMetaAddr([]string{m.metaDomain}); err == nil {
+			m.mu.Lock()
+			m.logger.Printf("resolved meta list %s to %v", m.metaDomain, newAddrs)
+			anyNew := false
+			for _, addr := range newAddrs {
+				if addMetaSession(&m.metaIPAddrs, &m.metas, addr) {
+					anyNew = true
+				}
+			}
+			if anyNew {
+				m.currentLeader = len(m.metas) - 1
+			}
+			m.mu.Unlock()
+
+		}
 	}
 	return resp, err
 }
 
+func addMetaSession(metaIPAddrs *[]string, metas *[]*metaSession, addr string) bool {
+	found := false
+	for _, oldAddr := range *metaIPAddrs {
+		if oldAddr == addr {
+			found = true
+			break
+		}
+	}
+
+	if !found {
+		*metaIPAddrs = append(*metaIPAddrs, addr)
+		*metas = append(*metas, &metaSession{
+			NodeSession: newNodeSession(addr, NodeTypeMeta, DisableMetrics),
+			logger:      pegalog.GetLogger(),
+		})
+		return true
+	}
+
+	return false
+}
+
 // QueryConfig queries table configuration from the leader of meta servers. If the leader was changed,
 // it retries for other servers until it finds the true leader, unless no leader exists.
 // Thread-Safe

go-client/session/meta_session_test.go

@@ -21,6 +21,7 @@ package session
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"testing"
 	"time"
@@ -142,3 +143,79 @@ func TestNodeSession_ForwardToPrimaryMeta(t *testing.T) {
 		assert.Equal(t, resp.Err.Errno, base.ERR_OK.String())
 	}
 }
+
+// TestMetaManager_DNSResolveHost tests that MetaManager can resolve meta IP addresses from a domain name.
+func TestMetaManager_DNSResolveHost(t *testing.T) {
+	defer leaktest.Check(t)()
+
+	originalResolveMetaAddr := ResolveMetaAddr
+	defer func() {
+		ResolveMetaAddr = originalResolveMetaAddr
+	}()
+
+	// Mock the ResolveMetaAddr function
+	ResolveMetaAddr = func(hosts []string) ([]string, error) {
+		return []string{"127.0.0.1:34601", "127.0.0.1:34602", "127.0.0.1:34603"}, nil
+	}
+
+	// Create a MetaManager with a domain name
+	mm := NewMetaManager([]string{"localhost:34601"}, NewNodeSession)
+	defer mm.Close()
+
+	// Verify initial resolution
+	assert.Equal(t, []string{"127.0.0.1:34601", "127.0.0.1:34602", "127.0.0.1:34603"}, mm.GetMetaIPAddrs())
+
+	_, err := mm.QueryConfig(context.Background(), "temp")
+	assert.Nil(t, err)
+}
+
+// TestMetaManager_DNSMetaAllChanged tests that MetaManager re-resolves DNS and updates meta servers when all meta servers fail.
+func TestMetaManager_DNSMetaAllChanged(t *testing.T) {
+	defer leaktest.Check(t)()
+
+	// Save original ResolveMetaAddr function
+	originalResolveMetaAddr := ResolveMetaAddr
+	defer func() {
+		ResolveMetaAddr = originalResolveMetaAddr
+	}()
+
+	// Mock the ResolveMetaAddr function to return addresses（unreachable ones）
+	ResolveMetaAddr = func(hosts []string) ([]string, error) {
+		return []string{"172.0.0.1:34601", "172.0.0.2:34601"}, nil
+	}
+
+	// Create a MetaManager with a domain name
+	mm := NewMetaManager([]string{"localhost:34601"}, NewNodeSession)
+	defer mm.Close()
+
+	// Verify initial resolution
+	assert.Equal(t, []string{"172.0.0.1:34601", "172.0.0.2:34601"}, mm.GetMetaIPAddrs())
+
+	// Create a context with timeout to trigger failure
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	// This should timeout because the initial meta servers are unreachable
+	_, err := mm.QueryConfig(ctx, "temp")
+	assert.NotNil(t, err)
+	assert.True(t, errors.Is(context.DeadlineExceeded, ctx.Err()))
+
+	// Mock the ResolveMetaAddr function to return new addresses（reachable ones）
+	ResolveMetaAddr = func(hosts []string) ([]string, error) {
+		return []string{"127.0.0.1:34601", "127.0.0.1:34602", "127.0.0.1:34603"}, nil
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	// This should also timeout because the initial meta servers are still unreachable, but it will update the meta servers after timeout.
+	_, err = mm.QueryConfig(ctx, "temp")
+	assert.NotNil(t, err)
+	assert.True(t, errors.Is(context.DeadlineExceeded, ctx.Err()))
+
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	// This should not timeout because meta servers are reachable now.
+	_, err = mm.QueryConfig(ctx, "temp")
+	assert.Nil(t, err)
+}