MM-55987: Bypass uploading to S3 for local mode mmctl import (#25591)

We directly instruct the server to read from the local filesystem
in case the local_mode key is set.

There is now no need to upload the file in --local mode and a warning
is thrown accordingly.

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

```release-note
Now the mmctl bulk import process command in local mode supports processing an import file without actually uploading it to the server. Simply pass the file path to the import file and the server will directly read from it, and pass the --bypass-upload flag. There is no need to use the import upload command. NOTE: all of this is applicable only in local mode.
```

Co-authored-by: Mattermost Build <build@mattermost.com>

server/channels/api4/import_test.go

@@ -6,12 +6,16 @@ package api4
 import (
 	"context"
 	"os"
+	"path"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
 	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/v8/channels/app"
 	"github.com/mattermost/mattermost/server/v8/channels/utils/fileutils"
 )
 
@@ -105,3 +109,41 @@ func TestListImports(t *testing.T) {
 		require.NoError(t, os.RemoveAll(importDir))
 	}, "change import directory")
 }
+
+func TestImportInLocalMode(t *testing.T) {
+	th := SetupWithServerOptions(t, []app.Option{app.RunEssentialJobs})
+	defer th.TearDown()
+
+	testsDir, _ := fileutils.FindDir("tests")
+	require.NotEmpty(t, testsDir)
+
+	job := &model.Job{
+		Type: model.JobTypeImportProcess,
+		Data: map[string]string{
+			"import_file": path.Join(testsDir, "import_test.zip"),
+			"local_mode":  "true",
+		},
+	}
+
+	received, _, err := th.SystemAdminClient.CreateJob(context.Background(), job)
+	require.NoError(t, err)
+	defer th.App.Srv().Store().Job().Delete(received.Id)
+
+	cnt1, err := th.App.Srv().Store().Post().AnalyticsPostCount(&model.PostCountOptions{UsersPostsOnly: true})
+	require.NoError(t, err)
+
+	var appErr *model.AppError
+	for !(received.Status == model.JobStatusSuccess || received.Status == model.JobStatusError) {
+		received, appErr = th.App.GetJob(th.Context, received.Id)
+		require.Nil(t, appErr)
+		time.Sleep(5 * time.Second)
+		th.Context.Logger().Debug("Job status", mlog.String("status", received.Status))
+	}
+
+	require.Equal(t, model.JobStatusSuccess, received.Status)
+
+	cnt2, err := th.App.Srv().Store().Post().AnalyticsPostCount(&model.PostCountOptions{UsersPostsOnly: true})
+	require.NoError(t, err)
+	// Just a sanity check to ensure new posts are actually added in the system.
+	require.Greater(t, cnt2, cnt1)
+}

server/channels/jobs/import_process/worker.go

@@ -5,8 +5,11 @@ package import_process
 
 import (
 	"archive/zip"
+	"errors"
+	"fmt"
 	"io"
 	"net/http"
+	"os"
 	"path/filepath"
 	"runtime"
 	"strconv"
@@ -45,29 +48,49 @@ func MakeWorker(jobServer *jobs.JobServer, app AppIface) *jobs.SimpleWorker {
 			return model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.missing_file", nil, "", http.StatusBadRequest)
 		}
 
-		importFilePath := filepath.Join(*app.Config().ImportSettings.Directory, importFileName)
-		if ok, err := app.FileExists(importFilePath); err != nil {
-			return err
-		} else if !ok {
-			return model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.file_exists", nil, "", http.StatusBadRequest)
-		}
+		var importFilePath string
+		var importFileSize int64
+		var importFile filestore.ReadCloseSeeker
+		if job.Data["local_mode"] == "true" {
+			// We simply read the file from the local filesystem.
+			info, err := os.Stat(importFileName)
+			if errors.Is(err, os.ErrNotExist) {
+				return fmt.Errorf("file %s doesn't exist.", importFile)
+			}
 
-		importFileSize, appErr := app.FileSize(importFilePath)
-		if appErr != nil {
-			return appErr
-		}
+			importFileSize = info.Size()
 
-		importFile, appErr := app.FileReader(importFilePath)
-		if appErr != nil {
-			return appErr
-		}
-		defer importFile.Close()
+			importFile, err = os.Open(importFileName)
+			if err != nil {
+				return err
+			}
+			defer importFile.Close()
+		} else {
+			importFilePath = filepath.Join(*app.Config().ImportSettings.Directory, importFileName)
+			if ok, err := app.FileExists(importFilePath); err != nil {
+				return err
+			} else if !ok {
+				return model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.file_exists", nil, "", http.StatusBadRequest)
+			}
 
-		// The import is a long running operation, try to cancel any timeouts attached to the reader.
-		type TimeoutCanceler interface{ CancelTimeout() bool }
-		if tc, ok := importFile.(TimeoutCanceler); ok {
-			if !tc.CancelTimeout() {
-				appContext.Logger().Warn("Could not cancel the timeout for the file reader. The import may fail due to a timeout.")
+			var appErr *model.AppError
+			importFileSize, appErr = app.FileSize(importFilePath)
+			if appErr != nil {
+				return appErr
+			}
+
+			importFile, appErr = app.FileReader(importFilePath)
+			if appErr != nil {
+				return appErr
+			}
+			defer importFile.Close()
+
+			// The import is a long running operation, try to cancel any timeouts attached to the reader.
+			type TimeoutCanceler interface{ CancelTimeout() bool }
+			if tc, ok := importFile.(TimeoutCanceler); ok {
+				if !tc.CancelTimeout() {
+					appContext.Logger().Warn("Could not cancel the timeout for the file reader. The import may fail due to a timeout.")
+				}
 			}
 		}
 
@@ -107,9 +130,12 @@ func MakeWorker(jobServer *jobs.JobServer, app AppIface) *jobs.SimpleWorker {
 			return appErr
 		}
 
-		// remove import file when done.
-		if appErr := app.RemoveFile(importFilePath); appErr != nil {
-			return appErr
+		// No need to remove the file in local mode.
+		if job.Data["local_mode"] != "true" {
+			// remove import file when done.
+			if appErr := app.RemoveFile(importFilePath); appErr != nil {
+				return appErr
+			}
 		}
 		return nil
 	}

server/cmd/mmctl/commands/import.go

@@ -9,6 +9,9 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path"
+	"path/filepath"
+	"strconv"
 	"strings"
 	"text/template"
 	"time"
@@ -108,6 +111,8 @@ func init() {
 	ImportValidateCmd.Flags().Bool("ignore-attachments", false, "Don't check if the attached files are present in the archive")
 	ImportValidateCmd.Flags().Bool("check-server-duplicates", true, "Set to false to ignore teams, channels, and users already present on the server")
 
+	ImportProcessCmd.Flags().Bool("bypass-upload", false, "If this is set, the file is not processed from the server, but rather directly read from the filesystem. Works only in --local mode.")
+
 	ImportListCmd.AddCommand(
 		ImportListAvailableCmd,
 		ImportListIncompleteCmd,
@@ -176,6 +181,11 @@ func importListAvailableCmdF(c client.Client, command *cobra.Command, args []str
 func importUploadCmdF(c client.Client, command *cobra.Command, args []string) error {
 	filepath := args[0]
 
+	isLocal, _ := command.Flags().GetBool("local")
+	if isLocal {
+		printer.PrintWarning("In --local mode, you don't need to upload the file to server any more. Directly use the import process command and pass the export file.")
+	}
+
 	file, err := os.Open(filepath)
 	if err != nil {
 		return fmt.Errorf("failed to open import file: %w", err)
@@ -240,10 +250,32 @@ func importUploadCmdF(c client.Client, command *cobra.Command, args []string) er
 func importProcessCmdF(c client.Client, command *cobra.Command, args []string) error {
 	importFile := args[0]
 
+	isLocal, _ := command.Flags().GetBool("local")
+	bypassUpload, _ := command.Flags().GetBool("bypass-upload")
+	if bypassUpload {
+		if isLocal {
+			// in local mode, we tell the server to directly read from this file.
+			if _, err := os.Stat(importFile); errors.Is(err, os.ErrNotExist) {
+				return fmt.Errorf("file %s doesn't exist. NOTE: If this file was uploaded to the server via mmctl import upload, please omit the --bypass-upload flag to revert to old behavior.", importFile)
+			}
+			// If it's not an absolute path, then we make it
+			if !path.IsAbs(importFile) {
+				var err2 error
+				importFile, err2 = filepath.Abs(importFile)
+				if err2 != nil {
+					return fmt.Errorf("error is getting the absolute path to %s: %w", importFile, err2)
+				}
+			}
+		} else {
+			printer.PrintWarning("--bypass-upload has no effect in non-local mode.")
+		}
+	}
+
 	job, _, err := c.CreateJob(context.TODO(), &model.Job{
 		Type: model.JobTypeImportProcess,
 		Data: map[string]string{
 			"import_file": importFile,
+			"local_mode":  strconv.FormatBool(isLocal && bypassUpload),
 		},
 	})
 	if err != nil {

server/cmd/mmctl/commands/import_test.go

@@ -210,7 +210,7 @@ func (s *MmctlUnitTestSuite) TestImportProcessCmdF() {
 	importFile := "import.zip"
 	mockJob := &model.Job{
 		Type: model.JobTypeImportProcess,
-		Data: map[string]string{"import_file": importFile},
+		Data: map[string]string{"import_file": importFile, "local_mode": "false"},
 	}
 
 	s.client.

server/cmd/mmctl/docs/mmctl_import_process.rst

@@ -27,7 +27,8 @@ Options
 
 ::
 
-  -h, --help   help for process
+      --bypass-upload   If this is set, the file is not processed from the server, but rather directly read from the filesystem. Works only in --local mode.
+  -h, --help            help for process
 
 Options inherited from parent commands
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~