Class OpenSearchIndexingTestsResource

java.lang.Object

ai.pipestream.module.pipelineprobe.opensearch.OpenSearchIndexingTestsResource

@Path("/api/opensearch-indexing-tests")
@Produces("application/json")
public class OpenSearchIndexingTestsResource
extends Object

REST resource backing the testing-sidecar's "OpenSearch Index Tests" tab.

Surfaces OSM-internal indexing-pipeline state that the existing tabs (Index Admin, Stress Test, E2E Tests) don't expose: per-plan redis stream depth, DLQ stream contents, live receipt-topic tail, corpus-driven push tester, and per-plan clean-up.

Mutiny-free reference

This class is the FIRST testing-sidecar resource written in the post-Mutiny style. Every other resource in this module (see IndexAdminResource, E2EPipelineTestService, etc.) returns Uni<Response> and chains via Mutiny operators; this one returns plain Response on a virtual thread and uses the synchronous GrpcClientFactory#getBlockingClient surface.

When the broader testing-sidecar Mutiny rip happens, copy from here. Rules in force:

• NO io.smallrye.mutiny.* imports, ever.
• NO Mutiny stubs (MutinyXxxServiceGrpc); use the plain blocking stubs (XxxServiceGrpc.XxxServiceBlockingStub).
• NO dynamic-grpc factory (task #40 is retiring it from this module). Use the standard GrpcClient injection + Stork discovery configured in application.properties.
• NO Uni<> or Multi<> on method signatures or fields.
• RunOnVirtualThread on every handler so blocking I/O is a non-issue.
• Plain List / Map return shapes; no reactive streams.

Endpoint surface (filled in across stacked PRs):

1. GET /plans — list READY plans (this PR).
2. GET /plans/{id}/state — XLEN, XPENDING, DLQ XLEN.
3. GET /plans/{id}/receipts/recent?limit=N — receipt tail.
4. GET /plans/{id}/dlq/recent?limit=N — DLQ tail.
5. POST /plans/{id}/push — drive court-1000 docs through the sink.
6. GET /jobs/{id} — push-job progress.
7. POST /plans/{id}/cleanup — XTRIM streams + drop test index.

Constructor Summary

Constructors

Constructor	Description
OpenSearchIndexingTestsResource()	CDI.

Method Summary

Modifier and Type	Method	Description
jakarta.ws.rs.core.Response	cleanupPlan(String planId)	Drop both per-plan redis streams and (when the index name passes the prefix safety guard) the underlying OpenSearch index.
jakarta.ws.rs.core.Response	dlqRecent(String planId, int limit)	Most recent N entries from the per-plan DLQ stream.
jakarta.ws.rs.core.Response	jobSnapshot(String jobId)	Snapshot of a push job's progress.
jakarta.ws.rs.core.Response	listJobs()	List all push jobs the sidecar has tracked since startup, newest first.
jakarta.ws.rs.core.Response	listPlans()	List every IndexPlanEntity known to OSM, projected to a UI-friendly shape.
jakarta.ws.rs.core.Response	planState(String planId)	Live redis-stream state for one plan: how many entries are sitting on the indexing stream (XLEN), how many are pending an XACK from the consumer group (XPENDING), and how many failures are in the DLQ (XLEN of the dlq stream).
jakarta.ws.rs.core.Response	setupCourtPlan(Map<String,Object> body)	One-shot setup: register ChunkerConfigs, EmbeddingConfig, VectorSets, and an IndexPlan matching the court-1000 chunker fixtures.
jakarta.ws.rs.core.Response	startPush(String planId, Map<String,Object> body)	Start a push job driving court-1000 PipeDocs at OSS over the OpenSearchIngestionService.StreamDocuments bidi RPC.

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

OpenSearchIndexingTestsResource

public OpenSearchIndexingTestsResource()

CDI.

Method Details

listPlans

@GET
@Path("/plans")
public jakarta.ws.rs.core.Response listPlans()

List every IndexPlanEntity known to OSM, projected to a UI-friendly shape. The frontend's plan picker reads this once on tab open and again whenever the user clicks refresh.

Returns plans in whatever order OSM serves them (currently insertion order). Filtering and pagination live on the gRPC surface but are not yet plumbed through this endpoint — the plan count in dev is small enough that the full list is fine.

Returns:

200 {plans: [...], total: N} on success, or a 502 {error: "..."} when the OSM gRPC call fails (service unreachable, gRPC error, timeout)

planState

@GET
@Path("/plans/{planId}/state")
public jakarta.ws.rs.core.Response planState(@PathParam("planId")
 String planId)

Live redis-stream state for one plan: how many entries are sitting on the indexing stream (XLEN), how many are pending an XACK from the consumer group (XPENDING), and how many failures are in the DLQ (XLEN of the dlq stream). The tab's per-plan status panel polls this every few seconds.

Missing keys are treated as zero — redis's XLEN on a nonexistent stream returns 0, and XPENDING on a missing group is caught and surfaced as 0 pending so the panel doesn't flap during the window before any sink publishes against a fresh plan.

Parameters:

planId - the IndexPlanEntity.id for the plan we're inspecting. The redis stream key is <streamKeyPrefix><planId>.

Returns:

200 { planId, indexingStreamKey, indexingXlen, pendingCount, dlqStreamKey, dlqXlen, consumerGroup } on success; 502 { error } when the redis call fails for any reason other than the missing-group case

cleanupPlan

@POST
@Path("/plans/{planId}/cleanup")
public jakarta.ws.rs.core.Response cleanupPlan(@PathParam("planId")
 String planId)

Drop both per-plan redis streams and (when the index name passes the prefix safety guard) the underlying OpenSearch index. Used by the tab's "Reset this plan" button so an operator can start a fresh push test without leftover state from a previous run.

Steps:

1. Look up the plan via IndexPlanService.GetIndexPlan — gives us the plan's indexName for the OS delete. Missing plan → 404.
2. XTRIM pipestream:indexing:<planId> MAXLEN 0 — drains the indexing stream.
3. XTRIM pipestream:indexing-dlq:<planId> MAXLEN 0 — drains the DLQ stream.
4. If indexName starts with an allowed prefix (see cleanup-allowed-prefixes), call OpenSearchManagerService.DeleteIndex. Otherwise skip with a "not-allowed-by-prefix" note in the response — protects real plans from accidental nuke.

Returns a per-step summary so the UI can show what actually happened. Every step is idempotent: re-running cleanup on an already-empty plan is a no-op.

Parameters:

planId - the plan to reset

Returns:

200 { planId, indexName, indexingTrimmed, dlqTrimmed, indexDeleted, indexDeleteSkippedReason }

dlqRecent

@GET
@Path("/plans/{planId}/dlq/recent")
public jakarta.ws.rs.core.Response dlqRecent(@PathParam("planId")
 String planId,
 @QueryParam("limit") @DefaultValue("20")
 int limit)

Most recent N entries from the per-plan DLQ stream. Designed for the tab's "recent failures" panel: shows the operator what failed and why without forcing them to redis-cli.

Returns entries newest-first via XREVRANGE. Each entry exposes its redis id plus a hand-picked subset of the original field map — request_payload is intentionally OMITTED because it's base64-encoded protobuf bytes that would just bloat the JSON without being human-readable. The failure_reason and failed_at_ms fields appended by IndexingDlqWriter are the load-bearing payload for triage.

Parameters:

planId - the IndexPlanEntity.id whose DLQ we're tailing

limit - cap on entries returned (default 20, max 200 to keep JSON payloads bounded)

Returns:

200 { planId, dlqStreamKey, entries: [...], count } on success; missing DLQ stream returns entries: []; 502 { error } on any other redis failure

setupCourtPlan

@POST
@Path("/setup")
public jakarta.ws.rs.core.Response setupCourtPlan(Map<String,Object> body)

One-shot setup: register ChunkerConfigs, EmbeddingConfig, VectorSets, and an IndexPlan matching the court-1000 chunker fixtures. Idempotent — re-runs return the same ids.

Body (optional):

{"indexName": "idx-test-court-1000"}

Response: planId / planName / planStatus / audit log of every step. Use the returned planName as planIds[0] on the subsequent POST /plans/{...}/push.

Parameters:

body - request parameters; may be null. Only indexName is read (defaulting when absent).

Returns:

200 with the setup result map on success; 502 with an error body when setup throws

startPush

@POST
@Path("/plans/{planId}/push")
public jakarta.ws.rs.core.Response startPush(@PathParam("planId")
 String planId,
 Map<String,Object> body)

Start a push job driving court-1000 PipeDocs at OSS over the OpenSearchIngestionService.StreamDocuments bidi RPC. The push runs on a virtual thread; this endpoint returns immediately with the job id so the FE can poll jobSnapshot(String).

Request body shape (all fields optional):

{
  "stage": "embedder",   // or "chunker"; default "embedder"
  "limit": 1000,         // clamped to [1, 1000]; default 1000
  "concurrency": 16      // in-flight cap; default from config
}

Parameters:

planId - target plan; must exist in OSM (no pre-validation here, OSS surfaces "unknown plan" as per-doc NACK)

body - push parameters (see above); may be null for defaults

Returns:

202 { jobId, ... initial snapshot fields } on success; 400 when the corpus is missing on the classpath; 503 when max-jobs active jobs are already running

jobSnapshot

@GET
@Path("/jobs/{jobId}")
public jakarta.ws.rs.core.Response jobSnapshot(@PathParam("jobId")
 String jobId)

Snapshot of a push job's progress. Polled by the FE every second or two while state == "RUNNING"; once state transitions to "COMPLETED" or "FAILED" the FE can stop polling.

Parameters:

jobId - job id returned by startPush(String, Map)

Returns:

200 with the snapshot, 404 when no such job

listJobs

@GET
@Path("/jobs")
public jakarta.ws.rs.core.Response listJobs()

List all push jobs the sidecar has tracked since startup, newest first. Used by the FE's history pane on the OpenSearch Index Tests tab.

Returns:

200 { jobs: [...], count: N }