scrub pii from observe logs (#2285)

## Description

This is a quick hack to satisfy a primary case of PII scrubbing.  We expect to revisit it in the future.

## Does this PR need a docs update or release note?

- [x] ⛔ No 

## Type of change

- [x] 🌻 Feature

## Issue(s)

* #2284

## Test Plan

- [x] 💪 Manual
- [x] ⚡ Unit test

src/internal/connector/exchange/data_collections.go

@@ -251,7 +251,10 @@ func createCollections(
 		Credentials:   creds,
 	}
 
-	foldersComplete, closer := observe.MessageWithCompletion(ctx, observe.Bulletf("%s - %s", qp.Category, user))
+	foldersComplete, closer := observe.MessageWithCompletion(ctx, observe.Bulletf(
+		"%s - %s",
+		observe.Safe(qp.Category.String()),
+		observe.PII(user)))
 	defer closer()
 	defer close(foldersComplete)
 

src/internal/connector/exchange/exchange_data_collection.go

@@ -183,9 +183,9 @@ func (col *Collection) streamItems(ctx context.Context) {
 		var closer func()
 		colProgress, closer = observe.CollectionProgress(
 			ctx,
-			user,
 			col.fullPath.Category().String(),
-			col.fullPath.Folder())
+			observe.PII(user),
+			observe.PII(col.fullPath.Folder()))
 
 		go closer()
 

src/internal/connector/exchange/service_restore.go

@@ -374,7 +374,11 @@ func restoreCollection(
 		user      = directory.ResourceOwner()
 	)
 
-	colProgress, closer := observe.CollectionProgress(ctx, user, category.String(), directory.Folder())
+	colProgress, closer := observe.CollectionProgress(
+		ctx,
+		category.String(),
+		observe.PII(user),
+		observe.PII(directory.Folder()))
 	defer closer()
 	defer close(colProgress)
 

src/internal/connector/onedrive/collection.go

@@ -191,7 +191,7 @@ func (oc *Collection) populateItems(ctx context.Context) {
 	folderProgress, colCloser := observe.ProgressWithCount(
 		ctx,
 		observe.ItemQueueMsg,
-		"/"+parentPathString,
+		observe.PII("/"+parentPathString),
 		int64(len(oc.driveItems)))
 	defer colCloser()
 	defer close(folderProgress)
@@ -267,7 +267,7 @@ func (oc *Collection) populateItems(ctx context.Context) {
 				}
 
 				// display/log the item download
-				progReader, closer := observe.ItemProgress(ctx, itemData, observe.ItemBackupMsg, itemName, itemSize)
+				progReader, closer := observe.ItemProgress(ctx, itemData, observe.ItemBackupMsg, observe.PII(itemName), itemSize)
 				go closer()
 
 				return progReader, nil

src/internal/connector/onedrive/collections.go

@@ -144,7 +144,7 @@ func (c *Collections) Get(ctx context.Context) ([]data.Collection, error) {
 		maps.Copy(excludedItems, excluded)
 	}
 
-	observe.Message(ctx, fmt.Sprintf("Discovered %d items to backup", c.NumItems))
+	observe.Message(ctx, observe.Safe(fmt.Sprintf("Discovered %d items to backup", c.NumItems)))
 
 	// Add an extra for the metadata collection.
 	collections := make([]data.Collection, 0, len(c.CollectionMap)+1)

src/internal/connector/onedrive/restore.go

@@ -243,7 +243,7 @@ func restoreItem(
 	}
 
 	iReader := itemData.ToReader()
-	progReader, closer := observe.ItemProgress(ctx, iReader, observe.ItemRestoreMsg, itemName, ss.Size())
+	progReader, closer := observe.ItemProgress(ctx, iReader, observe.ItemRestoreMsg, observe.PII(itemName), ss.Size())
 
 	go closer()
 

src/internal/connector/sharepoint/collection.go

@@ -158,9 +158,9 @@ func (sc *Collection) populate(ctx context.Context) {
 	// TODO: Insert correct ID for CollectionProgress
 	colProgress, closer := observe.CollectionProgress(
 		ctx,
-		"name",
 		sc.fullPath.Category().String(),
-		sc.fullPath.Folder())
+		observe.Safe("name"),
+		observe.PII(sc.fullPath.Folder()))
 	go closer()
 
 	defer func() {

src/internal/connector/sharepoint/data_collections.go

@@ -46,7 +46,8 @@ func DataCollections(
 	for _, scope := range b.Scopes() {
 		foldersComplete, closer := observe.MessageWithCompletion(ctx, observe.Bulletf(
 			"%s - %s",
-			scope.Category().PathType(), site))
+			observe.Safe(scope.Category().PathType().String()),
+			observe.PII(site)))
 		defer closer()
 		defer close(foldersComplete)
 

src/internal/observe/observe.go

@@ -138,8 +138,9 @@ const (
 // Progress Updates
 
 // Message is used to display a progress message
-func Message(ctx context.Context, message string) {
-	logger.Ctx(ctx).Info(message)
+func Message(ctx context.Context, msg cleanable) {
+	logger.Ctx(ctx).Info(msg.clean())
+	message := msg.String()
 
 	if cfg.hidden() {
 		return
@@ -163,9 +164,15 @@ func Message(ctx context.Context, message string) {
 
 // MessageWithCompletion is used to display progress with a spinner
 // that switches to "done" when the completion channel is signalled
-func MessageWithCompletion(ctx context.Context, message string) (chan<- struct{}, func()) {
+func MessageWithCompletion(
+	ctx context.Context,
+	msg cleanable,
+) (chan<- struct{}, func()) {
+	clean := msg.clean()
+	message := msg.String()
+
 	log := logger.Ctx(ctx)
-	log.Info(message)
+	log.Info(clean)
 
 	completionCh := make(chan struct{}, 1)
 
@@ -201,7 +208,7 @@ func MessageWithCompletion(ctx context.Context, message string) (chan<- struct{}
 	}(completionCh)
 
 	wacb := waitAndCloseBar(bar, func() {
-		log.Info("done - " + message)
+		log.Info("done - " + clean)
 	})
 
 	return completionCh, wacb
@@ -217,10 +224,11 @@ func MessageWithCompletion(ctx context.Context, message string) (chan<- struct{}
 func ItemProgress(
 	ctx context.Context,
 	rc io.ReadCloser,
-	header, iname string,
+	header string,
+	iname cleanable,
 	totalBytes int64,
 ) (io.ReadCloser, func()) {
-	log := logger.Ctx(ctx).With("item", iname, "size", humanize.Bytes(uint64(totalBytes)))
+	log := logger.Ctx(ctx).With("item", iname.clean(), "size", humanize.Bytes(uint64(totalBytes)))
 	log.Debug(header)
 
 	if cfg.hidden() || rc == nil || totalBytes == 0 {
@@ -232,7 +240,7 @@ func ItemProgress(
 	barOpts := []mpb.BarOption{
 		mpb.PrependDecorators(
 			decor.Name(header, decor.WCSyncSpaceR),
-			decor.Name(iname, decor.WCSyncSpaceR),
+			decor.Name(iname.String(), decor.WCSyncSpaceR),
 			decor.CountersKibiByte(" %.1f/%.1f ", decor.WC{W: 8}),
 			decor.NewPercentage("%d ", decor.WC{W: 4}),
 		),
@@ -256,9 +264,14 @@ func ItemProgress(
 // of the specified count.
 // Each write to the provided channel counts as a single increment.
 // The caller is expected to close the channel.
-func ProgressWithCount(ctx context.Context, header, message string, count int64) (chan<- struct{}, func()) {
+func ProgressWithCount(
+	ctx context.Context,
+	header string,
+	message cleanable,
+	count int64,
+) (chan<- struct{}, func()) {
 	log := logger.Ctx(ctx)
-	lmsg := fmt.Sprintf("%s %s - %d", header, message, count)
+	lmsg := fmt.Sprintf("%s %s - %d", header, message.clean(), count)
 	log.Info(lmsg)
 
 	progressCh := make(chan struct{})
@@ -281,7 +294,7 @@ func ProgressWithCount(ctx context.Context, header, message string, count int64)
 	barOpts := []mpb.BarOption{
 		mpb.PrependDecorators(
 			decor.Name(header, decor.WCSyncSpaceR),
-			decor.Name(message),
+			decor.Name(message.String()),
 			decor.Counters(0, " %d/%d "),
 		),
 	}
@@ -355,13 +368,17 @@ func makeSpinFrames(barWidth int) {
 // counts as a single increment.  The caller is expected to close the channel.
 func CollectionProgress(
 	ctx context.Context,
-	user, category, dirName string,
+	category string,
+	user, dirName cleanable,
 ) (chan<- struct{}, func()) {
-	log := logger.Ctx(ctx).With("user", user, "category", category, "dir", dirName)
-	message := "Collecting " + dirName
+	log := logger.Ctx(ctx).With(
+		"user", user.clean(),
+		"category", category,
+		"dir", dirName.clean())
+	message := "Collecting Directory"
 	log.Info(message)
 
-	if cfg.hidden() || len(user) == 0 || len(dirName) == 0 {
+	if cfg.hidden() || len(user.String()) == 0 || len(dirName.String()) == 0 {
 		ch := make(chan struct{})
 
 		go func(ci <-chan struct{}) {
@@ -379,7 +396,7 @@ func CollectionProgress(
 	wg.Add(1)
 
 	barOpts := []mpb.BarOption{
-		mpb.PrependDecorators(decor.Name(category)),
+		mpb.PrependDecorators(decor.Name(string(category))),
 		mpb.AppendDecorators(
 			decor.CurrentNoUnit("%d - ", decor.WCSyncSpace),
 			decor.Name(fmt.Sprintf("%s - %s", user, dirName)),
@@ -439,8 +456,65 @@ func waitAndCloseBar(bar *mpb.Bar, log func()) func() {
 // other funcs
 // ---------------------------------------------------------------------------
 
-// Bulletf prepends the message with "∙ ", and formats it.
-// Ex: Bulletf("%s", "foo") => "∙ foo"
-func Bulletf(template string, vs ...any) string {
-	return fmt.Sprintf("∙ "+template, vs...)
+const Bullet = "∙"
+
+// ---------------------------------------------------------------------------
+// PII redaction
+// ---------------------------------------------------------------------------
+
+type cleanable interface {
+	clean() string
+	String() string
+}
+
+type PII string
+
+func (p PII) clean() string {
+	return "***"
+}
+
+func (p PII) String() string {
+	return string(p)
+}
+
+type Safe string
+
+func (s Safe) clean() string {
+	return string(s)
+}
+
+func (s Safe) String() string {
+	return string(s)
+}
+
+type bulletPII struct {
+	tmpl string
+	vars []cleanable
+}
+
+func Bulletf(template string, vs ...cleanable) bulletPII {
+	return bulletPII{
+		tmpl: "∙ " + template,
+		vars: vs,
+	}
+}
+
+func (b bulletPII) clean() string {
+	vs := make([]any, 0, len(b.vars))
+
+	for _, v := range b.vars {
+		vs = append(vs, v.clean())
+	}
+
+	return fmt.Sprintf(b.tmpl, vs...)
+}
+
+func (b bulletPII) String() string {
+	vs := make([]any, 0, len(b.vars))
+
+	for _, v := range b.vars {
+		vs = append(vs, v.String())
+	}
+
+	return fmt.Sprintf(b.tmpl, vs...)
 }

src/internal/observe/observe_test.go

@@ -26,6 +26,12 @@ func TestObserveProgressUnitSuite(t *testing.T) {
 	suite.Run(t, new(ObserveProgressUnitSuite))
 }
 
+var (
+	tst        = observe.Safe("test")
+	testcat    = observe.Safe("testcat")
+	testertons = observe.Safe("testertons")
+)
+
 func (suite *ObserveProgressUnitSuite) TestItemProgress() {
 	ctx, flush := tester.NewContext()
 	defer flush()
@@ -47,7 +53,7 @@ func (suite *ObserveProgressUnitSuite) TestItemProgress() {
 		ctx,
 		io.NopCloser(bytes.NewReader(from)),
 		"folder",
-		"test",
+		tst,
 		100)
 	require.NotNil(t, prog)
 	require.NotNil(t, closer)
@@ -97,7 +103,7 @@ func (suite *ObserveProgressUnitSuite) TestCollectionProgress_unblockOnCtxCancel
 		observe.SeedWriter(context.Background(), nil, nil)
 	}()
 
-	progCh, closer := observe.CollectionProgress(ctx, "test", "testcat", "testertons")
+	progCh, closer := observe.CollectionProgress(ctx, "test", testcat, testertons)
 	require.NotNil(t, progCh)
 	require.NotNil(t, closer)
 
@@ -132,7 +138,7 @@ func (suite *ObserveProgressUnitSuite) TestCollectionProgress_unblockOnChannelCl
 		observe.SeedWriter(context.Background(), nil, nil)
 	}()
 
-	progCh, closer := observe.CollectionProgress(ctx, "test", "testcat", "testertons")
+	progCh, closer := observe.CollectionProgress(ctx, "test", testcat, testertons)
 	require.NotNil(t, progCh)
 	require.NotNil(t, closer)
 
@@ -164,7 +170,7 @@ func (suite *ObserveProgressUnitSuite) TestObserveProgress() {
 
 	message := "Test Message"
 
-	observe.Message(ctx, message)
+	observe.Message(ctx, observe.Safe(message))
 	observe.Complete()
 	require.NotEmpty(suite.T(), recorder.String())
 	require.Contains(suite.T(), recorder.String(), message)
@@ -185,7 +191,7 @@ func (suite *ObserveProgressUnitSuite) TestObserveProgressWithCompletion() {
 
 	message := "Test Message"
 
-	ch, closer := observe.MessageWithCompletion(ctx, message)
+	ch, closer := observe.MessageWithCompletion(ctx, observe.Safe(message))
 
 	// Trigger completion
 	ch <- struct{}{}
@@ -215,7 +221,7 @@ func (suite *ObserveProgressUnitSuite) TestObserveProgressWithChannelClosed() {
 
 	message := "Test Message"
 
-	ch, closer := observe.MessageWithCompletion(ctx, message)
+	ch, closer := observe.MessageWithCompletion(ctx, observe.Safe(message))
 
 	// Close channel without completing
 	close(ch)
@@ -247,7 +253,7 @@ func (suite *ObserveProgressUnitSuite) TestObserveProgressWithContextCancelled()
 
 	message := "Test Message"
 
-	_, closer := observe.MessageWithCompletion(ctx, message)
+	_, closer := observe.MessageWithCompletion(ctx, observe.Safe(message))
 
 	// cancel context
 	cancel()
@@ -278,7 +284,7 @@ func (suite *ObserveProgressUnitSuite) TestObserveProgressWithCount() {
 	message := "Test Message"
 	count := 3
 
-	ch, closer := observe.ProgressWithCount(ctx, header, message, int64(count))
+	ch, closer := observe.ProgressWithCount(ctx, header, observe.Safe(message), int64(count))
 
 	for i := 0; i < count; i++ {
 		ch <- struct{}{}
@@ -311,7 +317,7 @@ func (suite *ObserveProgressUnitSuite) TestObserveProgressWithCountChannelClosed
 	message := "Test Message"
 	count := 3
 
-	ch, closer := observe.ProgressWithCount(ctx, header, message, int64(count))
+	ch, closer := observe.ProgressWithCount(ctx, header, observe.Safe(message), int64(count))
 
 	close(ch)
 

src/internal/operations/backup.go

@@ -256,7 +256,7 @@ func produceBackupDataCollections(
 	metadata []data.Collection,
 	ctrlOpts control.Options,
 ) ([]data.Collection, error) {
-	complete, closer := observe.MessageWithCompletion(ctx, "Discovering items to backup")
+	complete, closer := observe.MessageWithCompletion(ctx, observe.Safe("Discovering items to backup"))
 	defer func() {
 		complete <- struct{}{}
 		close(complete)
@@ -338,7 +338,7 @@ func consumeBackupDataCollections(
 	backupID model.StableID,
 	isIncremental bool,
 ) (*kopia.BackupStats, *details.Builder, map[string]path.Path, error) {
-	complete, closer := observe.MessageWithCompletion(ctx, "Backing up data")
+	complete, closer := observe.MessageWithCompletion(ctx, observe.Safe("Backing up data"))
 	defer func() {
 		complete <- struct{}{}
 		close(complete)

src/internal/operations/operation.go

@@ -95,7 +95,7 @@ func connectToM365(
 	sel selectors.Selector,
 	acct account.Account,
 ) (*connector.GraphConnector, error) {
-	complete, closer := observe.MessageWithCompletion(ctx, "Connecting to M365")
+	complete, closer := observe.MessageWithCompletion(ctx, observe.Safe("Connecting to M365"))
 	defer func() {
 		complete <- struct{}{}
 		close(complete)

src/internal/operations/restore.go

@@ -170,9 +170,9 @@ func (op *RestoreOperation) Run(ctx context.Context) (restoreDetails *details.De
 
 	ctx = clues.Add(ctx, "details_paths", len(paths))
 
-	observe.Message(ctx, fmt.Sprintf("Discovered %d items in backup %s to restore", len(paths), op.BackupID))
+	observe.Message(ctx, observe.Safe(fmt.Sprintf("Discovered %d items in backup %s to restore", len(paths), op.BackupID)))
 
-	kopiaComplete, closer := observe.MessageWithCompletion(ctx, "Enumerating items in repository")
+	kopiaComplete, closer := observe.MessageWithCompletion(ctx, observe.Safe("Enumerating items in repository"))
 	defer closer()
 	defer close(kopiaComplete)
 
@@ -196,7 +196,7 @@ func (op *RestoreOperation) Run(ctx context.Context) (restoreDetails *details.De
 		return nil, opStats.readErr
 	}
 
-	restoreComplete, closer := observe.MessageWithCompletion(ctx, "Restoring data")
+	restoreComplete, closer := observe.MessageWithCompletion(ctx, observe.Safe("Restoring data"))
 	defer closer()
 	defer close(restoreComplete)
 

src/pkg/repository/repository.go

@@ -154,7 +154,7 @@ func Connect(
 	// their output getting clobbered (#1720)
 	defer observe.Complete()
 
-	complete, closer := observe.MessageWithCompletion(ctx, "Connecting to repository")
+	complete, closer := observe.MessageWithCompletion(ctx, observe.Safe("Connecting to repository"))
 	defer closer()
 	defer close(complete)