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/status.go

@@ -101,9 +101,7 @@ func (a *Application) restart(proc *process.Info) {
  a.stopWatcher(proc)
 
  // kill the process
- if proc != nil && proc.Process != nil {
- _ = proc.Process.Kill()
- }
+ _ = proc.Process.Kill()
 
  if proc != a.state.ProcessInfo {
  // we're restarting different process than actually running

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,64 @@
+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
+}