Merge #65207

65207: backupccl: encrypt returned SSTs in backup proc r=dt a=dt

This changes where backup SST files are encrypted if they are returned
to the SQL process that called ExportRequest rather than written directly
to S3 by the ExportRequest. Previously if a backup was being done with an
encryption key, all ExportRequests were sent with that key and the KV layer
would always encrypt files with that key, before either sending them to the
requested destination or replying with them depending on the request.

This however changes the sent requests to omit the encryption options if they
are requesting the file be returned, rather than written to a destination, so
that the files will be returned in cleartext and the backup processor can then
encrypt them before writing them to the desired destination. The backup processor
already could have decrypted them -- it has the key as it was the one sending it
to the KV layer after all -- so this doesn't significantly change what can be read
where, but does keep the chosen key within the tenant's SQL process and makes it
easier for the backup data processor to manipulate the file before sending it
along to the backup destination, in that it would otherwise need to decrypt and
then re-encrypt it.

Release note: none.

Co-authored-by: David Taylor <[email protected]>

pkg/ccl/backupccl/backup_processor.go

@@ -263,6 +263,11 @@ func runBackupProcessor(
 					TargetFileSize:                      targetFileSize,
 					ReturnSST:                           writeSSTsInProcessor,
 				}
+				// If we're sending the SST back, don't encrypt it -- we'll encrypt it
+				// here instead.
+				if req.ReturnSST {
+					req.Encryption = nil
+				}
 
 				// If we're doing re-attempts but are not yet in the priority regime,
 				// check to see if it is time to switch to priority.
@@ -361,9 +366,13 @@ func runBackupProcessor(
 
 				files := make([]BackupManifest_File, 0)
 				for _, file := range res.Files {
-					if writeSSTsInProcessor {
+					if len(file.SST) > 0 {
+						// TODO(dt): remove this when we add small-file returning.
+						if !writeSSTsInProcessor {
+							return errors.New("ExportRequest returned unexpected file payload")
+						}
 						file.Path = fmt.Sprintf("%d.sst", builtins.GenerateUniqueInt(flowCtx.EvalCtx.NodeID.SQLInstanceID()))
-						if err := writeFile(ctx, file, defaultStore, storeByLocalityKV); err != nil {
+						if err := writeFile(ctx, file, defaultStore, storeByLocalityKV, spec.Encryption); err != nil {
 							return err
 						}
 					}
@@ -441,6 +450,7 @@ func writeFile(
 	file roachpb.ExportResponse_File,
 	defaultStore cloud.ExternalStorage,
 	storeByLocalityKV map[string]cloud.ExternalStorage,
+	enc *roachpb.FileEncryptionOptions,
 ) error {
 	if defaultStore == nil {
 		return errors.New("no default store created when writing SST")
@@ -454,6 +464,14 @@ func writeFile(
 		exportStore = localitySpecificStore
 	}
 
+	if enc != nil {
+		var err error
+		data, err = storageccl.EncryptFile(data, enc.Key)
+		if err != nil {
+			return err
+		}
+	}
+
 	if err := cloud.WriteFile(ctx, exportStore, file.Path, bytes.NewReader(data)); err != nil {
 		log.VEventf(ctx, 1, "failed to put file: %+v", err)
 		return errors.Wrap(err, "writing SST")