<commit_before>package commands

import (
	"context"
	"fmt"
	"io"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/joyent/containerpilot/events"
	"github.com/joyent/containerpilot/utils"
)

const errNoChild = "wait: no child processes"

// Command wraps an os/exec.Cmd with a timeout, logging, and arg parsing.
type Command struct {
	Name       string // this gets used only in logs, defaults to Exec
	Cmd        *exec.Cmd
	Exec       string
	Args       []string
	Timeout    time.Duration
	logWriters []io.WriteCloser
	lock       *sync.Mutex
}

// NewCommand parses JSON config into a Command
func NewCommand(rawArgs interface{}, timeout time.Duration) (*Command, error) {
	exec, args, err := ParseArgs(rawArgs)
	if err != nil {
		return nil, err
	}
	cmd := &Command{
		Name:    exec, // override this in caller
		Exec:    exec,
		Args:    args,
		Timeout: timeout,
		lock:    &sync.Mutex{},
	} // cmd, logWriters all created at RunAndWait or RunWithTimeout
	return cmd, nil
}

// Run ...
func (c *Command) Run(pctx context.Context, bus *events.EventBus, fields log.Fields) {
	if c == nil {
		// TODO: will this ever get called like this?
		log.Debugf("nothing to run for %s", c.Name)
		return
	}
	// we should never have more than one instance running for any
	// realistic configuration but this ensures that's the case
	c.lock.Lock()
	log.Debugf("%s.Run start", c.Name)
	c.setUpCmd(fields)
	var (
		ctx    context.Context
		cancel context.CancelFunc
	)
	if c.Timeout > 0 {
		ctx, cancel = context.WithTimeout(pctx, c.Timeout)
	} else {
		ctx, cancel = context.WithCancel(pctx)
	}
	go func() {
		defer cancel()
		defer c.closeLogs()
		defer log.Debugf("%s.Run end", c.Name)
		if err := c.Cmd.Start(); err != nil {
			log.Errorf("unable to start %s: %v", c.Name, err)
			bus.Publish(events.Event{events.ExitFailed, c.Name})
			bus.Publish(events.Event{events.Error, err.Error()})
			return
		}
		// blocks this goroutine here; if the context gets cancelled
		// we'll return from wait() and do all the cleanup
		if _, err := c.wait(); err != nil {
			log.Errorf("%s exited with error: %v", c.Name, err)
			bus.Publish(events.Event{events.ExitFailed, c.Name})
			bus.Publish(events.Event{events.Error, err.Error()})
		} else {
			bus.Publish(events.Event{events.ExitSuccess, c.Name})
		}
	}()

	go func() {
		select {
		case <-ctx.Done():
			// unlock only here because we'll receive this cancel from
			// both a typical exit and a timeout
			defer c.lock.Unlock()
			// if the context was canceled we don't want to kill the
			// process because it's already gone
			if ctx.Err().Error() == "context deadline exceeded" {
				log.Warnf("%s timeout after %s: '%s'", c.Name, c.Timeout, c.Args)
				c.Kill()
			}
		}
	}()
}

// RunAndWaitForOutput runs the command and blocks until completed, then
// returns a string of the stdout
// TODO: remove this once the control plane is available for Sensors (the
// only caller) to send metrics to
func (c *Command) RunAndWaitForOutput(pctx context.Context, bus *events.EventBus) string {
	if c == nil {
		// TODO: will this ever get called like this?
		log.Debugf("nothing to run for %s", c.Name)
		return ""
	}
	// we should never have more than one instance running for any
	// realistic configuration but this ensures that's the case
	c.lock.Lock()
	log.Debugf("%s.Run start", c.Name)
	c.setUpCmd(nil)
	var (
		ctx    context.Context
		cancel context.CancelFunc
	)
	if c.Timeout > 0 {
		ctx, cancel = context.WithTimeout(pctx, c.Timeout)
	} else {
		ctx, cancel = context.WithCancel(pctx)
	}
	defer cancel()
	defer c.closeLogs()

	go func() {
		select {
		case <-ctx.Done():
			// unlock only here because we'll receive this cancel from
			// both a typical exit and a timeout
			defer c.lock.Unlock()
			// if the context was canceled we don't want to kill the
			// process because it's already gone
			if ctx.Err().Error() == "context deadline exceeded" {
				log.Warnf("%s timeout after %s: '%s'", c.Name, c.Timeout, c.Args)
				c.Kill()
			}
		}
	}()
	// we'll pass stderr to the container's stderr, but stdout must
	// be "clean" and not have anything other than what we intend
	// to write to our collector.
	c.Cmd.Stderr = os.Stderr
	log.Debugf("%s.Cmd.Output", c.Name)
	defer log.Debugf("%s.RunAndWaitForOutput end", c.Name)<commit_msg>minimize race for end of exec and closing logger's io.WriteCloser<commit_after>package commands

import (
	"context"
	"fmt"
	"io"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/joyent/containerpilot/events"
	"github.com/joyent/containerpilot/utils"
)

const errNoChild = "wait: no child processes"

// Command wraps an os/exec.Cmd with a timeout, logging, and arg parsing.
type Command struct {
	Name       string // this gets used only in logs, defaults to Exec
	Cmd        *exec.Cmd
	Exec       string
	Args       []string
	Timeout    time.Duration
	logWriters []io.WriteCloser
	lock       *sync.Mutex
}

// NewCommand parses JSON config into a Command
func NewCommand(rawArgs interface{}, timeout time.Duration) (*Command, error) {
	exec, args, err := ParseArgs(rawArgs)
	if err != nil {
		return nil, err
	}
	cmd := &Command{
		Name:    exec, // override this in caller
		Exec:    exec,
		Args:    args,
		Timeout: timeout,
		lock:    &sync.Mutex{},
	} // cmd, logWriters all created at RunAndWait or RunWithTimeout
	return cmd, nil
}

// Run ...
func (c *Command) Run(pctx context.Context, bus *events.EventBus, fields log.Fields) {
	if c == nil {
		// TODO: will this ever get called like this?
		log.Debugf("nothing to run for %s", c.Name)
		return
	}
	// we should never have more than one instance running for any
	// realistic configuration but this ensures that's the case
	c.lock.Lock()
	log.Debugf("%s.Run start", c.Name)
	c.setUpCmd(fields)
	var (
		ctx    context.Context
		cancel context.CancelFunc
	)
	if c.Timeout > 0 {
		ctx, cancel = context.WithTimeout(pctx, c.Timeout)
	} else {
		ctx, cancel = context.WithCancel(pctx)
	}
	go func() {
		defer cancel()
		defer log.Debugf("%s.Run end", c.Name)
		if err := c.Cmd.Start(); err != nil {
			log.Errorf("unable to start %s: %v", c.Name, err)
			bus.Publish(events.Event{events.ExitFailed, c.Name})
			bus.Publish(events.Event{events.Error, err.Error()})
			return
		}
		// blocks this goroutine here; if the context gets cancelled
		// we'll return from wait() and do all the cleanup
		if _, err := c.wait(); err != nil {
			log.Errorf("%s exited with error: %v", c.Name, err)
			bus.Publish(events.Event{events.ExitFailed, c.Name})
			bus.Publish(events.Event{events.Error, err.Error()})
		} else {
			bus.Publish(events.Event{events.ExitSuccess, c.Name})
		}
	}()

	go func() {
		select {
		case <-ctx.Done():
			// unlock and close logs only here because we'll receive this
			// cancel from both a typical exit and a timeout
			defer c.lock.Unlock()
			defer c.closeLogs()
			// if the context was canceled we don't want to kill the
			// process because it's already gone
			if ctx.Err().Error() == "context deadline exceeded" {
				log.Warnf("%s timeout after %s: '%s'", c.Name, c.Timeout, c.Args)
				c.Kill()
			}
		}
	}()
}

// RunAndWaitForOutput runs the command and blocks until completed, then
// returns a string of the stdout
// TODO: remove this once the control plane is available for Sensors (the
// only caller) to send metrics to
func (c *Command) RunAndWaitForOutput(pctx context.Context, bus *events.EventBus) string {
	if c == nil {
		// TODO: will this ever get called like this?
		log.Debugf("nothing to run for %s", c.Name)
		return ""
	}
	// we should never have more than one instance running for any
	// realistic configuration but this ensures that's the case
	c.lock.Lock()
	log.Debugf("%s.Run start", c.Name)
	c.setUpCmd(nil)
	var (
		ctx    context.Context
		cancel context.CancelFunc
	)
	if c.Timeout > 0 {
		ctx, cancel = context.WithTimeout(pctx, c.Timeout)
	} else {
		ctx, cancel = context.WithCancel(pctx)
	}
	defer cancel()

	go func() {
		select {
		case <-ctx.Done():
			// unlock and close logs only here because we'll receive this
			// cancel from both a typical exit and a timeout
			defer c.lock.Unlock()
			defer c.closeLogs()
			// if the context was canceled we don't want to kill the
			// process because it's already gone
			if ctx.Err().Error() == "context deadline exceeded" {
				log.Warnf("%s timeout after %s: '%s'", c.Name, c.Timeout, c.Args)
				c.Kill()
			}
		}
	}()
	// we'll pass stderr to the container's stderr, but stdout must
	// be "clean" and not have anything other than what we intend
	// to write to our collector.
	c.Cmd.Stderr = os.Stderr
	log.Debugf("%s.Cmd.Output", c.Name)
	defer log.Debugf("%s.RunAndWaitForOutput end", c.Name)
{'fraction_non_alphanumeric': 0.12821391840932464, 'fraction_numerical': 0.0004570906182150611, 'mean_word_length': 3.1715919923736893, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5132414', 'n_tokens_mistral': 3165, 'n_tokens_neox': 2916, 'n_words': 982}