ConnectAddress() was refactored

tengu-alt · tengu-alt · commit 6f8773abc84d · 2025-01-13T13:34:55.000+02:00
diff --git a/cassandra_test.go b/cassandra_test.go
@@ -32,7 +32,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/stretchr/testify/require"
 	"io"
 	"math"
 	"math/big"
@@ -45,7 +44,9 @@ import (
 	"time"
 	"unicode"
 
-	inf "gopkg.in/inf.v0"
+	"github.com/stretchr/testify/require"
+
+	"gopkg.in/inf.v0"
 )
 
 func TestEmptyHosts(t *testing.T) {
@@ -809,7 +810,11 @@ func TestReconnection(t *testing.T) {
 	defer session.Close()
 
 	h := session.ring.allHosts()[0]
-	session.handleNodeDown(h.ConnectAddress(), h.Port())
+	connAddr, err := h.ConnectAddress()
+	if err != nil {
+		t.Fatal(err)
+	}
+	session.handleNodeDown(connAddr, h.Port())
 
 	if h.State() != NodeDown {
 		t.Fatal("Host should be NodeDown but not.")
@@ -3086,7 +3091,11 @@ func TestDiscoverViaProxy(t *testing.T) {
 	for _, host := range clusterHosts {
 		found := false
 		for _, hi := range session.pool.hostConnPools {
-			if hi.host.ConnectAddress().String() == host {
+			connAddr, err := hi.host.ConnectAddress()
+			if err != nil {
+				t.Error(err)
+			}
+			if connAddr.String() == host {
 				found = true
 				break
 			}
diff --git a/conn.go b/conn.go
@@ -1697,7 +1697,11 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 		}
 
 		for _, row := range rows {
-			host, err := c.session.hostInfoFromMap(row, &HostInfo{connectAddress: c.host.ConnectAddress(), port: c.session.cfg.Port})
+			connAddr, err := c.host.ConnectAddress()
+			if err != nil {
+				goto cont
+			}
+			host, err := c.session.hostInfoFromMap(row, &HostInfo{connectAddress: connAddr, port: c.session.cfg.Port})
 			if err != nil {
 				goto cont
 			}
diff --git a/conn_test.go b/conn_test.go
@@ -350,7 +350,8 @@ type testQueryObserver struct {
 }
 
 func (o *testQueryObserver) ObserveQuery(ctx context.Context, q ObservedQuery) {
-	host := q.Host.ConnectAddress().String()
+	connAddr, _ := q.Host.ConnectAddress()
+	host := connAddr.String()
 	o.metrics[host] = q.Metrics
 	if o.verbose {
 		o.logger.Printf("Observed query %q. Returned %v rows, took %v on host %q with %v attempts and total latency %v. Error: %q\n",
@@ -359,7 +360,8 @@ func (o *testQueryObserver) ObserveQuery(ctx context.Context, q ObservedQuery) {
 }
 
 func (o *testQueryObserver) GetMetrics(host *HostInfo) *hostMetrics {
-	return o.metrics[host.ConnectAddress().String()]
+	connAddr, _ := host.ConnectAddress()
+	return o.metrics[connAddr.String()]
 }
 
 // TestQueryRetry will test to make sure that gocql will execute
diff --git a/connectionpool.go b/connectionpool.go
@@ -497,9 +497,11 @@ func (pool *hostConnPool) logConnectErr(err error) {
 
 // transition back to a not-filling state.
 func (pool *hostConnPool) fillingStopped(err error) {
+	connAddr, _ := pool.host.ConnectAddress()
+
 	if err != nil {
 		if gocqlDebug {
-			pool.logger.Printf("gocql: filling stopped %q: %v\n", pool.host.ConnectAddress(), err)
+			pool.logger.Printf("gocql: filling stopped %q: %v\n", connAddr, err)
 		}
 		// wait for some time to avoid back-to-back filling
 		// this provides some time between failed attempts
@@ -517,11 +519,11 @@ func (pool *hostConnPool) fillingStopped(err error) {
 	// if we errored and the size is now zero, make sure the host is marked as down
 	// see https://github.com/apache/cassandra-gocql-driver/issues/1614
 	if gocqlDebug {
-		pool.logger.Printf("gocql: conns of pool after stopped %q: %v\n", host.ConnectAddress(), count)
+		pool.logger.Printf("gocql: conns of pool after stopped %q: %v\n", connAddr, count)
 	}
 	if err != nil && count == 0 {
 		if pool.session.cfg.ConvictionPolicy.AddFailure(err, host) {
-			pool.session.handleNodeDown(host.ConnectAddress(), port)
+			pool.session.handleNodeDown(connAddr, port)
 		}
 	}
 }
@@ -575,8 +577,9 @@ func (pool *hostConnPool) connect() (err error) {
 			}
 		}
 		if gocqlDebug {
+			connAddr, _ := pool.host.ConnectAddress()
 			pool.logger.Printf("gocql: connection failed %q: %v, reconnecting with %T\n",
-				pool.host.ConnectAddress(), err, reconnectionPolicy)
+				connAddr, err, reconnectionPolicy)
 		}
 		time.Sleep(reconnectionPolicy.GetInterval(i))
 	}
diff --git a/control.go b/control.go
@@ -261,16 +261,21 @@ func (c *controlConn) connect(hosts []*HostInfo) error {
 	var conn *Conn
 	var err error
 	for _, host := range hosts {
+		connAddr, err := host.ConnectAddress()
+		if err != nil {
+			c.session.logger.Printf("gocql: %v\n", err)
+			continue
+		}
 		conn, err = c.session.dial(c.session.ctx, host, &cfg, c)
 		if err != nil {
-			c.session.logger.Printf("gocql: unable to dial control conn %v:%v: %v\n", host.ConnectAddress(), host.Port(), err)
+			c.session.logger.Printf("gocql: unable to dial control conn %v:%v: %v\n", connAddr, host.Port(), err)
 			continue
 		}
 		err = c.setupConn(conn)
 		if err == nil {
 			break
 		}
-		c.session.logger.Printf("gocql: unable setup control conn %v:%v: %v\n", host.ConnectAddress(), host.Port(), err)
+		c.session.logger.Printf("gocql: unable setup control conn %v:%v: %v\n", connAddr, host.Port(), err)
 		conn.Close()
 		conn = nil
 	}
@@ -300,9 +305,13 @@ func (c *controlConn) setupConn(conn *Conn) error {
 	}
 
 	host = c.session.ring.addOrUpdate(host)
+	connAddr, err := host.ConnectAddress()
+	if err != nil {
+		return err
+	}
 
 	if c.session.cfg.filterHost(host) {
-		return fmt.Errorf("host was filtered: %v", host.ConnectAddress())
+		return fmt.Errorf("host was filtered: %v", connAddr)
 	}
 
 	if err := c.registerEvents(conn); err != nil {
@@ -423,16 +432,21 @@ func (c *controlConn) attemptReconnectToAnyOfHosts(hosts []*HostInfo) (*Conn, er
 	var conn *Conn
 	var err error
 	for _, host := range hosts {
+		connAddr, err := host.ConnectAddress()
+		if err != nil {
+			c.session.logger.Printf("gocql: %v\n", err)
+			continue
+		}
 		conn, err = c.session.connect(c.session.ctx, host, c)
 		if err != nil {
-			c.session.logger.Printf("gocql: unable to dial control conn %v:%v: %v\n", host.ConnectAddress(), host.Port(), err)
+			c.session.logger.Printf("gocql: unable to dial control conn %v:%v: %v\n", connAddr, host.Port(), err)
 			continue
 		}
 		err = c.setupConn(conn)
 		if err == nil {
 			break
 		}
-		c.session.logger.Printf("gocql: unable setup control conn %v:%v: %v\n", host.ConnectAddress(), host.Port(), err)
+		c.session.logger.Printf("gocql: unable setup control conn %v:%v: %v\n", connAddr, host.Port(), err)
 		conn.Close()
 		conn = nil
 	}
diff --git a/control_ccm_test.go b/control_ccm_test.go
@@ -44,7 +44,8 @@ type TestHostFilter struct {
 func (f *TestHostFilter) Accept(h *HostInfo) bool {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	_, ok := f.allowedHosts[h.ConnectAddress().String()]
+	connAddr, _ := h.ConnectAddress()
+	_, ok := f.allowedHosts[connAddr.String()]
 	return ok
 }
 
@@ -100,7 +101,8 @@ func TestControlConn_ReconnectRefreshesRing(t *testing.T) {
 
 	var ccHostName string
 	for _, node := range allCcmHosts {
-		if node.Addr == ccHost.ConnectAddress().String() {
+		connAddr, _ := ccHost.ConnectAddress()
+		if node.Addr == connAddr.String() {
 			ccHostName = node.Name
 			break
 		}
diff --git a/control_test.go b/control_test.go
@@ -49,8 +49,13 @@ func TestHostInfo_Lookup(t *testing.T) {
 		}
 
 		host := hosts[0]
-		if !host.ConnectAddress().Equal(test.ip) {
-			t.Errorf("expected ip %v got %v for addr %q", test.ip, host.ConnectAddress(), test.addr)
+		connAddr, err := host.ConnectAddress()
+		if err != nil {
+			t.Errorf("%d: %v", i, err)
+			continue
+		}
+		if !connAddr.Equal(test.ip) {
+			t.Errorf("expected ip %v got %v for addr %q", test.ip, connAddr, test.addr)
 		}
 	}
 }
diff --git a/dial.go b/dial.go
@@ -60,7 +60,10 @@ type defaultHostDialer struct {
 }
 
 func (hd *defaultHostDialer) DialHost(ctx context.Context, host *HostInfo) (*DialedHost, error) {
-	ip := host.ConnectAddress()
+	ip, err := host.ConnectAddress()
+	if err != nil {
+		return nil, err
+	}
 	port := host.Port()
 
 	if !validIpAddr(ip) {
diff --git a/events.go b/events.go
@@ -241,7 +241,8 @@ func (s *Session) startPoolFill(host *HostInfo) {
 
 func (s *Session) handleNodeConnected(host *HostInfo) {
 	if gocqlDebug {
-		s.logger.Printf("gocql: Session.handleNodeConnected: %s:%d\n", host.ConnectAddress(), host.Port())
+		connAddr, _ := host.ConnectAddress()
+		s.logger.Printf("gocql: Session.handleNodeConnected: %s:%d\n", connAddr, host.Port())
 	}
 
 	host.setState(NodeUp)
diff --git a/filters.go b/filters.go
@@ -72,10 +72,13 @@ func WhiteListHostFilter(hosts ...string) HostFilter {
 
 	m := make(map[string]bool, len(hostInfos))
 	for _, host := range hostInfos {
-		m[host.ConnectAddress().String()] = true
+		connAddr, _ := host.ConnectAddress()
+		m[connAddr.String()] = true
 	}
 
 	return HostFilterFunc(func(host *HostInfo) bool {
-		return m[host.ConnectAddress().String()]
+		connAddr, _ := host.ConnectAddress()
+		return m[connAddr.String()]
+
 	})
 }
diff --git a/host_source.go b/host_source.go
@@ -164,8 +164,9 @@ func (h *HostInfo) Equal(host *HostInfo) bool {
 		// prevent rlock reentry
 		return true
 	}
-
-	return h.ConnectAddress().Equal(host.ConnectAddress())
+	hConnAddr, _ := h.ConnectAddress()
+	hostConnAddr, _ := host.ConnectAddress()
+	return hConnAddr.Equal(hostConnAddr)
 }
 
 func (h *HostInfo) Peer() net.IP {
@@ -220,14 +221,14 @@ func (h *HostInfo) nodeToNodeAddress() net.IP {
 // Returns the address that should be used to connect to the host.
 // If you wish to override this, use an AddressTranslator or
 // use a HostFilter to SetConnectAddress()
-func (h *HostInfo) ConnectAddress() net.IP {
+func (h *HostInfo) ConnectAddress() (net.IP, error) {
 	h.mu.RLock()
 	defer h.mu.RUnlock()
 
 	if addr, _ := h.connectAddressLocked(); validIpAddr(addr) {
-		return addr
+		return addr, nil
 	}
-	panic(fmt.Sprintf("no valid connect address for host: %v. Is your cluster configured correctly?", h))
+	return nil, fmt.Errorf("no valid connect address for host: %v. Is your cluster configured correctly?", h)
 }
 
 func (h *HostInfo) SetConnectAddress(address net.IP) *HostInfo {
@@ -582,8 +583,12 @@ func (s *Session) hostInfoFromMap(row map[string]interface{}, host *HostInfo) (*
 		// TODO(thrawn01): Add 'port'? once CASSANDRA-7544 is complete
 		// Not sure what the port field will be called until the JIRA issue is complete
 	}
+	connAddr, err := host.ConnectAddress()
+	if err != nil {
+		return nil, err
+	}
 
-	ip, port := s.cfg.translateAddressPort(host.ConnectAddress(), host.port)
+	ip, port := s.cfg.translateAddressPort(connAddr, host.port)
 	host.connectAddress = ip
 	host.port = port
 
diff --git a/host_source_test.go b/host_source_test.go
@@ -121,7 +121,7 @@ func TestHostInfo_ConnectAddress(t *testing.T) {
 				peer:             test.peer,
 			}
 
-			if addr := host.ConnectAddress(); !addr.Equal(localhost) {
+			if addr, _ := host.ConnectAddress(); !addr.Equal(localhost) {
 				t.Fatalf("expected ConnectAddress to be %s got %s", localhost, addr)
 			}
 		})
diff --git a/integration_test.go b/integration_test.go
@@ -95,7 +95,11 @@ func TestRingDiscovery(t *testing.T) {
 
 	if *clusterSize != size {
 		for p, pool := range session.pool.hostConnPools {
-			t.Logf("p=%q host=%v ips=%s", p, pool.host, pool.host.ConnectAddress().String())
+			connAddr, err := pool.host.ConnectAddress()
+			if err != nil {
+				t.Error(err)
+			}
+			t.Logf("p=%q host=%v ips=%s", p, pool.host, connAddr.String())
 
 		}
 		t.Errorf("Expected a cluster size of %d, but actual size was %d", *clusterSize, size)
@@ -115,7 +119,11 @@ func TestHostFilterDiscovery(t *testing.T) {
 	filtered := clusterHosts[1]
 	cluster.Hosts = clusterHosts[:1]
 	cluster.HostFilter = HostFilterFunc(func(host *HostInfo) bool {
-		if host.ConnectAddress().String() == filtered {
+		connAddr, err := host.ConnectAddress()
+		if err != nil {
+			t.Error(err)
+		}
+		if connAddr.String() == filtered {
 			return false
 		}
 		return true
@@ -139,7 +147,11 @@ func TestHostFilterInitial(t *testing.T) {
 	// we'll filter out the second host
 	filtered := clusterHosts[1]
 	cluster.HostFilter = HostFilterFunc(func(host *HostInfo) bool {
-		if host.ConnectAddress().String() == filtered {
+		connAddr, err := host.ConnectAddress()
+		if err != nil {
+			t.Error(err)
+		}
+		if connAddr.String() == filtered {
 			return false
 		}
 		return true
diff --git a/policies.go b/policies.go
diff --git a/policies_test.go b/policies_test.go
diff --git a/session.go b/session.go
diff --git a/token.go b/token.go
diff --git a/token_test.go b/token_test.go
diff --git a/topology.go b/topology.go

Original file line number	Diff line number	Diff line change
`@@ -1697,7 +1697,11 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {`
`1697`	`1697`	`}`
`1698`	`1698`
`1699`	`1699`	`for _, row := range rows {`
`1700`		`- host, err := c.session.hostInfoFromMap(row, &HostInfo{connectAddress: c.host.ConnectAddress(), port: c.session.cfg.Port})`
	`1700`	`+ connAddr, err := c.host.ConnectAddress()`
	`1701`	`+ if err != nil {`
	`1702`	`+ goto cont`
	`1703`	`+ }`
	`1704`	`+ host, err := c.session.hostInfoFromMap(row, &HostInfo{connectAddress: connAddr, port: c.session.cfg.Port})`
`1701`	`1705`	`if err != nil {`
`1702`	`1706`	`goto cont`
`1703`	`1707`	`}`
Original file line number	Diff line number	Diff line change
`@@ -350,7 +350,8 @@ type testQueryObserver struct {`
`350`	`350`	`}`
`351`	`351`
`352`	`352`	`func (o *testQueryObserver) ObserveQuery(ctx context.Context, q ObservedQuery) {`
`353`		`- host := q.Host.ConnectAddress().String()`
	`353`	`+ connAddr, _ := q.Host.ConnectAddress()`
	`354`	`+ host := connAddr.String()`
`354`	`355`	`o.metrics[host] = q.Metrics`
`355`	`356`	`if o.verbose {`
`356`	`357`	`o.logger.Printf("Observed query %q. Returned %v rows, took %v on host %q with %v attempts and total latency %v. Error: %q\n",`
`@@ -359,7 +360,8 @@ func (o *testQueryObserver) ObserveQuery(ctx context.Context, q ObservedQuery) {`
`359`	`360`	`}`
`360`	`361`
`361`	`362`	`func (o testQueryObserver) GetMetrics(host HostInfo) *hostMetrics {`
`362`		`- return o.metrics[host.ConnectAddress().String()]`
	`363`	`+ connAddr, _ := host.ConnectAddress()`
	`364`	`+ return o.metrics[connAddr.String()]`
`363`	`365`	`}`
`364`	`366`
`365`	`367`	`// TestQueryRetry will test to make sure that gocql will execute`
Original file line number	Diff line number	Diff line change
`@@ -497,9 +497,11 @@ func (pool *hostConnPool) logConnectErr(err error) {`
`497`	`497`
`498`	`498`	`// transition back to a not-filling state.`
`499`	`499`	`func (pool *hostConnPool) fillingStopped(err error) {`
	`500`	`+ connAddr, _ := pool.host.ConnectAddress()`
	`501`	`+`
`500`	`502`	`if err != nil {`
`501`	`503`	`if gocqlDebug {`
`502`		`- pool.logger.Printf("gocql: filling stopped %q: %v\n", pool.host.ConnectAddress(), err)`
	`504`	`+ pool.logger.Printf("gocql: filling stopped %q: %v\n", connAddr, err)`
`503`	`505`	`}`
`504`	`506`	`// wait for some time to avoid back-to-back filling`
`505`	`507`	`// this provides some time between failed attempts`
`@@ -517,11 +519,11 @@ func (pool *hostConnPool) fillingStopped(err error) {`
`517`	`519`	`// if we errored and the size is now zero, make sure the host is marked as down`
`518`	`520`	`// see https://github.com/apache/cassandra-gocql-driver/issues/1614`
`519`	`521`	`if gocqlDebug {`
`520`		`- pool.logger.Printf("gocql: conns of pool after stopped %q: %v\n", host.ConnectAddress(), count)`
	`522`	`+ pool.logger.Printf("gocql: conns of pool after stopped %q: %v\n", connAddr, count)`
`521`	`523`	`}`
`522`	`524`	`if err != nil && count == 0 {`
`523`	`525`	`if pool.session.cfg.ConvictionPolicy.AddFailure(err, host) {`
`524`		`- pool.session.handleNodeDown(host.ConnectAddress(), port)`
	`526`	`+ pool.session.handleNodeDown(connAddr, port)`
`525`	`527`	`}`
`526`	`528`	`}`
`527`	`529`	`}`
`@@ -575,8 +577,9 @@ func (pool *hostConnPool) connect() (err error) {`
`575`	`577`	`}`
`576`	`578`	`}`
`577`	`579`	`if gocqlDebug {`
	`580`	`+ connAddr, _ := pool.host.ConnectAddress()`
`578`	`581`	`pool.logger.Printf("gocql: connection failed %q: %v, reconnecting with %T\n",`
`579`		`- pool.host.ConnectAddress(), err, reconnectionPolicy)`
	`582`	`+ connAddr, err, reconnectionPolicy)`
`580`	`583`	`}`
`581`	`584`	`time.Sleep(reconnectionPolicy.GetInterval(i))`
`582`	`585`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,8 +49,13 @@ func TestHostInfo_Lookup(t *testing.T) {`
`49`	`49`	`}`
`50`	`50`
`51`	`51`	`host := hosts[0]`
`52`		`- if !host.ConnectAddress().Equal(test.ip) {`
`53`		`- t.Errorf("expected ip %v got %v for addr %q", test.ip, host.ConnectAddress(), test.addr)`
	`52`	`+ connAddr, err := host.ConnectAddress()`
	`53`	`+ if err != nil {`
	`54`	`+ t.Errorf("%d: %v", i, err)`
	`55`	`+ continue`
	`56`	`+ }`
	`57`	`+ if !connAddr.Equal(test.ip) {`
	`58`	`+ t.Errorf("expected ip %v got %v for addr %q", test.ip, connAddr, test.addr)`
`54`	`59`	`}`
`55`	`60`	`}`
`56`	`61`	`}`
Original file line number	Diff line number	Diff line change
`@@ -241,7 +241,8 @@ func (s Session) startPoolFill(host HostInfo) {`
`241`	`241`
`242`	`242`	`func (s Session) handleNodeConnected(host HostInfo) {`
`243`	`243`	`if gocqlDebug {`
`244`		`- s.logger.Printf("gocql: Session.handleNodeConnected: %s:%d\n", host.ConnectAddress(), host.Port())`
	`244`	`+ connAddr, _ := host.ConnectAddress()`
	`245`	`+ s.logger.Printf("gocql: Session.handleNodeConnected: %s:%d\n", connAddr, host.Port())`
`245`	`246`	`}`
`246`	`247`
`247`	`248`	`host.setState(NodeUp)`
Original file line number	Diff line number	Diff line change
`@@ -72,10 +72,13 @@ func WhiteListHostFilter(hosts ...string) HostFilter {`
`72`	`72`
`73`	`73`	`m := make(map[string]bool, len(hostInfos))`
`74`	`74`	`for _, host := range hostInfos {`
`75`		`- m[host.ConnectAddress().String()] = true`
	`75`	`+ connAddr, _ := host.ConnectAddress()`
	`76`	`+ m[connAddr.String()] = true`
`76`	`77`	`}`
`77`	`78`
`78`	`79`	`return HostFilterFunc(func(host *HostInfo) bool {`
`79`		`- return m[host.ConnectAddress().String()]`
	`80`	`+ connAddr, _ := host.ConnectAddress()`
	`81`	`+ return m[connAddr.String()]`
	`82`	`+`
`80`	`83`	`})`
`81`	`84`	`}`