From b946dad78de07e4a756ec4fa6bcfac61d34537bd Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Wed, 13 Dec 2023 09:38:00 +0530 Subject: [PATCH] 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 --- server/channels/api4/import_test.go | 42 +++++++++++ server/channels/jobs/import_process/worker.go | 72 +++++++++++++------ server/cmd/mmctl/commands/import.go | 32 +++++++++ server/cmd/mmctl/commands/import_test.go | 2 +- .../cmd/mmctl/docs/mmctl_import_process.rst | 3 +- 5 files changed, 126 insertions(+), 25 deletions(-) diff --git a/server/channels/api4/import_test.go b/server/channels/api4/import_test.go index 664a652e07..1f004c3818 100644 --- a/server/channels/api4/import_test.go +++ b/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) +} diff --git a/server/channels/jobs/import_process/worker.go b/server/channels/jobs/import_process/worker.go index 5c8e6fddfa..f7a0dfc520 100644 --- a/server/channels/jobs/import_process/worker.go +++ b/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 } diff --git a/server/cmd/mmctl/commands/import.go b/server/cmd/mmctl/commands/import.go index 114549ba8b..509bfe0eb0 100644 --- a/server/cmd/mmctl/commands/import.go +++ b/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 { diff --git a/server/cmd/mmctl/commands/import_test.go b/server/cmd/mmctl/commands/import_test.go index 4520ee504a..1549befa73 100644 --- a/server/cmd/mmctl/commands/import_test.go +++ b/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. diff --git a/server/cmd/mmctl/docs/mmctl_import_process.rst b/server/cmd/mmctl/docs/mmctl_import_process.rst index 0465375a26..40b0d99fb5 100644 --- a/server/cmd/mmctl/docs/mmctl_import_process.rst +++ b/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 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~