Add Sentry Spans for Bash execution.

internal/nomad/nomad.go

@@ -398,7 +398,8 @@ func (a *APIClient) ExecuteCommand(jobID string,
 		return a.executeCommandInteractivelyWithStderr(jobID, ctx, command, privilegedExecution, stdin, stdout, stderr)
 	}
 	command = prepareCommandWithoutTTY(command, privilegedExecution)
-	exitCode, err := a.apiQuerier.Execute(jobID, ctx, command, tty, stdin, stdout, stderr)
+	debugWriter := &SentryDebugWriter{Target: stdout, Ctx: ctx}
+	exitCode, err := a.apiQuerier.Execute(jobID, ctx, command, tty, stdin, debugWriter, stderr)
 	if err != nil {
 		return 1, fmt.Errorf("error executing command in job %s: %w", jobID, err)
 	}
@@ -422,8 +423,9 @@ func (a *APIClient) executeCommandInteractivelyWithStderr(allocationID string, c
 
 		// Catch stderr in separate execution.
 		logging.StartSpan("nomad.execute.stderr", "Execution for separate StdErr", ctx, func(ctx context.Context) {
+			debugWriterStderr := &SentryDebugWriter{Target: stderr, Ctx: ctx}
 			exit, err := a.Execute(allocationID, ctx, prepareCommandTTYStdErr(currentNanoTime, privilegedExecution), true,
-				nullio.Reader{Ctx: readingContext}, stderr, io.Discard)
+				nullio.Reader{Ctx: readingContext}, debugWriterStderr, io.Discard)
 			if err != nil {
 				log.WithError(err).WithField("runner", allocationID).Warn("Stderr task finished with error")
 			}
@@ -435,7 +437,8 @@ func (a *APIClient) executeCommandInteractivelyWithStderr(allocationID string, c
 	var exit int
 	var err error
 	logging.StartSpan("nomad.execute.tty", "Interactive Execution", ctx, func(ctx context.Context) {
-		exit, err = a.Execute(allocationID, ctx, command, true, stdin, stdout, io.Discard)
+		debugWriter := &SentryDebugWriter{Target: stdout, Ctx: ctx}
+		exit, err = a.Execute(allocationID, ctx, command, true, stdin, debugWriter, io.Discard)
 	})
 
 	// Wait until the stderr catch command finished to make sure we receive all output.
@@ -474,27 +477,42 @@ const (
 	UnprivilegedExecution = false
 )
 
-func prepareCommandWithoutTTY(commands []string, privilegedExecution bool) []string {
+func prepareCommandWithoutTTY(srcCommands []string, privilegedExecution bool) []string {
+	commands := make([]string, len(srcCommands)) // nozero The size is required for the copy.
+	copy(commands, srcCommands)
+
+	commands[len(commands)-1] = setInnerDebugMessages(commands[len(commands)-1])
 	commands = setUserCommand(commands, privilegedExecution)
 	commands[len(commands)-1] = fmt.Sprintf("'%s'", commands[len(commands)-1])
 	cmd := strings.Join(commands, " ")
-	return []string{"bash", "-c", fmt.Sprintf(unsetEnvironmentVariablesFormat, unsetEnvironmentVariablesShell, cmd)}
+	cmd = fmt.Sprintf(unsetEnvironmentVariablesFormat, unsetEnvironmentVariablesShell, cmd)
+	// Debug Message
+	cmd = fmt.Sprintf(timeDebugMessageFormatStart, "SetUserAndUnsetEnv", cmd)
+	return []string{"bash", "-c", cmd}
 }
 
-func prepareCommandTTY(commands []string, currentNanoTime int64, privilegedExecution bool) []string {
+func prepareCommandTTY(srcCommands []string, currentNanoTime int64, privilegedExecution bool) []string {
+	commands := make([]string, len(srcCommands)) // nozero The size is required for the copy.
+	copy(commands, srcCommands)
+
+	commands[len(commands)-1] = setInnerDebugMessages(commands[len(commands)-1])
 	commands = setUserCommand(commands, privilegedExecution)
 	// Take the last command which is the one to be executed and wrap it to redirect stderr.
 	stderrFifoPath := stderrFifo(currentNanoTime)
 	commands[len(commands)-1] =
 		fmt.Sprintf(stderrWrapperCommandFormat, stderrFifoPath, commands[len(commands)-1], stderrFifoPath)
 	cmd := strings.Join(commands, " ")
-	return []string{"bash", "-c", fmt.Sprintf(unsetEnvironmentVariablesFormat, unsetEnvironmentVariablesShell, cmd)}
+	cmd = fmt.Sprintf(unsetEnvironmentVariablesFormat, unsetEnvironmentVariablesShell, cmd)
+	// Debug Message
+	cmd = fmt.Sprintf(timeDebugMessageFormatStart, "SetUserAndUnsetEnv", cmd)
+	return []string{"bash", "-c", cmd}
 }
 
 func prepareCommandTTYStdErr(currentNanoTime int64, privilegedExecution bool) []string {
 	stderrFifoPath := stderrFifo(currentNanoTime)
-	command := []string{"bash", "-c", fmt.Sprintf(stderrFifoCommandFormat, stderrFifoPath, stderrFifoPath, stderrFifoPath)}
-	return setUserCommand(command, privilegedExecution)
+	cmd := fmt.Sprintf(stderrFifoCommandFormat, stderrFifoPath, stderrFifoPath, stderrFifoPath)
+	cmd = setInnerDebugMessages(cmd)
+	return setUserCommand([]string{"bash", "-c", cmd}, privilegedExecution)
 }
 
 // setUserCommand prefixes the passed command with the setUser command.
@@ -510,3 +528,9 @@ func setUserCommand(command []string, privilegedExecution bool) []string {
 func stderrFifo(id int64) string {
 	return fmt.Sprintf(stderrFifoFormat, id)
 }
+
+func setInnerDebugMessages(command string) (result string) {
+	result = fmt.Sprintf(timeDebugMessageFormatStart, "innerCommand", command)
+	result = strings.TrimSuffix(result, ";")
+	return fmt.Sprintf(timeDebugMessageFormatEnd, result, "End")
+}

internal/nomad/nomad_test.go

@@ -708,6 +708,7 @@ func (s *ExecuteCommandTestSuite) TestWithSeparateStderr() {
 		s.Require().NotNil(calledStdoutCommand)
 		stderrWrapperCommand := fmt.Sprintf(stderrWrapperCommandFormat, stderrFifoFormat, s.testCommand, stderrFifoFormat)
 		stdoutFifoRegexp := strings.ReplaceAll(regexp.QuoteMeta(stderrWrapperCommand), "%d", "\\d*")
+		stdoutFifoRegexp = strings.Replace(stdoutFifoRegexp, s.testCommand, ".*", 1)
 		s.Regexp(stdoutFifoRegexp, calledStdoutCommand[len(calledStdoutCommand)-1])
 	})
 

internal/nomad/sentry_debug_writer.go

@@ -0,0 +1,89 @@
+package nomad
+
+import (
+	"context"
+	"fmt"
+	"github.com/getsentry/sentry-go"
+	"io"
+	"regexp"
+	"strconv"
+	"time"
+)
+
+const (
+	// timeDebugMessageFormat is the format of messages that will be converted to debug messages.
+	timeDebugMessageFormat = "echo -ne \"\\x1EPoseidon %s $(date +%%s%%3N)\\x1E\""
+	// Format Parameters: 1. Debug Comment, 2. command.
+	timeDebugMessageFormatStart = timeDebugMessageFormat + "; %s"
+	// Format Parameters: 1. command, 2. Debug Comment.
+	timeDebugMessageFormatEnd = "%s && " + timeDebugMessageFormat
+
+	timeDebugMessagePatternGroupText = 1
+	timeDebugMessagePatternGroupTime = 2
+)
+
+var (
+	timeDebugMessagePattern      = regexp.MustCompile(`\x1EPoseidon (?P<text>\w+) (?P<time>\d+)\x1E`)
+	timeDebugMessagePatternStart = regexp.MustCompile(`\x1EPoseidon`)
+)
+
+// SentryDebugWriter is scanning the input for the debug message pattern.
+// For matches, it creates a Sentry Span. Otherwise, the data will be forwarded to the Target.
+// The passed context Ctx should contain the Sentry data.
+type SentryDebugWriter struct {
+	Target   io.Writer
+	Ctx      context.Context
+	lastSpan *sentry.Span
+}
+
+// Improve: Handling of a split debug messages (usually p is exactly one debug message, not less and not more).
+func (s *SentryDebugWriter) Write(p []byte) (n int, err error) {
+	if !timeDebugMessagePatternStart.Match(p) {
+		count, err := s.Target.Write(p)
+		if err != nil {
+			err = fmt.Errorf("SentryDebugWriter Forwarded: %w", err)
+		}
+		return count, err
+	}
+
+	loc := timeDebugMessagePattern.FindIndex(p)
+	if loc == nil {
+		log.WithField("data", p).Warn("Exec debug message could not be read completely")
+		return 0, nil
+	}
+
+	go s.handleTimeDebugMessage(p[loc[0]:loc[1]])
+
+	debugMessageLength := loc[1] - loc[0]
+	if debugMessageLength < len(p) {
+		count, err := s.Write(append(p[0:loc[0]], p[loc[1]:]...))
+		return debugMessageLength + count, err
+	} else {
+		return debugMessageLength, nil
+	}
+}
+
+func (s *SentryDebugWriter) handleTimeDebugMessage(message []byte) {
+	if s.lastSpan != nil {
+		s.lastSpan.Finish()
+	}
+
+	matches := timeDebugMessagePattern.FindSubmatch(message)
+	if matches == nil {
+		log.WithField("msg", message).Error("Cannot parse passed time debug message")
+		return
+	}
+
+	timestamp, err := strconv.ParseInt(string(matches[timeDebugMessagePatternGroupTime]), 10, 64)
+	if err != nil {
+		log.WithField("matches", matches).Warn("Could not parse Unix timestamp")
+		return
+	}
+	s.lastSpan = sentry.StartSpan(s.Ctx, "nomad.execute.pipe")
+	s.lastSpan.Description = string(matches[timeDebugMessagePatternGroupText])
+	s.lastSpan.StartTime = time.UnixMilli(timestamp)
+	s.lastSpan.Finish()
+
+	s.lastSpan = sentry.StartSpan(s.Ctx, "nomad.execute.bash")
+	s.lastSpan.Description = string(matches[timeDebugMessagePatternGroupText])
+}

internal/nomad/sentry_debug_writer_test.go

@@ -0,0 +1,21 @@
+package nomad
+
+import (
+	"bytes"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestSentryDebugWriter_Write(t *testing.T) {
+	buf := &bytes.Buffer{}
+	w := SentryDebugWriter{Target: buf}
+
+	description := "TestDebugMessageDescription"
+	data := "\x1EPoseidon " + description + " 1676646791482\x1E"
+	count, err := w.Write([]byte(data))
+
+	require.NoError(t, err)
+	assert.Equal(t, len(data), count)
+	assert.NotContains(t, buf.String(), description)
+}