centralize observe channel listener handling

To ensure that the observe channel handling
doesn't accidentally spawn unkillable routines,
adds a centralized channel listener func for
standard channel management in observe
progress bars.

src/cli/cli.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/alcionai/clues"
+	"github.com/pkg/profile"
 	"github.com/spf13/cobra"
 	"golang.org/x/exp/slices"
 
@@ -125,6 +126,8 @@ func Handle() {
 	loglevel, logfile := logger.PreloadLoggingFlags()
 	ctx, log := logger.Seed(ctx, loglevel, logfile)
 
+	defer profile.Start().Stop()
+
 	defer func() {
 		_ = log.Sync() // flush all logs in the buffer
 	}()

src/go.mod

@@ -39,11 +39,14 @@ require (
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d // indirect
 	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/dnaeon/go-vcr v1.2.0 // indirect
+	github.com/felixge/fgprof v0.9.3 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
+	github.com/google/pprof v0.0.0-20211214055906-6f57359322fd // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.6 // indirect
+	github.com/pkg/profile v1.7.0 // indirect
 	github.com/spf13/afero v1.9.3 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect

src/go.sum

@@ -107,6 +107,8 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m
 github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/felixge/fgprof v0.9.3 h1:VvyZxILNuCiUCSXtPtYmmtGvb65nqXh2QFWc0Wpf2/g=
+github.com/felixge/fgprof v0.9.3/go.mod h1:RdbpDgzqYVh/T9fPELJyV7EYJuHB55UTEULNun8eiPw=
 github.com/frankban/quicktest v1.14.3 h1:FJKSZTDHjyhriyC81FLQ0LY93eSai0ZyR/ZIkd3ZUKE=
 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
@@ -187,6 +189,8 @@ github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99/go.mod h1:ZgVRPoUq/hf
 github.com/google/pprof v0.0.0-20201023163331-3e6fc7fc9c4c/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20201203190320-1bf35d6f28c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20201218002935-b9804c9f04c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
+github.com/google/pprof v0.0.0-20211214055906-6f57359322fd h1:1FjCyPC+syAzJ5/2S8fqdZK1R22vvA0J7JZKcuOIQ7Y=
+github.com/google/pprof v0.0.0-20211214055906-6f57359322fd/go.mod h1:KgnwoLYCZ8IQu3XUZ8Nc/bM9CCZFOyjUNOSygVozoDg=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
@@ -209,6 +213,7 @@ github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
+github.com/ianlancetaylor/demangle v0.0.0-20210905161508-09a460cdf81d/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w=
 github.com/inconshreveable/mousetrap v1.0.1 h1:U3uMjPSQEBMNp1lFxmllqCPM6P5u/Xq7Pgzkat/bFNc=
 github.com/inconshreveable/mousetrap v1.0.1/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
@@ -315,6 +320,8 @@ github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/profile v1.7.0 h1:hnbDkaNWPCLMO9wGLdBFTIZvzDrDfBM2072E1S9gJkA=
+github.com/pkg/profile v1.7.0/go.mod h1:8Uer0jas47ZQMJ7VD+OHknK4YDY07LPUC6dEvqDjvNo=
 github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -602,6 +609,7 @@ golang.org/x/sys v0.0.0-20210423185535-09eb48e85fd7/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210616045830-e2b7044e8c71/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

src/internal/observe/observe.go

@@ -22,6 +22,9 @@ const (
 	progressBarWidth     = 32
 )
 
+// styling
+const Bullet = "∙"
+
 var (
 	wg sync.WaitGroup
 	// TODO: Revisit this being a global nd make it a parameter to the progress methods
@@ -168,16 +171,17 @@ func MessageWithCompletion(
 	ctx context.Context,
 	msg cleanable,
 ) (chan<- struct{}, func()) {
-	clean := msg.clean()
+	var (
-	message := msg.String()
+		clean   = msg.clean()
+		message = msg.String()
+		log     = logger.Ctx(ctx)
+		ch      = make(chan struct{}, 1)
+	)
 
-	log := logger.Ctx(ctx)
 	log.Info(clean)
 
-	completionCh := make(chan struct{}, 1)
-
 	if cfg.hidden() {
-		return completionCh, func() { log.Info("done - " + clean) }
+		return ch, func() { log.Info("done - " + clean) }
 	}
 
 	wg.Add(1)
@@ -194,24 +198,24 @@ func MessageWithCompletion(
 		mpb.BarFillerOnComplete("done"),
 	)
 
-	go func(ci <-chan struct{}) {
+	go listen(
-		for {
+		ctx,
-			select {
+		ch,
-			case <-contxt.Done():
+		func() {
-				bar.SetTotal(-1, true)
+			bar.SetTotal(-1, true)
-			case <-ci:
+			bar.Abort(true)
-				// We don't care whether the channel was signalled or closed
+		},
-				// Use either one as an indication that the bar is done
+		func() {
-				bar.SetTotal(-1, true)
+			// We don't care whether the channel was signalled or closed
-			}
+			// Use either one as an indication that the bar is done
-		}
+			bar.SetTotal(-1, true)
-	}(completionCh)
+		})
 
 	wacb := waitAndCloseBar(bar, func() {
 		log.Info("done - " + clean)
 	})
 
-	return completionCh, wacb
+	return ch, wacb
 }
 
 // ---------------------------------------------------------------------------
@@ -228,7 +232,9 @@ func ItemProgress(
 	iname cleanable,
 	totalBytes int64,
 ) (io.ReadCloser, func()) {
-	log := logger.Ctx(ctx).With("item", iname.clean(), "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 {
@@ -270,23 +276,17 @@ func ProgressWithCount(
 	message cleanable,
 	count int64,
 ) (chan<- struct{}, func()) {
-	log := logger.Ctx(ctx)
+	var (
-	lmsg := fmt.Sprintf("%s %s - %d", header, message.clean(), count)
+		log  = logger.Ctx(ctx)
+		lmsg = fmt.Sprintf("%s %s - %d", header, message.clean(), count)
+		ch   = make(chan struct{})
+	)
+
 	log.Info(lmsg)
 
-	progressCh := make(chan struct{})
-
 	if cfg.hidden() {
-		go func(ci <-chan struct{}) {
+		go listen(ctx, ch, nop, nop)
-			for {
+		return ch, func() { log.Info("done - " + lmsg) }
-				_, ok := <-ci
-				if !ok {
-					return
-				}
-			}
-		}(progressCh)
-
-		return progressCh, func() { log.Info("done - " + lmsg) }
 	}
 
 	wg.Add(1)
@@ -305,24 +305,11 @@ func ProgressWithCount(
 
 	bar := progress.New(count, mpb.NopStyle(), barOpts...)
 
-	ch := make(chan struct{})
+	go listen(
-	go func(ci <-chan struct{}) {
+		ctx,
-		for {
+		ch,
-			select {
+		func() { bar.Abort(true) },
-			case <-contxt.Done():
+		bar.Increment)
-				bar.Abort(true)
-				return
-
-			case _, ok := <-ci:
-				if !ok {
-					bar.Abort(true)
-					return
-				}
-
-				bar.Increment()
-			}
-		}
-	}(ch)
 
 	wacb := waitAndCloseBar(bar, func() {
 		log.Info("done - " + lmsg)
@@ -371,33 +358,28 @@ func CollectionProgress(
 	category string,
 	user, dirName cleanable,
 ) (chan<- struct{}, func()) {
-	log := logger.Ctx(ctx).With(
+	var (
-		"user", user.clean(),
+		counted int
-		"category", category,
+		ch      = make(chan struct{})
-		"dir", dirName.clean())
+		log     = logger.Ctx(ctx).With(
-	message := "Collecting Directory"
+			"user", user.clean(),
+			"category", category,
+			"dir", dirName.clean())
+		message = "Collecting Directory"
+	)
+
 	log.Info(message)
 
+	incCount := func() {
+		counted++
+		// Log every 1000 items that are processed
+		if counted%1000 == 0 {
+			log.Infow("uploading", "count", counted)
+		}
+	}
+
 	if cfg.hidden() || len(user.String()) == 0 || len(dirName.String()) == 0 {
-		ch := make(chan struct{})
+		go listen(ctx, ch, nop, incCount)
-
-		counted := 0
-
-		go func(ci <-chan struct{}) {
-			for {
-				_, ok := <-ci
-				if !ok {
-					return
-				}
-				counted++
-
-				// Log every 1000 items that are processed
-				if counted%1000 == 0 {
-					log.Infow("uploading", "count", counted)
-				}
-			}
-		}(ch)
-
 		return ch, func() { log.Infow("done - "+message, "count", counted) }
 	}
 
@@ -419,36 +401,16 @@ func CollectionProgress(
 	bar := progress.New(
 		-1, // -1 to indicate an unbounded count
 		mpb.SpinnerStyle(spinFrames...),
-		barOpts...,
+		barOpts...)
-	)
 
-	var counted int
+	go listen(
-
+		ctx,
-	ch := make(chan struct{})
+		ch,
-	go func(ci <-chan struct{}) {
+		func() { bar.SetTotal(-1, true) },
-		for {
+		func() {
-			select {
+			incCount()
-			case <-contxt.Done():
+			bar.Increment()
-				bar.SetTotal(-1, true)
+		})
-				return
-
-			case _, ok := <-ci:
-				if !ok {
-					bar.SetTotal(-1, true)
-					return
-				}
-
-				counted++
-
-				// Log every 1000 items that are processed
-				if counted%1000 == 0 {
-					log.Infow("uploading", "count", counted)
-				}
-
-				bar.Increment()
-			}
-		}
-	}(ch)
 
 	wacb := waitAndCloseBar(bar, func() {
 		log.Infow("done - "+message, "count", counted)
@@ -469,7 +431,26 @@ func waitAndCloseBar(bar *mpb.Bar, log func()) func() {
 // other funcs
 // ---------------------------------------------------------------------------
 
-const Bullet = "∙"
+var nop = func() {}
+
+// listen runs a
+func listen(ctx context.Context, ci <-chan struct{}, onEnd, onInc func()) {
+	for {
+		select {
+		case <-ctx.Done():
+			onEnd()
+			return
+
+		case _, ok := <-ci:
+			if !ok {
+				onEnd()
+				return
+			}
+
+			onInc()
+		}
+	}
+}
 
 // ---------------------------------------------------------------------------
 // PII redaction