Class ModuleEnginePerfSession

java.lang.Object
ai.pipestream.module.pipelineprobe.moduletest.engineperf.ModuleEnginePerfSession
@ApplicationScoped public class ModuleEnginePerfSession extends Object
Engine-path perf (Test 2): the sidecar acts as its OWN connector and pushes corpus docs through the production front door — connector-intake's uploadPipeDocStream — so the engine does everything it does for a real crawl: claim-check staging, fan-out to the subscribed graph, work dispensing over the bidi work stream, and the module's own worker loop self-ramps against the backlog. Subtracting the door test's raw numbers from this gives the per-step engine overhead.

Stable identity per module (graphs cannot be deleted, and every datasource binding fans out forever): one account engineperf-<module> (its own cluster + drives), one testing-sidecar-type datasource, one graph engineperf-<module> = intake node → the module node (terminal, no outgoing edges: the engine no-routes after the ack). Each RUN mints a fresh crawl_id — the measurement key — and rotates the datasource's API key, which doubles as the no-concurrent-tests lock (a still-running older stream loses its key validity at the next handshake).

Node config is the e2e config: chunker carries sentence10x3OnBody(true), semantic-graph the boundary config, opensearch-sink the opensearch_instance/plan_ids pair pointing at a STABLE plan (plan-idx-engineperf-opensearch-sink, provisioned READY on start, reused across runs — fresh-plan-per-run would change the graph content under a permanent graph id).

Measurement rides the audit plane: the engine's StepExecutionRecords (keyed by our crawl_id) flow into RichObservationCollector under (crawlId, "tap-<module>") — production telemetry, no new plumbing.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final String
    CONNECTOR_TYPE
    Connector type this sidecar registers itself under.

  • Constructor Summary

    Constructors
    Constructor
    Description
    ModuleEnginePerfSession()
    Creates the engine-perf session.

  • Method Summary

    Modifier and Type
    Method
    Description
    Map<String,Object>
    start(String module, int docs)
    Start an engine-path perf run: ensure the stable identity (account, datasource, graph — all idempotent get-or-create), rotate the API key (the run lock), then stream docs corpus docs through intake on a background thread.
    Map<String,Object>
    status()
    Live status: intake-side counters + audit-plane counts/rates.
    Map<String,Object>
    stop()
    Stop publishing, close the intake stream, unregister the crawl.

    Methods inherited from class Object

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

  • Field Details

    • CONNECTOR_TYPE

      public static final String CONNECTOR_TYPE
      Connector type this sidecar registers itself under.
      See Also:

  • Constructor Details

    • ModuleEnginePerfSession

      public ModuleEnginePerfSession()
      Creates the engine-perf session. Instances are managed by CDI.

  • Method Details

    • start

      public Map<String,Object> start(String module, int docs)
      Start an engine-path perf run: ensure the stable identity (account, datasource, graph — all idempotent get-or-create), rotate the API key (the run lock), then stream docs corpus docs through intake on a background thread.
      Parameters:
      module - the module to drive (selects the per-step scaffold, graph, and config)
      docs - the number of docs to stream; when <= 0 the module's full corpus size is used
      Returns:
      the initial run status snapshot
      Throws:
      IllegalStateException - if a run is already in progress

    • status

      public Map<String,Object> status()
      Live status: intake-side counters + audit-plane counts/rates.
      Returns:
      a status snapshot combining intake-side upload counters with audit-plane processing counts, backlog, rates, and the drained flag

    • stop

      public Map<String,Object> stop()
      Stop publishing, close the intake stream, unregister the crawl.
      Returns:
      the final status snapshot, with running forced to false