mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

[MM-56074] mmctl job commands (#26855)

Просмотр исходного кода 
* add job list and update  job status command to mmctl
Этот коммит содержится в:
Ben Cooke
2024-06-17 12:07:05 -04:00
коммит произвёл GitHub
родитель 5894abc36e
Коммит 9187c772b6
38 изменённых файлов: 1423 добавлений и 101 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

105
server/channels/api4/job.go
Просмотреть файл

@@ -23,6 +23,7 @@ func (api *API) InitJob() {
api.BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}/download", api.APISessionRequiredTrustRequester(downloadJob)).Methods("GET")
api.BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}/cancel", api.APISessionRequired(cancelJob)).Methods("POST")
api.BaseRoutes.Jobs.Handle("/type/{job_type:[A-Za-z0-9_-]+}", api.APISessionRequired(getJobsByType)).Methods("GET")
api.BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}/status", api.APISessionRequired(updateJobStatus)).Methods("PATCH")
}
func getJob(c *Context, w http.ResponseWriter, r *http.Request) {
@@ -147,23 +148,58 @@ func getJobs(c *Context, w http.ResponseWriter, r *http.Request) {
return
}
jobType := r.URL.Query().Get("job_type")
var validJobTypes []string
for _, jobType := range model.AllJobTypes {
if jobType != "" {
isValidJobType := model.IsValidJobType(jobType)
if !isValidJobType {
c.SetInvalidURLParam("job_type")
return
}
hasPermission, permissionRequired := c.App.SessionHasPermissionToReadJob(*c.AppContext.Session(), jobType)
if permissionRequired == nil {
c.Logger.Warn("The job types of a job you are trying to retrieve does not contain permissions", mlog.String("jobType", jobType))
continue
c.Err = model.NewAppError("getJobsByType", "api.job.retrieve.nopermissions", nil, "", http.StatusBadRequest)
return
}
if hasPermission {
validJobTypes = append(validJobTypes, jobType)
if !hasPermission {
c.SetPermissionError(permissionRequired)
return
}
validJobTypes = append(validJobTypes, jobType)
} else {
for _, jType := range model.AllJobTypes {
hasPermission, permissionRequired := c.App.SessionHasPermissionToReadJob(*c.AppContext.Session(), jType)
if permissionRequired == nil {
c.Logger.Warn("The job types of a job you are trying to retrieve does not contain permissions", mlog.String("jobType", jType))
continue
}
if hasPermission {
validJobTypes = append(validJobTypes, jType)
}
}
}
if len(validJobTypes) == 0 {
c.SetPermissionError()
return
}
jobs, appErr := c.App.GetJobsByTypesPage(c.AppContext, validJobTypes, c.Params.Page, c.Params.PerPage)
status := r.URL.Query().Get("status")
isValidStatus := model.IsValidJobStatus(status)
if status != "" && !isValidStatus {
c.Err = model.NewAppError("getJobs", "api.job.status.invalid", nil, "", http.StatusBadRequest)
}
var jobs []*model.Job
var appErr *model.AppError
if status == "" {
jobs, appErr = c.App.GetJobsByTypesPage(c.AppContext, validJobTypes, c.Params.Page, c.Params.PerPage)
} else {
jobs, appErr = c.App.GetJobsByTypeAndStatus(c.AppContext, validJobTypes, status, c.Params.Page, c.Params.PerPage)
}
if appErr != nil {
c.Err = appErr
return
@@ -248,3 +284,60 @@ func cancelJob(c *Context, w http.ResponseWriter, r *http.Request) {
ReturnStatusOK(w)
}
func updateJobStatus(c *Context, w http.ResponseWriter, r *http.Request) {
c.RequireJobId()
if c.Err != nil {
return
}
auditRec := c.MakeAuditRecord("updateJobStatus", audit.Fail)
defer c.LogAuditRec(auditRec)
audit.AddEventParameter(auditRec, "job_id", c.Params.JobId)
props := model.StringInterfaceFromJSON(r.Body)
status, ok := props["status"].(string)
if !ok {
c.SetInvalidParam("status")
return
}
force, ok := props["force"].(bool)
if !ok {
force = false
}
job, err := c.App.GetJob(c.AppContext, c.Params.JobId)
if err != nil {
c.Err = err
return
}
auditRec.AddEventPriorState(job)
auditRec.AddEventObjectType("job")
hasPermission, permissionRequired := c.App.SessionHasPermissionToManageJob(*c.AppContext.Session(), job)
if permissionRequired == nil {
c.Err = model.NewAppError("updateJobStatus", "api.job.unable_to_manage_job.incorrect_job_type", nil, "", http.StatusBadRequest)
return
}
if !hasPermission {
c.SetPermissionError(permissionRequired)
return
}
if !force && !job.IsValidStatusChange(status) {
c.Err = model.NewAppError("updateJobStatus", "api.job.status.invalid", nil, "", http.StatusBadRequest)
return
}
if err := c.App.UpdateJobStatus(c.AppContext, job, status); err != nil {
c.Err = err
return
}
auditRec.Success()
ReturnStatusOK(w)
}

1
server/channels/api4/job_local.go
Просмотреть файл

@@ -9,4 +9,5 @@ func (api *API) InitJobLocal() {
api.BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}", api.APILocal(getJob)).Methods("GET")
api.BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}/cancel", api.APILocal(cancelJob)).Methods("POST")
api.BaseRoutes.Jobs.Handle("/type/{job_type:[A-Za-z0-9_-]+}", api.APILocal(getJobsByType)).Methods("GET")
api.BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}/status", api.APILocal(updateJobStatus)).Methods("PATCH")
}

120
server/channels/api4/job_test.go
Просмотреть файл

@@ -102,6 +102,12 @@ func TestGetJobs(t *testing.T) {
Type: jobType,
CreateAt: t0 + 2,
},
{
Id: model.NewId(),
Type: model.JobTypeLdapSync,
CreateAt: t0 + 3,
Status: model.JobStatusPending,
},
}
for _, job := range jobs {
@@ -110,21 +116,47 @@ func TestGetJobs(t *testing.T) {
defer th.App.Srv().Store().Job().Delete(job.Id)
}
received, _, err := th.SystemAdminClient.GetJobs(context.Background(), 0, 2)
require.NoError(t, err)
t.Run("Get 2 jobs", func(t *testing.T) {
received, _, err := th.SystemAdminClient.GetJobs(context.Background(), "", "", 0, 2)
require.NoError(t, err)
require.Len(t, received, 2, "received wrong number of jobs")
require.Equal(t, jobs[2].Id, received[0].Id, "should've received newest job first")
require.Equal(t, jobs[0].Id, received[1].Id, "should've received second newest job second")
require.Len(t, received, 2, "received wrong number of jobs")
require.Equal(t, jobs[3].Id, received[0].Id, "should've received newest job first")
require.Equal(t, jobs[2].Id, received[1].Id, "should've received second newest job second")
})
received, _, err = th.SystemAdminClient.GetJobs(context.Background(), 1, 2)
require.NoError(t, err)
t.Run("Get oldest job using paging", func(t *testing.T) {
received, _, err := th.SystemAdminClient.GetJobs(context.Background(), "", "", 1, 3)
require.NoError(t, err)
require.Equal(t, jobs[1].Id, received[0].Id, "should've received oldest job last")
})
require.Equal(t, jobs[1].Id, received[0].Id, "should've received oldest job last")
t.Run("Return error fetching job without permissions", func(t *testing.T) {
_, resp, err := th.Client.GetJobs(context.Background(), "", "", 0, 60)
require.Error(t, err)
CheckForbiddenStatus(t, resp)
})
_, resp, err := th.Client.GetJobs(context.Background(), 0, 60)
require.Error(t, err)
CheckForbiddenStatus(t, resp)
t.Run("Get job by type", func(t *testing.T) {
received, _, err := th.SystemAdminClient.GetJobs(context.Background(), model.JobTypeLdapSync, "", 0, 3)
require.NoError(t, err)
require.Len(t, received, 1, "received wrong number of jobs")
require.Equal(t, jobs[3].Id, received[0].Id, "should've received the ldap sync job")
})
t.Run("Get job by status", func(t *testing.T) {
received, _, err := th.SystemAdminClient.GetJobs(context.Background(), "", model.JobStatusPending, 0, 3)
require.NoError(t, err)
require.Len(t, received, 1, "received wrong number of jobs")
require.Equal(t, jobs[3].Id, received[0].Id, "should've received the ldap sync job")
})
t.Run("Get job by type and status", func(t *testing.T) {
received, _, err := th.SystemAdminClient.GetJobs(context.Background(), model.JobTypeLdapSync, model.JobStatusPending, 0, 3)
require.NoError(t, err)
require.Len(t, received, 1, "received wrong number of jobs")
require.Equal(t, jobs[3].Id, received[0].Id, "should've received the ldap sync job")
})
}
func TestGetJobsByType(t *testing.T) {
@@ -336,3 +368,69 @@ func TestCancelJob(t *testing.T) {
require.Error(t, err)
CheckNotFoundStatus(t, resp)
}
func TestUpdateJobStatus(t *testing.T) {
th := Setup(t)
defer th.TearDown()
jobType := model.JobTypeDataRetention
jobs := []*model.Job{
{
Id: model.NewId(),
Type: jobType,
Status: model.JobStatusPending,
},
{
Id: model.NewId(),
Type: jobType,
Status: model.JobStatusInProgress,
},
{
Id: model.NewId(),
Type: jobType,
Status: model.JobStatusSuccess,
},
{
Id: model.NewId(),
Type: jobType,
Status: model.JobStatusPending,
},
}
for _, job := range jobs {
_, err := th.App.Srv().Store().Job().Save(job)
require.NoError(t, err)
defer th.App.Srv().Store().Job().Delete(job.Id)
}
t.Run("Fail to update job status without permission", func(t *testing.T) {
resp, err := th.Client.UpdateJobStatus(context.Background(), jobs[0].Id, model.JobStatusCancelRequested, false)
require.Error(t, err)
CheckForbiddenStatus(t, resp)
})
t.Run("Change a pending job to cancel requested without force with sysadmin client", func(t *testing.T) {
_, err := th.SystemAdminClient.UpdateJobStatus(context.Background(), jobs[0].Id, model.JobStatusCancelRequested, false)
require.NoError(t, err)
})
t.Run("Change a pending job to cancel requested without force with local client", func(t *testing.T) {
_, err := th.LocalClient.UpdateJobStatus(context.Background(), jobs[3].Id, model.JobStatusCancelRequested, false)
require.NoError(t, err)
})
t.Run("Fail to change a pending job to canceled without force", func(t *testing.T) {
th.TestForSystemAdminAndLocal(t, func(t *testing.T, client *model.Client4) {
resp, err := client.UpdateJobStatus(context.Background(), jobs[0].Id, model.JobStatusCanceled, false)
require.Error(t, err)
CheckBadRequestStatus(t, resp)
})
})
t.Run("Change a pending job to canceled with force", func(t *testing.T) {
th.TestForSystemAdminAndLocal(t, func(t *testing.T, client *model.Client4) {
_, err := client.UpdateJobStatus(context.Background(), jobs[0].Id, model.JobStatusCanceled, true)
require.NoError(t, err)
})
})
}

3
server/channels/app/app_iface.go
Просмотреть файл

@@ -720,6 +720,7 @@ type AppIface interface {
GetIncomingWebhooksPageByUser(userID string, page, perPage int) ([]*model.IncomingWebhook, *model.AppError)
GetJob(c request.CTX, id string) (*model.Job, *model.AppError)
GetJobsByType(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, *model.AppError)
GetJobsByTypeAndStatus(c request.CTX, jobTypes []string, status string, page int, perPage int) ([]*model.Job, *model.AppError)
GetJobsByTypePage(c request.CTX, jobType string, page int, perPage int) ([]*model.Job, *model.AppError)
GetJobsByTypes(c request.CTX, jobTypes []string, offset int, limit int) ([]*model.Job, *model.AppError)
GetJobsByTypesPage(c request.CTX, jobType []string, page int, perPage int) ([]*model.Job, *model.AppError)
@@ -1099,6 +1100,7 @@ type AppIface interface {
SessionHasPermissionToChannelByPost(session model.Session, postID string, permission *model.Permission) bool
SessionHasPermissionToCreateJob(session model.Session, job *model.Job) (bool, *model.Permission)
SessionHasPermissionToGroup(session model.Session, groupID string, permission *model.Permission) bool
SessionHasPermissionToManageJob(session model.Session, job *model.Job) (bool, *model.Permission)
SessionHasPermissionToReadJob(session model.Session, jobType string) (bool, *model.Permission)
SessionHasPermissionToTeam(session model.Session, teamID string, permission *model.Permission) bool
SessionHasPermissionToUser(session model.Session, userID string) bool
@@ -1173,6 +1175,7 @@ type AppIface interface {
UpdateHashedPassword(user *model.User, newHashedPassword string) *model.AppError
UpdateHashedPasswordByUserId(userID, newHashedPassword string) *model.AppError
UpdateIncomingWebhook(oldHook, updatedHook *model.IncomingWebhook) (*model.IncomingWebhook, *model.AppError)
UpdateJobStatus(c request.CTX, job *model.Job, newStatus string) *model.AppError
UpdateMfa(c request.CTX, activate bool, userID, token string) *model.AppError
UpdateMobileAppBadge(userID string)
UpdateOAuthApp(oldApp, updatedApp *model.OAuthApp) (*model.OAuthApp, *model.AppError)

59
server/channels/app/job.go
Просмотреть файл

@@ -52,6 +52,14 @@ func (a *App) GetJobsByTypes(c request.CTX, jobTypes []string, offset int, limit
return jobs, nil
}
func (a *App) GetJobsByTypeAndStatus(c request.CTX, jobTypes []string, status string, page int, perPage int) ([]*model.Job, *model.AppError) {
jobs, err := a.Srv().Store().Job().GetAllByTypeAndStatusPage(c, jobTypes, status, page*perPage, perPage)
if err != nil {
return nil, model.NewAppError("GetAllByTypeAndStatusPage", "app.job.get_all.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
}
return jobs, nil
}
func (a *App) CreateJob(c request.CTX, job *model.Job) (*model.Job, *model.AppError) {
return a.Srv().Jobs.CreateJob(c, job.Type, job.Data)
}
@@ -60,6 +68,19 @@ func (a *App) CancelJob(c request.CTX, jobId string) *model.AppError {
return a.Srv().Jobs.RequestCancellation(c, jobId)
}
func (a *App) UpdateJobStatus(c request.CTX, job *model.Job, newStatus string) *model.AppError {
switch newStatus {
case model.JobStatusPending:
return a.Srv().Jobs.SetJobPending(job)
case model.JobStatusCancelRequested:
return a.Srv().Jobs.RequestCancellation(c, job.Id)
case model.JobStatusCanceled:
return a.Srv().Jobs.SetJobCanceled(job)
default:
return model.NewAppError("UpdateJobStatus", "app.job.update_status.app_error", nil, "", http.StatusInternalServerError)
}
}
func (a *App) SessionHasPermissionToCreateJob(session model.Session, job *model.Job) (bool, *model.Permission) {
switch job.Type {
case model.JobTypeBlevePostIndexing:
@@ -92,6 +113,44 @@ func (a *App) SessionHasPermissionToCreateJob(session model.Session, job *model.
return false, nil
}
func (a *App) SessionHasPermissionToManageJob(session model.Session, job *model.Job) (bool, *model.Permission) {
var permission *model.Permission
switch job.Type {
case model.JobTypeBlevePostIndexing:
permission = model.PermissionManagePostBleveIndexesJob
case model.JobTypeDataRetention:
permission = model.PermissionManageDataRetentionJob
case model.JobTypeMessageExport:
permission = model.PermissionManageComplianceExportJob
case model.JobTypeElasticsearchPostIndexing:
permission = model.PermissionManageElasticsearchPostIndexingJob
case model.JobTypeElasticsearchPostAggregation:
permission = model.PermissionManageElasticsearchPostAggregationJob
case model.JobTypeLdapSync:
permission = model.PermissionManageLdapSyncJob
case
model.JobTypeMigrations,
model.JobTypePlugins,
model.JobTypeProductNotices,
model.JobTypeExpiryNotify,
model.JobTypeActiveUsers,
model.JobTypeImportProcess,
model.JobTypeImportDelete,
model.JobTypeExportProcess,
model.JobTypeExportDelete,
model.JobTypeCloud,
model.JobTypeExtractContent:
permission = model.PermissionManageJobs
}
if permission == nil {
return false, nil
}
return a.SessionHasPermissionTo(session, permission), permission
}
func (a *App) SessionHasPermissionToReadJob(session model.Session, jobType string) (bool, *model.Permission) {
switch jobType {
case model.JobTypeDataRetention:

61
server/channels/app/opentracing/opentracing_layer.go
Просмотреть файл

@@ -7308,6 +7308,28 @@ func (a *OpenTracingAppLayer) GetJobsByType(c request.CTX, jobType string, offse
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) GetJobsByTypeAndStatus(c request.CTX, jobTypes []string, status string, page int, perPage int) ([]*model.Job, *model.AppError) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetJobsByTypeAndStatus")
a.ctx = newCtx
a.app.Srv().Store().SetContext(newCtx)
defer func() {
a.app.Srv().Store().SetContext(origCtx)
a.ctx = origCtx
}()
defer span.Finish()
resultVar0, resultVar1 := a.app.GetJobsByTypeAndStatus(c, jobTypes, status, page, perPage)
if resultVar1 != nil {
span.LogFields(spanlog.Error(resultVar1))
ext.Error.Set(span, true)
}
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) GetJobsByTypePage(c request.CTX, jobType string, page int, perPage int) ([]*model.Job, *model.AppError) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetJobsByTypePage")
@@ -16316,6 +16338,23 @@ func (a *OpenTracingAppLayer) SessionHasPermissionToManageBot(rctx request.CTX,
return resultVar0
}
func (a *OpenTracingAppLayer) SessionHasPermissionToManageJob(session model.Session, job *model.Job) (bool, *model.Permission) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SessionHasPermissionToManageJob")
a.ctx = newCtx
a.app.Srv().Store().SetContext(newCtx)
defer func() {
a.app.Srv().Store().SetContext(origCtx)
a.ctx = origCtx
}()
defer span.Finish()
resultVar0, resultVar1 := a.app.SessionHasPermissionToManageJob(session, job)
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) SessionHasPermissionToReadJob(session model.Session, jobType string) (bool, *model.Permission) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SessionHasPermissionToReadJob")
@@ -18083,6 +18122,28 @@ func (a *OpenTracingAppLayer) UpdateIncomingWebhook(oldHook *model.IncomingWebho
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) UpdateJobStatus(c request.CTX, job *model.Job, newStatus string) *model.AppError {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateJobStatus")
a.ctx = newCtx
a.app.Srv().Store().SetContext(newCtx)
defer func() {
a.app.Srv().Store().SetContext(origCtx)
a.ctx = origCtx
}()
defer span.Finish()
resultVar0 := a.app.UpdateJobStatus(c, job, newStatus)
if resultVar0 != nil {
span.LogFields(spanlog.Error(resultVar0))
ext.Error.Set(span, true)
}
return resultVar0
}
func (a *OpenTracingAppLayer) UpdateMfa(c request.CTX, activate bool, userID string, token string) *model.AppError {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateMfa")

35
server/channels/app/permissions_migrations.go
Просмотреть файл

@@ -1181,6 +1181,40 @@ func (a *App) getAddChannelBookmarksPermissionsMigration() (permissionsMap, erro
return transformations, nil
}
func (a *App) getAddManageJobAncillaryPermissionsMigration() (permissionsMap, error) {
transformations := []permissionTransformation{}
transformations = append(transformations, permissionTransformation{
On: permissionExists(model.PermissionSysconsoleWriteAuthenticationLdap.Id),
Add: []string{model.PermissionManageLdapSyncJob.Id},
})
transformations = append(transformations, permissionTransformation{
On: permissionExists(model.PermissionSysconsoleWriteComplianceDataRetentionPolicy.Id),
Add: []string{model.PermissionManageDataRetentionJob.Id},
})
transformations = append(transformations, permissionTransformation{
On: permissionExists(model.PermissionSysconsoleWriteExperimentalBleve.Id),
Add: []string{model.PermissionManagePostBleveIndexesJob.Id},
})
transformations = append(transformations, permissionTransformation{
On: permissionExists(model.PermissionSysconsoleWriteComplianceComplianceExport.Id),
Add: []string{model.PermissionManageComplianceExportJob.Id},
})
transformations = append(transformations, permissionTransformation{
On: permissionExists(model.PermissionSysconsoleWriteEnvironmentElasticsearch.Id),
Add: []string{
model.PermissionManageElasticsearchPostIndexingJob.Id,
model.PermissionManageElasticsearchPostAggregationJob.Id,
},
})
return transformations, nil
}
// DoPermissionsMigrations execute all the permissions migrations need by the current version.
func (a *App) DoPermissionsMigrations() error {
return a.Srv().doPermissionsMigrations()
@@ -1228,6 +1262,7 @@ func (s *Server) doPermissionsMigrations() error {
{Key: model.MigrationKeyAddIPFilteringPermissions, Migration: a.getAddIPFilterPermissionsMigration},
{Key: model.MigrationKeyAddOutgoingOAuthConnectionsPermissions, Migration: a.getAddOutgoingOAuthConnectionsPermissions},
{Key: model.MigrationKeyAddChannelBookmarksPermissions, Migration: a.getAddChannelBookmarksPermissionsMigration},
{Key: model.MigrationKeyAddManageJobAncillaryPermissions, Migration: a.getAddManageJobAncillaryPermissionsMigration},
}
roles, err := s.Store().Role().GetAll()

18
server/channels/store/opentracinglayer/opentracinglayer.go
Просмотреть файл

@@ -5212,6 +5212,24 @@ func (s *OpenTracingLayerJobStore) GetAllByTypeAndStatus(c request.CTX, jobType
return result, err
}
func (s *OpenTracingLayerJobStore) GetAllByTypeAndStatusPage(c request.CTX, jobType []string, status string, offset int, limit int) ([]*model.Job, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "JobStore.GetAllByTypeAndStatusPage")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
result, err := s.JobStore.GetAllByTypeAndStatusPage(c, jobType, status, offset, limit)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return result, err
}
func (s *OpenTracingLayerJobStore) GetAllByTypePage(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "JobStore.GetAllByTypePage")

21
server/channels/store/retrylayer/retrylayer.go
Просмотреть файл

@@ -5903,6 +5903,27 @@ func (s *RetryLayerJobStore) GetAllByTypeAndStatus(c request.CTX, jobType string
}
func (s *RetryLayerJobStore) GetAllByTypeAndStatusPage(c request.CTX, jobType []string, status string, offset int, limit int) ([]*model.Job, error) {
tries := 0
for {
result, err := s.JobStore.GetAllByTypeAndStatusPage(c, jobType, status, offset, limit)
if err == nil {
return result, nil
}
if !isRepeatableError(err) {
return result, err
}
tries++
if tries >= 3 {
err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
return result, err
}
timepkg.Sleep(100 * timepkg.Millisecond)
}
}
func (s *RetryLayerJobStore) GetAllByTypePage(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, error) {
tries := 0

20
server/channels/store/sqlstore/job_store.go
Просмотреть файл

@@ -316,6 +316,26 @@ func (jss SqlJobStore) GetAllByStatus(c request.CTX, status string) ([]*model.Jo
return statuses, nil
}
func (jss SqlJobStore) GetAllByTypeAndStatusPage(c request.CTX, jobType []string, status string, offset int, limit int) ([]*model.Job, error) {
query, args, err := jss.getQueryBuilder().
Select("*").
From("Jobs").
Where(sq.Eq{"Type": jobType, "Status": status}).
OrderBy("CreateAt DESC").
Limit(uint64(limit)).
Offset(uint64(offset)).ToSql()
if err != nil {
return nil, errors.Wrap(err, "job_tosql")
}
jobs := []*model.Job{}
if err = jss.GetReplicaX().Select(&jobs, query, args...); err != nil {
return nil, errors.Wrapf(err, "failed to find Jobs with type=%s and status=%s", strings.Join(jobType, ","), status)
}
return jobs, nil
}
func (jss SqlJobStore) GetNewestJobByStatusAndType(status string, jobType string) (*model.Job, error) {
return jss.GetNewestJobByStatusesAndType([]string{status}, jobType)
}

1
server/channels/store/store.go
Просмотреть файл

@@ -761,6 +761,7 @@ type JobStore interface {
GetAllByTypePage(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, error)
GetAllByTypesPage(c request.CTX, jobTypes []string, offset int, limit int) ([]*model.Job, error)
GetAllByStatus(c request.CTX, status string) ([]*model.Job, error)
GetAllByTypeAndStatusPage(c request.CTX, jobType []string, status string, offset int, limit int) ([]*model.Job, error)
GetNewestJobByStatusAndType(status string, jobType string) (*model.Job, error)
GetNewestJobByStatusesAndType(statuses []string, jobType string) (*model.Job, error)
GetCountByStatusAndType(status string, jobType string) (int64, error)

57
server/channels/store/storetest/job_store.go
Просмотреть файл

@@ -24,6 +24,7 @@ func TestJobStore(t *testing.T, rctx request.CTX, ss store.Store) {
t.Run("JobGetAllByTypeAndStatus", func(t *testing.T) { testJobGetAllByTypeAndStatus(t, rctx, ss) })
t.Run("JobGetAllByTypePage", func(t *testing.T) { testJobGetAllByTypePage(t, rctx, ss) })
t.Run("JobGetAllByTypesPage", func(t *testing.T) { testJobGetAllByTypesPage(t, rctx, ss) })
t.Run("JobGetAllByTypeAndStatusPage", func(t *testing.T) { testJobGetAllByTypeAndStatusPage(t, rctx, ss) })
t.Run("JobGetAllByStatus", func(t *testing.T) { testJobGetAllByStatus(t, rctx, ss) })
t.Run("GetNewestJobByStatusAndType", func(t *testing.T) { testJobStoreGetNewestJobByStatusAndType(t, rctx, ss) })
t.Run("GetNewestJobByStatusesAndType", func(t *testing.T) { testJobStoreGetNewestJobByStatusesAndType(t, rctx, ss) })
@@ -259,6 +260,62 @@ func testJobGetAllByTypesPage(t *testing.T, rctx request.CTX, ss store.Store) {
require.Equal(t, received[0].Id, jobs[1].Id, "should've received oldest job last")
}
func testJobGetAllByTypeAndStatusPage(t *testing.T, rctx request.CTX, ss store.Store) {
jobType := model.NewId()
jobType2 := model.NewId()
t0 := model.GetMillis()
jobs := []*model.Job{
{
Id: model.NewId(),
Type: jobType,
Status: model.JobStatusPending,
CreateAt: t0,
},
{
Id: model.NewId(),
Type: jobType,
Status: model.JobStatusPending,
CreateAt: t0 + 1,
},
{
Id: model.NewId(),
Type: jobType2,
Status: model.JobStatusCanceled,
CreateAt: t0 + 2,
},
{
Id: model.NewId(),
Type: jobType2,
Status: model.JobStatusCanceled,
CreateAt: t0 + 3,
},
}
for _, job := range jobs {
_, err := ss.Job().Save(job)
require.NoError(t, err)
defer ss.Job().Delete(job.Id)
}
jobTypes := []string{jobType, jobType2}
received, err := ss.Job().GetAllByTypeAndStatusPage(rctx, jobTypes, model.JobStatusPending, 0, 4)
require.NoError(t, err)
require.Len(t, received, 2)
require.Equal(t, received[0].Id, jobs[1].Id, "should've received newest job first")
require.Equal(t, received[1].Id, jobs[0].Id, "should've received oldest job last")
received, err = ss.Job().GetAllByTypeAndStatusPage(rctx, jobTypes, model.JobStatusPending, 1, 1)
require.NoError(t, err)
require.Len(t, received, 1)
require.Equal(t, received[0].Id, jobs[0].Id, "should've received the oldest pending job")
received, err = ss.Job().GetAllByTypeAndStatusPage(rctx, []string{jobType2}, model.JobStatusCanceled, 1, 1)
require.NoError(t, err)
require.Len(t, received, 1)
require.Equal(t, received[0].Id, jobs[2].Id, "should've received the oldest canceled job")
}
func testJobGetAllByStatus(t *testing.T, rctx request.CTX, ss store.Store) {
jobType := model.NewId()
status := model.NewId()

30
server/channels/store/storetest/mocks/JobStore.go
Просмотреть файл

@@ -181,6 +181,36 @@ func (_m *JobStore) GetAllByTypeAndStatus(c request.CTX, jobType string, status
return r0, r1
}
// GetAllByTypeAndStatusPage provides a mock function with given fields: c, jobType, status, offset, limit
func (_m *JobStore) GetAllByTypeAndStatusPage(c request.CTX, jobType []string, status string, offset int, limit int) ([]*model.Job, error) {
ret := _m.Called(c, jobType, status, offset, limit)
if len(ret) == 0 {
panic("no return value specified for GetAllByTypeAndStatusPage")
}
var r0 []*model.Job
var r1 error
if rf, ok := ret.Get(0).(func(request.CTX, []string, string, int, int) ([]*model.Job, error)); ok {
return rf(c, jobType, status, offset, limit)
}
if rf, ok := ret.Get(0).(func(request.CTX, []string, string, int, int) []*model.Job); ok {
r0 = rf(c, jobType, status, offset, limit)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).([]*model.Job)
}
}
if rf, ok := ret.Get(1).(func(request.CTX, []string, string, int, int) error); ok {
r1 = rf(c, jobType, status, offset, limit)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// GetAllByTypePage provides a mock function with given fields: c, jobType, offset, limit
func (_m *JobStore) GetAllByTypePage(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, error) {
ret := _m.Called(c, jobType, offset, limit)

16
server/channels/store/timerlayer/timerlayer.go
Просмотреть файл

@@ -4731,6 +4731,22 @@ func (s *TimerLayerJobStore) GetAllByTypeAndStatus(c request.CTX, jobType string
return result, err
}
func (s *TimerLayerJobStore) GetAllByTypeAndStatusPage(c request.CTX, jobType []string, status string, offset int, limit int) ([]*model.Job, error) {
start := time.Now()
result, err := s.JobStore.GetAllByTypeAndStatusPage(c, jobType, status, offset, limit)
elapsed := float64(time.Since(start)) / float64(time.Second)
if s.Root.Metrics != nil {
success := "false"
if err == nil {
success = "true"
}
s.Root.Metrics.ObserveStoreMethodDuration("JobStore.GetAllByTypeAndStatusPage", success, elapsed)
}
return result, err
}
func (s *TimerLayerJobStore) GetAllByTypePage(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, error) {
start := time.Now()

1
server/channels/testlib/store.go
Просмотреть файл

@@ -76,6 +76,7 @@ func GetMockStoreForSetupFunctions() *mocks.Store {
systemStore.On("GetByName", model.MigrationKeyAddIPFilteringPermissions).Return(&model.System{Name: model.MigrationKeyAddIPFilteringPermissions, Value: "true"}, nil)
systemStore.On("GetByName", model.MigrationKeyAddOutgoingOAuthConnectionsPermissions).Return(&model.System{Name: model.MigrationKeyAddOutgoingOAuthConnectionsPermissions, Value: "true"}, nil)
systemStore.On("GetByName", model.MigrationKeyAddChannelBookmarksPermissions).Return(&model.System{Name: model.MigrationKeyAddChannelBookmarksPermissions, Value: "true"}, nil)
systemStore.On("GetByName", model.MigrationKeyAddManageJobAncillaryPermissions).Return(&model.System{Name: model.MigrationKeyAddManageJobAncillaryPermissions, Value: "true"}, nil)
systemStore.On("GetByName", "CustomGroupAdminRoleCreationMigrationComplete").Return(&model.System{Name: model.MigrationKeyAddPlayboosksManageRolesPermissions, Value: "true"}, nil)
systemStore.On("GetByName", "products_boards").Return(&model.System{Name: "products_boards", Value: "true"}, nil)
systemStore.On("GetByName", "elasticsearch_fix_channel_index_migration").Return(&model.System{Name: "elasticsearch_fix_channel_index_migration", Value: "true"}, nil)