commands support & parallel lock

archive.go

@@ -6,39 +6,40 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"sync"
 	"syscall"
 
 	"github.com/seletskiy/hierr"
 )
 
 func startArchiveReceivers(
 	lockedNodes *distributedLock,
-	args map[string]interface{},
-) (*archiveReceivers, error) {
-	var (
-		rootDir = args["--root"].(string)
-	)
-
+	rootDir string,
+) (*remoteExecution, error) {
 	archiveReceiverCommandString := fmt.Sprintf(
 		`tar -x --verbose --directory="%s"`,
 		rootDir,
 	)
 
 	unpackers := []io.WriteCloser{}
 
-	nodes := []archiveReceiverNode{}
+	nodes := []remoteExecutionNode{}
+
+	logMutex := &sync.Mutex{}
 
 	for _, node := range lockedNodes.nodes {
-		debugf(hierr.Errorf(
-			archiveReceiverCommandString,
-			"%s starting archive receiver command",
-			node.String(),
-		).Error())
+		tracef(
+			"%s",
+			hierr.Errorf(
+				archiveReceiverCommandString,
+				"%s starting archive receiver command",
+				node.String(),
+			).Error(),
+		)
 
 		archiveReceiverCommand, err := node.runner.Command(
 			archiveReceiverCommandString,
 		)
-
 		if err != nil {
 			return nil, hierr.Errorf(
 				err,
@@ -57,21 +58,24 @@ func startArchiveReceivers(
 		unpackers = append(unpackers, stdin)
 
 		stdout := newLineFlushWriter(
+			logMutex,
 			newPrefixWriter(
 				newDebugWriter(logger),
-				fmt.Sprintf("%s {tar} <stdout> ", node.String()),
+				node.String()+" {tar} <stdout> ",
 			),
+			true,
 		)
 
-		archiveReceiverCommand.SetStdout(stdout)
-
 		stderr := newLineFlushWriter(
+			logMutex,
 			newPrefixWriter(
 				newDebugWriter(logger),
-				fmt.Sprintf("%s {tar} <stderr> ", node.String()),
+				node.String()+" {tar} <stderr> ",
 			),
+			true,
 		)
 
+		archiveReceiverCommand.SetStdout(stdout)
 		archiveReceiverCommand.SetStderr(stderr)
 
 		err = archiveReceiverCommand.Start()
@@ -82,13 +86,16 @@ func startArchiveReceivers(
 			)
 		}
 
-		nodes = append(nodes, archiveReceiverNode{
+		nodes = append(nodes, remoteExecutionNode{
 			node:    node,
 			command: archiveReceiverCommand,
+
+			stdout: stdout,
+			stderr: stderr,
 		})
 	}
 
-	return &archiveReceivers{
+	return &remoteExecution{
 		stdin: multiWriteCloser{unpackers},
 		nodes: nodes,
 	}, nil
@@ -106,7 +113,7 @@ func archiveFilesToWriter(target io.Writer, files []string) error {
 	archive := tar.NewWriter(target)
 	for fileIndex, fileName := range files {
 		logger.Infof(
-			"(%d/%d) sending file: '%s'",
+			"%5d/%d sending file: '%s'",
 			fileIndex+1,
 			len(files),
 			fileName,
@@ -115,7 +122,7 @@ func archiveFilesToWriter(target io.Writer, files []string) error {
 		writeFileToArchive(fileName, archive, workDir)
 	}
 
-	debugf("closing archive stream, %d files sent", len(files))
+	tracef("closing archive stream, %d files sent", len(files))
 
 	err = archive.Close()
 	if err != nil {
@@ -168,7 +175,7 @@ func writeFileToArchive(
 		ModTime: fileInfo.ModTime(),
 	}
 
-	debugf(
+	tracef(
 		hierr.Errorf(
 			fmt.Sprintf(
 				"size: %d bytes; mode: %o; uid/gid: %d/%d; modtime: %s",

command.go

@@ -0,0 +1,137 @@
+package main
+
+import (
+	"io"
+	"os"
+	"strings"
+	"sync"
+
+	"github.com/seletskiy/hierr"
+)
+
+func runCommand(
+	lockedNodes *distributedLock,
+	command []string,
+	verbosityLevel verbosity,
+) error {
+	commandString := joinCommand(command)
+
+	remoteCommands := map[*distributedLockNode]*remoteExecutionNode{}
+
+	logMutex := &sync.Mutex{}
+
+	for _, node := range lockedNodes.nodes {
+		tracef(
+			"%s",
+			hierr.Errorf(
+				commandString,
+				"%s starting command",
+				node.String(),
+			).Error(),
+		)
+
+		remoteCommand, err := node.runner.Command(
+			commandString,
+		)
+		if err != nil {
+			return hierr.Errorf(
+				err,
+				`can't create remote command`,
+			)
+		}
+
+		var stdout io.WriteCloser
+		var stderr io.WriteCloser
+		switch verbosityLevel {
+		default:
+			stdout = newLineFlushWriter(
+				logMutex,
+				newPrefixWriter(
+					os.Stdout,
+					node.address.domain+" ",
+				),
+				true,
+			)
+
+			stderr = newLineFlushWriter(
+				logMutex,
+				newPrefixWriter(
+					os.Stderr,
+					node.address.domain+" ",
+				),
+				true,
+			)
+
+		case verbosityQuiet:
+			stdout = newLineFlushWriter(logMutex, os.Stdout, false)
+			stderr = newLineFlushWriter(logMutex, os.Stderr, false)
+
+		case verbosityDebug:
+			stdout = newLineFlushWriter(
+				logMutex,
+				newPrefixWriter(
+					newDebugWriter(logger),
+					node.String()+" {cmd} <stdout> ",
+				),
+				false,
+			)
+
+			stderr = newLineFlushWriter(
+				logMutex,
+				newPrefixWriter(
+					newDebugWriter(logger),
+					node.String()+" {cmd} <stderr> ",
+				),
+				false,
+			)
+		}
+
+		remoteCommand.SetStdout(stdout)
+		remoteCommand.SetStderr(stderr)
+
+		err = remoteCommand.Start()
+		if err != nil {
+			return hierr.Errorf(
+				err,
+				`can't start remote command`,
+			)
+		}
+
+		remoteCommands[node] = &remoteExecutionNode{
+			node:    node,
+			command: remoteCommand,
+
+			stdout: stdout,
+			stderr: stderr,
+		}
+	}
+
+	for node, remoteCommand := range remoteCommands {
+		err := remoteCommand.command.Wait()
+		_ = remoteCommand.stdout.Close()
+		_ = remoteCommand.stderr.Close()
+		if err != nil {
+			return hierr.Errorf(
+				err,
+				`%s can't wait for remote command to finish: '%s'`,
+				node.String(),
+				commandString,
+			)
+		}
+	}
+
+	return nil
+}
+
+func joinCommand(command []string) string {
+	escapedParts := []string{}
+
+	for _, part := range command {
+		part = strings.Replace(part, `"`, `\"`, -1)
+		part = strings.Replace(part, `\`, `\\`, -1)
+
+		escapedParts = append(escapedParts, part)
+	}
+
+	return strings.Join(escapedParts, " ")
+}