roachprod: return structured data from roachprod.Monitor

Previously, the `Monitor` function would return semi-structured data
in the form of a `(node, message)` pair. The `message` field
corresponded to the output (as in, stdout) of the bash script that
implements the monitor logic. This tied callers to the specific
implementation of the function and to the strings passed to `echo` in
that script (e.g., all the checks for the `"dead"` string in the
message).

This commit updates the function to instead return an event
channel. Events can be of different types corresponding to each of the
events that the shell script is able to emit (e.g., when the cockroach
process is found to be running, when it stops, etc). This makes the
parsing of the script output private, and simplifies the logic in most
callers.

The message returned by the `roachtest` wrapper is also changed
slightly in this commit to make it clearer where the error is coming
from: the function passed to `monitor.Go` or the monitor process
itself. This has been a source of confusion in the past.

Finally, the monitor implementation in roachprod is changed slightly
to avoid blocking on channel sends if the context has already been
canceled. This avoids leaked goroutines in tests, as canceling the
context passed to `Monitor()` should cause all goroutines to
eventually finish instead of blocking indefinitely.

Epic: CRDB-19321

Release note: None

Author: Renato Costa

pkg/cmd/roachprod/main.go

@@ -650,21 +650,15 @@ of nodes, outputting a line whenever a change is detected:
 `,
 	Args: cobra.ExactArgs(1),
 	Run: wrap(func(cmd *cobra.Command, args []string) error {
-		messages, err := roachprod.Monitor(context.Background(), config.Logger, args[0], monitorOpts)
+		eventChan, err := roachprod.Monitor(context.Background(), config.Logger, args[0], monitorOpts)
 		if err != nil {
 			return err
 		}
-		for msg := range messages {
-			if msg.Err != nil {
-				msg.Msg += "error: " + msg.Err.Error()
-			}
-			thisError := errors.Newf("%d: %s", msg.Node, msg.Msg)
-			if msg.Err != nil || strings.Contains(msg.Msg, "dead") {
-				err = errors.CombineErrors(err, thisError)
-			}
-			fmt.Println(thisError.Error())
+		for info := range eventChan {
+			fmt.Println(info.String())
 		}
-		return err
+
+		return nil
 	}),
 }
 

pkg/cmd/roachtest/cluster.go

@@ -1528,22 +1528,20 @@ func (c *clusterImpl) assertNoDeadNode(ctx context.Context, t test.Test) error {
 	}
 
 	t.L().Printf("checking for dead nodes")
-	ch, err := roachprod.Monitor(ctx, t.L(), c.name, install.MonitorOpts{OneShot: true, IgnoreEmptyNodes: true})
+	eventsCh, err := roachprod.Monitor(ctx, t.L(), c.name, install.MonitorOpts{OneShot: true, IgnoreEmptyNodes: true})
 
 	// An error here means there was a problem initialising a SyncedCluster.
 	if err != nil {
 		return err
 	}
 
 	deadNodes := 0
-	for n := range ch {
-		// If there's an error, it means either that the monitor command failed
-		// completely, or that it found a dead node worth complaining about.
-		if n.Err != nil || strings.HasPrefix(n.Msg, "dead") {
+	for info := range eventsCh {
+		t.L().Printf("%s", info)
+
+		if _, isDeath := info.Event.(install.MonitorNodeDead); isDeath {
 			deadNodes++
 		}
-
-		t.L().Printf("n%d: err=%v,msg=%s", n.Node, n.Err, n.Msg)
 	}
 
 	if deadNodes > 0 {

pkg/cmd/roachtest/monitor.go

@@ -13,7 +13,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -178,7 +177,7 @@ func (m *monitorImpl) wait() error {
 			m.cancel()
 			wg.Done()
 		}()
-		setErr(errors.Wrap(m.g.Wait(), "monitor task failed"))
+		setErr(errors.Wrap(m.g.Wait(), "function passed to monitor.Go failed"))
 	}()
 
 	// 2. The second goroutine reads from the monitoring channel, watching for any
@@ -190,28 +189,24 @@ func (m *monitorImpl) wait() error {
 			wg.Done()
 		}()
 
-		messagesChannel, err := roachprod.Monitor(m.ctx, m.l, m.nodes, install.MonitorOpts{})
+		eventsCh, err := roachprod.Monitor(m.ctx, m.l, m.nodes, install.MonitorOpts{})
 		if err != nil {
 			setErr(errors.Wrap(err, "monitor command failure"))
 			return
 		}
-		var monitorErr error
-		for msg := range messagesChannel {
-			if msg.Err != nil {
-				msg.Msg += "error: " + msg.Err.Error()
-			}
-			thisError := errors.Newf("%d: %s", msg.Node, msg.Msg)
-			if msg.Err != nil || strings.Contains(msg.Msg, "dead") {
-				monitorErr = errors.CombineErrors(monitorErr, thisError)
+
+		for info := range eventsCh {
+			_, isDeath := info.Event.(install.MonitorNodeDead)
+			isExpectedDeath := isDeath && atomic.AddInt32(&m.expDeaths, -1) >= 0
+			var expectedDeathStr string
+			if isExpectedDeath {
+				expectedDeathStr = ": expected"
 			}
-			var id int
-			var s string
-			newMsg := thisError.Error()
-			if n, _ := fmt.Sscanf(newMsg, "%d: %s", &id, &s); n == 2 {
-				if strings.Contains(s, "dead") && atomic.AddInt32(&m.expDeaths, -1) < 0 {
-					setErr(errors.Wrap(fmt.Errorf("unexpected node event: %s", newMsg), "monitor command failure"))
-					return
-				}
+			m.l.Printf("Monitor event: %s%s", info, expectedDeathStr)
+
+			if isDeath && !isExpectedDeath {
+				setErr(fmt.Errorf("unexpected node event: %s", info))
+				return
 			}
 		}
 	}()

pkg/roachprod/install/cluster_synced.go

@@ -583,20 +583,52 @@ fi
 	return results, nil
 }
 
+// MonitorNodeSkipped represents a node whose status was not checked.
+type MonitorNodeSkipped struct{}
+
+// MonitorNodeRunning represents the cockroach process running on a
+// node.
+type MonitorNodeRunning struct {
+	PID string
+}
+
+// MonitorNodeDead represents the cockroach process dying on a node.
+type MonitorNodeDead struct {
+	ExitCode string
+}
+
+type MonitorError struct {
+	Err error
+}
+
 // NodeMonitorInfo is a message describing a cockroach process' status.
 type NodeMonitorInfo struct {
 	// The index of the node (in a SyncedCluster) at which the message originated.
 	Node Node
-	// A message about the node. This is either a PID, "dead", "nc exited", or
-	// "skipped".
-	// Anything but a PID or "skipped" is an indication that there is some
-	// problem with the node and that the process is not running.
-	Msg string
-	// Err is an error that may occur when trying to probe the status of the node.
-	// If Err is non-nil, Msg is empty. After an error is returned, the node with
-	// the given index will no longer be probed. Errors typically indicate networking
-	// issues or nodes that have (physically) shut down.
-	Err error
+	// Event describes what happened to the node; it is one of
+	// MonitorNodeSkipped (no store directory was found);
+	// MonitorNodeRunning, sent when cockroach is running on a node;
+	// MonitorNodeDead, when the cockroach process stops running on a
+	// node; or MonitorError, typically indicate networking issues
+	// or nodes that have (physically) shut down.
+	Event interface{}
+}
+
+func (nmi NodeMonitorInfo) String() string {
+	var status string
+
+	switch event := nmi.Event.(type) {
+	case MonitorNodeRunning:
+		status = fmt.Sprintf("cockroach process is running (PID: %s)", event.PID)
+	case MonitorNodeSkipped:
+		status = "node skipped"
+	case MonitorNodeDead:
+		status = fmt.Sprintf("cockroach process died (exit code %s)", event.ExitCode)
+	case MonitorError:
+		status = fmt.Sprintf("error: %s", event.Err.Error())
+	}
+
+	return fmt.Sprintf("n%d: %s", nmi.Node, status)
 }
 
 // MonitorOpts is used to pass the options needed by Monitor.
@@ -606,16 +638,16 @@ type MonitorOpts struct {
 }
 
 // Monitor writes NodeMonitorInfo for the cluster nodes to the returned channel.
-// Infos sent to the channel always have the Index and exactly one of Msg or Err
-// set.
+// Infos sent to the channel always have the Node the event refers to, and the
+// event itself. See documentation for NodeMonitorInfo for possible event types.
 //
-// If oneShot is true, infos are retrieved only once for each node and the
+// If OneShot is true, infos are retrieved only once for each node and the
 // channel is subsequently closed; otherwise the process continues indefinitely
 // (emitting new information as the status of the cockroach process changes).
 //
-// If ignoreEmptyNodes is true, nodes on which no CockroachDB data is found
-// (in {store-dir}) will not be probed and single message, "skipped", will
-// be emitted for them.
+// If IgnoreEmptyNodes is true, nodes on which no CockroachDB data is found
+// (in {store-dir}) will not be probed and single event, MonitorNodeSkipped,
+// will be emitted for them.
 func (c *SyncedCluster) Monitor(
 	l *logger.Logger, ctx context.Context, opts MonitorOpts,
 ) chan NodeMonitorInfo {
@@ -624,10 +656,30 @@ func (c *SyncedCluster) Monitor(
 	var wg sync.WaitGroup
 	monitorCtx, cancel := context.WithCancel(ctx)
 
+	// sendEvent sends the NodeMonitorInfo passed through the channel
+	// that is listened to by the caller. Bails if the context is
+	// canceled.
+	sendEvent := func(info NodeMonitorInfo) {
+		select {
+		case ch <- info:
+			// We were able to send the info through the channel.
+		case <-monitorCtx.Done():
+			// Don't block trying to send the info.
+		}
+	}
+
+	const (
+		separator  = "|"
+		skippedMsg = "skipped"
+		runningMsg = "running"
+		deadMsg    = "dead"
+	)
+
 	for i := range nodes {
 		wg.Add(1)
 		go func(i int) {
 			defer wg.Done()
+
 			node := nodes[i]
 
 			// On each monitored node, we loop looking for a cockroach process.
@@ -637,18 +689,30 @@ func (c *SyncedCluster) Monitor(
 				Store       string
 				Port        int
 				Local       bool
+				Separator   string
+				SkippedMsg  string
+				RunningMsg  string
+				DeadMsg     string
 			}{
 				OneShot:     opts.OneShot,
 				IgnoreEmpty: opts.IgnoreEmptyNodes,
 				Store:       c.NodeDir(node, 1 /* storeIndex */),
 				Port:        c.NodePort(node),
 				Local:       c.IsLocal(),
+				Separator:   separator,
+				SkippedMsg:  skippedMsg,
+				RunningMsg:  runningMsg,
+				DeadMsg:     deadMsg,
 			}
 
+			// NB.: we parse the output of every line this script
+			// prints. Every call to `echo` must match the parsing logic
+			// down below in order to produce structured results to the
+			// caller.
 			snippet := `
 {{ if .IgnoreEmpty }}
 if ! ls {{.Store}}/marker.* 1> /dev/null 2>&1; then
-  echo "skipped"
+  echo "{{.SkippedMsg}}"
   exit 0
 fi
 {{- end}}
@@ -682,10 +746,10 @@ while :; do
         # the new incarnation. We lost the actual exit status of the old PID.
         status="unknown"
       fi
-    	echo "dead (exit status ${status})"
+    	echo "{{.DeadMsg}}{{.Separator}}${status}"
     fi
 		if [ "${pid}" != 0 ]; then
-			echo "${pid}"
+			echo "{{.RunningMsg}}{{.Separator}}${pid}"
     fi
     lastpid=${pid}
   fi
@@ -704,7 +768,8 @@ done
 			t := template.Must(template.New("script").Parse(snippet))
 			var buf bytes.Buffer
 			if err := t.Execute(&buf, data); err != nil {
-				ch <- NodeMonitorInfo{Node: node, Err: err}
+				err := errors.Wrap(err, "failed to execute template")
+				sendEvent(NodeMonitorInfo{Node: node, Event: MonitorError{err}})
 				return
 			}
 
@@ -713,14 +778,16 @@ done
 
 			p, err := sess.StdoutPipe()
 			if err != nil {
-				ch <- NodeMonitorInfo{Node: node, Err: err}
+				err := errors.Wrap(err, "failed to read stdout pipe")
+				sendEvent(NodeMonitorInfo{Node: node, Event: MonitorError{err}})
 				wg.Done()
 				return
 			}
 			// Request a PTY so that the script will receive a SIGPIPE when the
 			// session is closed.
 			if err := sess.RequestPty(); err != nil {
-				ch <- NodeMonitorInfo{Node: node, Err: err}
+				err := errors.Wrap(err, "failed to request PTY")
+				sendEvent(NodeMonitorInfo{Node: node, Event: MonitorError{err}})
 				return
 			}
 
@@ -734,12 +801,31 @@ done
 					if err == io.EOF {
 						return
 					}
-					ch <- NodeMonitorInfo{Node: node, Msg: string(line)}
+					if err != nil {
+						err := errors.Wrap(err, "error reading from session")
+						sendEvent(NodeMonitorInfo{Node: node, Event: MonitorError{err}})
+					}
+
+					parts := strings.Split(string(line), separator)
+					switch parts[0] {
+					case skippedMsg:
+						sendEvent(NodeMonitorInfo{Node: node, Event: MonitorNodeSkipped{}})
+					case runningMsg:
+						pid := parts[1]
+						sendEvent(NodeMonitorInfo{Node: node, Event: MonitorNodeRunning{pid}})
+					case deadMsg:
+						exitCode := parts[1]
+						sendEvent(NodeMonitorInfo{Node: node, Event: MonitorNodeDead{exitCode}})
+					default:
+						err := fmt.Errorf("internal error: unrecognized output from monitor: %s", line)
+						sendEvent(NodeMonitorInfo{Node: node, Event: MonitorError{err}})
+					}
 				}
 			}(p)
 
 			if err := sess.Start(); err != nil {
-				ch <- NodeMonitorInfo{Node: node, Err: err}
+				err := errors.Wrap(err, "failed to start session")
+				sendEvent(NodeMonitorInfo{Node: node, Event: MonitorError{err}})
 				return
 			}
 
@@ -755,7 +841,8 @@ done
 			// pipe. Otherwise it can be closed under us, causing the reader to loop
 			// infinitely receiving a non-`io.EOF` error.
 			if err := sess.Wait(); err != nil {
-				ch <- NodeMonitorInfo{Node: node, Err: err}
+				err := errors.Wrap(err, "failed to wait for session")
+				sendEvent(NodeMonitorInfo{Node: node, Event: MonitorError{err}})
 				return
 			}
 		}(i)

pkg/roachprod/install/session.go

@@ -112,7 +112,6 @@ func newRemoteSession(l *logger.Logger, command *remoteCommand) *remoteSession {
 		}
 	}
 
-	//const logfile = ""
 	args := []string{
 		command.user + "@" + command.host,