[dagster-aws] add Pipes cloudwatch message reader

[danielgafni]

Summary & Motivation

resolve #23056

Adds PipesCloudWatchMessageReader which can be used with different AWS services.

It's a solid default MessageReader since various AWS services already emit CloudWatch logs by default.

Right now it's reading the full CloudWatch log stream provided by the user and routing it to Dagster's stdout. That's the minimal functionality needed by Glue Pipes.

In the future, we can think about changing this behavior, mainly:

• optionally, filtering non-dagster logs in the CloudWatch query
• optionally, route logs to stderr

It's work mentioning that Glue routes both stdout and stderr to /output log group. Thus, we only need to read a single stream (with the job run id) from this group in order to receive all necessary information.

How I Tested These Changes

• [x]: add moto tests with emulating CloudWatch logs

[danielgafni]

• [dagster-aws] add Pipes cloudwatch message reader #23353 : 3 dependent PRs (#23354 , #23440 , #23441 ) 👈
• master

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

Join @danielgafni and the rest of your teammates on Graphite

[github-actions]

Deploy preview for dagster-docs ready!

Preview available at https://dagster-docs-pumhsafkc-elementl.vercel.app
https://pipes-cloudwatch-message-reader.dagster.dagster-docs.io

Direct link to changed pages:

• https://dagster-docs-pumhsafkc-elementl.vercel.app
  https://pipes-cloudwatch-message-reader.dagster.dagster-docs.io/concepts/dagster-pipes/aws-glue

[schrockn]

This looks great overall. Needs more tests and a more flesh-out test plan.

Can you also provide a more thorough "summary and motivation" section about why this and the tradeoffs involved. If memory serves there are a real cost tradeoffs here and it would be good to understand that and document them.

[danielgafni]

Side note: I discovered that moto + S3 tests fail if no credentials are configured. This is surprising a for a mocking software. Took me a while to figure this out on a new clean VM...

Found this issue here getmoto/moto#1941) describing similar situations.

Apparently, it's necessary to have at least some credentials configured.

Should we provide testing credentails for all our pytest tests automatically?

[github-actions]

Deploy preview for dagit-core-storybook ready!

✅ Preview
https://dagit-core-storybook-8ld7qz9sy-elementl.vercel.app
https://pipes-cloudwatch-message-reader.core-storybook.dagster-docs.io

Built with commit 2e31171.
This pull request is being automatically deployed with vercel-action

[danielgafni]

@schrockn hey, please take a look at moto tests.

I had to manually create CloudWatch logs (since moto doesn't actually run the Glue job and thus doesnt create any logs) in the fake Glue client.

I'm planning to move examples/docs changes into a separate PR tomorrow.

[schrockn]

I have some minor inline comments.

My big question here is what will appear in Dagster UI? We are now getting all the cloudwatch logs. Are they all going to show up in stdout? Where does stderr emitted from a glue job end up?

Paint me a picture of the end-to-end logging experience in Dagster in this case.

[danielgafni]

@schrockn As I suspected, AWS Glue routes both stdout and stderr to /output CloudWatch log group. The /error sterams in CloudWatch only contain Spark logs.

The following code:

with open_dagster_pipes(
        params_loader=PipesCliArgsParamsLoader(),
        context_loader=PipesS3ContextLoader(client=boto3.client("s3"))
    ) as pipes:
    pipes.log.info("Hello from external process!")
    pipes.report_asset_materialization(
            metadata={
                "some_metric": {"raw_value": 0, "type": "int"}
            },
            data_version="alpha",
        )
    
    print("hello from stdout")
    print("hello from stderr", file=sys.stderr)

produces this:

The line "hello from stderr" which is explicitly printed to stderr still ends up in the /output LogStream.

I don' think we can (or should) do anything about it. This is just how Glue is.

[schrockn]

@schrockn As I suspected, AWS Glue routes both stdout and stderr to /output CloudWatch log group. The /error sterams in CloudWatch only contain Spark logs.

The following code:

with open_dagster_pipes(
        params_loader=PipesCliArgsParamsLoader(),
        context_loader=PipesS3ContextLoader(client=boto3.client("s3"))
    ) as pipes:
    pipes.log.info("Hello from external process!")
    pipes.report_asset_materialization(
            metadata={
                "some_metric": {"raw_value": 0, "type": "int"}
            },
            data_version="alpha",
        )
    
    print("hello from stdout")
    print("hello from stderr", file=sys.stderr)

produces this:

The line "hello from stderr" which is explicitly printed to stderr still ends up in the /output LogStream.

I don' think we can (or should) do anything about it. This is just how Glue is.

Yup 100% agree. Can you document this behavior in both the PR summary and as a docblock?

[schrockn]

In terms of eliminating the glue-specific wrapper reader/writer classes, can you do a seperate PR stacked on this with colton included so we can discuss it separately?

[schrockn]

request for docs

[danielgafni]

I believe this PR should be ready