sql: make UI external active statements and open transactions consistent

Previously, the number of Active SQL Statements could be greater than
the number of Open SQL Transactions on the UI. This inconsistency was
due to filtering out internal Open SQL Transactions but keeping internal
Active SQL Statements. This change fixes this inconsistency by filtering
out both internal Active SQL Statements and internal Open SQL
Transactions.

Release note (sql change): filter out internal statements and
transactions from UI timeseries metrics.

pkg/sql/conn_executor.go

@@ -416,7 +416,8 @@ func makeMetrics(internal bool) Metrics {
 				6*metricsSampleInterval),
 			SQLTxnLatency: metric.NewLatency(getMetricMeta(MetaSQLTxnLatency, internal),
 				6*metricsSampleInterval),
-			SQLTxnsOpen: metric.NewGauge(getMetricMeta(MetaSQLTxnsOpen, internal)),
+			SQLTxnsOpen:         metric.NewGauge(getMetricMeta(MetaSQLTxnsOpen, internal)),
+			SQLActiveStatements: metric.NewGauge(getMetricMeta(MetaSQLActiveQueries, internal)),
 
 			TxnAbortCount:                     metric.NewCounter(getMetricMeta(MetaTxnAbort, internal)),
 			FailureCount:                      metric.NewCounter(getMetricMeta(MetaFailure, internal)),

pkg/sql/conn_executor_exec.go

@@ -316,6 +316,9 @@ func (ex *connExecutor) execStmtInOpenState(
 	}
 
 	ex.addActiveQuery(ast, formatWithPlaceholders(ast, ex.planner.EvalContext()), queryID, ex.state.cancel)
+	if ex.executorType != executorTypeInternal {
+		ex.metrics.EngineMetrics.SQLActiveStatements.Inc(1)
+	}
 
 	// Make sure that we always unregister the query. It also deals with
 	// overwriting res.Error to a more user-friendly message in case of query
@@ -329,6 +332,9 @@ func (ex *connExecutor) execStmtInOpenState(
 			}
 		}
 		ex.removeActiveQuery(queryID, ast)
+		if ex.executorType != executorTypeInternal {
+			ex.metrics.EngineMetrics.SQLActiveStatements.Dec(1)
+		}
 
 		// Detect context cancelation and overwrite whatever error might have been
 		// set on the result before. The idea is that once the query's context is
@@ -1982,7 +1988,9 @@ func (ex *connExecutor) recordTransactionStart(txnID uuid.UUID) {
 		ex.extraTxnState.shouldCollectTxnExecutionStats = txnExecStatsSampleRate > ex.rng.Float64()
 	}
 
-	ex.metrics.EngineMetrics.SQLTxnsOpen.Inc(1)
+	if ex.executorType != executorTypeInternal {
+		ex.metrics.EngineMetrics.SQLTxnsOpen.Inc(1)
+	}
 
 	ex.extraTxnState.shouldExecuteOnTxnFinish = true
 	ex.extraTxnState.txnFinishClosure.txnStartTime = txnStart
@@ -2019,7 +2027,9 @@ func (ex *connExecutor) recordTransactionFinish(
 
 	txnEnd := timeutil.Now()
 	txnTime := txnEnd.Sub(txnStart)
-	ex.metrics.EngineMetrics.SQLTxnsOpen.Dec(1)
+	if ex.executorType != executorTypeInternal {
+		ex.metrics.EngineMetrics.SQLTxnsOpen.Dec(1)
+	}
 	ex.metrics.EngineMetrics.SQLTxnLatency.RecordValue(txnTime.Nanoseconds())
 
 	ex.txnIDCacheWriter.Record(txnidcache.ResolvedTxnID{

pkg/sql/conn_executor_test.go

@@ -15,6 +15,7 @@ import (
 	gosql "database/sql"
 	"database/sql/driver"
 	"fmt"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"net/url"
 	"regexp"
 	"strings"
@@ -1404,6 +1405,111 @@ func TestInjectRetryErrors(t *testing.T) {
 	})
 }
 
+func TestTrackOnlyExternalOpenTransactionsAndActiveStatements(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	params := base.TestServerArgs{}
+	s, sqlDB, _ := serverutils.StartServer(t, params)
+	defer s.Stopper().Stop(ctx)
+
+	selectQuery := "SELECT * FROM t.foo"
+	selectInternalQueryActive := `SELECT count(*) FROM crdb_internal.cluster_queries WHERE query = '` + selectQuery + `'`
+	selectExternalQueryActive := `SELECT count(*) FROM [SHOW CLUSTER STATEMENTS] WHERE query = '` + selectQuery + `'`
+
+	sqlServer := s.SQLServer().(*sql.Server)
+	testDB := sqlutils.MakeSQLRunner(sqlDB)
+	testDB.Exec(t, "CREATE DATABASE t")
+	testDB.Exec(t, "CREATE TABLE t.foo (i INT PRIMARY KEY)")
+	testDB.Exec(t, "INSERT INTO t.foo VALUES (1)")
+
+	// Begin an external transaction.
+	testDB.Exec(t, "BEGIN")
+
+	// Check that the number of open transactions has incremented. (Might have to move this to later when a query has been executed)
+	require.Equal(t, int64(1), sqlServer.Metrics.EngineMetrics.SQLTxnsOpen.Value())
+
+	// Create a state of contention.
+	testDB.Exec(t, "SELECT * FROM t.foo WHERE i = 1 FOR UPDATE")
+
+	// Execute internal statement (this case is identical to opening an internal
+	// transaction).
+	go func() {
+		_, err := s.InternalExecutor().(*sql.InternalExecutor).ExecEx(ctx,
+			"test-internal-active-stmt-wait",
+			nil,
+			sessiondata.InternalExecutorOverride{User: security.RootUserName()},
+			selectQuery)
+		require.NoError(t, err, "expected internal SELECT query to be successful, but encountered an error")
+	}()
+
+	// Check that the internal statement is active.
+	testutils.SucceedsWithin(t, func() error {
+		row := testDB.QueryStr(t, selectInternalQueryActive)
+		if row[0][0] == "0" {
+			return errors.New("internal select query is not active yet")
+		}
+		return nil
+	}, 3*time.Second)
+
+	// Check that the number of open transactions has not incremented. We only
+	// want to track external open transactions. Open transaction count already
+	// at one from initial external transaction.
+	require.Equal(t, int64(1), sqlServer.Metrics.EngineMetrics.SQLTxnsOpen.Value())
+	// Check that the number of active statements has not incremented. We only
+	// want to track external active statements.
+	require.Equal(t, int64(0), sqlServer.Metrics.EngineMetrics.SQLActiveStatements.Value())
+
+	// Create active external statement.
+	go func() {
+		dbConn := serverutils.OpenDBConn(t, s.ServingSQLAddr(), "", false /* insecure */, s.Stopper())
+		_, err := dbConn.Exec(selectQuery)
+		require.NoError(t, err, "expected external SELECT query to be successful, but encountered an error")
+	}()
+
+	// Check that the external statement is active.
+	testutils.SucceedsWithin(t, func() error {
+		row := testDB.QueryStr(t, selectExternalQueryActive)
+		if row[0][0] == "0" {
+			return errors.New("external select query is not active yet")
+		}
+		return nil
+	}, 3*time.Second)
+
+	// Check that the number of open transactions has incremented. Second db
+	// connection creates an implicit external transaction.
+	require.Equal(t, int64(2), sqlServer.Metrics.EngineMetrics.SQLTxnsOpen.Value())
+	// Check that the number of active statements has incremented.
+	require.Equal(t, int64(1), sqlServer.Metrics.EngineMetrics.SQLActiveStatements.Value())
+
+	// Commit the external transaction. The internal and external select queries
+	// are no longer in contention.
+	testDB.Exec(t, "COMMIT")
+
+	// Check that both the internal & external statements are no longer active.
+	testutils.SucceedsWithin(t, func() error {
+		externalRow := testDB.QueryStr(t, selectExternalQueryActive)
+		internalRow := testDB.QueryStr(t, selectInternalQueryActive)
+
+		if externalRow[0][0] != "0" {
+			return errors.New("external select query is still active")
+		}
+		if internalRow[0][0] != "0" {
+			return errors.New("external select query is still active")
+		}
+		return nil
+	}, 3*time.Second)
+
+	// Check that the number of open transactions has decremented by 2 (should
+	// decrement for initial external transaction and external statement executed
+	// on second db connection).
+	require.Equal(t, int64(0), sqlServer.Metrics.EngineMetrics.SQLTxnsOpen.Value())
+	// Check that the number of active statements has decremented by 1 (should
+	// not decrement for the internal active statement).
+	require.Equal(t, int64(0), sqlServer.Metrics.EngineMetrics.SQLActiveStatements.Value())
+}
+
 // dynamicRequestFilter exposes a filter method which is a
 // kvserverbase.ReplicaRequestFilter but can be set dynamically.
 type dynamicRequestFilter struct {

pkg/sql/exec_util.go

@@ -793,6 +793,12 @@ var (
 		Measurement: "Open SQL Transactions",
 		Unit:        metric.Unit_COUNT,
 	}
+	MetaSQLActiveQueries = metric.Metadata{
+		Name:        "sql.statements.active",
+		Help:        "Number of SQL statements currently active",
+		Measurement: "Active Statements",
+		Unit:        metric.Unit_COUNT,
+	}
 	MetaFullTableOrIndexScan = metric.Metadata{
 		Name:        "sql.full.scan.count",
 		Help:        "Number of full table or index scans",

pkg/sql/executor_statement_metrics.go

@@ -39,6 +39,7 @@ type EngineMetrics struct {
 	SQLServiceLatency     *metric.Histogram
 	SQLTxnLatency         *metric.Histogram
 	SQLTxnsOpen           *metric.Gauge
+	SQLActiveStatements   *metric.Gauge
 
 	// TxnAbortCount counts transactions that were aborted, either due
 	// to non-retriable errors, or retriable errors when the client-side

pkg/ts/catalog/chart_catalog.go

@@ -2131,6 +2131,14 @@ var charts = []sectionDescription{
 				},
 				AxisLabel: "Transactions",
 			},
+			{
+				Title: "Active Statements",
+				Metrics: []string{
+					"sql.statements.active",
+					"sql.statements.active.internal",
+				},
+				AxisLabel: "Transactions",
+			},
 			{
 				Title: "Full Table Index Scans",
 				Metrics: []string{

pkg/ui/workspaces/db-console/src/views/cluster/containers/nodeGraphs/dashboards/sql.tsx

@@ -63,7 +63,7 @@ export default function(props: GraphDashboardProps) {
     >
       <Axis label="queries">
         <Metric
-          name="cr.node.sql.distsql.queries.active"
+          name="cr.node.sql.statements.active"
           title="Active Statements"
         />
       </Axis>