diff --git a/build.gradle b/build.gradle index 6cd0702..934b78b 100644 --- a/build.gradle +++ b/build.gradle @@ -53,7 +53,7 @@ dependencies { tasks.named('test') { useJUnitPlatform { - excludeTags 'benchmark', 'minio-integration', 'claim-concurrency', 'local-e2e', 'vector-search-performance', 'worker-indexing-throughput', 'worker-horizontal-scaling', 'worker-queue-backpressure' + excludeTags 'benchmark', 'minio-integration', 'claim-concurrency', 'local-e2e', 'vector-search-performance', 'worker-indexing-throughput', 'worker-horizontal-scaling', 'worker-queue-backpressure', 'document-indexing-e2e-load' } } @@ -219,6 +219,28 @@ tasks.register('workerHorizontalScalingTest', Test) { outputs.upToDateWhen { false } } +tasks.register('documentIndexingE2ELoadTest', Test) { + group = 'verification' + description = '실제 PDF·DOCX 50·100문서의 전체 인덱싱 처리량과 데이터 완전성을 측정합니다.' + testClassesDirs = sourceSets.test.output.classesDirs + classpath = sourceSets.test.runtimeClasspath + useJUnitPlatform { + includeTags 'document-indexing-e2e-load' + } + maxParallelForks = 1 + systemProperties System.properties.findAll { key, value -> + key.toString().startsWith('document.indexing.e2e.load.') + } + if (System.getProperty('document.indexing.e2e.load.output') == null) { + systemProperty( + 'document.indexing.e2e.load.output', + layout.buildDirectory.file('reports/document-indexing-e2e-load/document-indexing-e2e-load.json').get().asFile.absolutePath + ) + } + // 실제 외부 Service를 점유하고 큰 PDF·DOCX Queue를 만드는 장시간 Benchmark를 일반 Test와 분리한다. + outputs.upToDateWhen { false } +} + def configureOpenSqlDatabase = { Test task -> task.maxParallelForks = 1 task.outputs.upToDateWhen { false } diff --git a/docs/design/gimin-#143-pdf-docx-indexing-e2e-load-benchmark.md b/docs/design/gimin-#143-pdf-docx-indexing-e2e-load-benchmark.md new file mode 100644 index 0000000..8712cc7 --- /dev/null +++ b/docs/design/gimin-#143-pdf-docx-indexing-e2e-load-benchmark.md @@ -0,0 +1,178 @@ +# PDF·DOCX 전체 인덱싱 E2E 부하 Benchmark 설계 + +## 1. 배경 + +자동 Worker 전체 인덱싱 처리량 Benchmark는 실제 PostgreSQL 17, MinIO와 `BAAI/bge-m3`에서 +TXT 16·32문서의 기준선을 제공한다. 실제 PDF·DOCX는 소수 문서의 기능 E2E로 Parser와 Metadata를 +검증했지만, 여러 문서를 동시에 접수했을 때의 처리량과 전체 Pipeline 데이터 완전성은 측정하지 않았다. + +이번 작업은 실제 Text Layer PDF와 OOXML DOCX를 50·100문서 규모로 섞어 다음 질문에 답한다. + +1. Parser가 다른 PDF·DOCX 혼합 부하에서 전체 인덱싱 처리량과 지연은 어느 수준인가? +2. 형식별 Upload·Queue·처리·전체 지연에 의미 있는 차이가 있는가? +3. 큰 Queue를 모두 소진한 뒤 페이지·Section Metadata와 Vector 저장 불변식이 유지되는가? + +## 2. 범위 + +### 2.1 포함 + +- Memory에서 생성하는 실제 Text Layer PDF와 OOXML DOCX Binary +- 실제 인증과 Multipart HTTP 업로드 +- 실제 MinIO Object 저장과 Content-Type·크기 검증 +- 자동 Polling, Claim, Attempt, Parsing, Chunk 저장과 Lease 갱신 +- 실제 BGE-M3 Batch Embedding과 PostgreSQL `vector(1024)` 저장 +- 전체·PDF·DOCX별 처리량과 Upload·Queue·처리·E2E 지연 +- Document·Version·Job·Attempt·Event·현재 검색 Version 정합성 검증 +- PDF 페이지와 DOCX Section Metadata 보존 검증 +- Git 제외 JSON 원본 결과와 실행 결과 Markdown 기록 +- 일반 테스트와 분리한 전용 Gradle Task + +### 2.2 제외 + +- OCR과 스캔 PDF +- 구형 `.doc`와 HWP +- 운영 Admission Control과 Rate Limit +- 공식 OpenSQL 원격 장비의 절대 성능 판정 +- 제품 API·Entity·Migration 변경 + +## 3. Workload 계약 + +### 3.1 문서 Fixture + +- PDF는 두 페이지 Text Layer를 포함하고 페이지별 고유 주제 문구를 가진다. +- DOCX는 제목과 두 개 이상의 Heading·본문 Section을 포함한다. +- 문서별 고유 식별 문구 외에 형식별 본문 길이와 구조를 동일하게 유지한다. +- 파일 이름과 제목은 Profile, 반복, 형식과 순번을 포함해 충돌을 차단한다. +- PDF와 DOCX는 각 Profile에서 같은 개수로 섞는다. + +PDF는 PDFBox, DOCX는 Apache POI로 생성한다. 저장소 Fixture Binary를 추가하지 않고도 실제 Parser +경계를 통과하며, PDF는 OCR이 필요 없는 Text Layer만 검증한다. + +### 3.2 Profile + +| 구분 | 전체 문서 | PDF | DOCX | 반복 | 통계 포함 | +|---|---:|---:|---:|---:|---| +| Warm-up | 4 | 2 | 2 | 1 | 제외 | +| 중간 부하 | 50 | 25 | 25 | 2 | 포함 | +| 큰 부하 | 100 | 50 | 50 | 2 | 포함 | + +본 측정 대상은 총 300문서다. 문서 수와 반복은 `document.indexing.e2e.load.*` System Property로 +줄여 Smoke Test를 실행할 수 있다. Worker 최대 동시성은 같은 장비에서 문서 수 변화만 비교하도록 +의도적으로 `2`에 고정한다. Embedding Batch Size는 제품 설정을 사용하며 두 값 모두 결과 환경 지문에 +기록한다. + +### 3.3 측정 순서 + +1. Warm-up PDF·DOCX를 업로드하고 모두 `INDEXED`가 될 때까지 기다린다. +2. Profile 시작 시각부터 여러 Uploader Thread로 PDF·DOCX를 교대로 접수한다. +3. 각 Upload 응답 시각과 마지막 Upload 완료 시각을 기록한다. +4. 자동 Worker가 Profile의 모든 Job을 `INDEXED`로 전환할 때까지 기다린다. +5. 형식별 Metadata와 전체 DB 불변식을 검증한 뒤 통계를 계산한다. +6. Profile 데이터를 정리하고 다음 반복을 시작한다. + +Upload와 Worker 실행이 겹치는 실제 흐름을 유지한다. 모든 Uploader는 하나의 Profile 마감 시각을 공유해 +개별 Future마다 Timeout이 누적되지 않게 하고, 실패 시 완료된 Run 결과를 JSON에 보존한다. + +## 4. 지표 계약 + +전체와 PDF·DOCX별로 다음 지표를 기록한다. + +| 지표 | 계산 | +|---|---| +| documents/s | 완료 문서 수 / Profile 전체 경과 시간 | +| documents/min | documents/s × 60 | +| chunks/s | 저장 Chunk 수 / Profile 전체 경과 시간 | +| embeddings/s | 저장 Embedding 수 / Profile 전체 경과 시간 | +| Upload 지연 | 개별 HTTP 요청 시작부터 응답까지 | +| Queue 대기 | Job `created_at`부터 `LOCKED` Event까지 | +| 실제 처리 | `LOCKED`부터 `INDEXED` Event까지 | +| 전체 Job 지연 | Job `created_at`부터 `INDEXED` Event까지 | + +지연 분포는 선형 보간 p50·p95·p99와 max를 밀리초로 기록한다. Profile별 원본 Run과 Profile별 +중앙값을 JSON에 함께 기록한다. + +## 5. 정합성 계약 + +각 Profile은 다음 조건을 모두 검증한다. + +- Upload 실패 0건, 대상 Job 전부 `INDEXED` +- 전체 Schema에 `PENDING`·`PROCESSING` 잔여 Job 없음 +- Job별 `SUCCESS` Attempt 정확히 1개, 실패 Attempt와 Retry 0건 +- Job별 `LOCKED`, `PARSE_STARTED`, `CHUNKED`, `EMBEDDING_STARTED`, `INDEXED` 순서 유지 +- Document와 Version 모두 `INDEXED` +- `documents.current_version_id`가 측정 Version을 가리킴 +- 문서별 Chunk 수가 1 이상이고 Embedding 수와 일치 +- 같은 `(chunk_id, embedding_model_id)` 중복 없음 +- 모든 Vector 차원 1024, Embedding 상태 `ACTIVE` +- PDF의 모든 Chunk가 유효한 페이지 번호를 보존 +- DOCX의 모든 Chunk가 비어 있지 않은 Section 제목을 보존 +- MinIO Object 수, Content-Type과 크기가 Upload 결과와 일치 + +하나라도 실패하면 성능 숫자를 유효한 결과로 취급하지 않고 Benchmark를 실패시킨다. + +## 6. 환경과 결과 보존 + +시작 전에 다음 계약을 확인한다. + +- PostgreSQL Server `17.x` +- pgvector `0.8.1` +- 실행 전용 Test Schema와 MinIO Bucket +- BGE Health와 Model명 `BAAI/bge-m3` +- Worker 최대 동시성과 Embedding Batch Size + +Secret, JWT와 Object Storage Credential은 결과에 기록하지 않는다. 원본 JSON은 Git 제외 경로에 둔다. + +```text +build/reports/document-indexing-e2e-load/document-indexing-e2e-load.json +``` + +실행 환경, Profile 중앙값, 해석과 한계는 `docs/test-results/`에 기록한다. + +## 7. 실행 경계 + +일반 `./gradlew test`는 외부 Infrastructure에 의존하지 않는다. 전용 Task만 실제 PostgreSQL, MinIO와 +BGE-M3를 요구한다. + +```bash +docker compose up -d postgres minio embedding-server +./gradlew documentIndexingE2ELoadTest +``` + +작은 Smoke 실행은 다음과 같다. + +```bash +./gradlew documentIndexingE2ELoadTest \ + -Ddocument.indexing.e2e.load.document-counts=4 \ + -Ddocument.indexing.e2e.load.repetitions=1 +``` + +## 8. 실패 정책 + +- Infrastructure Health, DB·pgvector Version 또는 BGE Model 계약이 다르면 즉시 실패한다. +- Upload와 Worker 처리는 Profile 공통 제한 시간을 넘으면 상태 Snapshot과 함께 실패한다. +- 완료된 Run은 후속 진단을 위해 JSON에 보존하되 실패 Run은 중앙값에 포함하지 않는다. +- Profile별 식별자를 사용해 데이터를 분리하고, Class 종료 때 Schema와 Bucket을 정리한다. + +## 9. 검증 + +- 형식 혼합과 형식별 통계 계약 단위 테스트 +- 실제 Infrastructure를 사용하는 4문서 Smoke Benchmark +- 기본 50·100문서 Profile 각 2회 전체 Benchmark +- 기존 PDF·DOCX 로컬 E2E 회귀 +- 전체 일반 Java 테스트 + +## 10. 커밋 분할 + +1. `docs: #143 PDF DOCX E2E 부하 Benchmark 설계` +2. `test: #143 문서 형식별 부하 통계 계약 추가` +3. `perf: #143 PDF DOCX 전체 인덱싱 부하 Benchmark 추가` +4. `build: #143 E2E 부하 전용 테스트 작업 추가` +5. `perf: #143 PDF DOCX 50 100문서 실측 결과 기록` + +## 11. 완료 조건 + +- 실제 PDF·DOCX 혼합 Pipeline을 한 명령으로 50·100문서 규모에서 반복 측정할 수 있다. +- 본 측정 300문서가 모두 `INDEXED`로 수렴하고 실패·미완료·중복 Vector가 없다. +- 전체와 형식별 처리량 및 지연 분포가 구조화돼 기록된다. +- 페이지·Section Metadata와 Vector 1024차원 불변식이 모든 Profile에서 유지된다. +- 일반 테스트는 외부 Infrastructure 없이 계속 실행된다. diff --git a/docs/test-results/gimin-#143-pdf-docx-indexing-e2e-load-benchmark.md b/docs/test-results/gimin-#143-pdf-docx-indexing-e2e-load-benchmark.md new file mode 100644 index 0000000..d8e3d91 --- /dev/null +++ b/docs/test-results/gimin-#143-pdf-docx-indexing-e2e-load-benchmark.md @@ -0,0 +1,168 @@ +# Issue #143 PDF·DOCX 전체 인덱싱 E2E 부하 Benchmark 결과 + +## 1. 결과 요약 + +PostgreSQL 17.8, pgvector 0.8.1, MinIO와 실제 `BAAI/bge-m3`를 연결하고 Text Layer PDF와 +OOXML DOCX의 업로드부터 `INDEXED` 전환까지 전체 경로를 측정했다. PDF·DOCX 2개씩 4문서를 예열한 +뒤 50문서와 100문서 Profile을 각각 2회 실행했다. + +| 문서 수 | 구성 | 반복 | 중앙 총 시간 | 문서/분 | Chunk·Embedding/초 | 전체 P95 | +|---:|---|---:|---:|---:|---:|---:| +| 50 | PDF 25 + DOCX 25 | 2회 | 96.633초 | 31.046 | 2.070 | 92.178초 | +| 100 | PDF 50 + DOCX 50 | 2회 | 187.138초 | 32.063 | 2.138 | 178.533초 | + +본 측정 300문서에서 1,200개 Chunk와 1,200개 1024차원 Embedding이 저장됐다. 업로드 실패, +실패·재시도 Attempt, 미완료 Job과 중복 Vector는 모두 0건이었다. + +문서 수를 50개에서 100개로 두 배 늘려도 분당 처리량은 약 3.3% 증가한 범위에서 유지됐다. 처리 +P95는 4.093초에서 4.026초로 비슷했고, 전체 P95 증가는 Queue 대기 P95가 88.355초에서 +174.684초로 늘어난 영향이다. + +## 2. 공개 가능한 실행 환경 + +| 항목 | 값 | +|---|---| +| Database | PostgreSQL 17.8, Local Docker | +| pgvector | 0.8.1 | +| Object Storage | MinIO, Local Docker | +| Embedding Provider | `BAAI/bge-m3`, Local Docker CPU 추론 | +| Vector 차원 | 1024 | +| Embedding Batch Size | 32 | +| Worker 실행 슬롯 | 2 | +| Worker Polling 주기 | 50 ms | +| 업로더 Thread | 8 | +| PDF | 2페이지 Text Layer, 페이지당 1,600자 | +| DOCX | Heading·본문 2개 Section, Section당 1,600자 | +| 문서당 Chunk·Embedding | 각각 4개 | +| Warm-up | PDF 2 + DOCX 2 | +| 본 측정 | 50 / 100문서, Profile당 2회 | +| Application | Spring Boot 3.5.16, Java 17 | +| 실행 장비 | macOS `aarch64`, 가용 Processor 10개 | +| 실행 일자 | 2026-08-11 KST | + +DB·MinIO·JWT Credential은 결과에 기록하지 않았다. 이 수치는 단일 Apple Silicon 로컬 장비의 +개발 기준선이며, 공식 OpenSQL 원격 Server 성능이나 운영 SLO가 아니다. + +## 3. 측정 경로 + +각 문서는 다음 실제 경로를 통과했다. + +```text +PDF·DOCX Binary 생성 +→ 인증 Multipart HTTP 업로드 +→ MinIO 원본 저장 +→ Embedding Job PENDING +→ 자동 Worker Polling·Claim·Attempt +→ PDF·DOCX Parsing +→ Chunk·페이지·Section Metadata 저장 +→ 실제 BGE-M3 Batch 호출 +→ pgvector vector(1024) 저장 +→ Version·Document·Job INDEXED +→ current_version 전환 +→ Worker 실행 슬롯 반환 +``` + +Profile마다 전용 Schema의 Job·Document·Chunk·Embedding과 전용 MinIO Bucket의 Object를 초기화해 +이전 실행이 다음 수치에 포함되지 않게 했다. PDF와 DOCX는 업로드 순서에서 교대로 배치했다. + +## 4. 실행 방법 + +PostgreSQL, MinIO와 Embedding Server가 모두 건강한 로컬 환경에서 실행했다. + +```bash +docker compose up -d postgres minio embedding-server +DB_SSLMODE=disable ./gradlew documentIndexingE2ELoadTest +``` + +구조화 원시 결과는 Git에 포함하지 않는 다음 경로에 생성된다. + +```text +build/reports/document-indexing-e2e-load/document-indexing-e2e-load.json +``` + +작은 실제 환경 Smoke는 다음 설정으로 실행했다. + +```bash +DB_SSLMODE=disable ./gradlew documentIndexingE2ELoadTest \ + -Ddocument.indexing.e2e.load.warm-up-documents=2 \ + -Ddocument.indexing.e2e.load.document-counts=4 \ + -Ddocument.indexing.e2e.load.repetitions=1 \ + -Ddocument.indexing.e2e.load.output=build/reports/document-indexing-e2e-load/smoke.json +``` + +Smoke는 PDF 2 + DOCX 2, Chunk·Embedding 각 16개를 7.192초에 처리했고 33.370문서/분을 +기록했다. 전체 Gradle 실행은 19초였다. + +## 5. 반복별 결과 + +| 문서 수 | 회차 | 총 시간 | 문서/분 | Chunk·Embedding/초 | Upload P95 | Queue P95 | 처리 P95 | 전체 P95 | +|---:|---:|---:|---:|---:|---:|---:|---:|---:| +| 50 | 1 | 96.045초 | 31.235 | 2.082 | 165.7ms | 87.903초 | 3.961초 | 91.590초 | +| 50 | 2 | 97.222초 | 30.857 | 2.057 | 84.9ms | 88.808초 | 4.226초 | 92.767초 | +| 100 | 1 | 188.285초 | 31.867 | 2.124 | 106.3ms | 175.755초 | 4.189초 | 179.814초 | +| 100 | 2 | 185.991초 | 32.260 | 2.151 | 71.4ms | 173.612초 | 3.863초 | 177.252초 | + +50문서 처리량 범위는 30.857~31.235문서/분, 100문서는 31.867~32.260문서/분이었다. 단일 +최고값 대신 두 반복의 중앙값을 비교 기준으로 사용했다. + +## 6. 형식별 결과 + +각 Profile 전체 시간을 분모로 사용해 형식별 문서·Chunk 처리 기여도를 계산했다. + +| 문서 수 | 형식 | 문서/분 중앙값 | Chunk/초 중앙값 | Upload P95 | Queue P95 | 처리 P95 | 전체 P95 | +|---:|---|---:|---:|---:|---:|---:|---:| +| 50 | PDF | 15.523 | 1.035 | 124.1ms | 85.947초 | 4.098초 | 89.800초 | +| 50 | DOCX | 15.523 | 1.035 | 121.6ms | 87.994초 | 3.981초 | 91.820초 | +| 100 | PDF | 16.032 | 1.069 | 88.1ms | 173.244초 | 4.065초 | 177.075초 | +| 100 | DOCX | 16.032 | 1.069 | 88.3ms | 173.388초 | 4.021초 | 177.139초 | + +동일 길이의 결정적 Fixture에서는 PDF와 DOCX 처리 P95 차이가 50문서에서 약 0.12초, +100문서에서 약 0.04초였다. 이 결과는 Parser 비용이 현재 전체 처리량 병목이 아니며 Queue 대기가 +전체 꼬리 지연을 지배함을 보여준다. + +## 7. 데이터 완전성 + +| 문서 수 | 회차 | PDF·DOCX | Chunk | Embedding | 결과 | +|---:|---:|---:|---:|---:|---| +| 50 | 1 | 25 + 25 | 200 | 200 | PASS | +| 50 | 2 | 25 + 25 | 200 | 200 | PASS | +| 100 | 1 | 50 + 50 | 400 | 400 | PASS | +| 100 | 2 | 50 + 50 | 400 | 400 | PASS | + +각 Profile 완료 시 다음 불변식을 함께 검증했다. + +- 모든 HTTP 업로드가 성공하고 MinIO Object 수가 문서 수와 일치한다. +- MinIO Object와 DB FileObject의 크기·Content-Type이 원본과 일치한다. +- 모든 Job·Version·Document가 `INDEXED`이고 `current_version_id`가 측정 Version을 가리킨다. +- Job별 성공 Attempt가 정확히 하나이며 Retry Count는 0이다. +- `LOCKED → PARSE_STARTED → CHUNKED → EMBEDDING_STARTED → INDEXED` 순서를 유지한다. +- Chunk와 Embedding이 일대일이고 중복 Chunk Embedding이 없다. +- 모든 활성 Vector의 차원이 1024다. +- 모든 PDF Chunk가 페이지 번호를 가지며 문서마다 페이지 1·2가 보존된다. +- 모든 DOCX Chunk가 Section 제목을 가지며 문서마다 두 Section이 보존된다. +- Profile 종료 때 `PENDING`·`PROCESSING` Job과 점유된 Worker 슬롯이 없다. + +## 8. 검증 결과 + +| 검증 | 결과 | +|---|---| +| 형식별 통계 계약 단위 테스트 | PASS | +| 전용 Gradle Task 노출 | PASS | +| PDF 2 + DOCX 2 실제 Smoke | PASS, Gradle 19초 | +| 50·100문서 각 2회 본 측정 | PASS, Gradle 9분 44초 | +| 본 측정 300문서·1,200 Vector 완전성 | PASS | +| 기존 실제 PDF·DOCX 로컬 E2E | PASS, 2 tests | +| 전체 일반 Java 회귀 | PASS, 746 tests | +| `git diff --check` | PASS | + +## 9. 결론과 한계 + +- 구현됨: 실제 PDF·DOCX 혼합 전체 Pipeline을 50·100문서 규모로 반복 측정할 수 있다. +- 검증됨: 300문서가 실패·재시도 없이 모두 `INDEXED`로 수렴했다. +- 검증됨: 문서 수를 두 배로 늘려도 분당 처리량은 약 31~32문서로 유지됐다. +- 관찰됨: PDF와 DOCX의 처리 P95는 비슷하고 Queue 대기가 전체 P95 증가를 지배했다. +- 검증됨: 페이지·Section Metadata, Chunk·Embedding 일대일과 1024차원 Vector가 모두 유지됐다. +- 한계: Text Layer PDF만 포함하며 스캔 PDF와 OCR은 범위 밖이다. +- 한계: 동일 길이의 결정적 문서라 실제 사용자 파일의 크기·표·이미지 분포를 대표하지 않는다. +- 한계: 단일 Worker Node, 실행 슬롯 2개와 Local CPU BGE-M3 결과다. +- 후속: 실제 사용자 Corpus, 장애 주입과 공식 OpenSQL 원격 환경에서 같은 Harness를 재검증할 수 있다. diff --git a/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadBenchmark.java b/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadBenchmark.java new file mode 100644 index 0000000..484f3f7 --- /dev/null +++ b/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadBenchmark.java @@ -0,0 +1,927 @@ +package com.opensource.docgrid.e2e; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.Timeout; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.SpringBootVersion; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.springframework.web.client.RestClient; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.opensource.docgrid.domain.embedding.config.EmbeddingBatchProperties; +import com.opensource.docgrid.domain.worker.config.IndexingWorkerProperties; +import com.opensource.docgrid.domain.worker.execution.WorkerExecutionSlotPool; +import com.opensource.docgrid.domain.worker.lifecycle.WorkerExecutionLifecycleManager; +import com.opensource.docgrid.domain.worker.lifecycle.WorkerJobPollingScheduler; +import com.opensource.docgrid.domain.worker.lifecycle.WorkerLifecycleManager; +import com.opensource.docgrid.e2e.DocumentIndexingE2ELoadStatistics.DocumentFormat; +import com.opensource.docgrid.e2e.DocumentIndexingE2ELoadStatistics.DocumentMeasurement; +import com.opensource.docgrid.e2e.DocumentIndexingE2ELoadStatistics.FormatSummary; +import com.opensource.docgrid.e2e.LocalE2eApiClient.UploadedDocument; +import com.opensource.docgrid.e2e.LocalE2eDocumentFactory.DocumentPayload; +import com.opensource.docgrid.e2e.LocalE2eDocumentFactory.DocumentSection; + +import io.minio.MinioClient; +import io.minio.StatObjectResponse; +import lombok.extern.slf4j.Slf4j; + +/** + * 실제 PDF·DOCX 50·100문서를 PostgreSQL·MinIO·BGE-M3 전체 인덱싱 Pipeline으로 부하 검증한다. + * + *

Text Layer PDF와 OOXML DOCX를 실제 Multipart HTTP로 동시에 접수하고 자동 Worker가 Queue를 + * 소진하게 한다. 전체와 형식별 처리량·지연을 기록하면서 페이지·Section Metadata, Attempt, Event, + * Chunk와 1024차원 Vector의 완전성을 검증한다. + */ +@Slf4j +@Tag("document-indexing-e2e-load") +@ActiveProfiles({"test", "minio-integration"}) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +@DisplayName("PDF DOCX 전체 인덱싱 E2E 부하 Benchmark") +class DocumentIndexingE2ELoadBenchmark { + + private static final String EXECUTION_ID = UUID.randomUUID().toString().replace("-", ""); + // PostgreSQL 식별자 63자 제한 안에서 별도 실행이 Schema를 공유하지 않도록 격리한다. + private static final String TEST_SCHEMA = "docgrid_document_e2e_load_" + + EXECUTION_ID.substring(0, 24); + private static final String TEST_BUCKET = "docgrid-document-e2e-load-" + EXECUTION_ID; + private static final String EXPECTED_POSTGRES_VERSION_PREFIX = "17."; + private static final String EXPECTED_PGVECTOR_VERSION = "0.8.1"; + private static final String EXPECTED_MODEL = "BAAI/bge-m3"; + private static final int EXPECTED_VECTOR_DIMENSION = 1024; + // 같은 장비에서 문서 수 변화만 비교하도록 Worker 실행 슬롯을 고정한다. + private static final int WORKER_MAX_CONCURRENCY = 2; + private static final int SECTION_CHARACTER_COUNT = positiveIntegerProperty( + "document.indexing.e2e.load.section-characters", + 1_600 + ); + private static final int WARM_UP_DOCUMENT_COUNT = positiveEvenIntegerProperty( + "document.indexing.e2e.load.warm-up-documents", + 4 + ); + private static final List DOCUMENT_COUNTS = positiveEvenIntegerListProperty( + "document.indexing.e2e.load.document-counts", + List.of(50, 100) + ); + private static final int REPETITIONS = positiveIntegerProperty( + "document.indexing.e2e.load.repetitions", + 2 + ); + private static final int UPLOADER_THREADS = positiveIntegerProperty( + "document.indexing.e2e.load.uploader-threads", + 8 + ); + private static final long PROFILE_TIMEOUT_SECONDS = positiveLongProperty( + "document.indexing.e2e.load.profile-timeout-seconds", + 1_200L + ); + private static final long POLLING_SLEEP_MILLIS = positiveLongProperty( + "document.indexing.e2e.load.status-polling-ms", + 100L + ); + private static final Path OUTPUT_PATH = Path.of(System.getProperty( + "document.indexing.e2e.load.output", + "build/reports/document-indexing-e2e-load/document-indexing-e2e-load.json" + )); + + @Autowired private TestRestTemplate restTemplate; + @Autowired private JdbcTemplate jdbcTemplate; + @Autowired private MinioClient minioClient; + @Autowired private ObjectMapper objectMapper; + @Autowired private WorkerLifecycleManager workerLifecycleManager; + @Autowired private WorkerJobPollingScheduler pollingScheduler; + @Autowired private WorkerExecutionLifecycleManager executionLifecycleManager; + @Autowired private WorkerExecutionSlotPool executionSlotPool; + @Autowired private IndexingWorkerProperties workerProperties; + @Autowired private EmbeddingBatchProperties embeddingBatchProperties; + @Autowired @Qualifier("embeddingRestClient") private RestClient embeddingRestClient; + + private LocalE2eApiClient apiClient; + private LocalE2eMinioBucket minioBucket; + private String accessToken; + + @DynamicPropertySource + static void configureEnvironment(DynamicPropertyRegistry registry) { + registry.add("TEST_DB_SCHEMA", () -> TEST_SCHEMA); + registry.add("jwt.secret", () -> "docgrid-document-e2e-load-test-secret-key-2026"); + registry.add("minio.bucket", () -> TEST_BUCKET); + registry.add("indexing.worker.enabled", () -> "true"); + registry.add("indexing.worker.name", () -> "document-indexing-e2e-load-benchmark"); + registry.add("indexing.worker.polling-interval", () -> "50ms"); + registry.add("indexing.worker.heartbeat-interval", () -> "1s"); + registry.add("indexing.worker.dead-threshold", () -> "2m"); + registry.add("indexing.worker.max-concurrency", () -> Integer.toString(WORKER_MAX_CONCURRENCY)); + registry.add("indexing.worker.lease-duration", () -> "2m"); + registry.add("indexing.worker.lease-renewal-interval", () -> "10s"); + registry.add("indexing.worker.lease-recovery-interval", () -> "10m"); + registry.add("indexing.worker.shutdown-grace-period", () -> "30s"); + // CPU 기반 BGE-M3의 실측 추론 시간을 운영 기본 Timeout과 분리한다. + registry.add("embedding.server.read-timeout", () -> "2m"); + } + + @BeforeAll + void setUpInfrastructure() throws Exception { + apiClient = new LocalE2eApiClient(restTemplate); + minioBucket = new LocalE2eMinioBucket(minioClient, TEST_BUCKET); + minioBucket.create(); + accessToken = apiClient.loginAdmin(); + awaitCondition( + "Worker가 Application Ready 뒤 등록되지 않았습니다.", + () -> workerLifecycleManager.getWorkerId().isPresent() + ); + } + + @AfterAll + void cleanUpInfrastructure() throws Exception { + // 1. Scheduler와 실행 Thread를 먼저 닫아 자원 정리 뒤 DB 접근이 재개되지 않게 한다. + pollingScheduler.stopPolling(); + executionLifecycleManager.shutdown(); + workerLifecycleManager.stopWorker(); + + // 2. Benchmark 전용 Bucket과 Schema만 제거해 기존 개발 Data를 보존한다. + minioBucket.close(); + jdbcTemplate.execute("DROP SCHEMA IF EXISTS " + TEST_SCHEMA + " CASCADE"); + } + + @Test + @Timeout(3_600) + @DisplayName("실제 PDF DOCX 50 100문서의 전체 인덱싱 처리량과 완전성을 검증한다") + void measurePdfDocxIndexingE2ELoad() throws Exception { + // 1. 잘못된 DB나 외부 Service의 수치가 기록되지 않도록 환경 계약을 먼저 확인한다. + EnvironmentFingerprint environment = validateEnvironment(); + List runs = new ArrayList<>(); + writeReport(new BenchmarkReport(environment, runs, List.of())); + logJson("DOCUMENT_INDEXING_E2E_LOAD_ENV", environment); + + // 2. 실제 PDF·DOCX Parser, MinIO와 BGE-M3 경로를 예열하되 결과에는 포함하지 않는다. + runWarmUp(); + + // 3. 50·100문서의 균형 혼합을 반복 실행하며 완료된 Run을 즉시 보존한다. + for (int documentCount : DOCUMENT_COUNTS) { + for (int repetition = 1; repetition <= REPETITIONS; repetition++) { + resetProfileState(); + ProfileRun run = runMeasuredProfile(documentCount, repetition); + runs.add(run); + logJson("DOCUMENT_INDEXING_E2E_LOAD_RESULT", run); + writeReport(new BenchmarkReport(environment, runs, medians(runs))); + } + } + + // 4. 장비 내 변동을 줄인 Profile별 중앙값을 최종 결과와 Log에 남긴다. + List medians = medians(runs); + for (ProfileMedian median : medians) { + logJson("DOCUMENT_INDEXING_E2E_LOAD_MEDIAN", median); + } + writeReport(new BenchmarkReport(environment, runs, medians)); + } + + private EnvironmentFingerprint validateEnvironment() { + String postgresVersion = jdbcTemplate.queryForObject("SHOW server_version", String.class); + String pgvectorVersion = jdbcTemplate.queryForObject( + "SELECT extversion FROM pg_extension WHERE extname = 'vector'", + String.class + ); + Map model = jdbcTemplate.queryForMap( + "SELECT model_name, dimension FROM embedding_models " + + "WHERE is_active = TRUE AND is_searchable = TRUE" + ); + JsonNode health = embeddingRestClient.get() + .uri("/health") + .retrieve() + .body(JsonNode.class); + + assertThat(jdbcTemplate.queryForObject("SELECT current_schema()", String.class)) + .isEqualTo(TEST_SCHEMA); + assertThat(postgresVersion).startsWith(EXPECTED_POSTGRES_VERSION_PREFIX); + assertThat(pgvectorVersion).isEqualTo(EXPECTED_PGVECTOR_VERSION); + assertThat(model.get("model_name").toString()).isEqualTo(EXPECTED_MODEL); + assertThat(((Number) model.get("dimension")).intValue()).isEqualTo(EXPECTED_VECTOR_DIMENSION); + assertThat(health).isNotNull(); + assertThat(health.path("status").asText()).isEqualTo("ok"); + assertThat(workerProperties.getMaxConcurrency()).isEqualTo(WORKER_MAX_CONCURRENCY); + + return new EnvironmentFingerprint( + postgresVersion, + pgvectorVersion, + SpringBootVersion.getVersion(), + EXPECTED_MODEL, + EXPECTED_VECTOR_DIMENSION, + embeddingBatchProperties.getBatchSize(), + workerProperties.getMaxConcurrency(), + workerProperties.getPollingInterval().toString(), + UPLOADER_THREADS, + WARM_UP_DOCUMENT_COUNT, + DOCUMENT_COUNTS, + REPETITIONS, + SECTION_CHARACTER_COUNT, + System.getProperty("os.name"), + System.getProperty("os.arch"), + Runtime.getRuntime().availableProcessors() + ); + } + + private void runWarmUp() throws Exception { + resetProfileState(); + List uploads = uploadDocuments("warm-up", WARM_UP_DOCUMENT_COUNT); + awaitIndexedAndIdle(uploads, "Warm-up"); + assertProfileInvariants(uploads); + log.info("PDF DOCX E2E 부하 Benchmark 예열 완료. documentCount={}", WARM_UP_DOCUMENT_COUNT); + } + + private ProfileRun runMeasuredProfile(int documentCount, int repetition) throws Exception { + String profileName = "documents-" + documentCount + "-run-" + repetition; + long profileStartedAt = System.nanoTime(); + List uploads = uploadDocuments(profileName, documentCount); + long uploadCompletedAt = System.nanoTime(); + + awaitIndexedAndIdle(uploads, profileName); + long profileCompletedAt = System.nanoTime(); + ProfileData profileData = assertProfileInvariants(uploads); + + double elapsedSeconds = seconds(profileCompletedAt - profileStartedAt); + double uploadSeconds = seconds(uploadCompletedAt - profileStartedAt); + double queueDrainSeconds = seconds(profileCompletedAt - uploadCompletedAt); + List measurements = profileData.measurements(); + Map byFormat = + DocumentIndexingE2ELoadStatistics.summarizeByFormat(measurements, elapsedSeconds); + + return new ProfileRun( + documentCount, + repetition, + profileData.chunkCount(), + profileData.embeddingCount(), + elapsedSeconds, + uploadSeconds, + queueDrainSeconds, + documentCount / elapsedSeconds, + documentCount / elapsedSeconds * 60.0, + profileData.chunkCount() / elapsedSeconds, + profileData.embeddingCount() / elapsedSeconds, + LatencySummary.from(measurements.stream().map(DocumentMeasurement::uploadMillis).toList()), + LatencySummary.from(measurements.stream().map(DocumentMeasurement::queueMillis).toList()), + LatencySummary.from(measurements.stream().map(DocumentMeasurement::processingMillis).toList()), + LatencySummary.from(measurements.stream().map(DocumentMeasurement::e2eMillis).toList()), + byFormat + ); + } + + private List uploadDocuments(String profileName, int documentCount) throws Exception { + List documents = prepareDocuments(profileName, documentCount); + int threadCount = Math.min(UPLOADER_THREADS, documentCount); + ExecutorService uploader = Executors.newFixedThreadPool(threadCount); + List> futures = new ArrayList<>(documentCount); + + try { + // 1. PDF와 DOCX를 교대로 병렬 제출해 형식이 한쪽에 몰리지 않는 PENDING Queue를 만든다. + for (PreparedDocument document : documents) { + futures.add(uploader.submit(() -> { + long startedAt = System.nanoTime(); + UploadedDocument upload = apiClient.upload(accessToken, document.payload()); + return new UploadMeasurement( + document.format(), + document.payload(), + upload, + seconds(System.nanoTime() - startedAt) * 1_000.0 + ); + })); + } + uploader.shutdown(); + + // 2. 하나의 Profile 마감 시각을 공유해 Future마다 Timeout이 누적되지 않게 한다. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(PROFILE_TIMEOUT_SECONDS); + List uploads = new ArrayList<>(documentCount); + for (Future future : futures) { + long remainingNanos = deadline - System.nanoTime(); + if (remainingNanos <= 0L) { + throw new AssertionError(profileName + " Upload 공통 제한 시간을 초과했습니다."); + } + uploads.add(future.get(remainingNanos, TimeUnit.NANOSECONDS)); + } + return List.copyOf(uploads); + } finally { + uploader.shutdownNow(); + uploader.awaitTermination(10, TimeUnit.SECONDS); + } + } + + private List prepareDocuments(String profileName, int documentCount) + throws IOException { + List documents = new ArrayList<>(documentCount); + for (int index = 0; index < documentCount; index++) { + DocumentFormat format = index % 2 == 0 ? DocumentFormat.PDF : DocumentFormat.DOCX; + documents.add(new PreparedDocument(format, documentPayload(profileName, index, format))); + } + assertThat(documents.stream().filter(item -> item.format() == DocumentFormat.PDF).count()) + .isEqualTo(documentCount / 2L); + assertThat(documents.stream().filter(item -> item.format() == DocumentFormat.DOCX).count()) + .isEqualTo(documentCount / 2L); + return List.copyOf(documents); + } + + private DocumentPayload documentPayload( + String profileName, + int documentIndex, + DocumentFormat format + ) throws IOException { + String sequence = String.format(Locale.ROOT, "%04d", documentIndex); + String baseName = profileName + "-" + sequence; + if (format == DocumentFormat.PDF) { + return LocalE2eDocumentFactory.pdf( + baseName + ".pdf", + "PDF E2E Load " + baseName, + fixedLengthText(baseName, "PDF page one vector indexing"), + fixedLengthText(baseName, "PDF page two semantic retrieval") + ); + } + return LocalE2eDocumentFactory.docx( + baseName + ".docx", + "DOCX E2E Load " + baseName, + List.of( + new DocumentSection( + "DOCX Pipeline " + baseName, + fixedLengthText(baseName, "DOCX section one worker pipeline") + ), + new DocumentSection( + "DOCX Search " + baseName, + fixedLengthText(baseName, "DOCX section two searchable version") + ) + ) + ); + } + + private String fixedLengthText(String baseName, String subject) { + String marker = "DocGrid " + baseName + " " + subject + ". "; + String sentence = "Automatic workers parse real documents, store deterministic chunks, " + + "generate BGE-M3 vectors, and switch the searchable version atomically. "; + StringBuilder body = new StringBuilder(SECTION_CHARACTER_COUNT); + body.append(marker); + while (body.length() < SECTION_CHARACTER_COUNT) { + body.append(sentence); + } + body.setLength(SECTION_CHARACTER_COUNT); + return body.toString(); + } + + private void awaitIndexedAndIdle(List uploads, String profileName) + throws InterruptedException { + awaitCondition( + () -> profileName + " Profile이 완료되지 않았습니다. " + jobSnapshot(uploads), + () -> indexedJobCount(uploads) == uploads.size() && executionSlotPool.getActiveSlots() == 0 + ); + } + + private ProfileData assertProfileInvariants(List uploads) throws Exception { + List measurements = new ArrayList<>(uploads.size()); + Map> chunksByFormat = new EnumMap<>(DocumentFormat.class); + for (DocumentFormat format : DocumentFormat.values()) { + chunksByFormat.put(format, new ArrayList<>()); + } + int totalChunks = 0; + int totalEmbeddings = 0; + + for (UploadMeasurement measurement : uploads) { + UploadedDocument upload = measurement.upload(); + // 1. 실제 MinIO Object의 위치, 크기와 형식별 Content-Type을 DB와 교차 검증한다. + assertStoredObject(measurement); + + // 2. Job과 검색 Version 전이가 끝났고 재시도 없이 한 Attempt만 성공했는지 확인한다. + assertThat(queryString("SELECT status FROM embedding_jobs WHERE id = ?", upload.embeddingJobId())) + .isEqualTo("INDEXED"); + assertThat(queryInteger( + "SELECT retry_count FROM embedding_jobs WHERE id = ?", + upload.embeddingJobId() + )).isZero(); + assertThat(count( + "SELECT COUNT(*) FROM embedding_job_attempts WHERE embedding_job_id = ?", + upload.embeddingJobId() + )).isOne(); + assertThat(count( + "SELECT COUNT(*) FROM embedding_job_attempts " + + "WHERE embedding_job_id = ? AND status = 'SUCCESS'", + upload.embeddingJobId() + )).isOne(); + assertThat(queryString("SELECT status FROM documents WHERE id = ?", upload.documentId())) + .isEqualTo("INDEXED"); + assertThat(queryLong("SELECT current_version_id FROM documents WHERE id = ?", upload.documentId())) + .isEqualTo(upload.documentVersionId()); + assertThat(queryString( + "SELECT status FROM document_versions WHERE id = ?", + upload.documentVersionId() + )).isEqualTo("INDEXED"); + + // 3. Chunk와 Embedding이 일대일이고 모든 활성 Vector가 1024차원인지 확인한다. + int chunkCount = count( + "SELECT COUNT(*) FROM document_chunks WHERE document_version_id = ?", + upload.documentVersionId() + ); + int embeddingCount = count( + "SELECT COUNT(*) FROM embeddings WHERE document_version_id = ?", + upload.documentVersionId() + ); + assertThat(chunkCount).isPositive(); + assertThat(embeddingCount).isEqualTo(chunkCount); + assertThat(count( + "SELECT COUNT(DISTINCT chunk_id) FROM embeddings WHERE document_version_id = ?", + upload.documentVersionId() + )).isEqualTo(embeddingCount); + assertThat(queryInteger( + "SELECT MIN(vector_dims(vector)) FROM embeddings WHERE document_version_id = ?", + upload.documentVersionId() + )).isEqualTo(EXPECTED_VECTOR_DIMENSION); + assertThat(queryInteger( + "SELECT MAX(vector_dims(vector)) FROM embeddings WHERE document_version_id = ?", + upload.documentVersionId() + )).isEqualTo(EXPECTED_VECTOR_DIMENSION); + assertThat(count( + "SELECT COUNT(*) FROM embeddings WHERE document_version_id = ? AND status = 'ACTIVE'", + upload.documentVersionId() + )).isEqualTo(embeddingCount); + assertFormatMetadata(measurement.format(), upload.documentVersionId(), chunkCount); + + // 4. 정상 Event 순서와 실패·Retry 부재를 확인한 뒤 단계별 지연을 수집한다. + List events = jdbcTemplate.queryForList( + "SELECT event_type FROM indexing_events WHERE embedding_job_id = ? ORDER BY id", + String.class, + upload.embeddingJobId() + ); + assertThat(events).containsSubsequence( + "LOCKED", + "PARSE_STARTED", + "CHUNKED", + "EMBEDDING_STARTED", + "INDEXED" + ); + assertThat(events).doesNotContain( + "PARSE_FAILED", + "EMBEDDING_FAILED", + "LEASE_EXPIRED", + "FAILED", + "RETRY", + "MANUAL_RETRY" + ); + JobTiming timing = readJobTiming(upload.embeddingJobId()); + measurements.add(new DocumentMeasurement( + measurement.format(), + measurement.uploadMillis(), + timing.queueWaitMillis(), + timing.processingMillis(), + timing.endToEndMillis(), + chunkCount, + embeddingCount + )); + chunksByFormat.get(measurement.format()).add(chunkCount); + totalChunks += chunkCount; + totalEmbeddings += embeddingCount; + } + + // 5. 형식별 Fixture 분포, Object 수와 전체 Queue 소진 상태를 확인한다. + DocumentIndexingE2ELoadStatistics.validateBalancedMix(measurements); + for (List chunkCounts : chunksByFormat.values()) { + assertThat(chunkCounts).isNotEmpty(); + assertThat(chunkCounts).allMatch(count -> count.equals(chunkCounts.get(0))); + } + assertThat(minioBucket.objectKeys()).hasSize(uploads.size()); + assertThat(count( + "SELECT COUNT(*) FROM embedding_jobs WHERE status IN ('PENDING', 'PROCESSING')" + )).isZero(); + assertThat(workerLifecycleManager.getWorkerId()).isPresent(); + return new ProfileData(totalChunks, totalEmbeddings, List.copyOf(measurements)); + } + + private void assertStoredObject(UploadMeasurement measurement) throws Exception { + StoredObject storedObject = jdbcTemplate.queryForObject( + "SELECT bucket_name, object_key, content_type, file_size FROM file_objects WHERE id = ?", + (resultSet, rowNumber) -> new StoredObject( + resultSet.getString("bucket_name"), + resultSet.getString("object_key"), + resultSet.getString("content_type"), + resultSet.getLong("file_size") + ), + measurement.upload().fileObjectId() + ); + assertThat(storedObject).isNotNull(); + assertThat(storedObject.bucketName()).isEqualTo(TEST_BUCKET); + assertThat(storedObject.contentType()).isEqualTo(measurement.payload().mediaType().toString()); + assertThat(storedObject.fileSize()).isEqualTo(measurement.payload().content().length); + StatObjectResponse stat = minioBucket.stat(storedObject.objectKey()); + assertThat(stat.size()).isEqualTo(measurement.payload().content().length); + assertThat(stat.contentType()).isEqualTo(measurement.payload().mediaType().toString()); + } + + private void assertFormatMetadata(DocumentFormat format, Long versionId, int chunkCount) { + if (format == DocumentFormat.PDF) { + assertThat(count( + "SELECT COUNT(*) FROM document_chunks " + + "WHERE document_version_id = ? AND page_no IS NOT NULL", + versionId + )).isEqualTo(chunkCount); + assertThat(count( + "SELECT COUNT(DISTINCT page_no) FROM document_chunks WHERE document_version_id = ?", + versionId + )).isEqualTo(2); + return; + } + assertThat(count( + "SELECT COUNT(*) FROM document_chunks " + + "WHERE document_version_id = ? AND NULLIF(BTRIM(section_title), '') IS NOT NULL", + versionId + )).isEqualTo(chunkCount); + assertThat(count( + "SELECT COUNT(DISTINCT section_title) FROM document_chunks WHERE document_version_id = ?", + versionId + )).isEqualTo(2); + } + + private JobTiming readJobTiming(Long jobId) { + return jdbcTemplate.queryForObject( + """ + SELECT job.created_at AS job_created_at, + MIN(event.occurred_at) FILTER (WHERE event.event_type = 'LOCKED') AS first_locked_at, + MIN(event.occurred_at) FILTER (WHERE event.event_type = 'INDEXED') AS indexed_at + FROM embedding_jobs job + JOIN indexing_events event ON event.embedding_job_id = job.id + WHERE job.id = ? + GROUP BY job.id, job.created_at + """, + (resultSet, rowNumber) -> { + LocalDateTime createdAt = resultSet.getTimestamp("job_created_at").toLocalDateTime(); + LocalDateTime lockedAt = resultSet.getTimestamp("first_locked_at").toLocalDateTime(); + LocalDateTime indexedAt = resultSet.getTimestamp("indexed_at").toLocalDateTime(); + double queueWaitMillis = millis(Duration.between(createdAt, lockedAt)); + double processingMillis = millis(Duration.between(lockedAt, indexedAt)); + double endToEndMillis = millis(Duration.between(createdAt, indexedAt)); + assertThat(queueWaitMillis).isGreaterThanOrEqualTo(0.0); + assertThat(processingMillis).isGreaterThanOrEqualTo(0.0); + assertThat(endToEndMillis).isGreaterThanOrEqualTo(processingMillis); + return new JobTiming(queueWaitMillis, processingMillis, endToEndMillis); + }, + jobId + ); + } + + private void resetProfileState() throws Exception { + awaitCondition( + "이전 Profile의 Worker 실행 슬롯이 반환되지 않았습니다.", + () -> executionSlotPool.getActiveSlots() == 0 + ); + minioBucket.clear(); + jdbcTemplate.execute(""" + TRUNCATE TABLE + indexing_events, + embedding_job_attempts, + embeddings, + document_chunks, + embedding_jobs, + document_versions, + documents, + file_objects + RESTART IDENTITY CASCADE + """); + } + + private int indexedJobCount(List uploads) { + String placeholders = String.join(",", uploads.stream().map(upload -> "?").toList()); + Object[] jobIds = uploads.stream().map(item -> item.upload().embeddingJobId()).toArray(); + return count( + "SELECT COUNT(*) FROM embedding_jobs WHERE status = 'INDEXED' AND id IN (" + + placeholders + ")", + jobIds + ); + } + + private String jobSnapshot(List uploads) { + String placeholders = String.join(",", uploads.stream().map(upload -> "?").toList()); + Object[] jobIds = uploads.stream().map(item -> item.upload().embeddingJobId()).toArray(); + return jdbcTemplate.queryForList( + "SELECT id || ':' || status || ':retry=' || retry_count || ':error=' " + + "|| COALESCE(error_code, 'none') FROM embedding_jobs WHERE id IN (" + + placeholders + ") ORDER BY id", + String.class, + jobIds + ).toString(); + } + + private List medians(List runs) { + List medians = new ArrayList<>(); + for (int documentCount : DOCUMENT_COUNTS) { + List profileRuns = runs.stream() + .filter(run -> run.documentCount() == documentCount) + .toList(); + if (profileRuns.isEmpty()) { + continue; + } + Map formatMedians = new EnumMap<>(DocumentFormat.class); + for (DocumentFormat format : DocumentFormat.values()) { + formatMedians.put(format, formatMedian(profileRuns, format)); + } + medians.add(new ProfileMedian( + documentCount, + profileRuns.size(), + median(profileRuns, ProfileRun::documentsPerSecond), + median(profileRuns, ProfileRun::documentsPerMinute), + median(profileRuns, ProfileRun::chunksPerSecond), + median(profileRuns, ProfileRun::embeddingsPerSecond), + median(profileRuns, ProfileRun::totalElapsedSeconds), + median(profileRuns, ProfileRun::uploadElapsedSeconds), + median(profileRuns, ProfileRun::queueDrainSeconds), + median(profileRuns, run -> run.uploadLatency().p95Millis()), + median(profileRuns, run -> run.queueWait().p95Millis()), + median(profileRuns, run -> run.processing().p95Millis()), + median(profileRuns, run -> run.endToEnd().p95Millis()), + Map.copyOf(formatMedians) + )); + } + return List.copyOf(medians); + } + + private FormatMedian formatMedian(List runs, DocumentFormat format) { + return new FormatMedian( + median(runs, run -> run.byFormat().get(format).documentsPerMinute()), + median(runs, run -> run.byFormat().get(format).chunksPerSecond()), + median(runs, run -> run.byFormat().get(format).uploadLatencyMillis().p95()), + median(runs, run -> run.byFormat().get(format).queueLatencyMillis().p95()), + median(runs, run -> run.byFormat().get(format).processingLatencyMillis().p95()), + median(runs, run -> run.byFormat().get(format).e2eLatencyMillis().p95()) + ); + } + + private double median(List runs, ProfileMetric metric) { + return WorkerIndexingThroughputStatistics.median(runs.stream().map(metric::value).toList()); + } + + private void writeReport(BenchmarkReport report) throws IOException { + Path parent = OUTPUT_PATH.toAbsolutePath().getParent(); + if (parent != null) { + Files.createDirectories(parent); + } + objectMapper.writerWithDefaultPrettyPrinter().writeValue(OUTPUT_PATH.toFile(), report); + } + + private void logJson(String prefix, Object value) throws JsonProcessingException { + log.info("{} {}", prefix, objectMapper.writeValueAsString(value)); + } + + private void awaitCondition(String failureMessage, CheckedCondition condition) throws InterruptedException { + awaitCondition(() -> failureMessage, condition); + } + + private void awaitCondition(Supplier failureMessage, CheckedCondition condition) + throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(PROFILE_TIMEOUT_SECONDS); + while (System.nanoTime() < deadline) { + if (condition.evaluate()) { + return; + } + Thread.sleep(POLLING_SLEEP_MILLIS); + } + // 상태 Snapshot처럼 비용이 있는 메시지는 실제 Timeout이 발생했을 때만 계산한다. + throw new AssertionError(failureMessage.get()); + } + + private int count(String sql, Object... arguments) { + return jdbcTemplate.queryForObject(sql, Integer.class, arguments); + } + + private int queryInteger(String sql, Object... arguments) { + return jdbcTemplate.queryForObject(sql, Integer.class, arguments); + } + + private Long queryLong(String sql, Object... arguments) { + return jdbcTemplate.queryForObject(sql, Long.class, arguments); + } + + private String queryString(String sql, Object... arguments) { + return jdbcTemplate.queryForObject(sql, String.class, arguments); + } + + private static double seconds(long nanoseconds) { + return nanoseconds / 1_000_000_000.0; + } + + private static double millis(Duration duration) { + return duration.toNanos() / 1_000_000.0; + } + + private static int positiveIntegerProperty(String name, int defaultValue) { + String value = System.getProperty(name); + int parsed = value == null ? defaultValue : Integer.parseInt(value.trim()); + if (parsed < 1) { + throw new IllegalArgumentException(name + "은 1 이상이어야 합니다."); + } + return parsed; + } + + private static int positiveEvenIntegerProperty(String name, int defaultValue) { + int parsed = positiveIntegerProperty(name, defaultValue); + if (parsed % 2 != 0) { + throw new IllegalArgumentException(name + "은 PDF DOCX 균형을 위해 짝수여야 합니다."); + } + return parsed; + } + + private static long positiveLongProperty(String name, long defaultValue) { + String value = System.getProperty(name); + long parsed = value == null ? defaultValue : Long.parseLong(value.trim()); + if (parsed < 1L) { + throw new IllegalArgumentException(name + "은 1 이상이어야 합니다."); + } + return parsed; + } + + private static List positiveEvenIntegerListProperty( + String name, + List defaults + ) { + String value = System.getProperty(name); + if (value == null || value.isBlank()) { + return defaults; + } + List parsed = List.of(value.split(",")).stream() + .map(String::trim) + .map(Integer::parseInt) + .distinct() + .toList(); + if (parsed.isEmpty() || parsed.stream().anyMatch(item -> item < 2 || item % 2 != 0)) { + throw new IllegalArgumentException(name + "에는 2 이상의 짝수만 사용할 수 있습니다."); + } + return parsed; + } + + /** 실행 환경과 Workload 설정을 Secret 없이 재현할 수 있는 지문이다. */ + private record EnvironmentFingerprint( + String postgresVersion, + String pgvectorVersion, + String springBootVersion, + String embeddingModel, + int vectorDimension, + int embeddingBatchSize, + int workerMaxConcurrency, + String workerPollingInterval, + int uploaderThreads, + int warmUpDocumentCount, + List documentCounts, + int repetitions, + int sectionCharacterCount, + String osName, + String osArchitecture, + int availableProcessors + ) { + } + + /** 한 Profile 반복의 전체·형식별 처리량과 단계별 지연 원시 결과다. */ + private record ProfileRun( + int documentCount, + int repetition, + int chunkCount, + int embeddingCount, + double totalElapsedSeconds, + double uploadElapsedSeconds, + double queueDrainSeconds, + double documentsPerSecond, + double documentsPerMinute, + double chunksPerSecond, + double embeddingsPerSecond, + LatencySummary uploadLatency, + LatencySummary queueWait, + LatencySummary processing, + LatencySummary endToEnd, + Map byFormat + ) { + } + + /** 같은 문서 수 Profile 반복의 전체·형식별 지표 중앙값이다. */ + private record ProfileMedian( + int documentCount, + int completedRepetitions, + double documentsPerSecond, + double documentsPerMinute, + double chunksPerSecond, + double embeddingsPerSecond, + double totalElapsedSeconds, + double uploadElapsedSeconds, + double queueDrainSeconds, + double uploadP95Millis, + double queueWaitP95Millis, + double processingP95Millis, + double endToEndP95Millis, + Map byFormat + ) { + } + + /** 한 문서 형식의 핵심 처리량과 꼬리 지연 중앙값을 비교 가능하게 요약한다. */ + private record FormatMedian( + double documentsPerMinute, + double chunksPerSecond, + double uploadP95Millis, + double queueP95Millis, + double processingP95Millis, + double e2eP95Millis + ) { + } + + /** 한 지연 분포의 중앙값, 꼬리 지연과 최댓값을 밀리초로 보존한다. */ + private record LatencySummary( + double p50Millis, + double p95Millis, + double p99Millis, + double maxMillis + ) { + + private static LatencySummary from(List samples) { + return new LatencySummary( + WorkerIndexingThroughputStatistics.percentile(samples, 50.0), + WorkerIndexingThroughputStatistics.percentile(samples, 95.0), + WorkerIndexingThroughputStatistics.percentile(samples, 99.0), + samples.stream().mapToDouble(Double::doubleValue).max().orElseThrow() + ); + } + } + + /** 형식과 Payload를 Upload 실행 전 순서와 함께 보존한다. */ + private record PreparedDocument(DocumentFormat format, DocumentPayload payload) { + } + + /** 한 실제 HTTP Upload 결과에 형식·Payload·지연을 연결한다. */ + private record UploadMeasurement( + DocumentFormat format, + DocumentPayload payload, + UploadedDocument upload, + double uploadMillis + ) { + } + + /** MinIO Object 검증에 필요한 DB 저장 위치·형식·크기다. */ + private record StoredObject( + String bucketName, + String objectKey, + String contentType, + long fileSize + ) { + } + + /** Profile 검증에서 수집한 전체 저장량과 문서별 측정값이다. */ + private record ProfileData( + int chunkCount, + int embeddingCount, + List measurements + ) { + } + + /** 한 Job의 Queue, 실제 처리와 전체 지연을 밀리초로 표현한다. */ + private record JobTiming(double queueWaitMillis, double processingMillis, double endToEndMillis) { + } + + /** 완료된 Profile과 중앙값을 환경 지문과 함께 JSON으로 저장하는 최상위 결과다. */ + private record BenchmarkReport( + EnvironmentFingerprint environment, + List runs, + List medians + ) { + } + + /** Profile 중앙값을 계산할 실수 지표를 선택한다. */ + @FunctionalInterface + private interface ProfileMetric { + double value(ProfileRun run); + } + + /** 제한 시간 동안 반복 평가할 DB·Worker 완료 조건이다. */ + @FunctionalInterface + private interface CheckedCondition { + boolean evaluate(); + } +} diff --git a/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadStatistics.java b/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadStatistics.java new file mode 100644 index 0000000..5e2f2fe --- /dev/null +++ b/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadStatistics.java @@ -0,0 +1,142 @@ +package com.opensource.docgrid.e2e; + +import java.util.EnumMap; +import java.util.List; +import java.util.Map; + +/** + * PDF·DOCX E2E 부하 Benchmark의 형식 혼합과 형식별 처리량·지연 계산을 담당한다. + * + *

실제 Infrastructure I/O와 분리된 순수 계산만 제공해 Profile 결과 집계 계약을 일반 단위 + * 테스트에서 검증할 수 있게 한다. + */ +final class DocumentIndexingE2ELoadStatistics { + + private DocumentIndexingE2ELoadStatistics() { + } + + static void validateBalancedMix(List measurements) { + if (measurements == null || measurements.isEmpty()) { + throw new IllegalArgumentException("문서 측정값은 비어 있을 수 없습니다."); + } + + long pdfCount = measurements.stream() + .filter(measurement -> measurement.format() == DocumentFormat.PDF) + .count(); + long docxCount = measurements.stream() + .filter(measurement -> measurement.format() == DocumentFormat.DOCX) + .count(); + if (pdfCount != docxCount) { + throw new IllegalArgumentException("PDF와 DOCX 문서 수는 같아야 합니다."); + } + } + + static Map summarizeByFormat( + List measurements, + double elapsedSeconds + ) { + validateBalancedMix(measurements); + if (!Double.isFinite(elapsedSeconds) || elapsedSeconds <= 0.0) { + throw new IllegalArgumentException("Profile 경과 시간은 0보다 큰 유한한 값이어야 합니다."); + } + + Map summaries = new EnumMap<>(DocumentFormat.class); + for (DocumentFormat format : DocumentFormat.values()) { + List formatMeasurements = measurements.stream() + .filter(measurement -> measurement.format() == format) + .toList(); + summaries.put(format, summarize(formatMeasurements, elapsedSeconds)); + } + return Map.copyOf(summaries); + } + + private static FormatSummary summarize( + List measurements, + double elapsedSeconds + ) { + int documentCount = measurements.size(); + int chunkCount = measurements.stream().mapToInt(DocumentMeasurement::chunkCount).sum(); + int embeddingCount = measurements.stream().mapToInt(DocumentMeasurement::embeddingCount).sum(); + double documentsPerSecond = documentCount / elapsedSeconds; + + return new FormatSummary( + documentCount, + chunkCount, + embeddingCount, + documentsPerSecond, + documentsPerSecond * 60.0, + chunkCount / elapsedSeconds, + embeddingCount / elapsedSeconds, + latency(measurements.stream().map(DocumentMeasurement::uploadMillis).toList()), + latency(measurements.stream().map(DocumentMeasurement::queueMillis).toList()), + latency(measurements.stream().map(DocumentMeasurement::processingMillis).toList()), + latency(measurements.stream().map(DocumentMeasurement::e2eMillis).toList()) + ); + } + + private static LatencySummary latency(List values) { + return new LatencySummary( + WorkerIndexingThroughputStatistics.percentile(values, 50.0), + WorkerIndexingThroughputStatistics.percentile(values, 95.0), + WorkerIndexingThroughputStatistics.percentile(values, 99.0), + values.stream().mapToDouble(Double::doubleValue).max().orElseThrow() + ); + } + + /** 부하 Fixture와 결과 집계에서 공유하는 지원 문서 형식이다. */ + enum DocumentFormat { + PDF, + DOCX + } + + /** 한 문서의 형식, 구간별 지연과 저장 결과를 형식별 통계 입력으로 전달한다. */ + record DocumentMeasurement( + DocumentFormat format, + double uploadMillis, + double queueMillis, + double processingMillis, + double e2eMillis, + int chunkCount, + int embeddingCount + ) { + + DocumentMeasurement { + if (format == null) { + throw new IllegalArgumentException("문서 형식은 필수입니다."); + } + if (!isNonNegativeFinite(uploadMillis) + || !isNonNegativeFinite(queueMillis) + || !isNonNegativeFinite(processingMillis) + || !isNonNegativeFinite(e2eMillis)) { + throw new IllegalArgumentException("지연 측정값은 0 이상의 유한한 값이어야 합니다."); + } + if (chunkCount < 0 || embeddingCount < 0) { + throw new IllegalArgumentException("Chunk와 Embedding 수는 음수일 수 없습니다."); + } + } + + private static boolean isNonNegativeFinite(double value) { + return Double.isFinite(value) && value >= 0.0; + } + } + + /** 한 구간 지연 표본의 p50·p95·p99와 최댓값을 밀리초 단위로 보존한다. */ + record LatencySummary(double p50, double p95, double p99, double max) { + } + + /** 한 문서 형식의 처리량, 저장량과 구간별 지연 분포를 구조화한다. */ + record FormatSummary( + int documentCount, + int chunkCount, + int embeddingCount, + double documentsPerSecond, + double documentsPerMinute, + double chunksPerSecond, + double embeddingsPerSecond, + LatencySummary uploadLatencyMillis, + LatencySummary queueLatencyMillis, + LatencySummary processingLatencyMillis, + LatencySummary e2eLatencyMillis + ) { + } +} diff --git a/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadStatisticsTest.java b/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadStatisticsTest.java new file mode 100644 index 0000000..5fdf8ca --- /dev/null +++ b/src/test/java/com/opensource/docgrid/e2e/DocumentIndexingE2ELoadStatisticsTest.java @@ -0,0 +1,86 @@ +package com.opensource.docgrid.e2e; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import com.opensource.docgrid.e2e.DocumentIndexingE2ELoadStatistics.DocumentFormat; +import com.opensource.docgrid.e2e.DocumentIndexingE2ELoadStatistics.DocumentMeasurement; +import com.opensource.docgrid.e2e.DocumentIndexingE2ELoadStatistics.FormatSummary; + +/** + * PDF·DOCX E2E 부하 Benchmark가 동일한 형식 혼합과 형식별 통계 계약을 유지하는지 검증한다. + */ +@DisplayName("PDF DOCX 전체 인덱싱 부하 통계 단위 테스트") +class DocumentIndexingE2ELoadStatisticsTest { + + @Test + @DisplayName("균형 잡힌 PDF DOCX 표본을 형식별 처리량과 지연으로 집계한다") + void summarizeByFormat_aggregatesBalancedDocumentMix() { + List measurements = List.of( + measurement(DocumentFormat.PDF, 100.0, 1, 1), + measurement(DocumentFormat.DOCX, 200.0, 2, 2), + measurement(DocumentFormat.PDF, 300.0, 3, 3), + measurement(DocumentFormat.DOCX, 400.0, 4, 4) + ); + + Map summaries = + DocumentIndexingE2ELoadStatistics.summarizeByFormat(measurements, 2.0); + + assertThat(summaries).containsOnlyKeys(DocumentFormat.PDF, DocumentFormat.DOCX); + assertThat(summaries.get(DocumentFormat.PDF).documentCount()).isEqualTo(2); + assertThat(summaries.get(DocumentFormat.PDF).chunkCount()).isEqualTo(4); + assertThat(summaries.get(DocumentFormat.PDF).documentsPerSecond()).isEqualTo(1.0); + assertThat(summaries.get(DocumentFormat.PDF).uploadLatencyMillis().p50()).isEqualTo(200.0); + assertThat(summaries.get(DocumentFormat.DOCX).embeddingCount()).isEqualTo(6); + assertThat(summaries.get(DocumentFormat.DOCX).documentsPerMinute()).isEqualTo(60.0); + assertThat(summaries.get(DocumentFormat.DOCX).uploadLatencyMillis().max()).isEqualTo(400.0); + } + + @Test + @DisplayName("PDF DOCX 개수가 다르거나 Profile 시간이 유효하지 않으면 거부한다") + void summarizeByFormat_rejectsInvalidProfile() { + List unbalanced = List.of( + measurement(DocumentFormat.PDF, 100.0, 1, 1), + measurement(DocumentFormat.PDF, 200.0, 1, 1), + measurement(DocumentFormat.DOCX, 300.0, 1, 1) + ); + + assertThatThrownBy(() -> + DocumentIndexingE2ELoadStatistics.summarizeByFormat(unbalanced, 1.0) + ).isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("PDF와 DOCX"); + assertThatThrownBy(() -> + DocumentIndexingE2ELoadStatistics.summarizeByFormat( + List.of( + measurement(DocumentFormat.PDF, 100.0, 1, 1), + measurement(DocumentFormat.DOCX, 100.0, 1, 1) + ), + 0.0 + ) + ).isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("경과 시간"); + } + + private DocumentMeasurement measurement( + DocumentFormat format, + double uploadMillis, + int chunkCount, + int embeddingCount + ) { + return new DocumentMeasurement( + format, + uploadMillis, + uploadMillis + 10.0, + uploadMillis + 20.0, + uploadMillis + 30.0, + chunkCount, + embeddingCount + ); + } +} diff --git a/src/test/java/com/opensource/docgrid/e2e/LocalE2eDocumentFactory.java b/src/test/java/com/opensource/docgrid/e2e/LocalE2eDocumentFactory.java index aac7374..d328fcb 100644 --- a/src/test/java/com/opensource/docgrid/e2e/LocalE2eDocumentFactory.java +++ b/src/test/java/com/opensource/docgrid/e2e/LocalE2eDocumentFactory.java @@ -3,6 +3,7 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.List; import org.apache.pdfbox.pdmodel.PDDocument; import org.apache.pdfbox.pdmodel.PDPage; @@ -67,11 +68,25 @@ static DocumentPayload docx( String heading, String body ) throws IOException { + return docx(fileName, title, List.of(new DocumentSection(heading, body))); + } + + static DocumentPayload docx( + String fileName, + String title, + List sections + ) throws IOException { + if (sections == null || sections.isEmpty()) { + throw new IllegalArgumentException("DOCX Section은 하나 이상이어야 합니다."); + } + try (XWPFDocument document = new XWPFDocument()) { - XWPFParagraph headingParagraph = document.createParagraph(); - headingParagraph.setStyle("Heading1"); - headingParagraph.createRun().setText(heading); - document.createParagraph().createRun().setText(body); + for (DocumentSection section : sections) { + XWPFParagraph headingParagraph = document.createParagraph(); + headingParagraph.setStyle("Heading1"); + headingParagraph.createRun().setText(section.heading()); + document.createParagraph().createRun().setText(section.body()); + } ByteArrayOutputStream output = new ByteArrayOutputStream(); document.write(output); @@ -107,4 +122,14 @@ record DocumentPayload( byte[] content ) { } + + /** DOCX Fixture의 Heading과 그 Heading에 속한 본문을 한 Section으로 표현한다. */ + record DocumentSection(String heading, String body) { + + DocumentSection { + if (heading == null || heading.isBlank() || body == null || body.isBlank()) { + throw new IllegalArgumentException("DOCX Section 제목과 본문은 비어 있을 수 없습니다."); + } + } + } }