sqlstats,insights: adjust testing and move insights knobs

Now that the sql exec stats ingester lives in the sqlstats package we move the
relevant testing knobs from the insights package to the sql stats knobs. This
commit also adds the `OnIngesterFlush` testing knob to synchronously test the
ingester flush.

Epic: none
Release note: None

Author: xinhaoz

pkg/base/testing_knobs.go

@@ -54,7 +54,6 @@ type TestingKnobs struct {
 	KeyVisualizer                  ModuleTestingKnobs
 	TenantCapabilitiesTestingKnobs ModuleTestingKnobs
 	TableStatsKnobs                ModuleTestingKnobs
-	Insights                       ModuleTestingKnobs
 	TableMetadata                  ModuleTestingKnobs
 	LicenseTestingKnobs            ModuleTestingKnobs
 	VecIndexTestingKnobs           ModuleTestingKnobs

pkg/server/server_sql.go

@@ -105,7 +105,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness/slinstance"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness/slprovider"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
-	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/insights"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/stmtdiagnostics"
 	"github.com/cockroachdb/cockroach/pkg/sql/syntheticprivilegecache"
@@ -1144,10 +1143,6 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		execCfg.ExternalConnectionTestingKnobs = externalConnKnobs.(*externalconn.TestingKnobs)
 	}
 
-	if insightsKnobs := cfg.TestingKnobs.Insights; insightsKnobs != nil {
-		execCfg.InsightsTestingKnobs = insightsKnobs.(*insights.TestingKnobs)
-
-	}
 	var tableStatsTestingKnobs *stats.TableStatsTestingKnobs
 	if tableStatsKnobs := cfg.TestingKnobs.TableStatsKnobs; tableStatsKnobs != nil {
 		tableStatsTestingKnobs = tableStatsKnobs.(*stats.TableStatsTestingKnobs)

pkg/sql/conn_executor.go

@@ -441,7 +441,7 @@ type ServerMetrics struct {
 func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 	metrics := makeMetrics(false /* internal */, &cfg.Settings.SV)
 	serverMetrics := makeServerMetrics(cfg)
-	insightsProvider := insights.New(cfg.Settings, serverMetrics.InsightsMetrics, cfg.InsightsTestingKnobs)
+	insightsProvider := insights.New(cfg.Settings, serverMetrics.InsightsMetrics)
 	reportedSQLStats := sslocal.New(
 		cfg.Settings,
 		sqlstats.MaxMemReportedSQLStatsStmtFingerprints,
@@ -463,7 +463,7 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		reportedSQLStats,
 		cfg.SQLStatsTestingKnobs,
 	)
-	sqlStatsIngester := sslocal.NewSQLStatsIngester(insightsProvider)
+	sqlStatsIngester := sslocal.NewSQLStatsIngester(cfg.SQLStatsTestingKnobs, insightsProvider)
 	// 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) {

pkg/sql/exec_util.go

@@ -98,7 +98,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sessionphase"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
-	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/insights"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/stmtdiagnostics"
 	"github.com/cockroachdb/cockroach/pkg/sql/syntheticprivilegecache"
@@ -1407,7 +1406,6 @@ type ExecutorConfig struct {
 	UnusedIndexRecommendationsKnobs      *idxusage.UnusedIndexRecommendationTestingKnobs
 	ExternalConnectionTestingKnobs       *externalconn.TestingKnobs
 	EventLogTestingKnobs                 *EventLogTestingKnobs
-	InsightsTestingKnobs                 *insights.TestingKnobs
 	TableMetadataKnobs                   *tablemetadatacache_util.TestingKnobs
 
 	// HistogramWindowInterval is (server.Config).HistogramWindowInterval.

pkg/sql/sqlstats/insights/BUILD.bazel

@@ -12,7 +12,6 @@ go_library(
         "provider.go",
         "registry.go",
         "store.go",
-        "test_utils.go",
         "util.go",
     ],
     embed = [":insights_go_proto"],

pkg/sql/sqlstats/insights/insights.go

@@ -133,7 +133,7 @@ type PercentileValues struct {
 }
 
 // New builds a new Provider.
-func New(st *cluster.Settings, metrics Metrics, knobs *TestingKnobs) *Provider {
+func New(st *cluster.Settings, metrics Metrics) *Provider {
 	store := newStore(st)
 	anomalyDetector := newAnomalyDetector(st, metrics)
 
@@ -142,7 +142,7 @@ func New(st *cluster.Settings, metrics Metrics, knobs *TestingKnobs) *Provider {
 		registry: newRegistry(st, &compositeDetector{detectors: []detector{
 			&latencyThresholdDetector{st: st},
 			anomalyDetector,
-		}}, store, knobs),
+		}}, store),
 		anomalyDetector: anomalyDetector,
 	}
 }

pkg/sql/sqlstats/insights/insights_test.go

@@ -41,7 +41,7 @@ func BenchmarkInsights(b *testing.B) {
 	// down, guiding us as we tune buffer sizes, etc.
 	for _, numSessions := range []int{1, 10, 100, 1000, 10000} {
 		b.Run(fmt.Sprintf("numSessions=%d", numSessions), func(b *testing.B) {
-			provider := insights.New(settings, insights.NewMetrics(), nil)
+			provider := insights.New(settings, insights.NewMetrics())
 
 			// Spread the b.N work across the simulated SQL sessions, so that we
 			// can make apples-to-apples comparisons in the benchmark reports:

pkg/sql/sqlstats/insights/integration/BUILD.bazel

@@ -15,6 +15,7 @@ go_test(
         "//pkg/sql/clusterunique",
         "//pkg/sql/contention",
         "//pkg/sql/sessiondata",
+        "//pkg/sql/sqlstats",
         "//pkg/sql/sqlstats/insights",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",

pkg/sql/sqlstats/insights/integration/insights_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/clusterunique"
 	"github.com/cockroachdb/cockroach/pkg/sql/contention"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/insights"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
@@ -89,7 +90,7 @@ func TestInsightsIntegration(t *testing.T) {
 	require.NoError(t, err)
 
 	// Eventually see one recorded insight.
-	testutils.SucceedsWithin(t, func() error {
+	testutils.SucceedsSoon(t, func() error {
 		row = conn.QueryRowContext(ctx, "SELECT count(*), coalesce(string_agg(query, ';'),'') "+
 			"FROM crdb_internal.cluster_execution_insights where app_name = $1 ", appName)
 		if err = row.Scan(&count, &queryText); err != nil {
@@ -99,10 +100,10 @@ func TestInsightsIntegration(t *testing.T) {
 			return fmt.Errorf("expected 1, but was %d, queryText:%s", count, queryText)
 		}
 		return nil
-	}, 1*time.Second)
+	})
 
 	// Verify the table content is valid.
-	testutils.SucceedsWithin(t, func() error {
+	testutils.SucceedsSoon(t, func() error {
 		row = conn.QueryRowContext(ctx, "SELECT "+
 			"query, "+
 			"status, "+
@@ -147,12 +148,12 @@ func TestInsightsIntegration(t *testing.T) {
 		}
 
 		return nil
-	}, 1*time.Second)
+	})
 
 	// TODO (xzhang) Turn this into a datadriven test
 	// https://github.com/cockroachdb/cockroach/issues/95010
 	// Verify the txn table content is valid.
-	testutils.SucceedsWithin(t, func() error {
+	testutils.SucceedsSoon(t, func() error {
 		row = conn.QueryRowContext(ctx, "SELECT "+
 			"query, "+
 			"start_time, "+
@@ -194,7 +195,7 @@ func TestInsightsIntegration(t *testing.T) {
 		}
 
 		return nil
-	}, 1*time.Second)
+	})
 }
 
 func TestFailedInsights(t *testing.T) {
@@ -275,8 +276,7 @@ func TestFailedInsights(t *testing.T) {
 			_, _ = conn.ExecContext(ctx, tc.stmt)
 
 			var query, status, problem, errorCode, errorMsg string
-			testutils.SucceedsWithin(t, func() error {
-
+			testutils.SucceedsSoon(t, func() error {
 				// Query the node execution insights table.
 				row := conn.QueryRowContext(ctx, `
 SELECT query, 
@@ -289,7 +289,7 @@ WHERE query = $1 AND app_name = $2 `,
 					tc.fingerprint, appName)
 
 				return row.Scan(&query, &status, &problem, &errorCode, &errorMsg)
-			}, 1*time.Second)
+			})
 
 			require.Equal(t, tc.status, status)
 			require.Equal(t, tc.problem, problem)
@@ -362,7 +362,7 @@ WHERE query = $1 AND app_name = $2 `,
 			}
 
 			var query, problems, status, errorCode, errorMsg string
-			testutils.SucceedsWithin(t, func() error {
+			testutils.SucceedsSoon(t, func() error {
 
 				// Query the node txn execution insights table.
 				row := conn.QueryRowContext(ctx, `
@@ -375,7 +375,7 @@ FROM crdb_internal.node_txn_execution_insights
 WHERE query = $1 AND app_name = $2`, tc.fingerprint, appName)
 
 				return row.Scan(&query, &problems, &status, &errorCode, &errorMsg)
-			}, 1*time.Second)
+			})
 
 			require.Equal(t, tc.txnStatus, status)
 			require.Equal(t, tc.errorCode, errorCode)
@@ -587,7 +587,7 @@ func TestInsightsPriorityIntegration(t *testing.T) {
 	_, err = conn.ExecContext(ctx, "SELECT pg_sleep(.11)")
 	require.NoError(t, err)
 
-	testutils.SucceedsWithin(t, func() error {
+	testutils.SucceedsSoon(t, func() error {
 		row := conn.QueryRowContext(ctx, "SELECT "+
 			"implicit_txn "+
 			"FROM crdb_internal.node_execution_insights where "+
@@ -604,7 +604,7 @@ func TestInsightsPriorityIntegration(t *testing.T) {
 		}
 
 		return nil
-	}, 2*time.Second)
+	})
 
 	var priorities = []struct {
 		setPriorityQuery      string
@@ -639,7 +639,7 @@ func TestInsightsPriorityIntegration(t *testing.T) {
 	}
 
 	for _, p := range priorities {
-		testutils.SucceedsWithin(t, func() error {
+		testutils.SucceedsSoon(t, func() error {
 			tx, errTxn := conn.BeginTx(ctx, &gosql.TxOptions{})
 			require.NoError(t, errTxn)
 
@@ -654,9 +654,9 @@ func TestInsightsPriorityIntegration(t *testing.T) {
 			errTxn = tx.Commit()
 			require.NoError(t, errTxn)
 			return nil
-		}, 2*time.Second)
+		})
 
-		testutils.SucceedsWithin(t, func() error {
+		testutils.SucceedsSoon(t, func() error {
 			row := conn.QueryRowContext(ctx, "SELECT "+
 				"query, "+
 				"priority, "+
@@ -685,7 +685,7 @@ func TestInsightsPriorityIntegration(t *testing.T) {
 			}
 
 			return nil
-		}, 2*time.Second)
+		})
 	}
 }
 
@@ -911,7 +911,7 @@ func TestInsightsIndexRecommendationIntegration(t *testing.T) {
 	}
 
 	// Verify the table content is valid.
-	testutils.SucceedsWithin(t, func() error {
+	testutils.SucceedsSoon(t, func() error {
 		rows, err := sqlConn.QueryContext(ctx, "SELECT "+
 			"query, "+
 			"array_to_string(index_recommendations, ';') as cmb_index_recommendations "+
@@ -952,7 +952,7 @@ func TestInsightsIndexRecommendationIntegration(t *testing.T) {
 		}
 
 		return nil
-	}, 1*time.Second)
+	})
 }
 
 // TestInsightsClearsPerSessionMemory ensures that memory allocated
@@ -966,8 +966,8 @@ func TestInsightsClearsPerSessionMemory(t *testing.T) {
 	clearedSessionID := clusterunique.ID{}
 	ts := serverutils.StartServerOnly(t, base.TestServerArgs{
 		Knobs: base.TestingKnobs{
-			Insights: &insights.TestingKnobs{
-				OnSessionClear: func(sessionID clusterunique.ID) {
+			SQLStatsKnobs: &sqlstats.TestingKnobs{
+				OnIngesterSessionClear: func(sessionID clusterunique.ID) {
 					defer close(sessionClosedCh)
 					clearedSessionID = sessionID
 				},

pkg/sql/sqlstats/insights/registry.go

@@ -17,10 +17,9 @@ import (
 // statement execution to determine which statements are outliers and
 // writes insights into the provided sink.
 type lockingRegistry struct {
-	detector     detector
-	causes       *causes
-	store        *LockingStore
-	testingKnobs *TestingKnobs
+	detector detector
+	causes   *causes
+	store    *LockingStore
 }
 
 // Instead of creating and allocating a map to track duplicate
@@ -150,13 +149,10 @@ func (r *lockingRegistry) enabled() bool {
 	return r.detector.enabled()
 }
 
-func newRegistry(
-	st *cluster.Settings, detector detector, store *LockingStore, knobs *TestingKnobs,
-) *lockingRegistry {
+func newRegistry(st *cluster.Settings, detector detector, store *LockingStore) *lockingRegistry {
 	return &lockingRegistry{
-		detector:     detector,
-		causes:       &causes{st: st},
-		store:        store,
-		testingKnobs: knobs,
+		detector: detector,
+		causes:   &causes{st: st},
+		store:    store,
 	}
 }

pkg/sql/sqlstats/insights/registry_test.go

@@ -70,10 +70,9 @@ func TestRegistry(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(ctx, &st.SV, 1*time.Second)
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 
 		registry.observeTransaction(txns[0].txn, txns[0].stmts)
-		t.Log("aaaaaaaaaah ", store.stmtCount.Load())
 
 		expected := []*Insight{{
 			Session:     session,
@@ -121,7 +120,7 @@ func TestRegistry(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(ctx, &st.SV, 1*time.Second)
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 		// Transaction status is set during expectedTxnInsight stats recorded based on
 		// if the transaction committed. We'll inject the failure here to align
 		// it with the test. The insights integration tests will verify that this
@@ -156,7 +155,7 @@ func TestRegistry(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(ctx, &st.SV, 0)
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 		registry.observeTransaction(transaction, []*sqlstats.RecordedStmtStats{statement})
 
 		var actual []*Insight
@@ -180,7 +179,7 @@ func TestRegistry(t *testing.T) {
 			ServiceLatencySec: 0.5,
 		}
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 		registry.observeTransaction(transaction, []*sqlstats.RecordedStmtStats{stmt})
 
 		var actual []*Insight
@@ -234,7 +233,7 @@ func TestRegistry(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(ctx, &st.SV, 1*time.Second)
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 
 		expected := []*Insight{{
 			Session:     session,
@@ -317,7 +316,7 @@ func TestRegistry(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(ctx, &st.SV, 1*time.Second)
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 
 		registry.observeTransaction(transaction, []*sqlstats.RecordedStmtStats{
 			statement, siblingStatement,
@@ -335,14 +334,14 @@ func TestRegistry(t *testing.T) {
 	t.Run("txn with no stmts", func(t *testing.T) {
 		transaction := &sqlstats.RecordedTxnStats{TransactionID: uuid.MakeV4(), Committed: true, SessionID: session.ID}
 		st := cluster.MakeTestingClusterSettings()
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, newStore(st), nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, newStore(st))
 		require.NotPanics(t, func() { registry.observeTransaction(transaction, nil) })
 	})
 
 	t.Run("txn with high accumulated contention without high single stmt contention", func(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 		contentionDuration := 10 * time.Second
 		statement := &sqlstats.RecordedStmtStats{
 			SessionID:         session.ID,
@@ -439,7 +438,7 @@ func TestRegistry(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(ctx, &st.SV, 1*time.Second)
 		store := newStore(st)
-		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store, nil)
+		registry := newRegistry(st, &latencyThresholdDetector{st: st}, store)
 		registry.observeTransaction(transaction, stmts)
 
 		expected := []*Insight{