Skip to content

Handle exception during serializing incompatible executor_config object. #27428

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tirkarthi wants to merge 4 commits into from
Closed

Handle exception during serializing incompatible executor_config object. #27428

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0bc5a2d
Handle exception during serializing incompatible executor_config obje…
tirkarthi Nov 1, 2022
4c0b5ef
Update tests to patch executor_config instead of serialize method.
tirkarthi Nov 3, 2022
d53396e
Fix tests and typo in method call.
tirkarthi Nov 3, 2022
3958215
Fix tests.
tirkarthi Jan 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion airflow/api_connexion/schemas/task_instance_schema.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,14 @@
from airflow.utils.state import State


class _ExecutorConfigField(fields.String):
def _serialize(self, value, attr, obj, **kwargs):
try:
return super()._serialize(value, attr, obj, **kwargs)
except Exception:

@uranusjr uranusjr Nov 3, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What exceptions are possible? This except seems much too broad to me.

@tirkarthi tirkarthi Nov 3, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The executor_config in case of kubernetes can have values serialized and stored in one version to be picked up later by another kubernetes version in case of Airflow upgrade with old task instances. We noticed AttributeError in our case as per the issue but if the serialization fails due to other cases we just want to return {} so that other task instances are not affected. There could be other exceptions here as well. PR is similar to #24117

@dstandish dstandish Nov 3, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@uranusjr curious what's problematic with being very broad.... the goal i guess is to not fail the whole request just because one field fails to serialize.

@uranusjr uranusjr Nov 10, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the problem is mostly that we have no way to communicate the failed serialisation if we swallow everything here.

@dstandish dstandish Jan 7, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@uranusjr how would you recommend we communicate failed serialization? re what exceptions it could be... we should no longer have an issue with k8s pod objects. and the only thing i know of that is ever put in exec config is pod object. so i am not sure we have any way to know what other exceptions we might expect.

@dimberman dimberman Feb 13, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's generally bad form to catch all Exceptions because if you get an exception didn't expect it would be better for things to fail in a big way instead of a controlled way. A lot of linters explicitly call this out. I think it's reasonable to catch the exceptions we expect to catch given the knowledge we have instead of a broad Exception.

return "{}"
Comment on lines 38 to 41

@dstandish dstandish Nov 3, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
try:
return super()._serialize(value, attr, obj, **kwargs)
except Exception:
return "{}"
with suppress():
return super()._serialize(value, attr, obj, **kwargs)

(must be imported from contextlib)

Perhaps in this case it's best not to return empty dict because.... we actually don't know what it was, and saying it's empty dict is a lie. we could even instead return '' or something. but the above suggestion just returns None.... i'm not sure if that's handled well downstream. but you could try it.

@tirkarthi tirkarthi Dec 9, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess I received an error since this is not nullable in API spec and should be a string. I agree if there is a definite way to signal error in the string I would be happy to make the changes.

airflow/airflow/api_connexion/openapi/v1.yaml

Lines 3235 to 3236 in 2f5c77b

executor_config:
type: string

@dstandish dstandish Jan 7, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think you could make it optionally null. we can change schema if it's backward compatible.

@dstandish dstandish Jan 7, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what do you think would be best return value @uranusjr ?

@uranusjr uranusjr Jan 9, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we don’t expect this to happen (anymore), perhaps it is better to leave the exception (which would result in a 502 reponse, persumably?)

Or maybe an empty string would be viable? And we can probably document that behaviour in the API spec.

@dimberman dimberman Feb 13, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd rather an exception if possible. Even if documented it still fells... clunky...



class TaskInstanceSchema(SQLAlchemySchema):
"""Task instance schema."""

Expand Down Expand Up @@ -61,7 +69,7 @@ class Meta:
operator = auto_field()
queued_dttm = auto_field(data_key="queued_when")
pid = auto_field()
executor_config = auto_field()
executor_config = _ExecutorConfigField()
note = auto_field()
sla_miss = fields.Nested(SlaMissSchema, dump_default=None)
rendered_fields = JsonObjectField(dump_default={})
Expand Down
50 changes: 50 additions & 0 deletions tests/api_connexion/endpoints/test_task_instance_endpoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,11 @@
DEFAULT_DATETIME_STR_2 = "2020-01-02T00:00:00+00:00"


class _BadExecutorConfig:
def __str__(self):
raise Exception()
Comment on lines 43 to 45

@dstandish dstandish Nov 3, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
class _BadExecutorConfig:
def __str__(self):
raise Exception()
class BadExecutorConfig:
"""This is used to test what happens when executor config serialization fails"""
def __str__(self):
raise Exception("Failed to render as string")

to give a little context how this obj is used

@dimberman dimberman Feb 13, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If anything, we can create a custom exception class here. Somehting like BadExecutorConfigException(AirflowException).



@pytest.fixture(scope="module")
def configured_app(minimal_app_for_api):
app = minimal_app_for_api
Expand Down Expand Up @@ -220,6 +225,51 @@ def test_should_respond_200(self, username, session):
"triggerer_job": None,
}

@pytest.mark.parametrize(
"executor_config, expected_config",
[
pytest.param(_BadExecutorConfig(), "{}", id="Exception serialization"),
pytest.param({"foo": "bar"}, "{'foo': 'bar'}", id="Dict serialization"),
],
)
def test_task_instance_executor_config_exception(self, executor_config, expected_config, session):
tis = self.create_task_instances(session)
print_the_context = next(ti for ti in tis if ti.task_id == "print_the_context")
print_the_context.executor_config = executor_config
response = self.client.get(
"/api/v1/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances/print_the_context",
environ_overrides={"REMOTE_USER": "test"},
)
assert response.status_code == 200
assert response.json == {
"dag_id": "example_python_operator",
"duration": 10000.0,
"end_date": "2020-01-03T00:00:00+00:00",
"execution_date": "2020-01-01T00:00:00+00:00",
"executor_config": expected_config,
"hostname": "",
"map_index": -1,
"max_tries": 0,
"operator": "_PythonDecoratedOperator",
"note": "placeholder-note",
"pid": 100,
"pool": "default_pool",
"pool_slots": 1,
"priority_weight": 11,
"queue": "default_queue",
"queued_when": None,
"sla_miss": None,
"start_date": "2020-01-02T00:00:00+00:00",
"state": "running",
"task_id": "print_the_context",
"try_number": 0,
"unixname": getuser(),
"dag_run_id": "TEST_DAG_RUN_ID",
"rendered_fields": {},
"trigger": None,
"triggerer_job": None,
}

def test_should_respond_200_with_task_state_in_deferred(self, session):
now = pendulum.now("UTC")
ti = self.create_task_instances(
Expand Down