Make QueryMetrics (Latency and Attempts) opt-in.

Patch by João Reis; reviewed by TBD for CASSGO-90

Author: joao-r-reis

CHANGELOG.md

@@ -36,6 +36,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 #### 2.0.0
 
 - Remove release date from changelog and add 2.0.0-rc1 (CASSGO-86)
+- Make Query Metrics (Latency and Attempts) opt-in (CASSGO-90)
 
 #### 2.0.0-rc1
 

cluster.go

@@ -280,6 +280,11 @@ type ClusterConfig struct {
 	// default: 0.25.
 	NextPagePrefetch float64
 
+	// WithQueryMetrics sets whether query metrics should be enabled for all queries by default.
+	// Query metrics are available through Iter.Latency and Iter.Attempts.
+	// default: false.
+	WithQueryMetrics bool
+
 	// RegisteredTypes will be copied for all sessions created from this Cluster.
 	// If not provided, a copy of GlobalTypes will be used.
 	RegisteredTypes *RegisteredTypes

conn.go

@@ -1665,7 +1665,11 @@ func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter {
 			newQry := new(internalQuery)
 			*newQry = *q
 			newQry.pageState = copyBytes(x.meta.pagingState)
-			newQry.metrics = &queryMetrics{m: make(map[string]*hostMetrics)}
+			if qryOpts.withQueryMetrics {
+				newQry.metrics = &queryMetricsImpl{m: make(map[string]*hostMetrics)}
+			} else {
+				newQry.metrics = nilQueryMetrics
+			}
 
 			iter.next = &nextIter{
 				q:   newQry,

conn_test.go

@@ -348,7 +348,7 @@ func TestCancel(t *testing.T) {
 }
 
 type testQueryObserver struct {
-	metrics map[string]*hostMetrics
+	metrics map[string]hostMetrics
 	logger  StructuredLogger
 }
 
@@ -365,7 +365,7 @@ func (o *testQueryObserver) ObserveQuery(ctx context.Context, q ObservedQuery) {
 		newLogFieldError("err", q.Err))
 }
 
-func (o *testQueryObserver) GetMetrics(host *HostInfo) *hostMetrics {
+func (o *testQueryObserver) GetMetrics(host *HostInfo) hostMetrics {
 	return o.metrics[host.ConnectAddress().String()]
 }
 
@@ -395,7 +395,7 @@ func TestQueryRetry(t *testing.T) {
 
 	rt := &SimpleRetryPolicy{NumRetries: 1}
 
-	qry := db.Query("kill").RetryPolicy(rt)
+	qry := db.Query("kill").WithQueryMetrics(true).RetryPolicy(rt)
 	iter := qry.Iter()
 	err = iter.Close()
 	if err == nil {
@@ -443,8 +443,8 @@ func TestQueryMultinodeWithMetrics(t *testing.T) {
 
 	// 1 retry per host
 	rt := &SimpleRetryPolicy{NumRetries: 3}
-	observer := &testQueryObserver{metrics: make(map[string]*hostMetrics), logger: log}
-	qry := db.Query("kill").RetryPolicy(rt).Observer(observer).Idempotent(true)
+	observer := &testQueryObserver{metrics: make(map[string]hostMetrics), logger: log}
+	qry := db.Query("kill").RetryPolicy(rt).Observer(observer).Idempotent(true).WithQueryMetrics(true)
 	iter := qry.Iter()
 	err = iter.Close()
 	if err == nil {
@@ -453,7 +453,7 @@ func TestQueryMultinodeWithMetrics(t *testing.T) {
 
 	for i, ip := range addresses {
 		host := &HostInfo{connectAddress: net.ParseIP(ip)}
-		queryMetric := iter.metrics.hostMetrics(host)
+		queryMetric := iter.metrics.(*queryMetricsImpl).hostMetrics(host)
 		observedMetrics := observer.GetMetrics(host)
 
 		requests := int(atomic.LoadInt64(&nodes[i].nKillReq))

control.go

@@ -582,7 +582,7 @@ func (c *controlConn) query(statement string, values ...interface{}) (iter *Iter
 				newLogFieldString("statement", statement), newLogFieldError("err", iter.err))
 		}
 
-		iter.metrics.attempt(1, 0, c.getConn().host, false)
+		iter.metrics.attempt(1, 0, c.getConn().host)
 		if iter.err == nil || !c.retry.Attempt(qry) {
 			break
 		}

query_executor.go

@@ -61,7 +61,7 @@ type internalRequest interface {
 	attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo)
 	retryPolicy() RetryPolicy
 	speculativeExecutionPolicy() SpeculativeExecutionPolicy
-	getQueryMetrics() *queryMetrics
+	getQueryMetrics() queryMetrics
 	getRoutingInfo() *queryRoutingInfo
 	getKeyspaceFunc() func() string
 	RetryableQuery
@@ -280,6 +280,8 @@ type queryOptions struct {
 	// Protocol flag
 	disableSkipMetadata bool
 
+	withQueryMetrics bool
+
 	customPayload     map[string][]byte
 	prefetch          float64
 	rt                RetryPolicy
@@ -330,14 +332,15 @@ func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
 		nowInSecondsValue:     q.nowInSecondsValue,
 		keyspace:              q.keyspace,
 		hostID:                q.hostID,
+		withQueryMetrics:      q.withQueryMetrics,
 	}
 }
 
 type internalQuery struct {
 	originalQuery *Query
 	qryOpts       *queryOptions
 	pageState     []byte
-	metrics       *queryMetrics
+	metrics       queryMetrics
 	conn          *Conn
 	consistency   uint32
 	session       *Session
@@ -351,10 +354,16 @@ func newInternalQuery(q *Query, ctx context.Context) *internalQuery {
 		newPageState = make([]byte, len(pageState))
 		copy(newPageState, pageState)
 	}
+	var metrics queryMetrics
+	if q.withQueryMetrics {
+		metrics = &queryMetricsImpl{m: make(map[string]*hostMetrics)}
+	} else {
+		metrics = nilQueryMetrics
+	}
 	return &internalQuery{
 		originalQuery: q,
 		qryOpts:       newQueryOptions(q, ctx),
-		metrics:       &queryMetrics{m: make(map[string]*hostMetrics)},
+		metrics:       metrics,
 		consistency:   uint32(q.initialConsistency),
 		pageState:     newPageState,
 		conn:          nil,
@@ -370,7 +379,7 @@ func (q *internalQuery) Attempts() int {
 
 func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
 	latency := end.Sub(start)
-	attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+	attempt, metricsForHost := q.metrics.attempt(1, latency, host)
 
 	if q.qryOpts.observer != nil {
 		q.qryOpts.observer.ObserveQuery(q.qryOpts.context, ObservedQuery{
@@ -381,7 +390,7 @@ func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Ite
 			End:       end,
 			Rows:      iter.numRows,
 			Host:      host,
-			Metrics:   metricsForHost,
+			Metrics:   *metricsForHost,
 			Err:       iter.err,
 			Attempt:   attempt,
 			Query:     q.originalQuery,
@@ -457,7 +466,7 @@ func (q *internalQuery) IsIdempotent() bool {
 	return q.qryOpts.idempotent
 }
 
-func (q *internalQuery) getQueryMetrics() *queryMetrics {
+func (q *internalQuery) getQueryMetrics() queryMetrics {
 	return q.metrics
 }
 
@@ -548,17 +557,23 @@ func newBatchOptions(b *Batch, ctx context.Context) *batchOptions {
 type internalBatch struct {
 	originalBatch *Batch
 	batchOpts     *batchOptions
-	metrics       *queryMetrics
+	metrics       queryMetrics
 	consistency   uint32
 	routingInfo   *queryRoutingInfo
 	session       *Session
 }
 
 func newInternalBatch(batch *Batch, ctx context.Context) *internalBatch {
+	var metrics queryMetrics
+	if batch.withQueryMetrics {
+		metrics = &queryMetricsImpl{m: make(map[string]*hostMetrics)}
+	} else {
+		metrics = nilQueryMetrics
+	}
 	return &internalBatch{
 		originalBatch: batch,
 		batchOpts:     newBatchOptions(batch, ctx),
-		metrics:       &queryMetrics{m: make(map[string]*hostMetrics)},
+		metrics:       metrics,
 		routingInfo:   &queryRoutingInfo{},
 		session:       batch.session,
 		consistency:   uint32(batch.GetConsistency()),
@@ -572,7 +587,7 @@ func (b *internalBatch) Attempts() int {
 
 func (b *internalBatch) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
 	latency := end.Sub(start)
-	attempt, metricsForHost := b.metrics.attempt(1, latency, host, b.batchOpts.observer != nil)
+	attempt, metricsForHost := b.metrics.attempt(1, latency, host)
 
 	if b.batchOpts.observer == nil {
 		return
@@ -594,7 +609,7 @@ func (b *internalBatch) attempt(keyspace string, end, start time.Time, iter *Ite
 		End:        end,
 		// Rows not used in batch observations // TODO - might be able to support it when using BatchCAS
 		Host:    host,
-		Metrics: metricsForHost,
+		Metrics: *metricsForHost,
 		Err:     iter.err,
 		Attempt: attempt,
 		Batch:   b.originalBatch,
@@ -651,7 +666,7 @@ func (b *internalBatch) IsIdempotent() bool {
 	return b.batchOpts.idempotent
 }
 
-func (b *internalBatch) getQueryMetrics() *queryMetrics {
+func (b *internalBatch) getQueryMetrics() queryMetrics {
 	return b.metrics
 }