Skip to content

python: surface commit/ingest progress in wait loop warnings #6780

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
blp merged 2 commits into from
Jul 31, 2026
Merged

python: surface commit/ingest progress in wait loop warnings #6780

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1e9fe57
python: surface commit/ingest progress in wait loop warnings
blp Jul 31, 2026
9199a89
python: name wait-loop LongOperationWarning locals after what they await
blp Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 18 additions & 7 deletions python/feldera/pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,11 +387,19 @@ def wait_for_completion(
raise RuntimeError("Pipeline must be running to wait for completion")

start_time = time.monotonic()
long_op = LongOperationWarning(
# Reused by the warning message below; is_complete() would fetch and
# discard the same stats, hiding whether the wait is stalled or just
# slow (e.g. mid-transaction-commit).
latest_stats: Optional[PipelineStatistics] = None
wait_for_completion = LongOperationWarning(
logger,
lambda elapsed: f"still waiting for pipeline {self.name} to complete, "
f"waited {elapsed:.1f} seconds",
lambda elapsed: f"pipeline {self.name} completed after {elapsed:.1f} seconds",
lambda elapsed: (
f"still waiting for pipeline {self.name} to complete, "
f"waited {elapsed:.1f} seconds ({latest_stats.global_metrics.progress_summary()})"
),
lambda elapsed: (
f"pipeline {self.name} completed after {elapsed:.1f} seconds"
),
)

while True:
Expand All @@ -403,16 +411,19 @@ def wait_for_completion(
f" pipeline '{self.name}' to complete"
)

pipeline_complete: bool = self.is_complete()
latest_stats = self.stats()
pipeline_complete: Optional[bool] = (
latest_stats.global_metrics.pipeline_complete
)
if pipeline_complete is None:
raise RuntimeError(
"received unknown metrics from the pipeline, pipeline_complete is None"
)
elif pipeline_complete:
long_op.done()
wait_for_completion.done()
break

long_op.check()
wait_for_completion.check()
time.sleep(1)

if force_stop:
Expand Down
42 changes: 21 additions & 21 deletions python/feldera/rest/feldera_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,7 @@ def _wait_for_compilation(
"""Wait for pipeline compilation -- internal use only."""
wait = ["Pending", "CompilingSql", "SqlCompiled", "CompilingRust"]
start_time = time.monotonic()
long_op = LongOperationWarning(
wait_for_compilation = LongOperationWarning(
logger,
lambda elapsed: f"still compiling {name}, waited {elapsed:.1f} seconds",
lambda elapsed: f"{name} finished compiling after {elapsed:.1f} seconds",
Expand All @@ -193,7 +193,7 @@ def _wait_for_compilation(
status = p.program_status

if status == "Success":
long_op.done()
wait_for_compilation.done()
if expected_program_version is None:
return self.get_pipeline(name, PipelineFieldSelector.ALL)

Expand Down Expand Up @@ -231,7 +231,7 @@ def _wait_for_compilation(

raise RuntimeError(error_message)

long_op.check()
wait_for_compilation.check()
time.sleep(poll_interval_s)

def __wait_for_pipeline_state(
Expand All @@ -243,7 +243,7 @@ def __wait_for_pipeline_state(
poll_interval_s: float = 0.5,
):
start_time = time.monotonic()
long_op = LongOperationWarning(
wait_for_state = LongOperationWarning(
logger,
lambda elapsed: f"still waiting for {pipeline_name} to transition to "
f"'{state}', waited {elapsed:.1f} seconds",
Expand All @@ -264,7 +264,7 @@ def __wait_for_pipeline_state(
status = resp.deployment_status

if status.lower() == state.lower():
long_op.done()
wait_for_state.done()
break
elif (
status == "Stopped"
Expand All @@ -278,7 +278,7 @@ def __wait_for_pipeline_state(
{resp.deployment_error.get("message", "")}"""
)

long_op.check()
wait_for_state.check()
time.sleep(poll_interval_s)

def __wait_for_pipeline_state_one_of(
Expand All @@ -291,7 +291,7 @@ def __wait_for_pipeline_state_one_of(
) -> PipelineStatus:
start_time = time.monotonic()
states = [state.lower() for state in states]
long_op = LongOperationWarning(
wait_for_states = LongOperationWarning(
logger,
lambda elapsed: f"still waiting for {pipeline_name} to transition to "
f"one of {states}, waited {elapsed:.1f} seconds",
Expand All @@ -311,7 +311,7 @@ def __wait_for_pipeline_state_one_of(
status = resp.deployment_status

if status.lower() in states:
long_op.done()
wait_for_states.done()
return PipelineStatus.from_str(status)
elif (
status == "Stopped"
Expand All @@ -324,7 +324,7 @@ def __wait_for_pipeline_state_one_of(
Reason: The pipeline is in a STOPPED state due to the following error:
{resp.deployment_error.get("message", "")}"""
)
long_op.check()
wait_for_states.check()
time.sleep(poll_interval_s)

def create_pipeline(self, pipeline: Pipeline, wait: bool = True) -> Pipeline:
Expand Down Expand Up @@ -762,7 +762,7 @@ def stop_pipeline(
return

start = time.monotonic()
long_op = LongOperationWarning(
wait_for_stop = LongOperationWarning(
logger,
lambda elapsed: f"still stopping {pipeline_name}, waited {elapsed:.1f} seconds",
lambda elapsed: f"{pipeline_name} stopped after {elapsed:.1f} seconds",
Expand All @@ -779,10 +779,10 @@ def stop_pipeline(
).deployment_status

if status == "Stopped":
long_op.done()
wait_for_stop.done()
return

long_op.check()
wait_for_stop.check()
time.sleep(0.1)

def dismiss_error_pipeline(
Expand Down Expand Up @@ -823,7 +823,7 @@ def clear_storage(
return

start = time.monotonic()
long_op = LongOperationWarning(
wait_for_clear = LongOperationWarning(
logger,
lambda elapsed: f"still clearing {pipeline_name}, waited {elapsed:.1f} seconds",
lambda elapsed: f"{pipeline_name} storage cleared after {elapsed:.1f} seconds",
Expand All @@ -838,10 +838,10 @@ def clear_storage(
).storage_status

if status == "Cleared":
long_op.done()
wait_for_clear.done()
return

long_op.check()
wait_for_clear.check()
time.sleep(poll_interval_s)

def start_transaction(self, pipeline_name: str) -> int:
Expand Down Expand Up @@ -1037,7 +1037,7 @@ def commit_transaction(
if not wait:
return

long_op = LongOperationWarning(
wait_for_commit = LongOperationWarning(
logger,
lambda elapsed: f"transaction {transaction_id} on {pipeline_name} "
f"hasn't committed, waited {elapsed:.1f} seconds",
Expand All @@ -1052,10 +1052,10 @@ def commit_transaction(

stats = self.get_pipeline_stats(pipeline_name)
if stats["global_metrics"]["transaction_id"] != transaction_id:
long_op.done()
wait_for_commit.done()
return

long_op.check()
wait_for_commit.check()
time.sleep(poll_interval_s)

def checkpoint_pipeline(self, pipeline_name: str) -> int:
Expand Down Expand Up @@ -1268,7 +1268,7 @@ def wait_for_token(
max_backoff = 5
exponent = 1.2
retries = 0
long_op = LongOperationWarning(
wait_for_token_processed = LongOperationWarning(
logger,
lambda elapsed: f"still waiting for inputs represented by {token} "
f"to be processed, waited {elapsed:.1f} seconds",
Expand All @@ -1286,10 +1286,10 @@ def wait_for_token(
)

if self.completion_token_processed(pipeline_name, token):
long_op.done()
wait_for_token_processed.done()
break

long_op.check()
wait_for_token_processed.check()

retries += 1
backoff = min(max_backoff, initial_backoff * (exponent**retries))
Expand Down
24 changes: 24 additions & 0 deletions python/feldera/stats.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,8 +86,22 @@ def from_dict(cls, d: Mapping[str, Any]):
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"])
commit_progress = d.get("commit_progress")
metrics.commit_progress = (
CommitProgressSummary.from_dict(commit_progress)
if commit_progress is not None
else None
)
return metrics

def progress_summary(self) -> str:
"""Human-readable progress indicator for long-running wait loops:
commit progress while a transaction is committing, otherwise the
overall record count processed so far."""
if self.commit_progress is not None:
return str(self.commit_progress)
return f"{self.total_processed_records}/{self.total_input_records} records processed"


class ConnectorError:
"""Represents a connector error item reported by connector status endpoints."""
Expand Down Expand Up @@ -334,6 +348,16 @@ def from_dict(cls, d: Mapping[str, Any]):
status.__dict__.update(d)
return status

def __str__(self) -> str:
# Mirrors CommitProgressSummary's Display impl in

@mihaibudiu mihaibudiu Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this important?

@blp blp Jul 31, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, it's just informative.

# crates/feldera-types/src/transaction.rs.
return (
f"completed: {self.completed} operators, "
f"evaluating: {self.in_progress} operators "
f"[{self.in_progress_processed_records}/{self.in_progress_total_records} "
f"changes processed], remaining: {self.remaining} operators"
)


class TransactionInitiators:
"""Initiators for an ongoing transaction."""
Expand Down
31 changes: 30 additions & 1 deletion python/feldera/testutils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"Utility functions for writing tests against a Feldera instance."

import logging
import math
import os
import platform
Expand All @@ -10,13 +11,16 @@
from typing import List, Optional, cast
from datetime import datetime

from feldera._long_operation_warning import LongOperationWarning
from feldera.enums import CompilationProfile
from feldera.pipeline import Pipeline
from feldera.pipeline_builder import PipelineBuilder
from feldera.runtime_config import Resources, RuntimeConfig
from feldera.rest import FelderaClient
from feldera.rest._helpers import requests_verify_from_env

logger = logging.getLogger(__name__)

API_KEY = os.environ.get("FELDERA_API_KEY")


Expand Down Expand Up @@ -359,9 +363,34 @@ def check_end_of_input(pipeline: Pipeline) -> bool:

def wait_end_of_input(pipeline: Pipeline, timeout_s: Optional[int] = None):
start_time = time.monotonic()
while not check_end_of_input(pipeline):
# Reused by the warning message below so a stalled ingest can be told
# apart from a slow one; check_end_of_input() would fetch and discard it.
latest_stats = None
wait_for_input_end = LongOperationWarning(
logger,
lambda elapsed: (
f"still waiting for end of input on pipeline {pipeline.name}, "
f"waited {elapsed:.1f} seconds ({latest_stats.global_metrics.progress_summary()})"
),
lambda elapsed: (
f"end of input reached on pipeline {pipeline.name} "
f"after {elapsed:.1f} seconds"
),
)

while True:
latest_stats = pipeline.stats()
if all(
input_endpoint.metrics.end_of_input
for input_endpoint in latest_stats.inputs
):
wait_for_input_end.done()
return

if timeout_s is not None and time.monotonic() - start_time > timeout_s:
raise TimeoutError("Timeout waiting for end of input")

wait_for_input_end.check()
time.sleep(3)


Expand Down
74 changes: 74 additions & 0 deletions python/tests/unit/test_stats.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
"""Tests for GlobalPipelineMetrics.progress_summary(), which lets wait loops
(Pipeline.wait_for_completion, testutils.wait_end_of_input) report whether a
long wait is stalled or just slow."""

from feldera.stats import CommitProgressSummary, GlobalPipelineMetrics

BASE_METRICS_DICT = {
"state": "running",
"incarnation_uuid": "00000000-0000-0000-0000-000000000000",
"start_time": 0,
"transaction_status": "NoTransaction",
"total_processed_records": 42,
"total_input_records": 100,
}


def test_commit_progress_absent_when_no_transaction():
metrics = GlobalPipelineMetrics.from_dict(BASE_METRICS_DICT)
assert metrics.commit_progress is None


def test_commit_progress_parsed_into_object():
d = {
**BASE_METRICS_DICT,
"commit_progress": {
"completed": 3,
"in_progress": 2,
"remaining": 1,
"in_progress_processed_records": 10,
"in_progress_total_records": 50,
},
}
metrics = GlobalPipelineMetrics.from_dict(d)
assert isinstance(metrics.commit_progress, CommitProgressSummary)
assert metrics.commit_progress.completed == 3
assert metrics.commit_progress.in_progress_total_records == 50


def test_commit_progress_str_matches_rust_display_format():
progress = CommitProgressSummary.from_dict(
{
"completed": 3,
"in_progress": 2,
"remaining": 1,
"in_progress_processed_records": 10,
"in_progress_total_records": 50,
}
)
assert str(progress) == (
"completed: 3 operators, evaluating: 2 operators "
"[10/50 changes processed], remaining: 1 operators"
)


def test_progress_summary_falls_back_to_record_counts_outside_transaction():
metrics = GlobalPipelineMetrics.from_dict(BASE_METRICS_DICT)
assert metrics.progress_summary() == "42/100 records processed"


def test_progress_summary_prefers_commit_progress_during_transaction():
d = {
**BASE_METRICS_DICT,
"commit_progress": {
"completed": 0,
"in_progress": 1,
"remaining": 7,
"in_progress_processed_records": 0,
"in_progress_total_records": 169,
},
}
metrics = GlobalPipelineMetrics.from_dict(d)
summary = metrics.progress_summary()
assert "0/169 changes processed" in summary
assert "records processed" not in summary
Loading