Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Bugfix/reduce number of docs synced bulk #2396

Merged
merged 11 commits into from
May 10, 2024
Merged

Bugfix/reduce number of docs synced bulk #2396

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
c6d4fae
reduce number of docs synced in a single bulk request to 100 while sy…
epipav May 8, 2024
6109f1f
remove dd measure temporarily
epipav May 8, 2024
395803b
logging
epipav May 8, 2024
8b792a1
more logs
epipav May 8, 2024
d954b49
more logs
epipav May 8, 2024
e9b6fe1
testing
epipav May 8, 2024
055b101
flushing logic error fixed
epipav May 8, 2024
fadb4e7
cleaned logs
epipav May 8, 2024
d7f5d1a
do the fix also for orgs
epipav May 8, 2024
f8e1899
reduce number of docs sent per bulk request for orgs to 100
epipav May 8, 2024
62b1364
Merge branch 'crowd-linux' into bugfix/reduce-number-of-docs-synced-bulk
epipav May 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
22 changes: 8 additions & 14 deletions services/libs/opensearch/src/service/member.sync.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,7 +286,7 @@ export class MemberSyncService {

public async syncMembers(memberIds: string[], segmentIds?: string[]): Promise<IMemberSyncResult> {
const CONCURRENT_DATABASE_QUERIES = 10
const BULK_INDEX_DOCUMENT_BATCH_SIZE = 2500
const BULK_INDEX_DOCUMENT_BATCH_SIZE = 100

// get all memberId-segmentId couples
const memberSegmentCouples: IMemberSegmentMatrix =
Expand Down Expand Up @@ -399,7 +399,9 @@ export class MemberSyncService {
})

// databaseStreams will create syncStreams items in processSegmentsStream, which'll later be used to sync to opensearch in bulk
if (databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
const isLastSegment = i === totalMemberIds - 1 && j === totalSegments - 1

if (isLastSegment || databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
await processSegmentsStream(databaseStream)
databaseStream = []
}
Expand All @@ -413,18 +415,10 @@ export class MemberSyncService {
syncStream = syncStream.slice(BULK_INDEX_DOCUMENT_BATCH_SIZE)
}

// if we're processing the last segment
if (i === totalMemberIds - 1 && j === totalSegments - 1) {
// check if there are remaining databaseStreams to process
if (databaseStream.length > 0) {
await processSegmentsStream(databaseStream)
}

// check if there are remaining syncStreams to process
if (syncStream.length > 0) {
await this.openSearchService.bulkIndex(OpenSearchIndex.MEMBERS, syncStream)
documentsIndexed += syncStream.length
}
// check if there are remaining syncStreams to process
if (isLastSegment && syncStream.length > 0) {
await this.openSearchService.bulkIndex(OpenSearchIndex.MEMBERS, syncStream)
documentsIndexed += syncStream.length
}
successfullySyncedMembers.push(memberId)
}
Expand Down
22 changes: 8 additions & 14 deletions services/libs/opensearch/src/service/organization.sync.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ export class OrganizationSyncService {
segmentIds?: string[],
): Promise<IOrganizationSyncResult> {
const CONCURRENT_DATABASE_QUERIES = 5
const BULK_INDEX_DOCUMENT_BATCH_SIZE = 2500
const BULK_INDEX_DOCUMENT_BATCH_SIZE = 100

// get all orgId-segmentId couples
const orgSegmentCouples: IOrganizationSegmentMatrix =
Expand Down Expand Up @@ -360,7 +360,9 @@ export class OrganizationSyncService {
})

// databaseStreams will create syncStreams items in processSegmentsStream, which'll later be used to sync to opensearch in bulk
if (databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
const isLastSegment = i === totalOrgIds - 1 && j === totalSegments - 1

if (isLastSegment || databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
await processSegmentsStream(databaseStream)
databaseStream = []
}
Expand All @@ -374,18 +376,10 @@ export class OrganizationSyncService {
syncStream = syncStream.slice(BULK_INDEX_DOCUMENT_BATCH_SIZE)
}

// if we're processing the last segment
if (i === totalOrgIds - 1 && j === totalSegments - 1) {
// check if there are remaining databaseStreams to process
if (databaseStream.length > 0) {
await processSegmentsStream(databaseStream)
}

// check if there are remaining syncStreams to process
if (syncStream.length > 0) {
await this.openSearchService.bulkIndex(OpenSearchIndex.ORGANIZATIONS, syncStream)
documentsIndexed += syncStream.length
}
// check if there are remaining syncStreams to process
if (isLastSegment && syncStream.length > 0) {
await this.openSearchService.bulkIndex(OpenSearchIndex.ORGANIZATIONS, syncStream)
documentsIndexed += syncStream.length
}
}
}
Expand Down