Class ModuleTestResource

java.lang.Object
ai.pipestream.module.pipelineprobe.moduletest.ModuleTestResource
@Path("/test-sidecar/v1/module-test") @Produces("application/json") public class ModuleTestResource extends Object
Module Test — invoke a single module's processor directly via its dev test door (POST /module-runtime/v1/process-once), bypassing the engine pull-loop and queue (roadmap #3). Direct functional + stress testing of one module in isolation.

Corpus input comes from the <step>-input-court fixture jars, auto-discovered via SampleDataRegistry: the dataset whose stage equals a module name is that module's captured INPUT, so feeding it back exercises exactly that step. Every processing step that has a dataset on the classpath (chunker, embedder, semantic-graph, opensearch-sink) gets a corpus; modules without one (e.g. parser) use the ad-hoc /raw path.

This path is independent of the e2e pipeline-crawl — it never touches the engine, Kafka, or the crawl flow.

  • Constructor Summary

    Constructors
    Constructor
    Description
    ModuleTestResource()
    Creates the module-test JAX-RS resource.

  • Method Summary

    Modifier and Type
    Method
    Description
    jakarta.ws.rs.core.Response
    enginePerfStart(String module, int docs)
    Start an engine-path perf run: stream docs corpus docs through connector-intake (the production front door) into the module's stable single-step graph; the engine stages, fans out, and the module's own worker loop pulls.
    jakarta.ws.rs.core.Response
    enginePerfStatus()
    Live engine-perf status: intake counters + audit-plane counts/rates.
    jakarta.ws.rs.core.Response
    enginePerfStop()
    Stop publishing and close the intake stream.
    jakarta.ws.rs.core.Response
    perfPause()
    Pause the load (workers to zero) WITHOUT tearing down the scaffold — the plan/index stay live so downstream consumers can drain and be measured.
    jakarta.ws.rs.core.Response
    perfRampDown()
    Step concurrency down one geometric level.
    jakarta.ws.rs.core.Response
    perfRampUp()
    Step concurrency up one geometric level (1→2→4→8→16→32→64).
    jakarta.ws.rs.core.Response
    perfStart(String module, int buffer, String strategy)
    Start a perf session for module and begin a sustained door load at concurrency 1; drive it with ramp-up / ramp-down and poll /perf/stats.
    jakarta.ws.rs.core.Response
    perfStats()
    Live metrics snapshot for the current ramp level (min/max/mean/p50/p95/p99 + docs/sec).
    jakarta.ws.rs.core.Response
    perfStop()
    Stop the perf session and release the load pool.
    jakarta.ws.rs.core.Response
    rampGet(String module)
    Module pool state from its ops ramp surface (workers/min/max/ceiling).
    jakarta.ws.rs.core.Response
    rampSet(String module, String jsonBody)
    Set/clear the module's live concurrency ceiling ({"ceiling": N}; N <= 0 clears).
    jakarta.ws.rs.core.Response
    runCorpus(String module, int index)
    Run the module against one corpus doc: load its stage-appropriate input PipeDoc, wrap it, and POST to the door.
    jakarta.ws.rs.core.Response
    runRaw(String module, String kind, byte[] body)
    Run the module against an ad-hoc uploaded protobuf.
    jakarta.ws.rs.core.Response
    targets()
    List configured module targets and, for each, whether a corpus dataset feeds it and how many docs it carries (the expected stress count).

    Methods inherited from class Object

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

  • Constructor Details

    • ModuleTestResource

      public ModuleTestResource()
      Creates the module-test JAX-RS resource. Collaborators are injected by the container; instances are managed by CDI / JAX-RS.

  • Method Details

    • targets

      @GET @Path("/targets") public jakarta.ws.rs.core.Response targets()
      List configured module targets and, for each, whether a corpus dataset feeds it and how many docs it carries (the expected stress count).
      Returns:
      a 200 response whose JSON body maps each configured module key to its baseUrl, corpus name, and corpusDocCount

    • runCorpus

      @POST @Path("/{module}/corpus") public jakarta.ws.rs.core.Response runCorpus(@PathParam("module") String module, @QueryParam("index") @DefaultValue("1") int index)
      Run the module against one corpus doc: load its stage-appropriate input PipeDoc, wrap it, and POST to the door.
      Parameters:
      module - configured module key (== the dataset stage it feeds)
      index - 1-based corpus index
      Returns:
      the door's response (status + JSON body) proxied back to the caller, or a 400/404 JSON error when no corpus feeds the module or no doc exists at the requested index

    • perfStart

      @POST @Path("/{module}/perf/start") public jakarta.ws.rs.core.Response perfStart(@PathParam("module") String module, @QueryParam("buffer") @DefaultValue("200") int buffer, @QueryParam("strategy") String strategy)
      Start a perf session for module and begin a sustained door load at concurrency 1; drive it with ramp-up / ramp-down and poll /perf/stats. buffer sets the payload supply: >= corpus size caches every wrapped doc in memory once (raw baseline); smaller values stream the corpus through a bounded queue of that size.
      Parameters:
      module - configured module key to load
      buffer - payload supply size; >= corpus size caches every wrapped doc in memory, smaller values stream via a bounded queue of this capacity
      strategy - opensearch-sink only: indexing strategy passed to the session (null = default)
      Returns:
      a 200 response with the session's opening snapshot, or a 400 JSON error when the module/arguments are rejected

    • perfPause

      @POST @Path("/perf/pause") public jakarta.ws.rs.core.Response perfPause()
      Pause the load (workers to zero) WITHOUT tearing down the scaffold — the plan/index stay live so downstream consumers can drain and be measured. Resume with ramp-up.
      Returns:
      a 200 response with the post-pause session snapshot

    • perfRampUp

      @POST @Path("/perf/ramp-up") public jakarta.ws.rs.core.Response perfRampUp()
      Step concurrency up one geometric level (1→2→4→8→16→32→64).
      Returns:
      a 200 response with the session snapshot at the new ramp level

    • perfRampDown

      @POST @Path("/perf/ramp-down") public jakarta.ws.rs.core.Response perfRampDown()
      Step concurrency down one geometric level.
      Returns:
      a 200 response with the session snapshot at the new ramp level

    • perfStats

      @GET @Path("/perf/stats") public jakarta.ws.rs.core.Response perfStats()
      Live metrics snapshot for the current ramp level (min/max/mean/p50/p95/p99 + docs/sec).
      Returns:
      a 200 response with the current non-destructive session snapshot

    • perfStop

      @POST @Path("/perf/stop") public jakarta.ws.rs.core.Response perfStop()
      Stop the perf session and release the load pool.
      Returns:
      a 200 response with the final session snapshot

    • enginePerfStart

      @POST @Path("/{module}/engine-perf/start") public jakarta.ws.rs.core.Response enginePerfStart(@PathParam("module") String module, @QueryParam("docs") @DefaultValue("1000") int docs)
      Start an engine-path perf run: stream docs corpus docs through connector-intake (the production front door) into the module's stable single-step graph; the engine stages, fans out, and the module's own worker loop pulls. Measurement rides the audit plane under a fresh crawl_id.
      Parameters:
      module - configured module key whose stable single-step graph receives the streamed corpus
      docs - number of corpus docs to stream through connector-intake
      Returns:
      a 200 response with the engine-perf session's opening status, or a 400 JSON error when the module/arguments are rejected

    • enginePerfStatus

      @GET @Path("/engine-perf/status") public jakarta.ws.rs.core.Response enginePerfStatus()
      Live engine-perf status: intake counters + audit-plane counts/rates.
      Returns:
      a 200 response with the current engine-perf status

    • enginePerfStop

      @POST @Path("/engine-perf/stop") public jakarta.ws.rs.core.Response enginePerfStop()
      Stop publishing and close the intake stream.
      Returns:
      a 200 response with the final engine-perf status

    • rampGet

      @GET @Path("/{module}/ramp") public jakarta.ws.rs.core.Response rampGet(@PathParam("module") String module)
      Module pool state from its ops ramp surface (workers/min/max/ceiling).
      Parameters:
      module - configured module key whose ramp surface is queried
      Returns:
      the module's ramp response proxied back (status + JSON body), or a 400/502 JSON error when the module is unknown or unreachable

    • rampSet

      @POST @Path("/{module}/ramp") @Consumes("application/json") public jakarta.ws.rs.core.Response rampSet(@PathParam("module") String module, String jsonBody)
      Set/clear the module's live concurrency ceiling ({"ceiling": N}; N <= 0 clears).
      Parameters:
      module - configured module key whose ramp surface is updated
      jsonBody - the ramp request body; a blank/null body is treated as an empty JSON object {}
      Returns:
      the module's ramp response proxied back (status + JSON body), or a 400/502 JSON error when the module is unknown or unreachable

    • runRaw

      @POST @Path("/{module}/raw") @Consumes("application/octet-stream") public jakarta.ws.rs.core.Response runRaw(@PathParam("module") String module, @QueryParam("kind") @DefaultValue("pipedoc") String kind, byte[] body)
      Run the module against an ad-hoc uploaded protobuf. kind=pipedoc (default) wraps the bytes as a PipeStream; kind=pipestream forwards them as-is.
      Parameters:
      module - configured module key to invoke
      kind - pipedoc (default) to wrap the bytes as a PipeStream, or pipestream to forward them as-is
      body - the serialized protobuf payload
      Returns:
      the door's response (status + JSON body) proxied back, or a 400 JSON error when the body is empty or fails to decode as the kind