Merge #150535 #150771 #150947 #150949

150535: ui: show tenant dropdown in insecure mode when multiple tenants exist r=dt a=dt

Previously, the tenant dropdown would not appear in insecure mode because it required a tenant cookie to be set, which only happens after authentication. This change modifies the dropdown logic to show when running in insecure mode and the virtual_clusters endpoint returns more than one tenant.

In insecure mode, the dropdown now displays "default" when no tenant cookie is set, allowing users to select from available virtual clusters without requiring authentication.

Subsumes #150143.
Closes #150143.

Release note (ui change): The tenant dropdown now appears in insecure mode when multiple virtual clusters are available.

Epic: none.

150771: sql/ttl: improve TTL replan decision logic r=spilchen a=spilchen

Replace calculatePlanGrowth with detectNodeAvailabilityChanges to make TTL job replanning less sensitive to span changes. The new logic focuses specifically on detecting when nodes become unavailable rather than reacting to all plan differences.

The previous implementation would trigger replans for span splits/merges that don't actually indicate beneficial restart scenarios. The new approach only considers missing nodes from the original plan, which typically indicates node failures where work redistribution would benefit from restarting the job. It also supports a stability window so that replan decisions need to fire consecutively. This should help aleviate changes in plans due to range cache issues.

Fixes #150343

Epic: none
Release note (ops change): The 'sql.ttl.replan_flow_threshold' may have been set to 0 to work around the TTL replanner being too sensitive. This fix will alleviate that and any instance that had set replan_flow_threshold to 0 can be reset back to the default.


150947: asim: port over some trivial commits from prototype r=tbg a=wenyihu6

This commit ports over some trivial changes from the mma prototype branch.

Epic: CRDB-25222
Release note: none

150949: span: clean up test code after llrb frontier deletion r=rharding6373 a=andyyang890

This patch cleans up some code in the unit tests that previously
would toggle the frontier type but now just leads to each test
being running twice with the exact same configuration.

Informs #141668

Release note: None

Co-authored-by: David Taylor <[email protected]>
Co-authored-by: Matt Spilchen <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: Andy Yang <[email protected]>

Author: 5 people

pkg/kv/kvserver/asim/metrics/series.go

@@ -17,7 +17,7 @@ func MakeTS(metrics [][]StoreMetrics) map[string][][]float64 {
 	}
 
 	stores := len(metrics[0])
-	// TODO(kvoli): Either begin resuing prometheus metric definitions with a
+	// TODO(kvoli): Either begin reusing prometheus metric definitions with a
 	// custom scraper or provide definitions for each metric available. These
 	// are partially duplicated with the cluster tracker.
 	ret["qps"] = make([][]float64, stores)

pkg/kv/kvserver/asim/state/load.go

@@ -39,10 +39,6 @@ func LoadEventQPS(le workload.LoadEvent) float64 {
 
 // ReplicaLoadCounter is the sum of all key accesses and size of bytes, both written
 // and read.
-// TODO(kvoli): In the non-simulated code, replica_stats currently maintains
-// this structure, which is rated. This datastructure needs to be adapated by
-// the user to be rated over time. In the future we should introduce a better
-// general pupose stucture that enables rating.
 type ReplicaLoadCounter struct {
 	WriteKeys  int64
 	WriteBytes int64

pkg/kv/kvserver/asim/state/parser_replica_placement.go

@@ -121,6 +121,9 @@ func ParseReplicaPlacement(input string) ReplicaPlacement {
 			}
 			// For matches[0] and stores[0], storeID will be 1.
 			storeID, _ := strconv.Atoi(parts[0][1:])
+			if storeID == 0 {
+				panic(fmt.Sprintf("unable to parse store id: %s", parts[0][1:]))
+			}
 			storeSet = append(storeSet, storeID)
 
 			// If the replica type or leaseholder is not specified, artificially

pkg/kv/kvserver/asim/tests/datadriven_simulation_test.go

@@ -145,13 +145,16 @@ var runAsimTests = envutil.EnvOrDefaultBool("COCKROACH_RUN_ASIM_TESTS", false)
 //     constraints(violating) at the end of the evaluation.
 //
 //   - "setting" [replicate_queue_enabled=bool] [lease_queue_enabled=bool]
-//     [split_queue_enabled=bool] [rebalance_mode=<int>] [rebalance_interval=<duration>]
-//     [rebalance_qps_threshold=<float>] [split_qps_threshold=<float>]
+//     [split_queue_enabled=bool] [rebalance_interval=<duration>]
+//     [rebalance_mode=<int>] [split_qps_threshold=<float>]
 //     [rebalance_range_threshold=<float>] [gossip_delay=<duration>]
+//     [rebalance_objective=<int>]
 //     Configure the simulation's various settings. The default values are:
-//     rebalance_mode=2 (leases and replicas) rebalance_interval=1m (1 minute)
-//     rebalance_qps_threshold=0.1 split_qps_threshold=2500
-//     rebalance_range_threshold=0.05 gossip_delay=500ms.
+//     replicate_queue_enabled=true lease_queue_enabled=true
+//     split_queue_enabled=true rebalance_interval=1m (1 minute)
+//     rebalance_mode=2 (leases and replicas) split_qps_threshold=2500
+//     rebalance_range_threshold=0.05 gossip_delay=500ms rebalance_objective=0
+//     (QPS) (1=CPU)
 //
 //   - "eval" [duration=<string>] [samples=<int>] [seed=<int>]
 //     Run samples (e.g. samples=5) number of simulations for duration (e.g.

pkg/server/server.go

@@ -1303,6 +1303,7 @@ func NewServer(cfg Config, stopper *stop.Stopper) (serverctl.ServerStartupInterf
 		tenantCapabilitiesWatcher,
 		cfg.DisableSQLServer,
 		cfg.BaseConfig.DisableTLSForHTTP,
+		cfg.Insecure,
 	)
 	drain.serverCtl = sc
 

pkg/server/server_controller.go

@@ -100,6 +100,8 @@ type serverController struct {
 
 	disableTLSForHTTP bool
 
+	insecure bool
+
 	mu struct {
 		syncutil.RWMutex
 
@@ -136,6 +138,7 @@ func newServerController(
 	watcher *tenantcapabilitieswatcher.Watcher,
 	disableSQLServer bool,
 	disableTLSForHTTP bool,
+	insecure bool,
 ) *serverController {
 	c := &serverController{
 		AmbientContext:      ambientCtx,
@@ -150,6 +153,7 @@ func newServerController(
 		drainCh:             make(chan struct{}),
 		disableSQLServer:    disableSQLServer,
 		disableTLSForHTTP:   disableTLSForHTTP,
+		insecure:            insecure,
 	}
 	c.orchestrator = newChannelOrchestrator(parentStopper, c)
 	c.mu.servers = map[roachpb.TenantName]*serverState{

pkg/server/server_controller_http.go

@@ -7,6 +7,7 @@ package server
 
 import (
 	"bytes"
+	"encoding/json"
 	"io"
 	"net/http"
 	"strconv"
@@ -53,6 +54,26 @@ var ServerHTTPBasePath = settings.RegisterStringSetting(
 	settings.WithPublic,
 )
 
+func (c *serverController) insecureVirtualClusterList(w http.ResponseWriter, r *http.Request) {
+	tenantNames := c.getCurrentTenantNames()
+	tenants := make([]string, len(tenantNames))
+	for i, name := range tenantNames {
+		tenants[i] = string(name)
+	}
+
+	resp := &virtualClustersResp{
+		VirtualClusters: tenants,
+	}
+	respBytes, err := json.Marshal(resp)
+	if err != nil {
+		http.Error(w, "unable to marshal virtual clusters JSON", http.StatusInternalServerError)
+		return
+	}
+	if _, err := w.Write(respBytes); err != nil {
+		log.Errorf(r.Context(), "unable to write virtual clusters response: %s", err.Error())
+	}
+}
+
 // httpMux redirects incoming HTTP requests to the server selected by
 // the special HTTP request header.
 // If no tenant is specified, the default tenant is used.
@@ -78,6 +99,13 @@ func (c *serverController) httpMux(w http.ResponseWriter, r *http.Request) {
 		c.attemptLogoutFromAllTenants().ServeHTTP(w, r)
 		return
 	}
+	if c.insecure && r.URL.Path == virtualClustersPath {
+		// If the insecure flag is set, the virtual clusters endpoint should just
+		// return all tennants instead of delegating to a tenant server to read the
+		// auth cookie which doesn't exist.
+		c.insecureVirtualClusterList(w, r)
+		return
+	}
 	tenantName := getTenantNameFromHTTPRequest(c.st, r)
 	noFallback := false
 	if noFallbackValue := r.URL.Query().Get(NoFallbackParam); noFallbackValue != "" {

pkg/sql/ttl/ttljob/ttljob.go

@@ -8,6 +8,7 @@ package ttljob
 import (
 	"context"
 	"math/rand"
+	"sync/atomic"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
@@ -55,6 +56,14 @@ var replanFrequency = settings.RegisterDurationSetting(
 	settings.PositiveDuration,
 )
 
+var replanStabilityWindow = settings.RegisterIntSetting(
+	settings.ApplicationLevel,
+	"sql.ttl.replan_stability_window",
+	"number of consecutive replan evaluations required before triggering a replan; set to 1 to disable stability window",
+	2,
+	settings.PositiveInt,
+)
+
 // rowLevelTTLResumer implements the TTL job. The job can run on any node, but
 // the job node distributes SELECT/DELETE work via DistSQL to ttlProcessor
 // nodes. DistSQL divides work into spans that each ttlProcessor scans in a
@@ -65,6 +74,9 @@ type rowLevelTTLResumer struct {
 	physicalPlan *sql.PhysicalPlan
 	planCtx      *sql.PlanningCtx
 
+	// consecutiveReplanDecisions tracks how many consecutive times replan was deemed necessary.
+	consecutiveReplanDecisions *atomic.Int64
+
 	mu struct {
 		syncutil.Mutex
 		// lastUpdateTime is the wall time of the last job progress update.
@@ -296,7 +308,10 @@ func (t *rowLevelTTLResumer) Resume(ctx context.Context, execCtx interface{}) (r
 	// the TTL job to utilize those nodes for parallel work.
 	replanChecker, cancelReplanner := sql.PhysicalPlanChangeChecker(
 		ctx, t.physicalPlan, makePlan, jobExecCtx,
-		sql.ReplanOnChangedFraction(func() float64 { return replanThreshold.Get(&execCfg.Settings.SV) }),
+		replanDecider(t.consecutiveReplanDecisions,
+			func() int64 { return replanStabilityWindow.Get(&execCfg.Settings.SV) },
+			func() float64 { return replanThreshold.Get(&execCfg.Settings.SV) },
+		),
 		func() time.Duration { return replanFrequency.Get(&execCfg.Settings.SV) },
 	)
 
@@ -507,11 +522,91 @@ func (t *rowLevelTTLResumer) refreshProgress(
 	return newProgress, nil
 }
 
+// replanDecider returns a function that determines whether a TTL job should be
+// replanned based on changes in the physical execution plan. It compares the
+// old and new plans to detect node availability changes and decides if the
+// benefit of replanning (better parallelization) outweighs the cost of
+// restarting the job. It implements a stability window to avoid replanning
+// due to transient changes.
+func replanDecider(
+	consecutiveReplanDecisions *atomic.Int64,
+	stabilityWindowFn func() int64,
+	thresholdFn func() float64,
+) sql.PlanChangeDecision {
+	return func(ctx context.Context, oldPlan, newPlan *sql.PhysicalPlan) bool {
+		changed, growth := detectNodeAvailabilityChanges(oldPlan, newPlan)
+		threshold := thresholdFn()
+		shouldReplan := threshold != 0.0 && growth > threshold
+
+		stabilityWindow := stabilityWindowFn()
+
+		var currentDecisions int64
+		if shouldReplan {
+			currentDecisions = consecutiveReplanDecisions.Add(1)
+		} else {
+			consecutiveReplanDecisions.Store(0)
+			currentDecisions = 0
+		}
+
+		// If stability window is 1, replan immediately. Otherwise, require
+		// consecutive decisions to meet the window threshold.
+		replan := currentDecisions >= stabilityWindow
+
+		// Reset the counter when we decide to replan, since the job will restart
+		if replan {
+			consecutiveReplanDecisions.Store(0)
+		}
+
+		if shouldReplan || growth > 0.1 || log.V(1) {
+			log.Infof(ctx, "Re-planning would add or alter flows on %d nodes / %.2f, threshold %.2f, consecutive decisions %d/%d, replan %v",
+				changed, growth, threshold, currentDecisions, stabilityWindow, replan)
+		}
+
+		return replan
+	}
+}
+
+// detectNodeAvailabilityChanges analyzes differences between two physical plans
+// to determine if nodes have become unavailable. It returns the number of nodes
+// that are no longer available and the fraction of the original plan affected.
+//
+// The function focuses on detecting when nodes from the original plan are missing
+// from the new plan, which typically indicates node failures. When nodes fail,
+// their work gets redistributed to remaining nodes, making a job restart
+// beneficial for better parallelization. We ignore newly added nodes since
+// continuing the current job on existing nodes is usually more efficient than
+// restarting to incorporate new capacity.
+func detectNodeAvailabilityChanges(before, after *sql.PhysicalPlan) (int, float64) {
+	var changed int
+	beforeSpecs, beforeCleanup := before.GenerateFlowSpecs()
+	defer beforeCleanup(beforeSpecs)
+	afterSpecs, afterCleanup := after.GenerateFlowSpecs()
+	defer afterCleanup(afterSpecs)
+
+	// Count nodes from the original plan that are no longer present in the new plan.
+	// We only check nodes in beforeSpecs because we specifically want to detect
+	// when nodes that were doing work are no longer available, which typically
+	// indicates beneficial restart scenarios (node failures where work can be
+	// redistributed more efficiently).
+	for n := range beforeSpecs {
+		if _, ok := afterSpecs[n]; !ok {
+			changed++
+		}
+	}
+
+	var frac float64
+	if changed > 0 {
+		frac = float64(changed) / float64(len(beforeSpecs))
+	}
+	return changed, frac
+}
+
 func init() {
 	jobs.RegisterConstructor(jobspb.TypeRowLevelTTL, func(job *jobs.Job, settings *cluster.Settings) jobs.Resumer {
 		return &rowLevelTTLResumer{
-			job: job,
-			st:  settings,
+			job:                        job,
+			st:                         settings,
+			consecutiveReplanDecisions: &atomic.Int64{},
 		}
 	}, jobs.UsesTenantCostControl)
 }