Job metrics & dynamic Grafana annotations (#16650)

* docker: prometheus and grafana for development

Add prometheus and grafana docker configuration when developing with access to the enterprise repository. This simplifies the setup for developers adding new metrics or even examining events from their development server. The services are disabled by default when no enterprise source is detected.

Grafana is provisioned automatically with the Prometheus datasource, as well as the canonical dashboards used with Mattermost. Furthermore, no authentication is required to access Grafana to simplify access from a development environment. The default home dashboard is customized to automatically show links to the provisioned dashboards. Dashboards can be saved, and login remains available via the default `admin/admin` credentials (which initiates a password reset), but no dashboard or system configuration is (currently) persisted if the container is destroyed.

Linux requires slightly special handling (well, really it's Docker for MacOS/Windows), in that `host.docker.internal` is the canonical way to resolve the host on MacOS/Windows, but on Linux it's usually sufficient to just use `localhost`. Until https://github.com/docker/for-linux/issues/264 is resolved, this PR includes code to customize the Prometheus configuration to point at the required address for the running platform.

* metrics: track active jobs by type

* metrics: active jobs chart

* metrics: server start and job annotations

* Update build/docker-compose.common.yml

Co-authored-by: Claudio Costa <cstcld91@gmail.com>

* prometheus: resolve docker host via 172.17.0.1 instead

* skip enabling prometheus and grafana by default

* handle JOB_STATUS_ERROR and JOB_STATUS_CANCELED end states

* handle nil srv.metrics

* lookup job to determine type for metrics

* mocked unit tests for jobs

* goimports lint fixes

* missing license

* add instance to server start tag

* filter annotations by selected instance

Co-authored-by: Claudio Costa <cstcld91@gmail.com>
Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

app/server.go

@@ -1577,7 +1577,7 @@ func (s *Server) ClientConfigHash() string {
 }
 
 func (s *Server) initJobs() {
-	s.Jobs = jobs.NewJobServer(s, s.Store)
+	s.Jobs = jobs.NewJobServer(s, s.Store, s.Metrics)
 	if jobsDataRetentionJobInterface != nil {
 		s.Jobs.DataRetentionJob = jobsDataRetentionJobInterface(s)
 	}

build/docker/grafana/dashboards/mattermost/mattermost-performance-monitoring_rev2.json

@@ -9,6 +9,38 @@
         "iconColor": "rgba(0, 211, 255, 1)",
         "name": "Annotations & Alerts",
         "type": "dashboard"
+      },
+      {
+        "datasource": "Prometheus",
+        "enable": false,
+        "expr": "mattermost_system_server_start_time{instance=~\"$server\"}*1000",
+        "hide": false,
+        "iconColor": "#C8F2C2",
+        "limit": 100,
+        "name": "Server Start",
+        "showIn": 0,
+        "step": "5s",
+        "tagKeys": "instance",
+        "tags": [],
+        "titleFormat": "Server Started",
+        "type": "tags",
+        "useValueForTime": true
+      },
+      {
+        "datasource": "Prometheus",
+        "enable": false,
+        "expr": "mattermost_jobs_active{instance=~\"$server\"}",
+        "hide": false,
+        "iconColor": "#FFF899",
+        "limit": 100,
+        "name": "Active Jobs",
+        "showIn": 0,
+        "step": "5s",
+        "tagKeys": "type",
+        "tags": [],
+        "titleFormat": "Active Job",
+        "type": "tags",
+        "useValueForTime": false
       }
     ]
   },
@@ -1495,6 +1527,107 @@
         "align": false,
         "alignLevel": null
       }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": null,
+      "description": "A record of any job detected as being active within the configured Prometheus polling interval.",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 9,
+        "w": 24,
+        "x": 0,
+        "y": 117
+      },
+      "hiddenSeries": false,
+      "id": 15,
+      "legend": {
+        "alignAsTable": false,
+        "avg": false,
+        "current": false,
+        "hideEmpty": true,
+        "hideZero": true,
+        "max": false,
+        "min": false,
+        "rightSide": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.3.6",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": true,
+      "targets": [
+        {
+          "expr": "mattermost_jobs_active{instance=~\"$server\"}",
+          "instant": false,
+          "interval": "",
+          "legendFormat": "",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Active Jobs",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "label": "Active",
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        },
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false,
+        "alignLevel": null
+      }
     }
   ],
   "refresh": "5s",

einterfaces/metrics.go

@@ -66,4 +66,7 @@ type MetricsInterface interface {
 
 	ObserveEnabledUsers(users int64)
 	GetLoggerMetricsCollector() logr.MetricsCollector
+
+	IncrementJobActive(jobType string)
+	DecrementJobActive(jobType string)
 }

einterfaces/mocks/MetricsInterface.go

@@ -24,6 +24,11 @@ func (_m *MetricsInterface) AddMemCacheMissCounter(cacheName string, amount floa
 	_m.Called(cacheName, amount)
 }
 
+// DecrementJobActive provides a mock function with given fields: jobType
+func (_m *MetricsInterface) DecrementJobActive(jobType string) {
+	_m.Called(jobType)
+}
+
 // DecrementWebSocketBroadcastBufferSize provides a mock function with given fields: hub, amount
 func (_m *MetricsInterface) DecrementWebSocketBroadcastBufferSize(hub string, amount float64) {
 	_m.Called(hub, amount)
@@ -95,6 +100,11 @@ func (_m *MetricsInterface) IncrementHttpRequest() {
 	_m.Called()
 }
 
+// IncrementJobActive provides a mock function with given fields: jobType
+func (_m *MetricsInterface) IncrementJobActive(jobType string) {
+	_m.Called(jobType)
+}
+
 // IncrementLogin provides a mock function with given fields:
 func (_m *MetricsInterface) IncrementLogin() {
 	_m.Called()

jobs/jobs.go

@@ -59,6 +59,10 @@ func (srv *JobServer) ClaimJob(job *model.Job) (bool, *model.AppError) {
 		return false, model.NewAppError("ClaimJob", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
 
+	if updated && srv.metrics != nil {
+		srv.metrics.IncrementJobActive(job.Type)
+	}
+
 	return updated, nil
 }
 
@@ -83,6 +87,11 @@ func (srv *JobServer) SetJobSuccess(job *model.Job) *model.AppError {
 	if _, err := srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_SUCCESS); err != nil {
 		return model.NewAppError("SetJobSuccess", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
+
+	if srv.metrics != nil {
+		srv.metrics.DecrementJobActive(job.Type)
+	}
+
 	return nil
 }
 
@@ -93,6 +102,10 @@ func (srv *JobServer) SetJobError(job *model.Job, jobError *model.AppError) *mod
 			return model.NewAppError("SetJobError", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
 		}
 
+		if srv.metrics != nil {
+			srv.metrics.DecrementJobActive(job.Type)
+		}
+
 		return nil
 	}
 
@@ -109,6 +122,9 @@ func (srv *JobServer) SetJobError(job *model.Job, jobError *model.AppError) *mod
 	if err != nil {
 		return model.NewAppError("SetJobError", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
+	if updated && srv.metrics != nil {
+		srv.metrics.DecrementJobActive(job.Type)
+	}
 
 	if !updated {
 		updated, err = srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_CANCEL_REQUESTED)
@@ -127,6 +143,11 @@ func (srv *JobServer) SetJobCanceled(job *model.Job) *model.AppError {
 	if _, err := srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_CANCELED); err != nil {
 		return model.NewAppError("SetJobCanceled", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
+
+	if srv.metrics != nil {
+		srv.metrics.DecrementJobActive(job.Type)
+	}
+
 	return nil
 }
 
@@ -145,6 +166,15 @@ func (srv *JobServer) RequestCancellation(jobId string) *model.AppError {
 		return model.NewAppError("RequestCancellation", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
 	if updated {
+		if srv.metrics != nil {
+			job, err := srv.GetJob(jobId)
+			if err != nil {
+				return model.NewAppError("RequestCancellation", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
+			}
+
+			srv.metrics.DecrementJobActive(job.Type)
+		}
+
 		return nil
 	}
 

jobs/jobs_test.go

@@ -0,0 +1,545 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package jobs
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mattermost/mattermost-server/v5/einterfaces/mocks"
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/store"
+	"github.com/mattermost/mattermost-server/v5/store/storetest"
+	"github.com/mattermost/mattermost-server/v5/utils/testutils"
+)
+
+func makeJobServer(t *testing.T) (*JobServer, *storetest.Store, *mocks.MetricsInterface) {
+	configService := &testutils.StaticConfigService{}
+
+	mockStore := &storetest.Store{}
+	t.Cleanup(func() {
+		mockStore.AssertExpectations(t)
+	})
+
+	mockMetrics := &mocks.MetricsInterface{}
+	t.Cleanup(func() {
+		mockMetrics.AssertExpectations(t)
+	})
+
+	jobServer := NewJobServer(configService, mockStore, mockMetrics)
+
+	return jobServer, mockStore, mockMetrics
+}
+
+func expectErrorId(t *testing.T, errId string, appErr *model.AppError) {
+	t.Helper()
+	require.NotNil(t, appErr)
+	require.Equal(t, errId, appErr.Id)
+}
+
+func makeTeamEditionJobServer(t *testing.T) (*JobServer, *storetest.Store) {
+	configService := &testutils.StaticConfigService{}
+
+	mockStore := &storetest.Store{}
+	t.Cleanup(func() {
+		mockStore.AssertExpectations(t)
+	})
+
+	jobServer := NewJobServer(configService, mockStore, nil)
+
+	return jobServer, mockStore
+}
+
+func TestClaimJob(t *testing.T) {
+	t.Run("error claiming job", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_IN_PROGRESS).Return(false, &model.AppError{Message: "message"})
+
+		updated, err := jobServer.ClaimJob(job)
+		expectErrorId(t, "app.job.update.app_error", err)
+		require.False(t, updated)
+	})
+
+	t.Run("no existing job to update", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_IN_PROGRESS).Return(false, nil)
+
+		updated, err := jobServer.ClaimJob(job)
+		require.Nil(t, err)
+		require.False(t, updated)
+	})
+
+	t.Run("pending job updated", func(t *testing.T) {
+		jobServer, mockStore, mockMetrics := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_IN_PROGRESS).Return(true, nil)
+		mockMetrics.On("IncrementJobActive", "job_type")
+
+		updated, err := jobServer.ClaimJob(job)
+		require.Nil(t, err)
+		require.True(t, updated)
+	})
+
+	t.Run("pending job updated, nil metrics service", func(t *testing.T) {
+		jobServer, mockStore := makeTeamEditionJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_IN_PROGRESS).Return(true, nil)
+
+		updated, err := jobServer.ClaimJob(job)
+		require.Nil(t, err)
+		require.True(t, updated)
+	})
+}
+
+func TestSetJobProgress(t *testing.T) {
+	t.Run("error setting progress", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		progress := int64(50)
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		job.Status = model.JOB_STATUS_IN_PROGRESS
+		job.Progress = progress
+
+		mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(false, &model.AppError{Message: "message"})
+
+		err := jobServer.SetJobProgress(job, progress)
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("progress updated", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		progress := int64(50)
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		job.Status = model.JOB_STATUS_IN_PROGRESS
+		job.Progress = progress
+
+		mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(true, nil)
+
+		err := jobServer.SetJobProgress(job, progress)
+		require.Nil(t, err)
+	})
+}
+
+func TestSetJobWarning(t *testing.T) {
+	t.Run("error setting status", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_WARNING).Return(job, &model.AppError{Message: "message"})
+
+		err := jobServer.SetJobWarning(job)
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("status updated", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_WARNING).Return(job, nil)
+
+		err := jobServer.SetJobWarning(job)
+		require.Nil(t, err)
+	})
+}
+
+func TestSetJobSuccess(t *testing.T) {
+	t.Run("error setting status", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_SUCCESS).Return(job, &model.AppError{Message: "message"})
+
+		err := jobServer.SetJobSuccess(job)
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("status updated", func(t *testing.T) {
+		jobServer, mockStore, mockMetrics := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_SUCCESS).Return(job, nil)
+		mockMetrics.On("DecrementJobActive", "job_type")
+
+		err := jobServer.SetJobSuccess(job)
+		require.Nil(t, err)
+	})
+
+	t.Run("status updated, nil metrics service", func(t *testing.T) {
+		jobServer, mockStore := makeTeamEditionJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_SUCCESS).Return(job, nil)
+
+		err := jobServer.SetJobSuccess(job)
+		require.Nil(t, err)
+	})
+}
+
+func TestSetJobError(t *testing.T) {
+	t.Run("nil provided job error", func(t *testing.T) {
+		t.Run("error setting status", func(t *testing.T) {
+			jobServer, mockStore, _ := makeJobServer(t)
+
+			job := &model.Job{
+				Id:   "job_id",
+				Type: "job_type",
+			}
+
+			mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_ERROR).Return(job, &model.AppError{Message: "message"})
+
+			err := jobServer.SetJobError(job, nil)
+			expectErrorId(t, "app.job.update.app_error", err)
+		})
+
+		t.Run("status updated", func(t *testing.T) {
+			jobServer, mockStore, mockMetrics := makeJobServer(t)
+
+			job := &model.Job{
+				Id:   "job_id",
+				Type: "job_type",
+			}
+
+			mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_ERROR).Return(job, nil)
+			mockMetrics.On("DecrementJobActive", "job_type")
+
+			err := jobServer.SetJobError(job, nil)
+			require.Nil(t, err)
+		})
+
+		t.Run("status updated, nil metrics service", func(t *testing.T) {
+			jobServer, mockStore := makeTeamEditionJobServer(t)
+
+			job := &model.Job{
+				Id:   "job_id",
+				Type: "job_type",
+			}
+
+			mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_ERROR).Return(job, nil)
+
+			err := jobServer.SetJobError(job, nil)
+			require.Nil(t, err)
+		})
+	})
+
+	t.Run("provided job error", func(t *testing.T) {
+		t.Run("error setting status", func(t *testing.T) {
+			jobServer, mockStore, _ := makeJobServer(t)
+
+			jobError := &model.AppError{Message: "message"}
+
+			job := &model.Job{
+				Id:       "job_id",
+				Type:     "job_type",
+				Progress: -1,
+				Data:     map[string]string{"error": jobError.Message},
+			}
+
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(false, &model.AppError{Message: "message"})
+
+			err := jobServer.SetJobError(job, jobError)
+			expectErrorId(t, "app.job.update.app_error", err)
+		})
+
+		t.Run("status updated", func(t *testing.T) {
+			jobServer, mockStore, mockMetrics := makeJobServer(t)
+
+			jobError := &model.AppError{Message: "message"}
+
+			job := &model.Job{
+				Id:       "job_id",
+				Type:     "job_type",
+				Progress: -1,
+				Data:     map[string]string{"error": jobError.Message},
+			}
+
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(true, nil)
+			mockMetrics.On("DecrementJobActive", "job_type")
+
+			err := jobServer.SetJobError(job, jobError)
+			require.Nil(t, err)
+		})
+
+		t.Run("status updated, nil metrics service", func(t *testing.T) {
+			jobServer, mockStore := makeTeamEditionJobServer(t)
+
+			jobError := &model.AppError{Message: "message"}
+
+			job := &model.Job{
+				Id:       "job_id",
+				Type:     "job_type",
+				Progress: -1,
+				Data:     map[string]string{"error": jobError.Message},
+			}
+
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(true, nil)
+
+			err := jobServer.SetJobError(job, jobError)
+			require.Nil(t, err)
+		})
+
+		t.Run("status not updated, request cancellation, error setting status", func(t *testing.T) {
+			jobServer, mockStore, _ := makeJobServer(t)
+
+			jobError := &model.AppError{Message: "message"}
+
+			job := &model.Job{
+				Id:       "job_id",
+				Type:     "job_type",
+				Progress: -1,
+				Data:     map[string]string{"error": jobError.Message},
+			}
+
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(false, nil)
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_CANCEL_REQUESTED).Return(false, &model.AppError{Message: "message"})
+
+			err := jobServer.SetJobError(job, jobError)
+			expectErrorId(t, "app.job.update.app_error", err)
+		})
+
+		t.Run("status not updated, request cancellation, status not updated", func(t *testing.T) {
+			jobServer, mockStore, _ := makeJobServer(t)
+
+			jobError := &model.AppError{Message: "message"}
+
+			job := &model.Job{
+				Id:       "job_id",
+				Type:     "job_type",
+				Progress: -1,
+				Data:     map[string]string{"error": jobError.Message},
+			}
+
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(false, nil)
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_CANCEL_REQUESTED).Return(false, nil)
+
+			err := jobServer.SetJobError(job, jobError)
+			expectErrorId(t, "jobs.set_job_error.update.error", err)
+		})
+
+		t.Run("status not updated, request cancellation, status updated", func(t *testing.T) {
+			jobServer, mockStore, _ := makeJobServer(t)
+
+			jobError := &model.AppError{Message: "message"}
+
+			job := &model.Job{
+				Id:       "job_id",
+				Type:     "job_type",
+				Progress: -1,
+				Data:     map[string]string{"error": jobError.Message},
+			}
+
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(false, nil)
+			mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_CANCEL_REQUESTED).Return(true, nil)
+
+			err := jobServer.SetJobError(job, jobError)
+			require.Nil(t, err)
+		})
+	})
+}
+
+func TestSetJobCanceled(t *testing.T) {
+	t.Run("error setting status", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_CANCELED).Return(job, &model.AppError{Message: "message"})
+
+		err := jobServer.SetJobCanceled(job)
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("status updated", func(t *testing.T) {
+		jobServer, mockStore, mockMetrics := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_CANCELED).Return(job, nil)
+		mockMetrics.On("DecrementJobActive", "job_type")
+
+		err := jobServer.SetJobCanceled(job)
+		require.Nil(t, err)
+	})
+
+	t.Run("status updated, nil metrics service", func(t *testing.T) {
+		jobServer, mockStore := makeTeamEditionJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatus", "job_id", model.JOB_STATUS_CANCELED).Return(job, nil)
+
+		err := jobServer.SetJobCanceled(job)
+		require.Nil(t, err)
+	})
+}
+
+func TestUpdateInProgressJobData(t *testing.T) {
+	t.Run("error updating", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		job.Status = model.JOB_STATUS_IN_PROGRESS
+
+		mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(false, &model.AppError{Message: "message"})
+
+		err := jobServer.UpdateInProgressJobData(job)
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("progress updated", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		job.Status = model.JOB_STATUS_IN_PROGRESS
+
+		mockStore.JobStore.On("UpdateOptimistically", job, model.JOB_STATUS_IN_PROGRESS).Return(true, nil)
+
+		err := jobServer.UpdateInProgressJobData(job)
+		require.Nil(t, err)
+	})
+}
+
+func TestRequestCancellation(t *testing.T) {
+	t.Run("error cancelling", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED).Return(false, &model.AppError{Message: "message"})
+
+		err := jobServer.RequestCancellation("job_id")
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("cancelled, job not found", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED).Return(true, nil)
+		mockStore.JobStore.On("Get", "job_id").Return(nil, &store.ErrNotFound{})
+
+		err := jobServer.RequestCancellation("job_id")
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("cancelled, success", func(t *testing.T) {
+		jobServer, mockStore, mockMetrics := makeJobServer(t)
+
+		job := &model.Job{
+			Id:   "job_id",
+			Type: "job_type",
+		}
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED).Return(true, nil)
+		mockStore.JobStore.On("Get", "job_id").Return(job, nil)
+		mockMetrics.On("DecrementJobActive", "job_type")
+
+		err := jobServer.RequestCancellation("job_id")
+		require.Nil(t, err)
+	})
+
+	t.Run("cancelled, success, nil metrics service", func(t *testing.T) {
+		jobServer, mockStore := makeTeamEditionJobServer(t)
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED).Return(true, nil)
+
+		err := jobServer.RequestCancellation("job_id")
+		require.Nil(t, err)
+	})
+
+	t.Run("unable to cancel, requesting cancellation instead, error setting status", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED).Return(false, nil)
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_IN_PROGRESS, model.JOB_STATUS_CANCEL_REQUESTED).Return(false, &model.AppError{Message: "message"})
+
+		err := jobServer.RequestCancellation("job_id")
+		expectErrorId(t, "app.job.update.app_error", err)
+	})
+
+	t.Run("unable to cancel, requesting cancellation instead, success", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED).Return(false, nil)
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_IN_PROGRESS, model.JOB_STATUS_CANCEL_REQUESTED).Return(true, nil)
+
+		err := jobServer.RequestCancellation("job_id")
+		require.Nil(t, err)
+	})
+
+	t.Run("unable to cancel, requesting cancellation instead, unexpected state", func(t *testing.T) {
+		jobServer, mockStore, _ := makeJobServer(t)
+
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED).Return(false, nil)
+		mockStore.JobStore.On("UpdateStatusOptimistically", "job_id", model.JOB_STATUS_IN_PROGRESS, model.JOB_STATUS_CANCEL_REQUESTED).Return(false, nil)
+
+		err := jobServer.RequestCancellation("job_id")
+		expectErrorId(t, "jobs.request_cancellation.status.error", err)
+	})
+}

jobs/server.go

@@ -4,6 +4,7 @@
 package jobs
 
 import (
+	"github.com/mattermost/mattermost-server/v5/einterfaces"
 	ejobs "github.com/mattermost/mattermost-server/v5/einterfaces/jobs"
 	tjobs "github.com/mattermost/mattermost-server/v5/jobs/interfaces"
 	"github.com/mattermost/mattermost-server/v5/model"
@@ -14,6 +15,7 @@ import (
 type JobServer struct {
 	ConfigService configservice.ConfigService
 	Store         store.Store
+	metrics       einterfaces.MetricsInterface
 	Workers       *Workers
 	Schedulers    *Schedulers
 
@@ -32,10 +34,11 @@ type JobServer struct {
 	Cloud                   ejobs.CloudJobInterface
 }
 
-func NewJobServer(configService configservice.ConfigService, store store.Store) *JobServer {
+func NewJobServer(configService configservice.ConfigService, store store.Store, metrics einterfaces.MetricsInterface) *JobServer {
 	return &JobServer{
 		ConfigService: configService,
 		Store:         store,
+		metrics:       metrics,
 	}
 }