<commit_before>package condor

import (
	"fmt"
	"funnel/config"
	"funnel/logger"
	pbf "funnel/proto/funnel"
	"funnel/proto/tes"
	sched "funnel/scheduler"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strings"
	"text/template"
)

var log = logger.New("condor")

// prefix is a string prefixed to condor worker IDs, so that condor
// workers can be identified by ShouldStartWorker() below.
// TODO move to worker metadata to be consistent with GCE
const prefix = "condor-worker-"

// NewScheduler returns a new HTCondor Scheduler instance.
func NewScheduler(conf config.Config) sched.Scheduler {
	return &scheduler{conf}
}

type scheduler struct {
	conf config.Config
}

// Schedule schedules a job on the HTCondor queue and returns a corresponding Offer.
func (s *scheduler) Schedule(j *tes.Job) *sched.Offer {
	log.Debug("Running condor scheduler")

	disk := s.conf.Worker.Resources.Disk
	if disk == 0.0 {
		for _, v := range j.Task.GetResources().GetVolumes() {
			disk += v.GetSizeGb()
		}
	}

	cpus := s.conf.Worker.Resources.Cpus
	if cpus == 0 {
		cpus = j.Task.GetResources().GetMinimumCpuCores()
	}

	ram := s.conf.Worker.Resources.Ram
	if ram == 0.0 {
		ram = j.Task.GetResources().GetMinimumRamGb()
	}

	// TODO could we call condor_submit --dry-run to test if a job would succeed?
	w := &pbf.Worker{
		Id: prefix + j.JobID,
		Resources: &pbf.Resources{
			Cpus: cpus,
			Ram:  ram,
			Disk: disk,
		},
	}
	return sched.NewOffer(w, j, sched.Scores{})
}

func (s *scheduler) ShouldStartWorker(w *pbf.Worker) bool {
	return strings.HasPrefix(w.Id, prefix) &&
		w.State == pbf.WorkerState_Uninitialized
}

// StartWorker submits a job via "condor_submit" to start a new worker.
func (s *scheduler) StartWorker(w *pbf.Worker) error {
	log.Debug("Starting condor worker")
	var err error

	// TODO document that these working dirs need manual cleanup
	workdir := path.Join(s.conf.WorkDir, w.Id)
	workdir, _ = filepath.Abs(workdir)
	err = os.MkdirAll(workdir, 0755)
	if err != nil {
		return err
	}

	c := s.conf.Worker
	c.ID = w.Id
	// 5 second timeout
	c.Timeout = 5000000000
	c.Resources.Cpus = w.Resources.Cpus
	c.Resources.Ram = w.Resources.Ram
	c.Resources.Disk = w.Resources.Disk

	confPath := path.Join(workdir, "worker.conf.yml")
	c.ToYamlFile(confPath)

	workerPath := sched.DetectWorkerPath()

	submitPath := path.Join(workdir, "condor.submit")
	f, err := os.Create(submitPath)
	if err != nil {
		return err
	}

	submitTpl, err := template.New("condor.submit").Parse(`
		universe       = vanilla
		executable     = {{.Executable}}
		arguments      = worker --config worker.conf.yml
		environment    = "PATH=/usr/bin"
		log            = {{.WorkDir}}/condor-event-log
		error          = {{.WorkDir}}/tes-worker-stderr
		output         = {{.WorkDir}}/tes-worker-stdout
		input          = {{.Config}}
		{{.Resources}}
		should_transfer_files   = YES
		when_to_transfer_output = ON_EXIT
		queue
	`)
	if err != nil {
		return err
	}

	err = submitTpl.Execute(f, map[string]string{<commit_msg>fixed condor submit formatting<commit_after>package condor

import (
	"fmt"
	"funnel/config"
	"funnel/logger"
	pbf "funnel/proto/funnel"
	"funnel/proto/tes"
	sched "funnel/scheduler"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strings"
	"text/template"
)

var log = logger.New("condor")

// prefix is a string prefixed to condor worker IDs, so that condor
// workers can be identified by ShouldStartWorker() below.
// TODO move to worker metadata to be consistent with GCE
const prefix = "condor-worker-"

// NewScheduler returns a new HTCondor Scheduler instance.
func NewScheduler(conf config.Config) sched.Scheduler {
	return &scheduler{conf}
}

type scheduler struct {
	conf config.Config
}

// Schedule schedules a job on the HTCondor queue and returns a corresponding Offer.
func (s *scheduler) Schedule(j *tes.Job) *sched.Offer {
	log.Debug("Running condor scheduler")

	disk := s.conf.Worker.Resources.Disk
	if disk == 0.0 {
		for _, v := range j.Task.GetResources().GetVolumes() {
			disk += v.GetSizeGb()
		}
	}

	cpus := s.conf.Worker.Resources.Cpus
	if cpus == 0 {
		cpus = j.Task.GetResources().GetMinimumCpuCores()
	}

	ram := s.conf.Worker.Resources.Ram
	if ram == 0.0 {
		ram = j.Task.GetResources().GetMinimumRamGb()
	}

	// TODO could we call condor_submit --dry-run to test if a job would succeed?
	w := &pbf.Worker{
		Id: prefix + j.JobID,
		Resources: &pbf.Resources{
			Cpus: cpus,
			Ram:  ram,
			Disk: disk,
		},
	}
	return sched.NewOffer(w, j, sched.Scores{})
}

func (s *scheduler) ShouldStartWorker(w *pbf.Worker) bool {
	return strings.HasPrefix(w.Id, prefix) &&
		w.State == pbf.WorkerState_Uninitialized
}

// StartWorker submits a job via "condor_submit" to start a new worker.
func (s *scheduler) StartWorker(w *pbf.Worker) error {
	log.Debug("Starting condor worker")
	var err error

	// TODO document that these working dirs need manual cleanup
	workdir := path.Join(s.conf.WorkDir, w.Id)
	workdir, _ = filepath.Abs(workdir)
	err = os.MkdirAll(workdir, 0755)
	if err != nil {
		return err
	}

	c := s.conf.Worker
	c.ID = w.Id
	// 5 second timeout
	c.Timeout = 5000000000
	c.Resources.Cpus = w.Resources.Cpus
	c.Resources.Ram = w.Resources.Ram
	c.Resources.Disk = w.Resources.Disk

	confPath := path.Join(workdir, "worker.conf.yml")
	c.ToYamlFile(confPath)

	workerPath := sched.DetectWorkerPath()

	submitPath := path.Join(workdir, "condor.submit")
	f, err := os.Create(submitPath)
	if err != nil {
		return err
	}

	submitTpl, err := template.New("condor.submit").Parse(`
universe       = vanilla
executable     = {{.Executable}}
arguments      = worker --config worker.conf.yml
environment    = "PATH=/usr/bin"
log            = {{.WorkDir}}/condor-event-log
error          = {{.WorkDir}}/tes-worker-stderr
output         = {{.WorkDir}}/tes-worker-stdout
input          = {{.Config}}
{{.Resources}}
should_transfer_files   = YES
when_to_transfer_output = ON_EXIT
queue
`)
	if err != nil {
		return err
	}

	err = submitTpl.Execute(f, map[string]string{
{'fraction_non_alphanumeric': 0.13502673796791445, 'fraction_numerical': 0.0066844919786096255, 'mean_word_length': 3.4764397905759163, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '393334', 'n_tokens_mistral': 2275, 'n_tokens_neox': 2137, 'n_words': 542}