<commit_before>package job

import (
	"bytes"
	"errors"
	"fmt"
	"net/http"
	"os/exec"
	"strings"
	"time"

	"github.com/mattn/go-shellwords"
	log "github.com/sirupsen/logrus"
)

type JobRunner struct {
	job  *Job
	meta Metadata

	numberOfAttempts uint
	currentRetries   uint
	currentStat      *JobStat
}

var (
	ErrJobDisabled    = errors.New("Job cannot run, as it is disabled")
	ErrCmdIsEmpty     = errors.New("Job Command is empty.")
	ErrJobTypeInvalid = errors.New("Job Type is not valid.")
)

// Run calls the appropriate run function, collects metadata around the success
// or failure of the Job's execution, and schedules the next run.
func (j *JobRunner) Run(cache JobCache) (*JobStat, Metadata, error) {
	j.job.lock.RLock()
	defer j.job.lock.RUnlock()

	j.meta.LastAttemptedRun = j.job.clk.Time().Now()

	if j.job.Disabled {
		log.Infof("Job %s tried to run, but exited early because its disabled.", j.job.Name)
		return nil, j.meta, ErrJobDisabled
	}

	log.Infof("Job %s:%s started.", j.job.Name, j.job.Id)

	j.runSetup()

	for {
		var err error
		if j.job.JobType == LocalJob {
			err = j.LocalRun()
		} else if j.job.JobType == RemoteJob {
			err = j.RemoteRun()
		} else {
			err = ErrJobTypeInvalid
		}

		if err != nil {
			// Log Error in Metadata
			// TODO - Error Reporting, email error
			log.Errorln("Error running job:", j.currentStat.JobId)
			log.Errorln(err)

			j.meta.ErrorCount++
			j.meta.LastError = j.job.clk.Time().Now()

			// Handle retrying
			if j.shouldRetry() {
				j.currentRetries--
				continue
			}

			j.collectStats(false)
			j.meta.NumberOfFinishedRuns++

			// TODO: Wrap error into something better.
			return j.currentStat, j.meta, err
		} else {
			break
		}
	}

	log.Infof("Job %s:%s finished.", j.job.Name, j.job.Id)
	j.meta.SuccessCount++
	j.meta.NumberOfFinishedRuns++
	j.meta.LastSuccess = j.job.clk.Time().Now()

	j.collectStats(true)

	// Run Dependent Jobs
	if len(j.job.DependentJobs) != 0 {
		for _, id := range j.job.DependentJobs {
			newJob, err := cache.Get(id)
			if err != nil {
				log.Errorf("Error retrieving dependent job with id of %s", id)
			} else {
				newJob.Run(cache)
			}
		}
	}

	return j.currentStat, j.meta, nil
}

// LocalRun executes the Job's local shell command
func (j *JobRunner) LocalRun() error {
	return j.runCmd()
}

// RemoteRun sends a http request, and checks if the response is valid in time,
func (j *JobRunner) RemoteRun() error {
	// Calculate a response timeout
	timeout := j.responseTimeout()

	httpClient := http.Client{
		Timeout: timeout,
	}

	// Normalize the method passed by the user
	method := strings.ToUpper(j.job.RemoteProperties.Method)
	bodyBuffer := bytes.NewBufferString(j.job.RemoteProperties.Body)
	req, err := http.NewRequest(method, j.job.RemoteProperties.Url, bodyBuffer)
	if err != nil {
		return err
	}

	// Set default or user's passed headers
	j.setHeaders(req)

	// Do the request
	res, err := httpClient.Do(req)
	if err != nil {
		return err
	}

	// Check if we got any of the status codes the user asked for
	if j.checkExpected(res.StatusCode) {
		return nil
	} else {
		return errors.New(res.Status)
	}
}

func initShParser() *shellwords.Parser {
	shParser := shellwords.NewParser()
	shParser.ParseEnv = true
	shParser.ParseBacktick = true
	return shParser
}

func (j *JobRunner) runCmd() error {
	j.numberOfAttempts++

	// Execute command
	shParser := initShParser()<commit_msg>Changed to using the http.DefaultClient for remote jobs.<commit_after>package job

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"net/http"
	"os/exec"
	"strings"
	"time"

	"github.com/mattn/go-shellwords"
	log "github.com/sirupsen/logrus"
)

type JobRunner struct {
	job  *Job
	meta Metadata

	numberOfAttempts uint
	currentRetries   uint
	currentStat      *JobStat
}

var (
	ErrJobDisabled    = errors.New("Job cannot run, as it is disabled")
	ErrCmdIsEmpty     = errors.New("Job Command is empty.")
	ErrJobTypeInvalid = errors.New("Job Type is not valid.")
)

// Run calls the appropriate run function, collects metadata around the success
// or failure of the Job's execution, and schedules the next run.
func (j *JobRunner) Run(cache JobCache) (*JobStat, Metadata, error) {
	j.job.lock.RLock()
	defer j.job.lock.RUnlock()

	j.meta.LastAttemptedRun = j.job.clk.Time().Now()

	if j.job.Disabled {
		log.Infof("Job %s tried to run, but exited early because its disabled.", j.job.Name)
		return nil, j.meta, ErrJobDisabled
	}

	log.Infof("Job %s:%s started.", j.job.Name, j.job.Id)

	j.runSetup()

	for {
		var err error
		if j.job.JobType == LocalJob {
			err = j.LocalRun()
		} else if j.job.JobType == RemoteJob {
			err = j.RemoteRun()
		} else {
			err = ErrJobTypeInvalid
		}

		if err != nil {
			// Log Error in Metadata
			// TODO - Error Reporting, email error
			log.Errorln("Error running job:", j.currentStat.JobId)
			log.Errorln(err)

			j.meta.ErrorCount++
			j.meta.LastError = j.job.clk.Time().Now()

			// Handle retrying
			if j.shouldRetry() {
				j.currentRetries--
				continue
			}

			j.collectStats(false)
			j.meta.NumberOfFinishedRuns++

			// TODO: Wrap error into something better.
			return j.currentStat, j.meta, err
		} else {
			break
		}
	}

	log.Infof("Job %s:%s finished.", j.job.Name, j.job.Id)
	j.meta.SuccessCount++
	j.meta.NumberOfFinishedRuns++
	j.meta.LastSuccess = j.job.clk.Time().Now()

	j.collectStats(true)

	// Run Dependent Jobs
	if len(j.job.DependentJobs) != 0 {
		for _, id := range j.job.DependentJobs {
			newJob, err := cache.Get(id)
			if err != nil {
				log.Errorf("Error retrieving dependent job with id of %s", id)
			} else {
				newJob.Run(cache)
			}
		}
	}

	return j.currentStat, j.meta, nil
}

// LocalRun executes the Job's local shell command
func (j *JobRunner) LocalRun() error {
	return j.runCmd()
}

// RemoteRun sends a http request, and checks if the response is valid in time,
func (j *JobRunner) RemoteRun() error {
	// Calculate a response timeout
	timeout := j.responseTimeout()

	ctx, cncl := context.Background(), func() {}
	if timeout > 0 {
		ctx, cncl = context.WithTimeout(ctx, timeout)
		defer cncl()
	}

	// Normalize the method passed by the user
	method := strings.ToUpper(j.job.RemoteProperties.Method)
	bodyBuffer := bytes.NewBufferString(j.job.RemoteProperties.Body)
	req, err := http.NewRequest(method, j.job.RemoteProperties.Url, bodyBuffer)
	if err != nil {
		return err
	}

	// Set default or user's passed headers
	j.setHeaders(req)

	// Do the request
	res, err := http.DefaultClient.Do(req.WithContext(ctx))
	if err != nil {
		return err
	}

	// Check if we got any of the status codes the user asked for
	if j.checkExpected(res.StatusCode) {
		return nil
	} else {
		return errors.New(res.Status)
	}
}

func initShParser() *shellwords.Parser {
	shParser := shellwords.NewParser()
	shParser.ParseEnv = true
	shParser.ParseBacktick = true
	return shParser
}

func (j *JobRunner) runCmd() error {
	j.numberOfAttempts++

	// Execute command
	shParser := initShParser()
{'fraction_non_alphanumeric': 0.12898134863701577, 'fraction_numerical': 0.000430416068866571, 'mean_word_length': 3.31640866873065, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, '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': '6364947', 'n_tokens_mistral': 2652, 'n_tokens_neox': 2437, 'n_words': 672}