diff --git a/server/channels/api4/job.go b/server/channels/api4/job.go index 03afdb2071..b91bd42de0 100644 --- a/server/channels/api4/job.go +++ b/server/channels/api4/job.go @@ -234,7 +234,7 @@ func getJobs(c *Context, w http.ResponseWriter, r *http.Request) { 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) + jobs, appErr = c.App.GetJobsByTypesAndStatuses(c.AppContext, validJobTypes, []string{status}, c.Params.Page, c.Params.PerPage) } if appErr != nil { diff --git a/server/channels/app/job.go b/server/channels/app/job.go index 8d9a5cd4bf..6e62a040ce 100644 --- a/server/channels/app/job.go +++ b/server/channels/app/job.go @@ -43,10 +43,10 @@ func (a *App) GetJobsByTypesPage(c request.CTX, jobType []string, page int, perP 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) +func (a *App) GetJobsByTypesAndStatuses(c request.CTX, jobTypes []string, status []string, page int, perPage int) ([]*model.Job, *model.AppError) { + jobs, err := a.Srv().Store().Job().GetAllByTypesAndStatusesPage(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 nil, model.NewAppError("GetAllByTypesAndStatusesPage", "app.job.get_all.app_error", nil, "", http.StatusInternalServerError).Wrap(err) } return jobs, nil } diff --git a/server/channels/store/retrylayer/retrylayer.go b/server/channels/store/retrylayer/retrylayer.go index c741804a33..ef5ebaeac8 100644 --- a/server/channels/store/retrylayer/retrylayer.go +++ b/server/channels/store/retrylayer/retrylayer.go @@ -6404,11 +6404,11 @@ 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) { +func (s *RetryLayerJobStore) GetAllByTypePage(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, error) { tries := 0 for { - result, err := s.JobStore.GetAllByTypeAndStatusPage(c, jobType, status, offset, limit) + result, err := s.JobStore.GetAllByTypePage(c, jobType, offset, limit) if err == nil { return result, nil } @@ -6425,11 +6425,11 @@ func (s *RetryLayerJobStore) GetAllByTypeAndStatusPage(c request.CTX, jobType [] } -func (s *RetryLayerJobStore) GetAllByTypePage(c request.CTX, jobType string, offset int, limit int) ([]*model.Job, error) { +func (s *RetryLayerJobStore) GetAllByTypesAndStatusesPage(c request.CTX, jobType []string, status []string, offset int, limit int) ([]*model.Job, error) { tries := 0 for { - result, err := s.JobStore.GetAllByTypePage(c, jobType, offset, limit) + result, err := s.JobStore.GetAllByTypesAndStatusesPage(c, jobType, status, offset, limit) if err == nil { return result, nil } diff --git a/server/channels/store/sqlstore/job_store.go b/server/channels/store/sqlstore/job_store.go index 266e3836dd..187950f60c 100644 --- a/server/channels/store/sqlstore/job_store.go +++ b/server/channels/store/sqlstore/job_store.go @@ -378,12 +378,11 @@ 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, page int, perPage int) ([]*model.Job, error) { - offset := page * perPage +func (jss SqlJobStore) GetAllByTypesAndStatusesPage(c request.CTX, jobType []string, status []string, offset int, limit int) ([]*model.Job, error) { query, args, err := jss.jobQuery. Where(sq.Eq{"Type": jobType, "Status": status}). OrderBy("CreateAt DESC"). - Limit(uint64(perPage)). + Limit(uint64(limit)). Offset(uint64(offset)).ToSql() if err != nil { return nil, errors.Wrap(err, "job_tosql") @@ -391,7 +390,7 @@ func (jss SqlJobStore) GetAllByTypeAndStatusPage(c request.CTX, jobType []string jobs := []*model.Job{} if err = jss.GetReplica().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 nil, errors.Wrapf(err, "failed to find Jobs with types=%s and statuses=%s", strings.Join(jobType, ","), strings.Join(status, ",")) } return jobs, nil diff --git a/server/channels/store/store.go b/server/channels/store/store.go index 7047de907b..8fdf20d54a 100644 --- a/server/channels/store/store.go +++ b/server/channels/store/store.go @@ -796,7 +796,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) + GetAllByTypesAndStatusesPage(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) diff --git a/server/channels/store/storetest/job_store.go b/server/channels/store/storetest/job_store.go index d383b6a3af..52c82dfbe8 100644 --- a/server/channels/store/storetest/job_store.go +++ b/server/channels/store/storetest/job_store.go @@ -25,6 +25,7 @@ func TestJobStore(t *testing.T, rctx request.CTX, ss store.Store) { 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("JobGetAllByTypesAndStatusesPage", func(t *testing.T) { testJobGetAllByTypesAndStatusesPage(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) }) @@ -299,23 +300,133 @@ func testJobGetAllByTypeAndStatusPage(t *testing.T, rctx request.CTX, ss store.S } jobTypes := []string{jobType, jobType2} - received, err := ss.Job().GetAllByTypeAndStatusPage(rctx, jobTypes, model.JobStatusPending, 0, 4) + received, err := ss.Job().GetAllByTypesAndStatusesPage(rctx, jobTypes, []string{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) + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, jobTypes, []string{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) + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, []string{jobType2}, []string{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 testJobGetAllByTypesAndStatusesPage(t *testing.T, rctx request.CTX, ss store.Store) { + jobType1 := model.NewId() + jobType2 := model.NewId() + jobType3 := model.NewId() + status1 := model.JobStatusPending + status2 := model.JobStatusInProgress + status3 := model.JobStatusSuccess + t0 := model.GetMillis() + + jobs := []*model.Job{ + { + Id: model.NewId(), // 0: type1, status1, t0 + Type: jobType1, + Status: status1, + CreateAt: t0, + }, + { + Id: model.NewId(), // 1: type1, status2, t0+1 + Type: jobType1, + Status: status2, + CreateAt: t0 + 1, + }, + { + Id: model.NewId(), // 2: type2, status1, t0+2 + Type: jobType2, + Status: status1, + CreateAt: t0 + 2, + }, + { + Id: model.NewId(), // 3: type2, status2, t0+3 + Type: jobType2, + Status: status2, + CreateAt: t0 + 3, + }, + { + Id: model.NewId(), // 4: type1, status3, t0+4 + Type: jobType1, + Status: status3, + CreateAt: t0 + 4, + }, + { + Id: model.NewId(), // 5: type3, status1, t0+5 + Type: jobType3, + Status: status1, + CreateAt: t0 + 5, + }, + } + + for _, job := range jobs { + _, err := ss.Job().Save(job) + require.NoError(t, err) + defer ss.Job().Delete(job.Id) + } + + // Test case 1: Get jobs of type1 or type2 with status1 or status2, limit 4, offset 0 + types1 := []string{jobType1, jobType2} + statuses1 := []string{status1, status2} + received, err := ss.Job().GetAllByTypesAndStatusesPage(rctx, types1, statuses1, 0, 4) + require.NoError(t, err) + require.Len(t, received, 4) + require.Equal(t, jobs[3].Id, received[0].Id, "case 1: newest job type2/status2") + require.Equal(t, jobs[2].Id, received[1].Id, "case 1: second newest job type2/status1") + require.Equal(t, jobs[1].Id, received[2].Id, "case 1: third newest job type1/status2") + require.Equal(t, jobs[0].Id, received[3].Id, "case 1: oldest job type1/status1") + + // Test case 2: Get jobs of type1 or type2 with status1 or status2, limit 2, offset 2 + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, types1, statuses1, 2, 2) + require.NoError(t, err) + require.Len(t, received, 2) + require.Equal(t, jobs[1].Id, received[0].Id, "case 2: third newest job type1/status2") + require.Equal(t, jobs[0].Id, received[1].Id, "case 2: oldest job type1/status1") + + // Test case 3: Get jobs of type1 with status1 or status3, limit 5, offset 0 + types2 := []string{jobType1} + statuses2 := []string{status1, status3} + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, types2, statuses2, 0, 5) + require.NoError(t, err) + require.Len(t, received, 2) + require.Equal(t, jobs[4].Id, received[0].Id, "case 3: newest job type1/status3") + require.Equal(t, jobs[0].Id, received[1].Id, "case 3: oldest job type1/status1") + + // Test case 4: Get jobs of type3 with status1, limit 1, offset 0 + types3 := []string{jobType3} + statuses3 := []string{status1} + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, types3, statuses3, 0, 1) + require.NoError(t, err) + require.Len(t, received, 1) + require.Equal(t, jobs[5].Id, received[0].Id, "case 4: only job type3/status1") + + // Test case 5: Get jobs with non-existent type + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, []string{model.NewId()}, statuses1, 0, 5) + require.NoError(t, err) + require.Len(t, received, 0, "case 5: no jobs with non-existent type") + + // Test case 6: Get jobs with non-existent status + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, types1, []string{model.NewId()}, 0, 5) + require.NoError(t, err) + require.Len(t, received, 0, "case 6: no jobs with non-existent status") + + // Test case 7: Empty types slice + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, []string{}, statuses1, 0, 5) + require.NoError(t, err) + require.Len(t, received, 0, "case 7: empty types slice should return no jobs") + + // Test case 8: Empty statuses slice + received, err = ss.Job().GetAllByTypesAndStatusesPage(rctx, types1, []string{}, 0, 5) + require.NoError(t, err) + require.Len(t, received, 0, "case 8: empty statuses slice should return no jobs") +} + func testJobGetAllByStatus(t *testing.T, rctx request.CTX, ss store.Store) { jobType := model.NewId() status := model.NewId() diff --git a/server/channels/store/storetest/mocks/JobStore.go b/server/channels/store/storetest/mocks/JobStore.go index 2d963d852b..c68175b3b4 100644 --- a/server/channels/store/storetest/mocks/JobStore.go +++ b/server/channels/store/storetest/mocks/JobStore.go @@ -181,36 +181,6 @@ 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) @@ -241,6 +211,36 @@ func (_m *JobStore) GetAllByTypePage(c request.CTX, jobType string, offset int, return r0, r1 } +// GetAllByTypesAndStatusesPage provides a mock function with given fields: c, jobType, status, offset, limit +func (_m *JobStore) GetAllByTypesAndStatusesPage(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 GetAllByTypesAndStatusesPage") + } + + 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 +} + // GetAllByTypesPage provides a mock function with given fields: c, jobTypes, offset, limit func (_m *JobStore) GetAllByTypesPage(c request.CTX, jobTypes []string, offset int, limit int) ([]*model.Job, error) { ret := _m.Called(c, jobTypes, offset, limit) diff --git a/server/channels/store/timerlayer/timerlayer.go b/server/channels/store/timerlayer/timerlayer.go index 76b136a069..608138722c 100644 --- a/server/channels/store/timerlayer/timerlayer.go +++ b/server/channels/store/timerlayer/timerlayer.go @@ -5127,22 +5127,6 @@ 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() @@ -5159,6 +5143,22 @@ func (s *TimerLayerJobStore) GetAllByTypePage(c request.CTX, jobType string, off return result, err } +func (s *TimerLayerJobStore) GetAllByTypesAndStatusesPage(c request.CTX, jobType []string, status []string, offset int, limit int) ([]*model.Job, error) { + start := time.Now() + + result, err := s.JobStore.GetAllByTypesAndStatusesPage(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.GetAllByTypesAndStatusesPage", success, elapsed) + } + return result, err +} + func (s *TimerLayerJobStore) GetAllByTypesPage(c request.CTX, jobTypes []string, offset int, limit int) ([]*model.Job, error) { start := time.Now() diff --git a/server/cmd/mmctl/commands/compliance_export.go b/server/cmd/mmctl/commands/compliance_export.go index 563c66b689..614dfef8d4 100644 --- a/server/cmd/mmctl/commands/compliance_export.go +++ b/server/cmd/mmctl/commands/compliance_export.go @@ -5,11 +5,17 @@ package commands import ( "context" + "errors" "fmt" "os" + "path" + "strconv" + "time" + "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/v8/cmd/mmctl/client" "github.com/mattermost/mattermost/server/v8/cmd/mmctl/printer" + "github.com/mattermost/mattermost/server/v8/enterprise/message_export/shared" "github.com/spf13/cobra" ) @@ -44,12 +50,22 @@ var ComplianceExportCancelCmd = &cobra.Command{ var ComplianceExportDownloadCmd = &cobra.Command{ Use: "download [complianceExportJobID] [output filepath (optional)]", - Example: " compliance_export download o98rj3ur83dp5dppfyk5yk6osy", + Example: "compliance-export download o98rj3ur83dp5dppfyk5yk6osy", Short: "Download compliance export file", Args: cobra.MinimumNArgs(1), RunE: withClient(complianceExportDownloadCmdF), } +var ComplianceExportCreateCmd = &cobra.Command{ + Use: "create [complianceExportType] --date \"2025-03-27 -0400\"", + Example: "compliance-export create csv --date \"2025-03-27 -0400\"", + Long: "Create a compliance export job, of type 'csv' or 'actiance' or 'globalrelay'. If --date is set, the job will run for one day, from 12am to 12am (minus one millisecond) inclusively, in the format with timezone offset: `\"YYYY-MM-DD -0000\"`. E.g., \"2024-10-21 -0400\" for Oct 21, 2024 EDT timezone. \"2023-11-01 +0000\" for Nov 01, 2024 UTC. If set, the 'start' and 'end' flags will be ignored.\n\n" + + "Important: Running a compliance export job from mmctl will NOT affect the next scheduled job's batch_start_time. This means that if you run a compliance export job from mmctl, the next scheduled job will run from the batch_end_time of the previous scheduled job, as usual.", + Short: "Create a compliance export job, of type 'csv' or 'actiance' or 'globalrelay'", + Args: cobra.MinimumNArgs(1), + RunE: withClient(complianceExportCreateCmdF), +} + func init() { ComplianceExportListCmd.Flags().Int("page", 0, "Page number to fetch for the list of compliance export jobs") ComplianceExportListCmd.Flags().Int("per-page", DefaultPageSize, "Number of compliance export jobs to be fetched") @@ -57,11 +73,28 @@ func init() { ComplianceExportDownloadCmd.Flags().Int("num-retries", 5, "Number of retries if the download fails") + ComplianceExportCreateCmd.Flags().String( + "date", + "", + "Run the export for one day, from 12am to 12am (minus one millisecond) inclusively, in the format with timezone offset: `\"YYYY-MM-DD -0000\"`. E.g., `\"2024-10-21 -0400\"` for Oct 21, 2024 EDT timezone. `\"2023-11-01 +0000\"` for Nov 01, 2024 UTC. If set, the 'start' and 'end' flags will be ignored.", + ) + ComplianceExportCreateCmd.Flags().Int( + "start", + 0, + "The start timestamp in unix milliseconds. Posts with updateAt >= start will be exported. If set, 'end' must be set as well. eg, `1743048000000` for 2025-03-27 EDT.", + ) + ComplianceExportCreateCmd.Flags().Int( + "end", + 0, + "The end timestamp in unix milliseconds. Posts with updateAt <= end will be exported. If set, 'start' must be set as well. eg, `1743134400000` for 2025-03-28 EDT.", + ) + ComplianceExportCmd.AddCommand( ComplianceExportListCmd, ComplianceExportShowCmd, ComplianceExportCancelCmd, ComplianceExportDownloadCmd, + ComplianceExportCreateCmd, ) RootCmd.AddCommand(ComplianceExportCmd) } @@ -126,3 +159,85 @@ func complianceExportDownloadCmdF(c client.Client, command *cobra.Command, args printer.Print(fmt.Sprintf("Compliance export file downloaded to %q", path)) return nil } + +func complianceExportCreateCmdF(c client.Client, command *cobra.Command, args []string) error { + exportType := args[0] + if exportType != model.ComplianceExportTypeActiance && + exportType != model.ComplianceExportTypeCsv && + exportType != model.ComplianceExportTypeGlobalrelay { + return fmt.Errorf("invalid export type: %s, must be one of: csv, actiance, globalrelay", exportType) + } + + dateStr, err := command.Flags().GetString("date") + if err != nil { + return err + } + start, err := command.Flags().GetInt("start") + if err != nil { + return err + } + end, err := command.Flags().GetInt("end") + if err != nil { + return err + } + startTimestamp, endTimestamp, err := getStartAndEnd(dateStr, start, end) + if err != nil { + return err + } + startTime := strconv.FormatInt(startTimestamp, 10) + endTime := strconv.FormatInt(endTimestamp, 10) + exportDir := path.Join(model.ComplianceExportPath, fmt.Sprintf("%s-%s-%s", time.Now().Format(model.ComplianceExportDirectoryFormat), startTime, endTime)) + + // If start and end are 0, we need to not set those keys in the job data. + // This will make the job like a manual job (it will pick up where the previous job left off). + data := model.StringMap{ + shared.JobDataInitiatedBy: "mmctl", + shared.JobDataExportType: exportType, + shared.JobDataBatchStartId: "", + shared.JobDataJobStartId: "", + } + if startTimestamp != 0 && endTimestamp != 0 { + data[shared.JobDataBatchStartTime] = startTime + data[shared.JobDataJobStartTime] = startTime + data[shared.JobDataJobEndTime] = endTime + data[shared.JobDataExportDir] = exportDir + } + + job := &model.Job{ + Type: model.JobTypeMessageExport, + Data: data, + } + + if job, _, err = c.CreateJob(context.TODO(), job); err != nil { + return fmt.Errorf("failed to create compliance export job: %w", err) + } + + printer.Print(fmt.Sprintf("Compliance export job created with ID: %s", job.Id)) + + return nil +} + +// getStartAndEnd returns the start and end timestamps in unix milliseconds +func getStartAndEnd(dateStr string, start int, end int) (int64, int64, error) { + if dateStr == "" && start == 0 && end == 0 { + // return 0 so that the job will be like a manual job + return 0, 0, nil + } + + if dateStr != "" && (start > 0 || end > 0) { + return 0, 0, errors.New("if date is used, start and end must not be set") + } + + if dateStr != "" { + t, err := time.Parse("2006-01-02 -0700", dateStr) + if err != nil { + return 0, 0, fmt.Errorf("could not parse date string: %s, use the format with time zone offset: YYYY-MM-DD -0700, eg for EDT: `2024-12-24 -0400`, error details: %w", dateStr, err) + } + endTimestamp := t.AddDate(0, 0, 1).UnixMilli() - 1 + return t.UnixMilli(), endTimestamp, nil + } + if start <= 0 || end <= 0 || start >= end { + return 0, 0, fmt.Errorf("if date is not used, start: %d and end: %d must both be > 0, and start must be < end", start, end) + } + return int64(start), int64(end), nil +} diff --git a/server/cmd/mmctl/commands/compliance_export_e2e_test.go b/server/cmd/mmctl/commands/compliance_export_e2e_test.go index 6df88d7a9b..9252572c2a 100644 --- a/server/cmd/mmctl/commands/compliance_export_e2e_test.go +++ b/server/cmd/mmctl/commands/compliance_export_e2e_test.go @@ -10,12 +10,14 @@ import ( "io" "os" "path/filepath" + "strconv" "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/v8" st "github.com/mattermost/mattermost/server/v8/channels/store/storetest" "github.com/mattermost/mattermost/server/v8/cmd/mmctl/client" "github.com/mattermost/mattermost/server/v8/cmd/mmctl/printer" + "github.com/mattermost/mattermost/server/v8/enterprise/message_export/shared" "github.com/spf13/cobra" ) @@ -232,7 +234,7 @@ func (s *MmctlE2ETestSuite) TestComplianceExportCancelCmdE2E() { cmd := makeCmd() err = complianceExportCancelCmdF(s.th.Client, cmd, []string{job.Id}) - s.Require().EqualError(err, "failed to get compliance export job: You do not have the appropriate permissions.") + s.Require().EqualError(err, "failed to cancel compliance export job: You do not have the appropriate permissions.") s.Require().Empty(printer.GetLines()) s.Require().Empty(printer.GetErrorLines()) }) @@ -242,7 +244,7 @@ func (s *MmctlE2ETestSuite) TestComplianceExportCancelCmdE2E() { cmd := makeCmd() err := complianceExportCancelCmdF(c, cmd, []string{"non-existent-job-id"}) - s.Require().EqualError(err, "failed to get compliance export job: Sorry, we could not find the page., There doesn't appear to be an api call for the url='/api/v4/jobs/non-existent-job-id'. Typo? are you missing a team_id or user_id as part of the url?") + s.Require().EqualError(err, "failed to cancel compliance export job: Sorry, we could not find the page., There doesn't appear to be an api call for the url='/api/v4/jobs/non-existent-job-id/cancel'. Typo? are you missing a team_id or user_id as part of the url?") s.Require().Empty(printer.GetLines()) s.Require().Empty(printer.GetErrorLines()) }) @@ -585,3 +587,147 @@ func (s *MmctlE2ETestSuite) TestComplianceExportDownloadCmdE2E() { s.Require().True(foundExport2, "export2.zip not found in downloaded file") }) } + +func (s *MmctlE2ETestSuite) TestComplianceExportMmctlJobStartTimeE2E() { + s.SetupMessageExportTestHelper() + + s.RunForSystemAdminAndLocal("mmctl job uses batch_start_time from previous regular job", func(c client.Client) { + // Ensure no jobs exist before we start + jobs, _, err := s.th.SystemAdminClient.GetJobsByType(context.Background(), model.JobTypeMessageExport, 0, 1000) + s.Require().NoError(err) + for _, job := range jobs { + var result string + result, err = s.th.App.Srv().Store().Job().Delete(job.Id) + s.Require().NoError(err, "Failed to delete job (result: %v)", result) + } + + now := model.GetMillis() + + // Create a regular (non-mmctl) export job + regularStartTime := now - 10000 + regularEndTime := now - 5000 + regularJob := s.runJobForTest(map[string]string{ + shared.JobDataBatchStartTime: strconv.FormatInt(regularStartTime, 10), + shared.JobDataJobEndTime: strconv.FormatInt(regularEndTime, 10), + }) + + s.Require().Equal(model.JobStatusSuccess, regularJob.Status, "Regular job should complete successfully") + s.Require().NotEmpty(regularJob.Data[shared.JobDataBatchStartTime], "Regular job should have a batch start time") + regularJobBatchStartTime := regularJob.Data[shared.JobDataBatchStartTime] + + // Run an mmctl-initiated export job + cmd := &cobra.Command{} + cmd.Flags().String("date", "", "") + cmd.Flags().Int("start", 0, "") + cmd.Flags().Int("end", 0, "") + err = complianceExportCreateCmdF(c, cmd, []string{model.ComplianceExportTypeActiance}) + s.Require().NoError(err, "Should create mmctl job successfully") + + // Find the mmctl job + jobs, _, err = s.th.SystemAdminClient.GetJobsByType(context.Background(), model.JobTypeMessageExport, 0, 10) + s.Require().NoError(err) + s.Require().True(len(jobs) > 1, "Should have at least 2 jobs") + + // The most recent job should be the mmctl job + mmctlJob := jobs[0] + s.Require().Equal("mmctl", mmctlJob.Data[shared.JobDataInitiatedBy]) + + // Wait for the mmctl job to complete + s.checkJobForStatus(mmctlJob.Id, model.JobStatusSuccess) + mmctlJob = s.getMostRecentJobWithId(mmctlJob.Id) + + // The job_start_time should match the batch_start_time from the previous regular job + s.Require().Equal(regularJobBatchStartTime, mmctlJob.Data[shared.JobDataJobStartTime], + "mmctl job should use batch_start_time from previous regular job as its job_start_time") + + // Clean up jobs + for _, job := range jobs { + result, err := s.th.App.Srv().Store().Job().Delete(job.Id) + s.Require().NoError(err, "Failed to delete job (result: %v)", result) + } + }) + + s.RunForSystemAdminAndLocal("mmctl job ignores previous mmctl jobs and uses regular job", func(c client.Client) { + // Ensure no jobs exist before we start + jobs, _, err := s.th.SystemAdminClient.GetJobsByType(context.Background(), model.JobTypeMessageExport, 0, 1000) + s.Require().NoError(err) + for _, job := range jobs { + var result string + result, err = s.th.App.Srv().Store().Job().Delete(job.Id) + s.Require().NoError(err, "Failed to delete job (result: %v)", result) + } + + now := model.GetMillis() + + // Create a regular (non-mmctl) export job + regularStartTime := now - 10000 + regularEndTime := now - 5000 + regularJob := s.runJobForTest(map[string]string{ + shared.JobDataBatchStartTime: strconv.FormatInt(regularStartTime, 10), + shared.JobDataJobEndTime: strconv.FormatInt(regularEndTime, 10), + }) + + s.Require().Equal(model.JobStatusSuccess, regularJob.Status, "Regular job should complete successfully") + s.Require().NotEmpty(regularJob.Data[shared.JobDataBatchStartTime], "Regular job should have a batch start time") + regularJobBatchStartTime := regularJob.Data[shared.JobDataBatchStartTime] + + // Run an mmctl-initiated export job with an explicit start time (different from the regular job) + cmd := &cobra.Command{} + cmd.Flags().String("date", "", "") + cmd.Flags().Int("start", int(now-2000), "") + cmd.Flags().Int("end", int(now-1000), "") + err = complianceExportCreateCmdF(c, cmd, []string{model.ComplianceExportTypeActiance}) + s.Require().NoError(err, "Should create first mmctl job successfully") + + // Find the mmctl job + jobs, _, err = s.th.SystemAdminClient.GetJobsByType(context.Background(), model.JobTypeMessageExport, 0, 10) + s.Require().NoError(err) + s.Require().True(len(jobs) > 1, "Should have at least 2 jobs") + + // The most recent job should be the mmctl job + mmctlJob1 := jobs[0] + s.Require().Equal("mmctl", mmctlJob1.Data[shared.JobDataInitiatedBy]) + + // Wait for the mmctl job to complete + s.checkJobForStatus(mmctlJob1.Id, model.JobStatusSuccess) + mmctlJob1 = s.getMostRecentJobWithId(mmctlJob1.Id) + + // Verify this job has a different batch_start_time than the regular job + s.Require().NotEqual(regularJobBatchStartTime, mmctlJob1.Data[shared.JobDataBatchStartTime], + "First mmctl job should have a different batch_start_time than regular job") + + // Run a second mmctl-initiated export job WITHOUT a specified start time + cmd = &cobra.Command{} + cmd.Flags().String("date", "", "") + cmd.Flags().Int("start", 0, "") + cmd.Flags().Int("end", 0, "") + err = complianceExportCreateCmdF(c, cmd, []string{model.ComplianceExportTypeActiance}) + s.Require().NoError(err, "Should create second mmctl job successfully") + + // Find the second mmctl job + jobs, _, err = s.th.SystemAdminClient.GetJobsByType(context.Background(), model.JobTypeMessageExport, 0, 10) + s.Require().NoError(err) + s.Require().True(len(jobs) > 2, "Should have at least 3 jobs") + + // The most recent job should be the second mmctl job + mmctlJob2 := jobs[0] + s.Require().Equal("mmctl", mmctlJob2.Data[shared.JobDataInitiatedBy]) + + // Wait for the second mmctl job to complete + s.checkJobForStatus(mmctlJob2.Id, model.JobStatusSuccess) + mmctlJob2 = s.getMostRecentJobWithId(mmctlJob2.Id) + + // The job_start_time of the second mmctl job should match the batch_start_time from the regular job, + // not from the mmctl job that ran in between + s.Require().Equal(regularJobBatchStartTime, mmctlJob2.Data[shared.JobDataJobStartTime], + "Second mmctl job should use batch_start_time from previous regular job as its job_start_time, not from previous mmctl job") + s.Require().NotEqual(mmctlJob1.Data[shared.JobDataBatchStartTime], mmctlJob2.Data[shared.JobDataJobStartTime], + "Second mmctl job should not use batch_start_time from previous mmctl job as its job_start_time") + + // Clean up jobs + for _, job := range jobs { + result, err := s.th.App.Srv().Store().Job().Delete(job.Id) + s.Require().NoError(err, "Failed to delete job (result: %v)", result) + } + }) +} diff --git a/server/cmd/mmctl/commands/compliance_export_test.go b/server/cmd/mmctl/commands/compliance_export_test.go index 9b135439df..21dc2af2c0 100644 --- a/server/cmd/mmctl/commands/compliance_export_test.go +++ b/server/cmd/mmctl/commands/compliance_export_test.go @@ -7,6 +7,7 @@ import ( "context" "fmt" "os" + "testing" gomock "github.com/golang/mock/gomock" "github.com/mattermost/mattermost/server/public/model" @@ -298,6 +299,121 @@ func (s *MmctlUnitTestSuite) TestComplianceExportDownloadCmdF() { }) } +func TestGetStartAndEnd(t *testing.T) { + type args struct { + dateStr string + start int + end int + } + tests := []struct { + name string + args args + expectedStart int64 + expectedEnd int64 + wantErr bool + }{ + // check with: https://www.epochconverter.com/ + { + name: "parse a date in EDT (-0400)", + args: args{ + dateStr: "2024-10-21 -0400", + }, + expectedStart: 1729483200000, + expectedEnd: 1729569599999, + }, + { + name: "parse a date in UTC (+0)", + args: args{ + dateStr: "2024-10-21 +0000", + }, + expectedStart: 1729468800000, + expectedEnd: 1729555199999, + }, + { + name: "parse a date in CDT (-0500)", + args: args{ + dateStr: "2024-10-21 -0500", + }, + expectedStart: 1729486800000, + expectedEnd: 1729573199999, + }, + { + name: "bad format", + args: args{ + dateStr: "2024-10-21 CT", + }, + wantErr: true, + }, + { + name: "bad format", + args: args{ + dateStr: "2024-1-2 CDT", + }, + wantErr: true, + }, + { + name: "it's ok to not have date, start, or end", + args: args{}, + wantErr: false, + }, + { + name: "needs both start and end pt1", + args: args{ + start: 12345, + }, + wantErr: true, + }, + { + name: "needs both start and end pt2", + args: args{ + end: 12345, + }, + wantErr: true, + }, + { + name: "start and end", + args: args{ + start: 12345, + end: 678912, + }, + expectedStart: 12345, + expectedEnd: 678912, + wantErr: false, + }, + { + name: "date and start", + args: args{ + dateStr: "2024-10-21 -0400", + start: 12345, + }, + wantErr: true, + }, + { + name: "date and end", + args: args{ + dateStr: "2024-10-21 -0400", + end: 678912, + }, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotStart, gotEnd, err := getStartAndEnd(tt.args.dateStr, tt.args.start, tt.args.end) + if (err != nil) != tt.wantErr { + t.Errorf("getStartAndEnd() error = %v, wantErr %v", err, tt.wantErr) + return + } + if gotStart != tt.expectedStart { + t.Errorf("getStartAndEnd() got = %v, want %v", gotStart, tt.expectedStart) + } + if gotEnd != tt.expectedEnd { + t.Errorf("getStartAndEnd() got1 = %v, want %v", gotEnd, tt.expectedEnd) + } + }) + } +} + func makeCmd() *cobra.Command { cmd := &cobra.Command{} cmd.Flags().Int("page", 0, "") diff --git a/server/cmd/mmctl/commands/export.go b/server/cmd/mmctl/commands/export.go index 2e19cdd568..53282692d4 100644 --- a/server/cmd/mmctl/commands/export.go +++ b/server/cmd/mmctl/commands/export.go @@ -244,7 +244,7 @@ func exportDownloadCmdF(c client.Client, command *cobra.Command, args []string) return nil } -// downloadFile handles the common logic for downloading files in export and compliance_export commands +// downloadFile handles the common logic for downloading files in export and compliance-export commands func downloadFile(path string, downloadFn func(*os.File) (string, error), retries int, fileType string) (string, error) { var outFile *os.File info, err := os.Stat(path) diff --git a/server/cmd/mmctl/commands/mmctl_test.go b/server/cmd/mmctl/commands/mmctl_test.go index 2c35517c2c..24fa2b6d37 100644 --- a/server/cmd/mmctl/commands/mmctl_test.go +++ b/server/cmd/mmctl/commands/mmctl_test.go @@ -4,6 +4,10 @@ package commands import ( + "context" + "fmt" + "time" + "github.com/golang/mock/gomock" "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/v8/channels/api4" @@ -81,6 +85,8 @@ func (s *MmctlE2ETestSuite) SetupMessageExportTestHelper() *api4.TestHelper { s.th.App.Srv().Jobs.RegisterJobType(model.JobTypeMessageExport, messageExportImpl.MakeWorker(), messageExportImpl.MakeScheduler()) s.th.App.UpdateConfig(func(cfg *model.Config) { *cfg.MessageExportSettings.DownloadExportResults = true + *cfg.MessageExportSettings.EnableExport = true + *cfg.MessageExportSettings.ExportFormat = model.ComplianceExportTypeActiance }) err := s.th.App.Srv().Jobs.StartWorkers() @@ -125,3 +131,47 @@ func (s *MmctlE2ETestSuite) RunForAllClients(testName string, fn func(client.Cli func (s *MmctlE2ETestSuite) CheckErrorID(err error, errorId string) { api4.CheckErrorID(s.T(), err, errorId) } + +// Helper functions for compliance export job testing + +// getMostRecentJobWithId gets the most recent job with the specified ID +func (s *MmctlE2ETestSuite) getMostRecentJobWithId(id string) *model.Job { + list, _, err := s.th.SystemAdminClient.GetJobsByType(context.Background(), model.JobTypeMessageExport, 0, 1) + s.Require().NoError(err) + s.Require().Len(list, 1) + s.Require().Equal(id, list[0].Id) + return list[0] +} + +// checkJobForStatus polls until the job with the specified ID reaches the expected status +func (s *MmctlE2ETestSuite) checkJobForStatus(id string, status string) { + doneChan := make(chan bool) + var job *model.Job + go func() { + defer close(doneChan) + for { + job = s.getMostRecentJobWithId(id) + if job.Status == status { + break + } + time.Sleep(100 * time.Millisecond) + } + s.Require().Equal(status, job.Status) + }() + select { + case <-doneChan: + case <-time.After(15 * time.Second): + s.Require().Fail(fmt.Sprintf("expected job's status to be %s, got %s", status, job.Status)) + } +} + +// runJobForTest creates a job and waits for it to complete +func (s *MmctlE2ETestSuite) runJobForTest(jobData map[string]string) *model.Job { + job, _, err := s.th.SystemAdminClient.CreateJob(context.Background(), + &model.Job{Type: model.JobTypeMessageExport, Data: jobData}) + s.Require().NoError(err) + // poll until completion + s.checkJobForStatus(job.Id, model.JobStatusSuccess) + job = s.getMostRecentJobWithId(job.Id) + return job +} diff --git a/server/cmd/mmctl/docs/mmctl_compliance-export.rst b/server/cmd/mmctl/docs/mmctl_compliance-export.rst index f01019bb46..9628115a21 100644 --- a/server/cmd/mmctl/docs/mmctl_compliance-export.rst +++ b/server/cmd/mmctl/docs/mmctl_compliance-export.rst @@ -38,6 +38,7 @@ SEE ALSO * `mmctl `_ - Remote client for the Open Source, self-hosted Slack-alternative * `mmctl compliance-export cancel `_ - Cancel compliance export job +* `mmctl compliance-export create `_ - Create a compliance export job, of type 'csv' or 'actiance' or 'globalrelay' * `mmctl compliance-export download `_ - Download compliance export file * `mmctl compliance-export list `_ - List compliance export jobs, sorted by creation date descending (newest first) * `mmctl compliance-export show `_ - Show compliance export job diff --git a/server/cmd/mmctl/docs/mmctl_compliance-export_create.rst b/server/cmd/mmctl/docs/mmctl_compliance-export_create.rst new file mode 100644 index 0000000000..6fdd9b628d --- /dev/null +++ b/server/cmd/mmctl/docs/mmctl_compliance-export_create.rst @@ -0,0 +1,56 @@ +.. _mmctl_compliance-export_create: + +mmctl compliance-export create +------------------------------ + +Create a compliance export job, of type 'csv' or 'actiance' or 'globalrelay' + +Synopsis +~~~~~~~~ + + +Create a compliance export job, of type 'csv' or 'actiance' or 'globalrelay'. If --date is set, the job will run for one day, from 12am to 12am (minus one millisecond) inclusively, in the format with timezone offset: `"YYYY-MM-DD -0000"`. E.g., "2024-10-21 -0400" for Oct 21, 2024 EDT timezone. "2023-11-01 +0000" for Nov 01, 2024 UTC. If set, the 'start' and 'end' flags will be ignored. + +Important: Running a compliance export job from mmctl will NOT affect the next scheduled job's batch_start_time. This means that if you run a compliance export job from mmctl, the next scheduled job will run from the batch_end_time of the previous scheduled job, as usual. + +:: + + mmctl compliance-export create [complianceExportType] --date "2025-03-27 -0400" [flags] + +Examples +~~~~~~~~ + +:: + + compliance-export create csv --date "2025-03-27 -0400" + +Options +~~~~~~~ + +:: + + --date "YYYY-MM-DD -0000" Run the export for one day, from 12am to 12am (minus one millisecond) inclusively, in the format with timezone offset: "YYYY-MM-DD -0000". E.g., `"2024-10-21 -0400"` for Oct 21, 2024 EDT timezone. `"2023-11-01 +0000"` for Nov 01, 2024 UTC. If set, the 'start' and 'end' flags will be ignored. + --end 1743134400000 The end timestamp in unix milliseconds. Posts with updateAt <= end will be exported. If set, 'start' must be set as well. eg, 1743134400000 for 2025-03-28 EDT. + -h, --help help for create + --start 1743048000000 The start timestamp in unix milliseconds. Posts with updateAt >= start will be exported. If set, 'end' must be set as well. eg, 1743048000000 for 2025-03-27 EDT. + +Options inherited from parent commands +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:: + + --config string path to the configuration file (default "$XDG_CONFIG_HOME/mmctl/config") + --disable-pager disables paged output + --insecure-sha1-intermediate allows to use insecure TLS protocols, such as SHA-1 + --insecure-tls-version allows to use TLS versions 1.0 and 1.1 + --json the output format will be in json format + --local allows communicating with the server through a unix socket + --quiet prevent mmctl to generate output for the commands + --strict will only run commands if the mmctl version matches the server one + --suppress-warnings disables printing warning messages + +SEE ALSO +~~~~~~~~ + +* `mmctl compliance-export `_ - Management of compliance exports + diff --git a/server/cmd/mmctl/docs/mmctl_compliance-export_download.rst b/server/cmd/mmctl/docs/mmctl_compliance-export_download.rst index 0d893b3dab..07964aeaef 100644 --- a/server/cmd/mmctl/docs/mmctl_compliance-export_download.rst +++ b/server/cmd/mmctl/docs/mmctl_compliance-export_download.rst @@ -20,7 +20,7 @@ Examples :: - compliance_export download o98rj3ur83dp5dppfyk5yk6osy + compliance-export download o98rj3ur83dp5dppfyk5yk6osy Options ~~~~~~~ diff --git a/server/enterprise/message_export/shared/shared.go b/server/enterprise/message_export/shared/shared.go index f588a9dad3..a1911c98f2 100644 --- a/server/enterprise/message_export/shared/shared.go +++ b/server/enterprise/message_export/shared/shared.go @@ -42,6 +42,7 @@ const ( JobDataJobStartId = "job_start_id" JobDataExportType = "export_type" + JobDataInitiatedBy = "initiated_by" JobDataBatchSize = "batch_size" JobDataChannelBatchSize = "channel_batch_size" JobDataChannelHistoryBatchSize = "channel_history_batch_size" diff --git a/server/enterprise/message_export/worker.go b/server/enterprise/message_export/worker.go index 72170e174b..b81f410304 100644 --- a/server/enterprise/message_export/worker.go +++ b/server/enterprise/message_export/worker.go @@ -22,7 +22,10 @@ import ( "github.com/mattermost/mattermost/server/v8/platform/shared/templates" ) -const TimeBetweenBatchesMs = 100 +const ( + TimeBetweenBatchesMs = 100 + DefaultPreviousJobPageSize = 5 +) // testEndOfBatchCb is only used for testing var testEndOfBatchCb func(worker *MessageExportWorker) @@ -158,8 +161,9 @@ func (w *MessageExportWorker) DoJob(job *model.Job) { go w.jobServer.CancellationWatcher(cancelContext, job.Id, cancelWatcherChan) defer cancelCancelWatcher() + rctx := request.EmptyContext(logger).WithContext(w.context) // if job data is missing, we'll do our best to recover - w.initJobData(logger, job, time.Now()) + w.initJobData(rctx, logger, job, time.Now()) data, err := extractJobData(logger, job.Data) if err != nil { // Error in conversion. Not much we can do about that. But it shouldn't happen, unless someone edited the db. @@ -167,7 +171,6 @@ func (w *MessageExportWorker) DoJob(job *model.Job) { return } - rctx := request.EmptyContext(logger).WithContext(w.context) reportProgress := func(message string) { logger.Debug(message) // Don't fail because we couldn't update progress. @@ -260,7 +263,7 @@ func (w *MessageExportWorker) finishExport(rctx request.CTX, logger *mlog.Logger } // initializes job data if it's missing, allows us to recover from failed or improperly configured jobs -func (w *MessageExportWorker) initJobData(logger mlog.LoggerIFace, job *model.Job, now time.Time) { +func (w *MessageExportWorker) initJobData(rctx request.CTX, logger mlog.LoggerIFace, job *model.Job, now time.Time) { if job.Data == nil { job.Data = make(map[string]string) } @@ -307,7 +310,8 @@ func (w *MessageExportWorker) initJobData(logger mlog.LoggerIFace, job *model.Jo } if _, exists := job.Data[shared.JobDataBatchStartTime]; !exists { - previousJob, err := w.jobServer.Store.Job().GetNewestJobByStatusesAndType([]string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport) + previousJob, err := w.getPreviousNonCliJob(rctx) + if err != nil { exportFromTimestamp := strconv.FormatInt(*w.jobServer.Config().MessageExportSettings.ExportFromTimestamp, 10) logger.Info("Worker: No previously successful job found, falling back to configured MessageExportSettings.ExportFromTimestamp", mlog.String("export_from_timestamp", exportFromTimestamp)) @@ -362,6 +366,36 @@ func (w *MessageExportWorker) initJobData(logger mlog.LoggerIFace, job *model.Jo job.Data[shared.JobDataExportDir] = getJobExportDir(logger, job.Data, job.Data[shared.JobDataJobStartTime], job.Data[shared.JobDataJobEndTime]) } +// getPreviousNonCliJob returns the most recent job that was not initiated by mmctl +func (w *MessageExportWorker) getPreviousNonCliJob(rctx request.CTX) (*model.Job, error) { + offset := 0 + + for { + jobs, err := w.jobServer.Store.Job().GetAllByTypesAndStatusesPage(rctx, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + offset, DefaultPreviousJobPageSize) + if err != nil { + return nil, err + } + + // Find the first job not initiated by mmctl + for _, job := range jobs { + if job.Data == nil || job.Data[shared.JobDataInitiatedBy] != "mmctl" { + return job, nil + } + } + + // If we didn't get a full page of jobs, we've reached the end + if len(jobs) < DefaultPreviousJobPageSize { + return nil, nil + } + + // If we didn't find a non-mmctl job in this page, continue to the next page + offset += DefaultPreviousJobPageSize + } +} + func extractJobData(logger *mlog.Logger, strmap map[string]string) (shared.JobData, error) { data, err := shared.StringMapToJobDataWithZeroValues(strmap) if err != nil { diff --git a/server/enterprise/message_export/worker_test.go b/server/enterprise/message_export/worker_test.go index 390f45aa14..f64ec82b2a 100644 --- a/server/enterprise/message_export/worker_test.go +++ b/server/enterprise/message_export/worker_test.go @@ -43,7 +43,10 @@ func TestInitJobDataNoJobData(t *testing.T) { } // mock job store doesn't return a previously successful job, forcing fallback to config - mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test")) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return(nil, errors.New("test")) worker := &MessageExportWorker{ jobServer: &jobs.JobServer{ @@ -67,7 +70,7 @@ func TestInitJobDataNoJobData(t *testing.T) { } now := time.Now() - worker.initJobData(logger, job, now) + worker.initJobData(request.EmptyContext(logger), logger, job, now) assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType]) assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize]) @@ -98,7 +101,10 @@ func TestInitJobDataPreviousJobNoJobData(t *testing.T) { } // mock job store returns a previously successful job, but it doesn't have job data either, so we still fall back to config - mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(previousJob, nil) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return([]*model.Job{previousJob}, nil) worker := &MessageExportWorker{ jobServer: &jobs.JobServer{ @@ -122,7 +128,7 @@ func TestInitJobDataPreviousJobNoJobData(t *testing.T) { } now := time.Now() - worker.initJobData(logger, job, now) + worker.initJobData(request.EmptyContext(logger), logger, job, now) assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType]) assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize]) @@ -155,7 +161,10 @@ func TestInitJobDataPreviousJobWithJobData(t *testing.T) { } // mock job store returns a previously successful job that has the config that we're looking for, so we use it - mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(previousJob, nil) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return([]*model.Job{previousJob}, nil) worker := &MessageExportWorker{ jobServer: &jobs.JobServer{ @@ -179,7 +188,7 @@ func TestInitJobDataPreviousJobWithJobData(t *testing.T) { } now := time.Now() - worker.initJobData(logger, job, now) + worker.initJobData(request.EmptyContext(logger), logger, job, now) assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType]) assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize]) @@ -212,7 +221,10 @@ func TestInitJobDataPreviousJobWithJobDataPre105(t *testing.T) { } // mock job store returns a previously successful job that has the config that we're looking for, so we use it - mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(previousJob, nil) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return([]*model.Job{previousJob}, nil) worker := &MessageExportWorker{ jobServer: &jobs.JobServer{ @@ -236,7 +248,7 @@ func TestInitJobDataPreviousJobWithJobDataPre105(t *testing.T) { } now := time.Now() - worker.initJobData(logger, job, now) + worker.initJobData(request.EmptyContext(logger), logger, job, now) assert.Equal(t, model.ComplianceExportTypeActiance, job.Data[shared.JobDataExportType]) assert.Equal(t, strconv.Itoa(*worker.jobServer.Config().MessageExportSettings.BatchSize), job.Data[shared.JobDataBatchSize]) @@ -273,7 +285,10 @@ func TestDoJobNoPostsToExport(t *testing.T) { mockMetrics.On("IncrementJobActive", model.JobTypeMessageExport) // no previous job, data will be loaded from config - mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test")) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return(nil, errors.New("test")) // no channels with activity mockStore.ChannelMemberHistoryStore.On("GetChannelsWithActivityDuring", mock.Anything, mock.Anything). @@ -356,7 +371,10 @@ func TestDoJobWithDedicatedExportBackend(t *testing.T) { mockMetrics.On("IncrementJobActive", model.JobTypeMessageExport) // no previous job, data will be loaded from config - mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test")) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return(nil, errors.New("test")) channelId := st.NewTestID() channelName := st.NewTestID() @@ -521,7 +539,10 @@ func TestDoJobCancel(t *testing.T) { mockMetrics.On("IncrementJobActive", model.JobTypeMessageExport) // No previous job, data will be loaded from config - mockStore.JobStore.On("GetNewestJobByStatusesAndType", []string{model.JobStatusWarning, model.JobStatusSuccess}, model.JobTypeMessageExport).Return(nil, errors.New("test")) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return(nil, errors.New("test")) // Job updates the system console UI, once for getting channels, once for getting activity mockStore.JobStore.On("UpdateOptimistically", mock.AnythingOfType("*model.Job"), model.JobStatusInProgress).Return(true, nil).Times(2) @@ -571,3 +592,159 @@ func TestDoJobCancel(t *testing.T) { // Cleanup worker.Stop() } + +func TestGetPreviousJobNoJobs(t *testing.T) { + logger := mlog.CreateConsoleTestLogger(t) + mockStore := &storetest.Store{} + defer mockStore.AssertExpectations(t) + + // Mock the job store to return empty jobs list + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return([]*model.Job{}, nil).Once() + + worker := &MessageExportWorker{ + jobServer: &jobs.JobServer{ + Store: mockStore, + }, + logger: logger, + } + + rctx := request.EmptyContext(logger) + job, err := worker.getPreviousNonCliJob(rctx) + + require.NoError(t, err) + assert.Nil(t, job, "Expected nil job when no jobs are returned") +} + +func TestGetPreviousJobOneRegularJob(t *testing.T) { + logger := mlog.CreateConsoleTestLogger(t) + mockStore := &storetest.Store{} + defer mockStore.AssertExpectations(t) + + regularJob := &model.Job{ + Id: st.NewTestID(), + Status: model.JobStatusSuccess, + Type: model.JobTypeMessageExport, + Data: map[string]string{}, + } + + // Mock the job store to return one regular job + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return([]*model.Job{regularJob}, nil).Once() + + worker := &MessageExportWorker{ + jobServer: &jobs.JobServer{ + Store: mockStore, + }, + logger: logger, + } + + rctx := request.EmptyContext(logger) + job, err := worker.getPreviousNonCliJob(rctx) + + require.NoError(t, err) + assert.Equal(t, regularJob.Id, job.Id, "Expected to get the regular job") +} + +func TestGetPreviousJobOneMmctlJob(t *testing.T) { + logger := mlog.CreateConsoleTestLogger(t) + mockStore := &storetest.Store{} + defer mockStore.AssertExpectations(t) + + mmctlJob := &model.Job{ + Id: st.NewTestID(), + Status: model.JobStatusSuccess, + Type: model.JobTypeMessageExport, + Data: map[string]string{shared.JobDataInitiatedBy: "mmctl"}, + } + + // Mock the job store to return only mmctl jobs (4 jobs, not a full page) + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return([]*model.Job{mmctlJob, mmctlJob, mmctlJob, mmctlJob}, nil).Once() + + worker := &MessageExportWorker{ + jobServer: &jobs.JobServer{ + Store: mockStore, + }, + logger: logger, + } + + rctx := request.EmptyContext(logger) + job, err := worker.getPreviousNonCliJob(rctx) + + require.NoError(t, err) + assert.Nil(t, job, "Expected nil job when only mmctl jobs are found") +} + +func TestGetPreviousJobManyJobs(t *testing.T) { + logger := mlog.CreateConsoleTestLogger(t) + mockStore := &storetest.Store{} + defer mockStore.AssertExpectations(t) + + // Create DefaultPageSize mmctl jobs for first page + firstPageJobs := make([]*model.Job, DefaultPreviousJobPageSize) + for i := range DefaultPreviousJobPageSize { + firstPageJobs[i] = &model.Job{ + Id: st.NewTestID(), + Status: model.JobStatusSuccess, + Type: model.JobTypeMessageExport, + Data: map[string]string{shared.JobDataInitiatedBy: "mmctl"}, + } + } + + // Create DefaultPageSize mmctl jobs for second page + secondPageJobs := make([]*model.Job, DefaultPreviousJobPageSize) + for i := range DefaultPreviousJobPageSize { + secondPageJobs[i] = &model.Job{ + Id: st.NewTestID(), + Status: model.JobStatusSuccess, + Type: model.JobTypeMessageExport, + Data: map[string]string{shared.JobDataInitiatedBy: "mmctl"}, + } + } + + // Create 1 regular job for the third page (last job) + regularJob := &model.Job{ + Id: st.NewTestID(), + Status: model.JobStatusSuccess, + Type: model.JobTypeMessageExport, + Data: map[string]string{}, + } + thirdPageJobs := []*model.Job{regularJob} + + // Mock the job store to return the jobs in pages + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 0, DefaultPreviousJobPageSize).Return(firstPageJobs, nil).Once() + + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 1*DefaultPreviousJobPageSize, DefaultPreviousJobPageSize).Return(secondPageJobs, nil).Once() + + mockStore.JobStore.On("GetAllByTypesAndStatusesPage", mock.Anything, + []string{model.JobTypeMessageExport}, + []string{model.JobStatusWarning, model.JobStatusSuccess}, + 2*DefaultPreviousJobPageSize, DefaultPreviousJobPageSize).Return(thirdPageJobs, nil).Once() + + worker := &MessageExportWorker{ + jobServer: &jobs.JobServer{ + Store: mockStore, + }, + logger: logger, + } + + rctx := request.EmptyContext(logger) + job, err := worker.getPreviousNonCliJob(rctx) + + require.NoError(t, err) + assert.NotNil(t, job) + assert.Equal(t, regularJob.Id, job.Id, "Expected to find the regular job at the end") +}