Merge #143855 #147920

143855: sqlstats: move sql stats writing to be async  r=dhartunian a=xinhaoz

This commit removes the recording of sql stats at the end of statement and
transaction execution. The per-node SQLStats container has been added as a
sink to the async sql stats ingester. Instead of having sslocal.StatsCollector
record sql stats at the end of execution, it will now record the stats
asynchronously as part of event processing in sslocal.SQLStatsIngester's
ingest routine.

When creating the execution statistics structs, we now record the application
name from the application container currently that is currenlty set in
sslocal.StatsCollector. This is because the conn exec's app name may have been
mutated if the current statement is a `set application_name` statement but we
still want to record that statement in the previous application.

Epic: none
Closes: #141024

Release note: None

147920: cli: fix flake in TestDockerCLI_test_error_hints r=rafiss a=Dedej-Bergin

Previously a process was not exiting properly and was causing a test flake, this change adds a wait loop which checks that we have exited properly to avoid the flake.

Fixes: #147066
Fixes: #151413
Informs: #115626
Release note: None

Co-authored-by: Xin Hao Zhang <[email protected]>
Co-authored-by: David Hartunian <[email protected]>
Co-authored-by: Bergin Dedej <[email protected]>

Author: 4 people

docs/generated/metrics/metrics.yaml

@@ -9148,6 +9148,22 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.stats.ingester.num_processed
+      exported_name: sql_stats_ingester_num_processed
+      description: Number of items processed by the SQL stats ingester
+      y_axis_label: Items
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: sql.stats.ingester.queue_size
+      exported_name: sql_stats_ingester_queue_size
+      description: Current number of items queued in the SQL stats ingester
+      y_axis_label: Items
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
     - name: sql.stats.mem.current
       exported_name: sql_stats_mem_current
       description: Current memory usage for fingerprint storage

pkg/cli/interactive_tests/common.tcl

@@ -73,7 +73,7 @@ proc eexpect {text} {
 }
 
 # eexpect_re is like eexpect, but takes a regular expression argument
-# instead of a text string
+# instead of a text string.
 proc eexpect_re {text} {
   expect {
     -re $text {}
@@ -88,6 +88,44 @@ proc interrupt {} {
     sleep 0.4
 }
 
+# Convenience function that sends Ctrl+C and waits more reliably for process to exit.
+# This uses a retry approach similar to stop_server but adapted for expect environment.
+proc interrupt_and_wait {} {
+    report "INTERRUPT AND WAIT FOR FOREGROUND PROCESS"
+
+    # Send the interrupt signal
+    send "\003"
+
+    # Use a retry loop similar to stop_server's wait mechanism.
+    # Try up to 30 times with 1 second intervals, like stop_server does.
+    try {
+        # Set shorter timeout for the retry loop
+        set timeout 1
+
+        for {set i 1} {$i <= 30} {incr i} {
+            # Try to expect the shell prompt with a short timeout (1 second per attempt).
+            expect {
+                ":/# " {
+                    report "FOREGROUND PROCESS EXITED SUCCESSFULLY (after $i attempt(s))"
+                    return
+                }
+                timeout {
+                    # Process might still be running, continue waiting.
+                    report "still waiting for process to exit (attempt $i/30)"
+                    # The sleep is implicit in the timeout, so we don't need an additional sleep here.
+                }
+            }
+        }
+
+        # If we get here, the process didn't exit cleanly within 30 seconds.
+        report "TIMEOUT: foreground process still running after 30 attempts"
+        error "Failed to interrupt foreground process - it did not exit within 30 seconds"
+    } finally {
+        # Restore original timeout.
+        set timeout 45
+    }
+}
+
 # Convenience function that sends Ctrl+D to the monitored process.
 # Leaves some upfront delay to let the readline process the time
 # to initialize the key binding.

pkg/cli/interactive_tests/test_error_hints.tcl

@@ -69,9 +69,7 @@ end_test
 # server.
 
 set spawn_id $shell_spawn_id
-interrupt
-interrupt
-eexpect ":/# "
+interrupt_and_wait
 
 send "$argv start-single-node --listen-addr=localhost --certs-dir=$certs_dir --store=logs/db\r"
 eexpect "restarted pre-existing node"
@@ -90,13 +88,10 @@ eexpect "ERROR: node is running secure mode, SSL connection required"
 eexpect ":/# "
 end_test
 
-
 # Check what happens when attempting to connect to something
 # that is not a CockroachDB server.
 set spawn_id $shell_spawn_id
-interrupt
-interrupt
-eexpect ":/# "
+interrupt_and_wait
 
 start_test "Connecting an insecure RPC client to a non-CockroachDB server"
 # In one shell, start a bogus server
@@ -110,8 +105,7 @@ eexpect "insecure\r\n"
 set spawn_id $shell_spawn_id
 eexpect "connected"
 eexpect "PRI * HTTP/2.0"
-interrupt
-eexpect ":/# "
+interrupt_and_wait
 # Check that cockroach node drain becomes suitably confused.
 set spawn_id $client_spawn_id
 eexpect "ERROR: server closed the connection."
@@ -133,8 +127,7 @@ eexpect "insecure\r\n"
 set spawn_id $shell_spawn_id
 eexpect "connected"
 eexpect "cockroach sql"
-interrupt
-eexpect ":/# "
+interrupt_and_wait
 # Check that cockroach sql becomes suitably confused.
 set spawn_id $client_spawn_id
 eexpect "ERROR: server closed the connection."

pkg/server/application_api/sql_stats_test.go

@@ -59,6 +59,7 @@ func TestStatusAPICombinedTransactions(t *testing.T) {
 
 	var params base.TestServerArgs
 	params.Knobs.SpanConfig = &spanconfig.TestingKnobs{ManagerDisableJobCreation: true} // TODO(irfansharif): #74919.
+	params.Knobs.SQLStatsKnobs = &sqlstats.TestingKnobs{SynchronousSQLStats: true}
 	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: params,
 	})
@@ -130,8 +131,6 @@ func TestStatusAPICombinedTransactions(t *testing.T) {
 		}
 	}
 
-	// Wait for last app name to be flushed to in memory stats.
-
 	// Hit query endpoint.
 	var resp serverpb.StatementsResponse
 	if err := srvtestutils.GetStatusJSONProtoWithAdminAndTimeoutOption(firstServerProto, "combinedstmts", &resp, true, additionalTimeout); err != nil {
@@ -199,7 +198,15 @@ func TestStatusAPITransactions(t *testing.T) {
 	skip.UnderDeadlock(t, "test is very slow under deadlock")
 	skip.UnderRace(t, "test is too slow to run under race")
 
-	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{})
+	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			Knobs: base.TestingKnobs{
+				SQLStatsKnobs: &sqlstats.TestingKnobs{
+					SynchronousSQLStats: true,
+				},
+			},
+		},
+	})
 	ctx := context.Background()
 	defer testCluster.Stopper().Stop(ctx)
 
@@ -330,7 +337,15 @@ func TestStatusAPITransactionStatementFingerprintIDsTruncation(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{})
+	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			Knobs: base.TestingKnobs{
+				SQLStatsKnobs: &sqlstats.TestingKnobs{
+					SynchronousSQLStats: true,
+				},
+			},
+		},
+	})
 	defer testCluster.Stopper().Stop(context.Background())
 
 	firstServerProto := testCluster.Server(0).ApplicationLayer()
@@ -398,6 +413,7 @@ func TestStatusAPIStatements(t *testing.T) {
 	aggregatedTs := int64(1630353000)
 	statsKnobs := sqlstats.CreateTestingKnobs()
 	statsKnobs.StubTimeNow = func() time.Time { return timeutil.Unix(aggregatedTs, 0) }
+	statsKnobs.SynchronousSQLStats = true
 	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Knobs: base.TestingKnobs{
@@ -521,6 +537,7 @@ func TestStatusAPICombinedStatementsTotalLatency(t *testing.T) {
 	}
 
 	sqlStatsKnobs := sqlstats.CreateTestingKnobs()
+	sqlStatsKnobs.SynchronousSQLStats = true
 	// Start the cluster.
 	srv, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{
 		Insecure: true,
@@ -682,6 +699,7 @@ func TestStatusAPICombinedStatementsWithFullScans(t *testing.T) {
 	oneMinAfterAggregatedTs := aggregatedTs + 60
 	statsKnobs := sqlstats.CreateTestingKnobs()
 	statsKnobs.StubTimeNow = func() time.Time { return timeutil.Unix(aggregatedTs, 0) }
+	statsKnobs.SynchronousSQLStats = true
 	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Knobs: base.TestingKnobs{
@@ -856,6 +874,7 @@ func TestStatusAPICombinedStatements(t *testing.T) {
 	aggregatedTs := int64(1630353000)
 	statsKnobs := sqlstats.CreateTestingKnobs()
 	statsKnobs.StubTimeNow = func() time.Time { return timeutil.Unix(aggregatedTs, 0) }
+	statsKnobs.SynchronousSQLStats = true
 	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Knobs: base.TestingKnobs{
@@ -1031,6 +1050,7 @@ func TestStatusAPIStatementDetails(t *testing.T) {
 	aggregatedTs := int64(1630353000)
 	statsKnobs := sqlstats.CreateTestingKnobs()
 	statsKnobs.StubTimeNow = func() time.Time { return timeutil.Unix(aggregatedTs, 0) }
+	statsKnobs.SynchronousSQLStats = true
 	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Knobs: base.TestingKnobs{
@@ -1306,6 +1326,7 @@ func TestCombinedStatementUsesCorrectSourceTable(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	skip.UnderDuress(t)
 	ctx := context.Background()
 
 	// Disable flushing sql stats so we can manually set the table states
@@ -1314,6 +1335,7 @@ func TestCombinedStatementUsesCorrectSourceTable(t *testing.T) {
 	statsKnobs := sqlstats.CreateTestingKnobs()
 	defaultMockInsertedAggTs := timeutil.Unix(1696906800, 0)
 	statsKnobs.StubTimeNow = func() time.Time { return defaultMockInsertedAggTs }
+	statsKnobs.SynchronousSQLStats = true
 	persistedsqlstats.SQLStatsFlushEnabled.Override(ctx, &settings.SV, false)
 	srv := serverutils.StartServerOnly(t, base.TestServerArgs{
 		Settings: settings,
@@ -1550,7 +1572,15 @@ func TestDrainSqlStats(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 	appName := "drain_stats_app"
-	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{})
+	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			Knobs: base.TestingKnobs{
+				SQLStatsKnobs: &sqlstats.TestingKnobs{
+					SynchronousSQLStats: true,
+				},
+			},
+		},
+	})
 	ctx := context.Background()
 	defer testCluster.Stopper().Stop(ctx)
 
@@ -1587,7 +1617,15 @@ func TestDrainSqlStats_partialOutage(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 	appName := "drain_stats_app"
-	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{})
+	testCluster := serverutils.StartCluster(t, 3, base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			Knobs: base.TestingKnobs{
+				SQLStatsKnobs: &sqlstats.TestingKnobs{
+					SynchronousSQLStats: true,
+				},
+			},
+		},
+	})
 	ctx := context.Background()
 	defer testCluster.Stopper().Stop(ctx)
 
@@ -1636,15 +1674,18 @@ func TestClusterResetSQLStats(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	ctx := context.Background()
+	skip.UnderDuress(t)
 
+	ctx := context.Background()
+	knobs := sqlstats.CreateTestingKnobs()
+	knobs.SynchronousSQLStats = true
 	for _, flushed := range []bool{false, true} {
 		t.Run(fmt.Sprintf("flushed=%t", flushed), func(t *testing.T) {
 			testCluster := serverutils.StartCluster(t, 3 /* numNodes */, base.TestClusterArgs{
 				ServerArgs: base.TestServerArgs{
 					Insecure: true,
 					Knobs: base.TestingKnobs{
-						SQLStatsKnobs: sqlstats.CreateTestingKnobs(),
+						SQLStatsKnobs: knobs,
 					},
 				},
 			})

pkg/sql/conn_executor.go

@@ -430,6 +430,9 @@ type ServerMetrics struct {
 
 	// InsightsMetrics contains metrics related to outlier detection.
 	InsightsMetrics insights.Metrics
+
+	// IngesterMetrics contains metrics related to SQL stats ingestion.
+	IngesterMetrics sslocal.Metrics
 }
 
 // NewServer creates a new Server. Start() needs to be called before the Server
@@ -444,6 +447,7 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		sqlstats.MaxMemReportedSQLStatsTxnFingerprints,
 		serverMetrics.StatsMetrics.ReportedSQLStatsMemoryCurBytesCount,
 		serverMetrics.StatsMetrics.ReportedSQLStatsMemoryMaxBytesHist,
+		nil, /* discardedStatsCount */
 		pool,
 		nil, /* reportedProvider */
 		cfg.SQLStatsTestingKnobs,
@@ -454,11 +458,13 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		sqlstats.MaxMemSQLStatsTxnFingerprints,
 		serverMetrics.StatsMetrics.SQLStatsMemoryCurBytesCount,
 		serverMetrics.StatsMetrics.SQLStatsMemoryMaxBytesHist,
+		serverMetrics.StatsMetrics.DiscardedStatsCount,
 		pool,
 		reportedSQLStats,
 		cfg.SQLStatsTestingKnobs,
 	)
-	sqlStatsIngester := sslocal.NewSQLStatsIngester(cfg.SQLStatsTestingKnobs, insightsProvider)
+	sqlStatsIngester := sslocal.NewSQLStatsIngester(
+		cfg.Settings, cfg.SQLStatsTestingKnobs, serverMetrics.IngesterMetrics, insightsProvider, localSQLStats)
 	// TODO(117690): Unify StmtStatsEnable and TxnStatsEnable into a single cluster setting.
 	sqlstats.TxnStatsEnable.SetOnChange(&cfg.Settings.SV, func(_ context.Context) {
 		if !sqlstats.TxnStatsEnable.Get(&cfg.Settings.SV) {
@@ -672,6 +678,7 @@ func makeServerMetrics(cfg *ExecutorConfig) ServerMetrics {
 		},
 		ContentionSubsystemMetrics: txnidcache.NewMetrics(),
 		InsightsMetrics:            insights.NewMetrics(),
+		IngesterMetrics:            sslocal.NewIngesterMetrics(),
 	}
 }
 
@@ -1256,7 +1263,6 @@ func (s *Server) newConnExecutor(
 		s.sqlStatsIngester,
 		ex.phaseTimes,
 		s.localSqlStats.GetCounters(),
-		underOuterTxn,
 		s.cfg.SQLStatsTestingKnobs,
 	)
 	ex.dataMutatorIterator.onApplicationNameChange = func(newName string) {
@@ -4082,17 +4088,7 @@ func (ex *connExecutor) txnStateTransitionsApplyWrapper(
 				return advanceInfo{}, err
 			}
 			if advInfo.txnEvent.eventType == txnUpgradeToExplicit {
-				// TODO (xinhaoz): This is a temporary hook until
-				// https://github.com/cockroachdb/cockroach/issues/141024
-				// is resolved. The reason this exists is because we
-				// need to recompute the statement fingerprint id for
-				// statements currently in the stats collector, which
-				// were computed once already for insights. There is an
-				// acknowledgement that this means the fingerprint id
-				// given to insights for upgraded transactions are
-				// currently incorrect.
 				ex.extraTxnState.txnFinishClosure.implicit = false
-				ex.statsCollector.UpgradeToExplicitTransaction()
 			}
 		}
 	case txnStart: