feat(sentry_integration): exec app in subprocess #217

victoria-yining-huang · 2025-12-16T15:58:54Z

Summary

Refactor pipeline loading to use subprocess isolation for improved error handling and monitoring
Integrate Sentry SDK for streaming platform observability
Add test coverage for subprocess communication and error scenarios

Changes

Subprocess-based Pipeline Loading (sentry_streams/runner.py)

Introduced _load_pipeline_in_process() to execute customer pipeline code in an isolated subprocess using multiprocessing.Pool
Isolates customer code execution from the main runtime process
Pipeline object is returned directly from subprocess or propagates exceptions naturally

note:

this sentry integration support is not added to rust entry point

sentry_streams/sentry_streams/runner.py

sentry_streams/sentry_streams/test_sdk.py

sentry_streams/sentry_streams/pipeline_loader.py

sentry_streams/sentry_streams/runner.py

sentry_streams/sentry_streams/pipeline_loader.py

sentry_streams/sentry_streams/runner.py

untitaker · 2026-01-08T15:37:16Z

also it would be nice to have some sort of test for this.

you can test using custom transports in the SDK

sentry_streams/sentry_streams/runner.py

sentry_streams/sentry_streams/pipeline_loader.py

sentry_streams/sentry_streams/runner.py

fpacifici

The current approach means if no customer sentry configured, pickling errors and pipeline build errors will silently fail. Nobody gets alerted. Should we mandate every customer app must have a Sentry?

No, we build a platform, we would not mandate how the clients manage their errors.
Taking a step back I think we should be notified even when the application code fails when starting up, it is still useful for us.
I think we should get back those errors and send them to our integration as well. We can tag them or lower the severity so we can filter them out, but we should have visibility on them no matter how the customer set up their integrations.

We do not want product to get infra errors in their integration, but I think we should at least have visibility on product errors whether or not the product sets up sentry

sentry_streams/sentry_streams/runner.py

fpacifici · 2026-01-09T23:07:37Z

sentry_streams/sentry_streams/runner.py

    assigned_segment_id = int(segment_id) if segment_id else None
-    pipeline: Pipeline[Any] = pipeline_globals["pipeline"]
    runtime: Any = load_adapter(adapter, environment_config, assigned_segment_id, metric_config)
    translator = RuntimeTranslator(runtime)


I think there is a use case we did not consider and is an issue.
The pipeline definition is coupled with the configuration file. If I rename a step in the pipeline definition and the config file references that, the config may become invalid. Though we would know that only at this step. That error would go to streaming.

Though if we changed that and sent config file mistakes to product, the system would be wrong anyway, as streaming owns the config file content.

Do we have a plan to address this issue ?

I did not address this issue yet. The configuration yaml file should have its own validation step, and it's not been built yet. In any case like you said it will be a problem if this step is the first place an invalid deployment config file is erroring out

@fpacifici not sure what the course of action here is though, a mismatch between deployment config and pipeline settings could be either team's responsibility. if a team renames its steps and does not update the config file in advance then we should not have to react to that.

I think we should probably try to prevent this issue in CI if it becomes a big one.

I think we should probably try to prevent this issue in CI if it becomes a big one.

I think we do not haev a precise story for how to manage this use case in a CI/CD environment.
IF I change the application, the change may be breaking with respect to what the config says. The way this works today would require us to switch the configuraiton at the same time the new sha is deployed, which is not something we can really enforce now.

I think this should be discussed in a followup

https://linear.app/getsentry/issue/STREAM-680/clarify-the-process-to-deploy-changes-to-the-application-without

sentry_streams/sentry_streams/runner.py

victoria-yining-huang · 2026-01-12T22:50:04Z

@fpacifici added new feature:

when subprocess fails, main process sentry will receive a generic CalledProcessError exception. The subprocess' sentry (if configured) will receive specific Python Exceptions.

sentry_streams/sentry_streams/runner.py

victoria-yining-huang · 2026-01-12T23:14:11Z

@untitaker added tests using transports

sentry_streams/sentry_streams/runner.py

cursor

Cursor Bugbot has reviewed your changes and found 8 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

sentry_streams/sentry_streams/runner.py

sentry_streams/pyproject.toml

sentry_streams/tests/test_sentry_transports.py

github-actions · 2026-01-23T21:12:23Z

Semver Impact of This PR

🟡 Minor (new features)

📋 Changelog Preview

This is how your changes will appear in the changelog.
Entries from this PR are highlighted with a left border (blockquote style).

New Features ✨

(sentry_integration) Exec app in subprocess by victoria-yining-huang in #217

Other

Move gcssink config override out of the adapter by fpacifici in #228

_{🤖 This preview updates automatically when you update the PR.}

sentry_streams/sentry_streams/runner.py

cursor

Cursor Bugbot has reviewed your changes and found 3 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

sentry_streams/tests/test_sentry_transports.py

sentry_streams/sentry_streams/runner.py

cursor

Cursor Bugbot has reviewed your changes and found 3 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

sentry_streams/sentry_streams/runner.py

sentry_streams/uv.lock

'

cursor

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

sentry_streams/sentry_streams/runner.py

cursor

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

sentry_streams/sentry_streams/runner.py

cursor

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

cursor · 2026-01-26T23:17:02Z

sentry_streams/sentry_streams/runner.py

-    config: str,
    segment_id: Optional[str],
    application: str,
+    environment_config: Mapping[str, Any],


Rust caller uses incompatible old function signature

High Severity

The load_runtime function signature changed: the config parameter was removed, environment_config: Mapping[str, Any] was added, and parameter order shifted. However, the Rust code in run.rs still calls this function with the old positional arguments (name, log_level, adapter_name, config_file, segment_id, application_name). This causes a type mismatch where config_file (a string path) gets passed where segment_id is expected, and application_name (a string) gets passed where environment_config (a mapping) is expected, breaking the Rust integration entirely.

cursor · 2026-01-26T23:17:02Z

sentry_streams/sentry_streams/runner.py

+    # Note: Customer print() and logging statements (redirected to stderr)
+    # do not trigger platform Sentry alerts.
+    with multiprocessing.Pool(processes=1) as pool:
+        pipeline: Pipeline[Any] = pool.apply(_load_pipeline_in_process, (application,))


Subprocess pickle breaks pipelines containing lambdas

Medium Severity

The new multiprocessing.Pool.apply() approach requires the Pipeline object to be picklable when returning from the subprocess. However, Pipeline objects commonly contain Step instances with lambda functions (e.g., Map(function=lambda msg: ...)), and Python lambdas are not picklable. Existing examples like billing.py with aggregate_func=lambda: OutcomesBuffer() and tests using function=lambda msg: ... will fail with PicklingError at runtime.

cursor

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

cursor · 2026-01-26T23:35:21Z

sentry_streams/sentry_streams/runner.py

-    config: str,
    segment_id: Optional[str],
    application: str,
+    environment_config: Mapping[str, Any],


Rust caller uses outdated load_runtime function signature

High Severity

The load_runtime function signature changed from (name, log_level, adapter, config, segment_id, application) to (name, log_level, adapter, segment_id, application, environment_config), but the Rust caller in run.rs was not updated. The Rust code passes config_file where segment_id is now expected, segment_id where application is expected, and application_name (a string) where environment_config (a Mapping) is expected. This will crash when environment_config.get("metrics", {}) is called because strings don't have a .get() method.

cursor

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

cursor · 2026-01-27T00:19:28Z

sentry_streams/src/run.rs

-                runtime_config.config_file,
                runtime_config.segment_id,
                runtime_config.application_name,
+                environment_config,


Config validation skipped when using Rust runner

Medium Severity

The jsonschema.validate() call for config validation was moved from load_runtime() to main(). However, the Rust runner (run.rs) calls load_runtime() directly, bypassing main(). This means config files are no longer validated against the JSON schema when using the Rust entry point, which is a regression from the previous behavior where validation happened inside load_runtime() regardless of entry point.

Additional Locations (1)

sentry_streams/sentry_streams/runner.py#L195-L205

cursor · 2026-01-27T00:19:29Z

sentry_streams/src/run.rs

-                runtime_config.config_file,
                runtime_config.segment_id,
                runtime_config.application_name,
+                environment_config,


Sentry SDK not initialized when using Rust runner

Medium Severity

The Sentry SDK initialization code for platform observability was added only to the Python CLI's main() function. The Rust runner (run.rs) calls load_runtime() directly, bypassing main(), so sentry_sdk.init() is never called. This means platform errors occurring when using the Rust entry point won't be reported to Sentry, defeating the purpose of the Sentry integration feature described in the PR.

Additional Locations (1)

sentry_streams/sentry_streams/runner.py#L206-L212

fpacifici

Please see the comments inline

fpacifici · 2026-01-27T20:48:32Z

sentry_streams/src/run.rs

+        // Read and parse the config file as YAML to create environment_config
+        let yaml_module = py.import("yaml")?;
+        let config_path = runtime_config
+            .config_file
+            .to_str()
+            .ok_or_else(|| pyo3::exceptions::PyValueError::new_err("Invalid config file path"))?;
+
+        let config_file = std::fs::File::open(config_path).map_err(|e| {
+            pyo3::exceptions::PyIOError::new_err(format!("Failed to open config file: {}", e))
+        })?;
+        let config_reader = std::io::BufReader::new(config_file);
+        let config_str = std::io::read_to_string(config_reader).map_err(|e| {
+            pyo3::exceptions::PyIOError::new_err(format!("Failed to read config file: {}", e))
+        })?;
+
+        let environment_config = yaml_module.getattr("safe_load")?.call1((config_str,))?;
+


Please let's not duplicate this logic.
Every time we duplicate the logic between python and rust we make it harder to make changes as two places have to be updated.

If you are making this change because now the load_runtime method takes the config structure instead of the config file name. Please have two methods:
one takes the file name, reads it instantiates the sdk and then calls the other method passing the parsed config.
This rust function will call the first method.

More importantly, how do we initialize the platform sdk when we start the runner with this cli?

fpacifici · 2026-01-27T20:50:21Z

sentry_streams/sentry_streams/runner.py

                    step_streams[branch_name] = next_step_stream[branch_name]


+def _load_pipeline_in_process(application: str) -> Pipeline[Any]:


nit: this method does not know it runs in a separate process. It can work without the separate process. Just call it _load_pipeline. Then you can provide the rationale in the docstring

fpacifici · 2026-01-27T20:52:09Z

sentry_streams/sentry_streams/config.json

+                },
+                "streaming_platform_dsn": {
+                    "type": "string"


The DSN is not enough. At times we need to pass parameters to the sentry integraiotn.
Please make this an object and one of the fields will be the DSN. For now you do not have to add additional fields but we will certainly do. Turning a string into an object will be tricky as it will break existing configs. Adding a field to an object will be trivial.

fpacifici · 2026-01-27T20:53:01Z

sentry_streams/sentry_streams/runner.py

+    if streaming_platform_dsn:
+        sentry_sdk.init(
+            dsn=streaming_platform_dsn,
+            send_default_pii=True,


Why is this True? Usually we should not send pii.

fpacifici · 2026-01-27T21:01:13Z

sentry_streams/tests/test_sentry_transports.py

I am not sure I get what you are trying to validate with this test.

The test says test_sentry_transport, but there is no assertion whether the mock Transport provided is ever used.

Having a test for the load_runtime method is a good idea, we do not seem to have one, but, then I would assert the properties of the returned runtime using the Dummy runtime rather than mocking the graph construction methods which are the logic you are testing.

fpacifici · 2026-01-27T21:02:51Z

sentry_streams/tests/test_sentry_transports.py

+class CaptureTransport(Transport):
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.events: List[Any] = []
+        self.envelopes: List[Any] = []
+
+    def capture_event(self, event: Any) -> None:
+        self.events.append(event)
+        return None
+
+    def capture_envelope(self, envelope: Any) -> None:
+        self.envelopes.append(envelope)
+        return None
+
+    def flush(self, timeout: float, callback: Optional[Any] = None) -> None:
+        """Flush is called when SDK shuts down."""
+        pass
+
+
+@pytest.fixture
+def temp_fixture_dir(tmp_path: Any) -> Any:
+    fixture_dir = tmp_path / "fixtures"
+    fixture_dir.mkdir()
+    return fixture_dir
+
+
+@pytest.fixture(autouse=True)
+def reset_metrics_backend() -> Generator[None, None, None]:


These do not seem to be relevant for the test. No assertion is ran on any instance of these.
If you want to test load_runtime, then you can remove these and just test that load_Runtime returns the right runtime when the pipeline is good and that it returns an exception when it is not.

If you want to test the two separate sdk initializations then you will have to check that the Transport actually catches events when load fails.

fpacifici · 2026-01-27T21:03:14Z

sentry_streams/tests/test_sentry_transports.py

+    with (
+        patch("sentry_streams.runner.load_adapter") as mock_load_adapter,
+        patch("sentry_streams.runner.iterate_edges") as mock_iterate_edges,
+    ):
+        mock_runtime = type(
+            "MockRuntime",
+            (),
+            {
+                "run": lambda self: None,
+                "source": lambda self, step: "mock_stream",
+                "complex_step_override": lambda self: {},
+            },
+        )()
+        mock_load_adapter.return_value = mock_runtime


Please use the dummyAdapter rather than mocking everything

fpacifici · 2026-01-27T21:03:51Z

sentry_streams/tests/test_sentry_transports.py

+    app_file.write_text(
+        """
+from sentry_streams.pipeline import streaming_source
+pipeline = streaming_source(name="test", stream_name="test-stream")
+"""


Please create a module containing the pipeline you want to test, do not generate a python file on the fly. There is no gain in doing so while there is the downside that ype checking will nto understand this string.

victoria-yining-huang changed the title ~~i think i added python sentry~~ demo(sentry_integration): alerts in two different sentry projects Dec 16, 2025

victoria-yining-huang changed the title ~~demo(sentry_integration): alerts in two different sentry projects~~ wip(sentry_integration): exec app in subprocess Dec 17, 2025

victoria-yining-huang force-pushed the vic/subprocess_to_build_pipeline branch from d575941 to b0e6a25 Compare January 7, 2026 21:17

victoria-yining-huang marked this pull request as ready for review January 7, 2026 23:09

victoria-yining-huang requested a review from a team as a code owner January 7, 2026 23:09

victoria-yining-huang changed the title ~~wip(sentry_integration): exec app in subprocess~~ feat(sentry_integration): exec app in subprocess Jan 7, 2026

cursor bot reviewed Jan 7, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/test_sdk.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/pipeline_loader.py Outdated Show resolved Hide resolved

sentry bot reviewed Jan 7, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

untitaker reviewed Jan 8, 2026

View reviewed changes

sentry bot reviewed Jan 9, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

cursor bot reviewed Jan 9, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

cursor bot reviewed Jan 9, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/pipeline_loader.py Outdated Show resolved Hide resolved

cursor bot reviewed Jan 9, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

fpacifici reviewed Jan 9, 2026

View reviewed changes

sentry bot reviewed Jan 12, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

cursor bot reviewed Jan 12, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

This comment was marked as outdated.

Sign in to view

sentry bot reviewed Jan 22, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

This comment was marked as outdated.

Sign in to view

cursor bot reviewed Jan 22, 2026

View reviewed changes

sentry bot reviewed Jan 23, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Show resolved Hide resolved

cursor bot reviewed Jan 23, 2026

View reviewed changes

sentry_streams/tests/test_sentry_transports.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry bot reviewed Jan 23, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

cursor bot reviewed Jan 23, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry_streams/uv.lock Outdated Show resolved Hide resolved

victoria-yining-huang added 19 commits January 26, 2026 13:21

lol did not add tests now added

490fada

add test on customer print

f8a20a5

rename file

e2fed79

make dsn come from yaml

6133766

'

only read yaml once

9a64a32

use multiprocessing module

024c09e

remove module

a346fea

remove redundant raises

97c9720

remove third party multiprocessing

6b8736e

all tests verified and pass

29b794e

just one conn'

9e65bc6

remove lib

7823864

reset metrics backend in tests

8810710

that child close was not needed

b68fb44

typing annotations

1476cc8

remove daed code

071c87e

add more detailed tests

06f7e3a

refine error paths

5a8c1ec

fix return flow

8466979

victoria-yining-huang force-pushed the vic/subprocess_to_build_pipeline branch from 113a6e0 to 8466979 Compare January 26, 2026 18:21

cursor bot reviewed Jan 26, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/runner.py Show resolved Hide resolved

remove code by mistake

7e20803

cursor bot reviewed Jan 26, 2026

View reviewed changes

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

sentry_streams/sentry_streams/runner.py Outdated Show resolved Hide resolved

use multiprocessing pool

280fd9b

cursor bot reviewed Jan 26, 2026

View reviewed changes

typecheck

98c86f1

cursor bot reviewed Jan 26, 2026

View reviewed changes

match rust func signatures

c440385

cursor bot reviewed Jan 27, 2026

View reviewed changes

fpacifici requested changes Jan 27, 2026

View reviewed changes

		step_streams[branch_name] = next_step_stream[branch_name]


		def _load_pipeline_in_process(application: str) -> Pipeline[Any]:

Uh oh!

feat(sentry_integration): exec app in subprocess #217

Are you sure you want to change the base?

feat(sentry_integration): exec app in subprocess #217

Conversation

victoria-yining-huang commented Dec 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

untitaker commented Jan 8, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

fpacifici left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

victoria-yining-huang commented Jan 12, 2026

Uh oh!

Uh oh!

Uh oh!

victoria-yining-huang commented Jan 12, 2026

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

cursor bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Jan 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Semver Impact of This PR

New Features ✨

Other

Uh oh!

Uh oh!

cursor bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

victoria-yining-huang commented Dec 16, 2025 •

edited

Loading

fpacifici left a comment •

edited

Loading

github-actions bot commented Jan 23, 2026 •

edited

Loading

Rust caller uses outdated `load_runtime` function signature