mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

MM-31612 Optimize bulk import process by avoiding writing zip to disk (#17659)

Просмотр исходного кода
Этот коммит содержится в:
Ian Whitlock
2021-06-09 11:11:46 -05:00
коммит произвёл GitHub
родитель 72c86448b9
Коммит fe94a20ce7
12 изменённых файлов: 281 добавлений и 90 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

5
app/app_iface.go
Просмотреть файл

@@ -7,6 +7,7 @@
package app
import (
"archive/zip"
"bytes"
"context"
"crypto/ecdsa"
@@ -424,8 +425,8 @@ type AppIface interface {
BuildPushNotificationMessage(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(writer io.Writer, outPath string, opts BulkExportOpts) *model.AppError
BulkImport(c *request.Context, fileReader io.Reader, dryRun bool, workers int) (*model.AppError, int)
BulkImportWithPath(c *request.Context, fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int)
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)
CancelJob(jobId string) *model.AppError
ChannelMembersToRemove(teamID *string) ([]*model.ChannelMember, *model.AppError)
CheckAndSendUserLimitWarningEmails(c *request.Context) *model.AppError

20
app/export_test.go
Просмотреть файл

@@ -183,7 +183,7 @@ func TestExportAllUsers(t *testing.T) {
th2 := Setup(t)
defer th2.TearDown()
err, i := th2.App.BulkImport(th2.Context, &b, false, 5)
err, i := th2.App.BulkImport(th2.Context, &b, nil, false, 5)
assert.Nil(t, err)
assert.Equal(t, 0, i)
@@ -241,7 +241,7 @@ func TestExportDMChannel(t *testing.T) {
assert.Equal(t, 0, len(channels))
// import the exported channel
err, i := th2.App.BulkImport(th2.Context, &b, false, 5)
err, i := th2.App.BulkImport(th2.Context, &b, nil, false, 5)
require.Nil(t, err)
assert.Equal(t, 0, i)
@@ -274,7 +274,7 @@ func TestExportDMChannel(t *testing.T) {
defer th2.TearDown()
// import the exported channel
err, _ = th2.App.BulkImport(th2.Context, &b, true, 5)
err, _ = th2.App.BulkImport(th2.Context, &b, nil, true, 5)
require.Nil(t, err)
channels, nErr = th2.App.Srv().Store.Channel().GetAllDirectChannelsForExportAfter(1000, "00000000")
@@ -306,7 +306,7 @@ func TestExportDMChannelToSelf(t *testing.T) {
assert.Equal(t, 0, len(channels))
// import the exported channel
err, i := th2.App.BulkImport(th2.Context, &b, false, 5)
err, i := th2.App.BulkImport(th2.Context, &b, nil, false, 5)
assert.Nil(t, err)
assert.Equal(t, 0, i)
@@ -378,7 +378,7 @@ func TestExportGMandDMChannels(t *testing.T) {
assert.Equal(t, 0, len(channels))
// import the exported channel
err, i := th2.App.BulkImport(th2.Context, &b, false, 5)
err, i := th2.App.BulkImport(th2.Context, &b, nil, false, 5)
assert.Nil(t, err)
assert.Equal(t, 0, i)
@@ -457,7 +457,7 @@ func TestExportDMandGMPost(t *testing.T) {
assert.Equal(t, 0, len(posts))
// import the exported posts
appErr, i := th2.App.BulkImport(th2.Context, &b, false, 5)
appErr, i := th2.App.BulkImport(th2.Context, &b, nil, false, 5)
assert.Nil(t, appErr)
assert.Equal(t, 0, i)
@@ -532,7 +532,7 @@ func TestExportPostWithProps(t *testing.T) {
assert.Len(t, posts, 0)
// import the exported posts
appErr, i := th2.App.BulkImport(th2.Context, &b, false, 5)
appErr, i := th2.App.BulkImport(th2.Context, &b, nil, false, 5)
assert.Nil(t, appErr)
assert.Equal(t, 0, i)
@@ -574,7 +574,7 @@ func TestExportDMPostWithSelf(t *testing.T) {
assert.Equal(t, 0, len(posts))
// import the exported posts
err, i := th2.App.BulkImport(th2.Context, &b, false, 5)
err, i := th2.App.BulkImport(th2.Context, &b, nil, false, 5)
assert.Nil(t, err)
assert.Equal(t, 0, i)
@@ -614,7 +614,7 @@ func TestBulkExport(t *testing.T) {
jsonFile := extractImportFile(filepath.Join(testsDir, "import_test.zip"))
defer jsonFile.Close()
appErr, _ := th.App.BulkImportWithPath(th.Context, jsonFile, false, 1, dir)
appErr, _ := th.App.BulkImportWithPath(th.Context, jsonFile, nil, false, 1, dir)
require.Nil(t, appErr)
exportFile, err := os.Create(filepath.Join(dir, "export.zip"))
@@ -635,6 +635,6 @@ func TestBulkExport(t *testing.T) {
jsonFile = extractImportFile(filepath.Join(dir, "export.zip"))
defer jsonFile.Close()
appErr, _ = th.App.BulkImportWithPath(th.Context, jsonFile, false, 1, filepath.Join(dir, "data"))
appErr, _ = th.App.BulkImportWithPath(th.Context, jsonFile, nil, false, 1, filepath.Join(dir, "data"))
require.Nil(t, appErr)
}

36
app/import.go
Просмотреть файл

@@ -4,9 +4,11 @@
package app
import (
"archive/zip"
"bufio"
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"path/filepath"
@@ -123,16 +125,20 @@ func (a *App) bulkImportWorker(c *request.Context, dryRun bool, wg *sync.WaitGro
wg.Done()
}
func (a *App) BulkImport(c *request.Context, fileReader io.Reader, dryRun bool, workers int) (*model.AppError, int) {
return a.bulkImport(c, fileReader, dryRun, workers, "")
func (a *App) BulkImport(c *request.Context, jsonlReader io.Reader, attachmentsReader *zip.Reader, dryRun bool, workers int) (*model.AppError, int) {
return a.bulkImport(c, jsonlReader, attachmentsReader, dryRun, workers, "")
}
func (a *App) BulkImportWithPath(c *request.Context, fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
return a.bulkImport(c, fileReader, dryRun, workers, importPath)
func (a *App) BulkImportWithPath(c *request.Context, jsonlReader io.Reader, attachmentsReader *zip.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
return a.bulkImport(c, jsonlReader, attachmentsReader, dryRun, workers, importPath)
}
func (a *App) bulkImport(c *request.Context, fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
scanner := bufio.NewScanner(fileReader)
// bulkImport will extract attachments from attachmentsReader if it is
// not nil. If it is nil, it will look for attachments on the
// filesystem in the locations specified by the JSONL file according
// to the older behavior
func (a *App) bulkImport(c *request.Context, jsonlReader io.Reader, attachmentsReader *zip.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
scanner := bufio.NewScanner(jsonlReader)
buf := make([]byte, 0, 64*1024)
scanner.Buffer(buf, maxScanTokenSize)
@@ -146,6 +152,14 @@ func (a *App) bulkImport(c *request.Context, fileReader io.Reader, dryRun bool,
var linesChan chan LineImportWorkerData
lastLineType := ""
var attachedFiles map[string]*zip.File
if attachmentsReader != nil {
attachedFiles = make(map[string]*zip.File, len(attachmentsReader.File))
for _, fi := range attachmentsReader.File {
attachedFiles[fi.Name] = fi
}
}
for scanner.Scan() {
decoder := json.NewDecoder(bytes.NewReader(scanner.Bytes()))
lineNumber++
@@ -155,6 +169,16 @@ func (a *App) bulkImport(c *request.Context, fileReader io.Reader, dryRun bool,
return model.NewAppError("BulkImport", "app.import.bulk_import.json_decode.error", nil, err.Error(), http.StatusBadRequest), lineNumber
}
if len(attachedFiles) > 0 && line.Post != nil && line.Post.Attachments != nil {
for i, attachment := range *line.Post.Attachments {
var ok bool
path := *attachment.Path
if (*line.Post.Attachments)[i].Data, ok = attachedFiles[path]; !ok {
return model.NewAppError("BulkImport", "app.import.bulk_import.json_decode.error", nil, fmt.Sprintf("attachment '%s' not found in map", path), http.StatusBadRequest), lineNumber
}
}
}
if importPath != "" {
rewriteFilePaths(&line, importPath)
}

33
app/import_functions.go
Просмотреть файл

@@ -9,6 +9,7 @@ import (
"crypto/sha1"
"errors"
"fmt"
"io"
"io/ioutil"
"net/http"
"os"
@@ -1146,11 +1147,27 @@ func (a *App) importReplies(c *request.Context, data []ReplyImportData, post *mo
}
func (a *App) importAttachment(c *request.Context, data *AttachmentImportData, post *model.Post, teamID string) (*model.FileInfo, *model.AppError) {
file, err := os.Open(*data.Path)
if file == nil || err != nil {
return nil, model.NewAppError("BulkImport", "app.import.attachment.bad_file.error", map[string]interface{}{"FilePath": *data.Path}, "", http.StatusBadRequest)
var (
name string
file io.Reader
)
if data.Data != nil {
zipFile, err := data.Data.Open()
if err != nil {
return nil, model.NewAppError("BulkImport", "app.import.attachment.bad_file.error", map[string]interface{}{"FilePath": *data.Path}, err.Error(), http.StatusBadRequest)
}
defer zipFile.Close()
name = data.Data.Name
file = zipFile.(io.Reader)
} else {
realFile, err := os.Open(*data.Path)
if err != nil {
return nil, model.NewAppError("BulkImport", "app.import.attachment.bad_file.error", map[string]interface{}{"FilePath": *data.Path}, err.Error(), http.StatusBadRequest)
}
defer realFile.Close()
name = realFile.Name()
file = realFile
}
defer file.Close()
timestamp := utils.TimeFromMillis(post.CreateAt)
@@ -1166,7 +1183,7 @@ func (a *App) importAttachment(c *request.Context, data *AttachmentImportData, p
return nil, model.NewAppError("BulkImport", "app.import.attachment.file_upload.error", map[string]interface{}{"FilePath": *data.Path}, "", http.StatusBadRequest)
}
for _, oldFile := range oldFiles {
if oldFile.Name != path.Base(file.Name()) || oldFile.Size != int64(len(fileData)) {
if oldFile.Name != path.Base(name) || oldFile.Size != int64(len(fileData)) {
continue
}
// check md5
@@ -1178,15 +1195,15 @@ func (a *App) importAttachment(c *request.Context, data *AttachmentImportData, p
oldHash := sha1.Sum(oldFileData)
if bytes.Equal(oldHash[:], newHash[:]) {
mlog.Info("Skipping uploading of file because name already exists", mlog.Any("file_name", file.Name()))
mlog.Info("Skipping uploading of file because name already exists", mlog.Any("file_name", name))
return oldFile, nil
}
}
}
mlog.Info("Uploading file with name", mlog.String("file_name", file.Name()))
mlog.Info("Uploading file with name", mlog.String("file_name", name))
fileInfo, appErr := a.DoUploadFile(c, timestamp, teamID, post.ChannelId, post.UserId, file.Name(), fileData)
fileInfo, appErr := a.DoUploadFile(c, timestamp, teamID, post.ChannelId, post.UserId, name, fileData)
if appErr != nil {
mlog.Error("Failed to upload file:", mlog.Err(appErr))
return nil, appErr

161
app/import_functions_test.go
Просмотреть файл

@@ -4,6 +4,7 @@
package app
import (
"archive/zip"
"context"
"io/ioutil"
"os"
@@ -4025,7 +4026,7 @@ func TestImportPostAndRepliesWithAttachments(t *testing.T) {
AssertFileIdsInPost(attachments, th, t)
})
t.Run("Reply with Attachments in Direct Pos", func(t *testing.T) {
t.Run("Reply with Attachments in Direct Post", func(t *testing.T) {
directImportData := LineImportWorkerData{
LineImportData{
DirectPost: &DirectPostImportData{
@@ -4177,3 +4178,161 @@ func TestImportDirectPostWithAttachments(t *testing.T) {
require.Len(t, attachments, 3)
})
}
func TestZippedImportPostAndRepliesWithAttachments(t *testing.T) {
th := Setup(t)
defer th.TearDown()
// Create a Team.
teamName := model.NewRandomTeamName()
th.App.importTeam(th.Context, &TeamImportData{
Name: &teamName,
DisplayName: ptrStr("Display Name"),
Type: ptrStr("O"),
}, false)
team, appErr := th.App.GetTeamByName(teamName)
require.Nil(t, appErr, "Failed to get team from database.")
// Create a Channel.
channelName := model.NewId()
th.App.importChannel(th.Context, &ChannelImportData{
Team: &teamName,
Name: &channelName,
DisplayName: ptrStr("Display Name"),
Type: ptrStr("O"),
}, false)
_, appErr = th.App.GetChannelByName(channelName, team.Id, false)
require.Nil(t, appErr, "Failed to get channel from database.")
// Create users
username2 := model.NewId()
th.App.importUser(&UserImportData{
Username: &username2,
Email: ptrStr(model.NewId() + "@example.com"),
}, false)
user2, appErr := th.App.GetUserByUsername(username2)
require.Nil(t, appErr, "Failed to get user3 from database.")
// Create direct post users.
username3 := model.NewId()
th.App.importUser(&UserImportData{
Username: &username3,
Email: ptrStr(model.NewId() + "@example.com"),
}, false)
user3, appErr := th.App.GetUserByUsername(username3)
require.Nil(t, appErr, "Failed to get user3 from database.")
username4 := model.NewId()
th.App.importUser(&UserImportData{
Username: &username4,
Email: ptrStr(model.NewId() + "@example.com"),
}, false)
user4, appErr := th.App.GetUserByUsername(username4)
require.Nil(t, appErr, "Failed to get user3 from database.")
// Post with attachments
time := model.GetMillis()
attachmentsPostTime := time
attachmentsReplyTime := time + 1
testsDir, _ := fileutils.FindDir("tests")
testImage := filepath.Join(testsDir, "test.png")
testZipFileName := filepath.Join(testsDir, "import_test.zip")
testZip, _ := os.Open(testZipFileName)
fi, err := testZip.Stat()
require.NoError(t, err, "failed to get file info")
testZipReader, err := zip.NewReader(testZip, fi.Size())
require.NoError(t, err, "failed to read test zip")
require.NotEmpty(t, testZipReader.File)
imageData := testZipReader.File[0]
require.NoError(t, err, "failed to copy test Image file into zip")
testMarkDown := filepath.Join(testsDir, "test-attachments.md")
data := LineImportWorkerData{
LineImportData{
Post: &PostImportData{
Team: &teamName,
Channel: &channelName,
User: &username3,
Message: ptrStr("Message with reply"),
CreateAt: &attachmentsPostTime,
Attachments: &[]AttachmentImportData{{Path: &testImage}, {Path: &testMarkDown}},
Replies: &[]ReplyImportData{{
User: &user4.Username,
Message: ptrStr("Message reply"),
CreateAt: &attachmentsReplyTime,
Attachments: &[]AttachmentImportData{{Path: &testImage, Data: imageData}},
}},
},
},
19,
}
t.Run("import with attachment", func(t *testing.T) {
errLine, err := th.App.importMultiplePostLines(th.Context, []LineImportWorkerData{data}, false)
require.Nil(t, err)
require.Equal(t, 0, errLine)
attachments := GetAttachments(user3.Id, th, t)
require.Len(t, attachments, 2)
assert.Contains(t, attachments[0].Path, team.Id)
assert.Contains(t, attachments[1].Path, team.Id)
AssertFileIdsInPost(attachments, th, t)
attachments = GetAttachments(user4.Id, th, t)
require.Len(t, attachments, 1)
assert.Contains(t, attachments[0].Path, team.Id)
AssertFileIdsInPost(attachments, th, t)
})
t.Run("import existing post with new attachment", func(t *testing.T) {
data.Post.Attachments = &[]AttachmentImportData{{Path: &testImage}}
errLine, err := th.App.importMultiplePostLines(th.Context, []LineImportWorkerData{data}, false)
require.Nil(t, err)
require.Equal(t, 0, errLine)
attachments := GetAttachments(user3.Id, th, t)
require.Len(t, attachments, 1)
assert.Contains(t, attachments[0].Path, team.Id)
AssertFileIdsInPost(attachments, th, t)
attachments = GetAttachments(user4.Id, th, t)
require.Len(t, attachments, 1)
assert.Contains(t, attachments[0].Path, team.Id)
AssertFileIdsInPost(attachments, th, t)
})
t.Run("Reply with Attachments in Direct Post", func(t *testing.T) {
directImportData := LineImportWorkerData{
LineImportData{
DirectPost: &DirectPostImportData{
ChannelMembers: &[]string{
user3.Username,
user2.Username,
},
User: &user3.Username,
Message: ptrStr("Message with Replies"),
CreateAt: ptrInt64(model.GetMillis()),
Replies: &[]ReplyImportData{{
User: &user2.Username,
Message: ptrStr("Message reply with attachment"),
CreateAt: ptrInt64(model.GetMillis()),
Attachments: &[]AttachmentImportData{{Path: &testImage}},
}},
},
},
7,
}
errLine, err := th.App.importMultipleDirectPostLines(th.Context, []LineImportWorkerData{directImportData}, false)
require.Nil(t, err, "Expected success.")
require.Equal(t, 0, errLine)
attachments := GetAttachments(user2.Id, th, t)
require.Len(t, attachments, 1)
assert.Contains(t, attachments[0].Path, "noteam")
AssertFileIdsInPost(attachments, th, t)
})
}

14
app/import_test.go
Просмотреть файл

@@ -185,13 +185,13 @@ func TestImportBulkImport(t *testing.T) {
{"type": "direct_post", "direct_post": {"channel_members": ["` + username + `", "` + username2 + `", "` + username3 + `"], "user": "` + username + `", "message": "Hello Group Channel", "create_at": 123456789015}}
{"type": "emoji", "emoji": {"name": "` + emojiName + `", "image": "` + testImage + `"}}`
err, line := th.App.BulkImport(th.Context, strings.NewReader(data1), false, 2)
err, line := th.App.BulkImport(th.Context, strings.NewReader(data1), nil, false, 2)
require.Nil(t, err, "BulkImport should have succeeded")
require.Equal(t, 0, line, "BulkImport line should be 0")
// Run bulk import using a string that contains a line with invalid json.
data2 := `{"type": "version", "version": 1`
err, line = th.App.BulkImport(th.Context, strings.NewReader(data2), false, 2)
err, line = th.App.BulkImport(th.Context, strings.NewReader(data2), nil, false, 2)
require.NotNil(t, err, "Should have failed due to invalid JSON on line 1.")
require.Equal(t, 1, line, "Should have failed due to invalid JSON on line 1.")
@@ -200,7 +200,7 @@ func TestImportBulkImport(t *testing.T) {
{"type": "channel", "channel": {"type": "O", "display_name": "xr6m6udffngark2uekvr3hoeny", "team": "` + teamName + `", "name": "` + channelName + `"}}
{"type": "user", "user": {"username": "kufjgnkxkrhhfgbrip6qxkfsaa", "email": "kufjgnkxkrhhfgbrip6qxkfsaa@example.com"}}
{"type": "user", "user": {"username": "bwshaim6qnc2ne7oqkd5b2s2rq", "email": "bwshaim6qnc2ne7oqkd5b2s2rq@example.com", "teams": [{"name": "` + teamName + `", "channels": [{"name": "` + channelName + `"}]}]}}`
err, line = th.App.BulkImport(th.Context, strings.NewReader(data3), false, 2)
err, line = th.App.BulkImport(th.Context, strings.NewReader(data3), nil, false, 2)
require.NotNil(t, err, "Should have failed due to missing version line on line 1.")
require.Equal(t, 1, line, "Should have failed due to missing version line on line 1.")
@@ -212,7 +212,7 @@ func TestImportBulkImport(t *testing.T) {
{"type": "channel", "channel": {"type": "O", "display_name": "xr6m6udffngark2uekvr3hoeny", "team": "` + teamName + `", "name": "` + channelName + `"}}
{"type": "user", "user": {"username": "` + username + `", "email": "` + username + `@example.com", "teams": [{"name": "` + teamName + `","theme": "` + teamTheme1 + `", "channels": [{"name": "` + channelName + `"}]}]}}
{"type": "post", "post": {"team": "` + teamName + `", "channel": "` + channelName + `", "user": "` + username + `", "message": "Hello World", "create_at": 123456789012}}`
err, line = th.App.BulkImport(th.Context, strings.NewReader(data4+"\r\n"+posts), false, 2)
err, line = th.App.BulkImport(th.Context, strings.NewReader(data4+"\r\n"+posts), nil, false, 2)
require.Nil(t, err, "BulkImport should have succeeded")
require.Equal(t, 0, line, "BulkImport line should be 0")
})
@@ -220,7 +220,7 @@ func TestImportBulkImport(t *testing.T) {
t.Run("First item after version without type", func(t *testing.T) {
data := `{"type": "version", "version": 1}
{"name": "custom-emoji-troll", "image": "bulkdata/emoji/trollolol.png"}`
err, line := th.App.BulkImport(th.Context, strings.NewReader(data), false, 2)
err, line := th.App.BulkImport(th.Context, strings.NewReader(data), nil, false, 2)
require.NotNil(t, err, "Should have failed due to invalid type on line 2.")
require.Equal(t, 2, line, "Should have failed due to invalid type on line 2.")
})
@@ -234,7 +234,7 @@ func TestImportBulkImport(t *testing.T) {
{"type": "direct_channel", "direct_channel": {"members": ["` + username + `", "` + username + `"]}}
{"type": "direct_post", "direct_post": {"channel_members": ["` + username + `", "` + username + `"], "user": "` + username + `", "message": "Hello Direct Channel to myself", "create_at": 123456789014, "props":{"attachments":[{"id":0,"fallback":"[February 4th, 2020 2:46 PM] author: fallback","color":"D0D0D0","pretext":"","author_name":"author","author_link":"","title":"","title_link":"","text":"this post has props","fields":null,"image_url":"","thumb_url":"","footer":"Posted in #general","footer_icon":"","ts":"1580823992.000100"}]}}}}`
err, line := th.App.BulkImport(th.Context, strings.NewReader(data6), false, 2)
err, line := th.App.BulkImport(th.Context, strings.NewReader(data6), nil, false, 2)
require.Nil(t, err, "BulkImport should have succeeded")
require.Equal(t, 0, line, "BulkImport line should be 0")
})
@@ -393,7 +393,7 @@ func BenchmarkBulkImport(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
err, _ := th.App.BulkImportWithPath(th.Context, jsonFile, false, runtime.NumCPU(), dir)
err, _ := th.App.BulkImportWithPath(th.Context, jsonFile, nil, false, runtime.NumCPU(), dir)
require.Nil(b, err)
}
b.StopTimer()

5
app/import_types.go
Просмотреть файл

@@ -4,6 +4,8 @@
package app
import (
"archive/zip"
"github.com/mattermost/mattermost-server/v5/model"
)
@@ -195,7 +197,8 @@ type LineImportWorkerError struct {
}
type AttachmentImportData struct {
Path *string `json:"path"`
Path *string `json:"path"`
Data *zip.File `json:"-"`
}
type ComparablePreference struct {

9
app/opentracing/opentracing_layer.go
Просмотреть файл

@@ -7,6 +7,7 @@
package opentracing
import (
"archive/zip"
"bytes"
"context"
"crypto/ecdsa"
@@ -958,7 +959,7 @@ func (a *OpenTracingAppLayer) BulkExport(writer io.Writer, outPath string, opts
return resultVar0
}
func (a *OpenTracingAppLayer) BulkImport(c *request.Context, fileReader io.Reader, dryRun bool, workers int) (*model.AppError, int) {
func (a *OpenTracingAppLayer) BulkImport(c *request.Context, jsonlReader io.Reader, attachmentsReader *zip.Reader, dryRun bool, workers int) (*model.AppError, int) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.BulkImport")
@@ -970,7 +971,7 @@ func (a *OpenTracingAppLayer) BulkImport(c *request.Context, fileReader io.Reade
}()
defer span.Finish()
resultVar0, resultVar1 := a.app.BulkImport(c, fileReader, dryRun, workers)
resultVar0, resultVar1 := a.app.BulkImport(c, jsonlReader, attachmentsReader, dryRun, workers)
if resultVar0 != nil {
span.LogFields(spanlog.Error(resultVar0))
@@ -980,7 +981,7 @@ func (a *OpenTracingAppLayer) BulkImport(c *request.Context, fileReader io.Reade
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) BulkImportWithPath(c *request.Context, fileReader io.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
func (a *OpenTracingAppLayer) BulkImportWithPath(c *request.Context, jsonlReader io.Reader, attachmentsReader *zip.Reader, dryRun bool, workers int, importPath string) (*model.AppError, int) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.BulkImportWithPath")
@@ -992,7 +993,7 @@ func (a *OpenTracingAppLayer) BulkImportWithPath(c *request.Context, fileReader
}()
defer span.Finish()
resultVar0, resultVar1 := a.app.BulkImportWithPath(c, fileReader, dryRun, workers, importPath)
resultVar0, resultVar1 := a.app.BulkImportWithPath(c, jsonlReader, attachmentsReader, dryRun, workers, importPath)
if resultVar0 != nil {
span.LogFields(spanlog.Error(resultVar0))

2
cmd/mattermost/commands/import.go
Просмотреть файл

@@ -150,7 +150,7 @@ func bulkImportCmdF(command *cobra.Command, args []string) error {
CommandPrettyPrintln("")
if err, lineNumber := a.BulkImportWithPath(&request.Context{}, fileReader, !apply, workers, importPath); err != nil {
if err, lineNumber := a.BulkImportWithPath(&request.Context{}, fileReader, nil, !apply, workers, importPath); err != nil {
CommandPrintErrorln(err.Error())
if lineNumber != 0 {
CommandPrintErrorln(fmt.Sprintf("Error occurred on data file line %v", lineNumber))

3
cmd/mattermost/commands/sampledata.go
Просмотреть файл

@@ -374,7 +374,8 @@ func sampleDataCmdF(command *cobra.Command, args []string) error {
}
var importErr *model.AppError
importErr, lineNumber := a.BulkImport(&request.Context{}, bulkFile, false, workers)
importErr, lineNumber := a.BulkImport(&request.Context{}, bulkFile, nil, false, workers)
if importErr != nil {
return fmt.Errorf("%s: %s, %s (line: %d)", importErr.Where, importErr.Message, importErr.DetailedError, lineNumber)
}

8
i18n/en.json
Просмотреть файл

@@ -7510,14 +7510,6 @@
"id": "import_process.worker.do_job.open_file",
"translation": "Unable to process import: failed to open file."
},
{
"id": "import_process.worker.do_job.tmp_dir",
"translation": "Unable to process import: failed to create temporary directory."
},
{
"id": "import_process.worker.do_job.unzip",
"translation": "Unable to process import: failed to unzip file."
},
{
"id": "interactive_message.decode_trigger_id.base64_decode_failed",
"translation": "Failed to decode base64 for trigger ID for interactive dialog."

75
jobs/import_process/worker.go
Просмотреть файл

@@ -4,13 +4,13 @@
package import_process
import (
"archive/zip"
"io"
"io/ioutil"
"net/http"
"os"
"path/filepath"
"runtime"
"strconv"
"strings"
"github.com/mattermost/mattermost-server/v5/app"
"github.com/mattermost/mattermost-server/v5/app/request"
@@ -18,7 +18,6 @@ import (
tjobs "github.com/mattermost/mattermost-server/v5/jobs/interfaces"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
"github.com/mattermost/mattermost-server/v5/utils"
)
func init() {
@@ -125,51 +124,45 @@ func (w *ImportProcessWorker) doJob(job *model.Job) {
}
defer importFile.Close()
// TODO (MM-30187): improve this process by eliminating the need to unzip the import
// file locally and instead do the whole bulk import process in memory by
// streaming the import file.
// create a temporary dir to extract the zipped import file.
dir, err := ioutil.TempDir("", "import")
if err != nil {
appError := model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.tmp_dir", nil, err.Error(), http.StatusInternalServerError)
w.setJobError(job, appError)
return
}
defer os.RemoveAll(dir)
// extract the contents of the zipped file.
paths, err := utils.UnzipToPath(importFile.(io.ReaderAt), importFileSize, dir)
if err != nil {
appError := model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.unzip", nil, err.Error(), http.StatusInternalServerError)
w.setJobError(job, appError)
return
}
// find JSONL import file.
var jsonFilePath string
for _, path := range paths {
if filepath.Ext(path) == ".jsonl" {
jsonFilePath = path
break
}
}
if jsonFilePath == "" {
appError := model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.missing_jsonl", nil, "", http.StatusBadRequest)
w.setJobError(job, appError)
return
}
jsonFile, err := os.Open(jsonFilePath)
importZipReader, err := zip.NewReader(importFile.(io.ReaderAt), importFileSize)
if err != nil {
appError := model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.open_file", nil, err.Error(), http.StatusInternalServerError)
w.setJobError(job, appError)
return
}
// find JSONL import file.
var jsonFile io.ReadCloser
for _, f := range importZipReader.File {
if filepath.Ext(f.Name) != ".jsonl" {
continue
}
// avoid "zip slip"
if strings.Contains(f.Name, "..") {
appError := model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.open_file", nil, "jsonFilePath contains path traversal", http.StatusForbidden)
w.setJobError(job, appError)
return
}
jsonFile, err = f.Open()
if err != nil {
appError := model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.open_file", nil, err.Error(), http.StatusInternalServerError)
w.setJobError(job, appError)
return
}
defer jsonFile.Close()
break
}
if jsonFile == nil {
appError := model.NewAppError("ImportProcessWorker", "import_process.worker.do_job.missing_jsonl", nil, "jsonFile was nil", http.StatusBadRequest)
w.setJobError(job, appError)
return
}
// do the actual import.
appErr, lineNumber := w.app.BulkImportWithPath(w.appContext, jsonFile, false, runtime.NumCPU(), filepath.Join(dir, app.ExportDataDir))
appErr, lineNumber := w.app.BulkImport(w.appContext, jsonFile, importZipReader, false, runtime.NumCPU())
if appErr != nil {
job.Data["line_number"] = strconv.Itoa(lineNumber)
w.setJobError(job, appErr)