spanconfig: add Record constructor and validation

This change adds a constructor method that returns
a new `spanconfig.Record` and conditionally performs
some validation if the target is a SystemTarget.

Informs: cockroachdb#73727

Release note: None

Release justification: low-risk updates to new functionality

pkg/ccl/migrationccl/migrationsccl/seed_tenant_span_configs_external_test.go

@@ -75,7 +75,7 @@ func TestPreSeedSpanConfigsWrittenWhenActive(t *testing.T) {
 		})
 		require.NoError(t, err)
 		require.Len(t, records, 1)
-		require.Equal(t, records[0].Target.GetSpan(), tenantSeedSpan)
+		require.Equal(t, records[0].GetTarget().GetSpan(), tenantSeedSpan)
 	}
 }
 
@@ -144,7 +144,7 @@ func TestSeedTenantSpanConfigs(t *testing.T) {
 		})
 		require.NoError(t, err)
 		require.Len(t, records, 1)
-		require.Equal(t, records[0].Target.GetSpan(), tenantSeedSpan)
+		require.Equal(t, records[0].GetTarget().GetSpan(), tenantSeedSpan)
 	}
 }
 
@@ -200,7 +200,7 @@ func TestSeedTenantSpanConfigsWithExistingEntry(t *testing.T) {
 		})
 		require.NoError(t, err)
 		require.Len(t, records, 1)
-		require.Equal(t, records[0].Target.GetSpan(), tenantSeedSpan)
+		require.Equal(t, records[0].GetTarget().GetSpan(), tenantSeedSpan)
 	}
 
 	// Ensure the cluster version bump goes through successfully.
@@ -215,6 +215,6 @@ func TestSeedTenantSpanConfigsWithExistingEntry(t *testing.T) {
 		})
 		require.NoError(t, err)
 		require.Len(t, records, 1)
-		require.Equal(t, records[0].Target.GetSpan(), tenantSeedSpan)
+		require.Equal(t, records[0].GetTarget().GetSpan(), tenantSeedSpan)
 	}
 }

pkg/ccl/spanconfigccl/spanconfigreconcilerccl/datadriven_test.go

@@ -208,18 +208,18 @@ func TestDataDriven(t *testing.T) {
 				)
 				require.NoError(t, err)
 				sort.Slice(records, func(i, j int) bool {
-					return records[i].Target.Less(records[j].Target)
+					return records[i].GetTarget().Less(records[j].GetTarget())
 				})
 
 				lines := make([]string, len(records))
 				for i, record := range records {
 					switch {
-					case record.Target.IsSpanTarget():
-						lines[i] = fmt.Sprintf("%-42s %s", record.Target.GetSpan(),
-							spanconfigtestutils.PrintSpanConfigDiffedAgainstDefaults(record.Config))
-					case record.Target.IsSystemTarget():
-						lines[i] = fmt.Sprintf("%-42s %s", record.Target.GetSystemTarget(),
-							spanconfigtestutils.PrintSystemSpanConfigDiffedAgainstDefault(record.Config))
+					case record.GetTarget().IsSpanTarget():
+						lines[i] = fmt.Sprintf("%-42s %s", record.GetTarget().GetSpan(),
+							spanconfigtestutils.PrintSpanConfigDiffedAgainstDefaults(record.GetConfig()))
+					case record.GetTarget().IsSystemTarget():
+						lines[i] = fmt.Sprintf("%-42s %s", record.GetTarget().GetSystemTarget(),
+							spanconfigtestutils.PrintSystemSpanConfigDiffedAgainstDefault(record.GetConfig()))
 					default:
 						panic("unsupported target type")
 					}

pkg/ccl/spanconfigccl/spanconfigsqltranslatorccl/datadriven_test.go

@@ -161,18 +161,18 @@ func TestDataDriven(t *testing.T) {
 				records, _, err := sqlTranslator.Translate(ctx, descIDs, generateSystemSpanConfigs)
 				require.NoError(t, err)
 				sort.Slice(records, func(i, j int) bool {
-					return records[i].Target.Less(records[j].Target)
+					return records[i].GetTarget().Less(records[j].GetTarget())
 				})
 
 				var output strings.Builder
 				for _, record := range records {
 					switch {
-					case record.Target.IsSpanTarget():
-						output.WriteString(fmt.Sprintf("%-42s %s\n", record.Target.GetSpan(),
-							spanconfigtestutils.PrintSpanConfigDiffedAgainstDefaults(record.Config)))
-					case record.Target.IsSystemTarget():
-						output.WriteString(fmt.Sprintf("%-42s %s\n", record.Target.GetSystemTarget(),
-							spanconfigtestutils.PrintSystemSpanConfigDiffedAgainstDefault(record.Config)))
+					case record.GetTarget().IsSpanTarget():
+						output.WriteString(fmt.Sprintf("%-42s %s\n", record.GetTarget().GetSpan(),
+							spanconfigtestutils.PrintSpanConfigDiffedAgainstDefaults(record.GetConfig())))
+					case record.GetTarget().IsSystemTarget():
+						output.WriteString(fmt.Sprintf("%-42s %s\n", record.GetTarget().GetSystemTarget(),
+							spanconfigtestutils.PrintSystemSpanConfigDiffedAgainstDefault(record.GetConfig())))
 					default:
 						panic("unsupported target type")
 					}
@@ -185,12 +185,12 @@ func TestDataDriven(t *testing.T) {
 				require.NoError(t, err)
 
 				sort.Slice(records, func(i, j int) bool {
-					return records[i].Target.Less(records[j].Target)
+					return records[i].GetTarget().Less(records[j].GetTarget())
 				})
 				var output strings.Builder
 				for _, record := range records {
-					output.WriteString(fmt.Sprintf("%-42s %s\n", record.Target.GetSpan(),
-						spanconfigtestutils.PrintSpanConfigDiffedAgainstDefaults(record.Config)))
+					output.WriteString(fmt.Sprintf("%-42s %s\n", record.GetTarget().GetSpan(),
+						spanconfigtestutils.PrintSpanConfigDiffedAgainstDefaults(record.GetConfig())))
 				}
 				return output.String()
 

pkg/kv/kvserver/client_spanconfigs_test.go

@@ -70,15 +70,18 @@ func TestSpanConfigUpdateAppliedToReplica(t *testing.T) {
 	span := repl.Desc().RSpan().AsRawSpanWithNoLocals()
 	conf := roachpb.SpanConfig{NumReplicas: 5, NumVoters: 3}
 
+	add, err := spanconfig.Addition(spanconfig.MakeTargetFromSpan(span), conf)
+	require.NoError(t, err)
 	deleted, added := spanConfigStore.Apply(
 		ctx,
 		false, /* dryrun */
-		spanconfig.Addition(spanconfig.MakeTargetFromSpan(span), conf),
+		add,
 	)
 	require.Empty(t, deleted)
 	require.Len(t, added, 1)
-	require.True(t, added[0].Target.GetSpan().Equal(span))
-	require.True(t, added[0].Config.Equal(conf))
+	require.True(t, added[0].GetTarget().GetSpan().Equal(span))
+	addedCfg := added[0].GetConfig()
+	require.True(t, addedCfg.Equal(conf))
 
 	require.NotNil(t, mockSubscriber.callback)
 	mockSubscriber.callback(ctx, span) // invoke the callback

pkg/migration/migrations/seed_tenant_span_configs.go

@@ -67,12 +67,12 @@ func seedTenantSpanConfigsMigration(
 				Key:    tenantPrefix,
 				EndKey: tenantPrefix.Next(),
 			}
-			toUpsert := []spanconfig.Record{
-				{
-					Target: spanconfig.MakeTargetFromSpan(tenantSeedSpan),
-					Config: tenantSpanConfig,
-				},
+			record, err := spanconfig.MakeRecord(spanconfig.MakeTargetFromSpan(tenantSeedSpan),
+				tenantSpanConfig)
+			if err != nil {
+				return err
 			}
+			toUpsert := []spanconfig.Record{record}
 			scRecords, err := scKVAccessor.GetSpanConfigRecords(ctx, []spanconfig.Target{tenantTarget})
 			if err != nil {
 				return err

pkg/roachpb/span_config.go

@@ -14,6 +14,8 @@ import (
 	"fmt"
 	"strings"
 	"time"
+
+	"github.com/cockroachdb/errors"
 )
 
 // StoreMatchesConstraint returns whether a store's attributes or node's
@@ -50,6 +52,49 @@ func (s *SpanConfig) TTL() time.Duration {
 	return time.Duration(s.GCPolicy.TTLSeconds) * time.Second
 }
 
+// ValidateSystemTargetSpanConfig ensures that only protection policies
+// (GCPolicy.ProtectionPolicies) field is set on the underlying
+// roachpb.SpanConfig.
+func (s *SpanConfig) ValidateSystemTargetSpanConfig() error {
+	if s.RangeMinBytes != 0 {
+		return errors.AssertionFailedf("RangeMinBytes set on system span config")
+	}
+	if s.RangeMaxBytes != 0 {
+		return errors.AssertionFailedf("RangeMaxBytes set on system span config")
+	}
+	if s.GCPolicy.TTLSeconds != 0 {
+		return errors.AssertionFailedf("TTLSeconds set on system span config")
+	}
+	if s.GCPolicy.IgnoreStrictEnforcement {
+		return errors.AssertionFailedf("IgnoreStrictEnforcement set on system span config")
+	}
+	if s.GlobalReads {
+		return errors.AssertionFailedf("GlobalReads set on system span config")
+	}
+	if s.NumReplicas != 0 {
+		return errors.AssertionFailedf("NumReplicas set on system span config")
+	}
+	if s.NumVoters != 0 {
+		return errors.AssertionFailedf("NumVoters set on system span config")
+	}
+	if len(s.Constraints) != 0 {
+		return errors.AssertionFailedf("Constraints set on system span config")
+	}
+	if len(s.VoterConstraints) != 0 {
+		return errors.AssertionFailedf("VoterConstraints set on system span config")
+	}
+	if len(s.LeasePreferences) != 0 {
+		return errors.AssertionFailedf("LeasePreferences set on system span config")
+	}
+	if s.RangefeedEnabled {
+		return errors.AssertionFailedf("RangefeedEnabled set on system span config")
+	}
+	if s.ExcludeDataFromBackup {
+		return errors.AssertionFailedf("ExcludeDataFromBackup set on system span config")
+	}
+	return nil
+}
+
 // GetNumVoters returns the number of voting replicas as defined in the
 // span config.
 // TODO(arul): We can get rid of this now that we're correctly populating

pkg/roachpb/span_config.proto

@@ -185,6 +185,10 @@ message SpanConfig {
   bool exclude_data_from_backup = 11;
 
   // Next ID: 12
+  //
+  // When adding a field, also add a check a to `ValidateSystemTargetSpanConfig`
+  // if it is not expected to be set on a SpanConfig corresponding to a
+  // SystemTarget.
 }
 
 // SystemSpanConfigTarget specifies the target of system span configurations.

pkg/spanconfig/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "spanconfig",
     srcs = [
         "protectedts_state_reader.go",
+        "record.go",
         "spanconfig.go",
         "systemtarget.go",
         "target.go",
@@ -30,6 +31,7 @@ go_test(
     name = "spanconfig_test",
     srcs = [
         "protectedts_state_reader_test.go",
+        "record_test.go",
         "target_test.go",
     ],
     embed = [":spanconfig"],

pkg/spanconfig/record.go

@@ -0,0 +1,52 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package spanconfig
+
+import (
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/errors"
+)
+
+// Record ties a target to its corresponding config.
+type Record struct {
+	// target specifies the target (keyspan(s)) the config applies over.
+	target Target
+
+	// config is the set of attributes that apply over the corresponding target.
+	config roachpb.SpanConfig
+}
+
+// MakeRecord returns a Record with the specified Target and SpanConfig. If the
+// Record targets a SystemTarget, we also validate the SpanConfig.
+func MakeRecord(target Target, cfg roachpb.SpanConfig) (Record, error) {
+	if target.IsSystemTarget() {
+		if err := cfg.ValidateSystemTargetSpanConfig(); err != nil {
+			return Record{},
+				errors.NewAssertionErrorWithWrappedErrf(err, "failed to validate SystemTarget SpanConfig")
+		}
+	}
+	return Record{target: target, config: cfg}, nil
+}
+
+// IsEmpty returns true if the receiver is an empty Record.
+func (r *Record) IsEmpty() bool {
+	return r.target.isEmpty() && r.config.IsEmpty()
+}
+
+// GetTarget returns the Record target.
+func (r *Record) GetTarget() Target {
+	return r.target
+}
+
+// GetConfig returns the Record config.
+func (r *Record) GetConfig() roachpb.SpanConfig {
+	return r.config
+}