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 carrying a non-nil Err is
+	// emitted for this query, including errors a StatementExecutor sends itself
+	// while returning nil. 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),
 		},
@@ -308,14 +313,30 @@ func (e *Executor) executeQuery(query *influxql.Query, opt ExecutionOptions, clo
 
 	atomic.AddInt64(&e.stats.ActiveQueries, 1)
 	atomic.AddInt64(&e.stats.ExecutedQueries, 1)
+
+	// A query is counted as failed at most once, in the deferred cleanup below.
+	// Two signals feed it: failed, for terminal errors executeQuery emits
+	// directly on the results channel (before ctx exists, or bypassing it); 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.
+	// Panics are counted separately in recover(), which runs after this closure.
+	failed := false
+	var ctx *ExecutionContext
 	defer func(start time.Time) {
 		atomic.AddInt64(&e.stats.ActiveQueries, -1)
 		atomic.AddInt64(&e.stats.FinishedQueries, 1)
+		if failed || (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)
+	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,6 +383,7 @@ LOOP:
 						case "_tags":
 							command = "SHOW TAG VALUES"
 						}
+						failed = true
 						results <- &Result{
 							Err: fmt.Errorf("unable to use system source '%s': use %s instead", s.Name, command),
 						}
@@ -375,6 +397,7 @@ LOOP:
 		// This can occur on meta read statements which convert to SELECT statements.
 		newStmt, err := RewriteStatement(stmt)
 		if err != nil {
+			failed = true
 			results <- &Result{Err: err}
 			break
 		}
@@ -427,6 +450,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
 		}
 	}
@@ -456,6 +484,7 @@ func init() {
 func (e *Executor) recover(query *influxql.Query, results chan *Result) {
 	if err := recover(); err != nil {
 		atomic.AddInt64(&e.stats.RecoveredPanics, 1) // Capture the panic in _internal stats.
+		atomic.AddInt64(&e.stats.FailedQueries, 1)   // A panicked query is a failed query.
 		e.Logger.Error(fmt.Sprintf("%s [panic:%s] %s", query.String(), err, debug.Stack()))
 		results <- &Result{
 			StatementID: -1,

query/executor_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -730,3 +731,158 @@ func discardOutput(results <-chan *query.Result) {
 		// 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"))
+	})
+}
+
+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)
+	})
+}

query/task_manager.go

@@ -158,6 +158,29 @@ func (t *TaskManager) executeShowQueriesStatement(q *influxql.ShowQueriesStateme
 	}}, nil
 }
 
+// SlowQueryCount returns a snapshot of the number of currently-running queries
+// that have been executing for longer than LogQueriesAfter. It returns 0 when
+// LogQueriesAfter is unset (0), matching the slow-query logging semantics.
+func (t *TaskManager) SlowQueryCount() int64 {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	if t.LogQueriesAfter == 0 {
+		return 0
+	}
+
+	now := time.Now()
+	var n int64
+	for _, qi := range t.queries {
+		// >= matches the slow-query logging monitor, whose timer fires at
+		// exactly LogQueriesAfter.
+		if now.Sub(qi.startTime) >= t.LogQueriesAfter {
+			n++
+		}
+	}
+	return n
+}
+
 func prettyTime(d time.Duration) time.Duration {
 	switch {
 	case d >= time.Second: