Merge #145482 #146177 #146248 #146288

145482: util/mon: complete the revert of "nil as unlimited account" r=yuzefovich a=yuzefovich

**util/mon: remove temporary nil checks**

This commit finalizes the process of reverting of "nil as unlimited
account" behavior that was mostly done in b8d259d. The main leftover piece was
that we left the previous behavior in productions builds out of caution.
It's been a year now without any panics in our tests, so it should be
safe to remove this.

**util/mon: assume non-nil ConcurrentBoundAccount**

This commit completes the process of reverting of "nil as
unlimited account" behavior. The only remaining case was with
ConcurrentBoundAccount used only in a handful of places, so they were
audited to ensure that non-nil account was used.

Epic: None
Release note: None

146177: security: update TestTLSCipherRestrict test setup r=souravcrl a=souravcrl

Current test setup starts the test server only once and runs various tls cipher
configurations but this seems to fail the test as final cipher configuration set
by previous run of the test may not be overridden by current run, and we have
stale ciphers configured for the subtest. Another issue is the timout being set
is 2s which may fail and increasing it is not possible as the test may timeout
before that leading to leaky goroutines. Hence, handling the http client
response timeouts as a flake. Waitgroups are added for all client calls to
ensure there are no open connections during test breakdown.

fixes #145812
fixes #145527
fixes #145459
fixes #145313

Release Note: None

146248: changefeedccl: add timeout to nemeses test inserts r=aerfrei a=rharding6373

Some mutation queries take a long time to plan and execute even if they're safe. This PR adds a short timeout to each mutation, like tlp does for mutations, in order to avoid hanging.

Fixes: #146159
Epic: none

Release note: none

146288: admission: log the ingest model r=kvoli a=sumeerbhola

We were previously only logging the L0 ingest model, which is useful for modeling the growth of L0. For disk bandwidth accounting, the ingest model for the whole LSM is used, and with this change it is also logged. Logging these models helps understand the behavior of the ioLoadListener.

Epic: none

Release note: None

Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: souravcrl <[email protected]>
Co-authored-by: rharding6373 <[email protected]>
Co-authored-by: sumeerbhola <[email protected]>

Author: 5 people

pkg/ccl/changefeedccl/cdctest/nemeses.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"math/rand"
 	"strings"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/internal/sqlsmith"
@@ -418,11 +419,16 @@ func RunNemesis(
 
 		defer queryGen.Close()
 		const numInserts = 100
+		const insertTimeout = 5 * time.Second
 		time := timeutil.Now()
 		for i := 0; i < numInserts; i++ {
 			query := queryGen.Generate()
 			log.Infof(ctx, "Executing query: %s", query)
-			_, err := db.Exec(query)
+			err := timeutil.RunWithTimeout(ctx, "nemeses populate table",
+				insertTimeout, func(ctx context.Context) error {
+					_, err := db.ExecContext(ctx, query)
+					return err
+				})
 			log.Infof(ctx, "Time taken to execute last query: %s", timeutil.Since(time))
 			time = timeutil.Now()
 			if err != nil {

pkg/kv/kvclient/rangefeed/db_adapter.go

@@ -118,6 +118,8 @@ func (dbc *dbAdapter) Scan(
 	if cfg.mon != nil {
 		acc = cfg.mon.MakeConcurrentBoundAccount()
 		defer acc.Close(ctx)
+	} else {
+		acc = mon.NewStandaloneUnlimitedConcurrentAccount()
 	}
 
 	// If we don't have parallelism configured, just scan each span in turn.

pkg/security/BUILD.bazel

@@ -93,7 +93,6 @@ go_test(
         "//pkg/server",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",
-        "//pkg/testutils/skip",
         "//pkg/util/envutil",
         "//pkg/util/leaktest",
         "//pkg/util/log",

pkg/security/tls_test.go

@@ -13,16 +13,15 @@ import (
 	"net"
 	"net/http"
 	"strings"
+	"sync"
 	"testing"
-	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/security/certnames"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
-	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -124,50 +123,6 @@ func verifyX509Cert(cert *x509.Certificate, dnsName string, roots *x509.CertPool
 func TestTLSCipherRestrict(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	skip.UnderStress(t, "http server accessing previous test's restriction fn")
-	skip.UnderRace(t)
-
-	// since the listener does not return rpc/sql/http connection errors, we
-	// need to have a separate hook to obtain and validate it.
-	type cipherErrContainer struct {
-		syncutil.Mutex
-		err net.Error
-	}
-
-	cipherErrC := &cipherErrContainer{}
-	cipherRestrictFn := security.TLSCipherRestrict
-	defer testutils.TestingHook(&security.TLSCipherRestrict, func(conn net.Conn) (err net.Error) {
-		err = cipherRestrictFn(conn)
-		cipherErrC.Lock()
-		cipherErrC.err = err
-		cipherErrC.Unlock()
-		return err
-	})()
-
-	ctx := context.Background()
-
-	// Start with a clean cipher configuration state
-	err := security.SetTLSCipherSuitesConfigured([]string{})
-	require.NoError(t, err)
-
-	s := serverutils.StartServerOnly(t, base.TestServerArgs{})
-	defer s.Stopper().Stop(ctx)
-	defer require.NoError(t, security.SetTLSCipherSuitesConfigured([]string{}))
-
-	// setup for db console tests
-	httpClient, err := s.GetUnauthenticatedHTTPClient()
-	require.NoError(t, err)
-	httpClient.Timeout = 2 * time.Second
-	defer httpClient.CloseIdleConnections()
-
-	secureClient, err := s.GetAuthenticatedHTTPClient(false, serverutils.SingleTenantSession)
-	require.NoError(t, err)
-	secureClient.Timeout = 2 * time.Second
-	defer secureClient.CloseIdleConnections()
-
-	urlsToTest := []string{"/_status/vars", "/index.html", "/"}
-	adminURLHTTPS := s.AdminURL().String()
-	adminURLHTTP := strings.Replace(adminURLHTTPS, "https", "http", 1)
 
 	tests := []struct {
 		name      string
@@ -179,63 +134,106 @@ func TestTLSCipherRestrict(t *testing.T) {
 		cipherErr string
 	}{
 		{name: "no cipher set", ciphers: []string{}, wantErr: false},
-		{name: "valid ciphers", ciphers: []string{"TLS_AES_256_GCM_SHA384", "TLS_AES_128_GCM_SHA256"},
+		{name: "valid ciphers", ciphers: []string{"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", "TLS_AES_256_GCM_SHA384", "TLS_AES_128_GCM_SHA256"},
 			wantErr: false},
-		{name: "invalid ciphers", ciphers: []string{"TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256"}, wantErr: true,
+		{name: "invalid ciphers", ciphers: []string{"TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256"}, wantErr: true,
 			httpsErr:  []string{"\": EOF", "connect: connection refused", "read: connection reset by peer", "http: server closed idle connection"},
 			sqlErr:    "failed to connect to `host=127.0.0.1 user=root database=`: failed to receive message (unexpected EOF)",
 			rpcErr:    "initial connection heartbeat failed: grpc:",
 			cipherErr: "^presented cipher [^ ]+ not in allowed cipher suite list$"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			// First ensure we're starting with no restrictions
-			err := security.SetTLSCipherSuitesConfigured([]string{})
+			// since the listener does not return rpc/sql/http connection errors, we
+			// need to have a separate hook to obtain and validate it.
+			type cipherErrContainer struct {
+				syncutil.Mutex
+				err net.Error
+			}
+
+			cipherErrC := &cipherErrContainer{}
+			cipherRestrictFn := security.TLSCipherRestrict
+			defer testutils.TestingHook(&security.TLSCipherRestrict, func(conn net.Conn) (err net.Error) {
+				err = cipherRestrictFn(conn)
+				cipherErrC.Lock()
+				cipherErrC.err = err
+				cipherErrC.Unlock()
+				return err
+			})()
+			ctx := context.Background()
+
+			s := serverutils.StartServerOnly(t, base.TestServerArgs{})
+			defer s.Stopper().Stop(ctx)
+
+			// set the custom test ciphers
+			err := security.SetTLSCipherSuitesConfigured(tt.ciphers)
 			require.NoError(t, err)
 
+			// setup for db console tests
+			httpClient, transport, err := s.ApplicationLayer().GetUnauthenticatedHTTPClientWithTransport()
+			require.NoError(t, err)
+			const httpTimeout = "Client.Timeout exceeded while awaiting headers"
+			transport.TLSClientConfig.MinVersion = tls.VersionTLS13
+			defer httpClient.CloseIdleConnections()
+
+			urlsToTest := []string{"/_status/vars", "/index.html", "/"}
+			adminURLHTTPS := s.AdminURL().String()
+			adminURLHTTP := strings.Replace(adminURLHTTPS, "https", "http", 1)
+
 			// Reset the error container before each test
 			cipherErrC.Lock()
 			cipherErrC.err = nil
 			cipherErrC.Unlock()
 
-			// Now set the custom test ciphers
-			err = security.SetTLSCipherSuitesConfigured(tt.ciphers)
-			require.NoError(t, err)
-			// unset the ciphers after test
-			defer func() { _ = security.SetTLSCipherSuitesConfigured([]string{}) }()
-
 			// test db console tls access for cipher restriction.
 			for _, u := range urlsToTest {
-				for _, client := range []http.Client{httpClient, secureClient} {
-					resp, err := client.Get(adminURLHTTP + u)
-					if (err == nil) == tt.wantErr {
-						var body []byte
+				for _, client := range []http.Client{httpClient} {
+					var wg sync.WaitGroup
+					wg.Add(1)
+					var body []byte
+					go func() {
+						defer wg.Done()
+						var resp *http.Response
+						resp, err = client.Get(adminURLHTTP + u)
 						if resp != nil && resp.Body != nil {
 							defer resp.Body.Close()
 							body, err = io.ReadAll(resp.Body)
 						}
-						t.Fatalf("expected wantError=%t, got err=%v, resp=%v", tt.wantErr, err, string(body))
+					}()
+					wg.Wait()
+					if (err == nil) == tt.wantErr {
+						if !(err != nil && strings.Contains(err.Error(), httpTimeout)) {
+							t.Fatalf("expected wantError=%t, got err=%v, resp=%v", tt.wantErr, err, string(body))
+						}
 					}
 					if tt.wantErr {
-						cipherErrC.Lock()
-						errVal := cipherErrC.err
-						cipherErrC.Unlock()
-						require.Regexp(t, tt.cipherErr, errVal.Error())
 						var errMatch bool
 						for idx := range tt.httpsErr {
 							errMatch = errMatch || strings.Contains(err.Error(), tt.httpsErr[idx])
 						}
 						if !errMatch {
 							t.Fatalf("the provided error %s does not match any of the expected errors: %v", err.Error(), strings.Join(tt.httpsErr, ", "))
 						}
+						cipherErrC.Lock()
+						errVal := cipherErrC.err
+						cipherErrC.Unlock()
+						require.NotNil(t, errVal)
+						require.Regexp(t, tt.cipherErr, errVal.Error())
 					}
 				}
 			}
 
 			// test pgx connection for root user with cert auth
 			pgURL, cleanup := s.PGUrl(t, serverutils.User(username.RootUser), serverutils.ClientCerts(true))
 			defer cleanup()
-			rootConn, err := pgx.Connect(ctx, pgURL.String())
+			var wg sync.WaitGroup
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				_, err = pgx.Connect(ctx, pgURL.String())
+			}()
+			wg.Wait()
+
 			if (err == nil) == tt.wantErr {
 				t.Fatalf("expected wantError=%t, got err=%v", tt.wantErr, err)
 			}
@@ -245,12 +243,15 @@ func TestTLSCipherRestrict(t *testing.T) {
 				cipherErrC.Unlock()
 				require.Regexp(t, tt.cipherErr, errVal.Error())
 				require.Equal(t, tt.sqlErr, err.Error())
-			} else {
-				require.NoError(t, rootConn.Close(ctx))
 			}
 
 			// test rpc connection for root user.
-			conn, err := s.RPCClientConnE(username.RootUserName())
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				_, err = s.RPCClientConnE(username.RootUserName())
+			}()
+			wg.Wait()
 			if (err == nil) == tt.wantErr {
 				t.Fatalf("expected wantError=%t, got err=%v", tt.wantErr, err)
 			}
@@ -260,8 +261,6 @@ func TestTLSCipherRestrict(t *testing.T) {
 				cipherErrC.Unlock()
 				require.Regexp(t, tt.cipherErr, errVal.Error())
 				require.Contains(t, err.Error(), tt.rpcErr)
-			} else {
-				require.NoError(t, conn.Close()) // nolint:grpcconnclose
 			}
 		})
 	}

pkg/server/testserver_http.go

@@ -91,6 +91,28 @@ func (ts *httpTestServer) GetUnauthenticatedHTTPClient() (http.Client, error) {
 	return client, nil
 }
 
+// GetUnauthenticatedHTTPClientWithTransport implements TestServerInterface.
+func (ts *httpTestServer) GetUnauthenticatedHTTPClientWithTransport() (
+	http.Client,
+	*http.Transport,
+	error,
+) {
+	client, err := ts.t.sqlServer.execCfg.RPCContext.GetHTTPClient()
+	if err != nil {
+		return client, nil, err
+	}
+	transport := client.Transport.(*http.Transport)
+	client.Transport = &tenantHeaderDecorator{
+		RoundTripper: client.Transport,
+		tenantName:   ts.t.tenantName,
+	}
+	client.Timeout = 2 * time.Second
+	if util.RaceEnabled {
+		client.Timeout = 30 * time.Second
+	}
+	return client, transport, nil
+}
+
 // GetAdminHTTPClient implements the TestServerInterface.
 func (ts *httpTestServer) GetAdminHTTPClient() (http.Client, error) {
 	httpClient, _, err := ts.GetAuthenticatedHTTPClientAndCookie(

pkg/sql/sqlstats/ssmemstorage/ss_mem_storage.go

@@ -92,20 +92,22 @@ type Container struct {
 func New(
 	st *cluster.Settings,
 	uniqueServerCount *SQLStatsAtomicCounters,
-	mon *mon.BytesMonitor,
+	monitor *mon.BytesMonitor,
 	appName string,
 	knobs *sqlstats.TestingKnobs,
 ) *Container {
 	s := &Container{
 		st:                st,
 		appName:           appName,
-		mon:               mon,
+		mon:               monitor,
 		knobs:             knobs,
 		uniqueServerCount: uniqueServerCount,
 	}
 
-	if mon != nil {
-		s.acc = mon.MakeConcurrentBoundAccount()
+	if monitor != nil {
+		s.acc = monitor.MakeConcurrentBoundAccount()
+	} else {
+		s.acc = mon.NewStandaloneUnlimitedConcurrentAccount()
 	}
 
 	s.mu.stmts = make(map[stmtKey]*stmtStats)

pkg/testutils/serverutils/api.go

@@ -359,6 +359,12 @@ type ApplicationLayerInterface interface {
 	// with verbose method name.
 	GetUnauthenticatedHTTPClient() (http.Client, error)
 
+	// GetUnauthenticatedHTTPClientWithTransport returns an http client and its
+	// corresponding transport configured with the client TLS config required by
+	// the TestServer's configuration. Discourages implementer from using
+	// unauthenticated http connections with verbose method name.
+	GetUnauthenticatedHTTPClientWithTransport() (http.Client, *http.Transport, error)
+
 	// GetAdminHTTPClient returns an http client which has been
 	// authenticated to access Admin API methods (via a cookie).
 	// The user has admin privileges.

pkg/util/admission/io_load_listener.go

@@ -1341,10 +1341,14 @@ func (res adjustTokensResult) SafeFormat(p redact.SafePrinter, _ rune) {
 		res.aux.perWorkTokensAux.intL0WriteLinearModel.multiplier,
 		ib(res.aux.perWorkTokensAux.intL0WriteLinearModel.constant),
 		res.l0WriteLM.multiplier, ib(res.l0WriteLM.constant))
-	p.Printf("ingested-model %.2fx+%s (smoothed %.2fx+%s) + ",
+	p.Printf("l0-ingest-model %.2fx+%s (smoothed %.2fx+%s) + ",
 		res.aux.perWorkTokensAux.intL0IngestedLinearModel.multiplier,
 		ib(res.aux.perWorkTokensAux.intL0IngestedLinearModel.constant),
 		res.l0IngestLM.multiplier, ib(res.l0IngestLM.constant))
+	p.Printf("ingest-model %.2fx+%s (smoothed %.2fx+%s) + ",
+		res.aux.perWorkTokensAux.intIngestedLinearModel.multiplier,
+		ib(res.aux.perWorkTokensAux.intIngestedLinearModel.constant),
+		res.ingestLM.multiplier, ib(res.ingestLM.constant))
 	p.Printf("write-amp-model %.2fx+%s (smoothed %.2fx+%s) + ",
 		res.aux.perWorkTokensAux.intWriteAmpLinearModel.multiplier,
 		ib(res.aux.perWorkTokensAux.intWriteAmpLinearModel.constant),

pkg/util/admission/testdata/format_adjust_tokens_stats.txt

@@ -1,6 +1,6 @@
 echo
 ----
 zero:
-compaction score 0.000 (0 ssts, 0 sub-levels), L0 growth 0 B (write 0 B (ignored 0 B) ingest 0 B (ignored 0 B)): requests 0 (0 bypassed) with 0 B acc-write (0 B bypassed) + 0 B acc-ingest (0 B bypassed) + 0 B adjusted-LSM-writes + 0 B adjusted-disk-writes + write-model 0.00x+0 B (smoothed 0.00x+0 B) + ingested-model 0.00x+0 B (smoothed 0.00x+0 B) + write-amp-model 0.00x+0 B (smoothed 0.00x+0 B) + at-admission-tokens 0 B, compacted 0 B [≈0 B], flushed 0 B [≈0 B] (mult 1.00); admitting all (used total: 0 B elastic 0 B); write stalls 12
+compaction score 0.000 (0 ssts, 0 sub-levels), L0 growth 0 B (write 0 B (ignored 0 B) ingest 0 B (ignored 0 B)): requests 0 (0 bypassed) with 0 B acc-write (0 B bypassed) + 0 B acc-ingest (0 B bypassed) + 0 B adjusted-LSM-writes + 0 B adjusted-disk-writes + write-model 0.00x+0 B (smoothed 0.00x+0 B) + l0-ingest-model 0.00x+0 B (smoothed 0.00x+0 B) + ingest-model 0.00x+0 B (smoothed 0.00x+0 B) + write-amp-model 0.00x+0 B (smoothed 0.00x+0 B) + at-admission-tokens 0 B, compacted 0 B [≈0 B], flushed 0 B [≈0 B] (mult 1.00); admitting all (used total: 0 B elastic 0 B); write stalls 12
 real-numbers:
-compaction score 2.700[L0-overload] (195 ssts, 27 sub-levels), L0 growth 577 MiB (write 0 B (ignored 0 B) ingest 0 B (ignored 0 B)): requests 0 (0 bypassed) with 0 B acc-write (0 B bypassed) + 0 B acc-ingest (0 B bypassed) + 0 B adjusted-LSM-writes + 0 B adjusted-disk-writes + write-model 0.00x+0 B (smoothed 0.00x+0 B) + ingested-model 0.00x+0 B (smoothed 0.00x+0 B) + write-amp-model 0.00x+0 B (smoothed 0.00x+0 B) + at-admission-tokens 0 B, compacted 77 MiB [≈62 MiB], flushed 0 B [≈0 B] (mult 1.00); admitting 116 MiB (rate 7.7 MiB/s) (elastic 1 B rate 0 B/s) due to L0 growth (used total: 0 B elastic 0 B); write stalls 2
+compaction score 2.700[L0-overload] (195 ssts, 27 sub-levels), L0 growth 577 MiB (write 0 B (ignored 0 B) ingest 0 B (ignored 0 B)): requests 0 (0 bypassed) with 0 B acc-write (0 B bypassed) + 0 B acc-ingest (0 B bypassed) + 0 B adjusted-LSM-writes + 0 B adjusted-disk-writes + write-model 0.00x+0 B (smoothed 0.00x+0 B) + l0-ingest-model 0.00x+0 B (smoothed 0.00x+0 B) + ingest-model 0.00x+0 B (smoothed 0.00x+0 B) + write-amp-model 0.00x+0 B (smoothed 0.00x+0 B) + at-admission-tokens 0 B, compacted 77 MiB [≈62 MiB], flushed 0 B [≈0 B] (mult 1.00); admitting 116 MiB (rate 7.7 MiB/s) (elastic 1 B rate 0 B/s) due to L0 growth (used total: 0 B elastic 0 B); write stalls 2