initial server side continuous mapreduce runner

2 files changed, 126 insertions, 16 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 {