Rebuild channel index (#26080)

* WIP

* Added rebuild channels index functionality

* Added rough logic to send message to all sysadmins

* WIP

* WIP

* WIP

* Cleanup

* i18n fix

* reading through all pages of system admins

* Fixed webapp style

* i18n fix

* Added help text

* i18n fix

* i18n update

* Updated system console button action

* Updated snapshots

* some cleanup

* Updated snapshot

* Update server/channels/app/server.go

Co-authored-by: Daniel Espino García <larkox@gmail.com>

* fixed typo

* Refactoring to improve readibility

* moved index check to API later during config update

* Added some docs

* Updated get system bot

---------

Co-authored-by: Daniel Espino García <larkox@gmail.com>

server/channels/api4/config.go

@@ -167,6 +167,15 @@ func updateConfig(c *Context, w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	// if ES autocomplete was enabled, we need to make sure that index has been checked.
+	// we need to stop enabling ES autocomplete otherwise.
+	if !*appCfg.ElasticsearchSettings.EnableAutocomplete && *cfg.ElasticsearchSettings.EnableAutocomplete {
+		if !c.App.SearchEngine().ElasticsearchEngine.IsAutocompletionEnabled() {
+			c.Err = model.NewAppError("updateConfig", "api.config.update.elasticsearch.autocomplete_cannot_be_enabled_error", nil, "", http.StatusBadRequest)
+			return
+		}
+	}
+
 	c.App.HandleMessageExportConfig(cfg, appCfg)
 
 	if appErr := cfg.IsValid(); appErr != nil {

server/channels/api4/elasticsearch.go

@@ -71,7 +71,8 @@ func purgeElasticsearchIndexes(c *Context, w http.ResponseWriter, r *http.Reques
 		return
 	}
 
-	if err := c.App.PurgeElasticsearchIndexes(c.AppContext); err != nil {
+	specifiedIndexesQuery := r.URL.Query()["index"]
+	if err := c.App.PurgeElasticsearchIndexes(c.AppContext, specifiedIndexesQuery); err != nil {
 		c.Err = err
 		return
 	}

server/channels/app/app_iface.go

@@ -430,6 +430,7 @@ type AppIface interface {
 	AddDirectChannels(c request.CTX, teamID string, user *model.User) *model.AppError
 	AddLdapPrivateCertificate(fileData *multipart.FileHeader) *model.AppError
 	AddLdapPublicCertificate(fileData *multipart.FileHeader) *model.AppError
+	AddLicenseListener(listener func(oldLicense, newLicense *model.License)) string
 	AddRemoteCluster(rc *model.RemoteCluster) (*model.RemoteCluster, *model.AppError)
 	AddSamlIdpCertificate(fileData *multipart.FileHeader) *model.AppError
 	AddSamlPrivateCertificate(fileData *multipart.FileHeader) *model.AppError
@@ -988,7 +989,7 @@ type AppIface interface {
 	Publish(message *model.WebSocketEvent)
 	PublishUserTyping(userID, channelID, parentId string) *model.AppError
 	PurgeBleveIndexes(c request.CTX) *model.AppError
-	PurgeElasticsearchIndexes(c request.CTX) *model.AppError
+	PurgeElasticsearchIndexes(c request.CTX, indexes []string) *model.AppError
 	QueryLogs(rctx request.CTX, page, perPage int, logFilter *model.LogFilter) (map[string][]string, *model.AppError)
 	ReadFile(path string) ([]byte, *model.AppError)
 	RecycleDatabaseConnection(rctx request.CTX)

server/channels/app/cloud.go

@@ -16,16 +16,6 @@ import (
 	"github.com/mattermost/mattermost/server/v8/channels/store"
 )
 
-func (a *App) getSysAdminsEmailRecipients() ([]*model.User, *model.AppError) {
-	userOptions := &model.UserGetOptions{
-		Page:     0,
-		PerPage:  100,
-		Role:     model.SystemAdminRoleId,
-		Inactive: false,
-	}
-	return a.GetUsersFromProfiles(userOptions)
-}
-
 func getCurrentPlanName(a *App) (string, *model.AppError) {
 	subscription, err := a.Cloud().GetSubscription("")
 	if err != nil {
@@ -48,7 +38,7 @@ func getCurrentPlanName(a *App) (string, *model.AppError) {
 }
 
 func (a *App) SendPaymentFailedEmail(failedPayment *model.FailedPayment) *model.AppError {
-	sysAdmins, err := a.getSysAdminsEmailRecipients()
+	sysAdmins, err := a.getAllSystemAdmins()
 	if err != nil {
 		return err
 	}
@@ -77,7 +67,7 @@ func getCurrentProduct(subscriptionProductID string, products []*model.Product)
 }
 
 func (a *App) SendDelinquencyEmail(emailToSend model.DelinquencyEmail) *model.AppError {
-	sysAdmins, aErr := a.getSysAdminsEmailRecipients()
+	sysAdmins, aErr := a.getAllSystemAdmins()
 	if aErr != nil {
 		return aErr
 	}
@@ -161,7 +151,7 @@ func getNextBillingDateString() string {
 }
 
 func (a *App) SendUpgradeConfirmationEmail(isYearly bool) *model.AppError {
-	sysAdmins, e := a.getSysAdminsEmailRecipients()
+	sysAdmins, e := a.getAllSystemAdmins()
 	if e != nil {
 		return e
 	}
@@ -220,7 +210,7 @@ func (a *App) SendUpgradeConfirmationEmail(isYearly bool) *model.AppError {
 
 // SendNoCardPaymentFailedEmail
 func (a *App) SendNoCardPaymentFailedEmail() *model.AppError {
-	sysAdmins, err := a.getSysAdminsEmailRecipients()
+	sysAdmins, err := a.getAllSystemAdmins()
 	if err != nil {
 		return err
 	}
@@ -323,7 +313,7 @@ func (a *App) DoSubscriptionRenewalCheck() {
 		return
 	}
 
-	sysAdmins, aErr := a.getSysAdminsEmailRecipients()
+	sysAdmins, aErr := a.getAllSystemAdmins()
 	if aErr != nil {
 		a.Log().Error("Error getting sys admins", mlog.Err(aErr))
 		return

server/channels/app/config.go

@@ -64,6 +64,10 @@ func (a *App) AddConfigListener(listener func(*model.Config, *model.Config)) str
 	return a.Srv().platform.AddConfigListener(listener)
 }
 
+func (a *App) AddLicenseListener(listener func(oldLicense, newLicense *model.License)) string {
+	return a.Srv().platform.AddLicenseListener(listener)
+}
+
 // Removes a listener function by the unique ID returned when AddConfigListener was called
 func (a *App) RemoveConfigListener(id string) {
 	a.Srv().platform.RemoveConfigListener(id)

server/channels/app/elasticsearch.go

@@ -0,0 +1,142 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package app
+
+import (
+	"errors"
+	"net/url"
+
+	"github.com/mattermost/mattermost/server/public/model"
+	"github.com/mattermost/mattermost/server/public/shared/i18n"
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/v8/channels/utils"
+)
+
+func (a *App) initElasticsearchChannelIndexCheck() {
+	// the logic of when to perform the check has been derived from platform/searchengine.StartSearchEngine()
+	// Wherever we're starting the engine, we're checking the index mapping here.
+
+	a.elasticsearchChannelIndexCheckWithRetry()
+
+	a.AddConfigListener(func(oldConfig, newConfig *model.Config) {
+		if a.SearchEngine().ElasticsearchEngine == nil {
+			return
+		}
+
+		oldESConfig := oldConfig.ElasticsearchSettings
+		newESConfig := newConfig.ElasticsearchSettings
+
+		// if indexing is turned on, check.
+		if !*oldESConfig.EnableIndexing && *newESConfig.EnableIndexing {
+			a.elasticsearchChannelIndexCheckWithRetry()
+		} else if *newESConfig.EnableIndexing && (*oldESConfig.Password != *newESConfig.Password || *oldESConfig.Username != *newESConfig.Username || *oldESConfig.ConnectionURL != *newESConfig.ConnectionURL || *oldESConfig.Sniff != *newESConfig.Sniff) {
+			// ES client reconnects if credentials or address changes
+			a.elasticsearchChannelIndexCheckWithRetry()
+		}
+	})
+
+	a.AddLicenseListener(func(oldLicense, newLicense *model.License) {
+		if a.SearchEngine() == nil {
+			return
+		}
+
+		// if a license was added, and it has ES enabled-
+		if oldLicense == nil && newLicense != nil {
+			if a.SearchEngine().ElasticsearchEngine != nil {
+				a.elasticsearchChannelIndexCheckWithRetry()
+			}
+		}
+	})
+}
+
+func (a *App) elasticsearchChannelIndexCheckWithRetry() {
+	// this is being done async to not block license application and config
+	// processes as the listeners for those are called synchronously.
+	go func() {
+		// using progressive retry because ES client may take some time to connect and be ready.
+		_ = utils.LongProgressiveRetry(func() error {
+			if !*a.Config().ElasticsearchSettings.EnableIndexing {
+				a.Log().Debug("elasticsearchChannelIndexCheckWithRetry: skipping because elasticsearch indexing is disabled")
+				return nil
+			}
+
+			elastic := a.SearchEngine().ElasticsearchEngine
+			if elastic == nil {
+				a.Log().Debug("elasticsearchChannelIndexCheckWithRetry: skipping because elastic engine is nil")
+				return errors.New("retry")
+			}
+
+			if !elastic.IsActive() {
+				a.Log().Debug("elasticsearchChannelIndexCheckWithRetry: skipping because elastic.IsActive is false")
+				return errors.New("retry")
+			}
+
+			a.elasticsearchChannelIndexCheck()
+			return nil
+		})
+	}()
+}
+
+func (a *App) elasticsearchChannelIndexCheck() {
+	if needNotify := a.elasticChannelsIndexNeedNotifyAdmins(); !needNotify {
+		return
+	}
+
+	// notify all system admins
+	systemBot, appErr := a.GetSystemBot(request.EmptyContext(a.Log()))
+	if appErr != nil {
+		a.Log().Error("elasticsearchChannelIndexCheck: couldn't get system bot", mlog.Err(appErr))
+		return
+	}
+
+	sysAdmins, appErr := a.getAllSystemAdmins()
+	if appErr != nil {
+		a.Log().Error("elasticsearchChannelIndexCheck: error occurred fetching all system admins", mlog.Err(appErr))
+	}
+
+	elasticsearchSettingsSectionLink, err := url.JoinPath(*a.Config().ServiceSettings.SiteURL, "admin_console/environment/elasticsearch")
+	if err != nil {
+		a.Log().Error("elasticsearchChannelIndexCheck: error occurred constructing Elasticsearch system console section path")
+		return
+	}
+
+	// TODO include a link to changelog
+	postMessage := i18n.T("app.channel.elasticsearch_channel_index.notify_admin.message", map[string]interface{}{"ElasticsearchSection": elasticsearchSettingsSectionLink})
+
+	for _, sysAdmin := range sysAdmins {
+		var channel *model.Channel
+		channel, appErr = a.GetOrCreateDirectChannel(request.EmptyContext(a.Log()), sysAdmin.Id, systemBot.UserId)
+		if appErr != nil {
+			a.Log().Error("elasticsearchChannelIndexCheck: error occurred ensuring DM channel between system bot and sys admin", mlog.Err(appErr))
+			continue
+		}
+
+		post := &model.Post{
+			Message:   postMessage,
+			UserId:    systemBot.UserId,
+			ChannelId: channel.Id,
+		}
+		_, appErr = a.CreatePost(request.EmptyContext(a.Log()), post, channel, true, false)
+		if appErr != nil {
+			a.Log().Error("elasticsearchChannelIndexCheck: error occurred creating post", mlog.Err(appErr))
+			continue
+		}
+	}
+}
+
+func (a *App) elasticChannelsIndexNeedNotifyAdmins() bool {
+	elastic := a.SearchEngine().ElasticsearchEngine
+	if elastic == nil {
+		a.Log().Debug("elasticChannelsIndexNeedNotifyAdmins: skipping because elastic engine is nil")
+		return false
+	}
+
+	if elastic.IsChannelsIndexVerified() {
+		a.Log().Debug("elasticChannelsIndexNeedNotifyAdmins: skipping because channels index is verified")
+		return false
+	}
+
+	return true
+}

server/channels/app/enterprise.go

@@ -50,12 +50,6 @@ 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) {

server/channels/app/migrations.go

@@ -554,31 +554,6 @@ func (s *Server) doPostPriorityConfigDefaultTrueMigration() {
 	}
 }
 
-func (s *Server) doElasticsearchFixChannelIndex(c request.CTX) {
-	s.AddLicenseListener(func(oldLicense, newLicense *model.License) {
-		s.elasticsearchFixChannelIndex(c, newLicense)
-	})
-
-	s.elasticsearchFixChannelIndex(c, s.License())
-}
-
-func (s *Server) elasticsearchFixChannelIndex(c request.CTX, license *model.License) {
-	if model.BuildEnterpriseReady != "true" || license == nil || !*license.Features.Elasticsearch {
-		mlog.Debug("Skipping triggering Elasticsearch channel index fix job as build is not Enterprise ready")
-		return
-	}
-
-	// If the migration is already marked as completed, don't do it again.
-	if _, err := s.Store().System().GetByName(model.MigrationKeyElasticsearchFixChannelIndex); err == nil {
-		mlog.Debug("Skipping triggering Elasticsearch channel index fix job as it is already marked completed in database")
-		return
-	}
-
-	if _, appErr := s.Jobs.CreateJobOnce(c, model.JobTypeElasticsearchFixChannelIndex, nil); appErr != nil {
-		mlog.Fatal("failed to start job for fixing Elasticsearch channels index", mlog.Err(appErr))
-	}
-}
-
 func (s *Server) doCloudS3PathMigrations(c request.CTX) {
 	// This migration is only applicable for cloud environments
 	if os.Getenv("MM_CLOUD_FILESTORE_BIFROST") == "" {
@@ -672,7 +647,6 @@ func (s *Server) doAppMigrations() {
 	s.doFirstAdminSetupCompleteMigration()
 	s.doRemainingSchemaMigrations()
 	s.doPostPriorityConfigDefaultTrueMigration()
-	s.doElasticsearchFixChannelIndex(c)
 	s.doCloudS3PathMigrations(c)
 	s.doDeleteEmptyDraftsMigration(c)
 	s.doDeleteOrphanDraftsMigration(c)

server/channels/app/opentracing/opentracing_layer.go

@@ -247,6 +247,23 @@ func (a *OpenTracingAppLayer) AddLdapPublicCertificate(fileData *multipart.FileH
 	return resultVar0
 }
 
+func (a *OpenTracingAppLayer) AddLicenseListener(listener func(oldLicense, newLicense *model.License)) string {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.AddLicenseListener")
+
+	a.ctx = newCtx
+	a.app.Srv().Store().SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store().SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0 := a.app.AddLicenseListener(listener)
+
+	return resultVar0
+}
+
 func (a *OpenTracingAppLayer) AddPublicKey(name string, key io.Reader) *model.AppError {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.AddPublicKey")
@@ -13818,7 +13835,7 @@ func (a *OpenTracingAppLayer) PurgeBleveIndexes(c request.CTX) *model.AppError {
 	return resultVar0
 }
 
-func (a *OpenTracingAppLayer) PurgeElasticsearchIndexes(c request.CTX) *model.AppError {
+func (a *OpenTracingAppLayer) PurgeElasticsearchIndexes(c request.CTX, indexes []string) *model.AppError {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.PurgeElasticsearchIndexes")
 
@@ -13830,7 +13847,7 @@ func (a *OpenTracingAppLayer) PurgeElasticsearchIndexes(c request.CTX) *model.Ap
 	}()
 
 	defer span.Finish()
-	resultVar0 := a.app.PurgeElasticsearchIndexes(c)
+	resultVar0 := a.app.PurgeElasticsearchIndexes(c, indexes)
 
 	if resultVar0 != nil {
 		span.LogFields(spanlog.Error(resultVar0))

server/channels/app/searchengine.go

@@ -36,18 +36,21 @@ func (a *App) SetSearchEngine(se *searchengine.Broker) {
 	a.ch.srv.platform.SearchEngine = se
 }
 
-func (a *App) PurgeElasticsearchIndexes(c request.CTX) *model.AppError {
+func (a *App) PurgeElasticsearchIndexes(c request.CTX, indexes []string) *model.AppError {
 	engine := a.SearchEngine().ElasticsearchEngine
 	if engine == nil {
 		err := model.NewAppError("PurgeElasticsearchIndexes", "ent.elasticsearch.test_config.license.error", nil, "", http.StatusNotImplemented)
 		return err
 	}
 
-	if err := engine.PurgeIndexes(c); err != nil {
-		return err
+	var appErr *model.AppError
+	if len(indexes) > 0 {
+		appErr = engine.PurgeIndexList(c, indexes)
+	} else {
+		appErr = engine.PurgeIndexes(c)
 	}
 
-	return nil
+	return appErr
 }
 
 func (a *App) PurgeBleveIndexes(c request.CTX) *model.AppError {

server/channels/app/server.go

@@ -490,6 +490,8 @@ func NewServer(options ...Option) (*Server, error) {
 		}
 	})
 
+	app.initElasticsearchChannelIndexCheck()
+
 	return s, nil
 }
 
@@ -1479,11 +1481,6 @@ 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())

server/channels/app/user.go

@@ -2864,3 +2864,13 @@ func (a *App) UserIsFirstAdmin(user *model.User) bool {
 
 	return true
 }
+
+func (a *App) getAllSystemAdmins() ([]*model.User, *model.AppError) {
+	userOptions := &model.UserGetOptions{
+		Page:     0,
+		PerPage:  500,
+		Role:     model.SystemAdminRoleId,
+		Inactive: false,
+	}
+	return a.GetUsersFromProfiles(userOptions)
+}

server/channels/utils/backoff.go

@@ -7,10 +7,20 @@ import (
 	"time"
 )
 
-var backoffTimeouts = []time.Duration{50 * time.Millisecond, 100 * time.Millisecond, 200 * time.Millisecond, 200 * time.Millisecond, 400 * time.Millisecond, 400 * time.Millisecond}
+var shortBackoffTimeouts = []time.Duration{50 * time.Millisecond, 100 * time.Millisecond, 200 * time.Millisecond, 200 * time.Millisecond, 400 * time.Millisecond, 400 * time.Millisecond}
+
+var longBackoffTimeouts = []time.Duration{500 * time.Millisecond, 1 * time.Second, 2 * time.Second, 5 * time.Second, 8 * time.Second, 10 * time.Second}
 
 // ProgressiveRetry executes a BackoffOperation and waits an increasing time before retrying the operation.
 func ProgressiveRetry(operation func() error) error {
+	return CustomProgressiveRetry(operation, shortBackoffTimeouts)
+}
+
+func LongProgressiveRetry(operation func() error) error {
+	return CustomProgressiveRetry(operation, longBackoffTimeouts)
+}
+
+func CustomProgressiveRetry(operation func() error, backoffTimeouts []time.Duration) error {
 	var err error
 
 	for attempts := 0; attempts < len(backoffTimeouts); attempts++ {