This is an automated email from the ASF dual-hosted git repository.
joao-r-reis pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-gocql-driver.git
The following commit(s) were added to refs/heads/trunk by this push:
new 120af460 Propagate bytes transmitted and received into Observed*
120af460 is described below
commit 120af4602464d86b1a0299a9a6b4693bd9adbfe1
Author: Alex Benn <[email protected]>
AuthorDate: Wed Jul 22 16:44:58 2026 -0400
Propagate bytes transmitted and received into Observed*
Clients of the GoCQL driver lib may need to attribute traffic to individual
queries and batches, for example when a multitenant service shares one
session across tenants. This can surface anomalies and help clients better
understand usage patterns and load.
BytesTx and BytesRx count uncompressed message bodies rather than bytes on
the wire. With protocol v5, a segment can carry frames for multiple
requests,
so its wire size cannot be attributed to a single request without
approximation. Frame and segment headers and trailers are excluded for the
same reason.
Patch by Alex Benn ([email protected]; abenn135); reviewed by João Reis
(joao-r-reis), Bohdan Siryk (worryg0d) for CASSGO-136
Assisted-by: Claude Code:claude-opus-5-5
---
byte_metrics_test.go | 581 +++++++++++++++++++++++++++++++++++++++++++++++++++
conn.go | 77 +++++--
conn_test.go | 25 +++
frame.go | 5 +
query_executor.go | 81 ++++++-
session.go | 33 ++-
6 files changed, 768 insertions(+), 34 deletions(-)
diff --git a/byte_metrics_test.go b/byte_metrics_test.go
new file mode 100644
index 00000000..5e991148
--- /dev/null
+++ b/byte_metrics_test.go
@@ -0,0 +1,581 @@
+//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
+ * 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 (
+ "bufio"
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "net"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/apache/cassandra-gocql-driver/v2/internal/streams"
+ "github.com/apache/cassandra-gocql-driver/v2/lz4"
+ "github.com/apache/cassandra-gocql-driver/v2/snappy"
+ "github.com/stretchr/testify/require"
+)
+
+type unitFuncBatchObserver func(context.Context, ObservedBatch)
+
+func (f unitFuncBatchObserver) ObserveBatch(ctx context.Context, o
ObservedBatch) { f(ctx, o) }
+
+func TestObserveByteMetricsPrepareAndRecovery(t *testing.T) {
+ for _, batch := range []bool{false, true} {
+ for _, recoverUnprepared := range []bool{false, true} {
+ t.Run(fmt.Sprintf("batch=%t/reprepare=%t", batch,
recoverUnprepared), func(t *testing.T) {
+ var mu sync.Mutex
+ var expected byteCounts
+ var requests int
+ var rejected bool
+ srv := newTestServerOpts{
+ addr: "127.0.0.1:0", protocol:
protoVersion4,
+ customRequestHandler: func(_
*TestServer, req, resp *framer) error {
+ op := req.header.op
+ switch op {
+ case opStartup:
+ resp.writeHeader(0,
opReady, req.header.stream)
+ case opOptions:
+ resp.writeHeader(0,
opSupported, req.header.stream)
+ resp.writeShort(0)
+ case opPrepare:
+ resp.writeHeader(0,
opResult, req.header.stream)
+
resp.writeInt(resultKindPrepared)
+
resp.writeShortBytes([]byte{1})
+ resp.writeInt(0) //
prepared metadata flags
+ resp.writeInt(0) //
bound columns
+ resp.writeInt(0) //
partition key columns
+ resp.writeInt(0) //
result metadata flags
+ resp.writeInt(0) //
result columns
+ case opExecute, opBatch:
+ if recoverUnprepared &&
!rejected {
+ rejected = true
+
resp.writeHeader(0, opError, req.header.stream)
+
resp.writeInt(ErrCodeUnprepared)
+
resp.writeString("unprepared")
+
resp.writeShortBytes([]byte{1})
+ } else {
+
resp.writeHeader(0, opResult, req.header.stream)
+
resp.writeInt(resultKindVoid)
+ }
+ default:
+ return
fmt.Errorf("unexpected opcode %v", op)
+ }
+ if op == opPrepare || op ==
opExecute || op == opBatch {
+ mu.Lock()
+ requests++
+ expected.bytesTx +=
int64(req.header.length)
+ expected.bytesRx +=
int64(len(resp.buf) - frameHeadSize)
+ mu.Unlock()
+ }
+ return nil
+ },
+ }.newServer(t, context.Background())
+ defer srv.Stop()
+ cluster := testCluster(protoVersion4,
srv.Address)
+ cluster.NumConns = 1
+ db, err := cluster.CreateSession()
+ require.NoError(t, err)
+ defer db.Close()
+
+ var observed byteCounts
+ var events int
+ var execute func() error
+ if batch {
+ b := db.NewBatch(LoggedBatch)
+ b.Bind("select value from table",
func(*QueryInfo) ([]interface{}, error) { return nil, nil })
+ b.Observer(unitFuncBatchObserver(func(_
context.Context, o ObservedBatch) {
+ events++
+ observed =
byteCounts{o.BytesTx, o.BytesRx}
+ }))
+ execute = b.Exec
+ } else {
+ q := db.Query("select value from
table").RoutingKey([]byte{1})
+ q.Observer(unitFuncQueryObserver(func(_
context.Context, o ObservedQuery) {
+ events++
+ observed =
byteCounts{o.BytesTx, o.BytesRx}
+ }))
+ execute = q.Exec
+ }
+
+ // One observation must include PREPARE and all
UNPREPARED recovery
+ // exchanges, even though only the final
response reaches the iterator.
+ require.NoError(t, execute())
+ require.Equal(t, 1, events)
+ mu.Lock()
+ want := expected
+ requestCount := requests
+ expected = byteCounts{}
+ requests = 0
+ mu.Unlock()
+ require.Equal(t, want, observed)
+ wantRequests := 2
+ if recoverUnprepared {
+ wantRequests = 4
+ }
+ require.Equal(t, wantRequests, requestCount)
+
+ // A separate execution hits the prepared cache
and starts fresh.
+ require.NoError(t, execute())
+ require.Equal(t, 2, events)
+ mu.Lock()
+ want = expected
+ requestCount = requests
+ mu.Unlock()
+ require.Equal(t, want, observed)
+ require.Equal(t, 1, requestCount)
+ })
+ }
+ }
+}
+
+type sameHostByteMetricsRetryPolicy struct{ SimpleRetryPolicy }
+
+func (*sameHostByteMetricsRetryPolicy) GetRetryType(error) RetryType { return
Retry }
+
+func TestObserveByteMetricsRetries(t *testing.T) {
+ srv := NewTestServer(t, protoVersion4, context.Background())
+ defer srv.Stop()
+ db, err := newTestSession(protoVersion4, srv.Address)
+ require.NoError(t, err)
+ defer db.Close()
+ var events []ObservedQuery
+ observer := unitFuncQueryObserver(func(_ context.Context, o
ObservedQuery) { events = append(events, o) })
+ err =
db.Query("kill").Idempotent(true).RetryPolicy(&sameHostByteMetricsRetryPolicy{SimpleRetryPolicy{NumRetries:
2}}).Observer(observer).Exec()
+ require.Error(t, err)
+ require.Len(t, events, 3)
+ for i, o := range events {
+ require.Equal(t, i, o.Attempt)
+ require.Equal(t, events[0].BytesTx, o.BytesTx)
+ require.Equal(t, events[0].BytesRx, o.BytesRx)
+ require.Positive(t, o.BytesTx)
+ require.Equal(t, int64(4+2+len("query killed")), o.BytesRx) //
error code and message
+ }
+}
+
+type repeatingByteMetricsHostPolicy struct{ HostSelectionPolicy }
+
+func (p repeatingByteMetricsHostPolicy) Pick(q ExecutableStatement) NextHost {
+ host := p.HostSelectionPolicy.Pick(q)()
+ return func() SelectedHost { return host }
+}
+
+func TestObserveByteMetricsSpeculation(t *testing.T) {
+ srv := NewTestServer(t, protoVersion4, context.Background())
+ defer srv.Stop()
+ cluster := testCluster(protoVersion4, srv.Address)
+ // Allow two speculative attempts on one loopback host, using the real
+ // executor and connection pool without requiring extra loopback
aliases.
+ cluster.PoolConfig.HostSelectionPolicy =
repeatingByteMetricsHostPolicy{RoundRobinHostPolicy()}
+ db, err := cluster.CreateSession()
+ require.NoError(t, err)
+ defer db.Close()
+ events := make(chan ObservedQuery, 2)
+ observer := unitFuncQueryObserver(func(_ context.Context, o
ObservedQuery) { events <- o })
+ q :=
db.Query("slow").Idempotent(true).Observer(observer).SetSpeculativeExecutionPolicy(
+ &SimpleSpeculativeExecution{NumAttempts: 1, TimeoutDelay:
time.Millisecond},
+ )
+ require.NoError(t, q.Exec())
+ var success, canceled int
+ for i := 0; i < 2; i++ {
+ select {
+ case o := <-events:
+ require.Positive(t, o.BytesTx)
+ if o.Err == nil {
+ success++
+ require.Equal(t, int64(4), o.BytesRx)
+ } else {
+ canceled++
+ require.ErrorIs(t, o.Err, context.Canceled)
+ // A response may race with cancellation;
either way, bytes
+ // from the other attempt must not enter this
observation.
+ require.LessOrEqual(t, o.BytesRx, int64(4))
+ }
+ case <-time.After(time.Second):
+ t.Fatal("missing speculative observation")
+ }
+ }
+ require.GreaterOrEqual(t, success, 1)
+ require.Equal(t, 2, success+canceled)
+}
+
+func TestObserveByteMetricsPages(t *testing.T) {
+ var mu sync.Mutex
+ var page int
+ var expected []byteCounts
+ srv := newTestServerOpts{
+ addr: "127.0.0.1:0", protocol: protoVersion4,
+ customRequestHandler: func(_ *TestServer, req, resp *framer)
error {
+ switch req.header.op {
+ case opStartup:
+ resp.writeHeader(0, opReady, req.header.stream)
+ case opOptions:
+ resp.writeHeader(0, opSupported,
req.header.stream)
+ resp.writeShort(0)
+ case opQuery:
+ page++
+ resp.writeHeader(0, opResult, req.header.stream)
+ resp.writeInt(resultKindRows)
+ flags := flagGlobalTableSpec
+ if page == 1 {
+ flags |= flagHasMorePages
+ }
+ resp.writeInt(int32(flags))
+ resp.writeInt(1) // one column
+ if page == 1 {
+ resp.writeBytes([]byte{1})
+ }
+ resp.writeString("keyspace")
+ resp.writeString("table")
+ resp.writeString("value")
+ resp.writeShort(uint16(TypeInt))
+ resp.writeInt(1) // one row
+ resp.writeBytes([]byte{0, 0, 0, byte(page)})
+ mu.Lock()
+ expected = append(expected,
byteCounts{int64(req.header.length), int64(len(resp.buf) - frameHeadSize)})
+ mu.Unlock()
+ default:
+ return fmt.Errorf("unexpected opcode %v",
req.header.op)
+ }
+ return nil
+ },
+ }.newServer(t, context.Background())
+ defer srv.Stop()
+ cluster := testCluster(protoVersion4, srv.Address)
+ cluster.NumConns = 1
+ db, err := cluster.CreateSession()
+ require.NoError(t, err)
+ defer db.Close()
+ var events []ObservedQuery
+ observer := unitFuncQueryObserver(func(_ context.Context, o
ObservedQuery) {
+ mu.Lock()
+ defer mu.Unlock()
+ events = append(events, o)
+ })
+ iter := db.Query("paged").PageSize(1).Observer(observer).Iter()
+ var value int
+ require.True(t, iter.Scan(&value))
+ require.Equal(t, 1, value)
+ require.True(t, iter.Scan(&value))
+ require.Equal(t, 2, value)
+ require.False(t, iter.Scan(&value))
+ require.NoError(t, iter.Close())
+ mu.Lock()
+ defer mu.Unlock()
+ require.Len(t, events, 2)
+ require.Len(t, expected, 2)
+ for i, o := range events {
+ require.Zero(t, o.Attempt)
+ // Each page observation counts only its own request and
response.
+ require.Equal(t, expected[i], byteCounts{o.BytesTx, o.BytesRx})
+ }
+}
+
+func TestObserveByteMetricsSchemaAgreement(t *testing.T) {
+ var mu sync.Mutex
+ var expected byteCounts
+ var queries int
+ srv := newTestServerOpts{
+ addr: "127.0.0.1:0", protocol: protoVersion4,
+ customRequestHandler: func(_ *TestServer, req, resp *framer)
error {
+ switch req.header.op {
+ case opStartup:
+ resp.writeHeader(0, opReady, req.header.stream)
+ case opOptions:
+ resp.writeHeader(0, opSupported,
req.header.stream)
+ resp.writeShort(0)
+ case opQuery:
+ statement, err := req.readLongString()
+ if err != nil {
+ return err
+ }
+ resp.writeHeader(0, opResult, req.header.stream)
+ if statement == "ddl" {
+ resp.writeInt(resultKindSchemaChanged)
+ resp.writeString("CREATED")
+ resp.writeString("KEYSPACE")
+ resp.writeString("keyspace")
+ } else {
+ resp.writeInt(resultKindRows)
+ resp.writeInt(0) // metadata flags
+ resp.writeInt(0) // columns
+ resp.writeInt(0) // rows; no
conflicting schema versions
+ }
+ mu.Lock()
+ queries++
+ expected.bytesTx += int64(req.header.length)
+ expected.bytesRx += int64(len(resp.buf) -
frameHeadSize)
+ mu.Unlock()
+ default:
+ return fmt.Errorf("unexpected opcode %v",
req.header.op)
+ }
+ return nil
+ },
+ }.newServer(t, context.Background())
+ defer srv.Stop()
+ db, err := newTestSession(protoVersion4, srv.Address)
+ require.NoError(t, err)
+ defer db.Close()
+ var observed ObservedQuery
+ var events int
+ observer := unitFuncQueryObserver(func(_ context.Context, o
ObservedQuery) { observed = o; events++ })
+ require.NoError(t, db.Query("ddl").Observer(observer).Exec())
+ require.Equal(t, 1, events)
+ mu.Lock()
+ defer mu.Unlock()
+ // The DDL exchange and the peer/local schema queries belong to one
attempt.
+ require.Equal(t, 3, queries)
+ require.Equal(t, expected, byteCounts{observed.BytesTx,
observed.BytesRx})
+}
+
+func newByteMetricsConn(t *testing.T, proto byte, compressor Compressor)
(*Conn, net.Conn) {
+ t.Helper()
+ client, server := net.Pipe()
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(func() { cancel(); client.Close(); server.Close() })
+ c := &Conn{
+ r: &connReader{conn: client, r: bufio.NewReader(client)},
+ w: &deadlineContextWriter{
+ w: client, timeout: time.Second, semaphore: make(chan
struct{}, 1), quit: make(chan struct{}),
+ },
+ version: proto, compressor: compressor, streams:
streams.New(int(proto)),
+ calls: make(map[int]*callReq), ctx: ctx, cancel: cancel,
+ session: &Session{types: GlobalTypes}, requestTimeout:
time.Second, host: &HostInfo{},
+ logger: &defaultLogger{}, errorHandler:
connErrorHandlerFn(func(*Conn, error, bool) {}),
+ }
+ return c, server
+}
+
+// readByteMetricsRequest reads one request as the server, returning its header
+// and decompressed body.
+func readByteMetricsRequest(r io.Reader, proto byte, compressor Compressor)
(frameHeader, []byte, error) {
+ if proto == protoVersion5 {
+ var payload []byte
+ var err error
+ if compressor != nil {
+ payload, _, err = readCompressedSegment(r, compressor)
+ } else {
+ payload, _, err = readUncompressedSegment(r)
+ }
+ if err != nil {
+ return frameHeader{}, nil, err
+ }
+ r = bytes.NewReader(payload)
+ }
+ head, err := readHeader(r, make([]byte, frameHeadSize))
+ if err != nil {
+ return frameHeader{}, nil, err
+ }
+ request := newFramer(compressor, proto, GlobalTypes)
+ if err := request.readFrame(r, &head); err != nil {
+ return frameHeader{}, nil, err
+ }
+ return head, request.buf, nil
+}
+
+// newByteMetricsResponse builds a finished VOID result frame for stream,
padded
+// to bodySize bytes before compression.
+func newByteMetricsResponse(proto byte, compressor Compressor, stream,
bodySize int) (*framer, error) {
+ f := newFramer(compressor, proto, GlobalTypes)
+ f.writeHeader(f.flags, opResult, stream)
+ f.buf[0] |= 0x80
+ f.writeInt(resultKindVoid)
+ f.buf = append(f.buf, make([]byte, bodySize-4)...)
+ return f, f.finish()
+}
+
+func TestByteMetricsExec(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ proto byte
+ compressor Compressor
+ bodySize int
+ }{
+ {"v4-plain", protoVersion4, nil, 2048},
+ {"v4-snappy", protoVersion4, snappy.SnappyCompressor{}, 2048},
+ {"v5-plain", protoVersion5, nil, 2048},
+ {"v5-lz4", protoVersion5, lz4.LZ4Compressor{}, 2048},
+ // Responses larger than one segment are reassembled from
several.
+ {"v5-plain-fragmented", protoVersion5, nil, 2 *
maxSegmentPayloadSize},
+ {"v5-lz4-fragmented", protoVersion5, lz4.LZ4Compressor{}, 2 *
maxSegmentPayloadSize},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ c, server := newByteMetricsConn(t, tc.proto,
tc.compressor)
+ metrics := &byteMetrics{}
+ type result struct {
+ counts byteCounts
+ err error
+ }
+ serverResult := make(chan result, 1)
+ go func() {
+ head, body, err :=
readByteMetricsRequest(server, tc.proto, tc.compressor)
+ var response *framer
+ if err == nil {
+ response, err =
newByteMetricsResponse(tc.proto, tc.compressor, head.stream, tc.bodySize)
+ }
+ if err == nil && tc.proto == protoVersion5 {
+ err = response.prepareModernLayout()
+ }
+ if err == nil {
+ _, err = server.Write(response.buf)
+ }
+ // Bodies as they were before compression and
after decompression.
+ serverResult <- result{counts:
byteCounts{int64(len(body)), int64(tc.bodySize)}, err: err}
+ }()
+ received := make(chan error, 1)
+ go func() { received <- c.recv(c.ctx, true) }()
+ _, err := c.execWithMetrics(c.ctx, &writeQueryFrame{
+ statement: strings.Repeat("query ", 256),
params: queryParams{consistency: One},
+ }, nil, metrics)
+ require.NoError(t, err)
+ require.NoError(t, <-received)
+ actual := <-serverResult
+ require.NoError(t, actual.err)
+ require.Equal(t, actual.counts, metrics.snapshot())
+ })
+ }
+}
+
+func TestByteMetricsSharedSegment(t *testing.T) {
+ for _, compressed := range []bool{false, true} {
+ t.Run(fmt.Sprintf("compressed=%t", compressed), func(t
*testing.T) {
+ var compressor Compressor
+ if compressed {
+ compressor = lz4.LZ4Compressor{}
+ }
+ c, server := newByteMetricsConn(t, protoVersion5,
compressor)
+ statements := []string{"a", strings.Repeat("b", 29),
strings.Repeat("c", 113)}
+ metrics := make([]*byteMetrics, len(statements))
+ execErrs := make(chan error, len(statements))
+ for i, statement := range statements {
+ metrics[i] = &byteMetrics{}
+ go func(statement string, m *byteMetrics) {
+ _, err := c.execWithMetrics(c.ctx,
&writeQueryFrame{
+ statement: statement, params:
queryParams{consistency: One},
+ }, nil, m)
+ execErrs <- err
+ }(statement, metrics[i])
+ }
+ // Answer every request in one self-contained segment.
Each response
+ // body is 4 bytes longer than its request body, so
each attempt can
+ // be checked against its own request.
+ var payload []byte
+ for range statements {
+ head, body, err :=
readByteMetricsRequest(server, protoVersion5, compressor)
+ require.NoError(t, err)
+ response, err :=
newByteMetricsResponse(protoVersion5, nil, head.stream, len(body)+4)
+ require.NoError(t, err)
+ payload = append(payload, response.buf...)
+ }
+ var segment []byte
+ var err error
+ if compressed {
+ segment, err = newCompressedSegment(payload,
true, compressor)
+ } else {
+ segment, err = newUncompressedSegment(payload,
true)
+ }
+ require.NoError(t, err)
+ received := make(chan error, 1)
+ go func() { received <- c.recv(c.ctx, true) }()
+ _, err = server.Write(segment)
+ require.NoError(t, err)
+ require.NoError(t, <-received)
+ for range statements {
+ require.NoError(t, <-execErrs)
+ }
+ for _, m := range metrics {
+ counts := m.snapshot()
+ require.Positive(t, counts.bytesTx)
+ // Each response is charged only its own body,
not the shared segment.
+ require.Equal(t, counts.bytesTx+4,
counts.bytesRx)
+ }
+ })
+ }
+}
+
+// TestByteMetricsLateResponse checks that a response read after its request
+// timed out is not counted, even though the stream is still waiting for it.
+func TestByteMetricsLateResponse(t *testing.T) {
+ c, server := newByteMetricsConn(t, protoVersion4, nil)
+ c.requestTimeout = 10 * time.Millisecond
+ metrics := &byteMetrics{}
+ execErr := make(chan error, 1)
+ go func() {
+ _, err := c.execWithMetrics(c.ctx, &writeQueryFrame{statement:
"query", params: queryParams{consistency: One}}, nil, metrics)
+ execErr <- err
+ }()
+ head, body, err := readByteMetricsRequest(server, protoVersion4, nil)
+ require.NoError(t, err)
+ require.ErrorIs(t, <-execErr, ErrTimeoutNoResponse)
+
+ response, err := newByteMetricsResponse(protoVersion4, nil,
head.stream, 4)
+ require.NoError(t, err)
+ received := make(chan error, 1)
+ go func() { received <- c.recv(c.ctx, true) }()
+ _, err = server.Write(response.buf)
+ require.NoError(t, err)
+ require.NoError(t, <-received)
+ require.Equal(t, byteCounts{bytesTx: int64(len(body))},
metrics.snapshot())
+}
+
+type partialByteMetricsWriter struct{}
+
+func (partialByteMetricsWriter) writeContext(context.Context, []byte) (int,
error) {
+ return 5, io.ErrShortWrite
+}
+
+func TestByteMetricsPartialWrite(t *testing.T) {
+ c, _ := newByteMetricsConn(t, protoVersion4, nil)
+ c.w = partialByteMetricsWriter{}
+ metrics := &byteMetrics{}
+ _, err := c.execWithMetrics(c.ctx, &writeQueryFrame{statement: "query",
params: queryParams{consistency: One}}, nil, metrics)
+ require.ErrorIs(t, err, io.ErrShortWrite)
+ require.Equal(t, byteCounts{}, metrics.snapshot())
+}
+
+func TestByteMetricsPartialRead(t *testing.T) {
+ c, server := newByteMetricsConn(t, protoVersion4, nil)
+ metrics := &byteMetrics{}
+ execErr := make(chan error, 1)
+ go func() {
+ _, err := c.execWithMetrics(c.ctx, &writeQueryFrame{statement:
"query", params: queryParams{consistency: One}}, nil, metrics)
+ execErr <- err
+ }()
+ head, body, err := readByteMetricsRequest(server, protoVersion4, nil)
+ require.NoError(t, err)
+
+ response, err := newByteMetricsResponse(protoVersion4, nil,
head.stream, 24)
+ require.NoError(t, err)
+ received := make(chan error, 1)
+ go func() { received <- c.recv(c.ctx, true) }()
+ // Send the header and only part of the body before the connection
closes.
+ _, err = server.Write(response.buf[:frameHeadSize+5])
+ require.NoError(t, err)
+ require.NoError(t, server.Close())
+ require.NoError(t, <-received)
+ require.Error(t, <-execErr)
+ require.Equal(t, byteCounts{bytesTx: int64(len(body))},
metrics.snapshot())
+}
diff --git a/conn.go b/conn.go
index 54122445..ced68f18 100644
--- a/conn.go
+++ b/conn.go
@@ -436,7 +436,7 @@ func (s *startupCoordinator) write(ctx context.Context,
frame frameBuilder, star
return nil, ctx.Err()
}
- framer, err := s.conn.execInternal(ctx, frame, nil,
startupCompleted.Load())
+ framer, err := s.conn.execInternal(ctx, frame, nil,
startupCompleted.Load(), nil)
if err != nil {
return nil, err
}
@@ -1233,10 +1233,14 @@ func (c *Conn) addCall(call *callReq) error {
}
func (c *Conn) exec(ctx context.Context, req frameBuilder, tracer Tracer)
(*framer, error) {
- return c.execInternal(ctx, req, tracer, true)
+ return c.execWithMetrics(ctx, req, tracer, nil)
}
-func (c *Conn) execInternal(ctx context.Context, req frameBuilder, tracer
Tracer, startupCompleted bool) (*framer, error) {
+func (c *Conn) execWithMetrics(ctx context.Context, req frameBuilder, tracer
Tracer, bm *byteMetrics) (*framer, error) {
+ return c.execInternal(ctx, req, tracer, true, bm)
+}
+
+func (c *Conn) execInternal(ctx context.Context, req frameBuilder, tracer
Tracer, startupCompleted bool, bm *byteMetrics) (*framer, error) {
if ctxErr := ctx.Err(); ctxErr != nil {
return nil, ctxErr
}
@@ -1303,6 +1307,9 @@ func (c *Conn) execInternal(ctx context.Context, req
frameBuilder, tracer Tracer
}
if err == nil {
n, err = c.w.writeContext(ctx, framer.buf)
+ if err == nil {
+ bm.addTx(framer.uncompressedBodySize)
+ }
}
if err != nil {
// closeWithError will block waiting for this stream to either
receive a response
@@ -1367,6 +1374,9 @@ func (c *Conn) execInternal(ctx context.Context, req
frameBuilder, tracer Tracer
}
return nil, resp.err
}
+ // Count the response only once this request has received it,
so a
+ // response arriving after a timeout or cancellation is not
counted.
+ bm.addRx(resp.framer.uncompressedBodySize)
// dont release the stream if detect a timeout as another
request can reuse
// that stream and get a response for the old request, which we
have no
// easy way of detecting.
@@ -1474,6 +1484,10 @@ type inflightPrepare struct {
}
func (c *Conn) prepareStatement(ctx context.Context, stmt string, tracer
Tracer, keyspace string) (*preparedStatment, error) {
+ return c.prepareStatementWithMetrics(ctx, stmt, tracer, keyspace, nil)
+}
+
+func (c *Conn) prepareStatementWithMetrics(ctx context.Context, stmt string,
tracer Tracer, keyspace string, bm *byteMetrics) (*preparedStatment, error) {
stmtCacheKey := c.session.stmtsLRU.keyFor(c.host.HostID(), keyspace,
stmt)
flight, ok := c.session.stmtsLRU.execIfMissing(stmtCacheKey, func(lru
*lru.Cache) *inflightPrepare {
flight := &inflightPrepare{
@@ -1494,10 +1508,10 @@ func (c *Conn) prepareStatement(ctx context.Context,
stmt string, tracer Tracer,
prep.keyspace = keyspace
}
- // we won the race to do the load, if our context is
canceled we shouldnt
- // stop the load as other callers are waiting for it
but this caller should get
- // their context cancelled error.
- framer, err := c.exec(c.ctx, prep, tracer)
+ // Charge only the caller that initiated this prepare,
not cache hits
+ // or callers waiting for the same inflight prepare.
Keep c.ctx so a
+ // canceled caller does not cancel preparation for
other callers.
+ framer, err := c.execWithMetrics(c.ctx, prep, tracer,
bm)
if err != nil {
flight.err = err
c.session.stmtsLRU.remove(stmtCacheKey)
@@ -1570,6 +1584,10 @@ func marshalQueryValue(typ TypeInfo, value interface{},
dst *queryValues) error
}
func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter {
+ return c.executeQueryWithMetrics(ctx, q, nil)
+}
+
+func (c *Conn) executeQueryWithMetrics(ctx context.Context, q *internalQuery,
bm *byteMetrics) *Iter {
qryOpts := q.qryOpts
params := queryParams{
consistency: q.GetConsistency(),
@@ -1607,7 +1625,7 @@ func (c *Conn) executeQuery(ctx context.Context, q
*internalQuery) *Iter {
if !qryOpts.skipPrepare && shouldPrepare(qryOpts.stmt) {
// Prepare all DML queries. Other queries can not be prepared.
var err error
- info, err = c.prepareStatement(ctx, qryOpts.stmt,
qryOpts.trace, usedKeyspace)
+ info, err = c.prepareStatementWithMetrics(ctx, qryOpts.stmt,
qryOpts.trace, usedKeyspace, bm)
if err != nil {
iter.err = err
return iter
@@ -1671,7 +1689,7 @@ func (c *Conn) executeQuery(ctx context.Context, q
*internalQuery) *Iter {
}
}
- framer, err := c.exec(ctx, frame, qryOpts.trace)
+ framer, err := c.execWithMetrics(ctx, frame, qryOpts.trace, bm)
if err != nil {
iter.err = err
return iter
@@ -1749,6 +1767,9 @@ func (c *Conn) executeQuery(ctx context.Context, q
*internalQuery) *Iter {
q: newQry,
pos: int((1 - qryOpts.prefetch) *
float64(x.numRows)),
}
+ if newQry.conn != nil {
+ iter.next.bytes = bm
+ }
if iter.next.pos < 1 {
iter.next.pos = 1
@@ -1761,7 +1782,7 @@ func (c *Conn) executeQuery(ctx context.Context, q
*internalQuery) *Iter {
return iter
case *schemaChangeKeyspace, *schemaChangeTable, *schemaChangeFunction,
*schemaChangeAggregate, *schemaChangeType:
iter.framer = framer
- if err := c.awaitSchemaAgreement(ctx); err != nil {
+ if err := c.awaitSchemaAgreementWithMetrics(ctx,
c.session.cfg.MaxWaitSchemaAgreement, bm); err != nil {
// TODO: should have this behind a flag
c.logger.Warning("Error while awaiting for schema
agreement after a schema change event.", NewLogFieldError("err", err))
}
@@ -1772,7 +1793,7 @@ func (c *Conn) executeQuery(ctx context.Context, q
*internalQuery) *Iter {
case *RequestErrUnprepared:
stmtCacheKey := c.session.stmtsLRU.keyFor(c.host.HostID(),
usedKeyspace, qryOpts.stmt)
c.session.stmtsLRU.evictPreparedID(stmtCacheKey, x.StatementId)
- return c.executeQuery(ctx, q)
+ return c.executeQueryWithMetrics(ctx, q, bm)
case error:
iter.err = x
iter.framer = framer
@@ -1833,6 +1854,10 @@ func (c *Conn) UseKeyspace(keyspace string) error {
}
func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
+ return c.executeBatchWithMetrics(ctx, b, nil)
+}
+
+func (c *Conn) executeBatchWithMetrics(ctx context.Context, b *internalBatch,
bm *byteMetrics) *Iter {
iter := newIter(b.metrics, b.Keyspace(), b.routingInfo, nil)
n := len(b.batchOpts.entries)
req := &writeBatchFrame{
@@ -1868,7 +1893,7 @@ func (c *Conn) executeBatch(ctx context.Context, b
*internalBatch) *Iter {
batchStmt := &req.statements[i]
if len(entry.Args) > 0 || entry.binding != nil {
- info, err := c.prepareStatement(ctx, entry.Stmt,
b.batchOpts.trace, usedKeyspace)
+ info, err := c.prepareStatementWithMetrics(ctx,
entry.Stmt, b.batchOpts.trace, usedKeyspace, bm)
if err != nil {
iter.err = err
return iter
@@ -1920,7 +1945,7 @@ func (c *Conn) executeBatch(ctx context.Context, b
*internalBatch) *Iter {
}
}
- framer, err := c.exec(ctx, req, b.batchOpts.trace)
+ framer, err := c.execWithMetrics(ctx, req, b.batchOpts.trace, bm)
if err != nil {
iter.err = err
return iter
@@ -1946,7 +1971,7 @@ func (c *Conn) executeBatch(ctx context.Context, b
*internalBatch) *Iter {
key := c.session.stmtsLRU.keyFor(c.host.HostID(),
usedKeyspace, stmt)
c.session.stmtsLRU.evictPreparedID(key, x.StatementId)
}
- return c.executeBatch(ctx, b)
+ return c.executeBatchWithMetrics(ctx, b, bm)
case *resultRowsFrame:
iter.meta = x.meta
iter.framer = framer
@@ -1964,15 +1989,23 @@ func (c *Conn) executeBatch(ctx context.Context, b
*internalBatch) *Iter {
}
func (c *Conn) query(ctx context.Context, statement string, values
...interface{}) (iter *Iter) {
+ return c.queryWithMetrics(ctx, nil, statement, values...)
+}
+
+func (c *Conn) queryWithMetrics(ctx context.Context, bm *byteMetrics,
statement string, values ...interface{}) (iter *Iter) {
q := c.session.Query(statement, values...).Consistency(One).Trace(nil)
q.skipPrepare = true
q.disableSkipMetadata = true
// we want to keep the query on this connection
- return q.iterInternal(c, ctx)
+ return q.iterInternal(c, ctx, bm)
}
func (c *Conn) querySystemPeers(ctx context.Context, version cassVersion)
*Iter {
+ return c.querySystemPeersWithMetrics(ctx, version, nil)
+}
+
+func (c *Conn) querySystemPeersWithMetrics(ctx context.Context, version
cassVersion, bm *byteMetrics) *Iter {
const (
peerSchema = "SELECT * FROM system.peers"
peerV2Schemas = "SELECT * FROM system.peers_v2"
@@ -1984,7 +2017,7 @@ func (c *Conn) querySystemPeers(ctx context.Context,
version cassVersion) *Iter
if version.AtLeast(4, 0, 0) && isSchemaV2 {
// Try "system.peers_v2" and fallback to "system.peers" if it's
not found
- iter := c.query(ctx, peerV2Schemas)
+ iter := c.queryWithMetrics(ctx, bm, peerV2Schemas)
err := iter.checkErrAndNotFound()
if err != nil {
@@ -1993,14 +2026,14 @@ func (c *Conn) querySystemPeers(ctx context.Context,
version cassVersion) *Iter
c.mu.Lock()
c.isSchemaV2 = false
c.mu.Unlock()
- return c.query(ctx, peerSchema)
+ return c.queryWithMetrics(ctx, bm, peerSchema)
} else {
return iter
}
}
return iter
} else {
- return c.query(ctx, peerSchema)
+ return c.queryWithMetrics(ctx, bm, peerSchema)
}
}
@@ -2013,6 +2046,10 @@ func (c *Conn) awaitSchemaAgreement(ctx context.Context)
(err error) {
}
func (c *Conn) awaitSchemaAgreementWithTimeout(ctx context.Context, timeout
time.Duration) (err error) {
+ return c.awaitSchemaAgreementWithMetrics(ctx, timeout, nil)
+}
+
+func (c *Conn) awaitSchemaAgreementWithMetrics(ctx context.Context, timeout
time.Duration, bm *byteMetrics) (err error) {
const localSchemas = "SELECT schema_version FROM system.local WHERE
key='local'"
var versions map[string]struct{}
@@ -2022,7 +2059,7 @@ func (c *Conn) awaitSchemaAgreementWithTimeout(ctx
context.Context, timeout time
endDeadline := time.Now().Add(timeout)
for time.Now().Before(endDeadline) {
- iter := c.querySystemPeers(ctx, c.host.version)
+ iter := c.querySystemPeersWithMetrics(ctx, c.host.version, bm)
versions = make(map[string]struct{})
@@ -2049,7 +2086,7 @@ func (c *Conn) awaitSchemaAgreementWithTimeout(ctx
context.Context, timeout time
goto cont
}
- iter = c.query(ctx, localSchemas)
+ iter = c.queryWithMetrics(ctx, bm, localSchemas)
for iter.Scan(&schemaVersion) {
versions[schemaVersion] = struct{}{}
schemaVersion = ""
diff --git a/conn_test.go b/conn_test.go
index 1cb5ee3f..77bd2878 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -1613,3 +1613,28 @@ func TestConnProcessAllFramesInSingleSegment(t
*testing.T) {
require.NoError(t, err)
}
}
+
+type unitFuncQueryObserver func(context.Context, ObservedQuery)
+
+func (f unitFuncQueryObserver) ObserveQuery(ctx context.Context, o
ObservedQuery) { f(ctx, o) }
+
+// TestObserveQueryByteMetrics checks that the observer sees uncompressed body
+// sizes, excluding frame and segment overhead, for each protocol version.
+func TestObserveQueryByteMetrics(t *testing.T) {
+ for _, proto := range []byte{protoVersion4, protoVersion5} {
+ t.Run(fmt.Sprintf("v%d", proto), func(t *testing.T) {
+ srv := NewTestServer(t, proto, context.Background())
+ defer srv.Stop()
+ db, err := newTestSession(protoVersion(proto),
srv.Address)
+ require.NoError(t, err)
+ defer db.Close()
+
+ var observed ObservedQuery
+ observer := unitFuncQueryObserver(func(_
context.Context, o ObservedQuery) { observed = o })
+ require.NoError(t,
db.Query("void").Observer(observer).Exec())
+
+ require.Positive(t, observed.BytesTx)
+ require.Equal(t, int64(4), observed.BytesRx) //
resultKindVoid
+ })
+ }
+}
diff --git a/frame.go b/frame.go
index 7ad118c9..239fed86 100644
--- a/frame.go
+++ b/frame.go
@@ -384,6 +384,9 @@ type framer struct {
customPayload map[string][]byte
types *RegisteredTypes
+
+ // Body size before compression on writes, or after decompression on
reads.
+ uncompressedBodySize int
}
func newFramer(compressor Compressor, version byte, r *RegisteredTypes)
*framer {
@@ -494,6 +497,7 @@ func (f *framer) readFrame(r io.Reader, head *frameHeader)
error {
}
}
+ f.uncompressedBodySize = len(f.buf)
f.header = head
return nil
}
@@ -823,6 +827,7 @@ func (f *framer) finish() error {
return ErrFrameTooBig
}
+ f.uncompressedBodySize = len(f.buf) - frameHeadSize
if f.proto < protoVersion5 && f.buf[1]&flagCompress == flagCompress {
if f.compres == nil {
panic("compress flag set with no compressor")
diff --git a/query_executor.go b/query_executor.go
index 1ab75a4c..71fdd418 100644
--- a/query_executor.go
+++ b/query_executor.go
@@ -57,9 +57,10 @@ type Statement interface {
}
type internalRequest interface {
- execute(ctx context.Context, conn *Conn) *Iter
+ execute(ctx context.Context, conn *Conn, bm *byteMetrics) *Iter
getNextAttempt() int
- recordAttempt(attemptNum int, keyspace string, end, start time.Time,
iter *Iter, host *HostInfo)
+ recordAttempt(attemptNum int, keyspace string, end, start time.Time,
iter *Iter, host *HostInfo, bm *byteMetrics)
+ hasObserver() bool
retryPolicy() RetryPolicy
speculativeExecutionPolicy() SpeculativeExecutionPolicy
getQueryMetrics() *queryMetrics
@@ -188,6 +189,46 @@ func (c RequestInterceptorChain) InterceptPostAttempt(
}
}
+// byteMetrics belongs to one observer attempt, unlike queryMetrics, which is
+// shared across retries within a page. Preparation runs in its own goroutine
+// and can update it after the attempt has given up waiting for it.
+type byteMetrics struct {
+ mu sync.Mutex
+ byteCounts
+}
+
+// byteCounts holds uncompressed message body sizes; see ObservedQuery.BytesTx.
+type byteCounts struct {
+ bytesTx, bytesRx int64
+}
+
+func (m *byteMetrics) addTx(n int) {
+ if m == nil {
+ return
+ }
+ m.mu.Lock()
+ m.bytesTx += int64(n)
+ m.mu.Unlock()
+}
+
+func (m *byteMetrics) addRx(n int) {
+ if m == nil {
+ return
+ }
+ m.mu.Lock()
+ m.bytesRx += int64(n)
+ m.mu.Unlock()
+}
+
+func (m *byteMetrics) snapshot() byteCounts {
+ if m == nil {
+ return byteCounts{}
+ }
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return m.byteCounts
+}
+
func (q *queryExecutor) attemptQuery(ctx context.Context, iRequest
internalRequest, conn *Conn) *Iter {
var res InterceptResult
var interceptor RequestInterceptor
@@ -205,6 +246,10 @@ func (q *queryExecutor) attemptQuery(ctx context.Context,
iRequest internalReque
interceptor = iBatch.batchOpts.interceptor
}
attempt := iRequest.getNextAttempt()
+ var bm *byteMetrics
+ if iRequest.hasObserver() {
+ bm = &byteMetrics{}
+ }
if interceptor != nil {
var err error
res, err = interceptor.InterceptPreAttempt(ctx, iq)
@@ -216,7 +261,7 @@ func (q *queryExecutor) attemptQuery(ctx context.Context,
iRequest internalReque
// to abort the attempt, rather than getting into an
infinite retry loop with zero perceived attempts.
// Nothing was sent, so the attempt has zero latency.
now := time.Now()
- iRequest.recordAttempt(attempt, q.pool.keyspace, now,
now, iter, conn.host)
+ iRequest.recordAttempt(attempt, q.pool.keyspace, now,
now, iter, conn.host, bm)
return iter
}
if res.Ctx != nil {
@@ -226,12 +271,12 @@ func (q *queryExecutor) attemptQuery(ctx context.Context,
iRequest internalReque
// Time only the execution, so that time spent in interceptors isn't
reported as query latency.
start := time.Now()
- iter := iRequest.execute(ctx, conn)
+ iter := iRequest.execute(ctx, conn, bm)
end := time.Now()
if interceptor != nil {
interceptor.InterceptPostAttempt(res, iter.err)
}
- iRequest.recordAttempt(attempt, q.pool.keyspace, end, start, iter,
conn.host)
+ iRequest.recordAttempt(attempt, q.pool.keyspace, end, start, iter,
conn.host, bm)
return iter
}
@@ -541,11 +586,12 @@ func (q *internalQuery) Attempts() int {
return q.metrics.attempts()
}
-func (q *internalQuery) recordAttempt(attemptNum int, keyspace string, end,
start time.Time, iter *Iter, host *HostInfo) {
+func (q *internalQuery) recordAttempt(attemptNum int, keyspace string, end,
start time.Time, iter *Iter, host *HostInfo, bm *byteMetrics) {
latency := end.Sub(start)
q.metrics.recordAttempt(latency)
if q.qryOpts.observer != nil {
+ counts := bm.snapshot()
metricsForHost := q.hostMetricsManager.attempt(latency, host)
q.qryOpts.observer.ObserveQuery(q.qryOpts.context,
ObservedQuery{
Keyspace: keyspace,
@@ -565,12 +611,18 @@ func (q *internalQuery) recordAttempt(attemptNum int,
keyspace string, end, star
Attempt: attemptNum,
Query: q.originalQuery,
RejectedByInterceptor: iter.rejectedByInterceptor,
+ BytesTx: counts.bytesTx,
+ BytesRx: counts.bytesRx,
})
}
}
-func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
- return conn.executeQuery(ctx, q)
+func (q *internalQuery) execute(ctx context.Context, conn *Conn, bm
*byteMetrics) *Iter {
+ return conn.executeQueryWithMetrics(ctx, q, bm)
+}
+
+func (q *internalQuery) hasObserver() bool {
+ return q.qryOpts.observer != nil
}
func (q *internalQuery) retryPolicy() RetryPolicy {
@@ -778,7 +830,7 @@ func (b *internalBatch) Attempts() int {
return b.metrics.attempts()
}
-func (b *internalBatch) recordAttempt(attemptNum int, keyspace string, end,
start time.Time, iter *Iter, host *HostInfo) {
+func (b *internalBatch) recordAttempt(attemptNum int, keyspace string, end,
start time.Time, iter *Iter, host *HostInfo, bm *byteMetrics) {
latency := end.Sub(start)
b.metrics.recordAttempt(latency)
@@ -787,6 +839,7 @@ func (b *internalBatch) recordAttempt(attemptNum int,
keyspace string, end, star
}
metricsForHost := b.hostMetricsManager.attempt(latency, host)
+ counts := bm.snapshot()
n := len(b.batchOpts.entries)
statements := make([]string, n)
@@ -819,6 +872,8 @@ func (b *internalBatch) recordAttempt(attemptNum int,
keyspace string, end, star
Attempt: attemptNum,
Batch: b.originalBatch,
RejectedByInterceptor: iter.rejectedByInterceptor,
+ BytesTx: counts.bytesTx,
+ BytesRx: counts.bytesRx,
})
}
@@ -905,6 +960,10 @@ func (b *internalBatch) getKeyspaceFunc() func() string {
return nil
}
-func (b *internalBatch) execute(ctx context.Context, conn *Conn) *Iter {
- return conn.executeBatch(ctx, b)
+func (b *internalBatch) execute(ctx context.Context, conn *Conn, bm
*byteMetrics) *Iter {
+ return conn.executeBatchWithMetrics(ctx, b, bm)
+}
+
+func (b *internalBatch) hasObserver() bool {
+ return b.batchOpts.observer != nil
}
diff --git a/session.go b/session.go
index feb0752a..c35284a0 100644
--- a/session.go
+++ b/session.go
@@ -1455,11 +1455,11 @@ func (q *Query) IterContext(ctx context.Context) *Iter {
return q.session.executeQuery(internalQry)
}
-func (q *Query) iterInternal(c *Conn, ctx context.Context) *Iter {
+func (q *Query) iterInternal(c *Conn, ctx context.Context, bm *byteMetrics)
*Iter {
internalQry := newInternalQuery(q, ctx)
internalQry.conn = c
- iter := c.executeQuery(internalQry.Context(), internalQry)
+ iter := c.executeQueryWithMetrics(internalQry.Context(), internalQry,
bm)
if iter != nil {
// set iter.host so that the caller can retrieve the connect
address which should be preferable (if valid) for the local host
iter.host = c.host
@@ -2021,6 +2021,10 @@ type nextIter struct {
oncea sync.Once
once sync.Once
next *Iter
+ // Set only for queries pinned to a connection, whose pages can be part
of
+ // a surrounding attempt (for example, schema agreement checks). Pages
routed
+ // through the session get their own byteMetrics through queryExecutor.
+ bytes *byteMetrics
}
func (n *nextIter) fetchAsync() {
@@ -2034,7 +2038,7 @@ func (n *nextIter) fetch() *Iter {
// if the query was specifically run on a connection then
re-use that
// connection when fetching the next results
if n.q.conn != nil {
- n.next = n.q.conn.executeQuery(n.q.qryOpts.context, n.q)
+ n.next =
n.q.conn.executeQueryWithMetrics(n.q.qryOpts.context, n.q, n.bytes)
} else {
n.next = n.q.session.executeQuery(n.q)
}
@@ -2494,6 +2498,23 @@ type ObservedQuery struct {
// Whether the attempt was aborted by the interceptor before it was
issued. In this case, the request was never
// actually sent to the above host.
RejectedByInterceptor bool
+
+ // BytesTx is the total size of the uncompressed bodies of native
protocol
+ // messages fully written during this attempt. It is not the number of
bytes
+ // sent on the wire: compression is not reflected, and frame headers,
protocol
+ // v5+ segment headers and checksums, and TLS and TCP overhead are
excluded.
+ // A message whose write fails partway contributes zero. Fully written
messages
+ // remain counted if the attempt later fails, for example while
awaiting a response.
+ // Preparation, UNPREPARED recovery and schema agreement requests
initiated
+ // within the attempt contribute to these counts. Shared preparation is
charged
+ // only to the attempt that initiates it; cache hits and waiters are
not charged.
+ BytesTx int64
+ // BytesRx is the total size of the uncompressed (after decompression)
bodies
+ // of native protocol responses received by this attempt, with the same
+ // exclusions as BytesTx. A response is counted only once the request
waiting
+ // for it receives it, so a response that fails to read or decompress,
or that
+ // arrives after the request has timed out or been canceled, is not
counted.
+ BytesRx int64
}
// QueryObserver is the interface implemented by query observers / stat
collectors.
@@ -2547,6 +2568,12 @@ type ObservedBatch struct {
// Whether the attempt was aborted by the interceptor before it was
issued. In this case, the request was never
// actually sent to the above host.
RejectedByInterceptor bool
+
+ // BytesTx and BytesRx have the same semantics as the corresponding
fields
+ // of ObservedQuery, for this batch attempt: they count uncompressed
message
+ // bodies, not bytes on the wire.
+ BytesTx int64
+ BytesRx int64
}
// BatchObserver is the interface implemented by batch observers / stat
collectors.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
