Merge cockroachdb#107572 cockroachdb#107993 cockroachdb#108013

107572: changefeedccl: warn when using https sink urls r=[miretskiy] a=HonoreDB

Changefeeds previously used http as their URI to align with backups.
Backups no longer support http, and for changefeeds http is ambiguous,
leading users to create cloudstorage sinks when they meant webhook sinks.
This PR changes the scheme to file-http. Webhooks still use webhook-http.

Fixes cockroachdb#98719.

Release note (enterprise change): Changefeeds that create files over an http connection may now be specified via `INTO 'file-https://'` to disambiguate with `webhook-https`.

107993: sql: Deflake TestTxnObeysTableModificationTime r=Xiang-Gu a=Xiang-Gu

This commit refactors and simplifies this test, and a local execution with `--race --stress` succeeded for at least 10 minutes.

Fixes cockroachdb#107736
Fixes cockroachdb#107159
Epic: None
Release note: None

108013: c2c: reduce TestStreamingAutoReplan runtime r=stevendanna a=msbutler

In a suprising twist, cockroachdb#106853 deflaked this test, but also increased the runtime to 10 minutes. To understand why, consider that the default value of the stream_replication.replan_flow_frequency setting is 10 minutes, and if the user changes the cluster setting, the frequency will only take effect on the next replanning check. So, if a stream begins at t0, and the user changes the setting at t1, the replanning frequency will actually change 10 minutes after t0.

In cockroachdb#105853  patch reduced stream_replication.replan_flow_frequency setting to 500 ms _after_ the stream began, which caused the next replanninng check to occur a full 10 minutes after the stream started! This patch changes this cluster setting _before_ the stream begins, reducing the runtime of this test to a speedy 10 seconds.

Epic: none

Release note: none

Co-authored-by: Aaron Zinger <[email protected]>
Co-authored-by: Xiang Gu <[email protected]>
Co-authored-by: Michael Butler <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -904,6 +904,25 @@ func validateSink(
 	if err != nil {
 		return err
 	}
+	u, err := url.Parse(details.SinkURI)
+	if err != nil {
+		return err
+	}
+
+	ambiguousSchemes := map[string][2]string{
+		changefeedbase.DeprecatedSinkSchemeHTTP:  {changefeedbase.SinkSchemeCloudStorageHTTP, changefeedbase.SinkSchemeWebhookHTTP},
+		changefeedbase.DeprecatedSinkSchemeHTTPS: {changefeedbase.SinkSchemeCloudStorageHTTPS, changefeedbase.SinkSchemeWebhookHTTPS},
+	}
+
+	if disambiguations, isAmbiguous := ambiguousSchemes[u.Scheme]; isAmbiguous {
+		p.BufferClientNotice(ctx, pgnotice.Newf(
+			`Interpreting deprecated URI scheme %s as %s. For webhook semantics, use %s.`,
+			u.Scheme,
+			disambiguations[0],
+			disambiguations[1],
+		))
+	}
+
 	var nilOracle timestampLowerBoundOracle
 	canarySink, err := getAndDialSink(ctx, &p.ExecCfg().DistSQLSrv.ServerConfig, details,
 		nilOracle, p.User(), jobID, sli)

pkg/ccl/changefeedccl/changefeed_test.go

@@ -4828,7 +4828,7 @@ func TestChangefeedErrors(t *testing.T) {
 		`CREATE CHANGEFEED FOR foo INTO $1`, `webhook-https://fake-host?ca_cert=Zm9v`,
 	)
 	sqlDB.ExpectErr(
-		t, `sink requires https`,
+		t, `sink requires webhook-https`,
 		`CREATE CHANGEFEED FOR foo INTO $1`, `webhook-http://fake-host`,
 	)
 	sqlDB.ExpectErr(

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -162,6 +162,11 @@ const (
 	DeprecatedSinkSchemeCloudStorageNodelocal = `experimental-nodelocal`
 	DeprecatedSinkSchemeCloudStorageS3        = `experimental-s3`
 
+	// DeprecatedSinkSchemeHTTP is interpreted as cloudstorage over HTTP PUT.
+	DeprecatedSinkSchemeHTTP = `http`
+	// DeprecatedSinkSchemeHTTPS is interpreted as cloudstorage over HTTPS PUT.
+	DeprecatedSinkSchemeHTTPS = `https`
+
 	// OptKafkaSinkConfig is a JSON configuration for kafka sink (kafkaSinkConfig).
 	OptKafkaSinkConfig   = `kafka_sink_config`
 	OptPubsubSinkConfig  = `pubsub_sink_config`
@@ -186,13 +191,11 @@ const (
 	SinkParamTopicName              = `topic_name`
 	SinkSchemeCloudStorageAzure     = `azure`
 	SinkSchemeCloudStorageGCS       = `gs`
-	SinkSchemeCloudStorageHTTP      = `http`
-	SinkSchemeCloudStorageHTTPS     = `https`
+	SinkSchemeCloudStorageHTTP      = `file-http`
+	SinkSchemeCloudStorageHTTPS     = `file-https`
 	SinkSchemeCloudStorageNodelocal = `nodelocal`
 	SinkSchemeCloudStorageS3        = `s3`
 	SinkSchemeExperimentalSQL       = `experimental-sql`
-	SinkSchemeHTTP                  = `http`
-	SinkSchemeHTTPS                 = `https`
 	SinkSchemeKafka                 = `kafka`
 	SinkSchemeNull                  = `null`
 	SinkSchemeWebhookHTTP           = `webhook-http`

pkg/ccl/changefeedccl/sink_cloudstorage.go

@@ -49,6 +49,10 @@ func isCloudStorageSink(u *url.URL) bool {
 		changefeedbase.SinkSchemeCloudStorageNodelocal, changefeedbase.SinkSchemeCloudStorageHTTP,
 		changefeedbase.SinkSchemeCloudStorageHTTPS, changefeedbase.SinkSchemeCloudStorageAzure:
 		return true
+	// During the deprecation period, we need to keep parsing these as cloudstorage for backwards
+	// compatibility. Afterwards we'll either remove them or move them to webhook.
+	case changefeedbase.DeprecatedSinkSchemeHTTP, changefeedbase.DeprecatedSinkSchemeHTTPS:
+		return true
 	default:
 		return false
 	}
@@ -375,6 +379,7 @@ func makeCloudStorageSink(
 		}
 	}
 	u.Scheme = strings.TrimPrefix(u.Scheme, `experimental-`)
+	u.Scheme = strings.TrimPrefix(u.Scheme, `file-`)
 
 	sinkID := atomic.AddInt64(&cloudStorageSinkIDAtomic, 1)
 	sessID, err := generateChangefeedSessionID()

pkg/ccl/changefeedccl/sink_external_connection.go

@@ -92,6 +92,8 @@ var supportedExternalConnectionTypes = map[string]connectionpb.ConnectionProvide
 	GcpScheme:                                      connectionpb.ConnectionProvider_gcpubsub,
 	changefeedbase.SinkSchemeCloudStorageHTTP:      connectionpb.ConnectionProvider_http,
 	changefeedbase.SinkSchemeCloudStorageHTTPS:     connectionpb.ConnectionProvider_https,
+	changefeedbase.DeprecatedSinkSchemeHTTP:        connectionpb.ConnectionProvider_http,
+	changefeedbase.DeprecatedSinkSchemeHTTPS:       connectionpb.ConnectionProvider_https,
 	changefeedbase.SinkSchemeCloudStorageNodelocal: connectionpb.ConnectionProvider_nodelocal,
 	changefeedbase.SinkSchemeCloudStorageS3:        connectionpb.ConnectionProvider_s3,
 	changefeedbase.SinkSchemeKafka:                 connectionpb.ConnectionProvider_kafka,

pkg/ccl/changefeedccl/sink_webhook.go

@@ -233,7 +233,7 @@ func makeDeprecatedWebhookSink(
 	mb metricsRecorderBuilder,
 ) (Sink, error) {
 	if u.Scheme != changefeedbase.SinkSchemeWebhookHTTPS {
-		return nil, errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeHTTPS)
+		return nil, errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeWebhookHTTPS)
 	}
 	u.Scheme = strings.TrimPrefix(u.Scheme, `webhook-`)
 

pkg/ccl/changefeedccl/sink_webhook_v2.go

@@ -229,7 +229,7 @@ func validateWebhookOpts(
 	u sinkURL, encodingOpts changefeedbase.EncodingOptions, opts changefeedbase.WebhookSinkOptions,
 ) error {
 	if u.Scheme != changefeedbase.SinkSchemeWebhookHTTPS {
-		return errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeHTTPS)
+		return errors.Errorf(`this sink requires %s`, changefeedbase.SinkSchemeWebhookHTTPS)
 	}
 
 	switch encodingOpts.Format {

pkg/ccl/streamingccl/streamingest/replication_stream_e2e_test.go

@@ -729,6 +729,7 @@ func TestStreamingAutoReplan(t *testing.T) {
 	defer cleanup()
 	// Don't allow for replanning until the new nodes and scattered table have been created.
 	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_threshold", 0)
+	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_frequency", time.Millisecond*500)
 
 	// Begin the job on a single source node.
 	producerJobID, ingestionJobID := c.StartStreamReplication(ctx)
@@ -746,7 +747,6 @@ func TestStreamingAutoReplan(t *testing.T) {
 
 	// Configure the ingestion job to replan eagerly.
 	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_threshold", 0.1)
-	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_frequency", time.Millisecond*500)
 
 	// The ingestion job should eventually retry because it detects new nodes to add to the plan.
 	require.Error(t, <-retryErrorChan, sql.ErrPlanChanged)

pkg/sql/catalog/lease/BUILD.bazel

@@ -77,6 +77,7 @@ go_test(
     embed = [":lease"],
     deps = [
         "//pkg/base",
+        "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",
@@ -105,7 +106,6 @@ go_test(
         "//pkg/sql/rowenc/keyside",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondata",
-        "//pkg/sql/sqltestutils",
         "//pkg/sql/tests",
         "//pkg/sql/types",
         "//pkg/storage",
@@ -114,7 +114,6 @@ go_test(
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
-        "//pkg/util",
         "//pkg/util/admission",
         "//pkg/util/encoding",
         "//pkg/util/hlc",