1
0
Fork 0
mirror of https://github.com/Luzifer/elastic_cron.git synced 2024-12-23 02:41:20 +00:00
elastic_cron/main.go

187 lines
4.1 KiB
Go
Raw Normal View History

2016-02-26 17:44:10 +00:00
package main
import (
"fmt"
"io/ioutil"
"log"
"net/http"
2016-02-26 17:44:10 +00:00
"os"
"os/exec"
"strings"
"time"
"github.com/Luzifer/rconfig"
"github.com/robfig/cron"
"golang.org/x/net/context"
"gopkg.in/yaml.v2"
2016-02-26 17:44:10 +00:00
)
var (
cfg = struct {
ConfigFile string `flag:"config" default:"config.yaml" description:"Cron definition file"`
Hostname string `flag:"hostname" description:"Overwrite system hostname"`
PingTimout time.Duration `flag:"ping-timeout" default:"1s" description:"Timeout for success / failure pings"`
2016-02-26 17:44:10 +00:00
}{}
version = "dev"
logstream = make(chan *message, 1000)
)
type cronConfig struct {
RSyslogTarget string `yaml:"rsyslog_target"`
LogTemplate string `yaml:"log_template"`
2016-02-26 17:44:10 +00:00
Jobs []cronJob `yaml:"jobs"`
}
type cronJob struct {
Name string `yaml:"name"`
Schedule string `yaml:"schedule"`
Command string `yaml:"cmd"`
Arguments []string `yaml:"args"`
PingSuccess string `yaml:"ping_success"`
PingFailure string `yaml:"ping_failure"`
2016-02-26 17:44:10 +00:00
}
func init() {
rconfig.Parse(&cfg)
if cfg.Hostname == "" {
hostname, _ := os.Hostname()
cfg.Hostname = hostname
}
}
func main() {
body, err := ioutil.ReadFile(cfg.ConfigFile)
if err != nil {
log.Fatalf("Unable to read config file: %s", err)
2016-02-26 17:44:10 +00:00
}
cc := cronConfig{
LogTemplate: `<{{ syslogpri .Severity }}>{{ .Date.Format "Jan 02 15:04:05" }} {{ .Hostname }} {{ .JobName }}: {{ .Message }}`,
}
2016-02-26 17:44:10 +00:00
if err := yaml.Unmarshal(body, &cc); err != nil {
log.Fatalf("Unable to parse config file: %s", err)
}
c := cron.New()
for i := range cc.Jobs {
job := cc.Jobs[i]
if err := c.AddFunc(job.Schedule, getJobExecutor(job)); err != nil {
log.Fatalf("Unable to add job '%s': %s", job.Name, err)
}
}
c.Start()
logadapter, err := NewSyslogAdapter(cc.RSyslogTarget, cc.LogTemplate)
2016-02-26 17:44:10 +00:00
if err != nil {
log.Fatalf("Unable to open syslog connection: %s", err)
}
logadapter.Stream(logstream)
}
func getJobExecutor(job cronJob) func() {
return func() {
stdout := &messageChanWriter{
jobName: job.Name,
msgChan: logstream,
severity: 6, // Informational
}
stderr := &messageChanWriter{
jobName: job.Name,
msgChan: logstream,
severity: 3, // Error
}
fmt.Fprintln(stdout, "[SYS] Starting job")
cmd := exec.Command(job.Command, job.Arguments...)
cmd.Stdout = stdout
cmd.Stderr = stderr
err := cmd.Run()
switch err.(type) {
case nil:
fmt.Fprintln(stdout, "[SYS] Command execution successful")
go func(url string) {
if err := doPing(url); err != nil {
fmt.Fprintf(stderr, "[SYS] Ping to URL %q caused an error: %s", url, err)
}
}(job.PingSuccess)
2016-02-26 17:44:10 +00:00
case *exec.ExitError:
fmt.Fprintln(stderr, "[SYS] Command exited with unexpected exit code != 0")
go func(url string) {
if err := doPing(url); err != nil {
fmt.Fprintf(stderr, "[SYS] Ping to URL %q caused an error: %s", url, err)
}
}(job.PingFailure)
2016-02-26 17:44:10 +00:00
default:
fmt.Fprintf(stderr, "[SYS] Execution caused error: %s\n", err)
go func(url string) {
if err := doPing(url); err != nil {
fmt.Fprintf(stderr, "[SYS] Ping to URL %q caused an error: %s", url, err)
}
}(job.PingFailure)
2016-02-26 17:44:10 +00:00
}
}
}
func doPing(url string) error {
if url == "" {
return nil
}
ctx, cancel := context.WithTimeout(context.Background(), cfg.PingTimout)
defer cancel()
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return err
}
resp, err := http.DefaultClient.Do(req.WithContext(ctx))
if err != nil {
return err
}
if resp.StatusCode > 299 {
return fmt.Errorf("Expected HTTP2xx status, got HTTP%d", resp.StatusCode)
}
return nil
}
2016-02-26 17:44:10 +00:00
type messageChanWriter struct {
jobName string
msgChan chan *message
severity int
buffer []byte
}
func (m *messageChanWriter) Write(p []byte) (n int, err error) {
n = len(p)
err = nil
m.buffer = append(m.buffer, p...)
if strings.Contains(string(m.buffer), "\n") {
lines := strings.Split(string(m.buffer), "\n")
for _, l := range lines[:len(lines)-1] {
m.msgChan <- &message{
Date: time.Now(),
JobName: m.jobName,
Message: l,
Severity: m.severity,
}
}
m.buffer = []byte(lines[len(lines)-1])
}
return
}