This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 9a0a182c7 refactor(go/adbc): refactor logging instrumentation into
OTel tracing - part 3/3 (#4661)
9a0a182c7 is described below
commit 9a0a182c75e422331def2ce7b43f83f0187a01d8
Author: Bruce Irschick <[email protected]>
AuthorDate: Wed Aug 26 17:15:43 2026 -0700
refactor(go/adbc): refactor logging instrumentation into OTel tracing -
part 3/3 (#4661)
This pull request makes significant improvements to the FlightSQL
driver's tracing and logging, replacing legacy logging with structured
OpenTelemetry tracing, enhancing error recording, and ensuring better
test coverage for tracing-related cleanup. The changes modernize how
operation metadata is captured and reported, making tracing more
consistent and useful for observability and debugging.
**Tracing and Logging Modernization:**
* Replaced `slog`-based logging with OpenTelemetry tracing throughout
`flightsql_statement.go` and `flightsql_bulk_ingest.go`, using
`attribute.KeyValue` for structured event attributes and adding detailed
span events for operation start and finish. This includes updating
helper functions to return attributes in the new format.
* Updated error handling in tracing: errors are now consistently
recorded as span errors with stack traces, and operation-specific error
wrapping is improved for clarity.
**API and Internal Refactoring:**
* Refactored method signatures and internal calls to use the new tracing
and attribute formats, including changes to `SetSqlQuery`,
`ExecuteQuery`, `ExecuteUpdate`, and `Prepare` methods in `statement`.
* Changed the construction of the Flight client to add a unary
interceptor for response metadata, improving consistency for both unary
and stream calls.
**Testing Improvements:**
* Added a new test
`TestFlightSQLTracingCleansUpAfterConstructionFailure` to ensure that
tracing resources are properly cleaned up even when database
construction fails, preventing resource leaks.
**Dependency and Import Updates:**
* Updated imports to include OpenTelemetry packages and internal helpers
for tracing, and removed unused logging imports.
**Helper and Utility Updates:**
* Updated helper functions for generating tracing attributes and
correlation metadata to use the new attribute-based approach, improving
code clarity and maintainability.
These changes collectively ensure that the driver emits rich, structured
telemetry for all major operations, facilitates easier debugging, and
aligns with modern observability practices.
---
Extends: https://github.com/apache/arrow-adbc/pull/4659
Part 3/3
---------
Co-authored-by: Bruce Irschick (Bit Quill Technologies Inc)
<[email protected]>
---
go/adbc/driver/flightsql/flightsql_adbc_test.go | 36 +++++
go/adbc/driver/flightsql/flightsql_bulk_ingest.go | 68 +++++----
go/adbc/driver/flightsql/flightsql_connection.go | 24 ++--
go/adbc/driver/flightsql/flightsql_database.go | 5 +-
go/adbc/driver/flightsql/flightsql_statement.go | 166 ++++++++++------------
go/adbc/driver/flightsql/flightsql_tracing.go | 73 ++++++++++
go/adbc/driver/flightsql/logging.go | 79 ----------
go/adbc/driver/flightsql/record_reader.go | 16 ++-
go/adbc/driver/flightsql/record_reader_test.go | 6 +-
go/adbc/driver/flightsql/tracing_test.go | 58 ++++++++
10 files changed, 320 insertions(+), 211 deletions(-)
diff --git a/go/adbc/driver/flightsql/flightsql_adbc_test.go
b/go/adbc/driver/flightsql/flightsql_adbc_test.go
index 79d318b10..66019995f 100644
--- a/go/adbc/driver/flightsql/flightsql_adbc_test.go
+++ b/go/adbc/driver/flightsql/flightsql_adbc_test.go
@@ -369,9 +369,45 @@ func TestFlightSQLTracingProducesTraceFiles(t *testing.T) {
output := traceOutput.String()
require.Contains(t, output, "FlightSQL.Database.Open")
+ require.Contains(t, output, "FlightSQL.Database.Close")
require.Contains(t, output, "FlightSQL.Statement.ExecuteQuery")
}
+// TestFlightSQLTracingCleansUpAfterConstructionFailure verifies that failures
+// after tracing initialization closes the exporter and release its file
handles.
+// Removing the trace directory must succeed after URI or option validation
fails.
+func TestFlightSQLTracingCleansUpAfterConstructionFailure(t *testing.T) {
+ alloc := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer alloc.AssertSize(t, 0)
+ drv := driver.NewDriver(alloc)
+
+ for _, test := range []struct {
+ name string
+ uri string
+ extraOp map[string]string
+ }{
+ {name: "invalid URI", uri: "grpc://%"},
+ {name: "invalid option", uri: "grpc://localhost", extraOp:
map[string]string{"unknown option": "value"}},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ traceDir := t.TempDir()
+ opts := map[string]string{
+ adbc.OptionKeyURI:
test.uri,
+ adbc.OptionKeyTelemetryTracesExporter:
string(adbc.TelemetryExporterAdbcFile),
+ adbc.OptionKeyTelemetryTracesFolderPath:
traceDir,
+ }
+ for key, value := range test.extraOp {
+ opts[key] = value
+ }
+
+ _, err := drv.NewDatabase(opts)
+ require.Error(t, err)
+ require.IsType(t, adbc.Error{}, err)
+ require.NoError(t, os.RemoveAll(traceDir))
+ })
+ }
+}
+
// Run the test suite, but validating that a header set on the database is
ALWAYS passed
type FlightSQLWithHeaderQuirks struct {
diff --git a/go/adbc/driver/flightsql/flightsql_bulk_ingest.go
b/go/adbc/driver/flightsql/flightsql_bulk_ingest.go
index fea5604d5..b888a3c2a 100644
--- a/go/adbc/driver/flightsql/flightsql_bulk_ingest.go
+++ b/go/adbc/driver/flightsql/flightsql_bulk_ingest.go
@@ -20,14 +20,16 @@ package flightsql
import (
"context"
"fmt"
- "log/slog"
"time"
"github.com/apache/arrow-adbc/go/adbc"
+ "github.com/apache/arrow-adbc/go/adbc/driver/internal"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/flight/flightsql"
pb "github.com/apache/arrow-go/v18/arrow/flight/gen/flight"
+ "go.opentelemetry.io/otel/attribute"
+ "go.opentelemetry.io/otel/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
@@ -105,7 +107,18 @@ func createRecordReaderFromBatch(batch arrow.RecordBatch)
(array.RecordReader, e
// executeIngest performs bulk ingestion using the FlightSQL client's
ExecuteIngest method.
// This is called from the statement when a target table has been set for bulk
ingest.
-func (s *statement) executeIngest(ctx context.Context) (int64, error) {
+func (s *statement) executeIngest(ctx context.Context) (nRows int64, err
error) {
+ var startTime = time.Now()
+ ctx, span := internal.StartSpan(ctx, "FlightSQL.BulkIngest.Execute",
s.cnxn)
+ errorRecorded := false
+ defer func() {
+ internal.NewEndSpanHelper(span).
+ WithError(err).
+ WithStartTime(startTime).
+ WithRecordedError(errorRecorded).
+ EndSpan()
+ }()
+
if s.streamBind == nil && s.bound == nil {
return -1, adbc.Error{
Msg: "[Flight SQL Statement] must call Bind before
bulk ingestion",
@@ -113,7 +126,6 @@ func (s *statement) executeIngest(ctx context.Context)
(int64, error) {
}
}
- startTime := time.Now()
catalogStr := ""
if s.catalog != nil {
catalogStr = *s.catalog
@@ -122,16 +134,16 @@ func (s *statement) executeIngest(ctx context.Context)
(int64, error) {
if s.dbSchema != nil {
dbSchemaStr = *s.dbSchema
}
- startAttrs := []any{
- slog.String("target_table", s.targetTable),
- slog.String("mode", s.ingestMode),
- slog.String("catalog", catalogStr),
- slog.String("db_schema", dbSchemaStr),
- slog.Bool("temporary", s.temporary),
- slog.Bool("streamBind", s.streamBind != nil),
- slog.Bool("recordBound", s.bound != nil),
+ startAttrs := []attribute.KeyValue{
+ attribute.String("target_table", s.targetTable),
+ attribute.String("mode", s.ingestMode),
+ attribute.String("catalog", catalogStr),
+ attribute.String("db_schema", dbSchemaStr),
+ attribute.Bool("temporary", s.temporary),
+ attribute.Bool("streamBind", s.streamBind != nil),
+ attribute.Bool("recordBound", s.bound != nil),
}
- s.log.InfoContext(ctx, "FlightSQL ExecuteIngest start", startAttrs...)
+ span.AddEvent("flight.ingest.started",
trace.WithAttributes(startAttrs...))
opts := ingestOptions{
targetTable: s.targetTable,
@@ -145,16 +157,16 @@ func (s *statement) executeIngest(ctx context.Context)
(int64, error) {
// Get the record reader to ingest
var rdr array.RecordReader
- var err error
if s.streamBind != nil {
rdr = s.streamBind
} else {
rdr, err = createRecordReaderFromBatch(s.bound)
if err != nil {
- s.log.WarnContext(ctx, "FlightSQL ExecuteIngest
finished with error",
- slog.Duration("duration",
time.Since(startTime)),
- "err", err,
- )
+ span.RecordError(err, trace.WithAttributes(
+ attribute.String("flight.stage",
"create_record_reader"),
+ attribute.Float64("duration_s",
time.Since(startTime).Seconds()),
+ ), trace.WithStackTrace(true))
+ errorRecorded = true
return -1, err
}
}
@@ -163,20 +175,20 @@ func (s *statement) executeIngest(ctx context.Context)
(int64, error) {
var header, trailer metadata.MD
callOpts := append([]grpc.CallOption{}, grpc.Header(&header),
grpc.Trailer(&trailer), s.timeouts)
- nRows, err := s.cnxn.cl.ExecuteIngest(ctx, rdr, ingestOpts, callOpts...)
- finishAttrs := []any{
- slog.Duration("duration", time.Since(startTime)),
- slog.Int64("rowsIngested", nRows),
+ nRows, err = s.cnxn.cl.ExecuteIngest(ctx, rdr, ingestOpts, callOpts...)
+ finishAttrs := []attribute.KeyValue{
+ attribute.Float64("duration_s",
time.Since(startTime).Seconds()),
+ attribute.Int64("rowsIngested", nRows),
}
- finishAttrs = append(finishAttrs, correlationHeaderAttrs(header)...)
- finishAttrs = append(finishAttrs, correlationHeaderAttrs(trailer)...)
+ finishAttrs = append(finishAttrs, correlationHeaderKeyValues(header)...)
+ finishAttrs = append(finishAttrs,
correlationHeaderKeyValues(trailer)...)
if err != nil {
- wrapped := adbcFromFlightStatusWithDetails(err, header,
trailer, "ExecuteIngest")
- finishAttrs = append(finishAttrs, "err", wrapped)
- s.log.WarnContext(ctx, "FlightSQL ExecuteIngest finished with
error", finishAttrs...)
- return -1, wrapped
+ err = adbcFromFlightStatusWithDetails(err, header, trailer,
"FlightSQL.BulkIngest.Execute")
+ span.RecordError(err, trace.WithAttributes(finishAttrs...),
trace.WithStackTrace(true))
+ errorRecorded = true
+ return -1, err
}
- s.log.InfoContext(ctx, "FlightSQL ExecuteIngest finished",
finishAttrs...)
+ span.AddEvent("flight.ingest.completed",
trace.WithAttributes(finishAttrs...))
return nRows, nil
}
diff --git a/go/adbc/driver/flightsql/flightsql_connection.go
b/go/adbc/driver/flightsql/flightsql_connection.go
index ffacba5da..692f573e1 100644
--- a/go/adbc/driver/flightsql/flightsql_connection.go
+++ b/go/adbc/driver/flightsql/flightsql_connection.go
@@ -280,7 +280,9 @@ func doGetWithTracer(ctx context.Context, cl
*flightsql.Client, endpoint *flight
}
if err != nil {
attrs = append(attrs, attribute.String("flight.stage",
"do_get"))
- span.RecordError(err, trace.WithAttributes(attrs...),
trace.WithStackTrace(true))
+ if !isRecordReaderSiblingCancellation(ctx) {
+ span.RecordError(err,
trace.WithAttributes(attrs...), trace.WithStackTrace(true))
+ }
errorRecorded = true
} else {
span.AddEvent("flight.location.selected",
trace.WithAttributes(attrs...))
@@ -353,10 +355,12 @@ func doGetWithTracer(ctx context.Context, cl
*flightsql.Client, endpoint *flight
attribute.String("error.message", err.Error()),
))
err = fmt.Errorf("all DoGet attempts failed: %s; final:
%w", strings.Join(attemptErrors, "; "), err)
- span.RecordError(err, trace.WithAttributes(
- attribute.String("flight.stage",
"all_locations_failed"),
- attribute.Int("flight.location.attempt_count",
len(attemptErrors)),
- ), trace.WithStackTrace(true))
+ if !isRecordReaderSiblingCancellation(ctx) {
+ span.RecordError(err, trace.WithAttributes(
+ attribute.String("flight.stage",
"all_locations_failed"),
+
attribute.Int("flight.location.attempt_count", len(attemptErrors)),
+ ), trace.WithStackTrace(true))
+ }
errorRecorded = true
return nil, err
}
@@ -372,10 +376,12 @@ func doGetWithTracer(ctx context.Context, cl
*flightsql.Client, endpoint *flight
len(attemptErrors), strings.Join(attemptErrors, "; "),
err)
}
if err != nil {
- span.RecordError(err, trace.WithAttributes(
- attribute.String("flight.stage",
"all_locations_failed"),
- attribute.Int("flight.location.attempt_count",
len(attemptErrors)),
- ), trace.WithStackTrace(true))
+ if !isRecordReaderSiblingCancellation(ctx) {
+ span.RecordError(err, trace.WithAttributes(
+ attribute.String("flight.stage",
"all_locations_failed"),
+ attribute.Int("flight.location.attempt_count",
len(attemptErrors)),
+ ), trace.WithStackTrace(true))
+ }
errorRecorded = true
}
diff --git a/go/adbc/driver/flightsql/flightsql_database.go
b/go/adbc/driver/flightsql/flightsql_database.go
index 1a07c9f50..10b368c32 100644
--- a/go/adbc/driver/flightsql/flightsql_database.go
+++ b/go/adbc/driver/flightsql/flightsql_database.go
@@ -394,7 +394,10 @@ func getFlightClient(ctx context.Context, loc string, d
*databaseImpl, authMiddl
Unary: unaryTimeoutInterceptor,
Stream: streamTimeoutInterceptor,
},
- {Stream: responseMetadataStreamInterceptor},
+ {
+ Unary: responseMetadataUnaryInterceptor,
+ Stream: responseMetadataStreamInterceptor,
+ },
}
if d.enableCookies {
diff --git a/go/adbc/driver/flightsql/flightsql_statement.go
b/go/adbc/driver/flightsql/flightsql_statement.go
index 57aca4f94..3e6808f3a 100644
--- a/go/adbc/driver/flightsql/flightsql_statement.go
+++ b/go/adbc/driver/flightsql/flightsql_statement.go
@@ -36,6 +36,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/flight/flightsql"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/bluele/gcache"
+ "go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
@@ -483,35 +484,32 @@ func (s *statement) SetOptionDouble(key string, value
float64) error {
// The query can then be executed with any of the Execute methods.
// For queries expected to be executed repeatedly, Prepare should be
// called before execution.
-func (s *statement) SetSqlQuery(query string) error {
+func (s *statement) SetSqlQuery(query string) (err error) {
if s.prepared != nil {
- if err := s.closePreparedStatement(); err != nil {
+ if err = s.closePreparedStatement(); err != nil {
return err
}
s.prepared = nil
}
- if err := s.clearIncrementalQuery(); err != nil {
+ if err = s.clearIncrementalQuery(); err != nil {
return err
}
s.targetTable = ""
s.query.setSqlQuery(query)
- if s.log != nil {
- s.log.Debug("FlightSQL SetSqlQuery", s.queryAttrs()...)
- }
return nil
}
-func (s *statement) queryAttrs() []any {
+func (s *statement) queryAttrs() []attribute.KeyValue {
if s.query.sqlQuery != "" {
- return queryFingerprintAttrs(s.query.sqlQuery)
+ return queryFingerprintKeyValues(s.query.sqlQuery)
}
if s.query.substraitPlan != nil {
- return substraitFingerprintAttrs(s.query.substraitPlan,
s.query.substraitVersion)
+ return substraitFingerprintKeyValues(s.query.substraitPlan,
s.query.substraitVersion)
}
if s.targetTable != "" {
- return []any{slog.String("query_type", "ingest"),
slog.String("target_table", s.targetTable)}
+ return []attribute.KeyValue{attribute.String("query_type",
"ingest"), attribute.String("target_table", s.targetTable)}
}
- return []any{slog.String("query_type", "none")}
+ return []attribute.KeyValue{attribute.String("query_type", "none")}
}
// ExecuteQuery executes the current query or prepared statement
@@ -520,7 +518,20 @@ func (s *statement) queryAttrs() []any {
//
// This invalidates any prior result sets on this statement.
func (s *statement) ExecuteQuery(ctx context.Context) (rdr array.RecordReader,
nrec int64, err error) {
- if err := s.clearIncrementalQuery(); err != nil {
+ const (
+ operationName = "ExecuteQuery"
+ spanName = "FlightSQL.Statement." + operationName
+ )
+ startTime := time.Now()
+ ctx, span := internal.StartSpan(ctx, spanName, s.cnxn,
trace.WithAttributes(traceHeaderAttrsWithPrefix(s.hdrs,
traceRequestMetadataPrefix)...))
+ defer func() {
+ internal.NewEndSpanHelper(span).
+ WithError(err).
+ WithStartTime(startTime).
+ EndSpan()
+ }()
+
+ if err = s.clearIncrementalQuery(); err != nil {
return nil, -1, err
}
@@ -532,30 +543,17 @@ func (s *statement) ExecuteQuery(ctx context.Context)
(rdr array.RecordReader, n
}
}
- ctx, span := internal.StartSpan(
- ctx,
- "FlightSQL.Statement.ExecuteQuery",
- s.cnxn,
- trace.WithAttributes(traceHeaderAttrsWithPrefix(s.hdrs,
traceRequestMetadataPrefix)...),
- )
- defer func() {
- internal.NewEndSpanHelper(span).
- WithError(err).
- EndSpan()
- }()
-
// Handle bulk ingest
if s.targetTable != "" {
nrec, err = s.executeIngest(ctx)
return nil, nrec, err
}
- startTime := time.Now()
- startAttrs := append([]any{
- slog.Bool("prepared", s.prepared != nil),
- slog.Bool("hasTxn", s.cnxn.txn != nil),
+ startAttrs := append([]attribute.KeyValue{
+ attribute.Bool("prepared", s.prepared != nil),
+ attribute.Bool("hasTxn", s.cnxn.txn != nil),
}, s.queryAttrs()...)
- s.log.InfoContext(ctx, "FlightSQL ExecuteQuery start", startAttrs...)
+ span.AddEvent("starting", trace.WithAttributes(startAttrs...))
ctx = metadata.NewOutgoingContext(ctx, s.hdrs)
var info *flight.FlightInfo
@@ -568,25 +566,20 @@ func (s *statement) ExecuteQuery(ctx context.Context)
(rdr array.RecordReader, n
}
defer func() {
- finishAttrs := []any{
- slog.Duration("duration", time.Since(startTime)),
- slog.String("phase", "GetFlightInfo"),
+ finishAttrs := []attribute.KeyValue{
+ attribute.Float64("duration_s",
time.Since(startTime).Seconds()),
+ attribute.String("flight.stage", "get_flight_info"),
}
if info != nil {
- finishAttrs = append(finishAttrs,
flightInfoLogAttrs(info)...)
- }
- finishAttrs = append(finishAttrs,
correlationHeaderAttrs(header)...)
- finishAttrs = append(finishAttrs,
correlationHeaderAttrs(trailer)...)
- if err != nil {
- finishAttrs = append(finishAttrs, "err", err)
- s.log.WarnContext(ctx, "FlightSQL ExecuteQuery finished
with error", finishAttrs...)
- } else {
- s.log.InfoContext(ctx, "FlightSQL ExecuteQuery
finished", finishAttrs...)
+ finishAttrs = append(finishAttrs,
flightInfoTracingKeyValues(info)...)
}
+ finishAttrs = append(finishAttrs,
correlationHeaderKeyValues(header)...)
+ finishAttrs = append(finishAttrs,
correlationHeaderKeyValues(trailer)...)
+ span.AddEvent("finished", trace.WithAttributes(finishAttrs...))
}()
if err != nil {
- return nil, -1, adbcFromFlightStatusWithDetails(err, header,
trailer, "ExecuteQuery")
+ return nil, -1, adbcFromFlightStatusWithDetails(err, header,
trailer, operationName)
}
nrec = info.TotalRecords
@@ -596,7 +589,7 @@ func (s *statement) ExecuteQuery(ctx context.Context) (rdr
array.RecordReader, n
info: info,
clientCache: s.clientCache,
bufferSize: s.queueSize,
- logger: s.log,
+ tracing: s.cnxn,
}, s.timeouts)
return
}
@@ -604,7 +597,20 @@ func (s *statement) ExecuteQuery(ctx context.Context) (rdr
array.RecordReader, n
// ExecuteUpdate executes a statement that does not generate a result
// set. It returns the number of rows affected if known, otherwise -1.
func (s *statement) ExecuteUpdate(ctx context.Context) (n int64, err error) {
- if err := s.clearIncrementalQuery(); err != nil {
+ const (
+ operationName = "ExecuteUpdate"
+ spanName = "FlightSQL.Statement." + operationName
+ )
+ startTime := time.Now()
+ ctx, span := internal.StartSpan(ctx, spanName, s.cnxn,
trace.WithAttributes(traceHeaderAttrsWithPrefix(s.hdrs,
traceRequestMetadataPrefix)...))
+ defer func() {
+ internal.NewEndSpanHelper(span).
+ WithError(err).
+ WithStartTime(startTime).
+ EndSpan()
+ }()
+
+ if err = s.clearIncrementalQuery(); err != nil {
return -1, err
}
@@ -616,29 +622,16 @@ func (s *statement) ExecuteUpdate(ctx context.Context) (n
int64, err error) {
}
}
- ctx, span := internal.StartSpan(
- ctx,
- "FlightSQL.Statement.ExecuteUpdate",
- s.cnxn,
- trace.WithAttributes(traceHeaderAttrsWithPrefix(s.hdrs,
traceRequestMetadataPrefix)...),
- )
- defer func() {
- internal.NewEndSpanHelper(span).
- WithError(err).
- EndSpan()
- }()
-
// Handle bulk ingest
if s.targetTable != "" {
return s.executeIngest(ctx)
}
- startTime := time.Now()
- startAttrs := append([]any{
- slog.Bool("prepared", s.prepared != nil),
- slog.Bool("hasTxn", s.cnxn.txn != nil),
+ startAttrs := append([]attribute.KeyValue{
+ attribute.Bool("prepared", s.prepared != nil),
+ attribute.Bool("hasTxn", s.cnxn.txn != nil),
}, s.queryAttrs()...)
- s.log.InfoContext(ctx, "FlightSQL ExecuteUpdate start", startAttrs...)
+ span.AddEvent("starting", trace.WithAttributes(startAttrs...))
ctx = metadata.NewOutgoingContext(ctx, s.hdrs)
var header, trailer metadata.MD
@@ -650,22 +643,17 @@ func (s *statement) ExecuteUpdate(ctx context.Context) (n
int64, err error) {
}
defer func() {
- finishAttrs := []any{
- slog.Duration("duration", time.Since(startTime)),
- slog.Int64("rowsAffected", n),
- }
- finishAttrs = append(finishAttrs,
correlationHeaderAttrs(header)...)
- finishAttrs = append(finishAttrs,
correlationHeaderAttrs(trailer)...)
- if err != nil {
- finishAttrs = append(finishAttrs, "err", err)
- s.log.WarnContext(ctx, "FlightSQL ExecuteUpdate
finished with error", finishAttrs...)
- } else {
- s.log.InfoContext(ctx, "FlightSQL ExecuteUpdate
finished", finishAttrs...)
+ finishAttrs := []attribute.KeyValue{
+ attribute.Float64("duration_s",
time.Since(startTime).Seconds()),
+ attribute.Int64("rows_affected", n),
}
+ finishAttrs = append(finishAttrs,
correlationHeaderKeyValues(header)...)
+ finishAttrs = append(finishAttrs,
correlationHeaderKeyValues(trailer)...)
+ span.AddEvent("finished", trace.WithAttributes(finishAttrs...))
}()
if err != nil {
- err = adbcFromFlightStatusWithDetails(err, header, trailer,
"ExecuteQuery")
+ err = adbcFromFlightStatusWithDetails(err, header, trailer,
operationName)
}
return
@@ -674,39 +662,35 @@ func (s *statement) ExecuteUpdate(ctx context.Context) (n
int64, err error) {
// Prepare turns this statement into a prepared statement to be executed
// multiple times. This invalidates any prior result sets.
func (s *statement) Prepare(ctx context.Context) (err error) {
- ctx, span := internal.StartSpan(
- ctx,
- "FlightSQL.Statement.Prepare",
- s.cnxn,
- trace.WithAttributes(traceHeaderAttrsWithPrefix(s.hdrs,
traceRequestMetadataPrefix)...),
+ const (
+ operationName = "Prepare"
+ spanName = "FlightSQL.Statement." + operationName
)
+ startTime := time.Now()
+ ctx, span := internal.StartSpan(ctx, spanName, s.cnxn,
trace.WithAttributes(traceHeaderAttrsWithPrefix(s.hdrs,
traceRequestMetadataPrefix)...))
defer func() {
internal.NewEndSpanHelper(span).
WithError(err).
+ WithStartTime(startTime).
EndSpan()
}()
- startTime := time.Now()
- s.log.InfoContext(ctx, "FlightSQL Prepare start", s.queryAttrs()...)
+ span.AddEvent("starting", trace.WithAttributes(s.queryAttrs()...))
ctx = metadata.NewOutgoingContext(ctx, s.hdrs)
var header, trailer metadata.MD
- prep, err := s.query.prepare(ctx, s.cnxn, grpc.Header(&header),
grpc.Trailer(&trailer), s.timeouts)
+ var prep *flightsql.PreparedStatement
+ prep, err = s.query.prepare(ctx, s.cnxn, grpc.Header(&header),
grpc.Trailer(&trailer), s.timeouts)
defer func() {
- finishAttrs := []any{slog.Duration("duration",
time.Since(startTime))}
- finishAttrs = append(finishAttrs,
correlationHeaderAttrs(header)...)
- finishAttrs = append(finishAttrs,
correlationHeaderAttrs(trailer)...)
- if err != nil {
- finishAttrs = append(finishAttrs, "err", err)
- s.log.WarnContext(ctx, "FlightSQL Prepare finished with
error", finishAttrs...)
- } else {
- s.log.InfoContext(ctx, "FlightSQL Prepare finished",
finishAttrs...)
- }
+ finishAttrs :=
[]attribute.KeyValue{attribute.Float64("duration_s",
time.Since(startTime).Seconds())}
+ finishAttrs = append(finishAttrs,
correlationHeaderKeyValues(header)...)
+ finishAttrs = append(finishAttrs,
correlationHeaderKeyValues(trailer)...)
+ span.AddEvent("finished", trace.WithAttributes(finishAttrs...))
}()
if err != nil {
- return adbcFromFlightStatusWithDetails(err, header, trailer,
"Prepare")
+ return adbcFromFlightStatusWithDetails(err, header, trailer,
operationName)
}
s.prepared = prep
return nil
diff --git a/go/adbc/driver/flightsql/flightsql_tracing.go
b/go/adbc/driver/flightsql/flightsql_tracing.go
index 853be6897..18cf0f504 100644
--- a/go/adbc/driver/flightsql/flightsql_tracing.go
+++ b/go/adbc/driver/flightsql/flightsql_tracing.go
@@ -19,13 +19,17 @@ package flightsql
import (
"context"
+ "crypto/sha256"
"encoding/hex"
"fmt"
+ "slices"
"sync"
"time"
"github.com/apache/arrow-go/v18/arrow/flight"
"go.opentelemetry.io/otel/attribute"
+ "go.opentelemetry.io/otel/trace"
+ "golang.org/x/exp/maps"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
@@ -64,6 +68,41 @@ func (c *responseMetadataCollector) snapshot() metadata.MD {
return c.value.Copy()
}
+func responseMetadataUnaryInterceptor(ctx context.Context, method string, req,
reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts
...grpc.CallOption) (err error) {
+ span := trace.SpanFromContext(ctx)
+
+ // Ignore errors
+ outgoing, _ := metadata.FromOutgoingContext(ctx)
+ err = invoker(ctx, method, req, reply, cc, opts...)
+
+ if span.IsRecording() {
+ keys := maps.Keys(outgoing)
+ slices.Sort(keys)
+ args := []attribute.KeyValue{
+ attribute.String("target", cc.Target()),
+ attribute.StringSlice("metadata", keys),
+ }
+ // Surface curated outbound correlation IDs regardless of level.
+ args = append(args, outgoingCallHeaderKeyValues(ctx)...)
+ args = append(args, grpcStatusKeyValues(err)...)
+ span.AddEvent("Metadata.Unary.Interceptor."+method,
trace.WithAttributes(args...))
+ }
+ return err
+}
+
+// outgoingCallHeaderAttrs returns slog attributes for well-known correlation
+// headers on ctx's outbound gRPC metadata. Uses the "out_hdr_" prefix.
+func outgoingCallHeaderKeyValues(ctx context.Context) []attribute.KeyValue {
+ if ctx == nil {
+ return nil
+ }
+ md, ok := metadata.FromOutgoingContext(ctx)
+ if !ok {
+ return nil
+ }
+ return headerKeyValuesWithPrefix(md, "out_hdr_")
+}
+
func responseMetadataStreamInterceptor(ctx context.Context, desc
*grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer,
opts ...grpc.CallOption) (grpc.ClientStream, error) {
stream, err := streamer(ctx, desc, cc, method, opts...)
if err != nil {
@@ -176,6 +215,40 @@ func grpcStatusKeyValues(err error) []attribute.KeyValue {
}
}
+// queryFingerprintKeyValues builds OpenTelemetry attributes identifying a SQL
query
+// without exposing it: length and a SHA-256 prefix. The query text itself
+// is never recorded because it can embed end-user PII as literals.
+func queryFingerprintKeyValues(query string) []attribute.KeyValue {
+ if query == "" {
+ return []attribute.KeyValue{attribute.String("query_type",
"empty")}
+ }
+ h := sha256.Sum256([]byte(query))
+ return []attribute.KeyValue{
+ attribute.String("query_type", "sql"),
+ attribute.Int("query_length", len(query)),
+ attribute.String("query_sha256_prefix",
hex.EncodeToString(h[:8])),
+ }
+}
+
+// substraitFingerprintKeyValues builds OpenTelemetry attributes identifying a
Substrait
+// plan: length, SHA-256 prefix, and protocol version. Plan bytes are never
+// recorded.
+func substraitFingerprintKeyValues(plan []byte, version string)
[]attribute.KeyValue {
+ if len(plan) == 0 {
+ return []attribute.KeyValue{attribute.String("query_type",
"substrait_empty")}
+ }
+ h := sha256.Sum256(plan)
+ attrs := []attribute.KeyValue{
+ attribute.String("query_type", "substrait"),
+ attribute.Int("substrait_plan_bytes", len(plan)),
+ attribute.String("substrait_plan_sha256_prefix",
hex.EncodeToString(h[:8])),
+ }
+ if version != "" {
+ attrs = append(attrs, attribute.String("substrait_version",
version))
+ }
+ return attrs
+}
+
// flightInfoTracingKeyValues returns OpenTelemetry attributes describing a
FlightInfo:
// descriptor type and command prefix, AppMetadata prefix (some backends
// embed a server-side query handle there), and advisory record/byte
diff --git a/go/adbc/driver/flightsql/logging.go
b/go/adbc/driver/flightsql/logging.go
index e60627c33..34221841f 100644
--- a/go/adbc/driver/flightsql/logging.go
+++ b/go/adbc/driver/flightsql/logging.go
@@ -20,14 +20,12 @@ package flightsql
import (
"context"
"crypto/rand"
- "crypto/sha256"
"encoding/hex"
"io"
"log/slog"
"strconv"
"time"
- "github.com/apache/arrow-go/v18/arrow/flight"
"go.opentelemetry.io/otel/trace"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
@@ -230,80 +228,3 @@ func newRandomID(prefix string) string {
}
return prefix + "-" + hex.EncodeToString(b[:])
}
-
-// queryFingerprintAttrs builds slog attributes identifying a SQL query
-// without exposing it: length and a SHA-256 prefix. The query text itself
-// is never logged because it can embed end-user PII as literals.
-func queryFingerprintAttrs(query string) []any {
- if query == "" {
- return []any{slog.String("query_type", "empty")}
- }
- h := sha256.Sum256([]byte(query))
- return []any{
- slog.String("query_type", "sql"),
- slog.Int("query_length", len(query)),
- slog.String("query_sha256_prefix", hex.EncodeToString(h[:8])),
- }
-}
-
-// substraitFingerprintAttrs builds slog attributes identifying a Substrait
-// plan: length, SHA-256 prefix, and protocol version. Plan bytes are never
-// logged.
-func substraitFingerprintAttrs(plan []byte, version string) []any {
- if len(plan) == 0 {
- return []any{slog.String("query_type", "substrait_empty")}
- }
- h := sha256.Sum256(plan)
- attrs := []any{
- slog.String("query_type", "substrait"),
- slog.Int("substrait_plan_bytes", len(plan)),
- slog.String("substrait_plan_sha256_prefix",
hex.EncodeToString(h[:8])),
- }
- if version != "" {
- attrs = append(attrs, slog.String("substrait_version", version))
- }
- return attrs
-}
-
-// flightInfoLogAttrs returns slog attributes describing a FlightInfo:
-// descriptor type and command prefix, AppMetadata prefix (some backends
-// embed a server-side query handle there), and advisory record/byte
-// counts. Returns nil for a nil info.
-func flightInfoLogAttrs(info *flight.FlightInfo) []any {
- if info == nil {
- return nil
- }
- attrs := []any{
- slog.Int("numEndpoints", len(info.Endpoint)),
- slog.Int64("totalRecords", info.TotalRecords),
- slog.Int64("totalBytes", info.TotalBytes),
- slog.Bool("haveSchemaInFlightInfo", len(info.Schema) > 0),
- }
- if desc := info.FlightDescriptor; desc != nil {
- attrs = append(attrs, slog.String("descriptorType",
desc.Type.String()))
- if len(desc.Cmd) > 0 {
- limit := len(desc.Cmd)
- if limit > maxLoggedBlobBytes {
- limit = maxLoggedBlobBytes
- }
- attrs = append(attrs,
- slog.Int("descriptorCmdBytes", len(desc.Cmd)),
- slog.String("descriptorCmdPrefixHex",
hex.EncodeToString(desc.Cmd[:limit])),
- )
- }
- if len(desc.Path) > 0 {
- attrs = append(attrs, slog.Any("descriptorPath",
desc.Path))
- }
- }
- if len(info.AppMetadata) > 0 {
- limit := len(info.AppMetadata)
- if limit > maxLoggedBlobBytes {
- limit = maxLoggedBlobBytes
- }
- attrs = append(attrs,
- slog.Int("appMetadataBytes", len(info.AppMetadata)),
- slog.String("appMetadataPrefixHex",
hex.EncodeToString(info.AppMetadata[:limit])),
- )
- }
- return attrs
-}
diff --git a/go/adbc/driver/flightsql/record_reader.go
b/go/adbc/driver/flightsql/record_reader.go
index 2cfbeb9d9..289fc49a2 100644
--- a/go/adbc/driver/flightsql/record_reader.go
+++ b/go/adbc/driver/flightsql/record_reader.go
@@ -21,7 +21,6 @@ import (
"context"
"errors"
"fmt"
- "log/slog"
"sync/atomic"
"time"
@@ -54,6 +53,16 @@ type reader struct {
var errReaderReleased = errors.New("record reader released")
+type recordReaderCallerContextKey struct{}
+
+// isRecordReaderSiblingCancellation reports whether the record reader's
+// derived context was canceled while its original caller context remains
+// active, indicating that another endpoint goroutine triggered cancellation.
+func isRecordReaderSiblingCancellation(ctx context.Context) bool {
+ callerCtx, ok :=
ctx.Value(recordReaderCallerContextKey{}).(context.Context)
+ return ok && ctx.Err() == context.Canceled && callerCtx.Err() == nil
+}
+
// recordReaderConfig bundles the dependencies that newRecordReader
// needs to spin up its per-endpoint goroutines.
type recordReaderConfig struct {
@@ -63,7 +72,6 @@ type recordReaderConfig struct {
clientCache gcache.Cache
bufferSize int
tracing adbc.OTelTracing
- logger *slog.Logger
}
// newRecordReader kicks off a goroutine for each endpoint and returns a
@@ -108,6 +116,7 @@ func newRecordReader(ctx context.Context, cfg
recordReaderConfig, opts ...grpc.C
callerCtx := ctx
group, ctx := errgroup.WithContext(ctx)
ctx, cancelFn := context.WithCancelCause(ctx)
+ ctx = context.WithValue(ctx, recordReaderCallerContextKey{}, callerCtx)
goEndpoint := func(endpointFn func() error) {
group.Go(func() error {
err := endpointFn()
@@ -233,6 +242,9 @@ func newRecordReader(ctx context.Context, cfg
recordReaderConfig, opts ...grpc.C
endpointCtx, responseMetadata :=
withResponseMetadata(ctx)
rdr, err := doGetWithTracer(endpointCtx, cfg.cl,
endpoint, cfg.clientCache, cfg.tracing, opts...)
if err != nil {
+ if checkRecordReaderContext(err, ctx,
callerCtx) == nil {
+ return nil
+ }
span.RecordError(err, trace.WithAttributes(
append(
append([]attribute.KeyValue{},
epAttrs...),
diff --git a/go/adbc/driver/flightsql/record_reader_test.go
b/go/adbc/driver/flightsql/record_reader_test.go
index 987f533a7..75713dac8 100644
--- a/go/adbc/driver/flightsql/record_reader_test.go
+++ b/go/adbc/driver/flightsql/record_reader_test.go
@@ -75,6 +75,10 @@ func (f *testFlightService) DoGet(request *flight.Ticket,
stream flight.FlightSe
f.failureCount--
return fmt.Errorf("Failed request")
}
+ if request.Ticket[0] == 125 {
+ <-stream.Context().Done()
+ return stream.Context().Err()
+ }
schema := orderingSchema()
wr := flight.NewRecordWriter(stream, ipc.WithSchema(schema))
@@ -302,7 +306,7 @@ func (suite *RecordReaderTests)
TestSiblingCancellationRecordsOneException() {
Schema: flight.SerializeSchema(orderingSchema(),
suite.alloc),
Endpoint: []*flight.FlightEndpoint{
{Ticket: &flight.Ticket{Ticket: []byte{127}}},
- {Ticket: &flight.Ticket{Ticket: []byte{126}}},
+ {Ticket: &flight.Ticket{Ticket: []byte{125}}},
},
},
clientCache: suite.clCache,
diff --git a/go/adbc/driver/flightsql/tracing_test.go
b/go/adbc/driver/flightsql/tracing_test.go
index 8dfca8c1c..be1771582 100644
--- a/go/adbc/driver/flightsql/tracing_test.go
+++ b/go/adbc/driver/flightsql/tracing_test.go
@@ -26,9 +26,67 @@ import (
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"
"go.opentelemetry.io/otel/trace"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
)
+func TestResponseMetadataUnaryInterceptorDoesNotEndParentSpan(t *testing.T) {
+ recorder := tracetest.NewSpanRecorder()
+ tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder))
+ t.Cleanup(func() {
+ if err := tp.Shutdown(context.Background()); err != nil {
+ t.Fatalf("TracerProvider shutdown failed: %v", err)
+ }
+ })
+
+ conn, err := grpc.NewClient(
+ "passthrough:///test",
+ grpc.WithTransportCredentials(insecure.NewCredentials()),
+ )
+ if err != nil {
+ t.Fatalf("grpc.NewClient() failed: %v", err)
+ }
+ t.Cleanup(func() {
+ if err := conn.Close(); err != nil {
+ t.Fatalf("ClientConn.Close() failed: %v", err)
+ }
+ })
+
+ ctx, span := tp.Tracer("test.flightsql").Start(context.Background(),
"operation")
+ err = responseMetadataUnaryInterceptor(
+ ctx,
+ "/test.Service/Method",
+ nil,
+ nil,
+ conn,
+ func(context.Context, string, any, any, *grpc.ClientConn,
...grpc.CallOption) error {
+ return nil
+ },
+ )
+ if err != nil {
+ t.Fatalf("responseMetadataUnaryInterceptor() failed: %v", err)
+ }
+ if !span.IsRecording() {
+ t.Fatal("responseMetadataUnaryInterceptor() ended the parent
span")
+ }
+ if got := len(recorder.Ended()); got != 0 {
+ t.Fatalf("ended spans before parent Span.End() = %d, want 0",
got)
+ }
+
+ span.End()
+ spans := recorder.Ended()
+ if len(spans) != 1 {
+ t.Fatalf("ended spans len = %d, want 1", len(spans))
+ }
+ for _, event := range spans[0].Events() {
+ if event.Name ==
"Metadata.Unary.Interceptor./test.Service/Method" {
+ return
+ }
+ }
+ t.Fatal("parent span does not contain the unary metadata event")
+}
+
func TestTraceHeaderAttrsWithPrefix_AllowAndDeny(t *testing.T) {
md := metadata.New(map[string]string{
"x-request-id": "req-1",
