feat: add failed and slow queries to query stats

query/execution_context.go

@@ -3,6 +3,7 @@ package query
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 )
 
 // ExecutionContext contains state that the query is currently executing with.
@@ -27,6 +28,11 @@ type ExecutionContext struct {
 	mu   sync.RWMutex
 	done chan struct{}
 	err  error
+
+	// failed is set (via send/Send) whenever a Result with a non-nil Err is
+	// passed to the execution context for sending (even if the query is aborted
+	// before it can be delivered). Read by the Executor to count queriesFailed.
+	failed int32
 }
 
 func (ctx *ExecutionContext) watch() {
@@ -88,9 +94,18 @@ func (ctx *ExecutionContext) Value(key interface{}) interface{} {
 	return ctx.Context.Value(key)
 }
 
+// Failed reports whether any Result carrying a non-nil Err has been emitted for
+// this query.
+func (ctx *ExecutionContext) Failed() bool {
+	return atomic.LoadInt32(&ctx.failed) != 0
+}
+
 // send sends a Result to the Results channel and will exit if the query has
 // been aborted.
 func (ctx *ExecutionContext) send(result *Result) error {
+	if result.Err != nil {
+		atomic.StoreInt32(&ctx.failed, 1)
+	}
 	select {
 	case <-ctx.AbortCh:
 		return ErrQueryAborted
@@ -103,6 +118,9 @@ func (ctx *ExecutionContext) send(result *Result) error {
 // been interrupted or aborted.
 func (ctx *ExecutionContext) Send(result *Result) error {
 	result.StatementID = ctx.statementID
+	if result.Err != nil {
+		atomic.StoreInt32(&ctx.failed, 1)
+	}
 	select {
 	case <-ctx.Done():
 		return ctx.Err()

query/executor.go

@@ -46,6 +46,8 @@ const (
 	statQueriesActive          = "queriesActive"   // Number of queries currently being executed.
 	statQueriesExecuted        = "queriesExecuted" // Number of queries that have been executed (started).
 	statQueriesFinished        = "queriesFinished" // Number of queries that have finished.
+	statQueriesFailed          = "queriesFailed"   // Number of queries that have failed.
+	statQueriesSlow            = "queriesSlow"     // Snapshot of in-flight queries running longer than log-queries-after.
 	statQueryExecutionDuration = "queryDurationNs" // Total (wall) time spent executing queries.
 	statRecoveredPanics        = "recoveredPanics" // Number of panics recovered by Query Executor.
 
@@ -264,6 +266,7 @@ type Statistics struct {
 	ActiveQueries          int64
 	ExecutedQueries        int64
 	FinishedQueries        int64
+	FailedQueries          int64
 	QueryExecutionDuration int64
 	RecoveredPanics        int64
 }
@@ -277,6 +280,8 @@ func (e *Executor) Statistics(tags map[string]string) []models.Statistic {
 			statQueriesActive:          atomic.LoadInt64(&e.stats.ActiveQueries),
 			statQueriesExecuted:        atomic.LoadInt64(&e.stats.ExecutedQueries),
 			statQueriesFinished:        atomic.LoadInt64(&e.stats.FinishedQueries),
+			statQueriesFailed:          atomic.LoadInt64(&e.stats.FailedQueries),
+			statQueriesSlow:            e.TaskManager.SlowQueryCount(),
 			statQueryExecutionDuration: atomic.LoadInt64(&e.stats.QueryExecutionDuration),
 			statRecoveredPanics:        atomic.LoadInt64(&e.stats.RecoveredPanics),
 		},
@@ -304,18 +309,38 @@ func (e *Executor) ExecuteQuery(query *influxql.Query, opt ExecutionOptions, clo
 
 func (e *Executor) executeQuery(query *influxql.Query, opt ExecutionOptions, closing <-chan struct{}, results chan *Result) {
 	defer close(results)
-	defer e.recover(query, results)
 
 	atomic.AddInt64(&e.stats.ActiveQueries, 1)
 	atomic.AddInt64(&e.stats.ExecutedQueries, 1)
+
+	// A query is counted as failed at most once, here in the single cleanup
+	// closure. It is deferred BEFORE e.recover below so it runs AFTER recover
+	// during unwinding and can observe a recovered panic via panicked. Three
+	// signals feed it: failed, for terminal errors executeQuery emits directly
+	// on the results channel (before ctx exists, or bypassing it); panicked, set
+	// by recover when it recovers a panic; and ctx.Failed(), set whenever any
+	// Result with a non-nil Err is sent through the ExecutionContext — including
+	// errors a StatementExecutor sends itself while returning nil, and
+	// ErrNotExecuted emitted for interrupted queries.
+	failed := false
+	panicked := false
+	var ctx *ExecutionContext
 	defer func(start time.Time) {
 		atomic.AddInt64(&e.stats.ActiveQueries, -1)
 		atomic.AddInt64(&e.stats.FinishedQueries, 1)
+		if failed || panicked || (ctx != nil && ctx.Failed()) {
+			atomic.AddInt64(&e.stats.FailedQueries, 1)
+		}
 		atomic.AddInt64(&e.stats.QueryExecutionDuration, time.Since(start).Nanoseconds())
 	}(time.Now())
 
-	ctx, detach, err := e.TaskManager.AttachQuery(query, opt, closing)
+	defer e.recover(query, results, &panicked)
+
+	var detach func()
+	var err error
+	ctx, detach, err = e.TaskManager.AttachQuery(query, opt, closing)
 	if err != nil {
+		failed = true
 		select {
 		case results <- &Result{Err: err}:
 		case <-opt.AbortCh:
@@ -362,8 +387,11 @@ LOOP:
 						case "_tags":
 							command = "SHOW TAG VALUES"
 						}
-						results <- &Result{
-							Err: fmt.Errorf("unable to use system source '%s': use %s instead", s.Name, command),
+						if err := ctx.send(&Result{
+							StatementID: i,
+							Err:         fmt.Errorf("unable to use system source '%s': use %s instead", s.Name, command),
+						}); err == ErrQueryAborted {
+							return
 						}
 						break LOOP
 					}
@@ -375,7 +403,9 @@ LOOP:
 		// This can occur on meta read statements which convert to SELECT statements.
 		newStmt, err := RewriteStatement(stmt)
 		if err != nil {
-			results <- &Result{Err: err}
+			if err := ctx.send(&Result{StatementID: i, Err: err}); err == ErrQueryAborted {
+				return
+			}
 			break
 		}
 		stmt = newStmt
@@ -427,6 +457,11 @@ LOOP:
 		}
 
 		if interrupted {
+			// A query killed or timed out between statements is a failed query,
+			// consistent with one interrupted during a statement (err != nil
+			// above). The ErrNotExecuted fan-out below sets ctx.Failed() when
+			// statements remain, but not when this was the last statement.
+			failed = true
 			break
 		}
 	}
@@ -453,8 +488,13 @@ func init() {
 	}
 }
 
-func (e *Executor) recover(query *influxql.Query, results chan *Result) {
+// recover recovers a panic from executing a query. When it recovers a panic it
+// sets *panicked so the caller's cleanup counts the query as failed exactly once
+// (a panicked query is a failed query); recover does not touch FailedQueries
+// itself to avoid double-counting a query that also emitted an error Result.
+func (e *Executor) recover(query *influxql.Query, results chan *Result, panicked *bool) {
 	if err := recover(); err != nil {
+		*panicked = true
 		atomic.AddInt64(&e.stats.RecoveredPanics, 1) // Capture the panic in _internal stats.
 		e.Logger.Error(fmt.Sprintf("%s [panic:%s] %s", query.String(), err, debug.Stack()))
 		results <- &Result{

query/executor_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -725,8 +726,212 @@ func TestQueryExecutor_InvalidSource(t *testing.T) {
 	}
 }
 
+// TestQueryExecutor_InvalidSource_StatementID verifies that a system-source
+// error raised for a later statement in a multi-statement query is attributed
+// to that statement's index, not statement 0.
+func TestQueryExecutor_InvalidSource_StatementID(t *testing.T) {
+	e := NewQueryExecutor()
+	e.StatementExecutor = &StatementExecutor{
+		ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+			return nil // the first (valid) statement succeeds
+		},
+	}
+
+	q, err := influxql.ParseQuery(`SELECT value FROM cpu; SELECT fieldKey FROM _fieldKeys`)
+	require.NoError(t, err)
+
+	var errResult *query.Result
+	for r := range e.ExecuteQuery(q, query.ExecutionOptions{}, nil) {
+		if r.Err != nil {
+			errResult = r
+		}
+	}
+	require.NotNil(t, errResult)
+	require.EqualError(t, errResult.Err, `unable to use system source '_fieldKeys': use SHOW FIELD KEYS instead`)
+	require.Equal(t, 1, errResult.StatementID)
+}
+
 func discardOutput(results <-chan *query.Result) {
 	for range results {
 		// Read all results and discard.
 	}
 }
+
+// queryExecutorStat returns the named value from the executor's queryExecutor
+// statistic. It fails the test if the stat is missing or not an int64.
+func queryExecutorStat(t *testing.T, e *query.Executor, key string) int64 {
+	t.Helper()
+	stats := e.Statistics(nil)
+	require.Len(t, stats, 1)
+	v, ok := stats[0].Values[key].(int64)
+	require.Truef(t, ok, "stat %q missing or not int64: %v", key, stats[0].Values[key])
+	return v
+}
+
+func TestQueryExecutor_Statistics_QueriesFailed(t *testing.T) {
+	t.Run("success is not counted", func(t *testing.T) {
+		e := NewQueryExecutor()
+		e.StatementExecutor = &StatementExecutor{
+			ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+				return nil
+			},
+		}
+		q, err := influxql.ParseQuery(goodStatement)
+		require.NoError(t, err)
+		discardOutput(e.ExecuteQuery(q, query.ExecutionOptions{}, nil))
+		require.Equal(t, int64(0), queryExecutorStat(t, e, "queriesFailed"))
+	})
+
+	t.Run("single failure counts once", func(t *testing.T) {
+		e := NewQueryExecutor()
+		e.StatementExecutor = &StatementExecutor{
+			ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+				return errUnexpected
+			},
+		}
+		q, err := influxql.ParseQuery(goodStatement)
+		require.NoError(t, err)
+		discardOutput(e.ExecuteQuery(q, query.ExecutionOptions{}, nil))
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "queriesFailed"))
+	})
+
+	t.Run("multi-statement failure counts once", func(t *testing.T) {
+		e := NewQueryExecutor()
+		var calls int
+		e.StatementExecutor = &StatementExecutor{
+			ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+				calls++
+				return errUnexpected
+			},
+		}
+		queryStr := strings.Join([]string{goodStatement, goodStatement, goodStatement}, ";")
+		q, err := influxql.ParseQuery(queryStr)
+		require.NoError(t, err)
+		discardOutput(e.ExecuteQuery(q, query.ExecutionOptions{}, nil))
+		// The first statement fails; the remaining statements emit ErrNotExecuted
+		// without being executed. The query counts as failed exactly once.
+		require.Equal(t, 1, calls)
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "queriesFailed"))
+	})
+
+	t.Run("executor-streamed error counts", func(t *testing.T) {
+		e := NewQueryExecutor()
+		e.StatementExecutor = &StatementExecutor{
+			ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+				// Mirrors real statement executors that report an error by
+				// sending a Result and returning nil (see, e.g., the meta/SHOW
+				// paths in coordinator/statement_executor.go).
+				return ctx.Send(&query.Result{Err: errUnexpected})
+			},
+		}
+		q, err := influxql.ParseQuery(goodStatement)
+		require.NoError(t, err)
+		discardOutput(e.ExecuteQuery(q, query.ExecutionOptions{}, nil))
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "queriesFailed"))
+	})
+
+	t.Run("panic counts as failure", func(t *testing.T) {
+		e := NewQueryExecutor()
+		e.StatementExecutor = &StatementExecutor{
+			ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+				panic("test error")
+			},
+		}
+		q, err := influxql.ParseQuery(goodStatement)
+		require.NoError(t, err)
+		discardOutput(e.ExecuteQuery(q, query.ExecutionOptions{}, nil))
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "queriesFailed"))
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "recoveredPanics"))
+	})
+
+	t.Run("error then panic counts once", func(t *testing.T) {
+		e := NewQueryExecutor()
+		var call int
+		e.StatementExecutor = &StatementExecutor{
+			ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+				call++
+				if call == 1 {
+					// First statement reports an error by sending a Result and
+					// returning nil, which sets ctx.Failed().
+					return ctx.Send(&query.Result{Err: errUnexpected})
+				}
+				// A later statement panics after the failure was already
+				// recorded; the query must still be counted as failed only once.
+				panic("test error")
+			},
+		}
+		queryStr := strings.Join([]string{goodStatement, goodStatement}, ";")
+		q, err := influxql.ParseQuery(queryStr)
+		require.NoError(t, err)
+		discardOutput(e.ExecuteQuery(q, query.ExecutionOptions{}, nil))
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "queriesFailed"))
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "recoveredPanics"))
+	})
+}
+
+func TestTaskManager_SlowQueryCount(t *testing.T) {
+	// newBlockingExecutor returns an executor whose statement execution blocks
+	// until the returned release func is called (idempotent — safe to call more
+	// than once), signalling on started once the query is attached and running.
+	// Callers should defer release so a failed assertion cannot strand the
+	// blocked query goroutine.
+	newBlockingExecutor := func() (e *query.Executor, started <-chan struct{}, release func()) {
+		startedCh := make(chan struct{})
+		releaseCh := make(chan struct{})
+		var once sync.Once
+		e = NewQueryExecutor()
+		e.StatementExecutor = &StatementExecutor{
+			ExecuteStatementFn: func(stmt influxql.Statement, ctx *query.ExecutionContext) error {
+				close(startedCh)
+				<-releaseCh
+				return nil
+			},
+		}
+		return e, startedCh, func() { once.Do(func() { close(releaseCh) }) }
+	}
+
+	t.Run("counts in-flight queries slower than log-queries-after", func(t *testing.T) {
+		e, started, release := newBlockingExecutor()
+		defer e.Close()
+		defer release()
+		e.TaskManager.LogQueriesAfter = 20 * time.Millisecond
+
+		q, err := influxql.ParseQuery(goodStatement)
+		require.NoError(t, err)
+		results := e.ExecuteQuery(q, query.ExecutionOptions{}, nil)
+
+		<-started // query is attached and running
+
+		// Once it has been running longer than the threshold, it is counted.
+		require.Eventually(t, func() bool {
+			return e.TaskManager.SlowQueryCount() == 1
+		}, time.Second, 2*time.Millisecond)
+		require.Equal(t, int64(1), queryExecutorStat(t, e, "queriesSlow"))
+
+		release()
+		discardOutput(results)
+
+		// After the query finishes and detaches, the snapshot returns to zero.
+		require.Equal(t, int64(0), e.TaskManager.SlowQueryCount())
+	})
+
+	t.Run("zero threshold counts nothing", func(t *testing.T) {
+		e, started, release := newBlockingExecutor()
+		defer e.Close()
+		defer release()
+		e.TaskManager.LogQueriesAfter = 0 // slow-query detection disabled
+
+		q, err := influxql.ParseQuery(goodStatement)
+		require.NoError(t, err)
+		results := e.ExecuteQuery(q, query.ExecutionOptions{}, nil)
+
+		<-started // query is attached and running
+
+		// Even with a long-running in-flight query, a zero threshold never counts.
+		require.Equal(t, int64(0), e.TaskManager.SlowQueryCount())
+		require.Equal(t, int64(0), queryExecutorStat(t, e, "queriesSlow"))
+
+		release()
+		discardOutput(results)
+	})
+}