Merge #146813 #149904 #150191

146813: changefeedccl: support for constant headers for webhook and kafka sinks r=andyyang890,KeithCh a=asg0451

This adds support for constant headers for webhook
and kafka sinks, to be specified in the CREATE
CHANGEFEED statement as a JSON object.

Part of: #142273
Epic: CRDB-48880

Release note (general change): The CREATE
CHANGEFEED statement now supports the `headers`
option, which can be used to specify constant
headers for webhook and kafka sinks. This can be
used to add headers to all messages sent to the
sink.

149904: asim: final port over from mma prototype for asim  r=tbg a=wenyihu6

**asim: pass false to ReplicaChangeDelayFn for AllocationTransferLeaseOp**

Previously, we incorrectly passed true to ReplicaChangeDelayFn for
AllocationTransferLeaseOp, causing it to treat the operation like a replica
addition and extend the delay based on range size and snapshot rate. Since lease
transfers are expected to be much faster, this was unintended. This commit fixes
the issue by applying only the base delay (ReplicaChangeBaseDelay) to
AllocationTransferLeaseOp, which defaults to 100ms.

Epic: none
Release note: none

---

**asim: correctly update queue.next in pushReplicateChange**

Previously, we did not assign the next field of the queues correctly because we
passed the base queue by value into pushReplicateChange, which caused updates to
the next field to affect only the copy, not the original queue. This commit
fixes the issue by returning the correct next value and updating the queue
directly.

Epic: none
Release note: none

---

**asim: pass simulator.Replica instead to pushReplicateChange**

Previously, we passed state.Range to pushReplicateChange. This commit changes it
to pass the simulator.Replica directly, as future commits will require access to
additional fields such as Desc and RangeUsageInfo.

Epic: none
Release note: none

---

**asim: pass state changer directly to pushReplicateChange**

Previously, we passed base queue to pushReplicateChange, but only its state
changer was being used. This commit updates pushReplicateChange to only take in
the state changer.

Epic: none
Release note: none

---

**asim: integrate allocatorsync with lease and replicate queue properly**

This commit integrates lease and replicate queue changes with the mma allocator
properly by calling allocatorSync.PreApply* and allocatorSync.PostApply,
completing the allocator integration. Some data-driven test outputs changed
because allocatorSync.PostApply now updates store pool after operations (such as
UpdateLocalStoresAfterLeaseTransfer), which was missing previously in asim. It’s
unclear if this was an oversight or previously seemed as unnecessary. More
reading is needed.

Epic: none
Release note: none

---

**asim: pass queue names for logging**

This commit passes the name of the queue making changes to pushReplicateChange
to improve logging.

Epic: none
Release note: none

---

**asim: remove setting from store**

This commit removes the setting field from store struct as it is not used
anywhere.

Epic: none
Release note: none

---

**asim: add rebalance_objective and delay for rebalance_mode**

This commit adds rebalance_objective as a setting and allows changes to
rebalance_mode to be scheduled as a delayed simulation event. This enables
testing transitions between mma enabled and disabled mode.

Epic: none
Release note: none

---

**asim: check for LBRebalancingMultiMetric in ShouldRebalanceStore**

This commit updates the old store rebalancer's ShouldRebalanceStore to check the
LBRebalancingMultiMetric cluster setting. asim still invokes this method when
checking changes from the old store rebalancer.

Epic: none
Release note: none




150191: roachtest: fix check for npm being installed r=rafiss a=rafiss

We have seen that this check can return a 0 exit code even when the npm command doesn't exist. Using && should fix this by short-circuiting as soon as a non-zero exit code is received.

fixes #149536
fixes #149532

Release note: None

Co-authored-by: Miles Frankel <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/cdctest/mock_webhook_sink.go

@@ -17,6 +17,11 @@ import (
 	"github.com/klauspost/compress/zstd"
 )
 
+type RowWithHeaders struct {
+	Row     string
+	Headers http.Header
+}
+
 // MockWebhookSink is the Webhook sink used in tests.
 type MockWebhookSink struct {
 	basicAuth          bool
@@ -28,7 +33,7 @@ type MockWebhookSink struct {
 		responseBodies   map[int][]byte
 		statusCodes      []int
 		statusCodesIndex int
-		rows             []string
+		rows             []RowWithHeaders
 		lastHeaders      http.Header
 		notify           chan struct{}
 	}
@@ -160,7 +165,7 @@ func (s *MockWebhookSink) Latest() string {
 		return ""
 	}
 	latest := s.mu.rows[len(s.mu.rows)-1]
-	return latest
+	return latest.Row
 }
 
 // Pop deletes and returns the oldest message from MockWebhookSink.
@@ -170,11 +175,23 @@ func (s *MockWebhookSink) Pop() string {
 	if len(s.mu.rows) > 0 {
 		oldest := s.mu.rows[0]
 		s.mu.rows = s.mu.rows[1:]
-		return oldest
+		return oldest.Row
 	}
 	return ""
 }
 
+// PopWithHeaders deletes and returns the oldest message from MockWebhookSink along with its headers.
+func (s *MockWebhookSink) PopWithHeaders() RowWithHeaders {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.mu.rows) > 0 {
+		oldest := s.mu.rows[0]
+		s.mu.rows = s.mu.rows[1:]
+		return oldest
+	}
+	return RowWithHeaders{}
+}
+
 // NotifyMessage arranges for channel to be closed when message arrives.
 func (s *MockWebhookSink) NotifyMessage() chan struct{} {
 	c := make(chan struct{})
@@ -298,7 +315,7 @@ func (s *MockWebhookSink) publish(hw http.ResponseWriter, hr *http.Request) erro
 	}
 
 	s.mu.Lock()
-	s.mu.rows = append(s.mu.rows, string(row))
+	s.mu.rows = append(s.mu.rows, RowWithHeaders{Row: string(row), Headers: hr.Header})
 	if s.mu.notify != nil {
 		close(s.mu.notify)
 		s.mu.notify = nil

pkg/ccl/changefeedccl/cdctest/testfeed.go

@@ -7,6 +7,8 @@ package cdctest
 
 import (
 	"fmt"
+	"slices"
+	"strings"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -39,7 +41,10 @@ func (h Headers) String() string {
 		return ""
 	}
 	s := "("
-	for _, v := range h {
+	sorted := slices.SortedFunc(slices.Values(h), func(a, b Header) int {
+		return strings.Compare(a.K, b.K)
+	})
+	for _, v := range sorted {
 		s += fmt.Sprintf("%s: %s, ", v.K, v.V)
 	}
 	return s[:len(s)-2] + ")"

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11881,6 +11881,91 @@ func TestCloudstorageParallelCompression(t *testing.T) {
 	})
 }
 
+func TestChangefeedExtraHeaders(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		// Headers are not supported in the v1 kafka sink.
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.new_kafka_sink.enabled = true`)
+
+		sqlDB.Exec(t, `CREATE TABLE foo (key INT PRIMARY KEY);`)
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1);`)
+
+		cases := []struct {
+			name        string
+			headersArg  string
+			wantHeaders cdctest.Headers
+			expectErr   bool
+		}{
+			{
+				name:        "single header",
+				headersArg:  `{"X-Someheader": "somevalue"}`,
+				wantHeaders: cdctest.Headers{{K: "X-Someheader", V: []byte("somevalue")}},
+			},
+			{
+				name:       "multiple headers",
+				headersArg: `{"X-Someheader": "somevalue", "X-Someotherheader": "someothervalue"}`,
+				wantHeaders: cdctest.Headers{
+					{K: "X-Someheader", V: []byte("somevalue")},
+					{K: "X-Someotherheader", V: []byte("someothervalue")},
+				},
+			},
+			{
+				name:       "inappropriate json",
+				headersArg: `4`,
+				expectErr:  true,
+			},
+			{
+				name:       "also inappropriate json",
+				headersArg: `["X-Someheader", "somevalue"]`,
+				expectErr:  true,
+			},
+			{
+				name:       "invalid json",
+				headersArg: `xxxx`,
+				expectErr:  true,
+			},
+		}
+
+		for _, c := range cases {
+			feed, err := f.Feed(fmt.Sprintf(`CREATE CHANGEFEED FOR foo WITH extra_headers='%s'`, c.headersArg))
+			if c.expectErr {
+				require.Error(t, err)
+				continue
+			} else {
+				require.NoError(t, err)
+			}
+
+			assertPayloads(t, feed, []string{
+				fmt.Sprintf(`foo: [1]%s->{"after": {"key": 1}}`, c.wantHeaders.String()),
+			})
+			closeFeed(t, feed)
+		}
+	}
+
+	cdcTest(t, testFn, feedTestRestrictSinks("kafka", "webhook"))
+}
+func TestChangefeedAdditionalHeadersDoesntWorkWithV1KafkaSink(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.new_kafka_sink.enabled = false`)
+
+		sqlDB.Exec(t, `CREATE TABLE foo (key INT PRIMARY KEY);`)
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1);`)
+
+		_, err := f.Feed(`CREATE CHANGEFEED FOR foo WITH extra_headers='{"X-Someheader": "somevalue"}'`)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "headers are not supported for the v1 kafka sink")
+	}
+
+	cdcTest(t, testFn, feedTestForceSink("kafka"))
+}
+
 func TestDatabaseLevelChangefeed(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/changefeedbase/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "//pkg/sql/types",
         "//pkg/util",
         "//pkg/util/iterutil",
+        "//pkg/util/json",
         "//pkg/util/metamorphic",
         "@com_github_cockroachdb_errors//:errors",
     ],

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/cockroach/pkg/util/json"
 	"github.com/cockroachdb/errors"
 )
 
@@ -117,6 +118,7 @@ const (
 	// TODO(#142273): look into whether we want to add headers to pub/sub, and other
 	// sinks as well (eg cloudstorage, webhook, ..). Currently it's kafka-only.
 	OptHeadersJSONColumnName = `headers_json_column_name`
+	OptExtraHeaders          = `extra_headers`
 
 	OptVirtualColumnsOmitted VirtualColumnVisibility = `omitted`
 	OptVirtualColumnsNull    VirtualColumnVisibility = `null`
@@ -408,6 +410,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptEncodeJSONValueNullAsObject:        flagOption,
 	OptEnrichedProperties:                 csv(string(EnrichedPropertySource), string(EnrichedPropertySchema)),
 	OptHeadersJSONColumnName:              stringOption,
+	OptExtraHeaders:                       jsonOption,
 }
 
 // CommonOptions is options common to all sinks
@@ -428,13 +431,13 @@ var CommonOptions = makeStringSet(OptCursor, OptEndTime, OptEnvelope,
 var SQLValidOptions map[string]struct{} = nil
 
 // KafkaValidOptions is options exclusive to Kafka sink
-var KafkaValidOptions = makeStringSet(OptAvroSchemaPrefix, OptConfluentSchemaRegistry, OptKafkaSinkConfig, OptHeadersJSONColumnName)
+var KafkaValidOptions = makeStringSet(OptAvroSchemaPrefix, OptConfluentSchemaRegistry, OptKafkaSinkConfig, OptHeadersJSONColumnName, OptExtraHeaders)
 
 // CloudStorageValidOptions is options exclusive to cloud storage sink
 var CloudStorageValidOptions = makeStringSet(OptCompression)
 
 // WebhookValidOptions is options exclusive to webhook sink
-var WebhookValidOptions = makeStringSet(OptWebhookAuthHeader, OptWebhookClientTimeout, OptWebhookSinkConfig, OptCompression)
+var WebhookValidOptions = makeStringSet(OptWebhookAuthHeader, OptWebhookClientTimeout, OptWebhookSinkConfig, OptCompression, OptExtraHeaders)
 
 // PubsubValidOptions is options exclusive to pubsub sink
 var PubsubValidOptions = makeStringSet(OptPubsubSinkConfig)
@@ -477,6 +480,7 @@ func RedactUserFromURI(uri string) (string, error) {
 // RedactedOptions are options whose values should be replaced with "redacted" in job descriptions and errors.
 var RedactedOptions = map[string]redactionFunc{
 	OptWebhookAuthHeader:       redactSimple,
+	OptExtraHeaders:            redactSimple,
 	SinkParamClientKey:         redactSimple,
 	OptConfluentSchemaRegistry: RedactUserFromURI,
 }
@@ -1032,6 +1036,7 @@ func (s StatementOptions) GetFilters() Filters {
 type WebhookSinkOptions struct {
 	JSONConfig    SinkSpecificJSONConfig
 	AuthHeader    string
+	ExtraHeaders  map[string]string
 	ClientTimeout *time.Duration
 	Compression   string
 }
@@ -1049,13 +1054,66 @@ func (s StatementOptions) GetWebhookSinkOptions() (WebhookSinkOptions, error) {
 		return o, err
 	}
 	o.ClientTimeout = timeout
+
+	headersMap, err := parseHeaders[string](s.m[OptExtraHeaders])
+	if err != nil {
+		return o, err
+	}
+	o.ExtraHeaders = headersMap
 	return o, nil
 }
 
-// GetKafkaConfigJSON returns arbitrary json to be interpreted
-// by the kafka sink.
-func (s StatementOptions) GetKafkaConfigJSON() SinkSpecificJSONConfig {
-	return s.getJSONValue(OptKafkaSinkConfig)
+func parseHeaders[S interface{ string | []byte }](headers string) (map[string]S, error) {
+	if headers == "" {
+		return nil, nil
+	}
+	headersJ, err := json.ParseJSON(headers)
+	if err != nil {
+		return nil, errors.Wrap(err, "parsing headers")
+	}
+	it, err := headersJ.ObjectIter()
+	if err != nil {
+		return nil, errors.Wrap(err, "parsing headers as object")
+	}
+	if it == nil {
+		return nil, errors.Newf("headers is not a JSON object: %s", headers)
+	}
+	headersMap := make(map[string]S, headersJ.Len())
+	for it.Next() {
+		k := it.Key()
+		v := it.Value()
+		s, err := v.AsText()
+		if err != nil {
+			return nil, errors.Wrap(err, "parsing header value as text")
+		}
+		if s == nil {
+			continue
+		}
+		headersMap[k] = S(*s)
+	}
+	return headersMap, nil
+}
+
+type KafkaSinkOptions struct {
+	// JSONConfig is arbitrary json to be interpreted
+	// by the kafka sink.
+	JSONConfig SinkSpecificJSONConfig
+
+	// Headers is a map of header names to values.
+	Headers map[string][]byte
+}
+
+func (s StatementOptions) GetKafkaSinkOptions() (KafkaSinkOptions, error) {
+	headersMap, err := parseHeaders[[]byte](s.m[OptExtraHeaders])
+	if err != nil {
+		return KafkaSinkOptions{}, err
+	}
+
+	o := KafkaSinkOptions{
+		JSONConfig: s.getJSONValue(OptKafkaSinkConfig),
+		Headers:    headersMap,
+	}
+	return o, nil
 }
 
 // GetPubsubConfigJSON returns arbitrary json to be interpreted

pkg/ccl/changefeedccl/sink.go

@@ -249,20 +249,28 @@ func getSink(
 			return makeNullSink(&changefeedbase.SinkURL{URL: u}, metricsBuilder(nullIsAccounted))
 		case isKafkaSink(u):
 			return validateOptionsAndMakeSink(changefeedbase.KafkaValidOptions, func() (Sink, error) {
+				sinkOpts, err := opts.GetKafkaSinkOptions()
+				if err != nil {
+					return nil, err
+				}
 				if KafkaV2Enabled.Get(&serverCfg.Settings.SV) {
-					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
+					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), sinkOpts,
 						numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 						serverCfg.Settings, metricsBuilder, kafkaSinkV2Knobs{})
 				} else {
-					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(), serverCfg.Settings, metricsBuilder)
+					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), sinkOpts, serverCfg.Settings, metricsBuilder)
 				}
 			})
 		case isPulsarSink(u):
 			var testingKnobs *TestingKnobs
 			if knobs, ok := serverCfg.TestingKnobs.Changefeed.(*TestingKnobs); ok {
 				testingKnobs = knobs
 			}
-			return makePulsarSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
+			sinkOpts, err := opts.GetKafkaSinkOptions()
+			if err != nil {
+				return nil, err
+			}
+			return makePulsarSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, AllTargets(feedCfg), sinkOpts.JSONConfig,
 				serverCfg.Settings, metricsBuilder, testingKnobs)
 		case isWebhookSink(u):
 			webhookOpts, err := opts.GetWebhookSinkOptions()

pkg/ccl/changefeedccl/sink_kafka.go

@@ -1200,7 +1200,7 @@ func makeKafkaSink(
 	ctx context.Context,
 	u *changefeedbase.SinkURL,
 	targets changefeedbase.Targets,
-	jsonStr changefeedbase.SinkSpecificJSONConfig,
+	sinkOpts changefeedbase.KafkaSinkOptions,
 	settings *cluster.Settings,
 	mb metricsRecorderBuilder,
 ) (Sink, error) {
@@ -1210,6 +1210,12 @@ func makeKafkaSink(
 		return nil, errors.Errorf(`%s is not yet supported`, changefeedbase.SinkParamSchemaTopic)
 	}
 
+	jsonStr := sinkOpts.JSONConfig
+	if len(sinkOpts.Headers) > 0 {
+		return nil, errors.Newf("headers are not supported for the v1 kafka sink;"+
+			" use the v2 sink instead via the `%s` cluster setting", KafkaV2Enabled.Name())
+	}
+
 	m := mb(requiresResourceAccounting)
 	config, err := buildKafkaConfig(ctx, u, jsonStr, m.getKafkaThrottlingMetrics(settings), m.netMetrics())
 	if err != nil {

pkg/ccl/changefeedccl/sink_kafka_connection_test.go

@@ -403,7 +403,7 @@ func TestAzureKafkaDefaults(t *testing.T) {
 
 	assertExpectedKgoOpts := func(exp expectation, opts []kgo.Opt) {
 		sinkClient, err := newKafkaSinkClientV2(ctx, opts, sinkBatchConfig{},
-			"", cluster.MakeTestingClusterSettings(), kafkaSinkV2Knobs{}, nilMetricsRecorderBuilder, nil)
+			"", cluster.MakeTestingClusterSettings(), kafkaSinkV2Knobs{}, nilMetricsRecorderBuilder, nil, nil)
 		require.NoError(t, err)
 		defer func() { require.NoError(t, sinkClient.Close()) }()
 		client := sinkClient.client.(*kgo.Client)