Event submission on run_integration exit with exit code (#259)

* feat: integrations run by cmd channel submit event when finished containing exit code

* fix centos5 build

* refactor: var rename

* feat: cmd_stop_mode = process-not-found

* fix: notify 0 exit code

* feat: exit code is -3 on unknown errors

* fix: old import

* Update pkg/integrations/track/ctx/ctx.go

Co-authored-by: Carlos <croman@newrelic.com>

* Update pkg/integrations/track/tracker_test.go

Co-authored-by: Carlos <croman@newrelic.com>

Co-authored-by: Carlos <croman@newrelic.com>

Author: varas

cmd/newrelic-infra/newrelic-infra.go

@@ -30,7 +30,7 @@ import (
 	"github.com/newrelic/infrastructure-agent/internal/integrations/v4/integration"
 	"github.com/newrelic/infrastructure-agent/internal/integrations/v4/v3legacy"
 	"github.com/newrelic/infrastructure-agent/internal/socketapi"
-	"github.com/newrelic/infrastructure-agent/pkg/integrations/stoppable"
+	"github.com/newrelic/infrastructure-agent/pkg/integrations/track"
 	"github.com/newrelic/infrastructure-agent/pkg/plugins"
 	"github.com/sirupsen/logrus"
 
@@ -288,14 +288,14 @@ func initializeAgentAndRun(c *config.Config, logFwCfg config.LogForward) error {
 	// queues integration run requests
 	definitionQ := make(chan integration.Definition, 100)
 
-	// track stoppable integrations
-	tracker := stoppable.NewTracker()
-
 	emitterWithRegister := dm.NewEmitter(agt.GetContext(), dmSender, registerClient)
 	nonRegisterEmitter := dm.NewNonRegisterEmitter(agt.GetContext(), dmSender)
 
 	dmEmitter := dm.NewEmitterWithFF(emitterWithRegister, nonRegisterEmitter, ffManager)
 
+	// track stoppable integrations
+	tracker := track.NewTracker(dmEmitter)
+
 	integrationEmitter := emitter.NewIntegrationEmittor(agt, dmEmitter, ffManager)
 	integrationManager := v4.NewManager(integrationCfg, integrationEmitter, il, definitionQ, tracker)
 

internal/agent/cmdchannel/runintegration/runintegration.go

@@ -14,13 +14,16 @@ import (
 	"github.com/newrelic/infrastructure-agent/internal/integrations/v4/integration"
 	"github.com/newrelic/infrastructure-agent/pkg/backend/commandapi"
 	"github.com/newrelic/infrastructure-agent/pkg/fwrequest"
+	ctx2 "github.com/newrelic/infrastructure-agent/pkg/integrations/track/ctx"
 	"github.com/newrelic/infrastructure-agent/pkg/integrations/v4/config"
 	"github.com/newrelic/infrastructure-agent/pkg/integrations/v4/dm"
 	"github.com/newrelic/infrastructure-agent/pkg/integrations/v4/protocol"
 	"github.com/newrelic/infrastructure-agent/pkg/log"
 	"github.com/newrelic/infrastructure-agent/pkg/trace"
 )
 
+const cmdName = "run_integration"
+
 // Errors
 var (
 	ErrNoIntName = errors.New("missing required \"integration_name\"")
@@ -56,7 +59,9 @@ func NewHandler(definitionQ chan<- integration.Definition, il integration.Instan
 			LogDecorated(logger, cmd, args).WithError(err).Warn("cannot create handler for cmd channel run_integration requests")
 			return
 		}
-		def.CmdChannelHash = args.Hash()
+
+		cmdChanReq := ctx2.NewCmdChannelRequest(cmdName, args.Hash(), args.IntegrationName, args.IntegrationArgs)
+		def.CmdChanReq = &cmdChanReq
 
 		definitionQ <- def
 
@@ -67,7 +72,7 @@ func NewHandler(definitionQ chan<- integration.Definition, il integration.Instan
 		return
 	}
 
-	return cmdchannel.NewCmdHandler("run_integration", handleF)
+	return cmdchannel.NewCmdHandler(cmdName, handleF)
 }
 
 func NotifyPlatform(dmEmitter dm.Emitter, def integration.Definition, ev protocol.EventData) {

internal/agent/cmdchannel/stopintegration/stopintegration.go

@@ -12,19 +12,21 @@ import (
 	"github.com/newrelic/infrastructure-agent/internal/agent/cmdchannel/runintegration"
 	"github.com/newrelic/infrastructure-agent/internal/integrations/v4/integration"
 	"github.com/newrelic/infrastructure-agent/pkg/backend/commandapi"
-	"github.com/newrelic/infrastructure-agent/pkg/integrations/stoppable"
+	"github.com/newrelic/infrastructure-agent/pkg/integrations/track"
+	"github.com/newrelic/infrastructure-agent/pkg/integrations/track/ctx"
 	"github.com/newrelic/infrastructure-agent/pkg/integrations/v4/dm"
 	"github.com/newrelic/infrastructure-agent/pkg/log"
 	"github.com/newrelic/infrastructure-agent/pkg/trace"
 	"github.com/shirou/gopsutil/process"
 )
 
 const (
+	cmdName                = "stop_integration"
 	terminationGracePeriod = 1 * time.Minute
 )
 
 // NewHandler creates a cmd-channel handler for stop-integration requests.
-func NewHandler(tracker *stoppable.Tracker, il integration.InstancesLookup, dmEmitter dm.Emitter, l log.Entry) *cmdchannel.CmdHandler {
+func NewHandler(tracker *track.Tracker, il integration.InstancesLookup, dmEmitter dm.Emitter, l log.Entry) *cmdchannel.CmdHandler {
 	handleF := func(ctx context.Context, cmd commandapi.Command, initialFetch bool) (err error) {
 		if runtime.GOOS == "windows" {
 			return cmdchannel.ErrOSNotSupported
@@ -58,6 +60,7 @@ func NewHandler(tracker *stoppable.Tracker, il integration.InstancesLookup, dmEm
 		p, err := process.NewProcess(int32(<-pidC))
 		if err != nil {
 			runintegration.LogDecorated(l, cmd, args).WithError(err).Warn("cannot retrieve process")
+			notifyPlatformWithLog(dmEmitter, il, cmd, args, "process-not-found", l)
 			return
 		}
 
@@ -85,18 +88,21 @@ func NewHandler(tracker *stoppable.Tracker, il integration.InstancesLookup, dmEm
 			stopModeUsed = "sigkill"
 		}
 
-		// notify platform
-		if err = notifyPlatform(dmEmitter, il, cmd, args, stopModeUsed); err != nil {
-			runintegration.LogDecorated(l, cmd, args).WithError(err).Warn("cannot notify platform about command")
-		}
+		notifyPlatformWithLog(dmEmitter, il, cmd, args, stopModeUsed, l)
 
 		// no further error handling required
 		err = nil
 
 		return
 	}
 
-	return cmdchannel.NewCmdHandler("stop_integration", handleF)
+	return cmdchannel.NewCmdHandler(cmdName, handleF)
+}
+
+func notifyPlatformWithLog(dmEmitter dm.Emitter, il integration.InstancesLookup, cmd commandapi.Command, args runintegration.RunIntArgs, stopModeUsed string, l log.Entry) {
+	if err := notifyPlatform(dmEmitter, il, cmd, args, stopModeUsed); err != nil {
+		runintegration.LogDecorated(l, cmd, args).WithError(err).Warn("cannot notify platform about command")
+	}
 }
 
 func notifyPlatform(dmEmitter dm.Emitter, il integration.InstancesLookup, cmd commandapi.Command, args runintegration.RunIntArgs, stopModeUsed string) error {
@@ -105,7 +111,8 @@ func notifyPlatform(dmEmitter dm.Emitter, il integration.InstancesLookup, cmd co
 		return err
 	}
 
-	def.CmdChannelHash = args.Hash()
+	ccReq := ctx.NewCmdChannelRequest(cmdName, cmd.Hash, args.IntegrationName, args.IntegrationArgs)
+	def.CmdChanReq = &ccReq
 	ev := cmd.Event(args.IntegrationName, args.IntegrationArgs)
 	ev["cmd_stop_hash"] = args.Hash()
 	ev["cmd_stop_mode"] = stopModeUsed

internal/agent/cmdchannel/stopintegration/stopintegration_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/newrelic/infrastructure-agent/internal/agent/cmdchannel/runintegration"
 	"github.com/newrelic/infrastructure-agent/internal/integrations/v4/integration"
 	"github.com/newrelic/infrastructure-agent/pkg/backend/commandapi"
-	"github.com/newrelic/infrastructure-agent/pkg/integrations/stoppable"
+	"github.com/newrelic/infrastructure-agent/pkg/integrations/track"
 	dm "github.com/newrelic/infrastructure-agent/pkg/integrations/v4/dm/testutils"
 	"github.com/newrelic/infrastructure-agent/pkg/log"
 	"github.com/shirou/gopsutil/process"
@@ -30,13 +30,14 @@ func TestHandle_returnsErrorOnMissingName(t *testing.T) {
 		t.Skip("CC stop-intergation is not supported on Windows")
 	}
 
-	h := NewHandler(stoppable.NewTracker(), integration.ErrLookup, dm.NewNoopEmitter(), l)
+	h := NewHandler(track.NewTracker(nil), integration.ErrLookup, dm.NewNoopEmitter(), l)
 
 	cmdArgsMissingPID := commandapi.Command{
 		Args: []byte(`{ "integration_args": ["foo"] }`),
 	}
 
 	err := h.Handle(context.Background(), cmdArgsMissingPID, false)
+	require.Error(t, err)
 	assert.Equal(t, cmdchannel.NewArgsErr(runintegration.ErrNoIntName).Error(), err.Error())
 }
 
@@ -46,12 +47,12 @@ func TestHandle_signalStopProcess(t *testing.T) {
 	}
 
 	// Given a handler with an stoppables tracker
-	tracker := stoppable.NewTracker()
+	tracker := track.NewTracker(nil)
 	h := NewHandler(tracker, integration.ErrLookup, dm.NewNoopEmitter(), l)
 
 	// When a process context is tracked
 	ctx := context.Background()
-	ctx, pidC := tracker.Track(ctx, "foo#")
+	ctx, pidC := tracker.Track(ctx, "foo#", nil)
 
 	proc := exec.CommandContext(ctx, "sleep", "5")
 

internal/gobackfill/gte_1_12.go

@@ -0,0 +1,18 @@
+// Copyright 2020 New Relic Corporation. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//
+// Golang code for Go versions greater or equal than 1.12.
+//
+// +build go1.12
+
+package gobackfill
+
+import "os/exec"
+
+// ExitCode wrapping to backfill old Golang versions.
+// Negative values mean:
+// * -1: not exited
+// * -2: unknown
+func ExitCode(exitError *exec.ExitError) int {
+	return exitError.ExitCode()
+}

internal/gobackfill/lt_1_12.go

@@ -0,0 +1,15 @@
+// Copyright 2020 New Relic Corporation. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//
+// Golang code for Go versions lower than 1.12.
+//
+// +build !go1.12
+
+package gobackfill
+
+import "os/exec"
+
+// ExitCode not supported, always value -2, as -1 belong to not finished.
+func ExitCode(exitError *exec.ExitError) int {
+	return -2
+}

internal/integrations/v4/executor/executor.go

@@ -13,11 +13,14 @@ import (
 	"os/exec"
 	"sync"
 
+	"github.com/newrelic/infrastructure-agent/internal/gobackfill"
 	"github.com/newrelic/infrastructure-agent/internal/integrations/v4/constants"
 	"github.com/newrelic/infrastructure-agent/pkg/helpers"
 	"github.com/newrelic/infrastructure-agent/pkg/log"
 )
 
+const unknownErrExitCode = -3
+
 var illog = log.WithComponent("integrations.Executor")
 
 // Executor handles Executable commands asynchronously.
@@ -38,7 +41,7 @@ func FromCmdSlice(cmd []string, cfg *Config) Executor {
 // The executed process can be cancelled via the provided Context.
 // When writable PID channel is provided, generated PID will be written, so process could be signaled by 3rd parties.
 // When the process ends, all the channels are closed.
-func (r *Executor) Execute(ctx context.Context, pidChan chan<- int) OutputReceive {
+func (r *Executor) Execute(ctx context.Context, pidChan, exitCodeCh chan<- int) OutputReceive {
 	out, receiver := NewOutput()
 	commandCtx, cancelCommand := context.WithCancel(ctx)
 
@@ -99,7 +102,17 @@ func (r *Executor) Execute(ctx context.Context, pidChan chan<- int) OutputReceiv
 		<-commandCtx.Done()
 		if err := cmd.Wait(); err != nil {
 			out.Errors <- err
+			if exitCodeCh != nil {
+				if exitError, ok := err.(*exec.ExitError); ok {
+					exitCodeCh <- gobackfill.ExitCode(exitError)
+				} else {
+					exitCodeCh <- unknownErrExitCode
+				}
+			}
+		} else if exitCodeCh != nil {
+			exitCodeCh <- 0
 		}
+
 		allOutputForwarded.Wait() // waiting again to avoid closing output before the data is received during cancellation
 		out.Close()
 	}()

internal/integrations/v4/executor/executor_test.go

@@ -31,7 +31,7 @@ func TestRunnable_CLI_Execute(t *testing.T) {
 		r := FromCmdSlice(testhelp.Command(fixtures.BasicCmd), execConfig(t))
 
 		// WHEN it is executed
-		to := r.Execute(context.Background(), nil)
+		to := r.Execute(context.Background(), nil, nil)
 
 		// THEN no errors are returned
 		assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))
@@ -51,7 +51,7 @@ func TestRunnable_CLI_Execute_with_spaces(t *testing.T) {
 	r := FromCmdSlice(testhelp.Command(fixtures.BasicCmdWithSpace), execConfig(t))
 
 	// WHEN it is executed
-	to := r.Execute(context.Background(), nil)
+	to := r.Execute(context.Background(), nil, nil)
 
 	// THEN no errors are returned
 	assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))
@@ -76,7 +76,7 @@ func TestRunnable_Execute_WithUser(t *testing.T) {
 	r := FromCmdSlice(testhelp.Command(fixtures.BasicCmd), cfg)
 
 	// WHEN it is executed
-	to := r.Execute(context.Background(), nil)
+	to := r.Execute(context.Background(), nil, nil)
 
 	// THEN no errors are returned
 	assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))
@@ -95,7 +95,7 @@ func TestRunnable_Execute_WithArgs(t *testing.T) {
 	cfg := execConfig(t)
 	r := FromCmdSlice(testhelp.Command(fixtures.BasicCmd, "world"), cfg)
 
-	to := r.Execute(context.Background(), nil)
+	to := r.Execute(context.Background(), nil, nil)
 	assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))
 	assert.Equal(t, "stdout line", testhelp.ChannelRead(to.Stdout))
 	assert.Equal(t, "-world", testhelp.ChannelRead(to.Stdout))
@@ -113,7 +113,7 @@ func TestRunnable_Execute_WithArgs_WithEnv(t *testing.T) {
 	cfg.Environment = map[string]string{"PREFIX": "hello"}
 	r := FromCmdSlice(testhelp.Command(fixtures.BasicCmd, "world"), cfg)
 
-	to := r.Execute(context.Background(), nil)
+	to := r.Execute(context.Background(), nil, nil)
 	assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))
 	assert.Equal(t, "stdout line", testhelp.ChannelRead(to.Stdout))
 	assert.Equal(t, "hello-world", testhelp.ChannelRead(to.Stdout))
@@ -127,7 +127,7 @@ func TestRunnable_Execute_Error(t *testing.T) {
 	r := FromCmdSlice(testhelp.Command(fixtures.ErrorCmd), execConfig(t))
 
 	// WHEN it is executed
-	to := r.Execute(context.Background(), nil)
+	to := r.Execute(context.Background(), nil, nil)
 
 	// THEN an error is returned
 	assert.Error(t, testhelp.ChannelErrClosed(to.Errors))
@@ -149,7 +149,7 @@ func TestRunnable_Execute_Blocked(t *testing.T) {
 	r := FromCmdSlice(testhelp.Command(fixtures.BlockedCmd), cfg)
 
 	// THAT is normally working
-	to := r.Execute(ctx, nil)
+	to := r.Execute(ctx, nil, nil)
 	assert.Equal(t, "starting", testhelp.ChannelRead(to.Stdout))
 	assert.Error(t, testhelp.ChannelErrClosedTimeout(to.Errors, 100*time.Millisecond))
 
@@ -176,7 +176,7 @@ func TestNoRaces(t *testing.T) {
 	for i := 0; i < 100; i++ {
 		ctx, cancel := context.WithCancel(context.Background())
 		cmd := FromCmdSlice([]string{"echo", hugeLine}, &Config{})
-		go cmd.Execute(ctx, nil)
+		go cmd.Execute(ctx, nil, nil)
 		cancel()
 	}
 }
@@ -191,7 +191,7 @@ func TestRunnable_Execute_Verbose(t *testing.T) {
 	ctx := context.WithValue(context.Background(), constants.EnableVerbose, 1)
 
 	// WHEN it is executed
-	to := r.Execute(ctx, nil)
+	to := r.Execute(ctx, nil, nil)
 
 	// THEN no errors are returned
 	assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))
@@ -213,7 +213,7 @@ func TestRunnable_Execute_VerboseFalse(t *testing.T) {
 	ctx := context.WithValue(context.Background(), constants.EnableVerbose, 0)
 
 	// WHEN it is executed
-	to := r.Execute(ctx, nil)
+	to := r.Execute(ctx, nil, nil)
 
 	// THEN no errors are returned
 	assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))
@@ -232,7 +232,7 @@ func TestRunnable_Execute_NoVerboseSet(t *testing.T) {
 	r := FromCmdSlice(testhelp.Command(fixtures.IntegrationVerboseScript), execConfig(t))
 
 	// WHEN it is executed
-	to := r.Execute(context.Background(), nil)
+	to := r.Execute(context.Background(), nil, nil)
 
 	// THEN no errors are returned
 	assert.NoError(t, testhelp.ChannelErrClosed(to.Errors))