From f8caa6c84167e45519fc633d2cb5e063e179b724 Mon Sep 17 00:00:00 2001 From: Jesse Hallam Date: Wed, 20 Jan 2021 15:43:02 -0400 Subject: [PATCH] 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 * 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 Co-authored-by: Mattermod --- app/server.go | 2 +- ...attermost-performance-monitoring_rev2.json | 133 +++++ einterfaces/metrics.go | 3 + einterfaces/mocks/MetricsInterface.go | 10 + jobs/jobs.go | 30 + jobs/jobs_test.go | 545 ++++++++++++++++++ jobs/server.go | 5 +- 7 files changed, 726 insertions(+), 2 deletions(-) create mode 100644 jobs/jobs_test.go diff --git a/app/server.go b/app/server.go index df7669a2ad..1302d03b70 100644 --- a/app/server.go +++ b/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) } diff --git a/build/docker/grafana/dashboards/mattermost/mattermost-performance-monitoring_rev2.json b/build/docker/grafana/dashboards/mattermost/mattermost-performance-monitoring_rev2.json index 27ca40e54e..84544ad6bb 100644 --- a/build/docker/grafana/dashboards/mattermost/mattermost-performance-monitoring_rev2.json +++ b/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", diff --git a/einterfaces/metrics.go b/einterfaces/metrics.go index a2ae546fb5..ea4434aca0 100644 --- a/einterfaces/metrics.go +++ b/einterfaces/metrics.go @@ -66,4 +66,7 @@ type MetricsInterface interface { ObserveEnabledUsers(users int64) GetLoggerMetricsCollector() logr.MetricsCollector + + IncrementJobActive(jobType string) + DecrementJobActive(jobType string) } diff --git a/einterfaces/mocks/MetricsInterface.go b/einterfaces/mocks/MetricsInterface.go index 7fb2606153..f4547baf3e 100644 --- a/einterfaces/mocks/MetricsInterface.go +++ b/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() diff --git a/jobs/jobs.go b/jobs/jobs.go index 48f274175f..d4638f1cb5 100644 --- a/jobs/jobs.go +++ b/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 } diff --git a/jobs/jobs_test.go b/jobs/jobs_test.go new file mode 100644 index 0000000000..1571c10857 --- /dev/null +++ b/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) + }) +} diff --git a/jobs/server.go b/jobs/server.go index f69e0b1e65..b89ea595f2 100644 --- a/jobs/server.go +++ b/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, } }