watch: batch & de-duplicate file events (docker#10865)

Adjust the debouncing logic so that it applies to all inbound file
events, regardless of whether they match a sync or rebuild rule.

When the batch is flushed out, if any event for the service is a
rebuild event, then the service is rebuilt and all sync events for
the batch are ignored. If _all_ events in the batch are sync events,
then a sync is triggered, passing the entire batch at once. This
provides a substantial performance win for the new `tar`-based
implementation, as it can efficiently transfer the changes in bulk.

Additionally, this helps with jitter, e.g. it's not uncommon for
there to be double-writes in quick succession to a file, so even if
there's not many files being modified at once, it can still prevent
some unnecessary transfers.

Signed-off-by: Milas Bowman <[email protected]>

Author: milas

internal/sync/docker_cp.go

@@ -71,19 +71,19 @@ func (d *DockerCopy) sync(ctx context.Context, service types.ServiceConfig, path
 		if fi.IsDir() {
 			for i := 1; i <= scale; i++ {
 				_, err := d.client.Exec(ctx, d.projectName, api.RunOptions{
-					Service: pathMapping.Service,
+					Service: service.Name,
 					Command: []string{"mkdir", "-p", pathMapping.ContainerPath},
 					Index:   i,
 				})
 				if err != nil {
-					logrus.Warnf("failed to create %q from %s: %v", pathMapping.ContainerPath, pathMapping.Service, err)
+					logrus.Warnf("failed to create %q from %s: %v", pathMapping.ContainerPath, service.Name, err)
 				}
 			}
 			fmt.Fprintf(d.infoWriter, "%s created\n", pathMapping.ContainerPath)
 		} else {
 			err := d.client.Copy(ctx, d.projectName, api.CopyOptions{
 				Source:      pathMapping.HostPath,
-				Destination: fmt.Sprintf("%s:%s", pathMapping.Service, pathMapping.ContainerPath),
+				Destination: fmt.Sprintf("%s:%s", service.Name, pathMapping.ContainerPath),
 			})
 			if err != nil {
 				return err
@@ -93,12 +93,12 @@ func (d *DockerCopy) sync(ctx context.Context, service types.ServiceConfig, path
 	} else if errors.Is(statErr, fs.ErrNotExist) {
 		for i := 1; i <= scale; i++ {
 			_, err := d.client.Exec(ctx, d.projectName, api.RunOptions{
-				Service: pathMapping.Service,
+				Service: service.Name,
 				Command: []string{"rm", "-rf", pathMapping.ContainerPath},
 				Index:   i,
 			})
 			if err != nil {
-				logrus.Warnf("failed to delete %q from %s: %v", pathMapping.ContainerPath, pathMapping.Service, err)
+				logrus.Warnf("failed to delete %q from %s: %v", pathMapping.ContainerPath, service.Name, err)
 			}
 		}
 		fmt.Fprintf(d.infoWriter, "%s deleted from service\n", pathMapping.ContainerPath)

internal/sync/shared.go

@@ -22,8 +22,6 @@ import (
 
 // PathMapping contains the Compose service and modified host system path.
 type PathMapping struct {
-	// Service that the file event is for.
-	Service string
 	// HostPath that was created/modified/deleted outside the container.
 	//
 	// This is the path as seen from the user's perspective, e.g.

internal/sync/tar.go

@@ -121,9 +121,7 @@ func (t *Tar) Sync(ctx context.Context, service types.ServiceConfig, paths []Pat
 }
 
 type ArchiveBuilder struct {
-	tw    *tar.Writer
-	paths []string // local paths archived
-
+	tw *tar.Writer
 	// A shared I/O buffer to help with file copying.
 	copyBuf *bytes.Buffer
 }
@@ -168,7 +166,6 @@ func (a *ArchiveBuilder) ArchivePathsIfExist(paths []PathMapping) error {
 		if err != nil {
 			return fmt.Errorf("archiving %q: %w", entry.path, err)
 		}
-		a.paths = append(a.paths, entry.path)
 	}
 	return nil
 }

pkg/compose/compose.go

@@ -26,6 +26,8 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/jonboulle/clockwork"
+
 	"github.com/docker/docker/api/types/volume"
 
 	"github.com/compose-spec/compose-go/types"
@@ -58,13 +60,15 @@ func init() {
 func NewComposeService(dockerCli command.Cli) api.Service {
 	return &composeService{
 		dockerCli:      dockerCli,
+		clock:          clockwork.NewRealClock(),
 		maxConcurrency: -1,
 		dryRun:         false,
 	}
 }
 
 type composeService struct {
 	dockerCli      command.Cli
+	clock          clockwork.Clock
 	maxConcurrency int
 	dryRun         bool
 }