Change logger so it supports structured logging and log levels

The logger interface uses printf semantics and lacks the notion of log levels.
This commit changes (breaking change) the logger interface so printf semantics
are removed in favor of structured logging and adds log levels.

It provides 3 built in log implementations:
  - Default logger - uses log standard library
  - gocqlzap         - uses zap library
  - gocqlzerolog   - uses zerolog library

Users can use these implementations as is or they can implement their own
using these as examples.

Patch by João Reis; reviewed by James Hartig, Bohdan Siryk for CASSGO-9

Author: joao-r-reis

CHANGELOG.md

@@ -10,6 +10,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Removed
 
 - Drop support for old CQL protocol versions: 1 and 2 (CASSGO-75)
+- Cleanup of deprecated elements (CASSGO-12)
+- Remove global NewBatch function (CASSGO-15)
+- Remove deprecated global logger (CASSGO-24)
 
 ### Added
 
@@ -27,26 +30,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Move lz4 compressor to lz4 package within the gocql module (CASSGO-32)
 - Don't restrict server authenticator unless PasswordAuthentictor.AllowedAuthenticators is provided (CASSGO-19)
-- Cleanup of deprecated elements (CASSGO-12)
-- Remove global NewBatch function (CASSGO-15)
 - Detailed description for NumConns (CASSGO-3)
 - Change Batch API to be consistent with Query() (CASSGO-7)
-- Added Cassandra 4.0 table options support(CASSGO-13)
-- Remove deprecated global logger (CASSGO-24)
+- Added Cassandra 4.0 table options support (CASSGO-13)
 - Bumped actions/upload-artifact and actions/cache versions to v4 in CI workflow (CASSGO-48)
 - Keep nil slices in MapScan (CASSGO-44)
 - Improve error messages for marshalling (CASSGO-38)
 - Remove HostPoolHostPolicy from gocql package (CASSGO-21)
 - Standardized spelling of datacenter (CASSGO-35)
 - Refactor HostInfo creation and ConnectAddress() method (CASSGO-45)
-- gocql.Compressor interface changes to follow append-like design. Bumped Go version to 1.19 (CASSGO-1)
+- gocql.Compressor interface changes to follow append-like design (CASSGO-1)
 - Refactoring hostpool package test and Expose HostInfo creation (CASSGO-59)
 - Move "execute batch" methods to Batch type (CASSGO-57)
 - Make `Session` immutable by removing setters and associated mutex (CASSGO-23)
 - inet columns default to net.IP when using MapScan or SliceMap (CASSGO-43)
 - NativeType removed (CASSGO-43)
 - `New` and `NewWithError` removed and replaced with `Zero` (CASSGO-43)
 - Changes to Query and Batch to make them safely reusable (CASSGO-22)
+- Change logger interface so it supports structured logging and log levels (CASSGO-9)
 
 ### Fixed
 

address_translators_test.go

@@ -1,3 +1,6 @@
+//go:build all || unit
+// +build all unit
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file

cluster.go

@@ -259,9 +259,15 @@ type ClusterConfig struct {
 	// If not provided, Dialer will be used instead.
 	HostDialer HostDialer
 
-	// Logger for this ClusterConfig.
-	// If not specified, defaults to the gocql.defaultLogger.
-	Logger StdLogger
+	// StructuredLogger for this ClusterConfig.
+	//
+	// There are 3 built in implementations of StructuredLogger:
+	//  - std library "log" package: gocql.NewLogger
+	//  - zerolog: gocqlzerolog.NewZerologLogger
+	//  - zap: gocqlzap.NewZapLogger
+	//
+	// You can also provide your own logger implementation of the StructuredLogger interface.
+	Logger StructuredLogger
 
 	// Tracer will be used for all queries. Alternatively it can be set of on a
 	// per query basis.
@@ -318,11 +324,11 @@ func NewCluster(hosts ...string) *ClusterConfig {
 	return cfg
 }
 
-func (cfg *ClusterConfig) logger() StdLogger {
-	if cfg.Logger == nil {
-		return &defaultLogger{}
+func (cfg *ClusterConfig) newLogger() StructuredLogger {
+	if cfg.Logger != nil {
+		return cfg.Logger
 	}
-	return cfg.Logger
+	return NewLogger(LogLevelNone)
 }
 
 // CreateSession initializes the cluster based on this config and returns a
@@ -335,14 +341,14 @@ func (cfg *ClusterConfig) CreateSession() (*Session, error) {
 // if defined, to translate the given address and port into a possibly new address
 // and port, If no AddressTranslator or if an error occurs, the given address and
 // port will be returned.
-func (cfg *ClusterConfig) translateAddressPort(addr net.IP, port int) (net.IP, int) {
+func (cfg *ClusterConfig) translateAddressPort(addr net.IP, port int, logger StructuredLogger) (net.IP, int) {
 	if cfg.AddressTranslator == nil || len(addr) == 0 {
 		return addr, port
 	}
 	newAddr, newPort := cfg.AddressTranslator.Translate(addr, port)
-	if gocqlDebug {
-		cfg.logger().Printf("gocql: translating address '%v:%d' to '%v:%d'", addr, port, newAddr, newPort)
-	}
+	logger.Debug("Translating address.",
+		newLogFieldIp("old_addr", addr), newLogFieldInt("old_port", port),
+		newLogFieldIp("new_addr", newAddr), newLogFieldInt("new_port", newPort))
 	return newAddr, newPort
 }
 

cluster_test.go

@@ -1,3 +1,6 @@
+//go:build all || unit
+// +build all unit
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -60,23 +63,23 @@ func TestNewCluster_WithHosts(t *testing.T) {
 func TestClusterConfig_translateAddressAndPort_NilTranslator(t *testing.T) {
 	cfg := NewCluster()
 	assertNil(t, "cluster config address translator", cfg.AddressTranslator)
-	newAddr, newPort := cfg.translateAddressPort(net.ParseIP("10.0.0.1"), 1234)
+	newAddr, newPort := cfg.translateAddressPort(net.ParseIP("10.0.0.1"), 1234, nopLoggerSingleton)
 	assertTrue(t, "same address as provided", net.ParseIP("10.0.0.1").Equal(newAddr))
 	assertEqual(t, "translated host and port", 1234, newPort)
 }
 
 func TestClusterConfig_translateAddressAndPort_EmptyAddr(t *testing.T) {
 	cfg := NewCluster()
 	cfg.AddressTranslator = staticAddressTranslator(net.ParseIP("10.10.10.10"), 5432)
-	newAddr, newPort := cfg.translateAddressPort(net.IP([]byte{}), 0)
+	newAddr, newPort := cfg.translateAddressPort(net.IP([]byte{}), 0, nopLoggerSingleton)
 	assertTrue(t, "translated address is still empty", len(newAddr) == 0)
 	assertEqual(t, "translated port", 0, newPort)
 }
 
 func TestClusterConfig_translateAddressAndPort_Success(t *testing.T) {
 	cfg := NewCluster()
 	cfg.AddressTranslator = staticAddressTranslator(net.ParseIP("10.10.10.10"), 5432)
-	newAddr, newPort := cfg.translateAddressPort(net.ParseIP("10.0.0.1"), 2345)
+	newAddr, newPort := cfg.translateAddressPort(net.ParseIP("10.0.0.1"), 2345, nopLoggerSingleton)
 	assertTrue(t, "translated address", net.ParseIP("10.10.10.10").Equal(newAddr))
 	assertEqual(t, "translated port", 5432, newPort)
 }

common_test.go

@@ -1,3 +1,6 @@
+//go:build all || unit || integration || ccm || cassandra
+// +build all unit integration ccm cassandra
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -85,7 +88,7 @@ func addSslOptions(cluster *ClusterConfig) *ClusterConfig {
 var initOnce sync.Once
 
 func createTable(s *Session, table string) error {
-	// lets just be really sure
+	// let's just be really sure
 	if err := s.control.awaitSchemaAgreement(); err != nil {
 		log.Printf("error waiting for schema agreement pre create table=%q err=%v\n", table, err)
 		return err

conn.go

@@ -144,19 +144,12 @@ type ConnConfig struct {
 	Authenticator  Authenticator
 	AuthProvider   func(h *HostInfo) (Authenticator, error)
 	Keepalive      time.Duration
-	Logger         StdLogger
+	Logger         StructuredLogger
 
 	tlsConfig       *tls.Config
 	disableCoalesce bool
 }
 
-func (c *ConnConfig) logger() StdLogger {
-	if c.Logger == nil {
-		return &defaultLogger{}
-	}
-	return c.Logger
-}
-
 type ConnErrorHandler interface {
 	HandleError(conn *Conn, err error, closed bool)
 }
@@ -208,7 +201,7 @@ type Conn struct {
 
 	timeouts int64
 
-	logger StdLogger
+	logger StructuredLogger
 }
 
 // connect establishes a connection to a Cassandra node using session's connection config.
@@ -715,7 +708,7 @@ func (c *Conn) processFrame(ctx context.Context, r io.Reader) error {
 	delete(c.calls, head.stream)
 	c.mu.Unlock()
 	if call == nil || !ok {
-		c.logger.Printf("gocql: received response for stream which has no handler: header=%v\n", head)
+		c.logger.Warning("Received response for stream which has no handler.", newLogFieldString("header", head.String()))
 		return c.discardFrame(r, head)
 	} else if head.stream != call.streamID {
 		panic(fmt.Sprintf("call has incorrect streamID: got %d expected %d", call.streamID, head.stream))
@@ -1330,12 +1323,19 @@ func (c *Conn) execInternal(ctx context.Context, req frameBuilder, tracer Tracer
 		return resp.framer, nil
 	case <-timeoutCh:
 		close(call.timeout)
+		c.logger.Debug("Request timed out on connection.",
+			newLogFieldString("host_id", c.host.HostID()), newLogFieldIp("addr", c.host.ConnectAddress()))
 		c.handleTimeout()
 		return nil, ErrTimeoutNoResponse
 	case <-ctxDone:
+		c.logger.Debug("Request failed because context elapsed out on connection.",
+			newLogFieldString("host_id", c.host.HostID()), newLogFieldIp("addr", c.host.ConnectAddress()),
+			newLogFieldError("ctx_err", ctx.Err()))
 		close(call.timeout)
 		return nil, ctx.Err()
 	case <-c.ctx.Done():
+		c.logger.Debug("Request failed because connection closed.",
+			newLogFieldString("host_id", c.host.HostID()), newLogFieldIp("addr", c.host.ConnectAddress()))
 		close(call.timeout)
 		return nil, ErrConnectionClosed
 	}
@@ -1685,7 +1685,7 @@ func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter {
 		iter.framer = framer
 		if err := c.awaitSchemaAgreement(ctx); err != nil {
 			// TODO: should have this behind a flag
-			c.logger.Println(err)
+			c.logger.Warning("Error while awaiting for schema agreement after a schema change event.", newLogFieldError("err", err))
 		}
 		// dont return an error from this, might be a good idea to give a warning
 		// though. The impact of this returning an error would be that the cluster
@@ -1947,7 +1947,7 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 				goto cont
 			}
 			if !isValidPeer(host) || host.schemaVersion == "" {
-				c.logger.Printf("invalid peer or peer with empty schema_version: peer=%q", host)
+				c.logger.Warning("Invalid peer or peer with empty schema_version.", newLogFieldIp("peer", host.ConnectAddress()))
 				continue
 			}
 

conn_test.go

@@ -41,6 +41,7 @@ import (
 	"math/rand"
 	"net"
 	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -184,7 +185,7 @@ func newTestSession(proto protoVersion, addresses ...string) (*Session, error) {
 }
 
 func TestDNSLookupConnected(t *testing.T) {
-	log := &testLogger{}
+	log := newTestLogger(LogLevelDebug)
 
 	// Override the defaul DNS resolver and restore at the end
 	failDNS = true
@@ -205,13 +206,13 @@ func TestDNSLookupConnected(t *testing.T) {
 		t.Fatal("CreateSession() should have connected")
 	}
 
-	if !strings.Contains(log.String(), "gocql: dns error") {
+	if !strings.Contains(log.String(), "gocql: DNS error") {
 		t.Fatalf("Expected to receive dns error log message  - got '%s' instead", log.String())
 	}
 }
 
 func TestDNSLookupError(t *testing.T) {
-	log := &testLogger{}
+	log := newTestLogger(LogLevelDebug)
 
 	// Override the defaul DNS resolver and restore at the end
 	failDNS = true
@@ -229,7 +230,7 @@ func TestDNSLookupError(t *testing.T) {
 		t.Fatal("CreateSession() should have returned an error")
 	}
 
-	if !strings.Contains(log.String(), "gocql: dns error") {
+	if !strings.Contains(log.String(), "gocql: DNS error") {
 		t.Fatalf("Expected to receive dns error log message  - got '%s' instead", log.String())
 	}
 
@@ -240,7 +241,7 @@ func TestDNSLookupError(t *testing.T) {
 
 func TestStartupTimeout(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
-	log := &testLogger{}
+	log := newTestLogger(LogLevelDebug)
 
 	srv := NewTestServer(t, defaultProto, ctx)
 	defer srv.Stop()
@@ -348,17 +349,20 @@ func TestCancel(t *testing.T) {
 
 type testQueryObserver struct {
 	metrics map[string]*hostMetrics
-	verbose bool
-	logger  StdLogger
+	logger  StructuredLogger
 }
 
 func (o *testQueryObserver) ObserveQuery(ctx context.Context, q ObservedQuery) {
 	host := q.Host.ConnectAddress().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",
-			q.Statement, q.Rows, q.End.Sub(q.Start), host, q.Metrics.Attempts, q.Metrics.TotalLatency, q.Err)
-	}
+	o.logger.Debug("Observed query.",
+		newLogFieldString("stmt", q.Statement),
+		newLogFieldInt("rows", q.Rows),
+		newLogFieldString("duration", q.End.Sub(q.Start).String()),
+		newLogFieldString("host", host),
+		newLogFieldInt("attempts", q.Metrics.Attempts),
+		newLogFieldString("latency", strconv.FormatInt(q.Metrics.TotalLatency, 10)),
+		newLogFieldError("err", q.Err))
 }
 
 func (o *testQueryObserver) GetMetrics(host *HostInfo) *hostMetrics {
@@ -411,7 +415,7 @@ func TestQueryRetry(t *testing.T) {
 }
 
 func TestQueryMultinodeWithMetrics(t *testing.T) {
-	log := &testLogger{}
+	log := newTestLogger(LogLevelNone)
 	defer func() {
 		os.Stdout.WriteString(log.String())
 	}()
@@ -439,7 +443,7 @@ func TestQueryMultinodeWithMetrics(t *testing.T) {
 
 	// 1 retry per host
 	rt := &SimpleRetryPolicy{NumRetries: 3}
-	observer := &testQueryObserver{metrics: make(map[string]*hostMetrics), verbose: false, logger: log}
+	observer := &testQueryObserver{metrics: make(map[string]*hostMetrics), logger: log}
 	qry := db.Query("kill").RetryPolicy(rt).Observer(observer).Idempotent(true)
 	iter := qry.Iter()
 	err = iter.Close()
@@ -488,7 +492,7 @@ func (t *testRetryPolicy) GetRetryType(err error) RetryType {
 }
 
 func TestSpeculativeExecution(t *testing.T) {
-	log := &testLogger{}
+	log := newTestLogger(LogLevelDebug)
 	defer func() {
 		os.Stdout.WriteString(log.String())
 	}()
@@ -724,7 +728,7 @@ func TestStream0(t *testing.T) {
 		session: &Session{
 			types: GlobalTypes,
 		},
-		logger: &defaultLogger{},
+		logger: NewLogger(LogLevelNone),
 	}
 
 	err := conn.recv(context.Background(), false)