Add exit_code metadata to the command processor (#349)

Inspired from: redpanda-data/connect#3766

Signed-off-by: Mihai Todor <todormihai@gmail.com>
Co-authored-by: Ashley Jeffs <ash@jeffail.uk>

Author: mihaitodor

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file.
 
 ### Added
 
+- The `command` processor now emits the `exit_code` metadata field. (@mihaitodor)
 - Go API: `Message` now exposes `MetaGetImmut` and `MetaSetImmut` methods for reading and writing metadata values of any type with immutable semantics. (@Jeffail)
 
 ### Fixed

internal/impl/io/processor_command.go

@@ -26,15 +26,25 @@ func commandProcSpec() *service.ConfigSpec {
 		Description(`
 The specified command is executed for each message processed, with the raw bytes of the message being fed into the stdin of the command process, and the resulting message having its contents replaced with the stdout of it.
 
+== Metadata
+
+This input adds the following metadata fields to each message:
+
+`+"```text"+`
+- command_stderr - Contains the stderr output of a successful command, if any.
+- exit_code - The exit code returned by the command.
+`+"```"+`
+
+You can access these metadata fields using
+xref:configuration:interpolation.adoc#bloblang-queries[function interpolation].
+
 == Performance
 
 Since this processor executes a new process for each message performance will likely be an issue for high throughput streams. If this is the case then consider using the xref:components:processors/subprocess.adoc[`+"`subprocess` processor"+`] instead as it keeps the underlying process alive long term and uses codecs to insert and extract inputs and outputs to it via stdin/stdout.
 
 == Error handling
 
 If a non-zero error code is returned by the command then an error containing the entirety of stderr (or a generic message if nothing is written) is set on the message. These failed messages will continue through the pipeline unchanged, but can be dropped or placed in a dead letter queue according to your config, you can read about xref:configuration:error_handling.adoc[these patterns].
-
-If the command is successful but stderr is written to then a metadata field `+"`command_stderr`"+` is populated with its contents.
 `).
 		Fields(
 			service.NewInterpolatedStringField(cpNameField).
@@ -148,13 +158,16 @@ func (c *commandProc) Process(ctx context.Context, msg *service.Message) (servic
 	err = cmd.Run()
 	outBytes, errBytes := stdout.Bytes(), stderr.Bytes()
 	if err != nil {
-		return nil, fmt.Errorf("execution error: %w: %s", err, errBytes)
+		msg.MetaSetMut("exit_code", cmd.ProcessState.ExitCode())
+		return service.MessageBatch{msg}, fmt.Errorf("execution error: %w: %s", err, errBytes)
 	}
 
 	msg.SetBytes(outBytes)
 	if len(errBytes) > 0 {
 		msg.MetaSet("command_stderr", string(errBytes))
 	}
+	msg.MetaSetMut("exit_code", 0)
+
 	return service.MessageBatch{msg}, nil
 }
 

internal/impl/io/processor_command_test.go

@@ -87,13 +87,21 @@ args_mapping: '[ "-n" ]'
 			require.NoError(t, err)
 
 			res, err := cmdProc.Process(tCtx, service.NewMessage([]byte(test.input)))
+
+			exitCode, ok := res[0].MetaGetMut("exit_code")
+			assert.True(t, ok)
+
 			if test.errContains != "" {
 				require.Error(t, err)
 				assert.Contains(t, err.Error(), test.errContains)
+
+				assert.Equal(t, 2, exitCode)
 			} else {
 				require.NoError(t, err)
 				require.Len(t, res, 1)
 
+				assert.Equal(t, 0, exitCode)
+
 				resBytes, err := res[0].AsBytes()
 				require.NoError(t, err)
 				assert.Contains(t, string(resBytes), test.outputContains)