<commit_before>package poll

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/venicegeo/pzsvc-exec/dispatcher/cfwrapper"
	"github.com/venicegeo/pzsvc-exec/dispatcher/model"
	"github.com/venicegeo/pzsvc-exec/pzsvc"
)

const defaultTaskDiskMB = 6142
const defaultTaskMemoryMB = 3072

var pzsvcGetS3FileSizeInMegabytes = pzsvc.GetS3FileSizeInMegabytes
var pzsvcRequestKnownJSON = pzsvc.RequestKnownJSON
var pzsvcSendExecResultNoData = pzsvc.SendExecResultNoData

// Loop is an encapsulation of configuration and functionality needed for a job polling loop
type Loop struct {
	PzSession     *pzsvc.Session
	PzConfig      pzsvc.Config
	SvcID         string
	ConfigPath    string
	ClientFactory cfwrapper.Factory
	vcapID        string
	taskLimit     int
	intervalTick  time.Duration

	stopChan         chan bool
	runIterationFunc func(l Loop) error
}

// NewLoop creates a Loop and does starting configuration based on the given parameters
func NewLoop(s *pzsvc.Session, configObj pzsvc.Config, svcID string, configPath string, clientFactory cfwrapper.Factory) (*Loop, error) {
	pzsvc.LogInfo(*s, "Initializing polling loop object")

	appID, err := getVCAPApplicationID()
	if err != nil {
		return nil, err
	}
	pzsvc.LogInfo(*s, "Found application name from VCAP Tree: "+appID)

	// Read the # of simultaneous Tasks that are allowed to be run by the Dispatcher
	taskLimit := 3
	if envTaskLimit := os.Getenv("TASK_LIMIT"); envTaskLimit != "" {
		taskLimit, _ = strconv.Atoi(envTaskLimit)
	}

	return &Loop{
		PzSession:        s,
		PzConfig:         configObj,
		SvcID:            svcID,
		ConfigPath:       configPath,
		ClientFactory:    clientFactory,
		vcapID:           appID,
		taskLimit:        taskLimit,
		intervalTick:     5 * time.Second,
		stopChan:         nil, // initialized when loop starts
		runIterationFunc: runIteration,
	}, nil
}

// Start begins the polling interval loop and returns a channel that feeds
// through any errors encountered in each interval
func (l *Loop) Start() <-chan error {
	errChan := make(chan error)
	l.stopChan = make(chan bool)
	go func() {
		ticker := time.Tick(l.intervalTick)
		for {
			select {
			case <-ticker:
				err := l.runIterationFunc(*l)
				if err != nil {
					errChan <- err
				}
			case <-l.stopChan:
				close(errChan)
				return
			}
		}
	}()
	return errChan
}

// Stop halts the loop's iteration
func (l Loop) Stop() {
	l.stopChan <- true
	close(l.stopChan)
}

func runIteration(l Loop) error {
	pzsvc.LogInfo(*l.PzSession, "Starting polling loop iteration")

	cfSession, err := l.ClientFactory.GetSession()
	if err != nil {
		pzsvc.LogSimpleErr(*l.PzSession, "Error generating valid CF Client", err)
		return err
	}

	numTasks, err := cfSession.CountTasksForApp(l.vcapID)
	if err != nil {
		pzsvc.LogSimpleErr(*l.PzSession, "Error checking running tasks. ", err)
		return err
	}
	if numTasks >= l.taskLimit {
		pzsvc.LogInfo(*l.PzSession, "Too many tasks already running, skipping this iteration cycle")
		return nil
	}

	taskItem, taskBytes, err := l.getPzTaskItem()
	if err != nil {
		return err
	}

	jobID := taskItem.Data.SvcData.JobID
	jobData := taskItem.Data.SvcData.Data.DataInputs.Body.Content
	if jobData == "" {
		message := fmt.Sprintf("Received job with empty data, skipping this iteration cycle; (jobID='%s')\nPiazza Task Response was: %s", jobID, string(taskBytes))
		pzsvc.LogWarn(*l.PzSession, message)
		return nil
	}
	pzsvc.LogInfo(*l.PzSession, "New Task Grabbed.  JobID: "+jobID)

	jobInput, err := l.parseJobInput(jobData)
	if err != nil {
		return err
	}

	workerCommand, err := l.buildWorkerCommand(jobInput, jobID)
	if err != nil {
		return err
	}

	diskMB, memoryMB := l.calculateDiskAndMemoryLimits(jobInput)

	taskRequest := cfwrapper.TaskRequest{
		Command:          workerCommand,
		Name:             jobID,
		DropletGUID:      l.vcapID,
		DiskInMegabyte:   diskMB,
		MemoryInMegabyte: memoryMB,
	}

	serializedInput, _ := json.Marshal(jobInput)
	pzsvc.LogAudit(*l.PzSession, l.PzSession.UserID, "Creating CF Task for Job "+jobID+" : "+workerCommand, l.PzSession.AppName, string(serializedInput), pzsvc.INFO)

	if err = cfSession.CreateTask(taskRequest); err != nil {<commit_msg>Simplify 'no jobs available' log message<commit_after>package poll

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/venicegeo/pzsvc-exec/dispatcher/cfwrapper"
	"github.com/venicegeo/pzsvc-exec/dispatcher/model"
	"github.com/venicegeo/pzsvc-exec/pzsvc"
)

const defaultTaskDiskMB = 6142
const defaultTaskMemoryMB = 3072

var pzsvcGetS3FileSizeInMegabytes = pzsvc.GetS3FileSizeInMegabytes
var pzsvcRequestKnownJSON = pzsvc.RequestKnownJSON
var pzsvcSendExecResultNoData = pzsvc.SendExecResultNoData

// Loop is an encapsulation of configuration and functionality needed for a job polling loop
type Loop struct {
	PzSession     *pzsvc.Session
	PzConfig      pzsvc.Config
	SvcID         string
	ConfigPath    string
	ClientFactory cfwrapper.Factory
	vcapID        string
	taskLimit     int
	intervalTick  time.Duration

	stopChan         chan bool
	runIterationFunc func(l Loop) error
}

// NewLoop creates a Loop and does starting configuration based on the given parameters
func NewLoop(s *pzsvc.Session, configObj pzsvc.Config, svcID string, configPath string, clientFactory cfwrapper.Factory) (*Loop, error) {
	pzsvc.LogInfo(*s, "Initializing polling loop object")

	appID, err := getVCAPApplicationID()
	if err != nil {
		return nil, err
	}
	pzsvc.LogInfo(*s, "Found application name from VCAP Tree: "+appID)

	// Read the # of simultaneous Tasks that are allowed to be run by the Dispatcher
	taskLimit := 3
	if envTaskLimit := os.Getenv("TASK_LIMIT"); envTaskLimit != "" {
		taskLimit, _ = strconv.Atoi(envTaskLimit)
	}

	return &Loop{
		PzSession:        s,
		PzConfig:         configObj,
		SvcID:            svcID,
		ConfigPath:       configPath,
		ClientFactory:    clientFactory,
		vcapID:           appID,
		taskLimit:        taskLimit,
		intervalTick:     5 * time.Second,
		stopChan:         nil, // initialized when loop starts
		runIterationFunc: runIteration,
	}, nil
}

// Start begins the polling interval loop and returns a channel that feeds
// through any errors encountered in each interval
func (l *Loop) Start() <-chan error {
	errChan := make(chan error)
	l.stopChan = make(chan bool)
	go func() {
		ticker := time.Tick(l.intervalTick)
		for {
			select {
			case <-ticker:
				err := l.runIterationFunc(*l)
				if err != nil {
					errChan <- err
				}
			case <-l.stopChan:
				close(errChan)
				return
			}
		}
	}()
	return errChan
}

// Stop halts the loop's iteration
func (l Loop) Stop() {
	l.stopChan <- true
	close(l.stopChan)
}

func runIteration(l Loop) error {
	pzsvc.LogInfo(*l.PzSession, "Starting polling loop iteration")

	cfSession, err := l.ClientFactory.GetSession()
	if err != nil {
		pzsvc.LogSimpleErr(*l.PzSession, "Error generating valid CF Client", err)
		return err
	}

	numTasks, err := cfSession.CountTasksForApp(l.vcapID)
	if err != nil {
		pzsvc.LogSimpleErr(*l.PzSession, "Error checking running tasks. ", err)
		return err
	}
	if numTasks >= l.taskLimit {
		pzsvc.LogInfo(*l.PzSession, "Too many tasks already running, skipping this iteration cycle")
		return nil
	}

	taskItem, _, err := l.getPzTaskItem()
	if err != nil {
		return err
	}

	jobID := taskItem.Data.SvcData.JobID
	jobData := taskItem.Data.SvcData.Data.DataInputs.Body.Content
	if jobData == "" {
		message := fmt.Sprintf("No jobs available in task queue (jobID=''); skipping this iteration cycle")
		pzsvc.LogWarn(*l.PzSession, message)
		return nil
	}
	pzsvc.LogInfo(*l.PzSession, "New Task Grabbed.  JobID: "+jobID)

	jobInput, err := l.parseJobInput(jobData)
	if err != nil {
		return err
	}

	workerCommand, err := l.buildWorkerCommand(jobInput, jobID)
	if err != nil {
		return err
	}

	diskMB, memoryMB := l.calculateDiskAndMemoryLimits(jobInput)

	taskRequest := cfwrapper.TaskRequest{
		Command:          workerCommand,
		Name:             jobID,
		DropletGUID:      l.vcapID,
		DiskInMegabyte:   diskMB,
		MemoryInMegabyte: memoryMB,
	}

	serializedInput, _ := json.Marshal(jobInput)
	pzsvc.LogAudit(*l.PzSession, l.PzSession.UserID, "Creating CF Task for Job "+jobID+" : "+workerCommand, l.PzSession.AppName, string(serializedInput), pzsvc.INFO)

	if err = cfSession.CreateTask(taskRequest); err != nil {
{'fraction_non_alphanumeric': 0.10592160133444536, 'fraction_numerical': 0.0028595257953056116, 'mean_word_length': 3.6841517857142856, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 5, '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': '3589674', 'n_tokens_mistral': 3164, 'n_tokens_neox': 2925, 'n_words': 751}