diff options
| -rw-r--r-- | internal/server/continuous.go | 110 | ||||
| -rw-r--r-- | internal/server/scheduler.go | 32 | ||||
| -rwxr-xr-x | samples/dtail.schema.json | 5 |
3 files changed, 130 insertions, 17 deletions
diff --git a/internal/server/continuous.go b/internal/server/continuous.go new file mode 100644 index 0000000..bdb4079 --- /dev/null +++ b/internal/server/continuous.go @@ -0,0 +1,110 @@ +package server + +import ( + "context" + "fmt" + "os" + "strconv" + "strings" + "time" + + "github.com/mimecast/dtail/internal/clients" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/omode" + + gossh "golang.org/x/crypto/ssh" +) + +type continuous struct { +} + +func newContinuous() *continuous { + return &continuous{} +} + +func (s *continuous) start(ctx context.Context) { + // First run after just 10s! + time.Sleep(time.Second * 10) + s.runJobs(ctx) + + for { + select { + case <-time.After(time.Minute): + s.runJobs(ctx) + case <-ctx.Done(): + return + } + } +} + +func (s *continuous) runJobs(ctx context.Context) { + for _, job := range config.Server.Schedule { + if !job.Enable { + logger.Debug(job.Name, "Not running job as not enabled") + continue + } + + hour, err := strconv.Atoi(time.Now().Format("15")) + if err != nil { + logger.Error(job.Name, "Unable to create job job", err) + continue + } + + if hour < job.TimeRange[0] || hour >= job.TimeRange[1] { + logger.Debug(job.Name, "Not running job out of time range") + continue + } + + files := fillDates(job.Files) + outfile := fillDates(job.Outfile) + + _, err = os.Stat(outfile) + if !os.IsNotExist(err) { + logger.Debug(job.Name, "Not running job as outfile already exists", outfile) + continue + } + + servers := strings.Join(job.Servers, ",") + if servers == "" { + servers = config.Server.SSHBindAddress + } + + args := clients.Args{ + ConnectionsPerCPU: 10, + Discovery: job.Discovery, + ServersStr: servers, + What: files, + Mode: omode.MapClient, + UserName: config.BackgroundUser, + } + + args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(job.Name)) + + tmpOutfile := fmt.Sprintf("%s.tmp", outfile) + query := fmt.Sprintf("%s outfile %s", job.Query, tmpOutfile) + + client, err := clients.NewMaprClient(args, query) + if err != nil { + logger.Error(fmt.Sprintf("Unable to create job job %s", job.Name), err) + continue + } + + jobCtx, cancel := context.WithCancel(ctx) + defer cancel() + + logger.Info(fmt.Sprintf("Starting job job %s", job.Name)) + status := client.Start(jobCtx) + logMessage := fmt.Sprintf("Job exited with status %d", status) + + if err := os.Rename(tmpOutfile, outfile); err == nil { + logger.Info(job.Name, fmt.Sprintf("Renamed %s to %s", tmpOutfile, outfile)) + } + + if status != 0 { + logger.Warn(logMessage) + continue + } + logger.Info(logMessage) + } +} diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go index 8b073a6..1fdbeea 100644 --- a/internal/server/scheduler.go +++ b/internal/server/scheduler.go @@ -39,66 +39,66 @@ func (s *scheduler) start(ctx context.Context) { } func (s *scheduler) runJobs(ctx context.Context) { - for _, scheduled := range config.Server.Schedule { - if !scheduled.Enable { - logger.Debug(scheduled.Name, "Not running job as not enabled") + for _, job := range config.Server.Schedule { + if !job.Enable { + logger.Debug(job.Name, "Not running job as not enabled") continue } hour, err := strconv.Atoi(time.Now().Format("15")) if err != nil { - logger.Error(scheduled.Name, "Unable to create scheduled job", err) + logger.Error(job.Name, "Unable to create job job", err) continue } - if hour < scheduled.TimeRange[0] || hour >= scheduled.TimeRange[1] { - logger.Debug(scheduled.Name, "Not running job out of time range") + if hour < job.TimeRange[0] || hour >= job.TimeRange[1] { + logger.Debug(job.Name, "Not running job out of time range") continue } - files := fillDates(scheduled.Files) - outfile := fillDates(scheduled.Outfile) + files := fillDates(job.Files) + outfile := fillDates(job.Outfile) _, err = os.Stat(outfile) if !os.IsNotExist(err) { - logger.Debug(scheduled.Name, "Not running job as outfile already exists", outfile) + logger.Debug(job.Name, "Not running job as outfile already exists", outfile) continue } - servers := strings.Join(scheduled.Servers, ",") + servers := strings.Join(job.Servers, ",") if servers == "" { servers = config.Server.SSHBindAddress } args := clients.Args{ ConnectionsPerCPU: 10, - Discovery: scheduled.Discovery, + Discovery: job.Discovery, ServersStr: servers, What: files, Mode: omode.MapClient, UserName: config.BackgroundUser, } - args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(scheduled.Name)) + args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(job.Name)) tmpOutfile := fmt.Sprintf("%s.tmp", outfile) - query := fmt.Sprintf("%s outfile %s", scheduled.Query, tmpOutfile) + query := fmt.Sprintf("%s outfile %s", job.Query, tmpOutfile) client, err := clients.NewMaprClient(args, query) if err != nil { - logger.Error(fmt.Sprintf("Unable to create scheduled job %s", scheduled.Name), err) + logger.Error(fmt.Sprintf("Unable to create job job %s", job.Name), err) continue } jobCtx, cancel := context.WithCancel(ctx) defer cancel() - logger.Info(fmt.Sprintf("Starting scheduled job %s", scheduled.Name)) + logger.Info(fmt.Sprintf("Starting job job %s", job.Name)) status := client.Start(jobCtx) logMessage := fmt.Sprintf("Job exited with status %d", status) if err := os.Rename(tmpOutfile, outfile); err == nil { - logger.Info(scheduled.Name, fmt.Sprintf("Renamed %s to %s", tmpOutfile, outfile)) + logger.Info(job.Name, fmt.Sprintf("Renamed %s to %s", tmpOutfile, outfile)) } if status != 0 { diff --git a/samples/dtail.schema.json b/samples/dtail.schema.json index 73807fc..0497fbf 100755 --- a/samples/dtail.schema.json +++ b/samples/dtail.schema.json @@ -43,7 +43,7 @@ } } }, - "Monitor": { + "Continuous": { "type": "array", "items": { "type": "object", @@ -54,6 +54,9 @@ "Enable": { "type": "boolean" }, + "RestartOnDayChange": { + "type": "boolean" + }, "Files": { "type": "string" }, |