[dagster-pipes] Report / handle exceptions in the external process

[derek1032]

Hi Dagster team,

Thank you for creating this amazing feature pipes. We've been testing the pipes functionality using pipes_k8s_client, however, we encounter some issues that may require your support to resolve:

1. We discovered that when the external process raise exceptions, the behavior of the orchetration process is somewhat unpredictable. Sometimes it can successfuly capture external messages, but other times it displays a warning message and yields STEP_FAILURE event. The following is the warning message we received:

<img width="636" alt="Screen Shot 2023-11-14 at 12 17 25 PM" src="https://github.com/dagster-io/dagster/assets/13246451/5451aeb2-6f66-4aca-be9a-0ae35e528723">

[pipes] did not receive any messages from external process. Check stdout / stderr logs from the external process if possible.
PipesEnvContextInjector: Attempted to inject context directly, typically as an environment variable.
PipesK8sPodLogsMessageReader: Attempted to read messages by extracting them from kubernetes pod logs directly.

2. We've noticed there is an experimental change in dagster 1.5.7 about how pipes handles external exceptions. Thank you for developing this useful feature. After upgrading to 1.5.7, we can successfully see the message external process pipes closed with exception; However, the run can still end with a SUCCESS event. We are wondering what would be the best practice to report the red engine event, which contains external error messages, if we are using slack_on_run_failure_sensor.

3. If we are using pipes_k8s_client in partitioned ops instead of partitioned assets, the run will always produce a STEP_FAILURE event with error messages shown below. We would like to ask if current pipes feature is unable to be used on ops or if there is a way to resolve this issue?

More info:
Dagster version: 1.5.7
dagster-pipes: 1.5.7

Thank you,
Derek

[alangenfeld]

Thanks for the report, I will look in to this.

We discovered that when the external process raise exceptions, the behavior of the orchetration process is somewhat unpredictable

Just to confirm, when the external processes raises an exception does that bubble up and cause the container process to return a non zero exit code?

[alangenfeld]

hmm strange. This is the code I expect to raise (and what I see raise locally) https://github.com/dagster-io/dagster/blame/1.5.8/python_modules/libraries/dagster-k8s/dagster_k8s/client.py#L630-L635.

Can you share the body of k8s_pipes_asset ?

[derek1032]

Sure, here is the body of k8s_pipes_asset

[derek1032]

Hi @alangenfeld, I am suspecting when setting wait_for_state=WaitForPodState.Ready, the execution will break out of the while loop if the container status is ready and it may not reach the code you expect to raise DagsterK8sError

dagster/python_modules/libraries/dagster-k8s/dagster_k8s/pipes.py

Lines 247 to 259 in 41bc786

	try:
	# if were doing direct pod reading, wait for pod to start and then stream logs out
	if isinstance(self.message_reader, PipesK8sPodLogsMessageReader):
	client.wait_for_pod(
	pod_name,
	namespace,
	wait_for_state=WaitForPodState.Ready,
	)
	self.message_reader.consume_pod_logs(
	core_api=client.core_api,
	pod_name=pod_name,
	namespace=namespace,
	)

dagster/python_modules/libraries/dagster-k8s/dagster_k8s/client.py

Lines 579 to 588 in 1f15855

	if wait_for_state == WaitForPodState.Ready:
	# ready is boolean field of container status
	ready = container_status.ready
	if not ready:
	self.logger('Waiting for pod "%s" to become ready...' % pod_name)
	self.sleeper(wait_time_between_attempts)
	continue
	else:
	self.logger('Pod "%s" is ready, done waiting' % pod_name)
	break

[alangenfeld]

ah yep good catch, will get a fix out

[alangenfeld]

#18076

[alangenfeld]

in partitioned ops instead of partitioned assets, the run will always produce a STEP_FAILURE event with error messages shown below. We would like to ask if current pipes feature is unable to be used on ops or if there is a way to resolve this issue?

get_materialize_result() is designed to work in the context of an @asset, in an @op you should be using get_results(). #17921 is exploring some improvements here, but we may do some further improvements to the error message.

[derek1032]

@alangenfeld Thank you for your information. I've switched to using return pipes_k8s_client.run(...).get_results() in the context of @op. However, it appears that current behavior of the pipes will consistently yield a STEP_FAILUREwith the following error messages. Unsure if the improvements you are exploring can address this issue and enable the use of pipes_k8s_client in @ops.

[alangenfeld]

How are you defining your @op ? Will think on how to improve the ergonomics here but I expect this to work:

@op
def sample(context):
  yield from pipes_k8s_client.run().get_results()
  yield Output(...) 

and possibly this (not verified), assuming your op doesn't return anything

@op(out=Out(Nothing))
def sample(context):
  yield from pipes_k8s_client.run().get_results()

[derek1032]

The first method works, but the second one doesn't. Thank you!

[alangenfeld]

#18077