From f7446d7443b55c36465bd57b5831279ca1fb170e Mon Sep 17 00:00:00 2001 From: Devin Binnie <52460000+devinbinnie@users.noreply.github.com> Date: Fri, 19 Jan 2024 15:22:17 -0500 Subject: [PATCH] [MM-55726] Create batch report worker, add batch report job for exporting users to CSV (#25832) * Split out migration logic and create generic BatchWorker * WIP * WIP * POC batch reporting * Oops * Job hookup * Working export to file * PR feedback * Merge'd * Fix error handling * Add API to start report, translations, couple fixes * Add DMs to send reports to users * Merge'd * Update types * A bit of cleanup * Some fixes * Add missing API doc * PR feedback * Fix generated * Fix bug with post creation * PR feedback * Add some tests * PR feedback * Fix lint * Some test changes * Fix tests * Add comment to explain why we forcibly stop * Rework of some tests * Batch report test * Restrict batch exports to Pro and Enterprise licenses * Fix erroneous comment --------- Co-authored-by: Mattermost Build --- api/v4/source/reports.yaml | 72 +++++- server/channels/api4/report.go | 51 ++++- server/channels/app/app_iface.go | 6 + .../app/opentracing/opentracing_layer.go | 132 +++++++++++ server/channels/app/report.go | 212 ++++++++++++++++++ server/channels/app/report_test.go | 109 +++++++++ server/channels/app/server.go | 7 + server/channels/app/user_test.go | 2 - .../channels/jobs/batch_migration_worker.go | 205 ++++------------- .../jobs/batch_migration_worker_test.go | 186 +++------------ server/channels/jobs/batch_report_worker.go | 139 ++++++++++++ .../channels/jobs/batch_report_worker_test.go | 149 ++++++++++++ server/channels/jobs/batch_worker.go | 174 ++++++++++++++ server/channels/jobs/batch_worker_test.go | 147 ++++++++++++ .../export_users_to_csv.go | 107 +++++++++ server/channels/jobs/helper_test.go | 73 ++++++ server/i18n/en.json | 44 ++++ server/public/model/job.go | 1 + server/public/model/post.go | 4 +- server/public/model/report.go | 68 +++++- server/public/model/user.go | 1 - server/public/model/user_serial_gen.go | 35 +-- webapp/platform/client/src/client4.ts | 8 + 23 files changed, 1571 insertions(+), 361 deletions(-) create mode 100644 server/channels/app/report_test.go create mode 100644 server/channels/jobs/batch_report_worker.go create mode 100644 server/channels/jobs/batch_report_worker_test.go create mode 100644 server/channels/jobs/batch_worker.go create mode 100644 server/channels/jobs/batch_worker_test.go create mode 100644 server/channels/jobs/export_users_to_csv/export_users_to_csv.go diff --git a/api/v4/source/reports.yaml b/api/v4/source/reports.yaml index 59bfa05777..e13344bc38 100644 --- a/api/v4/source/reports.yaml +++ b/api/v4/source/reports.yaml @@ -110,13 +110,13 @@ summary: Gets the full count of users that match the filter. description: > Get the full count of users admin reporting purposes, based on provided parameters. + Must be a system admin to invoke this API. ##### Permissions Requires `sysconsole_read_user_management_users`. - operationId: GetUserCountForReporting parameters: - name: role_filter @@ -156,6 +156,37 @@ application/json: schema: type: number + /api/v4/reports/users/export: + post: + tags: + - reports + summary: Starts a job to export the users to a report file. + description: > + Starts a job to export the users to a report file. + + + Must be a system admin to invoke this API. + + ##### Permissions + + Requires `sysconsole_read_user_management_users`. + operationId: StartBatchUsersExport + parameters: + - name: date_range + in: query + description: The date range of the post statistics to display. Must be one of ("last30days", "previousmonth", "last6months", "alltime"). Will default to 'alltime' if the input is not valid. + schema: + type: string + default: 'alltime' + responses: + "200": + description: Job successfully started + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/UserReport" "400": $ref: "#/components/responses/BadRequest" "401": @@ -164,3 +195,42 @@ $ref: "#/components/responses/Forbidden" "500": $ref: "#/components/responses/InternalServerError" + /api/v4/reports/export/{report_id}: + get: + tags: + - reports + summary: Retrieves a compiled report file for the given report_id + description: > + Retrieves a compiled report file for the given report_id. + + Must be a system admin to invoke this API. + + ##### Permissions + + Requires `sysconsole_read_user_management_users`. + + operationId: RetrieveBatchReportFile + parameters: + - name: report_id + in: path + description: Report ID for the given batch job + required: true + schema: + type: string + - name: format + in: query + description: The format of the report generated (one of "csv") + required: true + schema: + type: string + responses: + "400": + $ref: "#/components/responses/BadRequest" + "401": + $ref: "#/components/responses/Unauthorized" + "403": + $ref: "#/components/responses/Forbidden" + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalServerError" diff --git a/server/channels/api4/report.go b/server/channels/api4/report.go index 5cf6898190..6a6aaccee0 100644 --- a/server/channels/api4/report.go +++ b/server/channels/api4/report.go @@ -17,10 +17,13 @@ import ( func (api *API) InitReports() { api.BaseRoutes.Reports.Handle("/users", api.APISessionRequired(getUsersForReporting)).Methods("GET") api.BaseRoutes.Reports.Handle("/users/count", api.APISessionRequired(getUserCountForReporting)).Methods("GET") + api.BaseRoutes.Reports.Handle("/users/export", api.APISessionRequired(startUsersBatchExport)).Methods("POST") + + api.BaseRoutes.Reports.Handle("/export/{report_id:[A-Za-z0-9]+}", api.APISessionRequired(retrieveBatchReportFile)).Methods("GET") } func getUsersForReporting(c *Context, w http.ResponseWriter, r *http.Request) { - if !(c.IsSystemAdmin() && c.App.SessionHasPermissionTo(*c.AppContext.Session(), model.PermissionSysconsoleReadUserManagementUsers)) { + if !(c.IsSystemAdmin()) { c.SetPermissionError(model.PermissionSysconsoleReadUserManagementUsers) return } @@ -51,7 +54,7 @@ func getUsersForReporting(c *Context, w http.ResponseWriter, r *http.Request) { } func getUserCountForReporting(c *Context, w http.ResponseWriter, r *http.Request) { - if !(c.IsSystemAdmin() && c.App.SessionHasPermissionTo(*c.AppContext.Session(), model.PermissionSysconsoleReadUserManagementUsers)) { + if !(c.IsSystemAdmin()) { c.SetPermissionError(model.PermissionSysconsoleReadUserManagementUsers) return } @@ -73,6 +76,50 @@ func getUserCountForReporting(c *Context, w http.ResponseWriter, r *http.Request } } +func startUsersBatchExport(c *Context, w http.ResponseWriter, r *http.Request) { + if !(c.IsSystemAdmin()) { + c.SetPermissionError(model.PermissionSysconsoleReadUserManagementUsers) + return + } + + startAt, endAt := model.GetReportDateRange(r.URL.Query().Get("date_range"), time.Now()) + if err := c.App.StartUsersBatchExport(c.AppContext, startAt, endAt); err != nil { + c.Err = err + return + } + + ReturnStatusOK(w) +} + +func retrieveBatchReportFile(c *Context, w http.ResponseWriter, r *http.Request) { + if !(c.IsSystemAdmin()) { + c.SetPermissionError(model.PermissionSysconsoleReadUserManagementUsers) + return + } + + reportId := c.Params.ReportId + if !model.IsValidId(reportId) { + c.Err = model.NewAppError("retrieveBatchReportFile", "api.retrieveBatchReportFile.invalid_report_id", nil, "", http.StatusBadRequest) + return + } + + format := r.URL.Query().Get("format") + if !model.IsValidReportExportFormat(format) { + c.Err = model.NewAppError("retrieveBatchReportFile", "api.retrieveBatchReportFile.invalid_format", nil, "", http.StatusBadRequest) + return + } + + file, name, err := c.App.RetrieveBatchReport(reportId, format) + if err != nil { + c.Err = err + return + } + defer file.Close() + + w.Header().Set("Content-Type", "text/csv") + http.ServeContent(w, r, name, time.Time{}, file) +} + func fillReportingBaseOptions(values url.Values) model.ReportingBaseOptions { sortColumn := "Username" if values.Get("sort_column") != "" { diff --git a/server/channels/app/app_iface.go b/server/channels/app/app_iface.go index 2ef5147fb0..af0a4ef3fa 100644 --- a/server/channels/app/app_iface.go +++ b/server/channels/app/app_iface.go @@ -484,6 +484,7 @@ type AppIface interface { CheckUserPostflightAuthenticationCriteria(rctx request.CTX, user *model.User) *model.AppError CheckUserPreflightAuthenticationCriteria(rctx request.CTX, user *model.User, mfaToken string) *model.AppError CheckWebConn(userID, connectionID string) *platform.CheckConnResult + CleanupReportChunks(format string, prefix string, numberOfChunks int) *model.AppError ClearChannelMembersCache(c request.CTX, channelID string) error ClearLatestVersionCache(rctx request.CTX) ClearSessionCacheForAllUsers() @@ -497,6 +498,7 @@ type AppIface interface { Cluster() einterfaces.ClusterInterface CompareAndDeletePluginKey(pluginID string, key string, oldValue []byte) (bool, *model.AppError) CompareAndSetPluginKey(pluginID string, key string, oldValue, newValue []byte) (bool, *model.AppError) + CompileReportChunks(format string, prefix string, numberOfChunks int, headers []string) *model.AppError CompleteOAuth(c request.CTX, service string, body io.ReadCloser, teamID string, props map[string]string, tokenUser *model.User) (*model.User, *model.AppError) CompleteOnboarding(c request.CTX, request *model.CompleteOnboardingRequest) *model.AppError CompleteSwitchWithOAuth(c request.CTX, service string, userData io.Reader, email string, tokenUser *model.User) (*model.User, *model.AppError) @@ -1023,6 +1025,7 @@ type AppIface interface { RestoreTeam(teamID string) *model.AppError RestrictUsersGetByPermissions(c request.CTX, userID string, options *model.UserGetOptions) (*model.UserGetOptions, *model.AppError) RestrictUsersSearchByPermissions(c request.CTX, userID string, options *model.UserSearchOptions) (*model.UserSearchOptions, *model.AppError) + RetrieveBatchReport(reportID string, format string) (filestore.ReadCloseSeeker, string, *model.AppError) ReturnSessionToPool(session *model.Session) RevokeAccessToken(c request.CTX, token string) *model.AppError RevokeAllSessions(c request.CTX, userID string) *model.AppError @@ -1043,6 +1046,7 @@ type AppIface interface { SaveBrandImage(rctx request.CTX, imageData *multipart.FileHeader) *model.AppError SaveComplianceReport(rctx request.CTX, job *model.Compliance) (*model.Compliance, *model.AppError) SaveReactionForPost(c request.CTX, reaction *model.Reaction) (*model.Reaction, *model.AppError) + SaveReportChunk(format string, prefix string, count int, reportData []model.ReportableObject) *model.AppError SaveSharedChannelRemote(remote *model.SharedChannelRemote) (*model.SharedChannelRemote, error) SaveUserTermsOfService(userID, termsOfServiceId string, accepted bool) *model.AppError SchemesIterator(scope string, batchSize int) func() []*model.Scheme @@ -1078,6 +1082,7 @@ type AppIface interface { SendPasswordReset(email string, siteURL string) (bool, *model.AppError) SendPaymentFailedEmail(failedPayment *model.FailedPayment) *model.AppError SendPersistentNotifications() error + SendReportToUser(rctx request.CTX, userID string, jobId string, format string) *model.AppError SendTestPushNotification(deviceID string) string SendUpgradeConfirmationEmail(isYearly bool) *model.AppError ServeInterPluginRequest(w http.ResponseWriter, r *http.Request, sourcePluginId, destinationPluginId string) @@ -1122,6 +1127,7 @@ type AppIface interface { SlackImport(c request.CTX, fileData multipart.File, fileSize int64, teamID string) (*model.AppError, *bytes.Buffer) SoftDeleteTeam(teamID string) *model.AppError Srv() *Server + StartUsersBatchExport(rctx request.CTX, startAt int64, endAt int64) *model.AppError SubmitInteractiveDialog(c request.CTX, request model.SubmitDialogRequest) (*model.SubmitDialogResponse, *model.AppError) SwitchEmailToLdap(c request.CTX, email, password, code, ldapLoginId, ldapPassword string) (string, *model.AppError) SwitchEmailToOAuth(c request.CTX, w http.ResponseWriter, r *http.Request, email, password, code, service string) (string, *model.AppError) diff --git a/server/channels/app/opentracing/opentracing_layer.go b/server/channels/app/opentracing/opentracing_layer.go index 76777673b9..428e1949c4 100644 --- a/server/channels/app/opentracing/opentracing_layer.go +++ b/server/channels/app/opentracing/opentracing_layer.go @@ -1438,6 +1438,28 @@ func (a *OpenTracingAppLayer) CheckWebConn(userID string, connectionID string) * return resultVar0 } +func (a *OpenTracingAppLayer) CleanupReportChunks(format string, prefix string, numberOfChunks int) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CleanupReportChunks") + + 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.CleanupReportChunks(format, prefix, numberOfChunks) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) ClearChannelMembersCache(c request.CTX, channelID string) error { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.ClearChannelMembersCache") @@ -1669,6 +1691,28 @@ func (a *OpenTracingAppLayer) CompareAndSetPluginKey(pluginID string, key string return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) CompileReportChunks(format string, prefix string, numberOfChunks int, headers []string) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CompileReportChunks") + + 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.CompileReportChunks(format, prefix, numberOfChunks, headers) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) CompleteOAuth(c request.CTX, service string, body io.ReadCloser, teamID string, props map[string]string, tokenUser *model.User) (*model.User, *model.AppError) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CompleteOAuth") @@ -14618,6 +14662,28 @@ func (a *OpenTracingAppLayer) RestrictUsersSearchByPermissions(c request.CTX, us return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) RetrieveBatchReport(reportID string, format string) (filestore.ReadCloseSeeker, string, *model.AppError) { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.RetrieveBatchReport") + + 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, resultVar2 := a.app.RetrieveBatchReport(reportID, format) + + if resultVar2 != nil { + span.LogFields(spanlog.Error(resultVar2)) + ext.Error.Set(span, true) + } + + return resultVar0, resultVar1, resultVar2 +} + func (a *OpenTracingAppLayer) ReturnSessionToPool(session *model.Session) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.ReturnSessionToPool") @@ -15051,6 +15117,28 @@ func (a *OpenTracingAppLayer) SaveReactionForPost(c request.CTX, reaction *model return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) SaveReportChunk(format string, prefix string, count int, reportData []model.ReportableObject) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SaveReportChunk") + + 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.SaveReportChunk(format, prefix, count, reportData) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) SaveSharedChannelRemote(remote *model.SharedChannelRemote) (*model.SharedChannelRemote, error) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SaveSharedChannelRemote") @@ -15872,6 +15960,28 @@ func (a *OpenTracingAppLayer) SendPersistentNotifications() error { return resultVar0 } +func (a *OpenTracingAppLayer) SendReportToUser(rctx request.CTX, userID string, jobId string, format string) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SendReportToUser") + + 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.SendReportToUser(rctx, userID, jobId, format) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) SendSubscriptionHistoryEvent(userID string) (*model.SubscriptionHistory, error) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SendSubscriptionHistoryEvent") @@ -16813,6 +16923,28 @@ func (a *OpenTracingAppLayer) SoftDeleteTeam(teamID string) *model.AppError { return resultVar0 } +func (a *OpenTracingAppLayer) StartUsersBatchExport(rctx request.CTX, startAt int64, endAt int64) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.StartUsersBatchExport") + + 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.StartUsersBatchExport(rctx, startAt, endAt) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) SubmitInteractiveDialog(c request.CTX, request model.SubmitDialogRequest) (*model.SubmitDialogResponse, *model.AppError) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SubmitInteractiveDialog") diff --git a/server/channels/app/report.go b/server/channels/app/report.go index e0507b1d89..96e6adf20b 100644 --- a/server/channels/app/report.go +++ b/server/channels/app/report.go @@ -4,11 +4,143 @@ package app import ( + "bytes" + "encoding/csv" + "fmt" "net/http" + "strconv" "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/i18n" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/public/shared/request" + "github.com/mattermost/mattermost/server/v8/platform/shared/filestore" ) +func (a *App) SaveReportChunk(format string, prefix string, count int, reportData []model.ReportableObject) *model.AppError { + switch format { + case "csv": + return a.saveCSVChunk(prefix, count, reportData) + } + return model.NewAppError("SaveReportChunk", "app.save_report_chunk.unsupported_format", nil, "unsupported report format", http.StatusBadRequest) +} + +func (a *App) saveCSVChunk(prefix string, count int, reportData []model.ReportableObject) *model.AppError { + var buf bytes.Buffer + w := csv.NewWriter(&buf) + + for _, report := range reportData { + err := w.Write(report.ToReport()) + if err != nil { + return model.NewAppError("saveCSVChunk", "app.save_csv_chunk.write_error", nil, "", http.StatusInternalServerError).Wrap(err) + } + } + + w.Flush() + if err := w.Error(); err != nil { + return model.NewAppError("saveCSVChunk", "app.save_csv_chunk.write_error", nil, "", http.StatusInternalServerError).Wrap(err) + } + _, appErr := a.WriteFile(&buf, makeFilePath(prefix, count, "csv")) + return appErr +} + +func (a *App) CompileReportChunks(format string, prefix string, numberOfChunks int, headers []string) *model.AppError { + switch format { + case "csv": + return a.compileCSVChunks(prefix, numberOfChunks, headers) + } + return model.NewAppError("CompileReportChunks", "app.compile_report_chunks.unsupported_format", nil, "", http.StatusBadRequest) +} + +func (a *App) compileCSVChunks(prefix string, numberOfChunks int, headers []string) *model.AppError { + filePath := makeCompiledFilePath(prefix, "csv") + + var headerBuf bytes.Buffer + w := csv.NewWriter(&headerBuf) + err := w.Write(headers) + if err != nil { + return model.NewAppError("compileCSVChunks", "app.compile_csv_chunks.header_error", nil, "", http.StatusInternalServerError).Wrap(err) + } + w.Flush() + if err = w.Error(); err != nil { + return model.NewAppError("saveCSVChunk", "app.save_csv_chunk.write_error", nil, "", http.StatusInternalServerError).Wrap(err) + } + _, appErr := a.WriteFile(&headerBuf, filePath) + if appErr != nil { + return appErr + } + + for i := 0; i < numberOfChunks; i++ { + chunkFilePath := makeFilePath(prefix, i, "csv") + chunk, err := a.ReadFile(chunkFilePath) + if err != nil { + return err + } + if _, err = a.AppendFile(bytes.NewReader(chunk), filePath); err != nil { + return err + } + } + + return nil +} + +func (a *App) SendReportToUser(rctx request.CTX, userID string, jobId string, format string) *model.AppError { + systemBot, err := a.GetSystemBot() + if err != nil { + return err + } + + channel, err := a.GetOrCreateDirectChannel(request.EmptyContext(a.Log()), userID, systemBot.UserId) + if err != nil { + return err + } + + post := &model.Post{ + ChannelId: channel.Id, + Message: i18n.T("app.report.send_report_to_user.export_finished", map[string]string{"Link": a.GetSiteURL() + "/api/v4/reports/export/" + jobId + "?format=" + format}), + Type: model.PostTypeAdminReport, + UserId: systemBot.UserId, + Props: model.StringInterface{ + "reportId": jobId, + "format": format, + }, + } + + _, err = a.CreatePost(rctx, post, channel, false, true) + return err +} + +func (a *App) CleanupReportChunks(format string, prefix string, numberOfChunks int) *model.AppError { + switch format { + case "csv": + return a.cleanupCSVChunks(prefix, numberOfChunks) + } + return model.NewAppError("CompileReportChunks", "app.compile_report_chunks.unsupported_format", nil, "", http.StatusBadRequest) +} + +func (a *App) cleanupCSVChunks(prefix string, numberOfChunks int) *model.AppError { + for i := 0; i < numberOfChunks; i++ { + chunkFilePath := makeFilePath(prefix, i, "csv") + if err := a.RemoveFile(chunkFilePath); err != nil { + return err + } + } + + return nil +} + +func makeFilePath(prefix string, count int, extension string) string { + return fmt.Sprintf("admin_reports/batch_report_%s__%d.%s", prefix, count, extension) +} + +func makeCompiledFilePath(prefix string, extension string) string { + return fmt.Sprintf("admin_reports/%s", makeCompiledFilename(prefix, extension)) +} + +func makeCompiledFilename(prefix string, extension string) string { + return fmt.Sprintf("batch_report_%s.%s", prefix, extension) +} + func (a *App) GetUsersForReporting(filter *model.UserReportOptions) ([]*model.UserReport, *model.AppError) { if appErr := filter.IsValid(); appErr != nil { return nil, appErr @@ -35,3 +167,83 @@ func (a *App) GetUserCountForReport(filter *model.UserReportOptions) (*int64, *m return &count, nil } + +func (a *App) StartUsersBatchExport(rctx request.CTX, startAt int64, endAt int64) *model.AppError { + if license := a.Srv().License(); license == nil || (license.SkuShortName != model.LicenseShortSkuProfessional && license.SkuShortName != model.LicenseShortSkuEnterprise) { + return model.NewAppError("StartUsersBatchExport", "app.report.start_users_batch_export.license_error", nil, "", http.StatusBadRequest) + } + + options := map[string]string{ + "requesting_user_id": rctx.Session().UserId, + "start_at": strconv.FormatInt(startAt, 10), + "end_at": strconv.FormatInt(endAt, 10), + } + + // Check for existing job + // TODO: Maybe make this a reusable function? + pendingJobs, err := a.Srv().Jobs.GetJobsByTypeAndStatus(rctx, model.JobTypeExportUsersToCSV, model.JobStatusPending) + if err != nil { + return err + } + for _, job := range pendingJobs { + if job.Data["start_at"] == options["start_at"] && job.Data["end_at"] == options["end_at"] && job.Data["requesting_user_id"] == rctx.Session().UserId { + return model.NewAppError("StartUsersBatchExport", "app.report.start_users_batch_export.job_exists", nil, "", http.StatusBadRequest) + } + } + + inProgressJobs, err := a.Srv().Jobs.GetJobsByTypeAndStatus(rctx, model.JobTypeExportUsersToCSV, model.JobStatusInProgress) + if err != nil { + return err + } + for _, job := range inProgressJobs { + if job.Data["start_at"] == options["start_at"] && job.Data["end_at"] == options["end_at"] && job.Data["requesting_user_id"] == rctx.Session().UserId { + return model.NewAppError("StartUsersBatchExport", "app.report.start_users_batch_export.job_exists", nil, "", http.StatusBadRequest) + } + } + + _, err = a.Srv().Jobs.CreateJobOnce(rctx, model.JobTypeExportUsersToCSV, options) + if err != nil { + return err + } + + a.Srv().Go(func() { + systemBot, err := a.GetSystemBot() + if err != nil { + rctx.Logger().Error("Failed to get the system bot", mlog.Err(err)) + return + } + + channel, err := a.GetOrCreateDirectChannel(request.EmptyContext(a.Log()), rctx.Session().UserId, systemBot.UserId) + if err != nil { + rctx.Logger().Error("Failed to get or create the DM", mlog.Err(err)) + return + } + + post := &model.Post{ + ChannelId: channel.Id, + Message: i18n.T("app.report.start_users_batch_export.started_export"), + Type: model.PostTypeDefault, + UserId: systemBot.UserId, + } + + if _, err := a.CreatePost(rctx, post, channel, false, true); err != nil { + rctx.Logger().Error("Failed to post batch export message", mlog.Err(err)) + } + }) + + return nil +} + +func (a *App) RetrieveBatchReport(reportID string, format string) (filestore.ReadCloseSeeker, string, *model.AppError) { + if license := a.Srv().License(); license == nil || (license.SkuShortName != model.LicenseShortSkuProfessional && license.SkuShortName != model.LicenseShortSkuEnterprise) { + return nil, "", model.NewAppError("RetrieveBatchReport", "app.report.retrieve_batch_report.license_error", nil, "", http.StatusBadRequest) + } + + filePath := makeCompiledFilePath(reportID, format) + reader, err := a.FileReader(filePath) + if err != nil { + return nil, "", err + } + + return reader, makeCompiledFilename(reportID, format), nil +} diff --git a/server/channels/app/report_test.go b/server/channels/app/report_test.go new file mode 100644 index 0000000000..b966a4d86c --- /dev/null +++ b/server/channels/app/report_test.go @@ -0,0 +1,109 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "fmt" + "strconv" + "testing" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/stretchr/testify/require" +) + +type MockReportable struct { + TestField1 string + TestField2 int + TestField3 time.Time +} + +func (mr *MockReportable) ToReport() []string { + return []string{ + mr.TestField1, + strconv.Itoa(mr.TestField2), + mr.TestField3.Format("2006-01-02"), + } +} + +var testData []model.ReportableObject = []model.ReportableObject{ + &MockReportable{ + TestField1: "some-name", + TestField2: 400, + TestField3: time.Date(2024, 1, 1, 0, 0, 0, 0, time.Local), + }, + &MockReportable{ + TestField1: "some-other-name", + TestField2: 500, + TestField3: time.Date(2023, 1, 1, 0, 0, 0, 0, time.Local), + }, + &MockReportable{ + TestField1: "some-other-other-name", + TestField2: 600, + TestField3: time.Date(2022, 1, 1, 0, 0, 0, 0, time.Local), + }, +} + +func TestSaveReportChunk(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + t.Run("should write CSV chunk to file", func(t *testing.T) { + prefix := model.NewId() + err := th.App.SaveReportChunk("csv", prefix, 999, []model.ReportableObject{testData[0]}) + require.Nil(t, err) + + filePath := fmt.Sprintf("admin_reports/batch_report_%s__999.csv", prefix) + bytes, err := th.App.ReadFile(filePath) + require.Nil(t, err) + require.NotNil(t, bytes) + require.Equal(t, "some-name,400,2024-01-01\n", string(bytes)) + }) + + t.Run("should fail if the report format is not supported", func(t *testing.T) { + err := th.App.SaveReportChunk("zzz", model.NewId(), 999, []model.ReportableObject{testData[0]}) + require.NotNil(t, err) + }) +} + +func TestCompileReportChunks(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + prefix := model.NewId() + err := th.App.SaveReportChunk("csv", prefix, 0, []model.ReportableObject{testData[0]}) + require.Nil(t, err) + err = th.App.SaveReportChunk("csv", prefix, 1, []model.ReportableObject{testData[1]}) + require.Nil(t, err) + err = th.App.SaveReportChunk("csv", prefix, 2, []model.ReportableObject{testData[2]}) + require.Nil(t, err) + + t.Run("should compile a bunch of report chunks", func(t *testing.T) { + compileErr := th.App.CompileReportChunks("csv", prefix, 3, []string{"Name", "NumPosts", "StartDate"}) + require.Nil(t, compileErr) + + filePath := fmt.Sprintf("admin_reports/batch_report_%s.csv", prefix) + bytes, readErr := th.App.ReadFile(filePath) + require.Nil(t, readErr) + require.NotNil(t, bytes) + + expected := + `Name,NumPosts,StartDate +some-name,400,2024-01-01 +some-other-name,500,2023-01-01 +some-other-other-name,600,2022-01-01 +` + require.Equal(t, expected, string(bytes)) + }) + + t.Run("should fail if the report format is not supported", func(t *testing.T) { + err = th.App.CompileReportChunks("zzz", prefix, 3, []string{"Name", "NumPosts", "StartDate"}) + require.NotNil(t, err) + }) + + t.Run("should fail if a chunk is missing", func(t *testing.T) { + err = th.App.CompileReportChunks("csv", prefix, 4, []string{"Name", "NumPosts", "StartDate"}) + require.NotNil(t, err) + }) +} diff --git a/server/channels/app/server.go b/server/channels/app/server.go index 81aa04ad2b..6d35b60cce 100644 --- a/server/channels/app/server.go +++ b/server/channels/app/server.go @@ -44,6 +44,7 @@ import ( "github.com/mattermost/mattermost/server/v8/channels/jobs/expirynotify" "github.com/mattermost/mattermost/server/v8/channels/jobs/export_delete" "github.com/mattermost/mattermost/server/v8/channels/jobs/export_process" + "github.com/mattermost/mattermost/server/v8/channels/jobs/export_users_to_csv" "github.com/mattermost/mattermost/server/v8/channels/jobs/extract_content" "github.com/mattermost/mattermost/server/v8/channels/jobs/hosted_purchase_screening" "github.com/mattermost/mattermost/server/v8/channels/jobs/import_delete" @@ -1677,6 +1678,12 @@ func (s *Server) initJobs() { refresh_post_stats.MakeScheduler(s.Jobs, *s.platform.Config().SqlSettings.DriverName), ) + s.Jobs.RegisterJobType( + model.JobTypeExportUsersToCSV, + export_users_to_csv.MakeWorker(s.Jobs, s.Store(), New(ServerConnector(s.Channels()))), + nil, + ) + s.platform.Jobs = s.Jobs } diff --git a/server/channels/app/user_test.go b/server/channels/app/user_test.go index d50d2468e3..e4a477627f 100644 --- a/server/channels/app/user_test.go +++ b/server/channels/app/user_test.go @@ -1991,8 +1991,6 @@ func TestGetUsersForReporting(t *testing.T) { CreateAt: 1000, FirstName: "Bob", LastName: "Bobson", - }, - UserPostStats: model.UserPostStats{ LastLogin: 1500, }, }, diff --git a/server/channels/jobs/batch_migration_worker.go b/server/channels/jobs/batch_migration_worker.go index 072c8e3174..4d615a8cb1 100644 --- a/server/channels/jobs/batch_migration_worker.go +++ b/server/channels/jobs/batch_migration_worker.go @@ -5,7 +5,6 @@ package jobs import ( "net/http" - "sync/atomic" "time" "github.com/mattermost/mattermost/server/public/model" @@ -29,83 +28,57 @@ type BatchMigrationWorkerAppIFace interface { // server in order to retry a failed migration job. Refactoring the job infrastructure is left as // a future exercise. type BatchMigrationWorker struct { - jobServer *JobServer - logger mlog.LoggerIFace - store store.Store - app BatchMigrationWorkerAppIFace - - stop chan struct{} - stopped chan bool - closed atomic.Bool - jobs chan model.Job - - migrationKey string - timeBetweenBatches time.Duration - doMigrationBatch func(data model.StringMap, store store.Store) (model.StringMap, bool, error) + *BatchWorker + app BatchMigrationWorkerAppIFace + migrationKey string + doMigrationBatch func(data model.StringMap, store store.Store) (model.StringMap, bool, error) } // MakeBatchMigrationWorker creates a worker to process the given migration batch function. -func MakeBatchMigrationWorker(jobServer *JobServer, store store.Store, app BatchMigrationWorkerAppIFace, migrationKey string, timeBetweenBatches time.Duration, doMigrationBatch func(data model.StringMap, store store.Store) (model.StringMap, bool, error)) model.Worker { +func MakeBatchMigrationWorker( + jobServer *JobServer, + store store.Store, + app BatchMigrationWorkerAppIFace, + migrationKey string, + timeBetweenBatches time.Duration, + doMigrationBatch func(data model.StringMap, store store.Store) (model.StringMap, bool, error), +) *BatchMigrationWorker { worker := &BatchMigrationWorker{ - jobServer: jobServer, - logger: jobServer.Logger().With(mlog.String("worker_name", migrationKey)), - store: store, - app: app, - stop: make(chan struct{}), - stopped: make(chan bool, 1), - jobs: make(chan model.Job), - migrationKey: migrationKey, - timeBetweenBatches: timeBetweenBatches, - doMigrationBatch: doMigrationBatch, + app: app, + migrationKey: migrationKey, + doMigrationBatch: doMigrationBatch, } + worker.BatchWorker = MakeBatchWorker(jobServer, store, timeBetweenBatches, worker.doBatch) return worker } -// Run starts the worker dedicated to the unique migration batch job it will be given to process. -func (worker *BatchMigrationWorker) Run() { - worker.logger.Debug("Worker started") - // We have to re-assign the stop channel again, because - // it might happen that the job was restarted due to a config change. - if worker.closed.CompareAndSwap(true, false) { - worker.stop = make(chan struct{}) +func (worker *BatchMigrationWorker) doBatch(rctx *request.Context, job *model.Job) bool { + // Ensure the cluster remains in sync, otherwise we restart the job to + // ensure a complete migration. Technically, the cluster could go out of + // sync briefly within a batch, but we accept that risk. + if !worker.checkIsClusterInSync(rctx) { + worker.logger.Warn("Worker: Resetting job") + worker.resetJob(worker.logger, job) + return true } - defer func() { - worker.logger.Debug("Worker finished") - worker.stopped <- true - }() - - for { - select { - case <-worker.stop: - worker.logger.Debug("Worker received stop signal") - return - case job := <-worker.jobs: - worker.DoJob(&job) - } - } -} - -// Stop interrupts the worker even if the migration has not yet completed. -func (worker *BatchMigrationWorker) Stop() { - // Set to close, and if already closed before, then return. - if !worker.closed.CompareAndSwap(false, true) { - return + nextData, done, err := worker.doMigrationBatch(job.Data, worker.store) + if err != nil { + worker.logger.Error("Worker: Failed to do migration batch. Exiting", mlog.Err(err)) + worker.setJobError(worker.logger, job, model.NewAppError("doMigrationBatch", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)) + return true + } else if done { + worker.logger.Info("Worker: Job is complete") + worker.setJobSuccess(worker.logger, job) + worker.markAsComplete() + return true } - worker.logger.Debug("Worker stopping") - close(worker.stop) - <-worker.stopped -} + job.Data = nextData -// JobChannel is the means by which the jobs infrastructure provides the worker the job to execute. -func (worker *BatchMigrationWorker) JobChannel() chan<- model.Job { - return worker.jobs -} - -// IsEnabled is always true for batch migrations. -func (worker *BatchMigrationWorker) IsEnabled(_ *model.Config) bool { - return true + // Migrations currently don't support reporting meaningful progress. + worker.jobServer.SetJobProgress(job, 0) + return false } // checkIsClusterInSync returns true if all nodes in the cluster are running the same version, @@ -128,108 +101,6 @@ func (worker *BatchMigrationWorker) checkIsClusterInSync(rctx request.CTX) bool return true } -// DoJob executes the job picked up through the job channel. -// -// Note that this is a lot of distracting machinery here to claim the job, then double check the -// status, and keep the status up to date in line with job infrastrcuture semantics. Unless an -// error occurs, this worker should hold onto the job until its completed. -func (worker *BatchMigrationWorker) DoJob(job *model.Job) { - logger := worker.logger.With(mlog.Any("job", job)) - logger.Debug("Worker received a new candidate job.") - defer worker.jobServer.HandleJobPanic(logger, job) - - if claimed, err := worker.jobServer.ClaimJob(job); err != nil { - logger.Warn("Worker experienced an error while trying to claim job", mlog.Err(err)) - return - } else if !claimed { - return - } - - c := request.EmptyContext(logger) - var appErr *model.AppError - - // We get the job again because ClaimJob changes the job status. - job, appErr = worker.jobServer.GetJob(c, job.Id) - if appErr != nil { - worker.logger.Error("Worker: job execution error", mlog.Err(appErr)) - worker.setJobError(logger, job, appErr) - return - } - - if job.Data == nil { - job.Data = make(model.StringMap) - } - - for { - select { - case <-worker.stop: - logger.Info("Worker: Migration has been canceled via Worker Stop. Setting the job back to pending.") - if err := worker.jobServer.SetJobPending(job); err != nil { - worker.logger.Error("Worker: Failed to mark job as pending", mlog.Err(err)) - } - return - case <-time.After(worker.timeBetweenBatches): - // Ensure the cluster remains in sync, otherwise we restart the job to - // ensure a complete migration. Technically, the cluster could go out of - // sync briefly within a batch, but we accept that risk. - if !worker.checkIsClusterInSync(c) { - worker.logger.Warn("Worker: Resetting job") - worker.resetJob(logger, job) - return - } - - nextData, done, err := worker.doMigrationBatch(job.Data, worker.store) - if err != nil { - worker.logger.Error("Worker: Failed to do migration batch. Exiting", mlog.Err(err)) - worker.setJobError(logger, job, model.NewAppError("doMigrationBatch", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)) - return - } else if done { - logger.Info("Worker: Job is complete") - worker.setJobSuccess(logger, job) - worker.markAsComplete() - return - } - - job.Data = nextData - - // Migrations currently don't support reporting meaningful progress. - worker.jobServer.SetJobProgress(job, 0) - } - } -} - -// resetJob erases the data tracking the next batch to execute and returns the job status to -// pending to allow the job infrastructure to requeue it. -func (worker *BatchMigrationWorker) resetJob(logger mlog.LoggerIFace, job *model.Job) { - job.Data = nil - job.Progress = 0 - job.Status = model.JobStatusPending - - if _, err := worker.store.Job().UpdateOptimistically(job, model.JobStatusInProgress); err != nil { - worker.logger.Error("Worker: Failed to reset job data. May resume instead of restarting.", mlog.Err(err)) - } -} - -// setJobSuccess records the job as successful. -func (worker *BatchMigrationWorker) setJobSuccess(logger mlog.LoggerIFace, job *model.Job) { - if err := worker.jobServer.SetJobProgress(job, 100); err != nil { - logger.Error("Worker: Failed to update progress for job", mlog.Err(err)) - worker.setJobError(logger, job, err) - } - - if err := worker.jobServer.SetJobSuccess(job); err != nil { - logger.Error("Worker: Failed to set success for job", mlog.Err(err)) - worker.setJobError(logger, job, err) - } -} - -// setJobError puts the job into an error state, preventing the job from running again. -func (worker *BatchMigrationWorker) setJobError(logger mlog.LoggerIFace, job *model.Job, appError *model.AppError) { - if err := worker.jobServer.SetJobError(job, appError); err != nil { - logger.Error("Worker: Failed to set job error", mlog.Err(err)) - } -} - // markAsComplete records a discrete migration key to prevent this job from ever running again. func (worker *BatchMigrationWorker) markAsComplete() { system := model.System{ diff --git a/server/channels/jobs/batch_migration_worker_test.go b/server/channels/jobs/batch_migration_worker_test.go index 8f24dca1fc..c97894abc9 100644 --- a/server/channels/jobs/batch_migration_worker_test.go +++ b/server/channels/jobs/batch_migration_worker_test.go @@ -45,51 +45,18 @@ func (ma *MockApp) SetOutOfSync() { } func TestBatchMigrationWorker(t *testing.T) { - waitDone := func(t *testing.T, done chan bool, msg string) { - t.Helper() - - require.Eventually(t, func() bool { - select { - case <-done: - return true - default: - return false - } - }, 5*time.Second, 100*time.Millisecond, msg) - } - setupBatchWorker := func(t *testing.T, th *TestHelper, mockApp *MockApp, doMigrationBatch func(model.StringMap, store.Store) (model.StringMap, bool, error)) (model.Worker, *model.Job) { t.Helper() - migrationKey := model.NewId() - timeBetweenBatches := 1 * time.Second - worker := jobs.MakeBatchMigrationWorker( th.Server.Jobs, th.Server.Store(), mockApp, - migrationKey, - timeBetweenBatches, + model.NewId(), + 1*time.Second, doMigrationBatch, ) - th.Server.Jobs.RegisterJobType(migrationKey, worker, nil) - - job, appErr := th.Server.Jobs.CreateJob(th.Context, migrationKey, nil) - require.Nil(t, appErr) - - done := make(chan bool) - go func() { - defer close(done) - worker.Run() - }() - - // When ending the test, ensure we wait for the worker to finish. - t.Cleanup(func() { - waitDone(t, done, "worker did not stop running") - }) - - // Give the worker time to start running - time.Sleep(500 * time.Millisecond) + job := th.SetupBatchWorker(t, worker.BatchWorker) return worker, job } @@ -106,18 +73,6 @@ func TestBatchMigrationWorker(t *testing.T) { waitDone(t, stopped, "worker did not stop") } - waitForJobStatus := func(t *testing.T, th *TestHelper, job *model.Job, status string) { - t.Helper() - - require.Eventuallyf(t, func() bool { - actualJob, appErr := th.Server.Jobs.GetJob(th.Context, job.Id) - require.Nil(t, appErr) - require.Equal(t, job.Id, actualJob.Id) - - return actualJob.Status == status - }, 5*time.Second, 250*time.Millisecond, "job never transitioned to %s", status) - } - assertJobReset := func(t *testing.T, th *TestHelper, job *model.Job) { actualJob, appErr := th.Server.Jobs.GetJob(th.Context, job.Id) require.Nil(t, appErr) @@ -144,6 +99,34 @@ func TestBatchMigrationWorker(t *testing.T) { return data } + t.Run("done after three batches", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + mockApp := &MockApp{} + + var worker model.Worker + var job *model.Job + worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) { + batchNumber := getBatchNumberFromData(t, data) + require.LessOrEqual(t, batchNumber, 3, "only 3 batches should have run") + + if batchNumber >= 3 { + go worker.Stop() // Shut down the worker when the job is done + return getDataFromBatchNumber(batchNumber), true, nil + } + + batchNumber++ + return getDataFromBatchNumber(batchNumber), false, nil + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForJobStatus(t, job, model.JobStatusSuccess) + th.WaitForBatchNumber(t, job, 3) + }) + t.Run("clusters not in sync before first batch", func(t *testing.T) { th := Setup(t).InitBasic() defer th.TearDown() @@ -165,65 +148,12 @@ func TestBatchMigrationWorker(t *testing.T) { // Queue the work to be done worker.JobChannel() <- *job - waitForJobStatus(t, th, job, model.JobStatusPending) + th.WaitForJobStatus(t, job, model.JobStatusPending) assertJobReset(t, th, job) stopWorker(t, worker) }) - t.Run("stop after first batch", func(t *testing.T) { - th := Setup(t).InitBasic() - defer th.TearDown() - - mockApp := &MockApp{} - - var worker model.Worker - var job *model.Job - worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) { - batchNumber := getBatchNumberFromData(t, data) - - require.Equal(t, 1, batchNumber, "only batch 1 should have run") - - // Shut down the worker after the first batch to prevent subsequent ones. - go worker.Stop() - - batchNumber++ - - return getDataFromBatchNumber(batchNumber), false, nil - }) - - // Queue the work to be done - worker.JobChannel() <- *job - - waitForJobStatus(t, th, job, model.JobStatusPending) - }) - - t.Run("stop after second batch", func(t *testing.T) { - th := Setup(t).InitBasic() - defer th.TearDown() - - mockApp := &MockApp{} - - var worker model.Worker - var job *model.Job - worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) { - batchNumber := getBatchNumberFromData(t, data) - - require.LessOrEqual(t, batchNumber, 2, "only batches 1 and 2 should have run") - - // Shut down the worker after the first batch to prevent subsequent ones. - go worker.Stop() - batchNumber++ - - return getDataFromBatchNumber(batchNumber), false, nil - }) - - // Queue the work to be done - worker.JobChannel() <- *job - - waitForJobStatus(t, th, job, model.JobStatusPending) - }) - t.Run("clusters not in sync after first batch", func(t *testing.T) { th := Setup(t).InitBasic() defer th.TearDown() @@ -248,59 +178,9 @@ func TestBatchMigrationWorker(t *testing.T) { // Queue the work to be done worker.JobChannel() <- *job - waitForJobStatus(t, th, job, model.JobStatusPending) + th.WaitForJobStatus(t, job, model.JobStatusPending) assertJobReset(t, th, job) stopWorker(t, worker) }) - - t.Run("done after first batch", func(t *testing.T) { - th := Setup(t).InitBasic() - defer th.TearDown() - - mockApp := &MockApp{} - - var worker model.Worker - var job *model.Job - worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) { - batchNumber := getBatchNumberFromData(t, data) - require.Equal(t, 1, batchNumber, "only batch 1 should have run") - - // Shut down the worker after the first batch to prevent subsequent ones. - go worker.Stop() - batchNumber++ - - return getDataFromBatchNumber(batchNumber), true, nil - }) - - // Queue the work to be done - worker.JobChannel() <- *job - - waitForJobStatus(t, th, job, model.JobStatusSuccess) - }) - - t.Run("done after three batches", func(t *testing.T) { - th := Setup(t).InitBasic() - defer th.TearDown() - - mockApp := &MockApp{} - - var worker model.Worker - var job *model.Job - worker, job = setupBatchWorker(t, th, mockApp, func(data model.StringMap, s store.Store) (model.StringMap, bool, error) { - batchNumber := getBatchNumberFromData(t, data) - require.LessOrEqual(t, batchNumber, 3, "only 3 batches should have run") - - // Shut down the worker after the first batch to prevent subsequent ones. - go worker.Stop() - batchNumber++ - - return getDataFromBatchNumber(batchNumber), true, nil - }) - - // Queue the work to be done - worker.JobChannel() <- *job - - waitForJobStatus(t, th, job, model.JobStatusSuccess) - }) } diff --git a/server/channels/jobs/batch_report_worker.go b/server/channels/jobs/batch_report_worker.go new file mode 100644 index 0000000000..4d341d8853 --- /dev/null +++ b/server/channels/jobs/batch_report_worker.go @@ -0,0 +1,139 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package jobs + +import ( + "net/http" + "strconv" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/public/shared/request" + "github.com/mattermost/mattermost/server/v8/channels/store" + "github.com/pkg/errors" +) + +type BatchReportWorkerAppIFace interface { + SaveReportChunk(format string, prefix string, count int, reportData []model.ReportableObject) *model.AppError + CompileReportChunks(format string, prefix string, numberOfChunks int, headers []string) *model.AppError + SendReportToUser(rctx request.CTX, userID string, jobId string, format string) *model.AppError + CleanupReportChunks(format string, prefix string, numberOfChunks int) *model.AppError +} + +type BatchReportWorker struct { + *BatchWorker + app BatchReportWorkerAppIFace + reportFormat string + headers []string + getData func(jobData model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error) +} + +func MakeBatchReportWorker( + jobServer *JobServer, + store store.Store, + app BatchReportWorkerAppIFace, + timeBetweenBatches time.Duration, + reportFormat string, + headers []string, + getData func(jobData model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error), +) *BatchReportWorker { + worker := &BatchReportWorker{ + app: app, + reportFormat: reportFormat, + headers: headers, + getData: getData, + } + worker.BatchWorker = MakeBatchWorker(jobServer, store, timeBetweenBatches, worker.doBatch) + return worker +} + +func (worker *BatchReportWorker) doBatch(rctx *request.Context, job *model.Job) bool { + reportData, nextData, done, err := worker.getData(job.Data) + if err != nil { + worker.logger.Error("Worker: Failed to get data for report batch. Exiting", mlog.Err(err)) + worker.setJobError(worker.logger, job, model.NewAppError("doBatch", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)) + return true + } else if done { + if err = worker.complete(rctx, job); err != nil { + worker.logger.Error("Worker: Failed to finish the batch report. Exiting", mlog.Err(err)) + worker.setJobError(worker.logger, job, model.NewAppError("doBatch", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)) + } else { + worker.logger.Info("Worker: Report job complete") + worker.setJobSuccess(worker.logger, job) + } + + return true + } + + err = worker.processChunk(job, reportData) + if err != nil { + worker.logger.Error("Worker: Failed to save report batch. Exiting", mlog.Err(err)) + worker.setJobError(worker.logger, job, model.NewAppError("doBatch", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)) + return true + } + + job.Data = nextData + + // We might be able to add progress for this type of job in the future + // But for now we can just set to 0 + worker.jobServer.SetJobProgress(job, 0) + return false +} + +func getFileCount(jobData model.StringMap) (int, error) { + if jobData["file_count"] != "" { + parsedFileCount, parseErr := strconv.Atoi(jobData["file_count"]) + if parseErr != nil { + return 0, errors.Wrap(parseErr, "failed to parse file_count") + } + return parsedFileCount, nil + } + + // Assume it hasn't been set + return 0, nil +} + +func (worker *BatchReportWorker) processChunk(job *model.Job, reportData []model.ReportableObject) error { + fileCount, err := getFileCount(job.Data) + if err != nil { + return err + } + + appErr := worker.app.SaveReportChunk(worker.reportFormat, job.Id, fileCount, reportData) + if appErr != nil { + return err + } + + fileCount++ + job.Data["file_count"] = strconv.Itoa(fileCount) + + return nil +} + +func (worker *BatchReportWorker) complete(rctx request.CTX, job *model.Job) error { + requestingUserId := job.Data["requesting_user_id"] + if requestingUserId == "" { + return errors.New("No user to send the report to") + } + fileCount, err := getFileCount(job.Data) + if err != nil { + return err + } + + appErr := worker.app.CompileReportChunks(worker.reportFormat, job.Id, fileCount, worker.headers) + if appErr != nil { + return appErr + } + + defer func() { + worker.app.CleanupReportChunks(worker.reportFormat, job.Id, fileCount) + }() + + if appErr = worker.app.SendReportToUser(rctx, requestingUserId, job.Id, worker.reportFormat); appErr != nil { + return appErr + } + + return nil +} diff --git a/server/channels/jobs/batch_report_worker_test.go b/server/channels/jobs/batch_report_worker_test.go new file mode 100644 index 0000000000..e51cc1a919 --- /dev/null +++ b/server/channels/jobs/batch_report_worker_test.go @@ -0,0 +1,149 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package jobs_test + +import ( + "errors" + "strconv" + "testing" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/request" + "github.com/mattermost/mattermost/server/v8/channels/jobs" + "github.com/stretchr/testify/require" +) + +type ReportMockApp struct{} + +func (rma *ReportMockApp) SaveReportChunk(format string, prefix string, count int, reportData []model.ReportableObject) *model.AppError { + return nil +} +func (rma *ReportMockApp) CompileReportChunks(format string, prefix string, numberOfChunks int, headers []string) *model.AppError { + return nil +} +func (rma *ReportMockApp) SendReportToUser(rctx request.CTX, userID string, jobId string, format string) *model.AppError { + return nil +} +func (rma *ReportMockApp) CleanupReportChunks(format string, prefix string, numberOfChunks int) *model.AppError { + return nil +} + +func TestBatchReportWorker(t *testing.T) { + setupBatchWorker := func( + t *testing.T, + th *TestHelper, + getData func(jobData model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error), + ) (*jobs.BatchReportWorker, *model.Job) { + t.Helper() + + worker := jobs.MakeBatchReportWorker( + th.Server.Jobs, + th.Server.Store(), + &ReportMockApp{}, + 1*time.Second, + "csv", + []string{}, + getData) + job := th.SetupBatchWorker(t, worker.BatchWorker) + return worker, job + } + + waitForFileCount := func(t *testing.T, th *TestHelper, job *model.Job, fileCount int) { + t.Helper() + + require.Eventuallyf(t, func() bool { + actualJob, appErr := th.Server.Jobs.GetJob(th.Context, job.Id) + require.Nil(t, appErr) + require.Equal(t, job.Id, actualJob.Id) + + finalFileCount, err := strconv.Atoi(actualJob.Data["file_count"]) + require.NoError(t, err) + return finalFileCount == fileCount + }, 5*time.Second, 250*time.Millisecond, "job did not stop at batch %d", fileCount) + } + + getFileCountFromData := func(t *testing.T, data model.StringMap) int { + t.Helper() + + if data["file_count"] == "" { + return 0 + } + + fileCount, err := strconv.Atoi(data["file_count"]) + require.NoError(t, err) + + return fileCount + } + + createData := func(th *TestHelper, data model.StringMap) model.StringMap { + data["requesting_user_id"] = th.SystemAdminUser.Id + return data + } + + t.Run("should finish when the report is done, incrementing file count along the way", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + var worker model.Worker + var job *model.Job + + iterations := 0 + + worker, job = setupBatchWorker(t, th, func(data model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error) { + fileCount := getFileCountFromData(t, data) + require.Equal(t, iterations, fileCount) + require.LessOrEqual(t, fileCount, 3, "only 3 batches should have run") + + iterations++ + + if fileCount >= 3 { + go worker.Stop() // Shut down the worker when the job is done + return []model.ReportableObject{}, createData(th, data), true, nil + } + + return []model.ReportableObject{}, createData(th, data), false, nil + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForJobStatus(t, job, model.JobStatusSuccess) + waitForFileCount(t, th, job, 3) + }) + + t.Run("should fail job when get data throws an error", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + var worker model.Worker + var job *model.Job + worker, job = setupBatchWorker(t, th, func(data model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error) { + go worker.Stop() // Shut down the worker right after this + return []model.ReportableObject{}, createData(th, data), false, errors.New("failed to fetch data") + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForJobStatus(t, job, model.JobStatusError) + }) + + t.Run("should fail if there is no user id to send the report to", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + var worker model.Worker + var job *model.Job + worker, job = setupBatchWorker(t, th, func(data model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error) { + go worker.Stop() // Shut down the worker right after this + return []model.ReportableObject{}, make(model.StringMap), true, nil + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForJobStatus(t, job, model.JobStatusError) + }) +} diff --git a/server/channels/jobs/batch_worker.go b/server/channels/jobs/batch_worker.go new file mode 100644 index 0000000000..8113595168 --- /dev/null +++ b/server/channels/jobs/batch_worker.go @@ -0,0 +1,174 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package jobs + +import ( + "sync/atomic" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/public/shared/request" + "github.com/mattermost/mattermost/server/v8/channels/store" +) + +type BatchWorker struct { + jobServer *JobServer + logger mlog.LoggerIFace + store store.Store + + stop chan struct{} + stopped chan bool + closed atomic.Bool + jobs chan model.Job + + timeBetweenBatches time.Duration + doBatch func(rctx *request.Context, job *model.Job) bool +} + +// MakeBatchWorker creates a worker to process the given batch function. +func MakeBatchWorker( + jobServer *JobServer, + store store.Store, + timeBetweenBatches time.Duration, + doBatch func(rctx *request.Context, job *model.Job) bool, +) *BatchWorker { + return &BatchWorker{ + jobServer: jobServer, + logger: jobServer.Logger(), + store: store, + stop: make(chan struct{}), + stopped: make(chan bool, 1), + jobs: make(chan model.Job), + timeBetweenBatches: timeBetweenBatches, + doBatch: doBatch, + } +} + +// Run starts the worker dedicated to the unique migration batch job it will be given to process. +func (worker *BatchWorker) Run() { + worker.logger.Debug("Worker started") + // We have to re-assign the stop channel again, because + // it might happen that the job was restarted due to a config change. + if worker.closed.CompareAndSwap(true, false) { + worker.stop = make(chan struct{}) + } + + defer func() { + worker.logger.Debug("Worker finished") + worker.stopped <- true + }() + + for { + select { + case <-worker.stop: + worker.logger.Debug("Worker received stop signal") + return + case job := <-worker.jobs: + worker.DoJob(&job) + } + } +} + +// Stop interrupts the worker even if the migration has not yet completed. +func (worker *BatchWorker) Stop() { + // Set to close, and if already closed before, then return. + if !worker.closed.CompareAndSwap(false, true) { + return + } + + worker.logger.Debug("Worker stopping") + close(worker.stop) + <-worker.stopped +} + +// JobChannel is the means by which the jobs infrastructure provides the worker the job to execute. +func (worker *BatchWorker) JobChannel() chan<- model.Job { + return worker.jobs +} + +// IsEnabled is always true for batches. +func (worker *BatchWorker) IsEnabled(_ *model.Config) bool { + return true +} + +// DoJob executes the job picked up through the job channel. +// +// Note that this is a lot of distracting machinery here to claim the job, then double check the +// status, and keep the status up to date in line with job infrastrcuture semantics. Unless an +// error occurs, this worker should hold onto the job until its completed. +func (worker *BatchWorker) DoJob(job *model.Job) { + logger := worker.logger.With(mlog.Any("job", job)) + logger.Debug("Worker received a new candidate job.") + defer worker.jobServer.HandleJobPanic(logger, job) + + if claimed, err := worker.jobServer.ClaimJob(job); err != nil { + logger.Warn("Worker experienced an error while trying to claim job", mlog.Err(err)) + return + } else if !claimed { + return + } + + c := request.EmptyContext(logger) + var appErr *model.AppError + + // We get the job again because ClaimJob changes the job status. + job, appErr = worker.jobServer.GetJob(c, job.Id) + if appErr != nil { + worker.logger.Error("Worker: job execution error", mlog.Err(appErr)) + worker.setJobError(logger, job, appErr) + return + } + + if job.Data == nil { + job.Data = make(model.StringMap) + } + + for { + select { + case <-worker.stop: + logger.Info("Worker: Batch has been canceled via Worker Stop. Setting the job back to pending.") + if err := worker.jobServer.SetJobPending(job); err != nil { + worker.logger.Error("Worker: Failed to mark job as pending", mlog.Err(err)) + } + return + case <-time.After(worker.timeBetweenBatches): + if stop := worker.doBatch(c, job); stop { + return + } + } + } +} + +// resetJob erases the data tracking the next batch to execute and returns the job status to +// pending to allow the job infrastructure to requeue it. +func (worker *BatchWorker) resetJob(logger mlog.LoggerIFace, job *model.Job) { + job.Data = nil + job.Progress = 0 + job.Status = model.JobStatusPending + + if _, err := worker.store.Job().UpdateOptimistically(job, model.JobStatusInProgress); err != nil { + worker.logger.Error("Worker: Failed to reset job data. May resume instead of restarting.", mlog.Err(err)) + } +} + +// setJobSuccess records the job as successful. +func (worker *BatchWorker) setJobSuccess(logger mlog.LoggerIFace, job *model.Job) { + if err := worker.jobServer.SetJobProgress(job, 100); err != nil { + logger.Error("Worker: Failed to update progress for job", mlog.Err(err)) + worker.setJobError(logger, job, err) + } + + if err := worker.jobServer.SetJobSuccess(job); err != nil { + logger.Error("Worker: Failed to set success for job", mlog.Err(err)) + worker.setJobError(logger, job, err) + } +} + +// setJobError puts the job into an error state, preventing the job from running again. +func (worker *BatchWorker) setJobError(logger mlog.LoggerIFace, job *model.Job, appError *model.AppError) { + if err := worker.jobServer.SetJobError(job, appError); err != nil { + logger.Error("Worker: Failed to set job error", mlog.Err(err)) + } +} diff --git a/server/channels/jobs/batch_worker_test.go b/server/channels/jobs/batch_worker_test.go new file mode 100644 index 0000000000..7a1bb3c674 --- /dev/null +++ b/server/channels/jobs/batch_worker_test.go @@ -0,0 +1,147 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package jobs_test + +import ( + "strconv" + "testing" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/request" + "github.com/mattermost/mattermost/server/v8/channels/jobs" + "github.com/stretchr/testify/require" +) + +func TestBatchWorker(t *testing.T) { + createBatchWorker := func(t *testing.T, th *TestHelper, doBatch func(rctx *request.Context, job *model.Job) bool) (*jobs.BatchWorker, *model.Job) { + t.Helper() + + worker := jobs.MakeBatchWorker(th.Server.Jobs, th.Server.Store(), 1*time.Second, doBatch) + job := th.SetupBatchWorker(t, worker) + return worker, job + } + + getBatchNumberFromData := func(t *testing.T, data model.StringMap) int { + t.Helper() + + batchNumber, err := strconv.Atoi(data["batch_number"]) + require.NoError(t, err) + + return batchNumber + } + + incrementBatchNumber := func(t *testing.T, th *TestHelper, job *model.Job) { + t.Helper() + + batchNumber, err := strconv.Atoi(job.Data["batch_number"]) + require.NoError(t, err) + + batchNumber++ + job.Data["batch_number"] = strconv.Itoa(batchNumber) + th.Server.Jobs.SetJobProgress(job, 0) + } + + t.Run("stop after first batch", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + var worker *jobs.BatchWorker + worker, job := createBatchWorker(t, th, func(rctx *request.Context, job *model.Job) bool { + batchNumber := getBatchNumberFromData(t, job.Data) + + require.Equal(t, 1, batchNumber, "only batch 1 should have run") + + // Shut down the worker after the first batch to prevent subsequent ones. + if batchNumber >= 1 { + go worker.Stop() + } else { + incrementBatchNumber(t, th, job) + } + + return false + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForJobStatus(t, job, model.JobStatusPending) + th.WaitForBatchNumber(t, job, 1) + }) + + t.Run("stop after second batch", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + var worker *jobs.BatchWorker + worker, job := createBatchWorker(t, th, func(rctx *request.Context, job *model.Job) bool { + batchNumber := getBatchNumberFromData(t, job.Data) + + require.LessOrEqual(t, batchNumber, 2, "only batches 1 and 2 should have run") + + // Shut down the worker after the second batch to prevent subsequent ones. + if batchNumber >= 2 { + go worker.Stop() + } else { + incrementBatchNumber(t, th, job) + } + + return false + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForJobStatus(t, job, model.JobStatusPending) + th.WaitForBatchNumber(t, job, 2) + }) + + t.Run("done after first batch", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + var worker *jobs.BatchWorker + worker, job := createBatchWorker(t, th, func(rctx *request.Context, job *model.Job) bool { + batchNumber := getBatchNumberFromData(t, job.Data) + require.Equal(t, 1, batchNumber, "only batch 1 should have run") + + if batchNumber >= 1 { + go worker.Stop() // Shut down the worker when the job is done + return true + } + + incrementBatchNumber(t, th, job) + return false + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForBatchNumber(t, job, 1) + }) + + t.Run("done after three batches", func(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + var worker *jobs.BatchWorker + worker, job := createBatchWorker(t, th, func(rctx *request.Context, job *model.Job) bool { + batchNumber := getBatchNumberFromData(t, job.Data) + require.LessOrEqual(t, batchNumber, 3, "only 3 batches should have run") + + if batchNumber >= 3 { + go worker.Stop() // Shut down the worker when the job is done + return true + } + + incrementBatchNumber(t, th, job) + return false + }) + + // Queue the work to be done + worker.JobChannel() <- *job + + th.WaitForBatchNumber(t, job, 3) + }) +} diff --git a/server/channels/jobs/export_users_to_csv/export_users_to_csv.go b/server/channels/jobs/export_users_to_csv/export_users_to_csv.go new file mode 100644 index 0000000000..f5dd4dc088 --- /dev/null +++ b/server/channels/jobs/export_users_to_csv/export_users_to_csv.go @@ -0,0 +1,107 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package export_users_to_csv + +import ( + "strconv" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/v8/channels/jobs" + "github.com/mattermost/mattermost/server/v8/channels/store" + "github.com/pkg/errors" +) + +const ( + timeBetweenBatches = 1 * time.Second +) + +type ExportUsersToCSVAppIFace interface { + jobs.BatchReportWorkerAppIFace + GetUsersForReporting(filter *model.UserReportOptions) ([]*model.UserReport, *model.AppError) +} + +// MakeWorker creates a batch report worker to generate CSV user reports. +func MakeWorker(jobServer *jobs.JobServer, store store.Store, app ExportUsersToCSVAppIFace) model.Worker { + return jobs.MakeBatchReportWorker( + jobServer, + store, + app, + timeBetweenBatches, + "csv", + []string{ + "Id", + "Username", + "Email", + "CreateAt", + "Name", + "Roles", + "LastLogin", + "LastStatusAt", + "LastPostDate", + "DaysActive", + "TotalPosts", + }, + getData(app), + ) +} + +// parseJobMetadata parses the opaque job metadata to return the information needed to decide which +// batch to process next. +func parseJobMetadata(data model.StringMap) (*model.UserReportOptions, error) { + startAt, err := strconv.ParseInt(data["start_at"], 10, 64) + if err != nil { + return nil, err + } + endAt, err := strconv.ParseInt(data["end_at"], 10, 64) + if err != nil { + return nil, err + } + + options := model.UserReportOptions{ + ReportingBaseOptions: model.ReportingBaseOptions{ + SortColumn: "Username", + PageSize: 100, + FromColumnValue: data["last_column_value"], + FromId: data["last_user_id"], + StartAt: startAt, + EndAt: endAt, + }, + } + + return &options, nil +} + +// makeJobMetadata encodes the information needed to decide which batch to process next back into +// the opaque job metadata. +func makeJobMetadata(jobData model.StringMap, lastColumnValue string, userID string) model.StringMap { + jobData["last_column_value"] = lastColumnValue + jobData["last_user_id"] = userID + return jobData +} + +func getData(app ExportUsersToCSVAppIFace) func(jobData model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error) { + return func(jobData model.StringMap) ([]model.ReportableObject, model.StringMap, bool, error) { + filter, err := parseJobMetadata(jobData) + if err != nil { + return nil, nil, false, errors.Wrap(err, "failed to parse job metadata") + } + + users, appErr := app.GetUsersForReporting(filter) + if appErr != nil { + return nil, nil, false, errors.Wrapf(err, "failed to get the next batch (column_value=%v, user_id=%v)", filter.FromColumnValue, filter.FromId) + } + + if len(users) == 0 { + return nil, nil, true, nil + } + + reportableObjects := []model.ReportableObject{} + for i := 0; i < len(users); i++ { + reportableObjects = append(reportableObjects, users[i]) + } + + return reportableObjects, makeJobMetadata(jobData, users[len(users)-1].Username, users[len(users)-1].Id), false, nil + } +} diff --git a/server/channels/jobs/helper_test.go b/server/channels/jobs/helper_test.go index b7f0f167f8..f44c78ba5e 100644 --- a/server/channels/jobs/helper_test.go +++ b/server/channels/jobs/helper_test.go @@ -6,6 +6,7 @@ package jobs_test import ( "os" "path/filepath" + "strconv" "sync" "testing" "time" @@ -14,8 +15,10 @@ import ( "github.com/mattermost/mattermost/server/public/shared/mlog" "github.com/mattermost/mattermost/server/public/shared/request" "github.com/mattermost/mattermost/server/v8/channels/app" + "github.com/mattermost/mattermost/server/v8/channels/jobs" "github.com/mattermost/mattermost/server/v8/channels/store" "github.com/mattermost/mattermost/server/v8/config" + "github.com/stretchr/testify/require" ) type TestHelper struct { @@ -221,3 +224,73 @@ func (th *TestHelper) TearDown() { os.RemoveAll(th.tempWorkspace) } } + +func (th *TestHelper) SetupBatchWorker(t *testing.T, worker *jobs.BatchWorker) *model.Job { + t.Helper() + + jobId := model.NewId() + th.Server.Jobs.RegisterJobType(jobId, worker, nil) + + jobData := make(model.StringMap) + jobData["batch_number"] = "1" + job, appErr := th.Server.Jobs.CreateJob(th.Context, jobId, jobData) + + if appErr != nil { + panic(appErr) + } + + done := make(chan bool) + go func() { + defer close(done) + worker.Run() + }() + + // When ending the test, ensure we wait for the worker to finish. + t.Cleanup(func() { + waitDone(t, done, "worker did not stop running") + }) + + // Give the worker time to start running + time.Sleep(500 * time.Millisecond) + + return job +} + +func (th *TestHelper) WaitForJobStatus(t *testing.T, job *model.Job, status string) { + t.Helper() + + require.Eventuallyf(t, func() bool { + actualJob, appErr := th.Server.Jobs.GetJob(th.Context, job.Id) + require.Nil(t, appErr) + require.Equal(t, job.Id, actualJob.Id) + + return actualJob.Status == status + }, 5*time.Second, 250*time.Millisecond, "job never transitioned to %s", status) +} + +func (th *TestHelper) WaitForBatchNumber(t *testing.T, job *model.Job, batchNumber int) { + t.Helper() + + require.Eventuallyf(t, func() bool { + actualJob, appErr := th.Server.Jobs.GetJob(th.Context, job.Id) + require.Nil(t, appErr) + require.Equal(t, job.Id, actualJob.Id) + + finalBatchNumber, err := strconv.Atoi(actualJob.Data["batch_number"]) + require.NoError(t, err) + return finalBatchNumber == batchNumber + }, 5*time.Second, 250*time.Millisecond, "job did not stop at batch %d", batchNumber) +} + +func waitDone(t *testing.T, done chan bool, msg string) { + t.Helper() + + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, 5*time.Second, 100*time.Millisecond, msg) +} diff --git a/server/i18n/en.json b/server/i18n/en.json index d87f4beeb5..e774ae6843 100644 --- a/server/i18n/en.json +++ b/server/i18n/en.json @@ -2706,6 +2706,14 @@ "id": "api.restricted_system_admin", "translation": "This action is forbidden to a restricted system admin." }, + { + "id": "api.retrieveBatchReportFile.invalid_format", + "translation": "Report format is invalid." + }, + { + "id": "api.retrieveBatchReportFile.invalid_report_id", + "translation": "Report ID is invalid." + }, { "id": "api.roles.get_multiple_by_name_too_many.request_error", "translation": "Unable to get that many roles by name. Only {{.MaxNames}} roles can be requested at once." @@ -5270,6 +5278,14 @@ "id": "app.command_webhook.try_use.invalid", "translation": "Invalid webhook." }, + { + "id": "app.compile_csv_chunks.header_error", + "translation": "Failed to write CSV headers." + }, + { + "id": "app.compile_report_chunks.unsupported_format", + "translation": "Unsupported report format." + }, { "id": "app.compliance.get.finding.app_error", "translation": "We encountered an error retrieving the compliance reports." @@ -6678,6 +6694,26 @@ "id": "app.report.get_user_report.store_error", "translation": "Failed to fetch user report." }, + { + "id": "app.report.retrieve_batch_report.license_error", + "translation": "Batch reporting export only available to Pro and Enterprise." + }, + { + "id": "app.report.send_report_to_user.export_finished", + "translation": "Report processing is finished. You can download the report [here]({{.Link}})" + }, + { + "id": "app.report.start_users_batch_export.job_exists", + "translation": "Job already exists for this user and date range." + }, + { + "id": "app.report.start_users_batch_export.license_error", + "translation": "Batch reporting export only available to Pro and Enterprise." + }, + { + "id": "app.report.start_users_batch_export.started_export", + "translation": "You have requested the export of user data. A CSV will be delivered to you when the export is complete." + }, { "id": "app.role.check_roles_exist.role_not_found", "translation": "The provided role does not exist" @@ -6718,6 +6754,14 @@ "id": "app.save_config.plugin_hook_error", "translation": "An error occurred running the plugin hook on configuration save." }, + { + "id": "app.save_csv_chunk.write_error", + "translation": "Failed to write CSV chunk." + }, + { + "id": "app.save_report_chunk.unsupported_format", + "translation": "Unsupported report format." + }, { "id": "app.scheme.delete.app_error", "translation": "Unable to delete this scheme." diff --git a/server/public/model/job.go b/server/public/model/job.go index a592e7b61d..9c91cd8d80 100644 --- a/server/public/model/job.go +++ b/server/public/model/job.go @@ -38,6 +38,7 @@ const ( JobTypeCleanupDesktopTokens = "cleanup_desktop_tokens" JobTypeDeleteEmptyDraftsMigration = "delete_empty_drafts_migration" JobTypeRefreshPostStats = "refresh_post_stats" + JobTypeExportUsersToCSV = "export_users_to_csv" JobStatusPending = "pending" JobStatusInProgress = "in_progress" diff --git a/server/public/model/post.go b/server/public/model/post.go index 4eb9563e22..a6b3dc6d23 100644 --- a/server/public/model/post.go +++ b/server/public/model/post.go @@ -52,6 +52,7 @@ const ( PostTypeMe = "me" PostCustomTypePrefix = "custom_" PostTypeReminder = "reminder" + PostTypeAdminReport = "system_admin_report" PostFileidsMaxRunes = 300 PostFilenamesMaxRunes = 4000 @@ -450,7 +451,8 @@ func (o *Post) IsValid(maxPostSize int) *AppError { PostTypeReminder, PostTypeMe, PostTypeWrangler, - PostTypeGMConvertedToChannel: + PostTypeGMConvertedToChannel, + PostTypeAdminReport: default: if !strings.HasPrefix(o.Type, PostCustomTypePrefix) { return NewAppError("Post.IsValid", "model.post.is_valid.type.app_error", nil, "id="+o.Type, http.StatusBadRequest) diff --git a/server/public/model/report.go b/server/public/model/report.go index 86a06997ed..c65453f999 100644 --- a/server/public/model/report.go +++ b/server/public/model/report.go @@ -5,6 +5,7 @@ package model import ( "net/http" + "strconv" "time" pUtils "github.com/mattermost/mattermost/server/public/utils" @@ -19,9 +20,15 @@ const ( ) var ( + ReportExportFormats = []string{"csv"} + UserReportSortColumns = []string{"CreateAt", "Username", "FirstName", "LastName", "Nickname", "Email", "Roles"} ) +type ReportableObject interface { + ToReport() []string +} + type ReportingBaseOptions struct { SortDesc bool Direction string // Accepts only "prev" or "next" @@ -34,20 +41,26 @@ type ReportingBaseOptions struct { EndAt int64 } -func (options *ReportingBaseOptions) PopulateDateRange(now time.Time) { +func GetReportDateRange(dateRange string, now time.Time) (int64, int64) { startAt := int64(0) endAt := int64(0) - if options.DateRange == ReportDurationLast30Days { + if dateRange == ReportDurationLast30Days { startAt = now.AddDate(0, 0, -30).UnixMilli() - } else if options.DateRange == ReportDurationPreviousMonth { + } else if dateRange == ReportDurationPreviousMonth { startOfMonth := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.Local) startAt = startOfMonth.AddDate(0, -1, 0).UnixMilli() endAt = startOfMonth.UnixMilli() - } else if options.DateRange == ReportDurationLast6Months { + } else if dateRange == ReportDurationLast6Months { startAt = now.AddDate(0, -6, -0).UnixMilli() } + return startAt, endAt +} + +func (options *ReportingBaseOptions) PopulateDateRange(now time.Time) { + startAt, endAt := GetReportDateRange(options.DateRange, now) + options.StartAt = startAt options.EndAt = endAt } @@ -70,6 +83,43 @@ type UserReport struct { UserPostStats } +func (u *UserReport) ToReport() []string { + lastStatusAt := "" + if u.LastStatusAt != nil { + lastStatusAt = time.UnixMilli(*u.LastStatusAt).String() + } + lastPostDate := "" + if u.LastPostDate != nil { + lastPostDate = time.UnixMilli(*u.LastPostDate).String() + } + daysActive := "" + if u.DaysActive != nil { + daysActive = strconv.Itoa(*u.DaysActive) + } + totalPosts := "" + if u.TotalPosts != nil { + totalPosts = strconv.Itoa(*u.TotalPosts) + } + lastLogin := "" + if u.LastLogin > 0 { + lastLogin = time.UnixMilli(u.LastLogin).String() + } + + return []string{ + u.Id, + u.Username, + u.Email, + time.UnixMilli(u.CreateAt).String(), + u.User.GetDisplayName(ShowNicknameFullName), + u.Roles, + lastLogin, + lastStatusAt, + lastPostDate, + daysActive, + totalPosts, + } +} + type UserReportOptions struct { ReportingBaseOptions Role string @@ -99,3 +149,13 @@ func (u *UserReportQuery) ToReport() *UserReport { UserPostStats: u.UserPostStats, } } + +func IsValidReportExportFormat(format string) bool { + for _, fmt := range ReportExportFormats { + if format == fmt { + return true + } + } + + return false +} diff --git a/server/public/model/user.go b/server/public/model/user.go index d29d282439..f9cdcb7b57 100644 --- a/server/public/model/user.go +++ b/server/public/model/user.go @@ -1015,7 +1015,6 @@ func (u *User) EmailDomain() string { } type UserPostStats struct { - LastLogin int64 `json:"last_login_at,omitempty"` LastStatusAt *int64 `json:"last_status_at,omitempty"` LastPostDate *int64 `json:"last_post_date,omitempty"` DaysActive *int `json:"days_active,omitempty"` diff --git a/server/public/model/user_serial_gen.go b/server/public/model/user_serial_gen.go index e828f29b67..e33b4e2f60 100644 --- a/server/public/model/user_serial_gen.go +++ b/server/public/model/user_serial_gen.go @@ -875,12 +875,6 @@ func (z *UserPostStats) DecodeMsg(dc *msgp.Reader) (err error) { return } switch msgp.UnsafeString(field) { - case "LastLogin": - z.LastLogin, err = dc.ReadInt64() - if err != nil { - err = msgp.WrapError(err, "LastLogin") - return - } case "LastStatusAt": if dc.IsNil() { err = dc.ReadNil() @@ -966,19 +960,9 @@ func (z *UserPostStats) DecodeMsg(dc *msgp.Reader) (err error) { // EncodeMsg implements msgp.Encodable func (z *UserPostStats) EncodeMsg(en *msgp.Writer) (err error) { - // map header, size 5 - // write "LastLogin" - err = en.Append(0x85, 0xa9, 0x4c, 0x61, 0x73, 0x74, 0x4c, 0x6f, 0x67, 0x69, 0x6e) - if err != nil { - return - } - err = en.WriteInt64(z.LastLogin) - if err != nil { - err = msgp.WrapError(err, "LastLogin") - return - } + // map header, size 4 // write "LastStatusAt" - err = en.Append(0xac, 0x4c, 0x61, 0x73, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x41, 0x74) + err = en.Append(0x84, 0xac, 0x4c, 0x61, 0x73, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x41, 0x74) if err != nil { return } @@ -1051,12 +1035,9 @@ func (z *UserPostStats) EncodeMsg(en *msgp.Writer) (err error) { // MarshalMsg implements msgp.Marshaler func (z *UserPostStats) MarshalMsg(b []byte) (o []byte, err error) { o = msgp.Require(b, z.Msgsize()) - // map header, size 5 - // string "LastLogin" - o = append(o, 0x85, 0xa9, 0x4c, 0x61, 0x73, 0x74, 0x4c, 0x6f, 0x67, 0x69, 0x6e) - o = msgp.AppendInt64(o, z.LastLogin) + // map header, size 4 // string "LastStatusAt" - o = append(o, 0xac, 0x4c, 0x61, 0x73, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x41, 0x74) + o = append(o, 0x84, 0xac, 0x4c, 0x61, 0x73, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x41, 0x74) if z.LastStatusAt == nil { o = msgp.AppendNil(o) } else { @@ -1104,12 +1085,6 @@ func (z *UserPostStats) UnmarshalMsg(bts []byte) (o []byte, err error) { return } switch msgp.UnsafeString(field) { - case "LastLogin": - z.LastLogin, bts, err = msgp.ReadInt64Bytes(bts) - if err != nil { - err = msgp.WrapError(err, "LastLogin") - return - } case "LastStatusAt": if msgp.IsNil(bts) { bts, err = msgp.ReadNilBytes(bts) @@ -1192,7 +1167,7 @@ func (z *UserPostStats) UnmarshalMsg(bts []byte) (o []byte, err error) { // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message func (z *UserPostStats) Msgsize() (s int) { - s = 1 + 10 + msgp.Int64Size + 13 + s = 1 + 13 if z.LastStatusAt == nil { s += msgp.NilSize } else { diff --git a/webapp/platform/client/src/client4.ts b/webapp/platform/client/src/client4.ts index 5a8e875d86..ae533cfac8 100644 --- a/webapp/platform/client/src/client4.ts +++ b/webapp/platform/client/src/client4.ts @@ -1015,6 +1015,14 @@ export default class Client4 { ); } + startUsersBatchExport = (dateRange: string) => { + const queryString = buildQueryString({date_range: dateRange}); + return this.doFetch( + `${this.getReportsRoute()}/users/export${queryString}`, + {method: 'post'}, + ); + } + /** * @deprecated */