diff --git a/python/changelog.d/+ingestion-backups-opt-in.breaking.md b/python/changelog.d/+ingestion-backups-opt-in.breaking.md new file mode 100644 index 0000000000..35646a0a78 --- /dev/null +++ b/python/changelog.d/+ingestion-backups-opt-in.breaking.md @@ -0,0 +1 @@ +Ingestion streaming clients now default to `StreamingMode.LIVE_ONLY`; pass `streaming_mode=StreamingMode.LIVE_WITH_BACKUPS` to enable local telemetry backups. diff --git a/python/docs/examples/ingestion.ipynb b/python/docs/examples/ingestion.ipynb index e746322b4d..c2167e4d22 100644 --- a/python/docs/examples/ingestion.ipynb +++ b/python/docs/examples/ingestion.ipynb @@ -17,6 +17,12 @@ "pip install sift-stack-py[sift-stream]\n", "```\n", "\n", + "## Streaming backups\n", + "\n", + "Both streaming client factories default to `StreamingMode.LIVE_ONLY`: telemetry streams to Sift without local disk backups.\n", + "\n", + "To enable local backups and replay after delivery failures, import `StreamingMode` from `sift_client.resources.ingestion` and pass `streaming_mode=StreamingMode.LIVE_WITH_BACKUPS` to `create_ingestion_config_streaming_client` or `create_auto_register_streaming_client`.\n", + "\n", "## Topics Covered\n", "\n", "1. **Basic Example**: Simple single flow sending using FlowConfig\n", diff --git a/python/examples/ingestion-tutorial/stream.py b/python/examples/ingestion-tutorial/stream.py index f49b4ced4f..2699ae3032 100644 --- a/python/examples/ingestion-tutorial/stream.py +++ b/python/examples/ingestion-tutorial/stream.py @@ -170,6 +170,10 @@ async def main() -> None: # All telemetry sent within this context will appear inside # this Run in Sift. # + # Streaming defaults to LIVE_ONLY (no local telemetry backups). + # To opt in, import StreamingMode from sift_client.resources.ingestion + # and pass streaming_mode=StreamingMode.LIVE_WITH_BACKUPS below. + # # WARNING: SiftStream should be reused for the entire program duration. # Do NOT recreate the streaming client per batch or per iteration — # each instantiation creates a new ingestion config, which introduces diff --git a/python/lib/sift_client/_internal/low_level_wrappers/ingestion.py b/python/lib/sift_client/_internal/low_level_wrappers/ingestion.py index 8cc0e57d7b..ac3fefb76c 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/ingestion.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/ingestion.py @@ -278,7 +278,7 @@ async def _build_sift_stream_instance( _default_logfile_prefix = "sift_stream_bindings.log" if streaming_mode is ...: # type: ignore[comparison-overlap] - streaming_mode = StreamingMode.LIVE_WITH_BACKUPS + streaming_mode = StreamingMode.LIVE_ONLY # None disables metrics streaming, so it passes through to the builder as-is. Zero is # rejected here because SiftStream cannot schedule metrics on a zero interval. @@ -328,7 +328,7 @@ async def _build_sift_stream_instance( fb_builder.disk_backup_policy = disk_backup_policy return await fb_builder.build() - else: # LIVE_WITH_BACKUPS (default) + else: # LIVE_WITH_BACKUPS lwb_builder = config_builder.live_with_backups() lwb_builder.metrics_streaming_interval = metrics_interval if retry_policy is not None: diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_ingestion_stream_builder.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_ingestion_stream_builder.py index 4ab0cf7002..9d59d0d3c8 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_ingestion_stream_builder.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_ingestion_stream_builder.py @@ -41,6 +41,36 @@ async def _build(**kwargs): ) +class TestStreamingMode: + @pytest.mark.asyncio + async def test_defaults_to_live_only(self, config_builder): + await _build() + + config_builder.live_only.return_value.build.assert_awaited_once() + config_builder.live_with_backups.assert_not_called() + config_builder.file_backup.assert_not_called() + + @pytest.mark.asyncio + async def test_backups_are_explicit_opt_in(self, config_builder): + backup_policy = sift_stream_bindings.DiskBackupPolicyPy.default() + retry_policy = sift_stream_bindings.RetryPolicyPy.default() + + await _build( + streaming_mode=StreamingMode.LIVE_WITH_BACKUPS, + disk_backup_policy=backup_policy, + retry_policy=retry_policy, + checkpoint_interval_seconds=30, + ) + + mode_builder = config_builder.live_with_backups.return_value + mode_builder.build.assert_awaited_once() + assert mode_builder.disk_backup_policy is backup_policy + assert mode_builder.retry_policy is retry_policy + assert mode_builder.checkpoint_interval.to_timedelta() == timedelta(seconds=30) + config_builder.live_only.assert_not_called() + config_builder.file_backup.assert_not_called() + + @pytest.mark.parametrize("streaming_mode", list(MODE_BUILDER_METHODS)) class TestMetricsStreamingInterval: @pytest.mark.asyncio diff --git a/python/lib/sift_client/_tests/resources/test_ingestion.py b/python/lib/sift_client/_tests/resources/test_ingestion.py index 192acbc9ef..7d1318aadf 100644 --- a/python/lib/sift_client/_tests/resources/test_ingestion.py +++ b/python/lib/sift_client/_tests/resources/test_ingestion.py @@ -1,4 +1,4 @@ -"""Tests that rate-limit errors from SiftStream reach callers of the streaming clients unchanged.""" +"""Tests for streaming client configuration and rate-limit error propagation.""" from datetime import timedelta from unittest.mock import AsyncMock, MagicMock @@ -7,12 +7,45 @@ from sift_client.resources.ingestion import ( AutoRegisterStreamingClient, + IngestionAPIAsync, IngestionConfigStreamingClient, + StreamingMode, ) sift_stream_bindings = pytest.importorskip("sift_stream_bindings") +@pytest.mark.asyncio +@pytest.mark.parametrize( + "factory_name", + ["create_ingestion_config_streaming_client", "create_auto_register_streaming_client"], +) +@pytest.mark.parametrize("streaming_mode", [None, StreamingMode.LIVE_WITH_BACKUPS]) +async def test_streaming_mode_default_and_opt_in( + mock_client, monkeypatch, factory_name, streaming_mode +): + build_stream = AsyncMock() + monkeypatch.setattr( + "sift_client._internal.low_level_wrappers.ingestion._build_sift_stream_instance", + build_stream, + ) + monkeypatch.setattr("sift_client.resources.ingestion._build_sift_stream_instance", build_stream) + monkeypatch.setattr( + sift_stream_bindings, + "SiftStreamAutoRegisterPy", + MagicMock(from_stream=AsyncMock()), + ) + api = IngestionAPIAsync(mock_client) + kwargs = {} if streaming_mode is None else {"streaming_mode": streaming_mode} + + await getattr(api, factory_name)(ingestion_config=MagicMock(), **kwargs) + + build_stream.assert_awaited_once() + assert build_stream.call_args.kwargs["streaming_mode"] == ( + StreamingMode.LIVE_ONLY if streaming_mode is None else streaming_mode + ) + + def _rate_limited_error(): return sift_stream_bindings.RateLimitedErrorPy( "[RateLimitedError] rate limited", diff --git a/python/lib/sift_client/resources/ingestion.py b/python/lib/sift_client/resources/ingestion.py index 462228451e..06680062c9 100644 --- a/python/lib/sift_client/resources/ingestion.py +++ b/python/lib/sift_client/resources/ingestion.py @@ -159,7 +159,7 @@ async def create_ingestion_config_streaming_client( run: RunCreate | dict | str | Run | None = None, asset_tags: list[str] | list[Tag] | None = None, asset_metadata: dict[str, str | float | bool] | None = None, - streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS, + streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY, retry_policy: RetryPolicyPy | None = None, disk_backup_policy: DiskBackupPolicyPy | None = None, checkpoint_interval_seconds: int | None = None, @@ -175,7 +175,8 @@ async def create_ingestion_config_streaming_client( run: The run to associate with ingestion. Can be a Run, RunCreate, dict, or run ID string. asset_tags: Tags to associate with the asset. asset_metadata: Metadata to associate with the asset. - streaming_mode: Transport mode for the stream. Defaults to LIVE_WITH_BACKUPS. + streaming_mode: Transport mode for the stream. Defaults to LIVE_ONLY, without disk + backups. Set to LIVE_WITH_BACKUPS to opt in to local disk backups. retry_policy: Retry policy for LIVE_WITH_BACKUPS mode. disk_backup_policy: Disk backup policy for LIVE_WITH_BACKUPS or FILE_BACKUP mode. checkpoint_interval_seconds: Checkpoint interval in seconds (LIVE_WITH_BACKUPS only). @@ -223,7 +224,7 @@ async def create_auto_register_streaming_client( run: RunCreate | dict | str | Run | None = None, asset_tags: list[str] | list[Tag] | None = None, asset_metadata: dict[str, str | float | bool] | None = None, - streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS, + streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY, retry_policy: RetryPolicyPy | None = None, disk_backup_policy: DiskBackupPolicyPy | None = None, checkpoint_interval_seconds: int | None = None, @@ -256,7 +257,8 @@ async def create_auto_register_streaming_client( run: The run to associate with ingestion. Can be a Run, RunCreate, dict, or run ID string. asset_tags: Tags to associate with the asset. asset_metadata: Metadata to associate with the asset. - streaming_mode: Transport mode for the stream. Defaults to LIVE_WITH_BACKUPS. + streaming_mode: Transport mode for the stream. Defaults to LIVE_ONLY, without disk + backups. Set to LIVE_WITH_BACKUPS to opt in to local disk backups. retry_policy: Retry policy for LIVE_WITH_BACKUPS mode. disk_backup_policy: Disk backup policy for LIVE_WITH_BACKUPS or FILE_BACKUP mode. checkpoint_interval_seconds: Checkpoint interval in seconds (LIVE_WITH_BACKUPS only). @@ -327,7 +329,7 @@ async def _create( run: RunCreate | dict | str | Run | RunFormPy | None = None, asset_tags: list[str] | list[Tag] | None = None, asset_metadata: dict[str, str | float | bool] | None = None, - streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS, + streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY, retry_policy: RetryPolicyPy | None = None, disk_backup_policy: DiskBackupPolicyPy | None = None, checkpoint_interval_seconds: int | None = None, @@ -344,7 +346,8 @@ async def _create( run: The run to associate with ingestion. Can be a Run, RunCreate, dict, or run ID string. asset_tags: Tags to associate with the asset. asset_metadata: Metadata to associate with the asset. - streaming_mode: Transport mode for the stream. Defaults to LIVE_WITH_BACKUPS. + streaming_mode: Transport mode for the stream. Defaults to LIVE_ONLY, without disk + backups. Set to LIVE_WITH_BACKUPS to opt in to local disk backups. retry_policy: Retry policy for LIVE_WITH_BACKUPS mode. disk_backup_policy: Disk backup policy for LIVE_WITH_BACKUPS or FILE_BACKUP mode. checkpoint_interval_seconds: Checkpoint interval in seconds (LIVE_WITH_BACKUPS only). @@ -686,7 +689,7 @@ async def _create( run: RunCreate | dict | str | Run | RunFormPy | None = None, asset_tags: list[str] | list[Tag] | None = None, asset_metadata: dict[str, str | float | bool] | None = None, - streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS, + streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY, retry_policy: RetryPolicyPy | None = None, disk_backup_policy: DiskBackupPolicyPy | None = None, checkpoint_interval_seconds: int | None = None,