diff --git a/app/app_iface.go b/app/app_iface.go index f10ca2d8b9..8c10cb16fc 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -435,7 +435,7 @@ type AppIface interface { BuildPostReactions(ctx request.CTX, postID string) (*[]ReactionImportData, *model.AppError) BuildPushNotificationMessage(c request.CTX, contentsConfig string, post *model.Post, user *model.User, channel *model.Channel, channelName string, senderName string, explicitMention bool, channelWideMention bool, replyToThreadType string) (*model.PushNotification, *model.AppError) BuildSamlMetadataObject(idpMetadata []byte) (*model.SamlMetadataResponse, *model.AppError) - BulkExport(ctx request.CTX, writer io.Writer, outPath string, opts model.BulkExportOpts) *model.AppError + BulkExport(ctx request.CTX, writer io.Writer, outPath string, job *model.Job, opts model.BulkExportOpts) *model.AppError BulkImport(c *request.Context, jsonlReader io.Reader, attachmentsReader *zip.Reader, dryRun bool, workers int) (*model.AppError, int) BulkImportWithPath(c *request.Context, jsonlReader io.Reader, attachmentsReader *zip.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) CanNotifyAdmin(trial bool) bool diff --git a/app/export.go b/app/export.go index 8e52fa74e7..ee427b6936 100644 --- a/app/export.go +++ b/app/export.go @@ -12,6 +12,7 @@ import ( "net/http" "os" "path/filepath" + "strconv" "strings" "time" @@ -65,7 +66,7 @@ var exportablePreferences = map[imports.ComparablePreference]string{{ }: "EmailInterval", } -func (a *App) BulkExport(ctx request.CTX, writer io.Writer, outPath string, opts model.BulkExportOpts) *model.AppError { +func (a *App) BulkExport(ctx request.CTX, writer io.Writer, outPath string, job *model.Job, opts model.BulkExportOpts) *model.AppError { var zipWr *zip.Writer if opts.CreateArchive { var err error @@ -78,46 +79,50 @@ func (a *App) BulkExport(ctx request.CTX, writer io.Writer, outPath string, opts } } + if job != nil && job.Data == nil { + job.Data = make(model.StringMap) + } + ctx.Logger().Info("Bulk export: exporting version") if err := a.exportVersion(writer); err != nil { return err } ctx.Logger().Info("Bulk export: exporting teams") - teamNames, err := a.exportAllTeams(writer) + teamNames, err := a.exportAllTeams(ctx, job, writer) if err != nil { return err } ctx.Logger().Info("Bulk export: exporting channels") - if err = a.exportAllChannels(writer, teamNames); err != nil { + if err = a.exportAllChannels(ctx, job, writer, teamNames); err != nil { return err } ctx.Logger().Info("Bulk export: exporting users") - if err = a.exportAllUsers(writer); err != nil { + if err = a.exportAllUsers(ctx, job, writer); err != nil { return err } ctx.Logger().Info("Bulk export: exporting posts") - attachments, err := a.exportAllPosts(ctx, writer, opts.IncludeAttachments) + attachments, err := a.exportAllPosts(ctx, job, writer, opts.IncludeAttachments) if err != nil { return err } ctx.Logger().Info("Bulk export: exporting emoji") - emojiPaths, err := a.exportCustomEmoji(ctx, writer, outPath, "exported_emoji", !opts.CreateArchive) + emojiPaths, err := a.exportCustomEmoji(ctx, job, writer, outPath, "exported_emoji", !opts.CreateArchive) if err != nil { return err } ctx.Logger().Info("Bulk export: exporting direct channels") - if err = a.exportAllDirectChannels(writer); err != nil { + if err = a.exportAllDirectChannels(ctx, job, writer); err != nil { return err } ctx.Logger().Info("Bulk export: exporting direct posts") - directAttachments, err := a.exportAllDirectPosts(ctx, writer, opts.IncludeAttachments) + directAttachments, err := a.exportAllDirectPosts(ctx, job, writer, opts.IncludeAttachments) if err != nil { return err } @@ -139,6 +144,8 @@ func (a *App) BulkExport(ctx request.CTX, writer io.Writer, outPath string, opts return err } } + + updateJobProgress(ctx.Logger(), a.Srv().Store(), job, "attachments_exported", len(attachments)+len(directAttachments)+len(emojiPaths)) } return nil @@ -175,9 +182,10 @@ func (a *App) exportVersion(writer io.Writer) *model.AppError { return a.exportWriteLine(writer, versionLine) } -func (a *App) exportAllTeams(writer io.Writer) (map[string]bool, *model.AppError) { +func (a *App) exportAllTeams(ctx request.CTX, job *model.Job, writer io.Writer) (map[string]bool, *model.AppError) { afterId := strings.Repeat("0", 26) teamNames := make(map[string]bool) + cnt := 0 for { teams, err := a.Srv().Store().Team().GetAllForExportAfter(1000, afterId) if err != nil { @@ -187,6 +195,8 @@ func (a *App) exportAllTeams(writer io.Writer) (map[string]bool, *model.AppError if len(teams) == 0 { break } + cnt += len(teams) + updateJobProgress(ctx.Logger(), a.Srv().Store(), job, "teams_exported", cnt) for _, team := range teams { afterId = team.Id @@ -207,8 +217,9 @@ func (a *App) exportAllTeams(writer io.Writer) (map[string]bool, *model.AppError return teamNames, nil } -func (a *App) exportAllChannels(writer io.Writer, teamNames map[string]bool) *model.AppError { +func (a *App) exportAllChannels(ctx request.CTX, job *model.Job, writer io.Writer, teamNames map[string]bool) *model.AppError { afterId := strings.Repeat("0", 26) + cnt := 0 for { channels, err := a.Srv().Store().Channel().GetAllChannelsForExportAfter(1000, afterId) @@ -219,6 +230,8 @@ func (a *App) exportAllChannels(writer io.Writer, teamNames map[string]bool) *mo if len(channels) == 0 { break } + cnt += len(channels) + updateJobProgress(ctx.Logger(), a.Srv().Store(), job, "channels_exported", cnt) for _, channel := range channels { afterId = channel.Id @@ -242,8 +255,9 @@ func (a *App) exportAllChannels(writer io.Writer, teamNames map[string]bool) *mo return nil } -func (a *App) exportAllUsers(writer io.Writer) *model.AppError { +func (a *App) exportAllUsers(ctx request.CTX, job *model.Job, writer io.Writer) *model.AppError { afterId := strings.Repeat("0", 26) + cnt := 0 for { users, err := a.Srv().Store().User().GetAllAfter(1000, afterId) @@ -254,6 +268,8 @@ func (a *App) exportAllUsers(writer io.Writer) *model.AppError { if len(users) == 0 { break } + cnt += len(users) + updateJobProgress(ctx.Logger(), a.Srv().Store(), job, "users_exported", cnt) for _, user := range users { afterId = user.Id @@ -395,12 +411,13 @@ func (a *App) buildUserNotifyProps(notifyProps model.StringMap) *imports.UserNot } } -func (a *App) exportAllPosts(ctx request.CTX, writer io.Writer, withAttachments bool) ([]imports.AttachmentImportData, *model.AppError) { +func (a *App) exportAllPosts(ctx request.CTX, job *model.Job, writer io.Writer, withAttachments bool) ([]imports.AttachmentImportData, *model.AppError) { var attachments []imports.AttachmentImportData afterId := strings.Repeat("0", 26) var postProcessCount uint64 logCheckpoint := time.Now() + cnt := 0 for { if time.Since(logCheckpoint) > 5*time.Minute { ctx.Logger().Debug(fmt.Sprintf("Bulk Export: processed %d posts", postProcessCount)) @@ -415,6 +432,8 @@ func (a *App) exportAllPosts(ctx request.CTX, writer io.Writer, withAttachments if len(posts) == 0 { return attachments, nil } + cnt += len(posts) + updateJobProgress(ctx.Logger(), a.Srv().Store(), job, "posts_exported", cnt) for _, post := range posts { afterId = post.Id @@ -538,9 +557,10 @@ func (a *App) buildPostAttachments(postID string) ([]imports.AttachmentImportDat return attachments, nil } -func (a *App) exportCustomEmoji(c request.CTX, writer io.Writer, outPath, exportDir string, exportFiles bool) ([]string, *model.AppError) { +func (a *App) exportCustomEmoji(c request.CTX, job *model.Job, writer io.Writer, outPath, exportDir string, exportFiles bool) ([]string, *model.AppError) { var emojiPaths []string pageNumber := 0 + cnt := 0 for { customEmojiList, err := a.GetEmojiList(c, pageNumber, 100, model.EmojiSortByName) @@ -551,6 +571,8 @@ func (a *App) exportCustomEmoji(c request.CTX, writer io.Writer, outPath, export if len(customEmojiList) == 0 { break } + cnt += len(customEmojiList) + updateJobProgress(c.Logger(), a.Srv().Store(), job, "emojis_exported", cnt) pageNumber++ @@ -619,8 +641,9 @@ func (a *App) copyEmojiImages(emojiId string, emojiImagePath string, pathToDir s return nil } -func (a *App) exportAllDirectChannels(writer io.Writer) *model.AppError { +func (a *App) exportAllDirectChannels(ctx request.CTX, job *model.Job, writer io.Writer) *model.AppError { afterId := strings.Repeat("0", 26) + cnt := 0 for { channels, err := a.Srv().Store().Channel().GetAllDirectChannelsForExportAfter(1000, afterId) if err != nil { @@ -630,6 +653,8 @@ func (a *App) exportAllDirectChannels(writer io.Writer) *model.AppError { if len(channels) == 0 { break } + cnt += len(channels) + updateJobProgress(ctx.Logger(), a.Srv().Store(), job, "direct_channels_exported", cnt) for _, channel := range channels { afterId = channel.Id @@ -682,12 +707,13 @@ func (a *App) buildFavoritedByList(channelID string) ([]string, *model.AppError) return userIDs, nil } -func (a *App) exportAllDirectPosts(ctx request.CTX, writer io.Writer, withAttachments bool) ([]imports.AttachmentImportData, *model.AppError) { +func (a *App) exportAllDirectPosts(ctx request.CTX, job *model.Job, writer io.Writer, withAttachments bool) ([]imports.AttachmentImportData, *model.AppError) { var attachments []imports.AttachmentImportData afterId := strings.Repeat("0", 26) var postProcessCount uint64 logCheckpoint := time.Now() + cnt := 0 for { if time.Since(logCheckpoint) > 5*time.Minute { ctx.Logger().Debug(fmt.Sprintf("Bulk Export: processed %d direct posts", postProcessCount)) @@ -702,6 +728,8 @@ func (a *App) exportAllDirectPosts(ctx request.CTX, writer io.Writer, withAttach if len(posts) == 0 { break } + cnt += len(posts) + updateJobProgress(ctx.Logger(), a.Srv().Store(), job, "direct_posts_exported", cnt) for _, post := range posts { afterId = post.Id @@ -815,3 +843,12 @@ func (a *App) DeleteExport(name string) *model.AppError { return a.RemoveFile(filePath) } + +func updateJobProgress(logger mlog.LoggerIFace, store store.Store, job *model.Job, key string, value int) { + if job != nil { + job.Data[key] = strconv.Itoa(value) + if _, err2 := store.Job().UpdateOptimistically(job, model.JobStatusInProgress); err2 != nil { + logger.Warn("Failed to update job status", mlog.Err(err2)) + } + } +} diff --git a/app/export_test.go b/app/export_test.go index 7dbaa984b7..e839e1dc13 100644 --- a/app/export_test.go +++ b/app/export_test.go @@ -164,7 +164,7 @@ func TestExportCustomEmoji(t *testing.T) { outPath, err := filepath.Abs(filePath) require.NoError(t, err) - _, appErr := th.App.exportCustomEmoji(th.Context, fileWriter, outPath, dirNameToExportEmoji, false) + _, appErr := th.App.exportCustomEmoji(th.Context, nil, fileWriter, outPath, dirNameToExportEmoji, false) require.Nil(t, appErr, "should not have failed") } @@ -178,7 +178,7 @@ func TestExportAllUsers(t *testing.T) { require.Nil(t, err) var b bytes.Buffer - err = th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err = th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) th2 := Setup(t) @@ -235,7 +235,7 @@ func TestExportDMChannel(t *testing.T) { }) var b bytes.Buffer - err := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) channels, nErr := th1.App.Srv().Store().Channel().GetAllDirectChannelsForExportAfter(1000, "00000000") @@ -282,7 +282,7 @@ func TestExportDMChannel(t *testing.T) { th1.App.PermanentDeleteUser(th1.Context, th1.BasicUser) var b bytes.Buffer - err := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) th2 := Setup(t).InitBasic() @@ -306,7 +306,7 @@ func TestExportDMChannelToSelf(t *testing.T) { th1.CreateDmChannel(th1.BasicUser) var b bytes.Buffer - err := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) channels, nErr := th1.App.Srv().Store().Channel().GetAllDirectChannelsForExportAfter(1000, "00000000") @@ -344,7 +344,7 @@ func TestExportGMChannel(t *testing.T) { th1.CreateGroupChannel(th1.Context, user1, user2) var b bytes.Buffer - err := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) channels, nErr := th1.App.Srv().Store().Channel().GetAllDirectChannelsForExportAfter(1000, "00000000") @@ -376,7 +376,7 @@ func TestExportGMandDMChannels(t *testing.T) { th1.CreateGroupChannel(th1.Context, user1, user2) var b bytes.Buffer - err := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) channels, nErr := th1.App.Srv().Store().Channel().GetAllDirectChannelsForExportAfter(1000, "00000000") @@ -459,7 +459,7 @@ func TestExportDMandGMPost(t *testing.T) { assert.Equal(t, 4, len(posts)) var b bytes.Buffer - appErr := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + appErr := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, appErr) th1.TearDown() @@ -534,7 +534,7 @@ func TestExportPostWithProps(t *testing.T) { require.NotEmpty(t, posts[1].Props) var b bytes.Buffer - appErr := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + appErr := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, appErr) th1.TearDown() @@ -572,7 +572,7 @@ func TestExportDMPostWithSelf(t *testing.T) { th1.CreatePost(dmChannel) var b bytes.Buffer - err := th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err := th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) posts, nErr := th1.App.Srv().Store().Post().GetDirectPostParentsForExportAfter(1000, "0000000") @@ -640,7 +640,7 @@ func TestBulkExport(t *testing.T) { IncludeAttachments: true, CreateArchive: true, } - appErr = th.App.BulkExport(th.Context, exportFile, dir, opts) + appErr = th.App.BulkExport(th.Context, exportFile, dir, nil, opts) require.Nil(t, appErr) th.TearDown() @@ -731,7 +731,7 @@ func TestExportDeletedTeams(t *testing.T) { require.Nil(t, err) var b bytes.Buffer - err = th1.App.BulkExport(th1.Context, &b, "somePath", model.BulkExportOpts{}) + err = th1.App.BulkExport(th1.Context, &b, "somePath", nil, model.BulkExportOpts{}) require.Nil(t, err) th2 := Setup(t) diff --git a/app/opentracing/opentracing_layer.go b/app/opentracing/opentracing_layer.go index 367d98ec39..608d007d7d 100644 --- a/app/opentracing/opentracing_layer.go +++ b/app/opentracing/opentracing_layer.go @@ -974,7 +974,7 @@ func (a *OpenTracingAppLayer) BuildSamlMetadataObject(idpMetadata []byte) (*mode return resultVar0, resultVar1 } -func (a *OpenTracingAppLayer) BulkExport(ctx request.CTX, writer io.Writer, outPath string, opts model.BulkExportOpts) *model.AppError { +func (a *OpenTracingAppLayer) BulkExport(ctx request.CTX, writer io.Writer, outPath string, job *model.Job, opts model.BulkExportOpts) *model.AppError { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.BulkExport") @@ -986,7 +986,7 @@ func (a *OpenTracingAppLayer) BulkExport(ctx request.CTX, writer io.Writer, outP }() defer span.Finish() - resultVar0 := a.app.BulkExport(ctx, writer, outPath, opts) + resultVar0 := a.app.BulkExport(ctx, writer, outPath, job, opts) if resultVar0 != nil { span.LogFields(spanlog.Error(resultVar0)) diff --git a/cmd/mattermost/commands/export.go b/cmd/mattermost/commands/export.go index d5573ee975..21ff04162e 100644 --- a/cmd/mattermost/commands/export.go +++ b/cmd/mattermost/commands/export.go @@ -240,7 +240,7 @@ func bulkExportCmdF(command *cobra.Command, args []string) error { var opts model.BulkExportOpts opts.IncludeAttachments = attachments opts.CreateArchive = archive - if err := a.BulkExport(request.EmptyContext(a.Log()), fileWriter, filepath.Dir(outPath), opts); err != nil { + if err := a.BulkExport(request.EmptyContext(a.Log()), fileWriter, filepath.Dir(outPath), nil /* nil job since it's spawned from CLI */, opts); err != nil { CommandPrintErrorln(err.Error()) return err } diff --git a/jobs/base_workers.go b/jobs/base_workers.go index 58bdc98ba7..71c360690f 100644 --- a/jobs/base_workers.go +++ b/jobs/base_workers.go @@ -78,6 +78,14 @@ func (worker *SimpleWorker) DoJob(job *model.Job) { return } + var appErr *model.AppError + // We get the job again because ClaimJob changes the job status. + job, appErr = worker.jobServer.GetJob(job.Id) + if appErr != nil { + mlog.Error("SimpleWorker: job execution error", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(appErr)) + worker.setJobError(job, appErr) + } + err := worker.execute(job) if err != nil { mlog.Error("SimpleWorker: job execution error", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) @@ -90,8 +98,13 @@ func (worker *SimpleWorker) DoJob(job *model.Job) { } func (worker *SimpleWorker) setJobSuccess(job *model.Job) { + if err := worker.jobServer.SetJobProgress(job, 100); err != nil { + mlog.Error("Worker: Failed to update progress for job", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) + worker.setJobError(job, err) + } + if err := worker.jobServer.SetJobSuccess(job); err != nil { - mlog.Error("SimpleWorker: Failed to set success for job", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error())) + mlog.Error("SimpleWorker: Failed to set success for job", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) worker.setJobError(job, err) } } diff --git a/jobs/export_process/worker.go b/jobs/export_process/worker.go index 7f80ac1950..48605b8fb6 100644 --- a/jobs/export_process/worker.go +++ b/jobs/export_process/worker.go @@ -21,7 +21,7 @@ type AppIface interface { configservice.ConfigService WriteFile(fr io.Reader, path string) (int64, *model.AppError) WriteFileContext(ctx context.Context, fr io.Reader, path string) (int64, *model.AppError) - BulkExport(ctx request.CTX, writer io.Writer, outPath string, opts model.BulkExportOpts) *model.AppError + BulkExport(ctx request.CTX, writer io.Writer, outPath string, job *model.Job, opts model.BulkExportOpts) *model.AppError Log() *mlog.Logger } @@ -56,7 +56,7 @@ func MakeWorker(jobServer *jobs.JobServer, app AppIface) model.Worker { }() logger := app.Log().With(mlog.String("job_id", job.Id)) - appErr := app.BulkExport(request.EmptyContext(logger), wr, outPath, opts) + appErr := app.BulkExport(request.EmptyContext(logger), wr, outPath, job, opts) wr.Close() // Close never returns an error if appErr != nil {