poc(edgejobs): use an in-process cron daemon

agent.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/portainer/portainer/pkg/libstack"
+	"github.com/robfig/cron/v3"
 )
 
 type (
@@ -148,6 +149,7 @@ type (
 	// Schedule represents a script that can be scheduled on the underlying host
 	Schedule struct {
 		ID             int
+		EntryID        cron.EntryID
 		CronExpression string
 		Script         string
 		Version        int

edge/scheduler/scheduler.go

@@ -6,34 +6,37 @@ package scheduler
 import (
 	"encoding/base64"
 	"fmt"
-	"strings"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"syscall"
 
 	"github.com/portainer/agent"
 	"github.com/portainer/agent/filesystem"
 
+	"github.com/robfig/cron/v3"
 	"github.com/rs/zerolog/log"
 )
 
-const (
-	cronDirectory = "/etc/cron.d"
-	cronFile      = "portainer_agent"
-	cronJobUser   = "root"
-)
-
 // CronManager is a service that manage schedules by creating a new entry inside the host filesystem under
 // the /etc/cron.d folder.
 type CronManager struct {
 	logsManager      *LogsManager
 	cronFileExists   bool
 	managedSchedules map[int]agent.Schedule
+	cron             *cron.Cron
 }
 
 // NewCronManager returns a pointer to a new instance of CronManager.
 func NewCronManager(logsManager *LogsManager) *CronManager {
+	cron := cron.New()
+	cron.Start()
+
 	return &CronManager{
 		logsManager:      logsManager,
 		cronFileExists:   false,
 		managedSchedules: make(map[int]agent.Schedule),
+		cron:             cron,
 	}
 }
 
@@ -101,70 +104,87 @@ func (manager *CronManager) Schedule(schedules []agent.Schedule) error {
 }
 
 func (manager *CronManager) removeCronFile() error {
+	for _, s := range manager.managedSchedules {
+		manager.cron.Remove(s.EntryID)
+	}
+
 	manager.managedSchedules = map[int]agent.Schedule{}
-	if manager.cronFileExists {
-		log.Debug().Msg("no schedules available, removing cron file")
 
-		manager.cronFileExists = false
-		return filesystem.RemoveFile(fmt.Sprintf("%s%s/%s", agent.HostRoot, cronDirectory, cronFile))
-	}
 	return nil
 }
 
 func (manager *CronManager) flushEntries(schedules map[int]agent.Schedule) error {
-	cronEntries := make([]string, 0)
-
-	header := []string{
-		"## This file is managed by the Portainer agent. DO NOT EDIT MANUALLY ALL YOUR CHANGES WILL BE OVERWRITTEN.",
-		"SHELL=/bin/sh",
-		"PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin",
-		"",
-	}
-
-	cronEntries = append(cronEntries, header...)
+	manager.cron.Stop()
+	manager.cron = cron.New()
+	manager.cron.Start()
 
-	for _, schedule := range schedules {
-		cronEntry, err := createCronEntry(&schedule)
+	for key, schedule := range schedules {
+		cronSpec, cronEntry, err := createCronEntry(&schedule)
 		if err != nil {
 			log.Error().Int("schedule_id", schedule.ID).Err(err).Msg("unable to create cron entry")
 
 			return err
 		}
 
-		cronEntries = append(cronEntries, cronEntry)
+		entryID, err := manager.cron.AddFunc(cronSpec, cronEntry)
+		if err != nil {
+			return err
+		}
+
+		s := schedules[key]
+		s.EntryID = entryID
+		schedules[key] = s
 	}
 
 	log.Debug().Int("schedule_count", len(manager.managedSchedules)).Msg("writing cron file on disk")
 
-	cronEntries = append(cronEntries, "")
-	cronFileContent := strings.Join(cronEntries, "\n")
-	err := filesystem.WriteFile(fmt.Sprintf("%s%s", agent.HostRoot, cronDirectory), cronFile, []byte(cronFileContent), 0644)
-	if err != nil {
-		return err
-	}
-
 	manager.cronFileExists = true
 	manager.managedSchedules = schedules
 
 	return nil
 }
 
-func createCronEntry(schedule *agent.Schedule) (string, error) {
+func createCronEntry(schedule *agent.Schedule) (string, cron.FuncJob, error) {
 	decodedScript, err := base64.RawStdEncoding.DecodeString(schedule.Script)
 	if err != nil {
-		return "", err
+		return "", nil, err
 	}
 
 	err = filesystem.WriteFile(fmt.Sprintf("%s%s", agent.HostRoot, agent.ScheduleScriptDirectory), fmt.Sprintf("schedule_%d", schedule.ID), decodedScript, 0744)
 	if err != nil {
-		return "", err
+		return "", nil, err
 	}
 
 	cronExpression := schedule.CronExpression
 	command := fmt.Sprintf("%s/schedule_%d", agent.ScheduleScriptDirectory, schedule.ID)
 	logFile := fmt.Sprintf("%s/schedule_%d.log", agent.ScheduleScriptDirectory, schedule.ID)
 
-	return fmt.Sprintf("%s %s %s > %s 2>&1", cronExpression, cronJobUser, command, logFile), nil
+	cronFn := func() {
+		log.Info().Str("command", command).Msg("running cron job")
+
+		logFileWriter, err := os.OpenFile(filepath.Join(agent.HostRoot, logFile), os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			log.Error().Err(err).Msg("could not open the log file")
+			return
+		}
+		defer logFileWriter.Close()
+
+		cmd := exec.Command("/bin/sh", command)
+		cmd.Dir = "/"
+		cmd.SysProcAttr = &syscall.SysProcAttr{Chroot: agent.HostRoot}
+		cmd.Env = []string{
+			"SHELL=/bin/sh",
+			"PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin",
+		}
+		cmd.Stdout = logFileWriter
+		cmd.Stderr = logFileWriter
+
+		if err := cmd.Run(); err != nil {
+			log.Error().Err(err).Msg("error encountered in cron job run")
+		}
+	}
+
+	return cronExpression, cronFn, nil
 }
 
 func (manager *CronManager) ProcessScheduleLogsCollection() {
@@ -189,9 +209,5 @@ func (manager *CronManager) AddSchedule(schedule agent.Schedule) error {
 func (manager *CronManager) RemoveSchedule(schedule agent.Schedule) error {
 	delete(manager.managedSchedules, schedule.ID)
 
-	if len(manager.managedSchedules) == 0 {
-		return manager.removeCronFile()
-	}
-
 	return manager.flushEntries(manager.managedSchedules)
 }

go.mod

@@ -111,6 +111,7 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
 	github.com/segmentio/asm v1.1.3 // indirect
 	github.com/segmentio/encoding v0.3.6 // indirect

go.sum

@@ -301,6 +301,8 @@ github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
 github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=