merge develop

1 files changed, 15 insertions, 22 deletions

diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go
index a1e9e36..0ba65f7 100644
--- a/internal/server/scheduler.go
+++ b/internal/server/scheduler.go
@@ -10,25 +10,23 @@ import (
 
 	"github.com/mimecast/dtail/internal/clients"
 	"github.com/mimecast/dtail/internal/config"
-	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/io/dlog"
 	"github.com/mimecast/dtail/internal/omode"
 
 	gossh "golang.org/x/crypto/ssh"
 )
 
-type scheduler struct {
-}
+type scheduler struct{}
 
 func newScheduler() *scheduler {
 	return &scheduler{}
 }
 
 func (s *scheduler) start(ctx context.Context) {
-	logger.Info("Starting scheduled job runner after 10s")
+	dlog.Server.Info("Starting scheduled job runner after 10s")
 	// First run after just 10s!
 	time.Sleep(time.Second * 10)
 	s.runJobs(ctx)
-
 	for {
 		select {
 		case <-time.After(time.Minute):
@@ -42,27 +40,24 @@ func (s *scheduler) start(ctx context.Context) {
 func (s *scheduler) runJobs(ctx context.Context) {
 	for _, job := range config.Server.Schedule {
 		if !job.Enable {
-			logger.Debug(job.Name, "Not running job as not enabled")
+			dlog.Server.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", err)
+			dlog.Server.Error(job.Name, "Unable to create job", err)
 			continue
 		}
-
 		if hour < job.TimeRange[0] || hour >= job.TimeRange[1] {
-			logger.Debug(job.Name, "Not running job out of time range")
+			dlog.Server.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)
+			dlog.Server.Debug(job.Name, "Not running job as outfile already exists", outfile)
 			continue
 		}
 
@@ -70,9 +65,8 @@ func (s *scheduler) runJobs(ctx context.Context) {
 		if servers == "" {
 			servers = config.Server.SSHBindAddress
 		}
-
-		args := clients.Args{
-			ConnectionsPerCPU: 10,
+		args := config.Args{
+			ConnectionsPerCPU: config.DefaultConnectionsPerCPU,
 			Discovery:         job.Discovery,
 			ServersStr:        servers,
 			What:              files,
@@ -81,25 +75,24 @@ func (s *scheduler) runJobs(ctx context.Context) {
 		}
 
 		args.SSHAuthMethods = append(args.SSHAuthMethods, gossh.Password(job.Name))
-
-		query := fmt.Sprintf("%s outfile %s", job.Query, outfile)
-		client, err := clients.NewMaprClient(args, query, clients.CumulativeMode)
+		args.QueryStr = fmt.Sprintf("%s outfile %s", job.Query, outfile)
+		client, err := clients.NewMaprClient(args, clients.CumulativeMode)
 		if err != nil {
-			logger.Error(fmt.Sprintf("Unable to create job %s", job.Name), err)
+			dlog.Server.Error(fmt.Sprintf("Unable to create job %s", job.Name), err)
 			continue
 		}
 
 		jobCtx, cancel := context.WithCancel(ctx)
 		defer cancel()
 
-		logger.Info(fmt.Sprintf("Starting job %s", job.Name))
+		dlog.Server.Info(fmt.Sprintf("Starting job %s", job.Name))
 		status := client.Start(jobCtx, make(chan string))
 		logMessage := fmt.Sprintf("Job exited with status %d", status)
 
 		if status != 0 {
-			logger.Warn(logMessage)
+			dlog.Server.Warn(logMessage)
 			continue
 		}
-		logger.Info(logMessage)
+		dlog.Server.Info(logMessage)
 	}
 }