From 3db6bb016e95b361140a273d55a44a482f029ac3 Mon Sep 17 00:00:00 2001 From: Harshil Sharma <18575143+harshilsharma63@users.noreply.github.com> Date: Fri, 28 Apr 2023 11:11:34 +0530 Subject: [PATCH] Channel index fix (#22947) * WIP * Fetched ES index for type checking * WIP * Moved migration key to model * CI * CI * Fixed existing tests * Updated einterface mocks * CI * WIP * Reverted test makefile changes * Fixed error level * CI * Not attempting to launch job if not enterprise ready * CI * Fixing job trigger condition * CI * Updated eemocks * CI --- server/channels/app/enterprise.go | 6 +++ server/channels/app/migrations.go | 19 ++++++++ server/channels/app/server.go | 5 ++ .../einterfaces/jobs/elasticsearch.go | 4 ++ .../ElasticsearchFixChannelIndexInterface.go | 46 +++++++++++++++++++ server/channels/testlib/store.go | 1 + server/model/job.go | 1 + server/model/migration.go | 1 + 8 files changed, 83 insertions(+) create mode 100644 server/channels/einterfaces/mocks/ElasticsearchFixChannelIndexInterface.go diff --git a/server/channels/app/enterprise.go b/server/channels/app/enterprise.go index 4edd75a534..bf7a18b05c 100644 --- a/server/channels/app/enterprise.go +++ b/server/channels/app/enterprise.go @@ -50,6 +50,12 @@ func RegisterJobsElasticsearchIndexerInterface(f func(*Server) ejobs.IndexerJobI jobsElasticsearchIndexerInterface = f } +var jobsElasticsearchFixChannelIndexInterface func(*Server) ejobs.ElasticsearchFixChannelIndexInterface + +func RegisterJobsElasticsearchFixChannelIndexInterface(f func(*Server) ejobs.ElasticsearchFixChannelIndexInterface) { + jobsElasticsearchFixChannelIndexInterface = f +} + var jobsLdapSyncInterface func(*App) ejobs.LdapSyncInterface func RegisterJobsLdapSyncInterface(f func(*App) ejobs.LdapSyncInterface) { diff --git a/server/channels/app/migrations.go b/server/channels/app/migrations.go index a90af5a333..66eab5f40f 100644 --- a/server/channels/app/migrations.go +++ b/server/channels/app/migrations.go @@ -559,6 +559,24 @@ func (s *Server) doPostPriorityConfigDefaultTrueMigration() { } } +func (s *Server) doElasticsearchFixChannelIndex() { + // If the migration is already marked as completed, don't do it again. + if _, err := s.Store().System().GetByName(model.MigrationKeyElasticsearchFixChannelIndex); err == nil { + return + } + + license := s.License() + if model.BuildEnterpriseReady != "true" || license == nil || !*license.Features.Elasticsearch { + mlog.Info("Skipping triggering Elasticsearch channel index fix job as build is not Enterprise ready") + return + } + + if _, appErr := s.Jobs.CreateJob(model.JobTypeElasticsearchFixChannelIndex, nil); appErr != nil { + mlog.Fatal("failed to start job for fixing Elasticsearch channels index", mlog.Err(appErr)) + return + } +} + func (a *App) DoAppMigrations() { a.Srv().doAppMigrations() } @@ -580,4 +598,5 @@ func (s *Server) doAppMigrations() { s.doFirstAdminSetupCompleteMigration() s.doRemainingSchemaMigrations() s.doPostPriorityConfigDefaultTrueMigration() + s.doElasticsearchFixChannelIndex() } diff --git a/server/channels/app/server.go b/server/channels/app/server.go index a9aeb47dc8..7bfe79fd3c 100644 --- a/server/channels/app/server.go +++ b/server/channels/app/server.go @@ -1476,6 +1476,11 @@ func (s *Server) initJobs() { s.Jobs.RegisterJobType(model.JobTypeElasticsearchPostIndexing, builder.MakeWorker(), nil) } + if jobsElasticsearchFixChannelIndexInterface != nil { + builder := jobsElasticsearchFixChannelIndexInterface(s) + s.Jobs.RegisterJobType(model.JobTypeElasticsearchFixChannelIndex, builder.MakeWorker(), nil) + } + if jobsLdapSyncInterface != nil { builder := jobsLdapSyncInterface(New(ServerConnector(s.Channels()))) s.Jobs.RegisterJobType(model.JobTypeLdapSync, builder.MakeWorker(), builder.MakeScheduler()) diff --git a/server/channels/einterfaces/jobs/elasticsearch.go b/server/channels/einterfaces/jobs/elasticsearch.go index aed3df9ef6..32ad94ff76 100644 --- a/server/channels/einterfaces/jobs/elasticsearch.go +++ b/server/channels/einterfaces/jobs/elasticsearch.go @@ -15,3 +15,7 @@ type ElasticsearchAggregatorInterface interface { MakeWorker() model.Worker MakeScheduler() model.Scheduler } + +type ElasticsearchFixChannelIndexInterface interface { + MakeWorker() model.Worker +} diff --git a/server/channels/einterfaces/mocks/ElasticsearchFixChannelIndexInterface.go b/server/channels/einterfaces/mocks/ElasticsearchFixChannelIndexInterface.go new file mode 100644 index 0000000000..08100b3cc4 --- /dev/null +++ b/server/channels/einterfaces/mocks/ElasticsearchFixChannelIndexInterface.go @@ -0,0 +1,46 @@ +// Code generated by mockery v2.23.2. DO NOT EDIT. + +// Regenerate this file using `make einterfaces-mocks`. + +package mocks + +import ( + model "github.com/mattermost/mattermost-server/server/v8/model" + mock "github.com/stretchr/testify/mock" +) + +// ElasticsearchFixChannelIndexInterface is an autogenerated mock type for the ElasticsearchFixChannelIndexInterface type +type ElasticsearchFixChannelIndexInterface struct { + mock.Mock +} + +// MakeWorker provides a mock function with given fields: +func (_m *ElasticsearchFixChannelIndexInterface) MakeWorker() model.Worker { + ret := _m.Called() + + var r0 model.Worker + if rf, ok := ret.Get(0).(func() model.Worker); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(model.Worker) + } + } + + return r0 +} + +type mockConstructorTestingTNewElasticsearchFixChannelIndexInterface interface { + mock.TestingT + Cleanup(func()) +} + +// NewElasticsearchFixChannelIndexInterface creates a new instance of ElasticsearchFixChannelIndexInterface. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +func NewElasticsearchFixChannelIndexInterface(t mockConstructorTestingTNewElasticsearchFixChannelIndexInterface) *ElasticsearchFixChannelIndexInterface { + mock := &ElasticsearchFixChannelIndexInterface{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} diff --git a/server/channels/testlib/store.go b/server/channels/testlib/store.go index 05e944fba9..c8c9067562 100644 --- a/server/channels/testlib/store.go +++ b/server/channels/testlib/store.go @@ -72,6 +72,7 @@ func GetMockStoreForSetupFunctions() *mocks.Store { systemStore.On("GetByName", model.MigrationKeyAddCustomUserGroupsPermissionRestore).Return(&model.System{Name: model.MigrationKeyAddCustomUserGroupsPermissionRestore, Value: "true"}, nil) systemStore.On("GetByName", "CustomGroupAdminRoleCreationMigrationComplete").Return(&model.System{Name: model.MigrationKeyAddPlayboosksManageRolesPermissions, Value: "true"}, nil) systemStore.On("GetByName", "products_boards").Return(&model.System{Name: "products_boards", Value: "true"}, nil) + systemStore.On("GetByName", "elasticsearch_fix_channel_index_migration_complete").Return(&model.System{Name: "elasticsearch_fix_channel_index_migration_complete", Value: "true"}, nil) systemStore.On("InsertIfExists", mock.AnythingOfType("*model.System")).Return(&model.System{}, nil).Once() systemStore.On("Save", mock.AnythingOfType("*model.System")).Return(nil) diff --git a/server/model/job.go b/server/model/job.go index a9c4c6f4a0..3801e1f01b 100644 --- a/server/model/job.go +++ b/server/model/job.go @@ -13,6 +13,7 @@ const ( JobTypeMessageExport = "message_export" JobTypeElasticsearchPostIndexing = "elasticsearch_post_indexing" JobTypeElasticsearchPostAggregation = "elasticsearch_post_aggregation" + JobTypeElasticsearchFixChannelIndex = "elasticsearch_fix_channel_index" JobTypeBlevePostIndexing = "bleve_post_indexing" JobTypeLdapSync = "ldap_sync" JobTypeMigrations = "migrations" diff --git a/server/model/migration.go b/server/model/migration.go index 766e51598a..3369851479 100644 --- a/server/model/migration.go +++ b/server/model/migration.go @@ -40,4 +40,5 @@ const ( MigrationKeyAddPlayboosksManageRolesPermissions = "playbooks_manage_roles" MigrationKeyAddProductsBoardsPermissions = "products_boards" MigrationKeyAddCustomUserGroupsPermissionRestore = "custom_groups_permission_restore" + MigrationKeyElasticsearchFixChannelIndex = "elasticsearch_fix_channel_index_migration_complete" )