mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

instrument workbench runtime otel spans

Browse Source
This commit is contained in:
lyon
2026-06-21 21:55:52 +08:00
parent 52abd338f7
commit ea0354aca2
4 changed files with 486 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workbenchruntime/otel.go
+321
View File
@@ -0,0 +1,321 @@
package workbenchruntime
import (
"bytes"
"context"
crand "crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"io"
"net/http"
"os"
"strconv"
"strings"
"time"
)
const (
otelTraceIDZero = "00000000000000000000000000000000"
otelSpanIDZero = "0000000000000000"
otelSpanKindInternal = 1
otelSpanKindServer = 2
otelExportTimeout = 1500 * time.Millisecond
)
type otelTraceContext struct {
TraceID string
ParentSpanID string
ServerSpanID string
}
type otelTraceContextKey struct{}
type queryStageError struct {
stage string
err error
}
func (e queryStageError) Error() string { return e.err.Error() }
func (e queryStageError) Unwrap() error { return e.err }
func wrapQueryStage(stage string, err error) error {
if err == nil || strings.TrimSpace(stage) == "" {
return err
}
if queryErrorStage(err) != "" {
return err
}
return queryStageError{stage: stage, err: err}
}
func queryErrorStage(err error) string {
var stageErr queryStageError
if errors.As(err, &stageErr) {
return stageErr.stage
}
return ""
}
func newOtelTraceContext(traceparent string) otelTraceContext {
traceID, parentSpanID, ok := parseOtelTraceparent(traceparent)
if !ok {
traceID = newOtelTraceID()
parentSpanID = ""
}
return otelTraceContext{TraceID: traceID, ParentSpanID: parentSpanID, ServerSpanID: newOtelSpanID()}
}
func (ctx otelTraceContext) traceparent() string {
if !validOtelHex(ctx.TraceID, 32, otelTraceIDZero) || !validOtelHex(ctx.ServerSpanID, 16, otelSpanIDZero) {
return ""
}
return "00-" + ctx.TraceID + "-" + ctx.ServerSpanID + "-01"
}
func newOtelTraceID() string { return randomOtelHex(16, otelTraceIDZero) }
func newOtelSpanID() string { return randomOtelHex(8, otelSpanIDZero) }
func randomOtelHex(byteCount int, zero string) string {
buf := make([]byte, byteCount)
if _, err := crand.Read(buf); err != nil {
return strings.TrimSuffix(zero, "0") + "1"
}
value := hex.EncodeToString(buf)
if value == zero {
return strings.TrimSuffix(zero, "0") + "1"
}
return value
}
func parseOtelTraceparent(value string) (string, string, bool) {
parts := strings.Split(strings.ToLower(strings.TrimSpace(value)), "-")
if len(parts) != 4 {
return "", "", false
}
traceID := parts[1]
spanID := parts[2]
if !validOtelHex(traceID, 32, otelTraceIDZero) || !validOtelHex(spanID, 16, otelSpanIDZero) {
return "", "", false
}
return traceID, spanID, true
}
func validOtelHex(value string, length int, zero string) bool {
if len(value) != length || value == zero {
return false
}
for _, ch := range value {
if (ch < '0' || ch > '9') && (ch < 'a' || ch > 'f') {
return false
}
}
return true
}
func (s *Server) otelHTTPHandler(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
startedAt := time.Now().UTC()
trace := newOtelTraceContext(r.Header.Get("traceparent"))
if traceparent := trace.traceparent(); traceparent != "" {
w.Header().Set("traceparent", traceparent)
w.Header().Set("x-hwlab-otel-trace-id", trace.TraceID)
}
ctx := context.WithValue(r.Context(), otelTraceContextKey{}, trace)
writer := &otelStatusWriter{ResponseWriter: w, statusCode: http.StatusOK}
next.ServeHTTP(writer, r.WithContext(ctx))
statusCode := writer.statusCode
errorCode := ""
if statusCode >= 500 {
errorCode = "http_" + strconv.Itoa(statusCode)
}
s.emitOtelSpanAsync("workbench-runtime.http "+r.Method+" "+r.URL.Path, trace, trace.ServerSpanID, trace.ParentSpanID, otelSpanKindServer, startedAt, map[string]any{
"http.request.method": r.Method,
"http.route": r.URL.Path,
"http.response.status_code": int64(statusCode),
"hwlab.runtime.stage": "http.server.request",
"valuesRedacted": true,
}, statusCode, errorCode)
})
}
type otelStatusWriter struct {
http.ResponseWriter
statusCode int
}
func (w *otelStatusWriter) WriteHeader(statusCode int) {
w.statusCode = statusCode
w.ResponseWriter.WriteHeader(statusCode)
}
func (w *otelStatusWriter) Write(payload []byte) (int, error) {
if w.statusCode == 0 {
w.statusCode = http.StatusOK
}
return w.ResponseWriter.Write(payload)
}
func otelTraceFromContext(ctx context.Context) (otelTraceContext, bool) {
trace, ok := ctx.Value(otelTraceContextKey{}).(otelTraceContext)
if !ok || !validOtelHex(trace.TraceID, 32, otelTraceIDZero) {
return otelTraceContext{}, false
}
return trace, true
}
func (s *Server) withOtelInternalSpan(ctx context.Context, name string, attrs map[string]any, fn func(context.Context) error) error {
trace, ok := otelTraceFromContext(ctx)
if !ok {
return fn(ctx)
}
startedAt := time.Now().UTC()
spanID := newOtelSpanID()
err := fn(ctx)
errorCode := ""
if err != nil {
errorCode = otelFirst(queryErrorStage(err), sqlErrorCode(err), "query_failed")
}
attributes := map[string]any{
"hwlab.runtime.stage": name,
"valuesRedacted": true,
}
for key, value := range attrs {
if value != nil {
attributes[key] = value
}
}
s.emitOtelSpanAsync(name, trace, spanID, trace.ServerSpanID, otelSpanKindInternal, startedAt, attributes, 0, errorCode)
return err
}
func (s *Server) emitOtelSpanAsync(name string, trace otelTraceContext, spanID string, parentSpanID string, kind int, startedAt time.Time, attrs map[string]any, httpStatus int, errorCode string) {
go s.emitOtelSpan(context.Background(), name, trace, spanID, parentSpanID, kind, startedAt, attrs, httpStatus, errorCode)
}
func (s *Server) emitOtelSpan(ctx context.Context, name string, trace otelTraceContext, spanID string, parentSpanID string, kind int, startedAt time.Time, attrs map[string]any, httpStatus int, errorCode string) {
endpoint := resolveOtelTracesEndpoint()
if endpoint == "" || !validOtelHex(trace.TraceID, 32, otelTraceIDZero) || !validOtelHex(spanID, 16, otelSpanIDZero) {
return
}
endedAt := time.Now().UTC()
if startedAt.IsZero() {
startedAt = endedAt
}
attributes := map[string]any{"otel.trace_id": trace.TraceID}
for key, value := range attrs {
if value != nil {
attributes[key] = value
}
}
if errorCode != "" {
attributes["error.code"] = errorCode
}
statusCode := 1
status := map[string]any{"code": statusCode}
if errorCode != "" || httpStatus >= 500 {
statusCode = 2
status = map[string]any{"code": statusCode, "message": otelFirst(errorCode, "request_failed")}
}
span := map[string]any{
"traceId": trace.TraceID,
"spanId": spanID,
"name": name,
"kind": kind,
"startTimeUnixNano": strconv.FormatInt(startedAt.UTC().UnixNano(), 10),
"endTimeUnixNano": strconv.FormatInt(endedAt.UnixNano(), 10),
"attributes": otelAttributes(attributes),
"status": status,
}
if validOtelHex(parentSpanID, 16, otelSpanIDZero) {
span["parentSpanId"] = parentSpanID
}
body := map[string]any{
"resourceSpans": []any{map[string]any{
"resource": map[string]any{"attributes": otelAttributes(s.otelResourceAttributes())},
"scopeSpans": []any{map[string]any{
"scope": map[string]any{"name": "hwlab.workbench-runtime", "version": "1"},
"spans": []any{span},
}},
}},
}
payload, err := json.Marshal(body)
if err != nil {
return
}
requestCtx, cancel := context.WithTimeout(ctx, otelExportTimeout)
defer cancel()
request, err := http.NewRequestWithContext(requestCtx, http.MethodPost, endpoint, bytes.NewReader(payload))
if err != nil {
return
}
request.Header.Set("Content-Type", "application/json")
client := http.Client{Timeout: otelExportTimeout}
response, err := client.Do(request)
if err != nil {
return
}
_, _ = io.Copy(io.Discard, response.Body)
_ = response.Body.Close()
}
func resolveOtelTracesEndpoint() string {
if endpoint := otelFirst(os.Getenv("HWLAB_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"), os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")); endpoint != "" {
return strings.TrimRight(endpoint, "/")
}
if endpoint := otelFirst(os.Getenv("HWLAB_OTEL_EXPORTER_OTLP_ENDPOINT"), os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")); endpoint != "" {
return strings.TrimRight(endpoint, "/") + "/v1/traces"
}
return ""
}
func (s *Server) otelResourceAttributes() map[string]any {
return map[string]any{
"service.name": otelFirst(os.Getenv("OTEL_SERVICE_NAME"), serviceID),
"deployment.environment": otelFirst(os.Getenv("HWLAB_ENVIRONMENT"), os.Getenv("HWLAB_RUNTIME_LANE"), "unknown"),
"hwlab.lane": otelFirst(os.Getenv("HWLAB_RUNTIME_LANE"), os.Getenv("HWLAB_GITOPS_PROFILE"), "unknown"),
"k8s.namespace.name": otelFirst(os.Getenv("POD_NAMESPACE"), os.Getenv("HWLAB_NAMESPACE"), "unknown"),
"git.commit": otelFirst(os.Getenv("HWLAB_COMMIT_ID"), os.Getenv("HWLAB_GITOPS_SOURCE_COMMIT"), os.Getenv("HWLAB_REVISION"), "unknown"),
}
}
func otelAttributes(values map[string]any) []map[string]any {
attrs := make([]map[string]any, 0, len(values))
for key, value := range values {
if key == "" || value == nil {
continue
}
attrs = append(attrs, map[string]any{"key": key, "value": otelAnyValue(value)})
}
return attrs
}
func otelAnyValue(value any) map[string]any {
switch typed := value.(type) {
case bool:
return map[string]any{"boolValue": typed}
case int:
return map[string]any{"intValue": strconv.FormatInt(int64(typed), 10)}
case int64:
return map[string]any{"intValue": strconv.FormatInt(typed, 10)}
case float64:
return map[string]any{"doubleValue": typed}
case string:
return map[string]any{"stringValue": typed}
default:
payload, err := json.Marshal(typed)
if err != nil {
return map[string]any{"stringValue": "{}"}
}
return map[string]any{"stringValue": string(payload)}
}
}
func otelFirst(values ...string) string {
for _, value := range values {
if text := strings.TrimSpace(value); text != "" {
return text
}
}
return ""
}
internal/workbenchruntime/service.go
+148 -106
View File
@@ -121,7 +121,7 @@ func Run(ctx context.Context) error {
db.SetMaxIdleConns(config.PoolMax)
}
server := NewServer(config, db)
httpServer := &http.Server{Addr: config.Addr, Handler: server.mux, ReadHeaderTimeout: 5 * time.Second}
httpServer := &http.Server{Addr: config.Addr, Handler: server.otelHTTPHandler(server.mux), ReadHeaderTimeout: 5 * time.Second}
errCh := make(chan error, 1)
go func() {
log.Printf("%s listening addr=%s", serviceID, config.Addr)
@@ -440,6 +440,8 @@ func (s *Server) queryFacts(ctx context.Context, query factQuery) (map[string][]
func (s *Server) queryFactRows(ctx context.Context, table string, jsonColumn string, query factQuery, fields []fieldColumn) ([]any, error) {
clauses, args := factClauses(query, fields)
limit := boundedLimit(query.Limit, 0, 1000)
family := factFamilyForTable(table)
stage := "workbench_runtime.db." + family
if table == "workbench_trace_events" {
if query.AfterProjectedSeq > 0 {
args = append(args, query.AfterProjectedSeq)
@@ -450,92 +452,115 @@ func (s *Server) queryFactRows(ctx context.Context, table string, jsonColumn str
args = append(args, limit)
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
}
return s.queryJSONColumn(ctx, sqlText, args)
return s.queryJSONColumn(ctx, stage, table, sqlText, args)
}
orderDirection := "ASC"
if factOrder(query, factFamilyForTable(table)) == "updated_desc" {
if factOrder(query, family) == "updated_desc" {
orderDirection = "DESC"
}
if table == "workbench_sessions" && query.SessionProjection == "summary" {
stage = "workbench_runtime.db.sessions_summary"
sqlText := fmt.Sprintf("SELECT %s FROM workbench_sessions%s ORDER BY updated_at %s", sessionSummarySelectClause(), whereSQL(clauses), orderDirection)
if limit > 0 {
args = append(args, limit)
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
}
return s.querySessionSummaries(ctx, sqlText, args)
return s.querySessionSummaries(ctx, stage, sqlText, args)
}
sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY updated_at %s", jsonColumn, table, whereSQL(clauses), orderDirection)
if limit > 0 {
args = append(args, limit)
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
}
return s.queryJSONColumn(ctx, sqlText, args)
return s.queryJSONColumn(ctx, stage, table, sqlText, args)
}
func (s *Server) queryJSONColumn(ctx context.Context, sqlText string, args []any) ([]any, error) {
rows, err := s.db.QueryContext(ctx, sqlText, args...)
if err != nil {
return nil, err
}
defer rows.Close()
func (s *Server) queryJSONColumn(ctx context.Context, stage string, table string, sqlText string, args []any) ([]any, error) {
result := []any{}
for rows.Next() {
var raw sql.NullString
if err := rows.Scan(&raw); err != nil {
return nil, err
err := s.withOtelInternalSpan(ctx, stage, map[string]any{
"db.system": "postgresql",
"db.operation.name": "SELECT",
"db.sql.table": table,
"db.query.arg_count": int64(len(args)),
}, func(spanCtx context.Context) error {
rows, err := s.db.QueryContext(spanCtx, sqlText, args...)
if err != nil {
return err
}
var item any
if raw.Valid && strings.TrimSpace(raw.String) != "" && json.Unmarshal([]byte(raw.String), &item) == nil && item != nil {
defer rows.Close()
for rows.Next() {
var raw sql.NullString
if err := rows.Scan(&raw); err != nil {
return err
}
var item any
if raw.Valid && strings.TrimSpace(raw.String) != "" && json.Unmarshal([]byte(raw.String), &item) == nil && item != nil {
result = append(result, item)
}
}
return rows.Err()
})
if err != nil {
return nil, wrapQueryStage(stage, err)
}
return result, nil
}
func (s *Server) querySessionSummaries(ctx context.Context, stage string, sqlText string, args []any) ([]any, error) {
result := []any{}
err := s.withOtelInternalSpan(ctx, stage, map[string]any{
"db.system": "postgresql",
"db.operation.name": "SELECT",
"db.sql.table": "workbench_sessions",
"db.query.arg_count": int64(len(args)),
}, func(spanCtx context.Context) error {
rows, err := s.db.QueryContext(spanCtx, sqlText, args...)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID, sourceEventID, providerProfile sql.NullString
var projectedSeq, sourceSeq sql.NullInt64
var terminal, sealed sql.NullBool
var createdAt, updatedAt sql.NullTime
if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &sourceSeq, &sourceEventID, &terminal, &sealed, &createdAt, &updatedAt, &providerProfile); err != nil {
return err
}
if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" {
continue
}
item := map[string]any{
"id": sessionID.String,
"sessionId": sessionID.String,
"ownerUserId": nullableString(ownerUserID),
"projectId": nullableString(projectID),
"conversationId": nullableString(conversationID),
"threadId": nullableString(threadID),
"status": first(nullableStringValue(status), "unknown"),
"lastTraceId": nullableString(lastTraceID),
"projectedSeq": nullableInt(projectedSeq),
"sourceSeq": nullableInt(sourceSeq),
"sourceEventId": nullableString(sourceEventID),
"terminal": nullableBool(terminal),
"sealed": nullableBool(sealed),
"createdAt": nullableTime(createdAt),
"updatedAt": first(nullableTime(updatedAt), nullableTime(createdAt)),
"valuesPrinted": false,
"valuesRedacted": true,
}
if providerProfile.Valid && strings.TrimSpace(providerProfile.String) != "" {
item["providerProfile"] = providerProfile.String
item["sessionJson"] = map[string]any{"providerProfile": providerProfile.String, "valuesRedacted": true, "secretMaterialStored": false}
}
result = append(result, item)
}
}
return result, rows.Err()
}
func (s *Server) querySessionSummaries(ctx context.Context, sqlText string, args []any) ([]any, error) {
rows, err := s.db.QueryContext(ctx, sqlText, args...)
return rows.Err()
})
if err != nil {
return nil, err
return nil, wrapQueryStage(stage, err)
}
defer rows.Close()
result := []any{}
for rows.Next() {
var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID, sourceEventID, providerProfile sql.NullString
var projectedSeq, sourceSeq sql.NullInt64
var terminal, sealed sql.NullBool
var createdAt, updatedAt sql.NullTime
if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &sourceSeq, &sourceEventID, &terminal, &sealed, &createdAt, &updatedAt, &providerProfile); err != nil {
return nil, err
}
if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" {
continue
}
item := map[string]any{
"id": sessionID.String,
"sessionId": sessionID.String,
"ownerUserId": nullableString(ownerUserID),
"projectId": nullableString(projectID),
"conversationId": nullableString(conversationID),
"threadId": nullableString(threadID),
"status": first(nullableStringValue(status), "unknown"),
"lastTraceId": nullableString(lastTraceID),
"projectedSeq": nullableInt(projectedSeq),
"sourceSeq": nullableInt(sourceSeq),
"sourceEventId": nullableString(sourceEventID),
"terminal": nullableBool(terminal),
"sealed": nullableBool(sealed),
"createdAt": nullableTime(createdAt),
"updatedAt": first(nullableTime(updatedAt), nullableTime(createdAt)),
"valuesPrinted": false,
"valuesRedacted": true,
}
if providerProfile.Valid && strings.TrimSpace(providerProfile.String) != "" {
item["providerProfile"] = providerProfile.String
item["sessionJson"] = map[string]any{"providerProfile": providerProfile.String, "valuesRedacted": true, "secretMaterialStored": false}
}
result = append(result, item)
}
return result, rows.Err()
return result, nil
}
func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQuery) ([]any, error) {
@@ -550,7 +575,7 @@ func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQue
args = append(args, limit)
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
}
return s.queryJSONColumn(ctx, sqlText, args)
return s.queryJSONColumn(ctx, "workbench_runtime.db.agent_trace_events", "agent_trace_events", sqlText, args)
}
func (s *Server) readProjectionOutbox(ctx context.Context, query outboxQuery) ([]any, error) {
@@ -561,43 +586,55 @@ func (s *Server) readProjectionOutbox(ctx context.Context, query outboxQuery) ([
addTextClause(&clauses, &args, "session_id", query.SessionID)
args = append(args, limit)
sqlText := fmt.Sprintf("SELECT outbox_seq, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE %s ORDER BY outbox_seq ASC LIMIT $%d", strings.Join(clauses, " AND "), len(args))
rows, err := s.db.QueryContext(ctx, sqlText, args...)
if err != nil {
return nil, err
}
defer rows.Close()
stage := "workbench_runtime.db.projection_outbox"
result := []any{}
for rows.Next() {
var outboxSeq, projectedSeq, sourceSeq sql.NullInt64
var traceID, sessionID, turnID, messageID, sourceEventID, commitType sql.NullString
var terminal, sealed sql.NullBool
var raw sql.NullString
var createdAt sql.NullTime
if err := rows.Scan(&outboxSeq, &traceID, &sessionID, &turnID, &messageID, &projectedSeq, &sourceSeq, &sourceEventID, &commitType, &terminal, &sealed, &raw, &createdAt); err != nil {
return nil, err
err := s.withOtelInternalSpan(ctx, stage, map[string]any{
"db.system": "postgresql",
"db.operation.name": "SELECT",
"db.sql.table": "workbench_projection_outbox",
"db.query.arg_count": int64(len(args)),
}, func(spanCtx context.Context) error {
rows, err := s.db.QueryContext(spanCtx, sqlText, args...)
if err != nil {
return err
}
var payload any
if raw.Valid && strings.TrimSpace(raw.String) != "" {
_ = json.Unmarshal([]byte(raw.String), &payload)
defer rows.Close()
for rows.Next() {
var outboxSeq, projectedSeq, sourceSeq sql.NullInt64
var traceID, sessionID, turnID, messageID, sourceEventID, commitType sql.NullString
var terminal, sealed sql.NullBool
var raw sql.NullString
var createdAt sql.NullTime
if err := rows.Scan(&outboxSeq, &traceID, &sessionID, &turnID, &messageID, &projectedSeq, &sourceSeq, &sourceEventID, &commitType, &terminal, &sealed, &raw, &createdAt); err != nil {
return err
}
var payload any
if raw.Valid && strings.TrimSpace(raw.String) != "" {
_ = json.Unmarshal([]byte(raw.String), &payload)
}
result = append(result, map[string]any{
"outboxSeq": nullableInt(outboxSeq),
"traceId": nullableString(traceID),
"sessionId": nullableString(sessionID),
"turnId": nullableString(turnID),
"messageId": nullableString(messageID),
"projectedSeq": nullableInt(projectedSeq),
"sourceSeq": nullableInt(sourceSeq),
"sourceEventId": nullableString(sourceEventID),
"commitType": nullableString(commitType),
"terminal": nullableBool(terminal),
"sealed": nullableBool(sealed),
"payload": payload,
"createdAt": nullableTime(createdAt),
"valuesPrinted": false,
})
}
result = append(result, map[string]any{
"outboxSeq": nullableInt(outboxSeq),
"traceId": nullableString(traceID),
"sessionId": nullableString(sessionID),
"turnId": nullableString(turnID),
"messageId": nullableString(messageID),
"projectedSeq": nullableInt(projectedSeq),
"sourceSeq": nullableInt(sourceSeq),
"sourceEventId": nullableString(sourceEventID),
"commitType": nullableString(commitType),
"terminal": nullableBool(terminal),
"sealed": nullableBool(sealed),
"payload": payload,
"createdAt": nullableTime(createdAt),
"valuesPrinted": false,
})
return rows.Err()
})
if err != nil {
return nil, wrapQueryStage(stage, err)
}
return result, rows.Err()
return result, nil
}
func factClauses(query factQuery, fields []fieldColumn) ([]string, []any) {
@@ -1198,16 +1235,21 @@ func decodeJSON(w http.ResponseWriter, r *http.Request, target any) bool {
func writeQueryError(w http.ResponseWriter, err error) {
log.Printf("%s query failed: %v", serviceID, err)
errorPayload := map[string]any{
"code": "workbench_runtime_query_failed",
"message": "workbench runtime query failed",
"retryable": true,
"transient": true,
"causeCode": sqlErrorCode(err),
"valuesRedacted": true,
}
if stage := queryErrorStage(err); stage != "" {
errorPayload["stage"] = stage
errorPayload["causeStage"] = stage
}
writeJSON(w, http.StatusServiceUnavailable, map[string]any{
"ok": false,
"error": map[string]any{
"code": "workbench_runtime_query_failed",
"message": "workbench runtime query failed",
"retryable": true,
"transient": true,
"causeCode": sqlErrorCode(err),
"valuesRedacted": true,
},
"ok": false,
"error": errorPayload,
"valuesRedacted": true,
})
}