diff --git a/go/adbc/driver/flightsql/flightsql_bulk_ingest.go b/go/adbc/driver/flightsql/flightsql_bulk_ingest.go index fea5604d5f..7952a020ec 100644 --- a/go/adbc/driver/flightsql/flightsql_bulk_ingest.go +++ b/go/adbc/driver/flightsql/flightsql_bulk_ingest.go @@ -20,14 +20,15 @@ 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 +106,14 @@ 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 span trace.Span + ctx, span = internal.StartSpan(ctx, "statement.executeIngest", s) + defer func() { + span.SetAttributes(attribute.Int64("ingested.row_count", nRows)) + internal.EndSpan(span, err) + }() + if s.streamBind == nil && s.bound == nil { return -1, adbc.Error{ Msg: "[Flight SQL Statement] must call Bind before bulk ingestion", @@ -113,7 +121,6 @@ func (s *statement) executeIngest(ctx context.Context) (int64, error) { } } - startTime := time.Now() catalogStr := "" if s.catalog != nil { catalogStr = *s.catalog @@ -122,16 +129,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.SetAttributes(startAttrs...) opts := ingestOptions{ targetTable: s.targetTable, @@ -145,16 +152,11 @@ 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, - ) return -1, err } } @@ -163,20 +165,17 @@ 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.Int64("rowsIngested", nRows), } finishAttrs = append(finishAttrs, correlationHeaderAttrs(header)...) finishAttrs = append(finishAttrs, correlationHeaderAttrs(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, "ExecuteIngest") + return -1, err } - s.log.InfoContext(ctx, "FlightSQL ExecuteIngest finished", finishAttrs...) + span.SetAttributes(finishAttrs...) return nRows, nil } diff --git a/go/adbc/driver/flightsql/flightsql_connection.go b/go/adbc/driver/flightsql/flightsql_connection.go index 11d9a6473d..3adfa152be 100644 --- a/go/adbc/driver/flightsql/flightsql_connection.go +++ b/go/adbc/driver/flightsql/flightsql_connection.go @@ -23,7 +23,6 @@ import ( "encoding/json" "fmt" "io" - "log/slog" "math" "strings" "time" @@ -39,6 +38,8 @@ import ( flightproto "github.com/apache/arrow-go/v18/arrow/flight/gen/flight" "github.com/apache/arrow-go/v18/arrow/ipc" "github.com/bluele/gcache" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" "google.golang.org/grpc" grpccodes "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" @@ -232,22 +233,38 @@ var adbcToFlightSQLInfo = map[adbc.InfoCode]flightsql.SqlInfo{ adbc.InfoVendorSubstraitMaxVersion: flightsql.SqlInfoFlightSqlServerSubstraitMaxVersion, } -// doGetWithLogger performs DoGet against an endpoint's locations, logging each +// doGetWithTracing performs DoGet against an endpoint's locations, logging each // attempt and joining all per-location failures into the returned error so the // caller can see every location that was tried. logger may be nil. -func doGetWithLogger(ctx context.Context, cl *flightsql.Client, endpoint *flight.FlightEndpoint, clientCache gcache.Cache, logger *slog.Logger, opts ...grpc.CallOption) (rdr *flight.Reader, err error) { - log := safeLogger(logger) +func doGetWithTracing( + ctx context.Context, + cl *flightsql.Client, + endpoint *flight.FlightEndpoint, + clientCache gcache.Cache, + tracing adbc.OTelTracing, + opts ...grpc.CallOption, +) (rdr *flight.Reader, err error) { + var span trace.Span + ctx, span = internal.StartSpan(ctx, "flightsql_connection.doGet", tracing) + defer func() { + internal.EndSpan(span, err) + }() + + const prefix string = "flightsql_connection.doGet" if len(endpoint.Location) == 0 { - log.DebugContext(ctx, "FlightSQL doGet", - "phase", "noLocations", - ) + span.AddEvent(prefix + ".noLocations") start := time.Now() rdr, err = cl.DoGet(ctx, endpoint.Ticket, opts...) - log.DebugContext(ctx, "FlightSQL doGet", - "phase", "defaultClientResult", - "duration", time.Since(start), - "err", err, - ) + if err != nil { + span.RecordError(err, trace.WithAttributes( + attribute.String("phase", prefix+".defaultClientResult"), + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + )) + } else { + span.AddEvent(prefix+".defaultClientResult.success", trace.WithAttributes( + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + )) + } return rdr, err } @@ -267,12 +284,11 @@ func doGetWithLogger(ctx context.Context, cl *flightsql.Client, endpoint *flight cc, err = clientCache.Get(loc.Uri) if err != nil { attemptErrors = append(attemptErrors, fmt.Sprintf("clientCache.Get(%q): %s", loc.Uri, err.Error())) - log.WarnContext(ctx, "FlightSQL doGet location attempt failed", - "phase", "clientCacheGet", - "location", loc.Uri, - "duration", time.Since(start), - "err", err, - ) + span.RecordError(err, trace.WithAttributes( + attribute.String("phase", prefix+".clientCacheGet"), + attribute.String("location", loc.Uri), + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + )) continue } @@ -280,20 +296,19 @@ func doGetWithLogger(ctx context.Context, cl *flightsql.Client, endpoint *flight rdr, err = conn.DoGet(ctx, endpoint.Ticket, opts...) if err != nil { attemptErrors = append(attemptErrors, fmt.Sprintf("DoGet(%q): %s", loc.Uri, err.Error())) - log.WarnContext(ctx, "FlightSQL doGet location attempt failed", - "phase", "doGet", - "location", loc.Uri, - "duration", time.Since(start), - "err", err, - ) + span.RecordError(err, trace.WithAttributes( + attribute.String("phase", prefix+".conn.doGet"), + attribute.String("location", loc.Uri), + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + )) continue } - log.DebugContext(ctx, "FlightSQL doGet succeeded", - "location", loc.Uri, - "duration", time.Since(start), - ) - return + span.AddEvent(prefix+".doGet.success", trace.WithAttributes( + attribute.String("location", loc.Uri), + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + )) + return rdr, nil } if hasFallback { @@ -301,15 +316,15 @@ func doGetWithLogger(ctx context.Context, cl *flightsql.Client, endpoint *flight rdr, err = cl.DoGet(ctx, endpoint.Ticket, opts...) if err != nil { attemptErrors = append(attemptErrors, fmt.Sprintf("DoGet(fallback to default client): %s", err.Error())) - log.WarnContext(ctx, "FlightSQL doGet fallback to default client failed", - "duration", time.Since(start), - "err", err, - ) + span.RecordError(err, trace.WithAttributes( + attribute.String("phase", prefix+".defaultClientFallback"), + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + )) return nil, fmt.Errorf("all DoGet attempts failed: %s; final: %w", strings.Join(attemptErrors, "; "), err) } - log.DebugContext(ctx, "FlightSQL doGet succeeded via default client fallback", - "duration", time.Since(start), - ) + span.AddEvent(prefix+".defaultClientFallback.success", trace.WithAttributes( + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + )) return rdr, nil } @@ -704,7 +719,7 @@ func (c *connectionImpl) PrepareDriverInfo(ctx context.Context, infoCodes []adbc // No error, go get the SqlInfo from the server for i, endpoint := range info.Endpoint { var header, trailer metadata.MD - rdr, err := doGetWithLogger(ctx, c.cl, endpoint, c.clientCache, c.Logger, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts) + rdr, err := doGetWithTracing(ctx, c.cl, endpoint, c.clientCache, c, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts) if err != nil { return adbcFromFlightStatusWithDetails(err, header, trailer, "GetInfo(DoGet): endpoint %d: %s", i, endpoint.Location) } @@ -769,7 +784,7 @@ func (c *connectionImpl) readInfo(ctx context.Context, expectedSchema *arrow.Sch info: info, clientCache: c.clientCache, bufferSize: 5, - logger: c.Logger, + tracing: c, }, opts...) if err != nil { return nil, adbcFromFlightStatus(err, "DoGet") @@ -967,7 +982,7 @@ func (c *connectionImpl) GetTableSchema(ctx context.Context, catalog *string, db header = metadata.MD{} trailer = metadata.MD{} - rdr, err := doGetWithLogger(ctx, c.cl, info.Endpoint[0], c.clientCache, c.Logger, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer)) + rdr, err := doGetWithTracing(ctx, c.cl, info.Endpoint[0], c.clientCache, c, c.timeouts, grpc.Header(&header), grpc.Trailer(&trailer)) if err != nil { return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "GetTableSchema(DoGet)") } @@ -1043,7 +1058,7 @@ func (c *connectionImpl) GetTableTypes(ctx context.Context) (array.RecordReader, info: info, clientCache: c.clientCache, bufferSize: 5, - logger: c.Logger, + tracing: c, }) } @@ -1096,21 +1111,15 @@ func (c *connectionImpl) Rollback(ctx context.Context) error { // NewStatement initializes a new statement object tied to this connection func (c *connectionImpl) NewStatement() (adbc.Statement, error) { id := newRandomID("stmt") - // Build a statement-scoped logger so every record emitted for this - // statement carries both connection_id (inherited from c.Logger via the - // With() called in databaseImpl.Open) and statement_id. The discard - // fallback in safeLogger keeps callers free of nil-checks if no logger - // is wired up by the host. - log := safeLogger(c.Logger).With("statement_id", id) return &statement{ - alloc: c.db.Alloc, - clientCache: c.clientCache, - hdrs: c.hdrs.Copy(), - queueSize: 5, - timeouts: c.timeouts, - cnxn: c, - id: id, - log: log, + StatementImplBase: driverbase.NewStatementImplBase(c.Base(), c.ErrorHelper), + alloc: c.db.Alloc, + clientCache: c.clientCache, + hdrs: c.hdrs.Copy(), + queueSize: 5, + timeouts: c.timeouts, + cnxn: c, + id: id, }, nil } @@ -1195,28 +1204,34 @@ func (c *connectionImpl) prepareSubstrait(ctx context.Context, plan flightsql.Su } // Close closes this connection and releases any associated resources. -func (c *connectionImpl) Close() error { +func (c *connectionImpl) Close() (err error) { + args := []attribute.KeyValue{ + attribute.String("connection_id", c.id), + } + ctx, span := internal.StartSpan(context.Background(), "connectionImpl.Close", c) + defer func() { + span.SetAttributes(args...) + internal.EndSpan(span, err) + }() if c.cl == nil { - return adbc.Error{ + err = adbc.Error{ Msg: "[Flight SQL Connection] trying to close already closed connection", Code: adbc.StatusInvalidState, } + return err } - closeStart := time.Now() // Snapshot fields before tearing down c.cl; log "closing" and // "closed" separately so a hung CloseSession is still visible. - logger := safeLogger(c.Logger) - connID := c.id - openedAt := c.openedAt - logger.Info("FlightSQL connection closing", - "connection_id", connID, - ) + connectionID := c.id + span.AddEvent("closing", trace.WithAttributes( + attribute.String("connection_id", connectionID), + )) - ctx := metadata.NewOutgoingContext(context.Background(), c.hdrs) + ctx = metadata.NewOutgoingContext(ctx, c.hdrs) var header, trailer metadata.MD - _, err := c.cl.CloseSession(ctx, &flight.CloseSessionRequest{}, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts) + _, err = c.cl.CloseSession(ctx, &flight.CloseSessionRequest{}, grpc.Header(&header), grpc.Trailer(&trailer), c.timeouts) if err != nil { grpcStatus := grpcstatus.Convert(err) // Ignore unimplemented @@ -1231,22 +1246,17 @@ func (c *connectionImpl) Close() error { err = c.cl.Close() c.cl = nil - args := []any{ - "connection_id", connID, - "close_duration", time.Since(closeStart), - } - if !openedAt.IsZero() { - args = append(args, "lifetime", time.Since(openedAt)) - } - if err != nil { - args = append(args, "err", err) - args = append(args, grpcStatusAttrs(err)...) - logger.Info("FlightSQL connection closed with error", args...) - } else { - logger.Info("FlightSQL connection closed", args...) + span.AddEvent("closed", trace.WithAttributes( + attribute.String("connection_id", connectionID), + )) + + if !c.openedAt.IsZero() { + args = append(args, attribute.Float64("lifetime_seconds", time.Since(c.openedAt).Seconds())) } + args = append(args, grpcStatusAttrs(err)...) - return adbcFromFlightStatus(err, "Close") + err = adbcFromFlightStatus(err, "Close") + return err } // ReadPartition constructs a statement for a partition of a query. The @@ -1271,7 +1281,7 @@ func (c *connectionImpl) ReadPartition(ctx context.Context, serializedPartition } ctx = metadata.NewOutgoingContext(ctx, c.hdrs) - rdr, err = doGetWithLogger(ctx, c.cl, info.Endpoint[0], c.clientCache, c.Logger, c.timeouts) + rdr, err = doGetWithTracing(ctx, c.cl, info.Endpoint[0], c.clientCache, c, c.timeouts) if err != nil { return nil, adbcFromFlightStatus(err, "ReadPartition(DoGet)") } diff --git a/go/adbc/driver/flightsql/flightsql_database.go b/go/adbc/driver/flightsql/flightsql_database.go index 259e534edb..dc1afcb413 100644 --- a/go/adbc/driver/flightsql/flightsql_database.go +++ b/go/adbc/driver/flightsql/flightsql_database.go @@ -30,11 +30,14 @@ import ( "time" "github.com/apache/arrow-adbc/go/adbc" + "github.com/apache/arrow-adbc/go/adbc/driver/internal" "github.com/apache/arrow-adbc/go/adbc/driver/internal/driverbase" "github.com/apache/arrow-go/v18/arrow/array" "github.com/apache/arrow-go/v18/arrow/flight" "github.com/apache/arrow-go/v18/arrow/flight/flightsql" "github.com/bluele/gcache" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" @@ -376,11 +379,18 @@ func (d *databaseImpl) Close() error { return nil } -func getFlightClient(ctx context.Context, loc string, d *databaseImpl, authMiddle *bearerAuthMiddleware, cookies flight.CookieMiddleware) (*flightsql.Client, error) { +func getFlightClient(ctx context.Context, loc string, d *databaseImpl, authMiddle *bearerAuthMiddleware, cookies flight.CookieMiddleware) (client *flightsql.Client, err error) { + var exitAttrs []attribute.KeyValue + var span trace.Span + ctx, span = internal.StartSpan(ctx, "flightsql_database.getFlightClient", d) + defer func() { + span.SetAttributes(exitAttrs...) + internal.EndSpan(span, err) + }() middleware := []flight.ClientMiddleware{ { - Unary: makeUnaryLoggingInterceptor(d.Logger), - Stream: makeStreamLoggingInterceptor(d.Logger), + Unary: makeUnaryLoggingInterceptor(d), + Stream: makeStreamLoggingInterceptor(d), }, flight.CreateClientMiddleware(authMiddle), { @@ -393,9 +403,11 @@ func getFlightClient(ctx context.Context, loc string, d *databaseImpl, authMiddl middleware = append(middleware, flight.CreateClientMiddleware(cookies)) } - uri, err := url.Parse(loc) + var uri *url.URL + uri, err = url.Parse(loc) if err != nil { - return nil, adbc.Error{Msg: fmt.Sprintf("Invalid URI '%s': %s", loc, err), Code: adbc.StatusInvalidArgument} + err = adbc.Error{Msg: fmt.Sprintf("Invalid URI '%s': %s", loc, err), Code: adbc.StatusInvalidArgument} + return nil, err } creds := d.creds @@ -417,7 +429,7 @@ func getFlightClient(ctx context.Context, loc string, d *databaseImpl, authMiddl dialOpts = append(dialOpts, grpc.WithPerRPCCredentials(d.oauthToken)) } - d.Logger.DebugContext(ctx, "new client", "location", loc) + span.AddEvent("new_client", trace.WithAttributes(attribute.String("location", loc))) cl, err := flightsql.NewClient(target, nil, middleware, dialOpts...) if err != nil { return nil, adbc.Error{ @@ -429,38 +441,44 @@ func getFlightClient(ctx context.Context, loc string, d *databaseImpl, authMiddl cl.Alloc = d.Alloc // Authorization header is already set, continue if len(authMiddle.hdrs.Get("authorization")) > 0 { - d.Logger.DebugContext(ctx, "reusing auth token", "location", loc) + span.SetAttributes(attribute.Bool("reusing_auth_token", true)) return cl, nil } + span.SetAttributes(attribute.Bool("reusing_auth_token", false)) var authValue string if d.user != "" || d.pass != "" { authStart := time.Now() - d.Logger.InfoContext(ctx, "FlightSQL basic auth started", - "target", loc, - "user", d.user, - ) + span.AddEvent("basic_auth.start", trace.WithAttributes( + attribute.String("target", loc), + attribute.String("user", d.user), + )) var header, trailer metadata.MD ctx, err = cl.Client.AuthenticateBasicToken(ctx, d.user, d.pass, grpc.Header(&header), grpc.Trailer(&trailer), d.timeout) if err != nil { - args := []any{ - "target", loc, - "user", d.user, - "duration", time.Since(authStart), - "err", err, + exitAttrs = []attribute.KeyValue{ + attribute.String("target", loc), + attribute.String("user", d.user), + attribute.String("err", err.Error()), } - args = append(args, correlationHeaderAttrs(header)...) - args = append(args, correlationHeaderAttrs(trailer)...) - args = append(args, grpcStatusAttrs(err)...) - d.Logger.InfoContext(ctx, "FlightSQL basic auth failed", args...) - return nil, adbcFromFlightStatusWithDetails(err, header, trailer, "AuthenticateBasicToken") + exitAttrs = append(exitAttrs, correlationHeaderAttrs(header)...) + exitAttrs = append(exitAttrs, correlationHeaderAttrs(trailer)...) + exitAttrs = append(exitAttrs, grpcStatusAttrs(err)...) + err = adbcFromFlightStatusWithDetails(err, header, trailer, "AuthenticateBasicToken") + return nil, err } if md, ok := metadata.FromOutgoingContext(ctx); ok { authValue = md.Get("Authorization")[0] } + span.AddEvent("basic_auth.end", trace.WithAttributes( + attribute.String("target", loc), + attribute.String("user", d.user), + attribute.Float64("duration_seconds", time.Since(authStart).Seconds()), + attribute.Int("token_length", len(authValue)), + )) d.Logger.InfoContext(ctx, "FlightSQL basic auth succeeded", "target", loc, "user", d.user, @@ -538,7 +556,7 @@ func (d *databaseImpl) Open(ctx context.Context) (adbc.Connection, error) { const int32code = 3 for _, endpoint := range info.Endpoint { - rdr, err := doGetWithLogger(ctx, cl, endpoint, cache, d.Logger, d.timeout) + rdr, err := doGetWithTracing(ctx, cl, endpoint, cache, d, d.timeout) if err != nil { continue } diff --git a/go/adbc/driver/flightsql/flightsql_statement.go b/go/adbc/driver/flightsql/flightsql_statement.go index 61705ae6f4..88b95a4b4f 100644 --- a/go/adbc/driver/flightsql/flightsql_statement.go +++ b/go/adbc/driver/flightsql/flightsql_statement.go @@ -20,7 +20,6 @@ package flightsql import ( "context" "fmt" - "log/slog" "math" "strconv" "strings" @@ -29,12 +28,17 @@ import ( "unsafe" "github.com/apache/arrow-adbc/go/adbc" + "github.com/apache/arrow-adbc/go/adbc/driver/internal" + "github.com/apache/arrow-adbc/go/adbc/driver/internal/driverbase" "github.com/apache/arrow-go/v18/arrow" "github.com/apache/arrow-go/v18/arrow/array" "github.com/apache/arrow-go/v18/arrow/flight" "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" + semconv "go.opentelemetry.io/otel/semconv/v1.30.0" + "go.opentelemetry.io/otel/trace" "google.golang.org/grpc" "google.golang.org/grpc/metadata" "google.golang.org/protobuf/proto" @@ -156,6 +160,7 @@ type incrementalState struct { } type statement struct { + driverbase.StatementImplBase alloc memory.Allocator cnxn *connectionImpl clientCache gcache.Cache @@ -181,7 +186,6 @@ type statement struct { bound arrow.RecordBatch streamBind array.RecordReader id string - log *slog.Logger } func (s *statement) closePreparedStatement() error { @@ -481,25 +485,34 @@ 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) { + var span trace.Span + _, span = internal.StartSpan(context.Background(), "statement.SetSqlQuery", s, + trace.WithAttributes( + append( + s.queryAttrs(), + semconv.DBQueryTextKey.String(query), + )..., + )) + defer func() { + span.SetAttributes(semconv.DBQueryTextKey.String(query)) + internal.EndSpan(span, err) + }() 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) } @@ -507,9 +520,9 @@ func (s *statement) queryAttrs() []any { return substraitFingerprintAttrs(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 @@ -518,16 +531,33 @@ 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) { + var span trace.Span + ctx, span = internal.StartSpan(ctx, "statement.ExecuteQuery", s, trace.WithAttributes( + append( + s.queryAttrs(), + attribute.Bool("prepared", s.prepared != nil), + attribute.Bool("hasTxn", s.cnxn.txn != nil), + )...)) + defer func() { + span.SetAttributes(semconv.DBResponseReturnedRowsKey.Int64(nrec)) + internal.EndSpan(span, err) + }() + + // startTime := time.Now() + span.AddEvent("start") + if err := s.clearIncrementalQuery(); err != nil { return nil, -1, err } // Reject staged binds if no ingest target was provided if s.targetTable == "" && s.prepared == nil && (s.bound != nil || s.streamBind != nil) { - return nil, -1, adbc.Error{ + nrec = -1 + err = adbc.Error{ Msg: "[Flight SQL Statement] must set IngestTargetTable before bulk ingestion", Code: adbc.StatusInvalidState, } + return nil, nrec, err } // Handle bulk ingest @@ -536,13 +566,6 @@ func (s *statement) ExecuteQuery(ctx context.Context) (rdr array.RecordReader, n return nil, nrec, err } - startTime := time.Now() - startAttrs := append([]any{ - slog.Bool("prepared", s.prepared != nil), - slog.Bool("hasTxn", s.cnxn.txn != nil), - }, s.queryAttrs()...) - s.log.InfoContext(ctx, "FlightSQL ExecuteQuery start", startAttrs...) - ctx = metadata.NewOutgoingContext(ctx, s.hdrs) var info *flight.FlightInfo var header, trailer metadata.MD @@ -554,25 +577,21 @@ 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.Duration("duration", time.Since(startTime)), + attribute.String("phase", "GetFlightInfo"), } 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...) - } + span.AddEvent("finished", trace.WithAttributes(finishAttrs...)) }() if err != nil { - return nil, -1, adbcFromFlightStatusWithDetails(err, header, trailer, "ExecuteQuery") + err = adbcFromFlightStatusWithDetails(err, header, trailer, "ExecuteQuery") + return nil, -1, err } nrec = info.TotalRecords @@ -582,7 +601,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, }, s.timeouts) return } @@ -590,6 +609,17 @@ 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) { + var span trace.Span + var finishAttrs []attribute.KeyValue + ctx, span = internal.StartSpan(ctx, "statement.ExecuteUpdate", s, trace.WithAttributes( + attribute.Bool("prepared", s.prepared != nil), + attribute.Bool("hasTxn", s.cnxn.txn != nil), + )) + defer func() { + span.SetAttributes(finishAttrs...) + internal.EndSpan(span, err) + }() + if err := s.clearIncrementalQuery(); err != nil { return -1, err } @@ -607,12 +637,12 @@ func (s *statement) ExecuteUpdate(ctx context.Context) (n int64, err error) { return s.executeIngest(ctx) } - startTime := time.Now() - startAttrs := append([]any{ - slog.Bool("prepared", s.prepared != nil), - slog.Bool("hasTxn", s.cnxn.txn != nil), + // startTime := time.Now() + 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("start", trace.WithAttributes(startAttrs...)) ctx = metadata.NewOutgoingContext(ctx, s.hdrs) var header, trailer metadata.MD @@ -624,18 +654,9 @@ 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, attribute.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...) - } }() if err != nil { @@ -647,28 +668,28 @@ 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) error { - startTime := time.Now() - s.log.InfoContext(ctx, "FlightSQL Prepare start", s.queryAttrs()...) +func (s *statement) Prepare(ctx context.Context) (err error) { + var span trace.Span + ctx, span = internal.StartSpan(ctx, "statement.Prepare", s, trace.WithAttributes(s.queryAttrs()...)) + finishAttrs := []attribute.KeyValue{} + defer func() { + span.SetAttributes(finishAttrs...) + internal.EndSpan(span, err) + }() 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...) - } }() if err != nil { - return adbcFromFlightStatusWithDetails(err, header, trailer, "Prepare") + err = adbcFromFlightStatusWithDetails(err, header, trailer, "Prepare") + return err } s.prepared = prep return nil diff --git a/go/adbc/driver/flightsql/logging.go b/go/adbc/driver/flightsql/logging.go index 9dae400b8d..af764c369e 100644 --- a/go/adbc/driver/flightsql/logging.go +++ b/go/adbc/driver/flightsql/logging.go @@ -22,12 +22,16 @@ import ( "crypto/rand" "crypto/sha256" "encoding/hex" + "fmt" "io" "log/slog" "strconv" "time" + "github.com/apache/arrow-adbc/go/adbc" + "github.com/apache/arrow-adbc/go/adbc/driver/internal" "github.com/apache/arrow-go/v18/arrow/flight" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" "golang.org/x/exp/maps" "golang.org/x/exp/slices" @@ -54,28 +58,28 @@ const maxLoggedBlobBytes = 32 // endpointLogAttrs builds slog attributes describing a Flight endpoint // (index, ticket length, locations) for per-endpoint log records. Ticket // contents are intentionally never logged. -func endpointLogAttrs(endpointIndex, numEndpoints int, endpoint *flight.FlightEndpoint) []any { - attrs := []any{ - slog.Int("endpointIndex", endpointIndex), - slog.Int("numEndpoints", numEndpoints), +func endpointLogAttrs(endpointIndex, numEndpoints int, endpoint *flight.FlightEndpoint) []attribute.KeyValue { + attrs := []attribute.KeyValue{ + attribute.Int("endpointIndex", endpointIndex), + attribute.Int("numEndpoints", numEndpoints), } if endpoint == nil { return attrs } if endpoint.Ticket != nil { - attrs = append(attrs, slog.Int("ticketBytes", len(endpoint.Ticket.Ticket))) + attrs = append(attrs, attribute.Int("ticketBytes", len(endpoint.Ticket.Ticket))) } if len(endpoint.Location) == 0 { - attrs = append(attrs, slog.String("locations", "")) + attrs = append(attrs, attribute.String("locations", "")) } else { uris := make([]string, 0, len(endpoint.Location)) for _, loc := range endpoint.Location { uris = append(uris, loc.Uri) } - attrs = append(attrs, slog.Any("locations", uris)) + attrs = append(attrs, attribute.StringSlice("locations", uris)) } if endpoint.ExpirationTime != nil { - attrs = append(attrs, slog.Time("expirationTime", endpoint.ExpirationTime.AsTime())) + attrs = append(attrs, attribute.String("expirationTime", endpoint.ExpirationTime.AsTime().Format(time.RFC3339))) } return attrs } @@ -109,20 +113,20 @@ func (p *streamProgress) recordBatch(rows int64, bytes int64) { } // logAttrs returns slog attributes summarizing this stream's progress. -func (p *streamProgress) logAttrs() []any { - attrs := []any{ - slog.Int64("batchesRead", p.batchesRead), - slog.Int64("recordsRead", p.recordsRead), - slog.Int64("approxBytesRead", p.bytesEstimate), - slog.Duration("elapsed", time.Since(p.start)), +func (p *streamProgress) logAttrs() []attribute.KeyValue { + attrs := []attribute.KeyValue{ + attribute.Int64("batchesRead", p.batchesRead), + attribute.Int64("recordsRead", p.recordsRead), + attribute.Int64("approxBytesRead", p.bytesEstimate), + attribute.Float64("elapsed", time.Since(p.start).Seconds()), } if !p.firstBatchAt.IsZero() { - attrs = append(attrs, slog.Duration("timeToFirstBatch", p.firstBatchAt.Sub(p.start))) + attrs = append(attrs, attribute.Float64("timeToFirstBatch", p.firstBatchAt.Sub(p.start).Seconds())) } else { - attrs = append(attrs, slog.String("timeToFirstBatch", "never")) + attrs = append(attrs, attribute.String("timeToFirstBatch", "never")) } if !p.lastBatchAt.IsZero() { - attrs = append(attrs, slog.Duration("timeSinceLastBatch", time.Since(p.lastBatchAt))) + attrs = append(attrs, attribute.Float64("timeSinceLastBatch", time.Since(p.lastBatchAt).Seconds())) } return attrs } @@ -143,46 +147,64 @@ func formatInt(n int64) string { return strconv.FormatInt(n, 10) } -func makeUnaryLoggingInterceptor(logger *slog.Logger) grpc.UnaryClientInterceptor { - interceptor := func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { +func makeUnaryLoggingInterceptor(tracing adbc.OTelTracing) grpc.UnaryClientInterceptor { + interceptor := func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) { + var span trace.Span + ctx, span = internal.StartSpan(ctx, method, tracing) + defer internal.EndSpan(span, err) + start := time.Now() // Ignore errors outgoing, _ := metadata.FromOutgoingContext(ctx) - err := invoker(ctx, method, req, reply, cc, opts...) - if logger.Enabled(ctx, slog.LevelDebug) { - args := []any{"target", cc.Target(), "duration", time.Since(start), "err", err, "metadata", outgoing} - args = append(args, outgoingCallHeaderAttrs(ctx)...) - args = append(args, grpcStatusAttrs(err)...) - logger.DebugContext(ctx, method, args...) - } else { - keys := maps.Keys(outgoing) - slices.Sort(keys) - args := []any{"target", cc.Target(), "duration", time.Since(start), "err", err, "metadata", keys} - // Surface curated outbound correlation IDs regardless of level. - args = append(args, outgoingCallHeaderAttrs(ctx)...) - args = append(args, grpcStatusAttrs(err)...) - logger.InfoContext(ctx, method, args...) + err = invoker(ctx, method, req, reply, cc, opts...) + if err != nil { + span.RecordError(err) + } + if span.IsRecording() { + attrs := []attribute.KeyValue{ + attribute.String("target", cc.Target()), + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + attribute.String("metadata", fmt.Sprint(outgoing)), + } + attrs = append(attrs, outgoingCallHeaderAttrs(ctx)...) + attrs = append(attrs, grpcStatusAttrs(err)...) + span.AddEvent(method, trace.WithAttributes(attrs...)) } return err } return interceptor } -func makeStreamLoggingInterceptor(logger *slog.Logger) grpc.StreamClientInterceptor { - interceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { +func makeStreamLoggingInterceptor(tracing adbc.OTelTracing) grpc.StreamClientInterceptor { + interceptor := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (stream grpc.ClientStream, err error) { + var span trace.Span + ctx, span = internal.StartSpan(ctx, method, tracing) + defer internal.EndSpan(span, err) start := time.Now() // Ignore errors outgoing, _ := metadata.FromOutgoingContext(ctx) - stream, err := streamer(ctx, desc, cc, method, opts...) + stream, err = streamer(ctx, desc, cc, method, opts...) if err != nil { - args := []any{"target", cc.Target(), "duration", time.Since(start), "err", err} + args := []attribute.KeyValue{ + attribute.String("target", cc.Target()), + attribute.Float64("duration_seconds", time.Since(start).Seconds()), + attribute.String("err_summary", err.Error()), + } args = append(args, outgoingCallHeaderAttrs(ctx)...) args = append(args, grpcStatusAttrs(err)...) - logger.InfoContext(ctx, method, args...) + span.RecordError(err, trace.WithAttributes(args...)) return stream, err } - return &loggedStream{ClientStream: stream, logger: logger, ctx: ctx, method: method, start: start, target: cc.Target(), outgoing: outgoing}, err + return &loggedStream{ + ClientStream: stream, + span: span, + ctx: ctx, + method: method, + start: start, + target: cc.Target(), + outgoing: outgoing, + }, err } return interceptor } @@ -190,7 +212,7 @@ func makeStreamLoggingInterceptor(logger *slog.Logger) grpc.StreamClientIntercep type loggedStream struct { grpc.ClientStream - logger *slog.Logger + span trace.Span ctx context.Context method string start time.Time @@ -215,31 +237,36 @@ func (stream *loggedStream) RecvMsg(m any) error { loggedErr = nil } + errSummary := "" + if loggedErr != nil { + errSummary = loggedErr.Error() + } + // Capture trailers from the terminated stream; they often carry // server-side diagnostic information for failure triage. trailer := stream.Trailer() - if stream.logger.Enabled(stream.ctx, slog.LevelDebug) { - stream.logger.DebugContext(stream.ctx, stream.method, - "target", stream.target, - "duration", time.Since(stream.start), - "err", loggedErr, - "recvMessages", stream.recvCount, - "metadata", stream.outgoing, - "trailer", trailer, - ) + if stream.span.IsRecording() { + stream.span.AddEvent(stream.method, trace.WithAttributes( + attribute.String("target", stream.target), + attribute.Float64("duration_seconds", time.Since(stream.start).Seconds()), + attribute.String("err_summary", errSummary), + attribute.Int64("recv_messages", stream.recvCount), + attribute.String("metadata", fmt.Sprint(stream.outgoing)), + attribute.String("trailer", fmt.Sprint(trailer)), + )) } else { keys := maps.Keys(stream.outgoing) slices.Sort(keys) trailerKeys := maps.Keys(trailer) slices.Sort(trailerKeys) - args := []any{ - "target", stream.target, - "duration", time.Since(stream.start), - "err", loggedErr, - "recvMessages", stream.recvCount, - "metadata", keys, - "trailer", trailerKeys, + args := []attribute.KeyValue{ + attribute.String("target", stream.target), + attribute.Float64("duration_seconds", time.Since(stream.start).Seconds()), + attribute.String("err_summary", errSummary), + attribute.Int64("recv_messages", stream.recvCount), + attribute.String("metadata", fmt.Sprint(keys)), + attribute.String("trailer", fmt.Sprint(trailerKeys)), } // Promote curated correlation headers from the trailer. args = append(args, correlationHeaderAttrs(trailer)...) @@ -250,7 +277,7 @@ func (stream *loggedStream) RecvMsg(m any) error { if loggedErr != nil { args = append(args, grpcStatusAttrs(loggedErr)...) } - stream.logger.InfoContext(stream.ctx, stream.method, args...) + stream.span.AddEvent(stream.method, trace.WithAttributes(args...)) } return err } @@ -289,14 +316,14 @@ var wellKnownCorrelationHeaders = []string{ // correlationHeaderAttrs (incoming) and outgoingCallHeaderAttrs // (outbound). Only headers in wellKnownCorrelationHeaders are emitted; // returns nil when none are present. -func headerAttrsWithPrefix(md metadata.MD, prefix string) []any { +func headerAttrsWithPrefix(md metadata.MD, prefix string) []attribute.KeyValue { if len(md) == 0 { return nil } - out := make([]any, 0, 4) + out := make([]attribute.KeyValue, 0, 4) for _, k := range wellKnownCorrelationHeaders { if vals := md.Get(k); len(vals) > 0 { - out = append(out, slog.Any(prefix+k, vals)) + out = append(out, attribute.StringSlice(prefix+k, vals)) } } return out @@ -305,13 +332,13 @@ func headerAttrsWithPrefix(md metadata.MD, prefix string) []any { // correlationHeaderAttrs returns slog attributes for well-known correlation // headers present in md (typically incoming headers/trailers). Uses the // "hdr_" prefix; only allow-listed headers are emitted. -func correlationHeaderAttrs(md metadata.MD) []any { +func correlationHeaderAttrs(md metadata.MD) []attribute.KeyValue { return headerAttrsWithPrefix(md, "hdr_") } // outgoingCallHeaderAttrs returns slog attributes for well-known correlation // headers on ctx's outbound gRPC metadata. Uses the "out_hdr_" prefix. -func outgoingCallHeaderAttrs(ctx context.Context) []any { +func outgoingCallHeaderAttrs(ctx context.Context) []attribute.KeyValue { if ctx == nil { return nil } @@ -324,7 +351,7 @@ func outgoingCallHeaderAttrs(ctx context.Context) []any { // grpcStatusAttrs returns "grpc_code" and "grpc_message" slog attributes // for the gRPC status embedded in err, or nil if err has no status. -func grpcStatusAttrs(err error) []any { +func grpcStatusAttrs(err error) []attribute.KeyValue { if err == nil { return nil } @@ -332,9 +359,9 @@ func grpcStatusAttrs(err error) []any { if !ok { return nil } - return []any{ - slog.String("grpc_code", st.Code().String()), - slog.String("grpc_message", st.Message()), + return []attribute.KeyValue{ + attribute.String("grpc_code", st.Code().String()), + attribute.String("grpc_message", st.Message()), } } @@ -397,33 +424,33 @@ func newRandomID(prefix string) string { // 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 { +func queryFingerprintAttrs(query string) []attribute.KeyValue { if query == "" { - return []any{slog.String("query_type", "empty")} + return []attribute.KeyValue{attribute.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])), + return []attribute.KeyValue{ + attribute.String("query_type", "sql"), + attribute.Int("query_length", len(query)), + attribute.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 { +func substraitFingerprintAttrs(plan []byte, version string) []attribute.KeyValue { if len(plan) == 0 { - return []any{slog.String("query_type", "substrait_empty")} + return []attribute.KeyValue{attribute.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])), + 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, slog.String("substrait_version", version)) + attrs = append(attrs, attribute.String("substrait_version", version)) } return attrs } @@ -432,30 +459,30 @@ func substraitFingerprintAttrs(plan []byte, version string) []any { // 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 { +func flightInfoLogAttrs(info *flight.FlightInfo) []attribute.KeyValue { 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), + attrs := []attribute.KeyValue{ + attribute.Int("numEndpoints", len(info.Endpoint)), + attribute.Int64("totalRecords", info.TotalRecords), + attribute.Int64("totalBytes", info.TotalBytes), + attribute.Bool("haveSchemaInFlightInfo", len(info.Schema) > 0), } if desc := info.FlightDescriptor; desc != nil { - attrs = append(attrs, slog.String("descriptorType", desc.Type.String())) + attrs = append(attrs, attribute.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])), + attribute.Int("descriptorCmdBytes", len(desc.Cmd)), + attribute.String("descriptorCmdPrefixHex", hex.EncodeToString(desc.Cmd[:limit])), ) } if len(desc.Path) > 0 { - attrs = append(attrs, slog.Any("descriptorPath", desc.Path)) + attrs = append(attrs, attribute.String("descriptorPath", fmt.Sprint(desc.Path))) } } if len(info.AppMetadata) > 0 { @@ -464,8 +491,8 @@ func flightInfoLogAttrs(info *flight.FlightInfo) []any { limit = maxLoggedBlobBytes } attrs = append(attrs, - slog.Int("appMetadataBytes", len(info.AppMetadata)), - slog.String("appMetadataPrefixHex", hex.EncodeToString(info.AppMetadata[:limit])), + attribute.Int("appMetadataBytes", len(info.AppMetadata)), + attribute.String("appMetadataPrefixHex", hex.EncodeToString(info.AppMetadata[:limit])), ) } return attrs diff --git a/go/adbc/driver/flightsql/logging_test.go b/go/adbc/driver/flightsql/logging_test.go index 4673a8c2c6..9790e0d87a 100644 --- a/go/adbc/driver/flightsql/logging_test.go +++ b/go/adbc/driver/flightsql/logging_test.go @@ -26,6 +26,7 @@ import ( "log/slog" "testing" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" @@ -296,15 +297,16 @@ func TestSafeLogger_AlwaysWrapsOtel(t *testing.T) { // (which is what slog.Any / slog.String produce). The map's values are // taken from the slog.Value's String() representation so callers can do // straightforward equality assertions without unwrapping Value kinds. -func slogAttrsToMap(t *testing.T, attrs []any) map[string]string { +func slogAttrsToMap(t *testing.T, attrs []attribute.KeyValue) map[string]string { t.Helper() out := make(map[string]string, len(attrs)) for i, a := range attrs { - attr, ok := a.(slog.Attr) + var iface interface{} = a + attr, ok := iface.(attribute.KeyValue) if !ok { - t.Fatalf("attrs[%d] is %T, want slog.Attr (value=%v)", i, a, a) + t.Fatalf("attrs[%d] is %T, want attribute.KeyValue (value=%v)", i, a, a) } - out[attr.Key] = attr.Value.String() + out[string(attr.Key)] = attr.Value.String() } return out } diff --git a/go/adbc/driver/flightsql/record_reader.go b/go/adbc/driver/flightsql/record_reader.go index 071cd1880b..1bdee0ea3b 100644 --- a/go/adbc/driver/flightsql/record_reader.go +++ b/go/adbc/driver/flightsql/record_reader.go @@ -20,10 +20,10 @@ package flightsql import ( "context" "fmt" - "log/slog" "sync/atomic" "github.com/apache/arrow-adbc/go/adbc" + "github.com/apache/arrow-adbc/go/adbc/driver/internal" "github.com/apache/arrow-adbc/go/adbc/utils" "github.com/apache/arrow-go/v18/arrow" "github.com/apache/arrow-go/v18/arrow/array" @@ -32,6 +32,8 @@ import ( "github.com/apache/arrow-go/v18/arrow/memory" "github.com/apache/arrow-go/v18/arrow/util" "github.com/bluele/gcache" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" "golang.org/x/sync/errgroup" "google.golang.org/grpc" "google.golang.org/grpc/metadata" @@ -56,14 +58,20 @@ type recordReaderConfig struct { info *flight.FlightInfo clientCache gcache.Cache bufferSize int - logger *slog.Logger + tracing adbc.OTelTracing } // newRecordReader kicks off a goroutine for each endpoint and returns a // reader which gathers all of the records as they come in. cfg.logger // may be nil. func newRecordReader(ctx context.Context, cfg recordReaderConfig, opts ...grpc.CallOption) (rdr array.RecordReader, err error) { - log := safeLogger(cfg.logger) + const prefix = "record_reader.newRecordReader" + var span trace.Span + ctx, span = internal.StartSpan(ctx, prefix, cfg.tracing) + defer func() { + internal.EndSpan(span, err) + }() + info := cfg.info endpoints := info.Endpoint var header, trailer metadata.MD @@ -92,11 +100,10 @@ func newRecordReader(ctx context.Context, cfg recordReaderConfig, opts ...grpc.C // We may mutate endpoints below numEndpoints := len(endpoints) - log.DebugContext(ctx, "FlightSQL newRecordReader start", - append([]any{ - slog.Int("bufferSize", cfg.bufferSize), - }, flightInfoLogAttrs(info)...)..., - ) + span.AddEvent(prefix+".start", trace.WithAttributes( + append(flightInfoLogAttrs(info), + attribute.Int("bufferSize", cfg.bufferSize), + )...)) defer func() { if err != nil { @@ -108,25 +115,27 @@ func newRecordReader(ctx context.Context, cfg recordReaderConfig, opts ...grpc.C if info.Schema != nil { schema, err = flight.DeserializeSchema(info.Schema, cfg.alloc) if err != nil { - return nil, adbc.Error{ + err = adbc.Error{ Msg: err.Error(), Code: adbc.StatusInvalidState} + return nil, err } } else { firstEndpoint := endpoints[0] epAttrs := endpointLogAttrs(0, numEndpoints, firstEndpoint) - log.DebugContext(ctx, "FlightSQL endpoint stream opening (schema discovery)", epAttrs...) + span.AddEvent("endpoint_stream.open.schema_discovery", trace.WithAttributes(epAttrs...)) startSchemaFetch := newStreamProgress() - rdr, err := doGetWithLogger(ctx, cfg.cl, firstEndpoint, cfg.clientCache, log, opts...) + rdr, err = doGetWithTracing(ctx, cfg.cl, firstEndpoint, cfg.clientCache, cfg.tracing, opts...) if err != nil { - log.ErrorContext(ctx, "FlightSQL endpoint DoGet failed (schema discovery)", - append(append([]any{}, epAttrs...), - "err", err, - "elapsed", startSchemaFetch.summary(), - )..., - ) - return nil, adbcFromFlightStatusWithDetails(err, header, trailer, + span.RecordError(err, trace.WithAttributes( + append( + epAttrs, + attribute.String("context", "FlightSQL endpoint DoGet failed (schema discovery)"), + attribute.String("elapsed", startSchemaFetch.summary()), + )...)) + err = adbcFromFlightStatusWithDetails(err, header, trailer, "DoGet: endpoint 0: remote: %s", firstEndpoint.Location) + return nil, err } schema = rdr.Schema() group.Go(func() error { @@ -142,20 +151,26 @@ func newRecordReader(ctx context.Context, cfg recordReaderConfig, opts ...grpc.C rec.Retain() ch <- rec } - if err := checkContext(rdr.Err(), ctx); err != nil { - log.ErrorContext(ctx, "FlightSQL endpoint stream ended with error", - append(append([]any{}, endpointLogAttrs(0, numEndpoints, firstEndpoint)...), - append([]any{"err", err}, progress.logAttrs()...)..., - )..., + if err = checkContext(rdr.Err(), ctx); err != nil { + attrs := append( + endpointLogAttrs(0, numEndpoints, firstEndpoint), + progress.logAttrs()..., ) - return adbcFromFlightStatusWithDetails(err, header, trailer, + attrs = append(attrs, + attribute.String("context", "FlightSQL endpoint stream ended with error"), + ) + err = adbcFromFlightStatusWithDetails(err, header, trailer, "DoGet: endpoint 0: remote: %s", firstEndpoint.Location) + + span.RecordError(err, trace.WithAttributes(attrs...)) + return err } - log.DebugContext(ctx, "FlightSQL endpoint stream completed", - append(append([]any{}, endpointLogAttrs(0, numEndpoints, firstEndpoint)...), - progress.logAttrs()..., + span.AddEvent("FlightSQL endpoint stream completed", trace.WithAttributes( + append(append([]attribute.KeyValue{}, + endpointLogAttrs(0, numEndpoints, firstEndpoint)..., + ), progress.logAttrs()..., )..., - ) + )) return nil }) @@ -192,30 +207,33 @@ func newRecordReader(ctx context.Context, cfg recordReaderConfig, opts ...grpc.C } epAttrs := endpointLogAttrs(logEndpointIndex, numEndpoints, endpoint) - log.DebugContext(ctx, "FlightSQL endpoint stream opening", epAttrs...) + span.AddEvent(prefix+".endpoint_stream_open", trace.WithAttributes(epAttrs...)) doGetStart := newStreamProgress() - rdr, err := doGetWithLogger(ctx, cfg.cl, endpoint, cfg.clientCache, log, opts...) + rdr, err := doGetWithTracing(ctx, cfg.cl, endpoint, cfg.clientCache, cfg.tracing, opts...) if err != nil { - log.ErrorContext(ctx, "FlightSQL endpoint DoGet failed", - append(append([]any{}, epAttrs...), - "err", err, - "elapsed", doGetStart.summary(), + span.RecordError(err, trace.WithAttributes( + append(epAttrs, + attribute.String("context", prefix+".failed"), + attribute.String("elapsed", doGetStart.summary()), )..., - ) - return adbcFromFlightStatusWithDetails(err, header, trailer, + )) + err = adbcFromFlightStatusWithDetails(err, header, trailer, "DoGet: endpoint %d: %s", logEndpointIndex, endpoint.Location) + return err } defer rdr.Release() streamSchema := utils.RemoveSchemaMetadata(rdr.Schema()) if !streamSchema.Equal(referenceSchema) { - log.ErrorContext(ctx, "FlightSQL endpoint returned inconsistent schema", - append(append([]any{}, epAttrs...), - "expectedSchema", referenceSchema.String(), - "actualSchema", streamSchema.String(), + err = fmt.Errorf("endpoint %d returned inconsistent schema: expected %s but got %s", logEndpointIndex, referenceSchema.String(), streamSchema.String()) + span.RecordError(err, trace.WithAttributes( + append(epAttrs, + attribute.String("context", prefix+".inconsistent_schema"), + attribute.String("expectedSchema", fmt.Sprint(referenceSchema)), + attribute.String("actualSchema", fmt.Sprint(streamSchema)), )..., - ) - return fmt.Errorf("endpoint %d returned inconsistent schema: expected %s but got %s", logEndpointIndex, referenceSchema.String(), streamSchema.String()) + )) + return err } progress := newStreamProgress() @@ -226,41 +244,37 @@ func newRecordReader(ctx context.Context, cfg recordReaderConfig, opts ...grpc.C chs[endpointIndex] <- rec } - if err := checkContext(rdr.Err(), ctx); err != nil { - log.ErrorContext(ctx, "FlightSQL endpoint stream ended with error", - append(append([]any{}, epAttrs...), - append([]any{"err", err}, progress.logAttrs()...)..., - )..., - ) - return adbcFromFlightStatusWithDetails(err, header, trailer, + if err = checkContext(rdr.Err(), ctx); err != nil { + span.RecordError(err, trace.WithAttributes( + append(epAttrs, append(progress.logAttrs(), + attribute.String("context", prefix+".ended_with_error"), + )...)..., + )) + err = adbcFromFlightStatusWithDetails(err, header, trailer, "DoGet: endpoint %d: %s", logEndpointIndex, endpoint.Location) + return err } - log.DebugContext(ctx, "FlightSQL endpoint stream completed", - append(append([]any{}, epAttrs...), - progress.logAttrs()..., - )..., - ) return nil }) } - go func() { + go func(span trace.Span) { err := group.Wait() reader.err = err if reader.err != nil { - log.WarnContext(ctx, "FlightSQL record reader finished with error", - "err", reader.err, - "numEndpoints", numEndpoints, - ) + span.RecordError(reader.err, trace.WithAttributes( + attribute.String("context", prefix+".finished_with_error"), + attribute.Int("numEndpoints", numEndpoints), + )) } else { - log.DebugContext(ctx, "FlightSQL record reader finished successfully", - "numEndpoints", numEndpoints, - ) + span.AddEvent(prefix+".success", trace.WithAttributes( + attribute.Int("numEndpoints", numEndpoints), + )) } // Don't close the last channel until after the group is finished, so that // Next() can only return after reader.err may have been set close(chs[lastChannelIndex]) - }() + }(span) return reader, nil } diff --git a/go/adbc/driver/flightsql/record_reader_test.go b/go/adbc/driver/flightsql/record_reader_test.go index ab7b5f1794..5c193ff023 100644 --- a/go/adbc/driver/flightsql/record_reader_test.go +++ b/go/adbc/driver/flightsql/record_reader_test.go @@ -83,7 +83,7 @@ func (f *testFlightService) DoGet(request *flight.Ticket, stream flight.FlightSe return nil } -func getFlightClientTest(ctx context.Context, loc string) (*flightsql.Client, error) { +func getFlightClientTest(_ context.Context, loc string) (*flightsql.Client, error) { uri, err := url.Parse(loc) if err != nil { return nil, err