MM-49374: Update job metadata for bulk export (#21983)

If a bulk export is triggered via a job, then
we update the metadata field of the job as it
progresses through the records.

https://mattermost.atlassian.net/browse/MM-49374

```release-note
Now you can monitor the progress of the bulk export job
via its metadata field. It is available at
`mmctl export job show <jobID>`.
```

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

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

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))
+		}
+	}
+}

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)

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))

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
 	}

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)
 	}
 }

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 {