diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 440aaa75a..44d7bd529 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -8,8 +8,8 @@ on: types: [ opened, synchronize, reopened ] jobs: - build: - name: Unit tests + unit-tests-amd64: + name: Unit tests AMD64 runs-on: ubuntu-latest strategy: matrix: @@ -23,11 +23,28 @@ jobs: run: make check - name: Run unit tests run: make test-unit + + unit-tests-arm64: + name: Unit tests ARM64 + runs-on: ubuntu-24.04-arm + strategy: + matrix: + go: [ '1.25', '1.26' ] + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-go@v6 + with: + go-version: ${{ matrix.go }} + - name: Run linting + run: make check + - name: Run unit tests + run: make test-unit integration-cassandra: timeout-minutes: 15 needs: - - build + - unit-tests-amd64 + - unit-tests-arm64 name: Integration Tests runs-on: ubuntu-latest strategy: @@ -93,7 +110,8 @@ jobs: integration-auth-cassandra: timeout-minutes: 15 needs: - - build + - unit-tests-amd64 + - unit-tests-arm64 name: Integration Tests with auth runs-on: ubuntu-latest strategy: diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 000000000..78d4c9cfc --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,16 @@ +# Agent guidance + +This file is read by automated agents (security scanners, code analyzers, +AI assistants) operating on this repository. It points them at the +human-authored references they should consult before producing output. + +## Security + +Security model: [SECURITY.md](./SECURITY.md), which links to the Apache +Cassandra project security model. + +This repository is part of the Apache Cassandra project. Its security model - +trust boundaries, in-scope / out-of-scope declarations, the security +properties the project provides and disclaims, and how findings are triaged - +is the umbrella Cassandra security model linked from SECURITY.md. Consult it +before reporting security issues. diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b07b9bfe..0b6f10457 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,27 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.2.0] + +### Added + +- Security-model discoverability (CASSANDRA-21464) +- Improve host_source locking and ring refresh concurrency (CASSGO-121) +- Add PreparedMetadata (Keyspace, Table) and IsPrepared fields to ObservedQuery, and parallel PreparedMetadata / IsPrepared slices to ObservedBatch, for statement-level observability without CQL parsing (CASSGO-119) +- Query.Binding() method to override binding function for a query object. + +### Fixed +- Correct protocol negotiation with non-Cassandra servers (CASSGO-131) +- LZ4 test fails on arm64 (CASSGO-128) + +## [2.1.2] + +### Fixed + +- Prevent panic when using a HostFilter and keyspace is not replicated to every DC (CASSGO-122) +- system.peers fallback doesn't work in some scenarios (CASSGO-126) +- Many "Pool connection error" with small Session.Timeout (CASSGO-125) + ## [2.1.1] ### Fixed diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 000000000..7ecb4e420 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,17 @@ +# Security Policy + +## Reporting a Vulnerability + +Please report suspected security vulnerabilities privately to the Apache +Security Team at , following the ASF process at +. Do not open public GitHub issues or pull +requests for security reports. + +## Security Model + +This repository is part of the Apache Cassandra project. The project's +security model - what is in and out of scope, the trust boundaries it +assumes, the security properties it provides and disclaims, and how findings +are triaged - is documented in the main apache/cassandra repository: + +https://github.com/apache/cassandra/blob/trunk/doc/modules/cassandra/pages/reference/security-model.adoc diff --git a/auth.go b/auth.go new file mode 100644 index 000000000..cb49c0985 --- /dev/null +++ b/auth.go @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package gocql + +import ( + "fmt" + "sync" +) + +// AuthRegistry is the interface for a registry of authenticators to support authentication negotiation between the client and the server (CEP-50). +type AuthRegistry interface { + // Returns the list of all registered authenticators. + Authenticators() []NegotiableAuthenticator + + // Registers a new authenticator. + // + // authenticator is the NegotiableAuthenticator to register. + Register(authenticator NegotiableAuthenticator) + + // Returns the authenticator for the given class name. + // If the authenticator is not found, returns false. + AuthenticatorFor(className string) (NegotiableAuthenticator, bool) +} + +// The default implementation of the AuthRegistry interface. +type defaultAuthRegistry struct { + mu sync.RWMutex + // Map of Java class name to authenticator. + byClassName map[string]NegotiableAuthenticator + // List of all registered authenticators. Used for iteration. + all []NegotiableAuthenticator +} + +// Creates a new default implementation of the AuthRegistry interface. +func NewDefaultAuthRegistry() AuthRegistry { + return &defaultAuthRegistry{ + byClassName: make(map[string]NegotiableAuthenticator), + all: make([]NegotiableAuthenticator, 0), + } +} + +func (r *defaultAuthRegistry) Register(authenticator NegotiableAuthenticator) { + r.mu.Lock() + defer r.mu.Unlock() + r.mustRegisterLocked(authenticator) +} + +// Registers the authenticator with the given class name. +// Panics if the authenticator is already registered. +// Must be called with the lock held. +func (r *defaultAuthRegistry) mustRegisterLocked(authenticator NegotiableAuthenticator) { + className := authenticator.ClassName() + if _, ok := r.byClassName[className]; ok { + panic(fmt.Sprintf("gocql: authenticator %s already registered", className)) + } + r.byClassName[className] = authenticator + r.all = append(r.all, authenticator) +} + +// Implements the AuthRegistry interface. Returns a copy of the list of all registered authenticators. +func (r *defaultAuthRegistry) Authenticators() []NegotiableAuthenticator { + r.mu.RLock() + defer r.mu.RUnlock() + return append([]NegotiableAuthenticator(nil), r.all...) +} + +// Returns the authenticator for the given class name. +// If the authenticator is not found, returns false. +func (r *defaultAuthRegistry) AuthenticatorFor(className string) (NegotiableAuthenticator, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + authenticator, ok := r.byClassName[className] + return authenticator, ok +} diff --git a/cassandra_test.go b/cassandra_test.go index 2f386c801..eda3eb9e9 100644 --- a/cassandra_test.go +++ b/cassandra_test.go @@ -172,9 +172,12 @@ func TestObserve(t *testing.T) { } var ( - observedErr error - observedKeyspace string - observedStmt string + observedErr error + observedKeyspace string + observedPreparedKeyspace string + observedStmt string + observedPreparedTable string + observedIsPrepared bool ) const keyspace = "gocql_test" @@ -182,12 +185,18 @@ func TestObserve(t *testing.T) { resetObserved := func() { observedErr = errors.New("placeholder only") // used to distinguish err=nil cases observedKeyspace = "" + observedPreparedKeyspace = "" observedStmt = "" + observedPreparedTable = "" + observedIsPrepared = false } observer := funcQueryObserver(func(ctx context.Context, o ObservedQuery) { observedKeyspace = o.Keyspace + observedPreparedKeyspace = o.PreparedMetadata.Keyspace observedStmt = o.Statement + observedPreparedTable = o.PreparedMetadata.Table + observedIsPrepared = o.IsPrepared observedErr = o.Err }) @@ -202,6 +211,12 @@ func TestObserve(t *testing.T) { t.Fatal("select: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `SELECT id FROM observe WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("select: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("select: unexpected observed prepared table", observedPreparedTable) } resetObserved() @@ -213,6 +228,12 @@ func TestObserve(t *testing.T) { t.Fatal("insert: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `INSERT INTO observe (id) VALUES (?)` { t.Fatal("insert: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("insert: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("insert: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("insert: unexpected observed prepared table", observedPreparedTable) } resetObserved() @@ -227,6 +248,12 @@ func TestObserve(t *testing.T) { t.Fatal("select: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `SELECT id FROM observe WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("select: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("select: unexpected observed prepared table", observedPreparedTable) } // also works from session observer @@ -240,6 +267,12 @@ func TestObserve(t *testing.T) { t.Fatal("select: unexpected observed keyspace", observedKeyspace) } else if observedStmt != `SELECT id FROM observe WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) + } else if !observedIsPrepared { + t.Fatal("select: expected observed IsPrepared to be true") + } else if observedPreparedKeyspace != keyspace { + t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace) + } else if observedPreparedTable != "observe" { + t.Fatal("select: unexpected observed prepared table", observedPreparedTable) } // reports errors when the query is poorly formed @@ -254,6 +287,24 @@ func TestObserve(t *testing.T) { } else if observedStmt != `SELECT id FROM unknown_table WHERE id = ?` { t.Fatal("select: unexpected observed stmt", observedStmt) } + + // statements that cannot be prepared (e.g. DDL) report IsPrepared=false and + // a zero PreparedMetadata, so the contract is explicit for observers. + resetObserved() + const ddlStmt = `CREATE TABLE IF NOT EXISTS gocql_test.observe_unprepared (id int primary key)` + if err := session.Query(ddlStmt).Observer(observer).Exec(); err != nil { + t.Fatal("ddl:", err) + } else if observedErr != nil { + t.Fatal("ddl:", observedErr) + } else if observedStmt != ddlStmt { + t.Fatal("ddl: unexpected observed stmt", observedStmt) + } else if observedIsPrepared { + t.Fatal("ddl: expected observed IsPrepared to be false for unprepared statement") + } else if observedPreparedKeyspace != "" { + t.Fatal("ddl: expected zero PreparedMetadata.Keyspace, got", observedPreparedKeyspace) + } else if observedPreparedTable != "" { + t.Fatal("ddl: expected zero PreparedMetadata.Table, got", observedPreparedTable) + } } func TestObserve_Pagination(t *testing.T) { @@ -2134,10 +2185,12 @@ func TestBatchObserve(t *testing.T) { } type observation struct { - observedErr error - observedKeyspace string - observedStmts []string - observedValues [][]interface{} + observedErr error + observedKeyspace string + observedPreparedMetadata []PreparedMetadata + observedStmts []string + observedIsPrepared []bool + observedValues [][]interface{} } var observedBatch *observation @@ -2149,10 +2202,12 @@ func TestBatchObserve(t *testing.T) { } observedBatch = &observation{ - observedKeyspace: o.Keyspace, - observedStmts: o.Statements, - observedErr: o.Err, - observedValues: o.Values, + observedKeyspace: o.Keyspace, + observedPreparedMetadata: o.PreparedMetadata, + observedStmts: o.Statements, + observedIsPrepared: o.IsPrepared, + observedErr: o.Err, + observedValues: o.Values, } })) for i := 0; i < 100; i++ { @@ -2175,11 +2230,29 @@ func TestBatchObserve(t *testing.T) { if observedBatch.observedKeyspace != "gocql_test" { t.Fatalf("expecting keyspace 'gocql_test', got %q", observedBatch.observedKeyspace) } + if len(observedBatch.observedPreparedMetadata) != 100 { + t.Fatal("expecting 100 observed prepared metadata entries, got", len(observedBatch.observedPreparedMetadata)) + } + if len(observedBatch.observedIsPrepared) != 100 { + t.Fatal("expecting 100 observed IsPrepared flags, got", len(observedBatch.observedIsPrepared)) + } for i, stmt := range observedBatch.observedStmts { if stmt != fmt.Sprintf(`INSERT INTO batch_observe_table (id,other) VALUES (?,%d)`, i) { t.Fatal("unexpected query", stmt) } + if !observedBatch.observedIsPrepared[i] { + t.Fatalf("expected observed IsPrepared at index %d to be true", i) + } + + if observedBatch.observedPreparedMetadata[i].Keyspace != "gocql_test" { + t.Fatalf("unexpected observed prepared keyspace at index %d: %q", i, observedBatch.observedPreparedMetadata[i].Keyspace) + } + + if observedBatch.observedPreparedMetadata[i].Table != "batch_observe_table" { + t.Fatalf("unexpected observed prepared table at index %d: %q", i, observedBatch.observedPreparedMetadata[i].Table) + } + assertDeepEqual(t, "observed value", []interface{}{i}, observedBatch.observedValues[i]) } } diff --git a/cluster.go b/cluster.go index 0c687d20d..f334aba4d 100644 --- a/cluster.go +++ b/cluster.go @@ -165,6 +165,10 @@ type ClusterConfig struct { // Default: nil AuthProvider func(h *HostInfo) (Authenticator, error) + // AuthRegistry is the registry of authenticators to support authentication negotiation between the client and the server (CEP-50). + // Default: nil + AuthRegistry AuthRegistry + // Default retry policy to use for queries. // Default: no retries. RetryPolicy RetryPolicy diff --git a/conn.go b/conn.go index a615129da..b1141ac25 100644 --- a/conn.go +++ b/conn.go @@ -74,6 +74,23 @@ type Authenticator interface { Success(data []byte) error } +// NegotiableAuthenticator is an authenticator that can be negotiated with the server. +// It is an optional interface that can be implemented by the authenticator to allow for negotiation with the server. +// If the authenticator implements this interface, it will be used to negotiate the authenticator with the server. +type NegotiableAuthenticator interface { + Authenticator + + // Java class name of the authenticator. Used to match the authenticator with the one chosen by the server for authentication. + // + // Example: "org.apache.cassandra.auth.PasswordAuthenticator" + ClassName() string + + // Authentication mode of the authenticator. Used as a payload in STARTUP frame. + // + // Example: "Unauthenticated", "Password", "MutualTLS" + AuthenticationMode() string +} + // PasswordAuthenticator specifies credentials to be used when authenticating. // It can be configured with an "allow list" of authenticator class names to avoid // attempting to authenticate with Cassandra if it doesn't provide an expected authenticator. @@ -102,6 +119,14 @@ func (p PasswordAuthenticator) Success(data []byte) error { return nil } +func (p PasswordAuthenticator) ClassName() string { + return "org.apache.cassandra.auth.PasswordAuthenticator" +} + +func (p PasswordAuthenticator) AuthenticationMode() string { + return "Password" +} + // SslOptions configures TLS use. // // Warning: Due to historical reasons, the SslOptions is insecure by default, so you need to set EnableHostVerification @@ -171,6 +196,7 @@ type Conn struct { w contextWriter writeTimeout time.Duration + requestTimeout time.Duration // Request timeout, used for setting up request timers cfg *ConnConfig frameObserver FrameHeaderObserver streamObserver StreamObserver @@ -279,6 +305,7 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg * logger: logger, streamObserver: s.streamObserver, writeTimeout: writeTimeout, + requestTimeout: cfg.ConnectTimeout, } if err := c.init(ctx, dialedHost); err != nil { @@ -312,6 +339,7 @@ func (c *Conn) init(ctx context.Context, dialedHost *DialedHost) error { } c.r.SetTimeout(c.cfg.Timeout) + c.requestTimeout = c.cfg.Timeout // dont coalesce startup frames if c.session.cfg.WriteCoalesceWaitTime > 0 && !c.cfg.disableCoalesce && !dialedHost.DisableCoalesce { @@ -335,8 +363,8 @@ type startupCoordinator struct { func (s *startupCoordinator) setupConn(ctx context.Context) error { var cancel context.CancelFunc - if s.conn.r.GetTimeout() > 0 { - ctx, cancel = context.WithTimeout(ctx, s.conn.r.GetTimeout()) + if s.conn.requestTimeout > 0 { + ctx, cancel = context.WithTimeout(ctx, s.conn.requestTimeout) } else { ctx, cancel = context.WithCancel(ctx) } @@ -479,6 +507,19 @@ func (s *startupCoordinator) startup(ctx context.Context, supported map[string][ } } + // If the server supports authentication negotiation and the client has a registry of authenticators, then we should negotiate authentication. + _, supportsNegotiation := supported["AUTHENTICATORS"] + shouldNegotiateAuthentication := supportsNegotiation && s.conn.session.cfg.AuthRegistry != nil + if shouldNegotiateAuthentication { + authenticators := s.conn.session.cfg.AuthRegistry.Authenticators() + authClasses := make([]string, len(authenticators)) + for i, authenticator := range authenticators { + authClasses[i] = authenticator.AuthenticationMode() + } + // The value should be a comma-separated list of Java class names according to the CEP-50 specification. + m["AUTHENTICATORS"] = strings.Join(authClasses, ",") + } + frame, err := s.write(ctx, &writeStartupFrame{opts: m}, startupCompleted) if err != nil { return err @@ -494,14 +535,23 @@ func (s *startupCoordinator) startup(ctx context.Context, supported map[string][ case *authenticateFrame: // Startup is successfully completed, so we could use Native Protocol 5 startupCompleted.Store(true) - return s.authenticateHandshake(ctx, v, startupCompleted) + return s.authenticateHandshake(ctx, v, startupCompleted, shouldNegotiateAuthentication) default: return NewErrProtocol("Unknown type of response to startup frame: %s", v) } } -func (s *startupCoordinator) authenticateHandshake(ctx context.Context, authFrame *authenticateFrame, startupCompleted *atomic.Bool) error { - if s.conn.auth == nil { +func (s *startupCoordinator) authenticateHandshake(ctx context.Context, authFrame *authenticateFrame, startupCompleted *atomic.Bool, shouldNegotiateAuthentication bool) error { + if shouldNegotiateAuthentication { + auth, ok := s.conn.session.cfg.AuthRegistry.AuthenticatorFor(authFrame.class) + // It should never happen, but we should handle it gracefully. + if !ok { + return fmt.Errorf("the server requested an unknown authenticator during authentication negotiation: %q", authFrame.class) + } + // Set the authenticator to use for the authentication handshake. + s.conn.auth = auth + s.conn.logger.Debug("Authenticator selected for authentication negotiation", NewLogFieldString("authenticator", auth.ClassName())) + } else if s.conn.auth == nil { return fmt.Errorf("authentication required (using %q)", authFrame.class) } @@ -688,8 +738,9 @@ func (c *Conn) processFrame(ctx context.Context, r io.Reader) error { // read a full header, ignore timeouts, as this is being ran in a loop // TODO: TCP level deadlines? or just query level deadlines? - if c.r.GetTimeout() > 0 { - c.r.SetReadDeadline(time.Time{}) + readTimeout := c.r.GetTimeout() + if readTimeout > 0 { + c.r.SetTimeout(0) } headStartTime := time.Now() @@ -700,6 +751,11 @@ func (c *Conn) processFrame(ctx context.Context, r io.Reader) error { return err } + // Set timeout back for reading frame body + if readTimeout > 0 { + c.r.SetTimeout(readTimeout) + } + if c.frameObserver != nil { c.frameObserver.ObserveFrameHeader(context.Background(), ObservedFrameHeader{ Version: protoVersion(head.version), @@ -802,12 +858,24 @@ func (c *Conn) recvSegment(ctx context.Context) error { err error ) + // Read segment without timeout, as this is being run in a loop waiting for the next segment + readTimeout := c.r.GetTimeout() + if readTimeout > 0 { + c.r.SetTimeout(0) + } + // Read frame based on compression if c.compressor != nil { frame, isSelfContained, err = readCompressedSegment(c.r, c.compressor) } else { frame, isSelfContained, err = readUncompressedSegment(c.r) } + + // Restore timeout for subsequent segment reads in multi-segment frames + if readTimeout > 0 { + c.r.SetTimeout(readTimeout) + } + if err != nil { return err } @@ -908,6 +976,8 @@ func (c *connReader) Read(p []byte) (n int, err error) { var nn int if c.timeout > 0 { c.conn.SetReadDeadline(time.Now().Add(c.timeout)) + } else if c.timeout == 0 { + c.conn.SetReadDeadline(time.Time{}) } nn, err = io.ReadFull(c.r, p[n:]) @@ -1309,7 +1379,7 @@ func (c *Conn) execInternal(ctx context.Context, req frameBuilder, tracer Tracer } var timeoutCh <-chan time.Time - if timeout := c.r.GetTimeout(); timeout > 0 { + if c.requestTimeout > 0 { if call.timer == nil { call.timer = time.NewTimer(0) <-call.timer.C @@ -1322,7 +1392,7 @@ func (c *Conn) execInternal(ctx context.Context, req frameBuilder, tracer Tracer } } - call.timer.Reset(timeout) + call.timer.Reset(c.requestTimeout) timeoutCh = call.timer.C } @@ -1638,6 +1708,7 @@ func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter { q.routingInfo.keyspace = usedKeyspace } q.routingInfo.table = info.request.table + q.routingInfo.prepared = true q.routingInfo.mu.Unlock() } else { frame = &writeQueryFrame{ @@ -1833,6 +1904,12 @@ func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter { stmts := make(map[string]string, len(b.batchOpts.entries)) + if b.batchOpts.observer != nil { + b.entryKeyspaces = make([]string, n) + b.entryTables = make([]string, n) + b.entryPrepared = make([]bool, n) + } + for i := 0; i < n; i++ { entry := &b.batchOpts.entries[i] batchStmt := &req.statements[i] @@ -1844,6 +1921,12 @@ func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter { return iter } + if b.entryTables != nil { + b.entryKeyspaces[i] = info.request.keyspace + b.entryTables[i] = info.request.table + b.entryPrepared[i] = true + } + var values []interface{} if entry.binding == nil { values = entry.Args @@ -1952,7 +2035,8 @@ func (c *Conn) querySystemPeers(ctx context.Context, version cassVersion) *Iter err := iter.checkErrAndNotFound() if err != nil { - if errFrame, ok := err.(errorFrame); ok && errFrame.code == ErrCodeInvalid { // system.peers_v2 not found, try system.peers + var requestErr RequestError + if errors.As(err, &requestErr) && requestErr.Code() == ErrCodeInvalid { // system.peers_v2 not found, try system.peers c.mu.Lock() c.isSchemaV2 = false c.mu.Unlock() diff --git a/conn_test.go b/conn_test.go index ad4e66e54..1cb5ee3f8 100644 --- a/conn_test.go +++ b/conn_test.go @@ -287,7 +287,9 @@ func TestTimeout(t *testing.T) { srv := NewTestServer(t, defaultProto, ctx) defer srv.Stop() - db, err := newTestSession(defaultProto, srv.Address) + cluster := testCluster(defaultProto, srv.Address) + cluster.Timeout = 2 * time.Second + db, err := cluster.CreateSession() if err != nil { t.Fatalf("NewCluster: %v", err) } @@ -306,12 +308,18 @@ func TestTimeout(t *testing.T) { } }() - if err := db.Query("kill").WithContext(ctx).Exec(); err == nil { + now := time.Now() + err = db.Query("timeout").ExecContext(ctx) + if err == nil { t.Fatal("expected error got nil") } cancel() - wg.Wait() + + elapsed := time.Since(now) + if elapsed < 1*time.Second || elapsed > 4*time.Second { + t.Fatalf("timeout is not respected (took %v)", elapsed.String()) + } } func TestCancel(t *testing.T) { @@ -329,13 +337,11 @@ func TestCancel(t *testing.T) { } defer db.Close() - qry := db.Query("timeout").WithContext(ctx) - // Make sure we finish the query without leftovers var wg sync.WaitGroup wg.Add(1) go func() { - err = qry.Exec() + err = db.Query("timeout").ExecContext(ctx) wg.Done() }() @@ -720,9 +726,14 @@ func TestStream0(t *testing.T) { t.Fatal(err) } + clientConn, serverConn := net.Pipe() + defer clientConn.Close() + defer serverConn.Close() + conn := &Conn{ r: &connReader{ - r: bufio.NewReader(&buf), + r: bufio.NewReader(&buf), + conn: clientConn, }, streams: streams.New(protoVersion4), session: &Session{ @@ -1248,7 +1259,12 @@ func (srv *TestServer) process(conn net.Conn, reqFrame *framer, useProtoV5, star srv.errorLocked("process frame with a nil header") return } - respFrame := newFramer(nil, byte(head.version), GlobalTypes) + // use the configured version unless it wasn't specified + version := srv.protocol + if version == 0 { + version = byte(head.version) + } + respFrame := newFramer(nil, version, GlobalTypes) if srv.customRequestHandler != nil { if err := srv.customRequestHandler(srv, reqFrame, respFrame); err != nil { diff --git a/control.go b/control.go index cc21e089e..9f56958a3 100644 --- a/control.go +++ b/control.go @@ -361,7 +361,8 @@ func (c *controlConn) setupConn(conn *Conn, sessionInit bool) error { c.conn.Store(ch) c.session.logger.Info("Control connection connected to host.", - NewLogFieldIP("host_addr", host.ConnectAddress()), NewLogFieldString("host_id", host.HostID())) + NewLogFieldIP("host_addr", host.ConnectAddress()), NewLogFieldString("host_id", host.HostID()), + NewLogFieldInt("protocol_version", c.session.cfg.ProtoVersion)) if c.session.initialized() { refreshErr := c.session.schemaDescriber.refreshSchemaMetadata() diff --git a/go.mod b/go.mod index f0428c40e..23e765e95 100644 --- a/go.mod +++ b/go.mod @@ -20,7 +20,7 @@ module github.com/apache/cassandra-gocql-driver/v2 require ( github.com/golang/snappy v0.0.3 github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed - github.com/pierrec/lz4/v4 v4.1.8 + github.com/pierrec/lz4/v4 v4.1.27 github.com/rs/zerolog v1.34.0 github.com/stretchr/testify v1.9.0 go.uber.org/zap v1.27.0 diff --git a/go.sum b/go.sum index c87ff1de5..5b6b743f6 100644 --- a/go.sum +++ b/go.sum @@ -22,6 +22,8 @@ github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APP github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/pierrec/lz4/v4 v4.1.8 h1:ieHkV+i2BRzngO4Wd/3HGowuZStgq6QkPsD1eolNAO4= github.com/pierrec/lz4/v4 v4.1.8/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= +github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= +github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/host_source.go b/host_source.go index e434b652c..a0e860f21 100644 --- a/host_source.go +++ b/host_source.go @@ -453,6 +453,14 @@ func (h *HostInfo) IsUp() bool { } func (h *HostInfo) HostnameAndPort() string { + h.mu.RLock() + if h.hostname != "" { + s := net.JoinHostPort(h.hostname, strconv.Itoa(h.port)) + h.mu.RUnlock() + return s + } + h.mu.RUnlock() + h.mu.Lock() defer h.mu.Unlock() if h.hostname == "" { @@ -463,8 +471,8 @@ func (h *HostInfo) HostnameAndPort() string { } func (h *HostInfo) ConnectAddressAndPort() string { - h.mu.Lock() - defer h.mu.Unlock() + h.mu.RLock() + defer h.mu.RUnlock() addr, _ := h.connectAddressLocked() return net.JoinHostPort(addr.String(), strconv.Itoa(h.port)) } @@ -484,10 +492,7 @@ func (h *HostInfo) String() string { // Polls system.peers at a specific interval to find new hosts type ringDescriber struct { - session *Session - mu sync.Mutex - prevHosts []*HostInfo - prevPartitioner string + session *Session } // Returns true if we are using system_schema.keyspaces instead of system.schema_keyspaces @@ -806,7 +811,7 @@ func (r *ringDescriber) getClusterPeerInfo(localHost *HostInfo) ([]*HostInfo, er // Return true if the host is a valid peer func isValidPeer(host *HostInfo) bool { - return !(len(host.RPCAddress()) == 0 || + return !(len(host.rpcAddress) == 0 || host.hostId == "" || host.dataCenter == "" || host.missingRack || @@ -815,17 +820,14 @@ func isValidPeer(host *HostInfo) bool { // GetHosts returns a list of hosts found via queries to system.local and system.peers func (r *ringDescriber) GetHosts() ([]*HostInfo, string, error) { - r.mu.Lock() - defer r.mu.Unlock() - localHost, err := r.getLocalHostInfo() if err != nil { - return r.prevHosts, r.prevPartitioner, err + return nil, "", err } peerHosts, err := r.getClusterPeerInfo(localHost) if err != nil { - return r.prevHosts, r.prevPartitioner, err + return nil, "", err } hosts := append([]*HostInfo{localHost}, peerHosts...) @@ -1048,13 +1050,13 @@ func (b *errorBroadcaster) newListener() <-chan error { func (b *errorBroadcaster) broadcast(err error) { b.mu.Lock() - defer b.mu.Unlock() curListeners := b.listeners - if len(curListeners) > 0 { - b.listeners = nil - } else { + if len(curListeners) == 0 { + b.mu.Unlock() return } + b.listeners = nil + b.mu.Unlock() for _, listener := range curListeners { listener <- err diff --git a/integration_test.go b/integration_test.go index bd6ccb5cc..c399c788d 100644 --- a/integration_test.go +++ b/integration_test.go @@ -38,6 +38,7 @@ import ( "testing" "time" + "github.com/stretchr/testify/require" inf "gopkg.in/inf.v0" ) @@ -977,3 +978,94 @@ func TestSliceMapMapScanCollectionTypes(t *testing.T) { }) } } + +// TestSmallTimeoutNoPoolErrors verifies that small Session.Timeout values +// don't cause connections to timeout and reconnect constantly. This is a +// regression test for https://github.com/apache/cassandra-gocql-driver/issues/1919 +// +// The issue was that the timeout was being applied to frame header reads, +// causing connections to timeout while waiting for the next frame. The fix +// ensures frame headers are read without timeout, while frame bodies are +// read with timeout. +func TestSmallTimeoutNoPoolErrors(t *testing.T) { + // Create a test logger to capture log messages + logger := newTestLogger(LogLevelDebug) + defer func() { + t.Log(logger.String()) + }() + + cluster := createCluster() + cluster.ConnectTimeout = 10 * time.Second + cluster.Timeout = 750 * time.Millisecond + cluster.NumConns = 1 + cluster.Logger = logger + + db, err := cluster.CreateSession() + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + defer db.Close() + + // Wait for connections to sit idle + // If the bug exists, connections will timeout while waiting for frame headers + // and "Pool connection error" messages will be logged repeatedly + time.Sleep(5 * time.Second) + + // Get log output for analysis + logOutput := strings.ToLower(logger.String()) + + // Count successful connection messages - should be exactly NumConns * number of nodes + connectedCount := strings.Count(logOutput, "pool connected to node") + if connectedCount != *clusterSize*cluster.NumConns { + t.Fatalf("Expected exactly %d 'Pool connected to node' messages, got %d:\n%s", + *clusterSize*cluster.NumConns, connectedCount, logOutput) + } + + // Count error messages - should be zero + // With the bug, we'd see many errors as connections constantly timeout and reconnect + errorCount := strings.Count(logOutput, "pool connection error") + if errorCount > 0 { + t.Fatalf("Found %d 'Pool connection error' messages - connections are timing out and reconnecting:\n%s", + errorCount, logOutput) + } +} + +// Implements the NegotiableAuthenticator interface. +type AllowAllAuthenticator struct{} + +func (p AllowAllAuthenticator) Challenge(req []byte) ([]byte, Authenticator, error) { + return nil, p, nil +} + +func (p AllowAllAuthenticator) Success(data []byte) error { + fmt.Println("Success", data) + return nil +} + +func (p AllowAllAuthenticator) AuthenticationMode() string { + return "Unauthenticated" +} + +func (p AllowAllAuthenticator) ClassName() string { + return "org.apache.cassandra.auth.AllowAllAuthenticator" +} + +func TestAuthenticationNegotiation(t *testing.T) { + cluster := createCluster() + cluster.Logger = NewLogger(LogLevelDebug) + cluster.AuthRegistry = NewDefaultAuthRegistry() + cluster.AuthRegistry.Register(PasswordAuthenticator{ + Username: "1cassandra", + Password: "cassandra", + }) + cluster.AuthRegistry.Register(AllowAllAuthenticator{}) + + session, err := cluster.CreateSession() + require.NoError(t, err) + defer session.Close() + + var hostID string + err = session.Query("SELECT host_id FROM system.local").Scan(&hostID) + require.NoError(t, err) + t.Fatal(hostID) +} diff --git a/policies.go b/policies.go index 41f407803..3d880e977 100644 --- a/policies.go +++ b/policies.go @@ -524,7 +524,7 @@ func (t *tokenAwareHostPolicy) updateReplicas(keyspace string) { if _, ok := meta.replicas[key]; !ok { metaUpdate := t.getMetadataForUpdate() newReplicas := make(map[string]tokenRingReplicas, len(meta.replicas)) - newReplicas[key] = strat.replicaMap(metaUpdate.tokenRing) + newReplicas[key] = strat.replicaMap(metaUpdate.tokenRing, t.logger) for k, replicas := range metaUpdate.replicas { newReplicas[k] = replicas } @@ -556,7 +556,7 @@ func (t *tokenAwareHostPolicy) updateAllReplicas(meta *clusterMeta, schemaMeta * if meta != nil && meta.tokenRing != nil { key := strat.strategyKey() if _, ok := newReplicas[key]; !ok { - newReplicas[key] = strat.replicaMap(meta.tokenRing) + newReplicas[key] = strat.replicaMap(meta.tokenRing, t.logger) } } } diff --git a/protocol_negotiation_test.go b/protocol_negotiation_test.go index 567c74e36..e58ec1dd2 100644 --- a/protocol_negotiation_test.go +++ b/protocol_negotiation_test.go @@ -231,9 +231,18 @@ func TestProtocolNegotiation(t *testing.T) { forceZeroStreamID: tc.forceZeroStreamID, } + // use the maximum protocol supported + protocol := uint8(0) + for _, supportedVersion := range tc.supportedVersions { + supportedProto := uint8(supportedVersion) + if supportedProto > protocol { + protocol = supportedProto + } + } + srv := newTestServerOpts{ addr: "127.0.0.1:0", - protocol: 5, + protocol: protocol, customRequestHandler: handler.handle, dontFailOnProtocolMismatch: true, }.newServer(t, context.Background()) diff --git a/query_executor.go b/query_executor.go index 2d7a62335..5d848c2e8 100644 --- a/query_executor.go +++ b/query_executor.go @@ -385,15 +385,20 @@ func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Ite q.qryOpts.observer.ObserveQuery(q.qryOpts.context, ObservedQuery{ Keyspace: keyspace, Statement: q.qryOpts.stmt, - Values: q.qryOpts.values, - Start: start, - End: end, - Rows: iter.numRows, - Host: host, - Metrics: metricsForHost, - Err: iter.err, - Attempt: attempt, - Query: q.originalQuery, + PreparedMetadata: PreparedMetadata{ + Keyspace: q.routingInfo.getKeyspace(), + Table: q.routingInfo.getTable(), + }, + IsPrepared: q.routingInfo.isPrepared(), + Values: q.qryOpts.values, + Start: start, + End: end, + Rows: iter.numRows, + Host: host, + Metrics: metricsForHost, + Err: iter.err, + Attempt: attempt, + Query: q.originalQuery, }) } } @@ -432,6 +437,7 @@ func (q *internalQuery) GetRoutingKey() ([]byte, error) { q.routingInfo.mu.Lock() q.routingInfo.keyspace = meta.Keyspace q.routingInfo.table = meta.Table + q.routingInfo.prepared = true q.routingInfo.mu.Unlock() } return createRoutingKey(meta, q.qryOpts.values) @@ -562,6 +568,19 @@ type internalBatch struct { session *Session metrics *queryMetrics hostMetricsManager hostMetricsManager + + // entryKeyspaces holds the keyspace for each batch entry, + // populated from prepared statement metadata during execution. + entryKeyspaces []string + + // entryTables holds the table name for each batch entry, + // populated from prepared statement metadata during execution. + entryTables []string + + // entryPrepared[i] reports whether the i-th batch entry was prepared by + // the driver, and therefore whether entryKeyspaces[i]/entryTables[i] hold + // valid prepared statement metadata. + entryPrepared []bool } func newInternalBatch(batch *Batch, ctx context.Context) *internalBatch { @@ -597,20 +616,30 @@ func (b *internalBatch) attempt(keyspace string, end, start time.Time, iter *Ite metricsForHost := b.hostMetricsManager.attempt(latency, host) - statements := make([]string, len(b.batchOpts.entries)) - values := make([][]interface{}, len(b.batchOpts.entries)) + n := len(b.batchOpts.entries) + statements := make([]string, n) + values := make([][]interface{}, n) + preparedMetadata := make([]PreparedMetadata, n) for i, entry := range b.batchOpts.entries { statements[i] = entry.Stmt values[i] = entry.Args + if i < len(b.entryKeyspaces) { + preparedMetadata[i] = PreparedMetadata{ + Keyspace: b.entryKeyspaces[i], + Table: b.entryTables[i], + } + } } b.batchOpts.observer.ObserveBatch(b.batchOpts.context, ObservedBatch{ - Keyspace: keyspace, - Statements: statements, - Values: values, - Start: start, - End: end, + Keyspace: keyspace, + Statements: statements, + PreparedMetadata: preparedMetadata, + IsPrepared: b.entryPrepared, + Values: values, + Start: start, + End: end, // Rows not used in batch observations // TODO - might be able to support it when using BatchCAS Host: host, Metrics: metricsForHost, @@ -652,6 +681,7 @@ func (b *internalBatch) GetRoutingKey() ([]byte, error) { b.routingInfo.mu.Lock() b.routingInfo.keyspace = meta.Keyspace b.routingInfo.table = meta.Table + b.routingInfo.prepared = true b.routingInfo.mu.Unlock() } diff --git a/session.go b/session.go index fcbdd703b..289fd8ee8 100644 --- a/session.go +++ b/session.go @@ -141,6 +141,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) { } // Check that either Authenticator is set or AuthProvider, not both + // TODO: add AuthRegistry check if cfg.Authenticator != nil && cfg.AuthProvider != nil { return nil, errors.New("Can't use both Authenticator and AuthProvider in cluster config.") } @@ -1086,6 +1087,8 @@ type queryRoutingInfo struct { keyspace string table string + + prepared bool } func (qr *queryRoutingInfo) getKeyspace() string { @@ -1100,6 +1103,12 @@ func (qr *queryRoutingInfo) getTable() string { return qr.table } +func (qr *queryRoutingInfo) isPrepared() bool { + qr.mu.RLock() + defer qr.mu.RUnlock() + return qr.prepared +} + func (q *Query) defaultsFromSession() { s := q.session @@ -1318,6 +1327,14 @@ func (q *Query) Idempotent(value bool) *Query { // For supported Go to CQL type conversions for query parameters, see Session.Query documentation. func (q *Query) Bind(v ...interface{}) *Query { q.values = v + q.binding = nil + return q +} + +// Binding sets a function for dynamic generation of query arguments. +func (q *Query) Binding(binding func(q *QueryInfo) ([]interface{}, error)) *Query { + q.values = nil + q.binding = binding return q } @@ -2370,10 +2387,28 @@ func (s *Session) GetHosts() []*HostInfo { return s.ring.allHosts() } +// PreparedMetadata holds metadata extracted from a prepared statement +// (returned by Cassandra during statement preparation). It is exposed via +// ObservedQuery and ObservedBatch when the driver has prepared the statement. +type PreparedMetadata struct { + Keyspace string + Table string +} + type ObservedQuery struct { Keyspace string Statement string + // PreparedMetadata holds keyspace/table information from prepared statement + // metadata returned by Cassandra. Only valid when IsPrepared is true; the + // zero value is reported otherwise (for example, for DDL statements or + // statements not prepared by the driver). + PreparedMetadata PreparedMetadata + + // IsPrepared reports whether the statement was prepared by the driver and + // therefore whether PreparedMetadata holds valid information. + IsPrepared bool + // Values holds a slice of bound values for the query. // Do not modify the values here, they are shared with multiple goroutines. Values []interface{} @@ -2418,6 +2453,15 @@ type ObservedBatch struct { Keyspace string Statements []string + // PreparedMetadata holds prepared statement metadata for each batch statement. + // PreparedMetadata[i] corresponds to Statements[i]. Each entry is only valid + // when IsPrepared[i] is true; the zero value is reported otherwise. + PreparedMetadata []PreparedMetadata + + // IsPrepared reports, for each batch statement, whether the statement was + // prepared by the driver. IsPrepared[i] corresponds to Statements[i]. + IsPrepared []bool + // Values holds a slice of bound values for each statement. // Values[i] are bound values passed to Statements[i]. // Do not modify the values here, they are shared with multiple goroutines. @@ -2497,6 +2541,10 @@ var ( // ErrProtocol represents a protocol-level error. type ErrProtocol struct{ error } +func (e ErrProtocol) Unwrap() error { + return e.error +} + // NewErrProtocol creates a new protocol error with the specified format and arguments. func NewErrProtocol(format string, args ...interface{}) error { return ErrProtocol{fmt.Errorf(format, args...)} diff --git a/session_test.go b/session_test.go index b1c7b7476..aa1f45210 100644 --- a/session_test.go +++ b/session_test.go @@ -403,3 +403,60 @@ func TestRetryType_IgnoreRethrow(t *testing.T) { resetObserved() } } + +func TestStaticQueryInfo_OverrideBindingFunction(t *testing.T) { + session := createSession(t) + defer session.Close() + + if err := createTable(session, "CREATE TABLE IF NOT EXISTS gocql_test.static_query_info_override (id int, value text, PRIMARY KEY (id))"); err != nil { + t.Fatalf("failed to create table with error '%v'", err) + } + + if err := session.Query("INSERT INTO static_query_info_override (id, value) VALUES (?, ?)", 1, "foo").Exec(); err != nil { + t.Fatalf("insert into static_query_info_override failed, err '%v'", err) + } + + if err := session.Query("INSERT INTO static_query_info_override (id, value) VALUES (?, ?)", 2, "bar").Exec(); err != nil { + t.Fatalf("insert into static_query_info_override failed, err '%v'", err) + } + + qry := session.Bind("SELECT id, value FROM static_query_info_override WHERE id = ?", func(q *QueryInfo) ([]interface{}, error) { + values := make([]interface{}, 1) + values[0] = 1 + return values, nil + }) + + iter := qry.Iter() + var id int + var value string + iter.Scan(&id, &value) + if err := iter.Close(); err != nil { + t.Fatalf("query with exposed info failed, err '%v'", err) + } + + if id != 1 { + t.Fatalf("Expected id %d, but got %d", 113, id) + } + if value != "foo" { + t.Fatalf("Expected value %s, but got %s", "foo", value) + } + + qry.Binding(func(q *QueryInfo) ([]interface{}, error) { + values := make([]interface{}, 1) + values[0] = 2 + return values, nil + }) + + iter = qry.Iter() + iter.Scan(&id, &value) + if err := iter.Close(); err != nil { + t.Fatalf("query with exposed info failed, err '%v'", err) + } + + if id != 2 { + t.Fatalf("Expected id %d, but got %d", 2, id) + } + if value != "bar" { + t.Fatalf("Expected value %s, but got %s", "bar", value) + } +} diff --git a/topology.go b/topology.go index 3ec20b15e..0481b2ff8 100644 --- a/topology.go +++ b/topology.go @@ -66,7 +66,7 @@ func (h tokenRingReplicas) replicasFor(t token) *hostTokens { } type placementStrategy interface { - replicaMap(tokenRing *tokenRing) tokenRingReplicas + replicaMap(tokenRing *tokenRing, logger StructuredLogger) tokenRingReplicas replicationFactor(dc string) int // strategyKey returns a unique identifier string for this strategy instance. // Two strategy instances with identical configuration should return the same key. @@ -162,7 +162,7 @@ func (s *simpleStrategy) replicationFactor(dc string) int { return s.rf } -func (s *simpleStrategy) replicaMap(tokenRing *tokenRing) tokenRingReplicas { +func (s *simpleStrategy) replicaMap(tokenRing *tokenRing, _ StructuredLogger) tokenRingReplicas { tokens := tokenRing.tokens ring := make(tokenRingReplicas, len(tokens)) @@ -255,7 +255,7 @@ func (n *networkTopology) haveRF(replicaCounts map[string]int) bool { return true } -func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas { +func (n *networkTopology) replicaMap(tokenRing *tokenRing, logger StructuredLogger) tokenRingReplicas { dcRacks := make(map[string]map[string]struct{}, len(n.dcs)) // skipped hosts in a dc skipped := make(map[string][]*HostInfo, len(n.dcs)) @@ -324,7 +324,11 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas { continue } else if replicasInDC[dc] >= rf { if replicasInDC[dc] > rf { - panic(fmt.Sprintf("replica overflow. rf=%d have=%d in dc %q", rf, replicasInDC[dc], dc)) + logger.Warning("Replica overflow. Returning empty map.", + NewLogFieldInt("rf", rf), + NewLogFieldInt("have", replicasInDC[dc]), + NewLogFieldString("dc", dc)) + return tokenRingReplicas{} } // have enough replicas in this DC @@ -372,23 +376,36 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas { } if len(replicas) == 0 { - panic(fmt.Sprintf("no replicas for token: %v", th.token)) + logger.Warning("No replicas for token. Returning empty map.", + NewLogFieldString("token", th.token.String())) + return tokenRingReplicas{} } else if !replicas[0].Equal(th.host) { - panic(fmt.Sprintf("first replica is not the primary replica for the token: expected %v got %v", replicas[0].ConnectAddress(), th.host.ConnectAddress())) + logger.Warning("First replica is not the primary replica for the token. Returning empty map.", + NewLogFieldString("token", th.token.String()), + NewLogFieldIP("expected", replicas[0].ConnectAddress()), + NewLogFieldIP("got", th.host.ConnectAddress())) + return tokenRingReplicas{} } replicaRing = append(replicaRing, hostTokens{th.token, replicas}) } dcsWithReplicas := 0 - for _, dc := range n.dcs { - if dc > 0 { + for dc, rf := range n.dcs { + // We should count only DCs that driver is aware of and have a replication factor > 0 + if _, knownDc := dcRacks[dc]; knownDc && rf > 0 { dcsWithReplicas++ } } if dcsWithReplicas == len(dcRacks) && len(replicaRing) != len(tokens) { - panic(fmt.Sprintf("token map different size to token ring: got %d expected %d", len(replicaRing), len(tokens))) + logger.Warning("Unexpected state while building replica map. Returning empty map.", + NewLogFieldString("strategy_key", n.strategyKey()), + NewLogFieldInt("dcs_with_replicas", dcsWithReplicas), + NewLogFieldInt("dcs_in_ring", len(dcRacks)), + NewLogFieldInt("token_ring_size", len(tokens)), + NewLogFieldInt("replica_ring_size", len(replicaRing))) + return tokenRingReplicas{} } return replicaRing diff --git a/topology_test.go b/topology_test.go index 10ee85e53..3e8fb9a2a 100644 --- a/topology_test.go +++ b/topology_test.go @@ -31,6 +31,8 @@ import ( "fmt" "sort" "testing" + + "github.com/stretchr/testify/require" ) func TestPlacementStrategy_SimpleStrategy(t *testing.T) { @@ -49,7 +51,7 @@ func TestPlacementStrategy_SimpleStrategy(t *testing.T) { hosts := []*HostInfo{host0, host25, host50, host75} strat := newSimpleStrategy(2) - tokenReplicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}) + tokenReplicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}, nopLoggerSingleton) if len(tokenReplicas) != len(tokens) { t.Fatalf("expected replica map to have %d items but has %d", len(tokens), len(tokenReplicas)) } @@ -157,7 +159,7 @@ func TestPlacementStrategy_NetworkStrategy(t *testing.T) { expReplicas += rf } - tokenReplicas := test.strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}) + tokenReplicas := test.strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}, nopLoggerSingleton) if len(tokenReplicas) != test.expectedReplicaMapSize { t.Fatalf("expected replica map to have %d items but has %d", test.expectedReplicaMapSize, len(tokenReplicas)) @@ -224,3 +226,33 @@ func TestPlacementStrategy_NetworkStrategy(t *testing.T) { }) } } + +// Regression test for CASSGO-122: +// when the token ring only contains hosts from a DC that has RF=0/unspecified for a keyspace, +// networkTopology.replicaMap should return an empty replica map. +func TestPlacementStrategy_NetworkStrategy_ReturnEmptyReplicaMapWhenNoReplicasInRing(t *testing.T) { + strat := newNetworkTopology(map[string]int{ + "dc1": 3, // replicated only in dc1 + }) + + // Hosts in ring only from dc2, so no replicas should be returned. + // hostId format: dc:rack:host which is used as a token in the token ring. + // It makes sense to use the hostId as a token in the token ring because it is unique and deterministic for test purpose. + hosts := []*HostInfo{ + {hostId: "dc2:rack1:0", dataCenter: "dc2", rack: "rack1"}, + {hostId: "dc2:rack2:1", dataCenter: "dc2", rack: "rack2"}, + {hostId: "dc2:rack3:2", dataCenter: "dc2", rack: "rack3"}, + } + + tokens := make([]hostToken, 0, len(hosts)) + for _, h := range hosts { + tokens = append(tokens, hostToken{ + token: orderedToken(h.hostId), + host: h, + }) + } + sort.Sort(&tokenRing{tokens: tokens}) + + replicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens}, nopLoggerSingleton) + require.Empty(t, replicas, "expected no replicas, got %d", len(replicas)) +}