Collect process (beats) stdout and stderr

CHANGELOG.next.asciidoc

@@ -101,6 +101,7 @@
 - Add fleet-server to output of elastic-agent inspect output command (and diagnostic bundle). {pull}243[243]
 - Update API calls that the agent makes to Kibana when running the container command. {pull}253[253]
 - diagnostics collect log names are fixed on Windows machines, command will ignore failures. AgentID is included in diagnostics(and diagnostics collect) output. {issue}81[81] {issue}92[92] {issue}190[190] {pull}262[262]
+- Collects stdout and stderr of applications run as a process and logs them. {issue}[88]
 
 ==== New features
 

internal/pkg/agent/control/client/client.go

@@ -8,7 +8,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-
 	"sync"
 	"time"
 
@@ -44,7 +43,8 @@ type Version struct {
 	Snapshot  bool
 }
 
-// ApplicationStatus is a status of an application inside of Elastic Agent.
+// ApplicationStatus is a status of an application managed by the Elastic Agent.
+// TODO(Anderson): Implement sort.Interface and sort it.
 type ApplicationStatus struct {
 	ID      string
 	Name    string

internal/pkg/core/plugin/process/configure.go

@@ -51,7 +51,7 @@ func (a *Application) Configure(ctx context.Context, config map[string]interface
 		a.appLock.Unlock()
 		a.Stop()
 		err = a.Start(ctx, a.desc, config)
-		// lock back so it wont panic on deferred unlock
+		// lock back so it won't panic on deferred unlock
 		a.appLock.Lock()
 	}
 

internal/pkg/core/plugin/process/start.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os/exec"
 	"path/filepath"
 
 	"gopkg.in/yaml.v2"
@@ -132,7 +133,10 @@ func (a *Application) start(ctx context.Context, t app.Taggable, cfg map[string]
 		a.processConfig,
 		a.uid,
 		a.gid,
-		spec.Args)
+		spec.Args, func(c *exec.Cmd) {
+			c.Stdout = newLoggerWriter(a.Name(), logStdOut, a.logger)
+			c.Stderr = newLoggerWriter(a.Name(), logStdErr, a.logger)
+		})
 	if err != nil {
 		return fmt.Errorf("%q failed to start %q: %w",
 			a.Name(), spec.BinaryPath, err)

internal/pkg/core/plugin/process/stdlogger.go

@@ -0,0 +1,52 @@
+package process
+
+import (
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+)
+
+type logStd int
+
+const (
+	agentConsoleName = "agent.console.name"
+	agentConsoleType = "agent.console.type"
+
+	logStdOut logStd = iota
+	logStdErr
+)
+
+func (l logStd) String() string {
+	switch l {
+	case logStdOut:
+		return "stdout"
+	case logStdErr:
+		return "stderr"
+	}
+
+	return "unknown"
+}
+
+type loggerWriter struct {
+	format string
+	logf   func(format string, args ...interface{})
+}
+
+func newLoggerWriter(appName string, std logStd, log *logger.Logger) loggerWriter {
+	log = log.With(
+		agentConsoleName, appName,
+		agentConsoleType, std.String())
+
+	logf := log.Infof
+	if std == logStdErr {
+		logf = log.Errorf
+	}
+
+	return loggerWriter{
+		format: appName + " " + std.String() + ": %q",
+		logf:   logf,
+	}
+}
+
+func (l loggerWriter) Write(p []byte) (n int, err error) {
+	l.logf(l.format, string(p))
+	return len(p), nil
+}

internal/pkg/core/plugin/process/stdlogger_test.go

@@ -0,0 +1,66 @@
+package process
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zapcore"
+
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+)
+
+func Test_loggerWriter(t *testing.T) {
+	tc := []struct {
+		name string
+		args struct {
+			appName string
+			logTo   logStd
+		}
+		logMsg   string
+		logLevel zapcore.Level
+	}{
+		{
+			name: "capture stdout",
+			args: struct {
+				appName string
+				logTo   logStd
+			}{
+				appName: "somebeats",
+				logTo:   logStdOut,
+			},
+			logMsg:   "stdout log",
+			logLevel: zapcore.InfoLevel,
+		},
+		{
+			name: "capture stderr",
+			args: struct {
+				appName string
+				logTo   logStd
+			}{
+				appName: "somebeats",
+				logTo:   logStdErr,
+			},
+			logMsg:   "stderr log",
+			logLevel: zapcore.ErrorLevel,
+		},
+	}
+
+	for _, tt := range tc {
+		logg, obs := logger.NewTesting("test-loggerWriter")
+		logg = logg.With("previous-field", "previous-value")
+
+		l := newLoggerWriter(tt.args.appName, tt.args.logTo, logg)
+		_, _ = l.Write([]byte(tt.logMsg))
+
+		logs := obs.All()
+		require.Equal(t, 1, len(logs))
+
+		log := logs[0]
+		assert.Equal(t, log.Level, tt.logLevel)
+		assert.Contains(t, log.Message, tt.logMsg)
+		assert.Equal(t, log.ContextMap()[agentConsoleName], tt.args.appName)
+		assert.Equal(t, log.ContextMap()[agentConsoleType], tt.args.logTo.String())
+		assert.Equal(t, log.ContextMap()["previous-field"], "previous-value")
+	}
+}

pkg/core/logger/testing.go

@@ -0,0 +1,21 @@
+package logger
+
+import (
+	"github.com/elastic/elastic-agent-libs/logp"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
+)
+
+// NewTesting creates a testing logger that buffers the logs in memory and
+// logs in debug level. Check observer.ObservedLogs for more details.
+func NewTesting(name string) (*Logger, *observer.ObservedLogs) {
+	core, obs := observer.New(zapcore.DebugLevel)
+
+	logger := logp.NewLogger(
+		name,
+		zap.WrapCore(func(in zapcore.Core) zapcore.Core {
+			return zapcore.NewTee(in, core)
+		}))
+	return logger, obs
+}