team/server/enterprise/elasticsearch/opensearch/bulk_test.go at release-11.0

hanzoai/team

Fork 0

mirror of https://github.com/hanzoai/team.git synced 2026-10-30 14:49:35 +00:00

Files

T

24045f5b7c MM-60441: Re-index public channels when a user joins a team (#33400 )

* Index all public channels when a user joins a team

* Precompute team members for indexChannelsForTeam

* Refactor RequestContextWithMaster to store package

This way, we can import it from both the sqlstore and the searchlayer
packages. The alternative for this is duplicating the code in those two
packages, but that will *not* work:

The context package expects custom types for the keys stored in it, so
that different packages never clash with each other when trying to
register a new key. See the docs for the WithValue function:
https://pkg.go.dev/context#WithValue

If we try to duplicate the storeContextKey type in both the sqlstore and
searchlayer packages, although they *look* the same, they are not, and
HasMaster will fail to get the value of the storeContextKey(useMaster)
key if it's from the other package.

* Use master in call to GetTeamMembersForChannel

In GetTeamMembersForChannel, use the DB from the newly passed context,
which will be the receiving context everywhere except in the call done
from indexChannelsForTeam, to avoid the read after write issue when
saving a team member.

* Fix GetPublicChannelsForTeam paging

We were using the page and perPage arguments as is in the call to
GetPublicChannelsForTeam, but that function expects and offset and a
limit as understood by SQL. Although perPage and limit are
interchangeable, offset is not equal to page, but to page * perPage.

* Add a synchronous bulk indexer for Opensearch

* Implement Opensearch's SyncBulkIndexChannels

* Add a synchronous bulk indexer for Elasticsearch

* Implement Elasticsearch's SynkBulkIndexChannels

* Test SyncBulkIndexChannels

* make mocks

* Bulk index channels on indexChannelsForTeam

* Handle error from SyncBulkIndexChannels

* Fix style

* Revert indexChannelWithTeamMembers refactor

* Remove defensive code on sync bulk processor

* Revert "Add a synchronous bulk indexer for Opensearch"

This reverts commit bfe4671d96bffa9ca27ed3c655fc5527b72bbafb.

* Revert "Add a synchronous bulk indexer for Elasticsearch"

This reverts commit 6643ae3f30c461544d0861aec7dee1f24e507c37.

* Refactor bulk indexers with a common interface

* Test all the different implementations

Assisted by Claude

* Remove debug statements

* Refactor common code into _stop

* Rename getUserIDsFor{,Private}Channel

* Wrap error

* Make perPage a const

* Fix typos

* Call GetTeamsForUser only if needed

* Differentiate errors for sync/async processors

---------

Co-authored-by: Ibrahim Serdar Acikgoz <[email protected]>
Co-authored-by: Mattermost Build <[email protected]>

2025-08-25 19:28:19 +02:00

483 lines

14 KiB

Go

Raw Permalink Blame History

 // Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
 // See LICENSE.enterprise for license information.
 package opensearch
 import (
 	"fmt"
 	"os"
 	"testing"
 	"time"
 	"github.com/elastic/go-elasticsearch/v8/typedapi/types"
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/mattermost/mattermost/server/v8/channels/api4"
 	"github.com/mattermost/mattermost/server/v8/enterprise/elasticsearch/common"
 	"github.com/stretchr/testify/require"
 )
 // setupBulkClient creates a test bulk client with common setup
 func setupBulkClient(t *testing.T, flushBytes int, flushNumReqs int, flushInterval time.Duration) (*Bulk, *api4.TestHelper) {
 	th := api4.SetupEnterprise(t)
 	if os.Getenv("IS_CI") == "true" {
 		os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://opensearch:9201")
 		os.Setenv("MM_ELASTICSEARCHSETTINGS_BACKEND", "opensearch")
 	}
 	th.App.UpdateConfig(func(cfg *model.Config) {
 		if os.Getenv("IS_CI") == "true" {
 			*cfg.ElasticsearchSettings.ConnectionURL = "http://opensearch:9201"
 		} else {
 			*cfg.ElasticsearchSettings.ConnectionURL = "http://localhost:9201"
 		}
 		*cfg.ElasticsearchSettings.Backend = model.ElasticsearchSettingsOSBackend
 		*cfg.ElasticsearchSettings.EnableIndexing = true
 		*cfg.ElasticsearchSettings.EnableSearching = true
 		*cfg.ElasticsearchSettings.EnableAutocomplete = true
 	})
 	client := createTestClient(t, th.Context, th.App.Config(), th.App.FileBackend())
 	bulk := NewBulk(
 		common.BulkSettings{
 			FlushBytes:    flushBytes,
 			FlushInterval: flushInterval,
 			FlushNumReqs:  flushNumReqs,
 		},
 		client,
 		time.Duration(*th.App.Config().ElasticsearchSettings.RequestTimeoutSeconds)*time.Second,
 		th.Server.Platform().Log())
 	return bulk, th
 }
 // createTestPost creates a test post for indexing
 func createTestPost(t *testing.T, message string) *common.ESPost {
 	post, err := common.ESPostFromPost(&model.Post{
 		Id:      model.NewId(),
 		Message: message,
 	}, "myteam")
 	require.NoError(t, err)
 	return post
 }
 func TestBulkProcessor(t *testing.T) {
 	bulk, th := setupBulkClient(t, 0, 10, 0)
 	defer th.TearDown()
 	defer func() {
 		if os.Getenv("IS_CI") == "true" {
 			os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 			os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 		}
 	}()
 	post := createTestPost(t, "hello world")
 	err := bulk.IndexOp(&types.IndexOperation{
 		Index_: model.NewPointer("myindex"),
 		Id_:    model.NewPointer(post.Id),
 	}, post)
 	require.NoError(t, err)
 	require.Equal(t, 1, bulk.pendingRequests)
 	err = bulk.Stop()
 	require.NoError(t, err)
 	require.Equal(t, 0, bulk.pendingRequests)
 }
 func TestNewBulk(t *testing.T) {
 	bulk, th := setupBulkClient(t, 1024, 10, 0)
 	defer th.TearDown()
 	defer func() {
 		if os.Getenv("IS_CI") == "true" {
 			os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 			os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 		}
 	}()
 	t.Run("creates bulk client without periodic flusher", func(t *testing.T) {
 		require.NotNil(t, bulk)
 		require.NotNil(t, bulk.client)
 		require.NotNil(t, bulk.logger)
 		require.NotNil(t, bulk.buf)
 		require.Equal(t, 0, bulk.pendingRequests)
 		require.Equal(t, 1024, bulk.bulkSettings.FlushBytes)
 		require.Equal(t, 10, bulk.bulkSettings.FlushNumReqs)
 	})
 	t.Run("creates bulk client with periodic flusher", func(t *testing.T) {
 		bulkWithTimer, th2 := setupBulkClient(t, 1024, 10, 100*time.Millisecond)
 		defer th2.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		require.NotNil(t, bulkWithTimer)
 		require.Equal(t, 100*time.Millisecond, bulkWithTimer.bulkSettings.FlushInterval)
 		err := bulkWithTimer.Stop()
 		require.NoError(t, err)
 	})
 	err := bulk.Stop()
 	require.NoError(t, err)
 }
 func TestIndexOp(t *testing.T) {
 	bulk, th := setupBulkClient(t, 0, 10, 0)
 	defer th.TearDown()
 	defer func() {
 		if os.Getenv("IS_CI") == "true" {
 			os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 			os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 		}
 	}()
 	t.Run("single index operation with struct", func(t *testing.T) {
 		post := createTestPost(t, "test message")
 		err := bulk.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(post.Id),
 		}, post)
 		require.NoError(t, err)
 		require.Equal(t, 1, bulk.pendingRequests)
 		// Verify buffer has content
 		require.Greater(t, bulk.buf.Len(), 0)
 	})
 	t.Run("index operation with []byte", func(t *testing.T) {
 		initialRequests := bulk.pendingRequests
 		docId := model.NewId()
 		data := []byte(`{"message": "test byte slice"}`)
 		err := bulk.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(docId),
 		}, data)
 		require.NoError(t, err)
 		require.Equal(t, initialRequests+1, bulk.pendingRequests)
 	})
 	t.Run("index operation with json.RawMessage", func(t *testing.T) {
 		initialRequests := bulk.pendingRequests
 		docId := model.NewId()
 		jsonData := []byte(`{"message": "test raw message"}`)
 		err := bulk.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(docId),
 		}, jsonData)
 		require.NoError(t, err)
 		require.Equal(t, initialRequests+1, bulk.pendingRequests)
 	})
 	t.Run("multiple index operations", func(t *testing.T) {
 		initialRequests := bulk.pendingRequests
 		for i := range 5 {
 			post := createTestPost(t, fmt.Sprintf("test message %d", i))
 			err := bulk.IndexOp(&types.IndexOperation{
 				Index_: model.NewPointer("testindex"),
 				Id_:    model.NewPointer(post.Id),
 			}, post)
 			require.NoError(t, err)
 		}
 		require.Equal(t, initialRequests+5, bulk.pendingRequests)
 	})
 	t.Run("auto flush on request threshold", func(t *testing.T) {
 		// Create a new client with low flush threshold
 		bulk2, th2 := setupBulkClient(t, 0, 2, 0)
 		defer th2.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		post1 := createTestPost(t, "first message")
 		err := bulk2.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(post1.Id),
 		}, post1)
 		require.NoError(t, err)
 		require.Equal(t, 1, bulk2.pendingRequests)
 		post2 := createTestPost(t, "second message")
 		err = bulk2.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(post2.Id),
 		}, post2)
 		require.NoError(t, err)
 		require.Equal(t, 2, bulk2.pendingRequests)
 		// Third operation should trigger flush
 		post3 := createTestPost(t, "third message")
 		err = bulk2.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(post3.Id),
 		}, post3)
 		require.NoError(t, err)
 		require.Equal(t, 0, bulk2.pendingRequests)
 		err = bulk2.Stop()
 		require.NoError(t, err)
 	})
 	err := bulk.Stop()
 	require.NoError(t, err)
 }
 func TestDeleteOp(t *testing.T) {
 	bulk, th := setupBulkClient(t, 0, 10, 0)
 	defer th.TearDown()
 	defer func() {
 		if os.Getenv("IS_CI") == "true" {
 			os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 			os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 		}
 	}()
 	t.Run("single delete operation", func(t *testing.T) {
 		docId := model.NewId()
 		err := bulk.DeleteOp(&types.DeleteOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(docId),
 		})
 		require.NoError(t, err)
 		require.Equal(t, 1, bulk.pendingRequests)
 		// Verify buffer has content
 		require.Greater(t, bulk.buf.Len(), 0)
 	})
 	t.Run("multiple delete operations", func(t *testing.T) {
 		initialRequests := bulk.pendingRequests
 		for range 3 {
 			docId := model.NewId()
 			err := bulk.DeleteOp(&types.DeleteOperation{
 				Index_: model.NewPointer("testindex"),
 				Id_:    model.NewPointer(docId),
 			})
 			require.NoError(t, err)
 		}
 		require.Equal(t, initialRequests+3, bulk.pendingRequests)
 	})
 	t.Run("auto flush on request threshold", func(t *testing.T) {
 		// Create a new client with low flush threshold
 		bulk2, th2 := setupBulkClient(t, 0, 2, 0)
 		defer th2.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		// Add two delete operations
 		for range 2 {
 			docId := model.NewId()
 			err := bulk2.DeleteOp(&types.DeleteOperation{
 				Index_: model.NewPointer("testindex"),
 				Id_:    model.NewPointer(docId),
 			})
 			require.NoError(t, err)
 		}
 		require.Equal(t, 2, bulk2.pendingRequests)
 		// Third operation should trigger flush
 		docId := model.NewId()
 		err := bulk2.DeleteOp(&types.DeleteOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(docId),
 		})
 		require.NoError(t, err)
 		require.Equal(t, 0, bulk2.pendingRequests)
 		err = bulk2.Stop()
 		require.NoError(t, err)
 	})
 	err := bulk.Stop()
 	require.NoError(t, err)
 }
 func TestFlush(t *testing.T) {
 	bulk, th := setupBulkClient(t, 0, 10, 0)
 	defer th.TearDown()
 	defer func() {
 		if os.Getenv("IS_CI") == "true" {
 			os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 			os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 		}
 	}()
 	t.Run("flush with pending operations", func(t *testing.T) {
 		post := createTestPost(t, "test message")
 		err := bulk.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(post.Id),
 		}, post)
 		require.NoError(t, err)
 		require.Equal(t, 1, bulk.pendingRequests)
 		err = bulk.Flush()
 		require.NoError(t, err)
 		require.Equal(t, 0, bulk.pendingRequests)
 		// Verify buffer is empty after flush
 		require.Equal(t, 0, bulk.buf.Len())
 	})
 	t.Run("flush with no pending operations", func(t *testing.T) {
 		require.Equal(t, 0, bulk.pendingRequests)
 		err := bulk.Flush()
 		require.NoError(t, err)
 		require.Equal(t, 0, bulk.pendingRequests)
 	})
 	err := bulk.Stop()
 	require.NoError(t, err)
 }
 func TestStop(t *testing.T) {
 	t.Run("stop with pending operations", func(t *testing.T) {
 		bulk, th := setupBulkClient(t, 0, 10, 0)
 		defer th.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		post := createTestPost(t, "test message")
 		err := bulk.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(post.Id),
 		}, post)
 		require.NoError(t, err)
 		require.Equal(t, 1, bulk.pendingRequests)
 		err = bulk.Stop()
 		require.NoError(t, err)
 		require.Equal(t, 0, bulk.pendingRequests)
 	})
 	t.Run("stop with no pending operations", func(t *testing.T) {
 		bulk, th := setupBulkClient(t, 0, 10, 0)
 		defer th.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		require.Equal(t, 0, bulk.pendingRequests)
 		err := bulk.Stop()
 		require.NoError(t, err)
 		require.Equal(t, 0, bulk.pendingRequests)
 	})
 	t.Run("stop with periodic flusher", func(t *testing.T) {
 		bulk, th := setupBulkClient(t, 0, 10, 100*time.Millisecond)
 		defer th.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		post := createTestPost(t, "test message")
 		err := bulk.IndexOp(&types.IndexOperation{
 			Index_: model.NewPointer("testindex"),
 			Id_:    model.NewPointer(post.Id),
 		}, post)
 		require.NoError(t, err)
 		require.Equal(t, 1, bulk.pendingRequests)
 		// Stop should flush pending operations and stop the periodic flusher
 		err = bulk.Stop()
 		require.NoError(t, err)
 		require.Equal(t, 0, bulk.pendingRequests)
 	})
 }
 func TestFlushThresholds(t *testing.T) {
 	t.Run("flush on bytes threshold", func(t *testing.T) {
 		// Create a client with very small byte threshold
 		bulk, th := setupBulkClient(t, 100, 0, 0) // 100 bytes threshold
 		defer th.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		// Add operations that should exceed the byte threshold
 		for range 5 {
 			post := createTestPost(t, "This is a long message that should help us exceed the byte threshold for testing")
 			err := bulk.IndexOp(&types.IndexOperation{
 				Index_: model.NewPointer("testindex"),
 				Id_:    model.NewPointer(post.Id),
 			}, post)
 			require.NoError(t, err)
 		}
 		// Should have been flushed due to byte threshold
 		require.Equal(t, 0, bulk.pendingRequests)
 		err := bulk.Stop()
 		require.NoError(t, err)
 	})
 	t.Run("no flush when thresholds not met", func(t *testing.T) {
 		bulk, th := setupBulkClient(t, 100000, 10, 0) // High thresholds
 		defer th.TearDown()
 		defer func() {
 			if os.Getenv("IS_CI") == "true" {
 				os.Setenv("MM_ELASTICSEARCHSETTINGS_CONNECTIONURL", "http://elasticsearch:9201")
 				os.Unsetenv("MM_ELASTICSEARCHSETTINGS_BACKEND")
 			}
 		}()
 		// Add a few operations that shouldn't trigger flush
 		for range 3 {
 			post := createTestPost(t, "short")
 			err := bulk.IndexOp(&types.IndexOperation{
 				Index_: model.NewPointer("testindex"),
 				Id_:    model.NewPointer(post.Id),
 			}, post)
 			require.NoError(t, err)
 			fmt.Println("PENDING REQS:", bulk.pendingRequests)
 		}
 		// Should not have been flushed
 		require.Equal(t, 3, bulk.pendingRequests)
 		err := bulk.Stop()
 		require.NoError(t, err)
 	})
 }

483 lines 14 KiB Go Raw Permalink Blame History

483 lines

14 KiB

Go

Raw Permalink Blame History