Improve bulk export logging (#21935)

This change includes the following:
 - Adds the job ID as a log field for bulk exports
 - Adds checkpoint logging for posts and direct posts so that
   progress can be tracked.

app/export.go

@@ -398,8 +398,15 @@ 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) {
 	var attachments []imports.AttachmentImportData
 	afterId := strings.Repeat("0", 26)
+	var postProcessCount uint64
+	logCheckpoint := time.Now()
 
 	for {
+		if time.Since(logCheckpoint) > 5*time.Minute {
+			ctx.Logger().Debug(fmt.Sprintf("Bulk Export: processed %d posts", postProcessCount))
+			logCheckpoint = time.Now()
+		}
+
 		posts, nErr := a.Srv().Store().Post().GetParentsForExportAfter(1000, afterId)
 		if nErr != nil {
 			return nil, model.NewAppError("exportAllPosts", "app.post.get_posts.app_error", nil, "", http.StatusInternalServerError).Wrap(nErr)
@@ -411,6 +418,7 @@ func (a *App) exportAllPosts(ctx request.CTX, writer io.Writer, withAttachments
 
 		for _, post := range posts {
 			afterId = post.Id
+			postProcessCount++
 
 			// Skip deleted.
 			if post.DeleteAt != 0 {
@@ -677,7 +685,15 @@ func (a *App) buildFavoritedByList(channelID string) ([]string, *model.AppError)
 func (a *App) exportAllDirectPosts(ctx request.CTX, writer io.Writer, withAttachments bool) ([]imports.AttachmentImportData, *model.AppError) {
 	var attachments []imports.AttachmentImportData
 	afterId := strings.Repeat("0", 26)
+	var postProcessCount uint64
+	logCheckpoint := time.Now()
+
 	for {
+		if time.Since(logCheckpoint) > 5*time.Minute {
+			ctx.Logger().Debug(fmt.Sprintf("Bulk Export: processed %d direct posts", postProcessCount))
+			logCheckpoint = time.Now()
+		}
+
 		posts, err := a.Srv().Store().Post().GetDirectPostParentsForExportAfter(1000, afterId)
 		if err != nil {
 			return nil, model.NewAppError("exportAllDirectPosts", "app.post.get_direct_posts.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
@@ -689,6 +705,7 @@ func (a *App) exportAllDirectPosts(ctx request.CTX, writer io.Writer, withAttach
 
 		for _, post := range posts {
 			afterId = post.Id
+			postProcessCount++
 
 			// Skip deleted.
 			if post.DeleteAt != 0 {

jobs/export_process/worker.go

@@ -55,7 +55,8 @@ func MakeWorker(jobServer *jobs.JobServer, app AppIface) model.Worker {
 			}
 		}()
 
-		appErr := app.BulkExport(request.EmptyContext(app.Log()), wr, outPath, opts)
+		logger := app.Log().With(mlog.String("job_id", job.Id))
+		appErr := app.BulkExport(request.EmptyContext(logger), wr, outPath, opts)
 		wr.Close() // Close never returns an error
 
 		if appErr != nil {