gogocron/gogocron.go

/*

	Copyright by Vladimir Smagin (21h) 2018
	http://blindage.org email: 21h@blindage.org

	Project page: http://gogocron.blindage.org

*/

package main

import (
	"context"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"
)

const (
	cronDefaultConfigLocation string = "/etc/gogocron/gogocron.yml"
	cronDefaultTasksLocation  string = "/etc/gogocron/configs"
	cronDefaultPrecision      string = "100ms"
)

type cronConfig struct {
	Precision       time.Duration
	PrecisionString string `yaml:"ticktack_precision,omitempty"` // how frequently check seconds change
	TasksLocation   string `yaml:"tasks_location,omitempty"`     // where tasks configs stored
}

type cronTask struct {
	Name         string   `yaml:"name"`                   // name of task
	AsUser       string   `yaml:"user,omitempty"`         // run as user
	RunSecond    string   `yaml:"runsecond,omitempty"`    // second
	RunMinute    string   `yaml:"runminute,omitempty"`    // minute
	RunHour      string   `yaml:"runhour,omitempty"`      // hour
	RunDom       string   `yaml:"rundom,omitempty"`       // day of month
	RunMonth     string   `yaml:"runmonth,omitempty"`     // month
	RunDow       string   `yaml:"rundow,omitempty"`       // day of week
	TimeOut      string   `yaml:"timeout,omitempty"`      // exec with timeout, seconds
	Env          []string `yaml:"env,omitempty"`          // array of env variables
	Commands     []string `yaml:"commands"`               // array of commands to exec
	LogType      string   `yaml:"logtype,omitempty"`      // type of logging: info, full, disabled
	GracefulStop string   `yaml:"gracefulstop,omitempty"` // seconds between task term and kill
}

type cronTasks []cronTask

// search for tasks that must be executed now
func filterTasksToExecute(tasks cronTasks) cronTasks {
	now := time.Now()
	var tasksToExecute cronTasks
	for _, task := range tasks {
		taskIsForExecute := true

		if task.RunSecond != "" && taskIsForExecute {
			taskIsForExecute = isReadyToExec(task.RunSecond, now.Second())
		}

		if task.RunMinute != "" && taskIsForExecute {
			taskIsForExecute = isReadyToExec(task.RunMinute, now.Minute())
		}

		if task.RunHour != "" && taskIsForExecute {
			taskIsForExecute = isReadyToExec(task.RunHour, now.Hour())
		}

		if task.RunDom != "" && taskIsForExecute {
			taskIsForExecute = isReadyToExec(task.RunDom, now.Day())
		}

		if task.RunMonth != "" && taskIsForExecute {
			taskIsForExecute = isReadyToExec(task.RunMonth, int(now.Month()))
		}

		if task.RunDow != "" && taskIsForExecute {
			taskIsForExecute = isReadyToExec(task.RunDow, int(now.Weekday()))
		}

		if taskIsForExecute {
			tasksToExecute = append(tasksToExecute, task)
		}
	}
	return tasksToExecute
}

// executeTasks runs batch of planned tasks on go routines with context
func executeTasks(tasks cronTasks) {
	//now := time.Now()
	//log.Printf("Execute started: %v", now)

	// run every task in own context
	ctx, ctxCancel := context.WithCancel(context.Background())
	defer ctxCancel()

	// iterate tasks and run them in go routine
	for _, task := range tasks {

		switch task.LogType {
		case "full":
			task.Env = append(task.Env, "LOGTYPE=full")
		case "disabled":
			task.Env = append(task.Env, "LOGTYPE=disabled")
		default:
			task.LogType = "info"
			task.Env = append(task.Env, "LOGTYPE=info")
		}

		if task.LogType != "disabled" {
			log.Printf("TASK: %#v TIMEOUT: %#v COMMANDS: %#v", task.Name, task.TimeOut, task.Commands)
		}

		if task.TimeOut != "" {
			timeout, err := time.ParseDuration(task.TimeOut)
			if err != nil {
				log.Fatalln("Error, task definition malformed!", task.Name)
			}
			ctx, ctxCancelTimeout := context.WithTimeout(ctx, timeout)
			defer ctxCancelTimeout()
			go executeTask(ctx, task)
		} else {
			go executeTask(ctx, task)
		}

	}

	select {}

}

func executeTask(ctx context.Context, task cronTask) {
	// flag to stop timed out tasks
	timeoutFlag := false

	// send gracefulstop with context
	type contextKey string
	gracefulStop := contextKey("gracefulstopvalue")
	ctxRunCmd := context.WithValue(ctx, gracefulStop, task.GracefulStop)

	// iterate commands list
	for _, cmd := range task.Commands {
		if timeoutFlag {
			break
		}

		select {
		case <-ctxRunCmd.Done():
			log.Printf("Timeout exceeded for task %v", task.Name)
			timeoutFlag = true
		default:
			// su required to run as other users, requires pty to run
			runCmd(ctxRunCmd, task.Env, "su", "-c", cmd, task.AsUser)
		}
	}

}

func main() {
	log.Println("gogocron 1.1.1 by Vladimir Smagin (http://gogocron.blindage.org)")

	// Load config at startup, once
	gogoConfig := loadConfig()

	// Load tasks
	tasks := loadTasks(gogoConfig)

	// Goroutine reloads tasks if SIGHUP received
	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, syscall.SIGHUP)
	go func() {
		for signame := range sigchan {
			log.Println("Received signal HUP, reloading configuration", signame)
			tasks = loadTasks(gogoConfig)
		}
	}()

	tickchan := make(chan struct{})
	go ticktack(tickchan, gogoConfig)

	for {
		select {
		case <-tickchan:
			tasksToExecute := filterTasksToExecute(tasks)
			if len(tasksToExecute) > 0 {
				go executeTasks(tasksToExecute)
			}
		}
	}

}