[dagster-pipes] Add key param to PipesS3MessageReader

[danielgafni]

Summary & Motivation

This PR adds key parameter to PipesS3MessageReader.

Previously, it could only be used when paired with PipesS3MessageWriter on the external process side.

This change allows using PipesS3MessageReader in situations where we expect an external service to create an S3 object containing Pipes messages (typically by dumping stdout/stderr logs).

A new expect_s3_message_writer parameter is introduced to distinguish between these two operational modes (defaults to True).

A new key parameter can either be passed to the constructor or set later on via PipesSession.report_launched hook (if not known during MessageReader instantiation).

How I Tested These Changes

Added a unit test for the new functionality.

Changelog

Insert changelog entry or "NOCHANGELOG" here.

• NEW (added new feature or capability)

[danielgafni]

• [dagster-pipes] Add key param to PipesS3MessageReader #24884 👈
• [dagster-aws, dagster-pipes] Add PipesS3LogReader #25224 : 1 other dependent PR (#23998 )
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @danielgafni and the rest of your teammates on Graphite

[danielgafni]

I suspect it would be better to just have a separate class

I think we should spend some time discussing these options.

---

This will be a common pattern with all PipesMessageReaders. It should always be possible to create the message passing channel on Dagster's side (and require a matching PipesMessageWriter on the remote process side).

Let's consider our file-based message readers:

• PipesFileMessageReader
• PipesTempFileMessageReader

We have 2 classes for these 2 options. I my opinion, that's fine.

However, we do not follow the same naming/functionality mapping with other message readers. Consider the corresponding S3 message readers:

• PipesFileMessageReader (reads from any file) -> ?
• PipesTempFileMessageReader (reads from a file created by Dagster) -> PipesS3MessageReader.

The current PipesS3MessageReader actually acts as PipesTemp3MessageReader, because it's reading from objects created by Pipes.

That's why I didn't add a separate class - the logical decision would be to rename the existing PipesS3MessageReader to PipesTempS3MessageReader (or another name strongly indicating that it isn't supposed to be reading just any S3 objects) and add the new class as just PipesS3MessageReader, but this would mean breaking changes in Pipes.

So, I decided to make a parameter to switch between these 2 options instead.

---

We should have this naming/functionality consistent across different MessageReaders. If we decide to have 2 classes, we should have 2 classes everywhere (but this means breaking changes). If we decide to have an argument we should also have it everywhere (and it will be less confusing since it will be all over our docs and impossible to miss).

[smackesey]

We should have this naming/functionality consistent across different MessageReaders. If we decide to have 2 classes, we should have 2 classes everywhere (but this means breaking changes). If we decide to have an argument we should also have it everywhere (and it will be less confusing since it will be all over our docs and impossible to miss).

I agree and think you've hit on a rough area in the design of pipes. We should discuss with @schrockn. AFAICT the EMR client (the last one in the stack we aimed to merge this week) need not depend on this functionality, so please move that off of this one so we can merge it independently, then we can discuss this question of message readers dependence on a particular message writer next week.

[danielgafni]

@schrockn I believe this topic above needs your attention