refactor query outcome metrics (#779)

Author: bill-ph

README.md

@@ -64,8 +64,8 @@ Duckgres exposes Prometheus metrics on `:9090/metrics`. The metrics port is curr
 | Metric | Type | Description |
 |--------|------|-------------|
 | `duckgres_connections_open` | Gauge | Number of currently open client connections |
-| `duckgres_query_duration_seconds` | Histogram | Query execution duration (includes `_count`, `_sum`, `_bucket`) |
-| `duckgres_query_errors_total` | Counter | Total number of failed queries |
+| `duckgres_query_total{org,outcome}` | Counter | Total number of non-empty query attempts by terminal outcome (`success`, `error`, `canceled`) |
+| `duckgres_query_duration_seconds{org}` | Histogram | Simple/extended query execution latency (includes `_count`, `_sum`, `_bucket`); use `duckgres_query_total` for attempt totals |
 | `duckgres_auth_failures_total` | Counter | Total number of authentication failures |
 | `duckgres_rate_limit_rejects_total` | Counter | Total number of connections rejected due to rate limiting |
 | `duckgres_rate_limited_ips` | Gauge | Number of currently rate-limited IP addresses |

grafana/dashboards/duckgres.json

@@ -229,7 +229,7 @@
       },
       "targets": [
         {
-          "expr": "rate(duckgres_query_duration_seconds_count[5m])",
+          "expr": "sum(rate(duckgres_query_total[5m]))",
           "legendFormat": "Queries/s",
           "refId": "A"
         }
@@ -261,7 +261,7 @@
       },
       "targets": [
         {
-          "expr": "rate(duckgres_query_errors_total[5m])",
+          "expr": "sum(rate(duckgres_query_total{outcome=\"error\"}[5m]))",
           "legendFormat": "Errors/s",
           "refId": "A"
         }

metrics-compose.yml

@@ -9,8 +9,8 @@
 #
 # Available metrics:
 #   - duckgres_connections_open: current number of open connections
-#   - duckgres_query_duration_seconds: histogram of query latencies
-#   - duckgres_query_errors_total: counter of failed queries
+#   - duckgres_query_total: counter of query attempts by terminal outcome
+#   - duckgres_query_duration_seconds: histogram of query execution latencies
 #   - duckgres_perf_query_duration_seconds: per-query perf harness latency histograms
 #
 # Note: Uses port 9091 for Prometheus UI to avoid conflict with duckgres metrics on 9090.

scripts/test_metrics.sh

@@ -25,7 +25,9 @@ PGPASSWORD=postgres psql "host=127.0.0.1 port=$PORT user=postgres sslmode=requir
 PGPASSWORD=postgres psql "host=127.0.0.1 port=$PORT user=postgres sslmode=require" -c "SELECT 2" >/dev/null
 
 # Check metrics
-QUERY_COUNT=$(curl -s "http://localhost:$METRICS_PORT/metrics" | grep 'duckgres_query_duration_seconds_count' | awk '{print $2}')
+METRICS=$(curl -s "http://localhost:$METRICS_PORT/metrics")
+QUERY_COUNT=$(echo "$METRICS" | awk '/^duckgres_query_duration_seconds_count/ {sum += $2} END {print sum}')
+QUERY_SUCCESS_TOTAL=$(echo "$METRICS" | awk '/^duckgres_query_total\{.*outcome="success"/ {sum += $2} END {print sum}')
 
 if [ -z "$QUERY_COUNT" ]; then
     echo "FAIL: could not find 'duckgres_query_duration_seconds_count' metric in metrics output"
@@ -42,3 +44,19 @@ else
     echo "FAIL: query count is $QUERY_COUNT (expected >= 2)"
     exit 1
 fi
+
+if [ -z "$QUERY_SUCCESS_TOTAL" ]; then
+    echo "FAIL: could not find 'duckgres_query_total{outcome=\"success\"}' metric in metrics output"
+    exit 1
+fi
+
+if ! [[ "$QUERY_SUCCESS_TOTAL" =~ ^[0-9]+$ ]]; then
+    echo "FAIL: query success total '$QUERY_SUCCESS_TOTAL' is not a valid integer metric value"
+    exit 1
+fi
+if [ "$QUERY_SUCCESS_TOTAL" -ge 2 ]; then
+    echo "PASS: query success total is $QUERY_SUCCESS_TOTAL (expected >= 2)"
+else
+    echo "FAIL: query success total is $QUERY_SUCCESS_TOTAL (expected >= 2)"
+    exit 1
+fi

server/conn.go

@@ -167,6 +167,8 @@ type clientConn struct {
 	txStatus               byte                     // current transaction status ('I', 'T', or 'E')
 	ignoreTillSync         bool                     // discard extended-query messages until Sync after an error; see runExtendedQueryMessage
 	errorResponsesSent     uint64                   // ErrorResponses sent via sendError; observed by runExtendedQueryMessage
+	lastErrorCode          string                   // most recent SQLSTATE sent via sendError; observed by query metrics
+	activeQueryMetrics     *queryMetricsScope       // active query attempt metrics scope for non-ErrorResponse failures
 	passthrough            bool                     // true for passthrough users (skip transpiler + pg_catalog)
 	cursors                map[string]*cursorState  // server-side cursor emulation
 	catalogUseRewrite      bool                     // true when bare `USE ducklake`/`USE iceberg` should expand to the reliable two-part target
@@ -873,10 +875,10 @@ func (c *clientConn) serve() error {
 	c.sendInitialParams()
 
 	// Send ready for query
-	if err := wire.WriteReadyForQuery(c.writer, c.txStatus); err != nil {
+	if err := c.writeReadyForQuery(c.txStatus); err != nil {
 		return err
 	}
-	if err := c.writer.Flush(); err != nil {
+	if err := c.flushWriter(); err != nil {
 		return fmt.Errorf("failed to flush writer: %w", err)
 	}
 
@@ -978,7 +980,7 @@ func (c *clientConn) handleStartup() error {
 	if err := wire.WriteAuthCleartextPassword(c.writer); err != nil {
 		return err
 	}
-	if err := c.writer.Flush(); err != nil {
+	if err := c.flushWriter(); err != nil {
 		return fmt.Errorf("failed to flush writer: %w", err)
 	}
 
@@ -1096,10 +1098,10 @@ func (c *clientConn) messageLoop() error {
 			// Extended query protocol - Sync: ends any skip-until-Sync error
 			// recovery, then reports readiness.
 			c.ignoreTillSync = false
-			if err := wire.WriteReadyForQuery(c.writer, c.txStatus); err != nil {
+			if err := c.writeReadyForQuery(c.txStatus); err != nil {
 				return err
 			}
-			_ = c.writer.Flush()
+			_ = c.flushWriter()
 
 		case wire.MsgClose:
 			// Extended query protocol - Close
@@ -1109,7 +1111,7 @@ func (c *clientConn) messageLoop() error {
 			// Discarded during skip-until-Sync error recovery, like real
 			// PostgreSQL (the ErrorResponse was already flushed by sendError).
 			if !c.ignoreTillSync {
-				_ = c.writer.Flush()
+				_ = c.flushWriter()
 			}
 
 		case wire.MsgTerminate:
@@ -1121,16 +1123,16 @@ func (c *clientConn) messageLoop() error {
 	}
 }
 
-func (c *clientConn) handleQuery(body []byte) error {
+func (c *clientConn) handleQuery(body []byte) (retErr error) {
 	query := string(bytes.TrimRight(body, "\x00"))
 	query = strings.TrimSpace(query)
 
 	// Treat empty queries or queries with just semicolons as empty
 	// PostgreSQL returns EmptyQueryResponse for queries like "" or ";" or ";;;"
 	if query == "" || isEmptyQuery(query) {
 		_ = wire.WriteEmptyQueryResponse(c.writer)
-		_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-		_ = c.writer.Flush()
+		_ = c.writeReadyForQuery(c.txStatus)
+		_ = c.flushWriter()
 		return nil
 	}
 
@@ -1146,7 +1148,13 @@ func (c *clientConn) handleQuery(body []byte) error {
 	}()
 
 	start := time.Now()
-	defer func() { queryDurationHistogram.WithLabelValues(c.orgID).Observe(time.Since(start).Seconds()) }()
+	queryMetrics := c.beginQueryMetrics(start)
+	defer func() {
+		if retErr != nil {
+			queryMetrics.markError(retErr)
+		}
+		c.finishQueryMetrics(queryMetrics)
+	}()
 
 	ctx, span := observe.Tracer().Start(c.ctx, "duckgres.query",
 		trace.WithAttributes(
@@ -1244,8 +1252,8 @@ func (c *clientConn) handleQuery(body []byte) error {
 	if err != nil {
 		// Transform error - send error to client
 		c.sendError("ERROR", "42601", fmt.Sprintf("syntax error: %v", err))
-		_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-		_ = c.writer.Flush()
+		_ = c.writeReadyForQuery(c.txStatus)
+		_ = c.flushWriter()
 		return nil
 	}
 
@@ -1254,8 +1262,8 @@ func (c *clientConn) handleQuery(body []byte) error {
 		if err := c.validateWithDuckDB(query); err != nil {
 			// Neither PostgreSQL nor DuckDB can parse this query
 			c.sendError("ERROR", "42601", fmt.Sprintf("syntax error: %v", err))
-			_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-			_ = c.writer.Flush()
+			_ = c.writeReadyForQuery(c.txStatus)
+			_ = c.flushWriter()
 			return nil
 		}
 		c.logger().Debug("Fallback to native DuckDB: query not valid PostgreSQL but valid DuckDB.", "query", loggableQuery)
@@ -1264,8 +1272,8 @@ func (c *clientConn) handleQuery(body []byte) error {
 	// Handle transform-detected errors (e.g., unrecognized config parameter)
 	if result.Error != nil {
 		c.sendError("ERROR", transformErrorSQLState(result.Error), result.Error.Error())
-		_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-		_ = c.writer.Flush()
+		_ = c.writeReadyForQuery(c.txStatus)
+		_ = c.flushWriter()
 		return nil
 	}
 
@@ -1278,18 +1286,18 @@ func (c *clientConn) handleQuery(body []byte) error {
 	// Handle ignored SET parameters
 	if result.IsIgnoredSet {
 		c.logger().Debug("Ignoring PostgreSQL-specific SET.", "query", query)
-		_ = wire.WriteCommandComplete(c.writer, "SET")
-		_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-		_ = c.writer.Flush()
+		_ = c.writeCommandComplete("SET")
+		_ = c.writeReadyForQuery(c.txStatus)
+		_ = c.flushWriter()
 		return nil
 	}
 
 	// Handle no-op commands (CREATE INDEX, VACUUM, etc.)
 	if result.IsNoOp {
 		c.logger().Debug("No-op command (DuckLake limitation).", "query", query)
-		_ = wire.WriteCommandComplete(c.writer, result.NoOpTag)
-		_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-		_ = c.writer.Flush()
+		_ = c.writeCommandComplete(result.NoOpTag)
+		_ = c.writeReadyForQuery(c.txStatus)
+		_ = c.flushWriter()
 		return nil
 	}
 
@@ -1323,9 +1331,9 @@ func (c *clientConn) handleQuery(body []byte) error {
 		// while DuckDB throws an error. Match PostgreSQL behavior.
 		if cmdType == "BEGIN" && c.txStatus == txStatusTransaction {
 			c.sendNotice("WARNING", "25001", "there is already a transaction in progress")
-			_ = wire.WriteCommandComplete(c.writer, "BEGIN")
-			_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-			_ = c.writer.Flush()
+			_ = c.writeCommandComplete("BEGIN")
+			_ = c.writeReadyForQuery(c.txStatus)
+			_ = c.flushWriter()
 			return nil
 		}
 
@@ -1377,8 +1385,8 @@ func (c *clientConn) handleQuery(body []byte) error {
 				c.sendError("ERROR", errCode, errMsg)
 				c.setTxError()
 				c.logQuery(start, originalQuery, query, cmdType, 0, 0, errCode, errMsg, "simple")
-				_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-				_ = c.writer.Flush()
+				_ = c.writeReadyForQuery(c.txStatus)
+				_ = c.flushWriter()
 				return nil
 			}
 		}
@@ -1389,10 +1397,10 @@ func (c *clientConn) handleQuery(body []byte) error {
 		}
 		c.updateTxStatus(cmdType)
 		tag := c.buildCommandTag(cmdType, execResult)
-		_ = wire.WriteCommandComplete(c.writer, tag)
+		_ = c.writeCommandComplete(tag)
 		c.logQuery(start, originalQuery, query, cmdType, 0, writtenRows, "", "", "simple")
-		_ = wire.WriteReadyForQuery(c.writer, c.txStatus)
-		_ = c.writer.Flush()
+		_ = c.writeReadyForQuery(c.txStatus)
+		_ = c.flushWriter()
 		return nil
 	}