Repository: gocql/gocql
Pull request #1553 (2021-04-14 11:00:27)

Add per-session logger
There are race conditions in tests that swap the global logger,
we shouldn't modify any global state in tests or otherwise.


Patch:
diff --git a/cluster.go b/cluster.go
index 2c29437..e7c4669 100644
--- a/cluster.go
+++ b/cluster.go
@@ -149,6 +149,10 @@ type ClusterConfig struct {
 	// If not provided, a default dialer configured with ConnectTimeout will be used.
 	Dialer Dialer
 
+	// Logger for this ClusterConfig.
+	// If not specified, defaults to the global gocql.Logger.
+	Logger StdLogger
+
 	// internal config for testing
 	disableControlConn bool
 }
@@ -188,6 +192,13 @@ func NewCluster(hosts ...string) *ClusterConfig {
 	return cfg
 }
 
+func (cfg *ClusterConfig) logger() StdLogger {
+	if cfg.Logger == nil {
+		return Logger
+	}
+	return cfg.Logger
+}
+
 // CreateSession initializes the cluster based on this config and returns a
 // session object that can be used to interact with the database.
 func (cfg *ClusterConfig) CreateSession() (*Session, error) {
@@ -204,7 +215,7 @@ func (cfg *ClusterConfig) translateAddressPort(addr net.IP, port int) (net.IP, i
 	}
 	newAddr, newPort := cfg.AddressTranslator.Translate(addr, port)
 	if gocqlDebug {
-		Logger.Printf("gocql: translating address '%v:%d' to '%v:%d'", addr, port, newAddr, newPort)
+		cfg.logger().Printf("gocql: translating address '%v:%d' to '%v:%d'", addr, port, newAddr, newPort)
 	}
 	return newAddr, newPort
 }
diff --git a/conn.go b/conn.go
index bc7ed44..374d951 100644
--- a/conn.go
+++ b/conn.go
@@ -121,11 +121,19 @@ type ConnConfig struct {
 	Authenticator  Authenticator
 	AuthProvider   func(h *HostInfo) (Authenticator, error)
 	Keepalive      time.Duration
+	Logger         StdLogger
 
 	tlsConfig       *tls.Config
 	disableCoalesce bool
 }
 
+func (c *ConnConfig) logger() StdLogger {
+	if c.Logger == nil {
+		return Logger
+	}
+	return c.Logger
+}
+
 type ConnErrorHandler interface {
 	HandleError(conn *Conn, err error, closed bool)
 }
@@ -178,6 +186,8 @@ type Conn struct {
 	cancel context.CancelFunc
 
 	timeouts int64
+
+	logger StdLogger
 }
 
 // connect establishes a connection to a Cassandra node using session's connection config.
@@ -276,6 +286,7 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg *
 		},
 		ctx:    ctx,
 		cancel: cancel,
+		logger: cfg.logger(),
 	}
 
 	if err := c.init(ctx); err != nil {
@@ -685,7 +696,7 @@ func (c *Conn) recv(ctx context.Context) error {
 	delete(c.calls, head.stream)
 	c.mu.Unlock()
 	if call == nil || call.framer == nil || !ok {
-		Logger.Printf("gocql: received response for stream which has no handler: header=%v\n", head)
+		c.logger.Printf("gocql: received response for stream which has no handler: header=%v\n", head)
 		return c.discardFrame(head)
 	} else if head.stream != call.streamID {
 		panic(fmt.Sprintf("call has incorrect streamID: got %d expected %d", call.streamID, head.stream))
@@ -1227,7 +1238,7 @@ func (c *Conn) executeQuery(ctx context.Context, qry *Query) *Iter {
 		iter := &Iter{framer: framer}
 		if err := c.awaitSchemaAgreement(ctx); err != nil {
 			// TODO: should have this behind a flag
-			Logger.Println(err)
+			c.logger.Println(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
@@ -1429,7 +1440,7 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context) (err error) {
 				goto cont
 			}
 			if !isValidPeer(host) || host.schemaVersion == "" {
-				Logger.Printf("invalid peer or peer with empty schema_version: peer=%q", host)
+				c.logger.Printf("invalid peer or peer with empty schema_version: peer=%q", host)
 				continue
 			}
 
diff --git a/connectionpool.go b/connectionpool.go
index 75c28bc..8d0a942 100644
--- a/connectionpool.go
+++ b/connectionpool.go
@@ -113,6 +113,7 @@ func connConfig(cfg *ClusterConfig) (*ConnConfig, error) {
 		Authenticator:   cfg.Authenticator,
 		AuthProvider:    cfg.AuthProvider,
 		Keepalive:       cfg.SocketKeepalive,
+		Logger:          cfg.logger(),
 		tlsConfig:       tlsConfig,
 		disableCoalesce: tlsConfig != nil, // write coalescing doesn't work with framing on top of TCP like in TLS.
 	}, nil
@@ -273,7 +274,8 @@ type hostConnPool struct {
 	closed  bool
 	filling bool
 
-	pos uint32
+	pos    uint32
+	logger StdLogger
 }
 
 func (h *hostConnPool) String() string {
@@ -296,6 +298,7 @@ func newHostConnPool(session *Session, host *HostInfo, port, size int,
 		conns:    make([]*Conn, 0, size),
 		filling:  false,
 		closed:   false,
+		logger:   session.logger,
 	}
 
 	// the pool is not filled or connected
@@ -463,11 +466,11 @@ func (pool *hostConnPool) logConnectErr(err error) {
 		// connection refused
 		// these are typical during a node outage so avoid log spam.
 		if gocqlDebug {
-			Logger.Printf("unable to dial %q: %v\n", pool.host.ConnectAddress(), err)
+			pool.logger.Printf("unable to dial %q: %v\n", pool.host.ConnectAddress(), err)
 		}
 	} else if err != nil {
 		// unexpected error
-		Logger.Printf("error: failed to connect to %s due to error: %v", pool.addr, err)
+		pool.logger.Printf("error: failed to connect to %s due to error: %v", pool.addr, err)
 	}
 }
 
@@ -534,7 +537,7 @@ func (pool *hostConnPool) connect() (err error) {
 			}
 		}
 		if gocqlDebug {
-			Logger.Printf("connection failed %q: %v, reconnecting with %T\n",
+			pool.logger.Printf("connection failed %q: %v, reconnecting with %T\n",
 				pool.host.ConnectAddress(), err, reconnectionPolicy)
 		}
 		time.Sleep(reconnectionPolicy.GetInterval(i))
diff --git a/control.go b/control.go
index 9ab9c1c..c933697 100644
--- a/control.go
+++ b/control.go
@@ -177,7 +177,7 @@ func (c *controlConn) shuffleDial(endpoints []*HostInfo) (*Conn, error) {
 			return conn, nil
 		}
 
-		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", host.ConnectAddress(), host.Port(), err)
 	}
 
 	return nil, err
@@ -372,7 +372,7 @@ func (c *controlConn) reconnect(refreshring bool) {
 
 	if err := c.setupConn(newConn); err != nil {
 		newConn.Close()
-		Logger.Printf("gocql: control unable to register events: %v\n", err)
+		c.session.logger.Printf("gocql: control unable to register events: %v\n", err)
 		return
 	}
 
@@ -456,7 +456,7 @@ func (c *controlConn) query(statement string, values ...interface{}) (iter *Iter
 		})
 
 		if gocqlDebug && iter.err != nil {
-			Logger.Printf("control: error executing %q: %v\n", statement, iter.err)
+			c.session.logger.Printf("control: error executing %q: %v\n", statement, iter.err)
 		}
 
 		q.AddAttempts(1, c.getConn().host)
diff --git a/events.go b/events.go
index c633825..aaaed9e 100644
--- a/events.go
+++ b/events.go
@@ -14,14 +14,17 @@ type eventDebouncer struct {
 
 	callback func([]frame)
 	quit     chan struct{}
+
+	logger StdLogger
 }
 
-func newEventDebouncer(name string, eventHandler func([]frame)) *eventDebouncer {
+func newEventDebouncer(name string, eventHandler func([]frame), logger StdLogger) *eventDebouncer {
 	e := &eventDebouncer{
 		name:     name,
 		quit:     make(chan struct{}),
 		timer:    time.NewTimer(eventDebounceTime),
 		callback: eventHandler,
+		logger:   logger,
 	}
 	e.timer.Stop()
 	go e.flusher()
@@ -73,7 +76,7 @@ func (e *eventDebouncer) debounce(frame frame) {
 	if len(e.events) < eventBufferSize {
 		e.events = append(e.events, frame)
 	} else {
-		Logger.Printf("%s: buffer full, dropping event frame: %s", e.name, frame)
+		e.logger.Printf("%s: buffer full, dropping event frame: %s", e.name, frame)
 	}
 
 	e.mu.Unlock()
@@ -82,13 +85,12 @@ func (e *eventDebouncer) debounce(frame frame) {
 func (s *Session) handleEvent(framer *framer) {
 	frame, err := framer.parseFrame()
 	if err != nil {
-		// TODO: logger
-		Logger.Printf("gocql: unable to parse event frame: %v\n", err)
+		s.logger.Printf("gocql: unable to parse event frame: %v\n", err)
 		return
 	}
 
 	if gocqlDebug {
-		Logger.Printf("gocql: handling frame: %v\n", frame)
+		s.logger.Printf("gocql: handling frame: %v\n", frame)
 	}
 
 	switch f := frame.(type) {
@@ -99,7 +101,7 @@ func (s *Session) handleEvent(framer *framer) {
 	case *topologyChangeEventFrame, *statusChangeEventFrame:
 		s.nodeEvents.debounce(frame)
 	default:
-		Logger.Printf("gocql: invalid event frame (%T): %v\n", f, f)
+		s.logger.Printf("gocql: invalid event frame (%T): %v\n", f, f)
 	}
 }
 
@@ -159,7 +161,7 @@ func (s *Session) handleNodeEvent(frames []frame) {
 
 	for _, f := range events {
 		if gocqlDebug {
-			Logger.Printf("gocql: dispatching event: %+v\n", f)
+			s.logger.Printf("gocql: dispatching event: %+v\n", f)
 		}
 
 		switch f.change {
@@ -182,7 +184,7 @@ func (s *Session) addNewNode(ip net.IP, port int) {
 	// Get host info and apply any filters to the host
 	hostInfo, err := s.hostSource.getHostInfo(ip, port)
 	if err != nil {
-		Logger.Printf("gocql: events: unable to fetch host info for (%s:%d): %v\n", ip, port, err)
+		s.logger.Printf("gocql: events: unable to fetch host info for (%s:%d): %v\n", ip, port, err)
 		return
 	} else if hostInfo == nil {
 		// ignore if it's null because we couldn't find it
@@ -210,7 +212,7 @@ func (s *Session) addNewNode(ip net.IP, port int) {
 
 func (s *Session) handleNewNode(ip net.IP, port int) {
 	if gocqlDebug {
-		Logger.Printf("gocql: Session.handleNewNode: %s:%d\n", ip.String(), port)
+		s.logger.Printf("gocql: Session.handleNewNode: %s:%d\n", ip.String(), port)
 	}
 
 	ip, port = s.cfg.translateAddressPort(ip, port)
@@ -226,7 +228,7 @@ func (s *Session) handleNewNode(ip net.IP, port int) {
 
 func (s *Session) handleRemovedNode(ip net.IP, port int) {
 	if gocqlDebug {
-		Logger.Printf("gocql: Session.handleRemovedNode: %s:%d\n", ip.String(), port)
+		s.logger.Printf("gocql: Session.handleRemovedNode: %s:%d\n", ip.String(), port)
 	}
 
 	ip, port = s.cfg.translateAddressPort(ip, port)
@@ -251,7 +253,7 @@ func (s *Session) handleRemovedNode(ip net.IP, port int) {
 
 func (s *Session) handleNodeUp(eventIp net.IP, eventPort int) {
 	if gocqlDebug {
-		Logger.Printf("gocql: Session.handleNodeUp: %s:%d\n", eventIp.String(), eventPort)
+		s.logger.Printf("gocql: Session.handleNodeUp: %s:%d\n", eventIp.String(), eventPort)
 	}
 
 	ip, port := s.cfg.translateAddressPort(eventIp, eventPort)
@@ -271,7 +273,7 @@ func (s *Session) handleNodeUp(eventIp net.IP, eventPort int) {
 
 func (s *Session) handleNodeDown(ip net.IP, port int) {
 	if gocqlDebug {
-		Logger.Printf("gocql: Session.handleNodeDown: %s:%d\n", ip.String(), port)
+		s.logger.Printf("gocql: Session.handleNodeDown: %s:%d\n", ip.String(), port)
 	}
 
 	host := s.ring.getHost(ip)
diff --git a/filters.go b/filters.go
index 32e6ce6..ecd9c77 100644
--- a/filters.go
+++ b/filters.go
@@ -40,7 +40,7 @@ func DataCentreHostFilter(dataCentre string) HostFilter {
 // WhiteListHostFilter filters incoming hosts by checking that their address is
 // in the initial hosts whitelist.
 func WhiteListHostFilter(hosts ...string) HostFilter {
-	hostInfos, err := addrsToHosts(hosts, 9042)
+	hostInfos, err := addrsToHosts(hosts, 9042, nopLogger{})
 	if err != nil {
 		// dont want to panic here, but rather not break the API
 		panic(fmt.Errorf("unable to lookup host info from address: %v", err))
diff --git a/helpers.go b/helpers.go
index cafb508..142577c 100644
--- a/helpers.go
+++ b/helpers.go
@@ -130,38 +130,38 @@ func getCassandraBaseType(name string) Type {
 	}
 }
 
-func getCassandraType(name string) TypeInfo {
+func getCassandraType(name string, logger StdLogger) TypeInfo {
 	if strings.HasPrefix(name, "frozen<") {
-		return getCassandraType(strings.TrimPrefix(name[:len(name)-1], "frozen<"))
+		return getCassandraType(strings.TrimPrefix(name[:len(name)-1], "frozen<"), logger)
 	} else if strings.HasPrefix(name, "set<") {
 		return CollectionType{
 			NativeType: NativeType{typ: TypeSet},
-			Elem:       getCassandraType(strings.TrimPrefix(name[:len(name)-1], "set<")),
+			Elem:       getCassandraType(strings.TrimPrefix(name[:len(name)-1], "set<"), logger),
 		}
 	} else if strings.HasPrefix(name, "list<") {
 		return CollectionType{
 			NativeType: NativeType{typ: TypeList},
-			Elem:       getCassandraType(strings.TrimPrefix(name[:len(name)-1], "list<")),
+			Elem:       getCassandraType(strings.TrimPrefix(name[:len(name)-1], "list<"), logger),
 		}
 	} else if strings.HasPrefix(name, "map<") {
 		names := splitCompositeTypes(strings.TrimPrefix(name[:len(name)-1], "map<"))
 		if len(names) != 2 {
-			Logger.Printf("Error parsing map type, it has %d subelements, expecting 2\n", len(names))
+			logger.Printf("Error parsing map type, it has %d subelements, expecting 2\n", len(names))
 			return NativeType{
 				typ: TypeCustom,
 			}
 		}
 		return CollectionType{
 			NativeType: NativeType{typ: TypeMap},
-			Key:        getCassandraType(names[0]),
-			Elem:       getCassandraType(names[1]),
+			Key:        getCassandraType(names[0], logger),
+			Elem:       getCassandraType(names[1], logger),
 		}
 	} else if strings.HasPrefix(name, "tuple<") {
 		names := splitCompositeTypes(strings.TrimPrefix(name[:len(name)-1], "tuple<"))
 		types := make([]TypeInfo, len(names))
 
 		for i, name := range names {
-			types[i] = getCassandraType(name)
+			types[i] = getCassandraType(name, logger)
 		}
 
 		return TupleTypeInfo{
diff --git a/host_source.go b/host_source.go
index f6716ca..e426148 100644
--- a/host_source.go
+++ b/host_source.go
@@ -544,7 +544,7 @@ func (r *ringDescriber) getClusterPeerInfo() ([]*HostInfo, error) {
 			return nil, err
 		} else if !isValidPeer(host) {
 			// If it's not a valid peer
-			Logger.Printf("Found invalid peer '%s' "+
+			r.session.logger.Printf("Found invalid peer '%s' "+
 				"Likely due to a gossip or snitch issue, this host will be ignored", host)
 			continue
 		}
diff --git a/logger.go b/logger.go
index bd16d41..8ff6658 100644
--- a/logger.go
+++ b/logger.go
@@ -12,6 +12,14 @@ type StdLogger interface {
 	Println(v ...interface{})
 }
 
+type nopLogger struct{}
+
+func (n nopLogger) Print(_ ...interface{}) {}
+
+func (n nopLogger) Printf(_ string, _ ...interface{}) {}
+
+func (n nopLogger) Println(_ ...interface{}) {}
+
 type testLogger struct {
 	capture bytes.Buffer
 }
@@ -27,4 +35,6 @@ func (l *defaultLogger) Print(v ...interface{})                 { log.Print(v...
 func (l *defaultLogger) Printf(format string, v ...interface{}) { log.Printf(format, v...) }
 func (l *defaultLogger) Println(v ...interface{})               { log.Println(v...) }
 
+// Logger for logging messages.
+// Deprecated: Use ClusterConfig.Logger instead.
 var Logger StdLogger = &defaultLogger{}
diff --git a/metadata.go b/metadata.go
index 6cd2b4b..faf1ee8 100644
--- a/metadata.go
+++ b/metadata.go
@@ -290,7 +290,8 @@ func (s *schemaDescriber) refreshSchema(keyspaceName string) error {
 	}
 
 	// organize the schema data
-	compileMetadata(s.session.cfg.ProtoVersion, keyspace, tables, columns, functions, aggregates, views, materializedViews)
+	compileMetadata(s.session.cfg.ProtoVersion, keyspace, tables, columns, functions, aggregates, views,
+		materializedViews, s.session.logger)
 
 	// update the cache
 	s.cache[keyspaceName] = keyspace
@@ -312,6 +313,7 @@ func compileMetadata(
 	aggregates []AggregateMetadata,
 	views []ViewMetadata,
 	materializedViews []MaterializedViewMetadata,
+	logger StdLogger,
 ) {
 	keyspace.Tables = make(map[string]*TableMetadata)
 	for i := range tables {
@@ -357,13 +359,13 @@ func compileMetadata(
 		col := &columns[i]
 		// decode the validator for TypeInfo and order
 		if col.ClusteringOrder != "" { // Cassandra 3.x+
-			col.Type = getCassandraType(col.Validator)
+			col.Type = getCassandraType(col.Validator, logger)
 			col.Order = ASC
 			if col.ClusteringOrder == "desc" {
 				col.Order = DESC
 			}
 		} else {
-			validatorParsed := parseType(col.Validator)
+			validatorParsed := parseType(col.Validator, logger)
 			col.Type = validatorParsed.types[0]
 			col.Order = ASC
 			if validatorParsed.reversed[0] {
@@ -385,9 +387,9 @@ func compileMetadata(
 	}
 
 	if protoVersion == protoVersion1 {
-		compileV1Metadata(tables)
+		compileV1Metadata(tables, logger)
 	} else {
-		compileV2Metadata(tables)
+		compileV2Metadata(tables, logger)
 	}
 }
 
@@ -396,14 +398,14 @@ func compileMetadata(
 // column metadata as V2+ (because V1 doesn't support the "type" column in the
 // system.schema_columns table) so determining PartitionKey and ClusterColumns
 // is more complex.
-func compileV1Metadata(tables []TableMetadata) {
+func compileV1Metadata(tables []TableMetadata, logger StdLogger) {
 	for i := range tables {
 		table := &tables[i]
 
 		// decode the key validator
-		keyValidatorParsed := parseType(table.KeyValidator)
+		keyValidatorParsed := parseType(table.KeyValidator, logger)
 		// decode the comparator
-		comparatorParsed := parseType(table.Comparator)
+		comparatorParsed := parseType(table.Comparator, logger)
 
 		// the partition key length is the same as the number of types in the
 		// key validator
@@ -489,7 +491,7 @@ func compileV1Metadata(tables []TableMetadata) {
 				alias = table.ValueAlias
 			}
 			// decode the default validator
-			defaultValidatorParsed := parseType(table.DefaultValidator)
+			defaultValidatorParsed := parseType(table.DefaultValidator, logger)
 			column := &ColumnMetadata{
 				Keyspace: table.Keyspace,
 				Table:    table.Name,
@@ -503,7 +505,7 @@ func compileV1Metadata(tables []TableMetadata) {
 }
 
 // The simpler compile case for V2+ protocol
-func compileV2Metadata(tables []TableMetadata) {
+func compileV2Metadata(tables []TableMetadata, logger StdLogger) {
 	for i := range tables {
 		table := &tables[i]
 
@@ -511,7 +513,7 @@ func compileV2Metadata(tables []TableMetadata) {
 		table.ClusteringColumns = make([]*ColumnMetadata, clusteringColumnCount)
 
 		if table.KeyValidator != "" {
-			keyValidatorParsed := parseType(table.KeyValidator)
+			keyValidatorParsed := parseType(table.KeyValidator, logger)
 			table.PartitionKey = make([]*ColumnMetadata, len(keyValidatorParsed.types))
 		} else { // Cassandra 3.x+
 			partitionKeyCount := componentColumnCountOfType(table.Columns, ColumnPartitionKey)
@@ -921,11 +923,11 @@ func getColumnMetadata(session *Session, keyspaceName string) ([]ColumnMetadata,
 	return columns, nil
 }
 
-func getTypeInfo(t string) TypeInfo {
+func getTypeInfo(t string, logger StdLogger) TypeInfo {
 	if strings.HasPrefix(t, apacheCassandraTypePrefix) {
 		t = apacheToCassandraType(t)
 	}
-	return getCassandraType(t)
+	return getCassandraType(t, logger)
 }
 
 func getViewsMetadata(session *Session, keyspaceName string) ([]ViewMetadata, error) {
@@ -961,7 +963,7 @@ func getViewsMetadata(session *Session, keyspaceName string) ([]ViewMetadata, er
 		}
 		view.FieldTypes = make([]TypeInfo, len(argumentTypes))
 		for i, argumentType := range argumentTypes {
-			view.FieldTypes[i] = getTypeInfo(argumentType)
+			view.FieldTypes[i] = getTypeInfo(argumentType, session.logger)
 		}
 		views = append(views, view)
 	}
@@ -1082,10 +1084,10 @@ func getFunctionsMetadata(session *Session, keyspaceName string) ([]FunctionMeta
 		if err != nil {
 			return nil, err
 		}
-		function.ReturnType = getTypeInfo(returnType)
+		function.ReturnType = getTypeInfo(returnType, session.logger)
 		function.ArgumentTypes = make([]TypeInfo, len(argumentTypes))
 		for i, argumentType := range argumentTypes {
-			function.ArgumentTypes[i] = getTypeInfo(argumentType)
+			function.ArgumentTypes[i] = getTypeInfo(argumentType, session.logger)
 		}
 		functions = append(functions, function)
 	}
@@ -1139,11 +1141,11 @@ func getAggregatesMetadata(session *Session, keyspaceName string) ([]AggregateMe
 		if err != nil {
 			return nil, err
 		}
-		aggregate.ReturnType = getTypeInfo(returnType)
-		aggregate.StateType = getTypeInfo(stateType)
+		aggregate.ReturnType = getTypeInfo(returnType, session.logger)
+		aggregate.StateType = getTypeInfo(stateType, session.logger)
 		aggregate.ArgumentTypes = make([]TypeInfo, len(argumentTypes))
 		for i, argumentType := range argumentTypes {
-			aggregate.ArgumentTypes[i] = getTypeInfo(argumentType)
+			aggregate.ArgumentTypes[i] = getTypeInfo(argumentType, session.logger)
 		}
 		aggregates = append(aggregates, aggregate)
 	}
@@ -1157,8 +1159,9 @@ func getAggregatesMetadata(session *Session, keyspaceName string) ([]AggregateMe
 
 // type definition parser state
 type typeParser struct {
-	input string
-	index int
+	input  string
+	index  int
+	logger StdLogger
 }
 
 // the type definition parser result
@@ -1170,8 +1173,8 @@ type typeParserResult struct {
 }
 
 // Parse the type definition used for validator and comparator schema data
-func parseType(def string) typeParserResult {
-	parser := &typeParser{input: def}
+func parseType(def string, logger StdLogger) typeParserResult {
+	parser := &typeParser{input: def, logger: logger}
 	return parser.parse()
 }
 
@@ -1231,7 +1234,7 @@ func (t *typeParser) parse() typeParserResult {
 				var name string
 				decoded, err := hex.DecodeString(*param.name)
 				if err != nil {
-					Logger.Printf(
+					t.logger.Printf(
 						"Error parsing type '%s', contains collection name '%s' with an invalid format: %v",
 						t.input,
 						*param.name,
diff --git a/policies.go b/policies.go
index 06e1c13..1446dec 100644
--- a/policies.go
+++ b/policies.go
@@ -6,7 +6,6 @@ package gocql
 
 //This file will be the future home for more policies
 
-
 import (
 	"context"
 	"errors"
@@ -216,11 +215,6 @@ func (d *DowngradingConsistencyRetryPolicy) Attempt(q RetryableQuery) bool {
 		return false
 	} else if currentAttempt > 0 {
 		q.SetConsistency(d.ConsistencyLevelsToTry[currentAttempt-1])
-		if gocqlDebug {
-			Logger.Printf("%T: set consistency to %q\n",
-				d,
-				d.ConsistencyLevelsToTry[currentAttempt-1])
-		}
 	}
 	return true
 }
@@ -389,11 +383,14 @@ type tokenAwareHostPolicy struct {
 	hosts       cowHostList
 	partitioner string
 	metadata    atomic.Value // *clusterMeta
+
+	logger StdLogger
 }
 
 func (t *tokenAwareHostPolicy) Init(s *Session) {
 	t.getKeyspaceMetadata = s.KeyspaceMetadata
 	t.getKeyspaceName = func() string { return s.cfg.Keyspace }
+	t.logger = s.logger
 }
 
 func (t *tokenAwareHostPolicy) IsLocal(host *HostInfo) bool {
@@ -416,7 +413,7 @@ func (t *tokenAwareHostPolicy) updateReplicas(meta *clusterMeta, keyspace string
 
 	ks, err := t.getKeyspaceMetadata(keyspace)
 	if err == nil {
-		strat := getStrategy(ks)
+		strat := getStrategy(ks, t.logger)
 		if strat != nil {
 			if meta != nil && meta.tokenRing != nil {
 				newReplicas[keyspace] = strat.replicaMap(meta.tokenRing)
@@ -441,7 +438,7 @@ func (t *tokenAwareHostPolicy) SetPartitioner(partitioner string) {
 		t.fallback.SetPartitioner(partitioner)
 		t.partitioner = partitioner
 		meta := t.getMetadataForUpdate()
-		meta.resetTokenRing(t.partitioner, t.hosts.get())
+		meta.resetTokenRing(t.partitioner, t.hosts.get(), t.logger)
 		t.updateReplicas(meta, t.getKeyspaceName())
 		t.metadata.Store(meta)
 	}
@@ -451,7 +448,7 @@ func (t *tokenAwareHostPolicy) AddHost(host *HostInfo) {
 	t.mu.Lock()
 	if t.hosts.add(host) {
 		meta := t.getMetadataForUpdate()
-		meta.resetTokenRing(t.partitioner, t.hosts.get())
+		meta.resetTokenRing(t.partitioner, t.hosts.get(), t.logger)
 		t.updateReplicas(meta, t.getKeyspaceName())
 		t.metadata.Store(meta)
 	}
@@ -468,7 +465,7 @@ func (t *tokenAwareHostPolicy) AddHosts(hosts []*HostInfo) {
 	}
 
 	meta := t.getMetadataForUpdate()
-	meta.resetTokenRing(t.partitioner, t.hosts.get())
+	meta.resetTokenRing(t.partitioner, t.hosts.get(), t.logger)
 	t.updateReplicas(meta, t.getKeyspaceName())
 	t.metadata.Store(meta)
 
@@ -483,7 +480,7 @@ func (t *tokenAwareHostPolicy) RemoveHost(host *HostInfo) {
 	t.mu.Lock()
 	if t.hosts.remove(host.ConnectAddress()) {
 		meta := t.getMetadataForUpdate()
-		meta.resetTokenRing(t.partitioner, t.hosts.get())
+		meta.resetTokenRing(t.partitioner, t.hosts.get(), t.logger)
 		t.updateReplicas(meta, t.getKeyspaceName())
 		t.metadata.Store(meta)
 	}
@@ -523,7 +520,7 @@ func (t *tokenAwareHostPolicy) getMetadataForUpdate() *clusterMeta {
 
 // resetTokenRing creates a new tokenRing.
 // It must be called with t.mu locked.
-func (m *clusterMeta) resetTokenRing(partitioner string, hosts []*HostInfo) {
+func (m *clusterMeta) resetTokenRing(partitioner string, hosts []*HostInfo, logger StdLogger) {
 	if partitioner == "" {
 		// partitioner not yet set
 		return
@@ -532,7 +529,7 @@ func (m *clusterMeta) resetTokenRing(partitioner string, hosts []*HostInfo) {
 	// create a new token ring
 	tokenRing, err := newTokenRing(partitioner, hosts)
 	if err != nil {
-		Logger.Printf("Unable to update the token ring due to error: %s", err)
+		logger.Printf("Unable to update the token ring due to error: %s", err)
 		return
 	}
 
diff --git a/session.go b/session.go
index 27aca92..1353870 100644
--- a/session.go
+++ b/session.go
@@ -72,6 +72,8 @@ type Session struct {
 
 	closeMu  sync.RWMutex
 	isClosed bool
+
+	logger StdLogger
 }
 
 var queryPool = &sync.Pool{
@@ -80,14 +82,14 @@ var queryPool = &sync.Pool{
 	},
 }
 
-func addrsToHosts(addrs []string, defaultPort int) ([]*HostInfo, error) {
+func addrsToHosts(addrs []string, defaultPort int, logger StdLogger) ([]*HostInfo, error) {
 	var hosts []*HostInfo
 	for _, hostport := range addrs {
 		resolvedHosts, err := hostInfo(hostport, defaultPort)
 		if err != nil {
 			// Try other hosts if unable to resolve DNS name
 			if _, ok := err.(*net.DNSError); ok {
-				Logger.Printf("gocql: dns error: %v\n", err)
+				logger.Printf("gocql: dns error: %v\n", err)
 				continue
 			}
 			return nil, err
@@ -125,12 +127,13 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 		connectObserver: cfg.ConnectObserver,
 		ctx:             ctx,
 		cancel:          cancel,
+		logger:          cfg.logger(),
 	}
 
 	s.schemaDescriber = newSchemaDescriber(s)
 
-	s.nodeEvents = newEventDebouncer("NodeEvents", s.handleNodeEvent)
-	s.schemaEvents = newEventDebouncer("SchemaEvents", s.handleSchemaEvent)
+	s.nodeEvents = newEventDebouncer("NodeEvents", s.handleNodeEvent, s.logger)
+	s.schemaEvents = newEventDebouncer("SchemaEvents", s.handleSchemaEvent, s.logger)
 
 	s.routingKeyInfoCache.lru = lru.New(cfg.MaxRoutingKeyInfo)
 
@@ -178,7 +181,7 @@ func NewSession(cfg ClusterConfig) (*Session, error) {
 }
 
 func (s *Session) init() error {
-	hosts, err := addrsToHosts(s.cfg.Hosts, s.cfg.Port)
+	hosts, err := addrsToHosts(s.cfg.Hosts, s.cfg.Port, s.logger)
 	if err != nil {
 		return err
 	}
@@ -347,7 +350,7 @@ func (s *Session) reconnectDownedHosts(intv time.Duration) {
 				for _, h := range hosts {
 					buf.WriteString("[" + h.ConnectAddress().String() + ":" + h.State().String() + "]")
 				}
-				Logger.Println(buf.String())
+				s.logger.Println(buf.String())
 			}
 
 			for _, h := range hosts {
diff --git a/topology.go b/topology.go
index a46c665..a154f0f 100644
--- a/topology.go
+++ b/topology.go
@@ -66,12 +66,12 @@ func getReplicationFactorFromOpts(val interface{}) (int, error) {
 	}
 }
 
-func getStrategy(ks *KeyspaceMetadata) placementStrategy {
+func getStrategy(ks *KeyspaceMetadata, logger StdLogger) placementStrategy {
 	switch {
 	case strings.Contains(ks.StrategyClass, "SimpleStrategy"):
 		rf, err := getReplicationFactorFromOpts(ks.StrategyOptions["replication_factor"])
 		if err != nil {
-			Logger.Printf("parse rf for keyspace %q: %v", ks.Name, err)
+			logger.Printf("parse rf for keyspace %q: %v", ks.Name, err)
 			return nil
 		}
 		return &simpleStrategy{rf: rf}
@@ -84,7 +84,7 @@ func getStrategy(ks *KeyspaceMetadata) placementStrategy {
 
 			rf, err := getReplicationFactorFromOpts(rf)
 			if err != nil {
-				Logger.Println("parse rf for keyspace %q, dc %q: %v", err)
+				logger.Println("parse rf for keyspace %q, dc %q: %v", err)
 				// skip DC if the rf is invalid/unsupported, so that we can at least work with other working DCs.
 				continue
 			}
@@ -95,7 +95,7 @@ func getStrategy(ks *KeyspaceMetadata) placementStrategy {
 	case strings.Contains(ks.StrategyClass, "LocalStrategy"):
 		return nil
 	default:
-		Logger.Printf("parse rf for keyspace %q: unsupported strategy class: %v", ks.StrategyClass)
+		logger.Printf("parse rf for keyspace %q: unsupported strategy class: %v", ks.StrategyClass)
 		return nil
 	}
 }


Test patch:
diff --git a/conn_test.go b/conn_test.go
index 9f1f64b..801afa6 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -151,10 +151,6 @@ func newTestSession(proto protoVersion, addresses ...string) (*Session, error) {
 
 func TestDNSLookupConnected(t *testing.T) {
 	log := &testLogger{}
-	Logger = log
-	defer func() {
-		Logger = &defaultLogger{}
-	}()
 
 	// Override the defaul DNS resolver and restore at the end
 	failDNS = true
@@ -164,6 +160,7 @@ func TestDNSLookupConnected(t *testing.T) {
 	defer srv.Stop()
 
 	cluster := NewCluster("cassandra1.invalid", srv.Address, "cassandra2.invalid")
+	cluster.Logger = log
 	cluster.ProtoVersion = int(defaultProto)
 	cluster.disableControlConn = true
 
@@ -181,16 +178,13 @@ func TestDNSLookupConnected(t *testing.T) {
 
 func TestDNSLookupError(t *testing.T) {
 	log := &testLogger{}
-	Logger = log
-	defer func() {
-		Logger = &defaultLogger{}
-	}()
 
 	// Override the defaul DNS resolver and restore at the end
 	failDNS = true
 	defer func() { failDNS = false }()
 
 	cluster := NewCluster("cassandra1.invalid", "cassandra2.invalid")
+	cluster.Logger = log
 	cluster.ProtoVersion = int(defaultProto)
 	cluster.disableControlConn = true
 
@@ -213,10 +207,6 @@ func TestDNSLookupError(t *testing.T) {
 func TestStartupTimeout(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	log := &testLogger{}
-	Logger = log
-	defer func() {
-		Logger = &defaultLogger{}
-	}()
 
 	srv := NewTestServer(t, defaultProto, ctx)
 	defer srv.Stop()
@@ -226,6 +216,7 @@ func TestStartupTimeout(t *testing.T) {
 
 	startTime := time.Now()
 	cluster := NewCluster(srv.Address)
+	cluster.Logger = log
 	cluster.ProtoVersion = int(defaultProto)
 	cluster.disableControlConn = true
 	// Set very long query connection timeout
@@ -323,13 +314,14 @@ func TestCancel(t *testing.T) {
 type testQueryObserver struct {
 	metrics map[string]*hostMetrics
 	verbose bool
+	logger  StdLogger
 }
 
 func (o *testQueryObserver) ObserveQuery(ctx context.Context, q ObservedQuery) {
 	host := q.Host.ConnectAddress().String()
 	o.metrics[host] = q.Metrics
 	if o.verbose {
-		Logger.Printf("Observed query %q. Returned %v rows, took %v on host %q with %v attempts and total latency %v. Error: %q\n",
+		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)
 	}
 }
@@ -383,9 +375,7 @@ func TestQueryRetry(t *testing.T) {
 
 func TestQueryMultinodeWithMetrics(t *testing.T) {
 	log := &testLogger{}
-	Logger = log
 	defer func() {
-		Logger = &defaultLogger{}
 		os.Stdout.WriteString(log.String())
 	}()
 
@@ -412,7 +402,7 @@ func TestQueryMultinodeWithMetrics(t *testing.T) {
 
 	// 1 retry per host
 	rt := &SimpleRetryPolicy{NumRetries: 3}
-	observer := &testQueryObserver{metrics: make(map[string]*hostMetrics), verbose: false}
+	observer := &testQueryObserver{metrics: make(map[string]*hostMetrics), verbose: false, logger: log}
 	qry := db.Query("kill").RetryPolicy(rt).Observer(observer)
 	if err := qry.Exec(); err == nil {
 		t.Fatalf("expected error")
@@ -460,9 +450,7 @@ func (t *testRetryPolicy) GetRetryType(err error) RetryType {
 
 func TestSpeculativeExecution(t *testing.T) {
 	log := &testLogger{}
-	Logger = log
 	defer func() {
-		Logger = &defaultLogger{}
 		os.Stdout.WriteString(log.String())
 	}()
 
@@ -689,6 +677,7 @@ func TestStream0(t *testing.T) {
 	conn := &Conn{
 		r:       bufio.NewReader(&buf),
 		streams: streams.New(protoVersion4),
+		logger:  &defaultLogger{},
 	}
 
 	err := conn.recv(context.Background())
diff --git a/events_test.go b/events_test.go
index c56bc07..37aefdc 100644
--- a/events_test.go
+++ b/events_test.go
@@ -15,7 +15,7 @@ func TestEventDebounce(t *testing.T) {
 	debouncer := newEventDebouncer("testDebouncer", func(events []frame) {
 		defer wg.Done()
 		eventsSeen += len(events)
-	})
+	}, &defaultLogger{})
 	defer debouncer.stop()
 
 	for i := 0; i < eventCount; i++ {
diff --git a/helpers_test.go b/helpers_test.go
index ff057bd..0f6cedc 100644
--- a/helpers_test.go
+++ b/helpers_test.go
@@ -6,7 +6,7 @@ import (
 )
 
 func TestGetCassandraType_Set(t *testing.T) {
-	typ := getCassandraType("set<text>")
+	typ := getCassandraType("set<text>", &defaultLogger{})
 	set, ok := typ.(CollectionType)
 	if !ok {
 		t.Fatalf("expected CollectionType got %T", typ)
@@ -203,7 +203,7 @@ func TestGetCassandraType(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.input, func(t *testing.T) {
-			got := getCassandraType(test.input)
+			got := getCassandraType(test.input, &defaultLogger{})
 
 			// TODO(zariel): define an equal method on the types?
 			if !reflect.DeepEqual(got, test.exp) {
diff --git a/metadata_test.go b/metadata_test.go
index c874864..d9651e0 100644
--- a/metadata_test.go
+++ b/metadata_test.go
@@ -13,6 +13,7 @@ import (
 // from metadata schema queries (see getKeyspaceMetadata, getTableMetadata, and getColumnMetadata)
 func TestCompileMetadata(t *testing.T) {
 	// V1 tests - these are all based on real examples from the integration test ccm cluster
+	log := &defaultLogger{}
 	keyspace := &KeyspaceMetadata{
 		Name: "V1Keyspace",
 	}
@@ -94,7 +95,7 @@ func TestCompileMetadata(t *testing.T) {
 		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "schema_version", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.UUIDType"},
 		{Keyspace: "V1Keyspace", Table: "peers", Kind: ColumnRegular, Name: "tokens", ComponentIndex: 0, Validator: "org.apache.cassandra.db.marshal.SetType(org.apache.cassandra.db.marshal.UTF8Type)"},
 	}
-	compileMetadata(1, keyspace, tables, columns, nil, nil, nil, nil)
+	compileMetadata(1, keyspace, tables, columns, nil, nil, nil, nil, log)
 	assertKeyspaceMetadata(
 		t,
 		keyspace,
@@ -375,7 +376,7 @@ func TestCompileMetadata(t *testing.T) {
 			Validator: "org.apache.cassandra.db.marshal.UTF8Type",
 		},
 	}
-	compileMetadata(2, keyspace, tables, columns, nil, nil, nil, nil)
+	compileMetadata(2, keyspace, tables, columns, nil, nil, nil, nil, log)
 	assertKeyspaceMetadata(
 		t,
 		keyspace,
@@ -674,7 +675,8 @@ func assertParseNonCompositeType(
 	typeExpected assertTypeInfo,
 ) {
 
-	result := parseType(def)
+	log := &defaultLogger{}
+	result := parseType(def, log)
 	if len(result.reversed) != 1 {
 		t.Errorf("%s expected %d reversed values but there were %d", def, 1, len(result.reversed))
 	}
@@ -704,7 +706,8 @@ func assertParseCompositeType(
 	collectionsExpected map[string]assertTypeInfo,
 ) {
 
-	result := parseType(def)
+	log := &defaultLogger{}
+	result := parseType(def, log)
 	if len(result.reversed) != len(typesExpected) {
 		t.Errorf("%s expected %d reversed values but there were %d", def, len(typesExpected), len(result.reversed))
 	}
diff --git a/session_test.go b/session_test.go
index cbc64f9..5051aed 100644
--- a/session_test.go
+++ b/session_test.go
@@ -16,6 +16,7 @@ func TestSessionAPI(t *testing.T) {
 		cfg:    *cfg,
 		cons:   Quorum,
 		policy: RoundRobinHostPolicy(),
+		logger: cfg.logger(),
 	}
 
 	s.pool = cfg.PoolConfig.buildPool(s)
@@ -184,8 +185,9 @@ func TestBatchBasicAPI(t *testing.T) {
 	cfg := &ClusterConfig{RetryPolicy: &SimpleRetryPolicy{NumRetries: 2}}
 
 	s := &Session{
-		cfg:  *cfg,
-		cons: Quorum,
+		cfg:    *cfg,
+		cons:   Quorum,
+		logger: cfg.logger(),
 	}
 	defer s.Close()