server: add cluster setting to control tenant trace redaction

Previously, all operations run on the server on behalf of a secondary
tenant would have their logs redacted. This patch introduces a new
cluster setting that controls whether logs should be redacted or
not. This is only settable by the system tenant and maintains the
previous behaviour by default.

Release note (ops change): introduce a new cluster setting
(`server.secondary_tenants.redact_trace`) which controls if traces
should be redacted for ops run on behalf of secondary tenants.

pkg/server/node.go

@@ -152,6 +152,12 @@ var (
 		10*time.Second,
 		settings.NonNegativeDurationWithMaximum(maxGraphiteInterval),
 	).WithPublic()
+	redactServerTracesForSecondaryTenants = settings.RegisterBoolSetting(
+		settings.TenantReadOnly,
+		"server.secondary_tenants.redact_trace",
+		"controls if server side traces are redacted for tenant operations",
+		true,
+	).WithPublic()
 )
 
 type nodeMetrics struct {
@@ -233,6 +239,8 @@ type Node struct {
 
 	spanConfigAccessor spanconfig.KVAccessor // powers the span configuration RPCs
 
+	settings *cluster.Settings
+
 	// Turns `Node.writeNodeStatus` into a no-op. This is a hack to enable the
 	// COCKROACH_DEBUG_TS_IMPORT_FILE env var.
 	suppressNodeStatus syncutil.AtomicBool
@@ -375,6 +383,7 @@ func NewNode(
 		tenantSettingsWatcher: tenantSettingsWatcher,
 		spanConfigAccessor:    spanConfigAccessor,
 		testingErrorEvent:     cfg.TestingKnobs.TestingResponseErrorEvent,
+		settings:              cfg.Settings,
 	}
 	n.storeCfg.KVAdmissionController = n.admissionController
 	n.perReplicaServer = kvserver.MakeServer(&n.Descriptor, n.stores)
@@ -1097,6 +1106,7 @@ type spanForRequest struct {
 	sp            *tracing.Span
 	needRecording bool
 	tenID         roachpb.TenantID
+	settings      *cluster.Settings
 }
 
 // finish finishes the span. If the span was recording and br is not nil, the
@@ -1114,17 +1124,21 @@ func (sp *spanForRequest) finish(ctx context.Context, br *roachpb.BatchResponse)
 
 	rec = sp.sp.FinishAndGetConfiguredRecording()
 	if rec != nil {
-		// Decide if the trace for this RPC, if any, will need to be redacted. It
-		// needs to be redacted if the response goes to a tenant. In case the request
-		// is local, then the trace might eventually go to a tenant (and tenID might
-		// be set), but it will go to the tenant only indirectly, through the response
-		// of a parent RPC. In that case, that parent RPC is responsible for the
-		// redaction.
-		//
-		// Tenants get a redacted recording, i.e. with anything
-		// sensitive stripped out of the verbose messages. However,
 		// structured payloads stay untouched.
-		needRedaction := sp.tenID != roachpb.SystemTenantID
+		// Decide if the trace for this RPC, if any, will need to be redacted. In
+		// general, responses sent to a tenant are redacted unless indicated
+		// otherwise by the cluster setting below.
+		//
+		// In case the request is local, the trace might eventually go to a tenant
+		// (and tenID might be set), but it will go to the tenant only indirectly
+		// through the response of a parent RPC. In that case, the parent RPC is
+		// responsible for the redaction.
+		//
+		// Even if the recording sent to a tenant is redacted (anything sensitive
+		// is stripped out of the verbose messages). However, structured payloads
+		// stay untouched.
+		needRedaction := sp.tenID != roachpb.SystemTenantID &&
+			redactServerTracesForSecondaryTenants.Get(&sp.settings.SV)
 		if needRedaction {
 			if err := redactRecordingForTenant(sp.tenID, rec); err != nil {
 				log.Errorf(ctx, "error redacting trace recording: %s", err)
@@ -1151,11 +1165,15 @@ func (sp *spanForRequest) finish(ctx context.Context, br *roachpb.BatchResponse)
 func (n *Node) setupSpanForIncomingRPC(
 	ctx context.Context, tenID roachpb.TenantID, ba *roachpb.BatchRequest,
 ) (context.Context, spanForRequest) {
-	return setupSpanForIncomingRPC(ctx, tenID, ba, n.storeCfg.AmbientCtx.Tracer)
+	return setupSpanForIncomingRPC(ctx, tenID, ba, n.storeCfg.AmbientCtx.Tracer, n.settings)
 }
 
 func setupSpanForIncomingRPC(
-	ctx context.Context, tenID roachpb.TenantID, ba *roachpb.BatchRequest, tr *tracing.Tracer,
+	ctx context.Context,
+	tenID roachpb.TenantID,
+	ba *roachpb.BatchRequest,
+	tr *tracing.Tracer,
+	settings *cluster.Settings,
 ) (context.Context, spanForRequest) {
 	var newSpan *tracing.Span
 	parentSpan := tracing.SpanFromContext(ctx)
@@ -1199,6 +1217,7 @@ func setupSpanForIncomingRPC(
 		needRecording: needRecordingCollection,
 		tenID:         tenID,
 		sp:            newSpan,
+		settings:      settings,
 	}
 }