From 6d6e589c11d513a4df0b9008102eb99143f43e48 Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Tue, 25 Jul 2023 08:38:35 +0530 Subject: [PATCH] MM-53747: Create job to encode older image paths (#24073) Bifrost now encodes all image paths. Due to this one-way translation, we need to encode all the older image paths as well. After this is done, we can remove the double-lookup. https://mattermost.atlassian.net/browse/MM-53747 ```release-note NONE ``` Co-authored-by: Mattermost Build --- server/channels/app/migrations.go | 31 ++ server/channels/app/server.go | 6 + .../s3_path_migration/s3_path_migration.go | 265 ++++++++++++++++++ .../opentracinglayer/opentracinglayer.go | 4 +- .../channels/store/retrylayer/retrylayer.go | 4 +- .../store/sqlstore/file_info_store.go | 14 +- server/channels/store/store.go | 2 +- .../store/storetest/file_info_store.go | 18 +- .../store/storetest/mocks/FileInfoStore.go | 18 +- .../channels/store/timerlayer/timerlayer.go | 4 +- .../bleveengine/indexer/indexing_job.go | 2 +- server/platform/shared/filestore/s3store.go | 57 +++- server/public/model/job.go | 1 + server/public/model/migration.go | 1 + 14 files changed, 399 insertions(+), 28 deletions(-) create mode 100644 server/channels/jobs/s3_path_migration/s3_path_migration.go diff --git a/server/channels/app/migrations.go b/server/channels/app/migrations.go index fa643174ff..8281533963 100644 --- a/server/channels/app/migrations.go +++ b/server/channels/app/migrations.go @@ -6,6 +6,7 @@ package app import ( "context" "fmt" + "os" "reflect" "github.com/mattermost/mattermost/server/public/model" @@ -571,6 +572,35 @@ func (s *Server) doElasticsearchFixChannelIndex() { } } +func (s *Server) doCloudS3PathMigrations() { + // This migration is only applicable for cloud environments + if os.Getenv("MM_CLOUD_FILESTORE_BIFROST") == "" { + return + } + + // If the migration is already marked as completed, don't do it again. + if _, err := s.Store().System().GetByName(model.MigrationKeyS3Path); err == nil { + return + } + + // If there is a job already pending, no need to schedule again. + // This is possible if the pod was rolled over. + jobs, err := s.Store().Job().GetAllByTypeAndStatus(model.JobTypeS3PathMigration, model.JobStatusPending) + if err != nil { + mlog.Fatal("failed to get jobs by type and status", mlog.Err(err)) + return + } + if len(jobs) > 0 { + return + } + + if _, appErr := s.Jobs.CreateJob(model.JobTypeS3PathMigration, nil); appErr != nil { + mlog.Fatal("failed to start job for migrating s3 file paths", mlog.Err(appErr)) + return + } + +} + func (a *App) DoAppMigrations() { a.Srv().doAppMigrations() } @@ -593,4 +623,5 @@ func (s *Server) doAppMigrations() { s.doRemainingSchemaMigrations() s.doPostPriorityConfigDefaultTrueMigration() s.doElasticsearchFixChannelIndex() + s.doCloudS3PathMigrations() } diff --git a/server/channels/app/server.go b/server/channels/app/server.go index ed6fc7faf4..a45d7cc8b7 100644 --- a/server/channels/app/server.go +++ b/server/channels/app/server.go @@ -54,6 +54,7 @@ import ( "github.com/mattermost/mattermost/server/v8/channels/jobs/post_persistent_notifications" "github.com/mattermost/mattermost/server/v8/channels/jobs/product_notices" "github.com/mattermost/mattermost/server/v8/channels/jobs/resend_invitation_email" + "github.com/mattermost/mattermost/server/v8/channels/jobs/s3_path_migration" "github.com/mattermost/mattermost/server/v8/channels/product" "github.com/mattermost/mattermost/server/v8/channels/store" "github.com/mattermost/mattermost/server/v8/channels/utils" @@ -1568,6 +1569,11 @@ func (s *Server) initJobs() { import_delete.MakeScheduler(s.Jobs), ) + s.Jobs.RegisterJobType( + model.JobTypeS3PathMigration, + s3_path_migration.MakeWorker(s.Jobs, s.Store(), s.FileBackend()), + nil) + s.Jobs.RegisterJobType( model.JobTypeExportDelete, export_delete.MakeWorker(s.Jobs, New(ServerConnector(s.Channels()))), diff --git a/server/channels/jobs/s3_path_migration/s3_path_migration.go b/server/channels/jobs/s3_path_migration/s3_path_migration.go new file mode 100644 index 0000000000..1157b7646e --- /dev/null +++ b/server/channels/jobs/s3_path_migration/s3_path_migration.go @@ -0,0 +1,265 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package s3_path_migration + +import ( + "errors" + "net/http" + "os" + "strconv" + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/v8/channels/jobs" + "github.com/mattermost/mattermost/server/v8/channels/store" + "github.com/mattermost/mattermost/server/v8/platform/shared/filestore" +) + +const ( + JobName = "S3PathMigration" + + timeBetweenBatches = 1 * time.Second +) + +type S3PathMigrationWorker struct { + name string + jobServer *jobs.JobServer + store store.Store + fileBackend *filestore.S3FileBackend + + stop chan bool + stopped chan bool + jobs chan model.Job +} + +func MakeWorker(jobServer *jobs.JobServer, store store.Store, fileBackend filestore.FileBackend) model.Worker { + // If the type cast fails, it will be nil + // which is checked later. + s3Backend, _ := fileBackend.(*filestore.S3FileBackend) + worker := &S3PathMigrationWorker{ + jobServer: jobServer, + store: store, + fileBackend: s3Backend, + name: JobName, + stop: make(chan bool, 1), + stopped: make(chan bool, 1), + jobs: make(chan model.Job), + } + return worker +} + +func (worker *S3PathMigrationWorker) Run() { + mlog.Debug("Worker started", mlog.String("worker", worker.name)) + + defer func() { + mlog.Debug("Worker finished", mlog.String("worker", worker.name)) + worker.stopped <- true + }() + + for { + select { + case <-worker.stop: + mlog.Debug("Worker received stop signal", mlog.String("worker", worker.name)) + return + case job := <-worker.jobs: + mlog.Debug("Worker received a new candidate job.", mlog.String("worker", worker.name)) + worker.DoJob(&job) + } + } +} + +func (worker *S3PathMigrationWorker) Stop() { + mlog.Debug("Worker stopping", mlog.String("worker", worker.name)) + close(worker.stop) + <-worker.stopped +} + +func (worker *S3PathMigrationWorker) JobChannel() chan<- model.Job { + return worker.jobs +} + +func (worker *S3PathMigrationWorker) IsEnabled(_ *model.Config) bool { + return os.Getenv("MM_CLOUD_FILESTORE_BIFROST") != "" +} + +func (worker *S3PathMigrationWorker) getJobMetadata(job *model.Job, key string) (int, *model.AppError) { + countStr := job.Data[key] + count := 0 + var err error + if countStr != "" { + count, err = strconv.Atoi(countStr) + if err != nil { + return 0, model.NewAppError("getJobMetadata", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err) + } + } + return count, nil +} + +func (worker *S3PathMigrationWorker) DoJob(job *model.Job) { + defer worker.jobServer.HandleJobPanic(job) + + if claimed, err := worker.jobServer.ClaimJob(job); err != nil { + mlog.Warn("S3PathMigrationWorker experienced an error while trying to claim job", + mlog.String("worker", worker.name), + mlog.String("job_id", job.Id), + mlog.Err(err)) + return + } else if !claimed { + return + } + + if worker.fileBackend == nil { + err := errors.New("no S3 file backend found") + mlog.Error("S3PathMigrationWorker: ", mlog.Err(err)) + worker.setJobError(job, model.NewAppError("DoJob", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)) + 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("S3PathMigrationWorker: job execution error", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(appErr)) + worker.setJobError(job, appErr) + return + } + + // Check if there is metadata for that job. + // If there isn't, it will be empty by default, which is the right value. + startFileID := job.Data["start_file_id"] + + doneCount, appErr := worker.getJobMetadata(job, "done_file_count") + if appErr != nil { + mlog.Error("S3PathMigrationWorker: failed to get done file count", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(appErr)) + worker.setJobError(job, appErr) + return + } + + startTime, appErr := worker.getJobMetadata(job, "start_create_at") + if appErr != nil { + mlog.Error("S3PathMigrationWorker: failed to get start create_at", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(appErr)) + worker.setJobError(job, appErr) + return + } + if startTime == 0 { + // Time of the commit because we know no files older than that are affected. + // Exact commit was done on 09:54AM June 27, IST. + // We take June 26 as an approximation. + startTime = int(time.Date(2023, time.June, 26, 0, 0, 0, 0, time.UTC).UnixMilli()) + } + + const pageSize = 100 + + for { + select { + case <-worker.stop: + mlog.Info("Worker: S3 Migration has been canceled via Worker Stop. Setting the job back to pending.", + mlog.String("workername", worker.name), + mlog.String("job_id", job.Id)) + if err := worker.jobServer.SetJobPending(job); err != nil { + mlog.Error("Worker: Failed to mark job as pending", + mlog.String("workername", worker.name), + mlog.String("job_id", job.Id), + mlog.Err(err)) + } + return + case <-time.After(timeBetweenBatches): + var files []*model.FileForIndexing + tries := 0 + for files == nil { + var err error + // Take batches of `pageSize` + files, err = worker.store.FileInfo().GetFilesBatchForIndexing(int64(startTime), startFileID, true, pageSize) + if err != nil { + if tries > 3 { + mlog.Error("Worker: Failed to get files after multiple retries. Exiting") + worker.setJobError(job, model.NewAppError("DoJob", model.NoTranslation, nil, "", http.StatusInternalServerError).Wrap(err)) + return + } + mlog.Warn("Failed to get file info for s3 migration. Retrying .. ", mlog.Err(err)) + + // Wait a bit before trying again. + time.Sleep(15 * time.Second) + } + + tries++ + } + + if len(files) == 0 { + mlog.Info("S3PathMigrationWorker: Job is complete", mlog.String("worker", worker.name), mlog.String("job_id", job.Id)) + worker.setJobSuccess(job) + worker.markAsComplete() + return + } + + // Iterate through the rows in each page. + for _, f := range files { + mlog.Debug("Processing file ID", mlog.String("id", f.Id), mlog.String("worker", worker.name), mlog.String("job_id", job.Id)) + // We do not fail the job if a single image failed to encode. + if f.Path != "" { + if err := worker.fileBackend.DecodeFilePathIfNeeded(f.Path); err != nil { + mlog.Warn("Failed to encode S3 file path", mlog.String("path", f.Path), mlog.String("id", f.Id), mlog.Err(err)) + } + } + if f.PreviewPath != "" { + if err := worker.fileBackend.DecodeFilePathIfNeeded(f.PreviewPath); err != nil { + mlog.Warn("Failed to encode S3 file path", mlog.String("path", f.PreviewPath), mlog.String("id", f.Id), mlog.Err(err)) + } + } + if f.ThumbnailPath != "" { + if err := worker.fileBackend.DecodeFilePathIfNeeded(f.ThumbnailPath); err != nil { + mlog.Warn("Failed to encode S3 file path", mlog.String("path", f.ThumbnailPath), mlog.String("id", f.Id), mlog.Err(err)) + } + } + } + + // Work on each batch and save the batch starting ID in metadata + lastFile := files[len(files)-1] + startFileID = lastFile.Id + startTime = int(lastFile.CreateAt) + + if job.Data == nil { + job.Data = make(model.StringMap) + } + job.Data["start_file_id"] = startFileID + job.Data["start_create_at"] = strconv.Itoa(startTime) + doneCount += len(files) + job.Data["done_file_count"] = strconv.Itoa(doneCount) + } + } +} + +func (worker *S3PathMigrationWorker) markAsComplete() { + system := model.System{ + Name: model.MigrationKeyS3Path, + Value: "true", + } + + // Note that if this fails, then the job would have still succeeded. + // So it will try to run the same job again next time, but then + // it will just fall through everything because all files would have + // converted. The actual job is idempotent, so there won't be a problem. + if err := worker.jobServer.Store.System().Save(&system); err != nil { + mlog.Error("Worker: Failed to mark s3 path migration as completed in the systems table.", mlog.String("workername", worker.name), mlog.Err(err)) + } +} + +func (worker *S3PathMigrationWorker) 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("S3PathMigrationWorker: Failed to set success for job", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) + worker.setJobError(job, err) + } +} + +func (worker *S3PathMigrationWorker) setJobError(job *model.Job, appError *model.AppError) { + if err := worker.jobServer.SetJobError(job, appError); err != nil { + mlog.Error("S3PathMigrationWorker: Failed to set job error", mlog.String("worker", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) + } +} diff --git a/server/channels/store/opentracinglayer/opentracinglayer.go b/server/channels/store/opentracinglayer/opentracinglayer.go index e91e0353f6..d71e322124 100644 --- a/server/channels/store/opentracinglayer/opentracinglayer.go +++ b/server/channels/store/opentracinglayer/opentracinglayer.go @@ -3608,7 +3608,7 @@ func (s *OpenTracingLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, return result, err } -func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { +func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetFilesBatchForIndexing") s.Root.Store.SetContext(newCtx) @@ -3617,7 +3617,7 @@ func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64 }() defer span.Finish() - result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit) + result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, includeDeleted, limit) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) diff --git a/server/channels/store/retrylayer/retrylayer.go b/server/channels/store/retrylayer/retrylayer.go index f5a8c5c539..4dc9633ded 100644 --- a/server/channels/store/retrylayer/retrylayer.go +++ b/server/channels/store/retrylayer/retrylayer.go @@ -4034,11 +4034,11 @@ func (s *RetryLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error } -func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { +func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) { tries := 0 for { - result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit) + result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, includeDeleted, limit) if err == nil { return result, nil } diff --git a/server/channels/store/sqlstore/file_info_store.go b/server/channels/store/sqlstore/file_info_store.go index 6c67ad3cd2..00afdea04c 100644 --- a/server/channels/store/sqlstore/file_info_store.go +++ b/server/channels/store/sqlstore/file_info_store.go @@ -691,9 +691,10 @@ func (fs SqlFileInfoStore) CountAll() (int64, error) { return count, nil } -func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { +func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) { files := []*model.FileForIndexing{} - sql, args, _ := fs.getQueryBuilder(). + + query := fs.getQueryBuilder(). Select(fs.queryFields...). From("FileInfo"). Where(sq.Or{ @@ -704,10 +705,13 @@ func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID }, }). OrderBy("FileInfo.CreateAt ASC, FileInfo.Id ASC"). - Limit(uint64(limit)). - ToSql() + Limit(uint64(limit)) - err := fs.GetSearchReplicaX().Select(&files, sql, args...) + if !includeDeleted { + query = query.Where(sq.Eq{"FileInfo.DeleteAt": 0}) + } + + err := fs.GetSearchReplicaX().SelectBuilder(&files, query) if err != nil { return nil, errors.Wrap(err, "failed to find Files") } diff --git a/server/channels/store/store.go b/server/channels/store/store.go index 47492889f9..81215b164e 100644 --- a/server/channels/store/store.go +++ b/server/channels/store/store.go @@ -703,7 +703,7 @@ type FileInfoStore interface { SetContent(fileID, content string) error Search(paramsList []*model.SearchParams, userID, teamID string, page, perPage int) (*model.FileInfoList, error) CountAll() (int64, error) - GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) + GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) ClearCaches() GetStorageUsage(allowFromCache, includeDeleted bool) (int64, error) // GetUptoNSizeFileTime returns the CreateAt time of the last accessible file with a running-total size upto n bytes. diff --git a/server/channels/store/storetest/file_info_store.go b/server/channels/store/storetest/file_info_store.go index a1c100c400..5ee5a41c42 100644 --- a/server/channels/store/storetest/file_info_store.go +++ b/server/channels/store/storetest/file_info_store.go @@ -748,21 +748,29 @@ func testFileInfoStoreGetFilesBatchForIndexing(t *testing.T, ss store.Store) { ss.FileInfo().PermanentDelete(f3.Id) }() + // Soft-deleting one file info + _, err = ss.FileInfo().DeleteForPost(f1.PostId) + require.NoError(t, err) + // Getting all - r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 100) + r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", true, 100) require.NoError(t, err) require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r)) - // Testing pagination - r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 2) + r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", false, 100) require.NoError(t, err) require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r)) - r, err = ss.FileInfo().GetFilesBatchForIndexing(r[1].CreateAt, r[1].Id, 2) + // Testing pagination + r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", true, 2) + require.NoError(t, err) + require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r)) + + r, err = ss.FileInfo().GetFilesBatchForIndexing(r[1].CreateAt, r[1].Id, true, 2) require.NoError(t, err) require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r)) - r, err = ss.FileInfo().GetFilesBatchForIndexing(r[0].CreateAt, r[0].Id, 2) + r, err = ss.FileInfo().GetFilesBatchForIndexing(r[0].CreateAt, r[0].Id, true, 2) require.NoError(t, err) require.Len(t, r, 0, "Expected 0 posts in results. Got %v", len(r)) } diff --git a/server/channels/store/storetest/mocks/FileInfoStore.go b/server/channels/store/storetest/mocks/FileInfoStore.go index fd10afe000..9436e64d2d 100644 --- a/server/channels/store/storetest/mocks/FileInfoStore.go +++ b/server/channels/store/storetest/mocks/FileInfoStore.go @@ -159,25 +159,25 @@ func (_m *FileInfoStore) GetByPath(path string) (*model.FileInfo, error) { return r0, r1 } -// GetFilesBatchForIndexing provides a mock function with given fields: startTime, startFileID, limit -func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { - ret := _m.Called(startTime, startFileID, limit) +// GetFilesBatchForIndexing provides a mock function with given fields: startTime, startFileID, includeDeleted, limit +func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) { + ret := _m.Called(startTime, startFileID, includeDeleted, limit) var r0 []*model.FileForIndexing var r1 error - if rf, ok := ret.Get(0).(func(int64, string, int) ([]*model.FileForIndexing, error)); ok { - return rf(startTime, startFileID, limit) + if rf, ok := ret.Get(0).(func(int64, string, bool, int) ([]*model.FileForIndexing, error)); ok { + return rf(startTime, startFileID, includeDeleted, limit) } - if rf, ok := ret.Get(0).(func(int64, string, int) []*model.FileForIndexing); ok { - r0 = rf(startTime, startFileID, limit) + if rf, ok := ret.Get(0).(func(int64, string, bool, int) []*model.FileForIndexing); ok { + r0 = rf(startTime, startFileID, includeDeleted, limit) } else { if ret.Get(0) != nil { r0 = ret.Get(0).([]*model.FileForIndexing) } } - if rf, ok := ret.Get(1).(func(int64, string, int) error); ok { - r1 = rf(startTime, startFileID, limit) + if rf, ok := ret.Get(1).(func(int64, string, bool, int) error); ok { + r1 = rf(startTime, startFileID, includeDeleted, limit) } else { r1 = ret.Error(1) } diff --git a/server/channels/store/timerlayer/timerlayer.go b/server/channels/store/timerlayer/timerlayer.go index 4c54c34ac8..21e6655a83 100644 --- a/server/channels/store/timerlayer/timerlayer.go +++ b/server/channels/store/timerlayer/timerlayer.go @@ -3303,10 +3303,10 @@ func (s *TimerLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error return result, err } -func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { +func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, includeDeleted bool, limit int) ([]*model.FileForIndexing, error) { start := time.Now() - result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit) + result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, includeDeleted, limit) elapsed := float64(time.Since(start)) / float64(time.Second) if s.Root.Metrics != nil { diff --git a/server/platform/services/searchengine/bleveengine/indexer/indexing_job.go b/server/platform/services/searchengine/bleveengine/indexer/indexing_job.go index 33e4da6c70..58110842b9 100644 --- a/server/platform/services/searchengine/bleveengine/indexer/indexing_job.go +++ b/server/platform/services/searchengine/bleveengine/indexer/indexing_job.go @@ -404,7 +404,7 @@ func (worker *BleveIndexerWorker) IndexFilesBatch(progress IndexingProgress) (In tries := 0 for files == nil { var err error - files, err = worker.jobServer.Store.FileInfo().GetFilesBatchForIndexing(progress.LastEntityTime, progress.LastFileID, *worker.jobServer.Config().BleveSettings.BatchSize) + files, err = worker.jobServer.Store.FileInfo().GetFilesBatchForIndexing(progress.LastEntityTime, progress.LastFileID, true, *worker.jobServer.Config().BleveSettings.BatchSize) if err != nil { if tries >= 10 { return progress, model.NewAppError("IndexFilesBatch", "app.post.get_files_batch_for_indexing.get.app_error", nil, "", http.StatusInternalServerError).Wrap(err) diff --git a/server/platform/shared/filestore/s3store.go b/server/platform/shared/filestore/s3store.go index eae68e3510..67b330d5e4 100644 --- a/server/platform/shared/filestore/s3store.go +++ b/server/platform/shared/filestore/s3store.go @@ -368,6 +368,62 @@ func (b *S3FileBackend) CopyFile(oldPath, newPath string) error { return nil } +// DecodeFilePathIfNeeded is a special method to URL decode all older +// file paths. It is only needed for the migration, and will be removed +// as soon as the migration is complete. +func (b *S3FileBackend) DecodeFilePathIfNeeded(path string) error { + // Encode and check if file path changes. + // If there is no change, then there is no need to do anything. + if path == s3utils.EncodePath(path) { + return nil + } + + // Check if encoded path exists. + exists, err := b.lookupOriginalPath(s3utils.EncodePath(path)) + if err != nil { + return err + } + + if !exists { + return nil + } + + // If yes, then it needs to be migrated. + // This is basically a copy of MoveFile without the path encoding. + // We avoid any further refactoring because this method will be removed anyways. + oldPath := filepath.Join(b.pathPrefix, s3utils.EncodePath(path)) + newPath := filepath.Join(b.pathPrefix, path) + srcOpts := s3.CopySrcOptions{ + Bucket: b.bucket, + Object: oldPath, + } + if b.encrypt { + srcOpts.Encryption = encrypt.NewSSE() + } + + dstOpts := s3.CopyDestOptions{ + Bucket: b.bucket, + Object: newPath, + } + if b.encrypt { + dstOpts.Encryption = encrypt.NewSSE() + } + + ctx, cancel := context.WithTimeout(context.Background(), b.timeout) + defer cancel() + if _, err := b.client.CopyObject(ctx, dstOpts, srcOpts); err != nil { + return errors.Wrapf(err, "unable to copy the file to %s to the new destination", newPath) + } + + ctx2, cancel2 := context.WithTimeout(context.Background(), b.timeout) + defer cancel2() + if err := b.client.RemoveObject(ctx2, b.bucket, oldPath, s3.RemoveObjectOptions{}); err != nil { + return errors.Wrapf(err, "unable to remove the file old file %s", oldPath) + } + + return nil +} + func (b *S3FileBackend) MoveFile(oldPath, newPath string) error { oldPath, err := b.prefixedPath(oldPath) if err != nil { @@ -666,7 +722,6 @@ func (b *S3FileBackend) prefixedPath(s string) (string, error) { // More info at: https://github.com/aws/aws-sdk-go/blob/a57c4d92784a43b716645a57b6fa5fb94fb6e419/aws/signer/v4/v4.go#L8 s = s3utils.EncodePath(s) } - } return filepath.Join(b.pathPrefix, s), nil } diff --git a/server/public/model/job.go b/server/public/model/job.go index 9c3b010c30..5b2dea6731 100644 --- a/server/public/model/job.go +++ b/server/public/model/job.go @@ -35,6 +35,7 @@ const ( JobTypePostPersistentNotifications = "post_persistent_notifications" JobTypeInstallPluginNotifyAdmin = "install_plugin_notify_admin" JobTypeHostedPurchaseScreening = "hosted_purchase_screening" + JobTypeS3PathMigration = "s3_path_migration" JobStatusPending = "pending" JobStatusInProgress = "in_progress" diff --git a/server/public/model/migration.go b/server/public/model/migration.go index ad7f286c88..480080b872 100644 --- a/server/public/model/migration.go +++ b/server/public/model/migration.go @@ -42,4 +42,5 @@ const ( MigrationKeyAddCustomUserGroupsPermissionRestore = "custom_groups_permission_restore" MigrationKeyAddReadChannelContentPermissions = "read_channel_content_permissions" MigrationKeyElasticsearchFixChannelIndex = "elasticsearch_fix_channel_index_migration" + MigrationKeyS3Path = "s3_path_migration" )