[adapters] Report total_processed_steps for output connectors in API.

blp · blp · commit 3bce09eac676 · 2026-02-27T20:13:04.000Z
This has been available inside the pipeline for a while, but we forgot to
expose it through the API.  This fixes that.

This also fixes up and enables the corresponding Python integration test.

Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs
@@ -1310,6 +1310,10 @@ impl ControllerStatus {
                         .metrics
                         .total_processed_input_records
                         .load(Ordering::Relaxed),
+                    total_processed_steps: output
+                        .metrics
+                        .total_processed_steps
+                        .load(Ordering::Relaxed),
                     memory: output.metrics.memory.load(Ordering::Relaxed),
                 },
                 fatal_error: output.fatal_error.lock().unwrap().clone(),
@@ -2201,11 +2205,18 @@ pub struct OutputEndpointMetrics {
     /// This metric tracks the end-to-end progress of the pipeline: the output
     /// of this endpoint is equal to the output of the circuit after
     /// processing `total_processed_input_records` records.
+    ///
+    /// In a multihost pipeline, this count reflects only the input records
+    /// processed on the same host as the output endpoint, which is not usually
+    /// meaningful.
     pub total_processed_input_records: AtomicU64,
 
     /// The number of steps whose input records have been processed by the
     /// endpoint.
     ///
+    /// This is meaningful in a multihost pipeline because steps are
+    /// synchronized across all of the hosts.
+    ///
     /// # Interpretation
     ///
     /// This is a count, not a step number.  If `total_processed_steps` is 0, no
@@ -2272,6 +2283,7 @@ impl OutputEndpointMetrics {
             total_processed_input_records: self
                 .total_processed_input_records
                 .load(Ordering::Relaxed),
+            total_processed_steps: self.total_processed_steps.load(Ordering::Relaxed),
             memory: self.memory.load(Ordering::Relaxed),
         }
     }
diff --git a/crates/feldera-types/src/adapter_stats.rs b/crates/feldera-types/src/adapter_stats.rs
@@ -206,7 +206,30 @@ pub struct ExternalOutputEndpointMetrics {
     /// Number of transport errors.
     pub num_transport_errors: u64,
     /// The number of input records processed by the circuit.
+    ///
+    /// This metric tracks the end-to-end progress of the pipeline: the output
+    /// of this endpoint is equal to the output of the circuit after
+    /// processing `total_processed_input_records` records.
+    ///
+    /// In a multihost pipeline, this count reflects only the input records
+    /// processed on the same host as the output endpoint, which is not usually
+    /// meaningful.
     pub total_processed_input_records: u64,
+    /// The number of steps whose input records have been processed by the
+    /// endpoint.
+    ///
+    /// This is meaningful in a multihost pipeline because steps are
+    /// synchronized across all of the hosts.
+    ///
+    /// # Interpretation
+    ///
+    /// This is a count, not a step number.  If `total_processed_steps` is 0, no
+    /// steps have been processed to completion.  If `total_processed_steps >
+    /// 0`, then the last step whose input records have been processed to
+    /// completion is `total_processed_steps - 1`. A record that was ingested in
+    /// step `n` is fully processed when `total_processed_steps > n`.
+    #[schema(value_type = u64)]
+    pub total_processed_steps: Step,
     /// Extra memory in use beyond that used for queuing records.
     pub memory: u64,
 }
diff --git a/openapi.json b/openapi.json
@@ -9398,6 +9398,7 @@
           "num_encode_errors",
           "num_transport_errors",
           "total_processed_input_records",
+          "total_processed_steps",
           "memory"
         ],
         "properties": {
@@ -9446,7 +9447,13 @@
           "total_processed_input_records": {
             "type": "integer",
             "format": "int64",
-            "description": "The number of input records processed by the circuit.",
+            "description": "The number of input records processed by the circuit.\n\nThis metric tracks the end-to-end progress of the pipeline: the output\nof this endpoint is equal to the output of the circuit after\nprocessing `total_processed_input_records` records.\n\nIn a multihost pipeline, this count reflects only the input records\nprocessed on the same host as the output endpoint, which is not usually\nmeaningful.",
+            "minimum": 0
+          },
+          "total_processed_steps": {
+            "type": "integer",
+            "format": "int64",
+            "description": "The number of steps whose input records have been processed by the\nendpoint.\n\nThis is meaningful in a multihost pipeline because steps are\nsynchronized across all of the hosts.\n\n# Interpretation\n\nThis is a count, not a step number.  If `total_processed_steps` is 0, no\nsteps have been processed to completion.  If `total_processed_steps >\n0`, then the last step whose input records have been processed to\ncompletion is `total_processed_steps - 1`. A record that was ingested in\nstep `n` is fully processed when `total_processed_steps > n`.",
             "minimum": 0
           },
           "transmitted_bytes": {
diff --git a/python/feldera/stats.py b/python/feldera/stats.py
@@ -43,20 +43,30 @@ def __init__(self):
         """
         self.state: Optional[PipelineStatus] = None
         self.bootstrap_in_progress: Optional[bool] = None
+        self.transaction_status: Optional[TransactionStatus] = None
+        self.transaction_id: Optional[int] = None
+        self.commit_progress: Optional[CommitProgressSummary] = None
+        self.transaction_initiators: Optional[TransactionInitiators] = None
         self.rss_bytes: Optional[int] = None
         self.cpu_msecs: Optional[int] = None
+        self.uptime_msecs: Optional[float] = None
         self.start_time: Optional[datetime] = None
         self.incarnation_uuid: Optional[uuid.UUID] = None
+        self.initial_start_time: Optional[datetime] = None
         self.storage_bytes: Optional[int] = None
         self.storage_mb_secs: Optional[int] = None
         self.runtime_elapsed_msecs: Optional[int] = None
         self.buffered_input_records: Optional[int] = None
         self.total_input_records: Optional[int] = None
+        self.buffered_input_bytes: Optional[int] = None
+        self.total_input_bytes: Optional[int] = None
         self.total_processed_records: Optional[int] = None
+        self.total_processed_bytes: Optional[int] = None
         self.total_completed_records: Optional[int] = None
+        self.output_stall_msecs: Optional[int] = None
+        self.total_initiated_steps: Optional[int] = None
+        self.total_completed_steps: Optional[int] = None
         self.pipeline_complete: Optional[bool] = None
-        self.transaction_status: Optional[TransactionStatus] = None
-        self.transaction_id: Optional[int] = None
 
     @classmethod
     def from_dict(cls, d: Mapping[str, Any]):
@@ -65,6 +75,7 @@ def from_dict(cls, d: Mapping[str, Any]):
         metrics.state = PipelineStatus.from_str(d["state"])
         metrics.incarnation_uuid = uuid.UUID(d["incarnation_uuid"])
         metrics.start_time = datetime.fromtimestamp(d["start_time"])
+        metrics.initial_start_time = datetime.fromtimestamp(d["start_time"])
         metrics.transaction_status = TransactionStatus.from_str(d["transaction_status"])
         return metrics
 
@@ -100,6 +111,7 @@ def __init__(self):
         self.total_bytes: Optional[int] = None
         self.total_records: Optional[int] = None
         self.buffered_records: Optional[int] = None
+        self.buffered_bytes: Optional[int] = None
         self.num_transport_errors: Optional[int] = None
         self.num_parse_errors: Optional[int] = None
         self.end_of_input: Optional[bool] = None
@@ -141,12 +153,69 @@ def __init__(self):
         self.transmitted_bytes: Optional[int] = None
         self.queued_records: Optional[int] = None
         self.queued_batches: Optional[int] = None
+        self.buffered_records: Optional[int] = None
+        self.buffered_batches: Optional[int] = None
         self.num_encode_errors: Optional[int] = None
         self.num_transport_errors: Optional[int] = None
         self.total_processed_input_records: Optional[int] = None
+        self.total_processed_steps: Optional[int] = None
+        self.memory: Optional[int] = None
 
     @classmethod
     def from_dict(cls, d: Mapping[str, Any]):
         metrics = cls()
         metrics.__dict__.update(d)
         return metrics
+
+
+class CommitProgressSummary:
+    """Progress of a transaction commit.
+    """
+
+    def __init__(self):
+        """Initializes an empty status."""
+        self.completed: Optional[int] = None
+        self.in_progress: Optional[int] = None
+        self.remaining: Optional[int] = None
+        self.in_progress_processed_records: Optional[int] = None
+        self.in_progress_total_records: Optional[int] = None
+
+    @classmethod
+    def from_dict(cls, d: Mapping[str, Any]):
+        status = cls()
+        status.__dict__.update(d)
+        return status
+
+
+class TransactionInitiators:
+    """Initiators for an ongoing transaction.
+    """
+
+    def __init__(self):
+        """Initializes an empty status."""
+        self.transaction_id: Optional[int] = None
+        self.initiated_by_api: Optional[str] = None
+        self.initiated_by_connectors: Optional[Mapping[str, ConnectorTransactionPhase]] = None
+
+    @classmethod
+    def from_dict(cls, d: Mapping[str, Any]):
+        status = cls()
+        status.__dict__.update(d)
+        status.initiated_by_connectors = ConnectorTransactionPhase.from_dict(d["initiated_by_connectors"])
+        return status
+
+
+class ConnectorTransactionPhase:
+    """Connector transaction phase with optional label
+    """
+
+    def __init__(self):
+        """Initializes an empty status."""
+        self.phase: Optional[str] = None
+        self.label: Optional[str] = None
+
+    @classmethod
+    def from_dict(cls, d: Mapping[str, Any]):
+        status = cls()
+        status.__dict__.update(d)
+        return status
diff --git a/python/tests/platform/test_metrics_logs.py b/python/tests/platform/test_metrics_logs.py
@@ -18,6 +18,8 @@
 )
 
 from feldera.testutils import FELDERA_TEST_NUM_HOSTS
+from feldera.stats import PipelineStatistics
+from feldera.enums import PipelineStatus
 
 
 def _ingest_lines(name: str, table: str, body: str):
@@ -83,7 +85,6 @@ def test_pipeline_metrics(pipeline_name):
 
 
 @gen_pipeline_name
-@single_host_only
 def test_pipeline_stats(pipeline_name):
     """
     Tests retrieving pipeline statistics via `/stats`.
@@ -102,12 +103,14 @@ def test_pipeline_stats(pipeline_name):
     """.strip()
 
     create_pipeline(pipeline_name, sql)
-    start_pipeline(pipeline_name)
+    start_pipeline_as_paused(pipeline_name)
 
     # Create output connector on v1 (egress)
     r_out = post_no_body(api_url(f"/pipelines/{pipeline_name}/egress/v1"), stream=True)
     assert r_out.status_code == HTTPStatus.OK, (r_out.status_code, r_out.text)
 
+    resume_pipeline(pipeline_name)
+
     # Wait for datagen completion
     time.sleep(3)
     deadline = time.time() + 10
@@ -118,37 +121,48 @@ def test_pipeline_stats(pipeline_name):
         time.sleep(1)
     assert _adhoc_count(pipeline_name, "t1") == 5, "Did not ingest expected 5 rows"
 
+    # Wait for all the steps to be completed.
+    deadline = time.time() + 10
+    while time.time() < deadline:
+        r_stats = get(api_url(f"/pipelines/{pipeline_name}/stats"))
+        assert r_stats.status_code == HTTPStatus.OK, (r_stats.status_code, r_stats.text)
+        stats = PipelineStatistics.from_dict(r_stats.json())
+        gm = stats.global_metrics
+        steps = gm.total_initiated_steps
+        if steps is not None and steps == gm.total_completed_steps:
+            break
+
     r_stats = get(api_url(f"/pipelines/{pipeline_name}/stats"))
     assert r_stats.status_code == HTTPStatus.OK, (r_stats.status_code, r_stats.text)
-    stats = r_stats.json()
-    keys = sorted(stats.keys())
+    r_stats_json = r_stats.json()
+    keys = sorted(r_stats_json.keys())
     assert keys == ["global_metrics", "inputs", "outputs", "suspend_error"]
+    stats = PipelineStatistics.from_dict(r_stats_json)
 
-    gm = stats["global_metrics"]
-    assert gm.get("state") == "Running"
-    assert gm.get("total_input_records") == 5
-    assert gm.get("total_processed_records") == 5
-    assert gm.get("pipeline_complete")
-    assert gm.get("buffered_input_records") == 0
-    assert gm.get("buffered_input_bytes") == 0
+    assert gm.state == PipelineStatus.RUNNING
+    assert gm.total_input_records == 5
+    assert gm.total_processed_records == 5
+    assert gm.pipeline_complete
+    assert gm.buffered_input_records == 0
+    assert gm.buffered_input_bytes == 0
 
-    inputs = stats["inputs"]
-    assert isinstance(inputs, list) and len(inputs) == 1
+    inputs = stats.inputs
+    assert len(inputs) == 1
     inp = inputs[0]
-    assert inp.get("config", {}).get("stream") == "t1"
-    assert inp.get("metrics", {}).get("buffered_bytes") == 0
-    assert inp.get("metrics", {}).get("buffered_records") == 0
-    assert inp.get("metrics", {}).get("end_of_input")
-    assert inp.get("metrics", {}).get("num_parse_errors") == 0
-    assert inp.get("metrics", {}).get("num_transport_errors") == 0
-    assert inp.get("metrics", {}).get("total_bytes") == 40
-    assert inp.get("metrics", {}).get("total_records") == 5
-
-    outputs = stats["outputs"]
-    assert isinstance(outputs, list) and len(outputs) == 1
+    assert inp.config["stream"] == "t1"
+    assert inp.metrics.buffered_bytes == 0
+    assert inp.metrics.buffered_records == 0
+    assert inp.metrics.end_of_input
+    assert inp.metrics.num_parse_errors == 0
+    assert inp.metrics.num_transport_errors == 0
+    assert inp.metrics.total_bytes == 40
+    assert inp.metrics.total_records == 5
+
+    outputs = stats.outputs
+    assert len(outputs) == 1
     out = outputs[0]
-    assert out.get("config", {}).get("stream") == "v1"
-    assert out.get("metrics", {}).get("total_processed_input_records") == 5
+    assert out.config["stream"] == "v1"
+    assert out.metrics.total_processed_steps == steps
 
     # /time_series
     r_ts = get(api_url(f"/pipelines/{pipeline_name}/time_series"))
