From 0495a519499d6cefa289982a94d8f42de541c1f0 Mon Sep 17 00:00:00 2001
From: George Goldberg <george@gberg.me>
Date: Fri, 7 Jul 2017 15:21:02 +0100
Subject: PLT-6916: Redesign the jobs package and Jobserver. (#6733)

This commit redesigns the jobserver to be based around an architecture
of "workers", which carry out jobs of a particular type, and "jobs"
which are a unit of work carried by a particular worker. It also
introduces "schedulers" which are responsible for scheduling jobs of a
particular type automatically (jobs can also be scheduled manually when
apropriate).

Workers may be run many times, either in instances of the platform
binary, or the standalone jobserver binary. In any mattermost cluster,
only one instance of platform OR jobserver must run the schedulers. At
the moment this is controlled by a config variable, but in future will
be controlled through the cluster leader election process.
---
 jobs/jobs_watcher.go | 85 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 85 insertions(+)
 create mode 100644 jobs/jobs_watcher.go

(limited to 'jobs/jobs_watcher.go')

diff --git a/jobs/jobs_watcher.go b/jobs/jobs_watcher.go
new file mode 100644
index 000000000..ada957ccc
--- /dev/null
+++ b/jobs/jobs_watcher.go
@@ -0,0 +1,85 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package jobs
+
+import (
+	"math/rand"
+	"time"
+
+	l4g "github.com/alecthomas/log4go"
+	"github.com/mattermost/platform/model"
+)
+
+const (
+	WATCHER_POLLING_INTERVAL = 15000
+)
+
+type Watcher struct {
+	workers *Workers
+
+	stop    chan bool
+	stopped chan bool
+}
+
+func MakeWatcher(workers *Workers) *Watcher {
+	return &Watcher{
+		stop:    make(chan bool, 1),
+		stopped: make(chan bool, 1),
+		workers: workers,
+	}
+}
+
+func (watcher *Watcher) Start() {
+	l4g.Debug("Watcher Started")
+
+	// Delay for some random number of milliseconds before starting to ensure that multiple
+	// instances of the jobserver  don't poll at a time too close to each other.
+	rand.Seed(time.Now().UTC().UnixNano())
+	_ = <-time.After(time.Duration(rand.Intn(WATCHER_POLLING_INTERVAL)) * time.Millisecond)
+
+	defer func(){
+		l4g.Debug("Watcher Finished")
+		watcher.stopped <- true
+	}()
+
+	for {
+		select {
+		case <-watcher.stop:
+			l4g.Debug("Watcher: Received stop signal")
+			return
+		case <-time.After(WATCHER_POLLING_INTERVAL * time.Millisecond):
+			watcher.PollAndNotify()
+		}
+	}
+}
+
+func (watcher *Watcher) Stop() {
+	l4g.Debug("Watcher Stopping")
+	watcher.stop <- true
+	<-watcher.stopped
+}
+
+func (watcher *Watcher) PollAndNotify() {
+	if result := <-Srv.Store.Job().GetAllByStatus(model.JOB_STATUS_PENDING); result.Err != nil {
+		l4g.Error("Error occured getting all pending statuses: %v", result.Err.Error())
+	} else {
+		jobStatuses := result.Data.([]*model.Job)
+
+		for _, js := range jobStatuses {
+			j := model.Job{
+				Type: js.Type,
+				Id: js.Id,
+			}
+
+			if js.Type == model.JOB_TYPE_DATA_RETENTION {
+				if watcher.workers.DataRetention != nil {
+					select {
+					case watcher.workers.DataRetention.JobChannel() <- j:
+					default:
+					}
+				}
+			}
+		}
+	}
+}
-- 
cgit v1.2.3-1-g7c22