Resources cannot be used due to issues with pickling

[awildturtok]

Apache Airflow version:

2.0.1

Environment:

• OS (e.g. from /etc/os-release): Container using Ubuntu Groovy 20.10
• Kernel (e.g. uname -a): 4.18.0-305.7.1.el8_4.x86_64

What happened:

When using resources in a dag, it has some issues related to resources and pickling.

When I add the following to any working Task:

resources={"ram" : 15000}

it breaks with the following message

File "/usr/local/lib/python3.8/dist-packages/airflow/serialization/serialized_objects.py", line 184, in serialize_to_json
    if cls._is_excluded(value, key, object_to_serialize):
  File "/usr/local/lib/python3.8/dist-packages/airflow/serialization/serialized_objects.py", line 501, in _is_excluded
    return super()._is_excluded(var, attrname, op)
  File "/usr/local/lib/python3.8/dist-packages/airflow/serialization/serialized_objects.py", line 170, in _is_excluded
    return isinstance(var, cls._excluded_types) or cls._value_is_hardcoded_default(
  File "/usr/local/lib/python3.8/dist-packages/airflow/serialization/serialized_objects.py", line 328, in _value_is_hardcoded_default
    cls._CONSTRUCTOR_PARAMS[attrname] is value or (value in [{}, []])
  File "/usr/local/lib/python3.8/dist-packages/airflow/utils/operator_resources.py", line 136, in __eq__
    return self.__dict__ == other.__dict__
AttributeError: 'dict' object has no attribute '__dict__'

What you expected to happen:

Using resources shouldn't be an issue. As far as I tracked the error down, it looks like _is_excluded makes wrong assumptions about the self object causing problems.

How to reproduce it:

def do_echo():
    logging.info("hi")

with DAG(
        'an_echoing_dag',
        schedule_interval=None,
        max_active_runs=1,
        start_date=datetime(2020, 1, 1),
) as dag:

    PythonOperator(
        task_id="task",
        python_callable=do_echo,
        resources={"ram" : 15000}
    )

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[uranusjr]

I think this can basically be fixed by changing Resources.__eq__() to something like

if not isintance(other, type(self)):
    return False
return self.__dict__ == other.__dict__

The same should probably also be done for Resource.__eq__(). Would you be ionterested in submitting a PR for this?

[awildturtok]

I think I can do that. Where would be a sensible place to add de-/serialization tests for resources? Or a test that checks exactly this code path?

[uranusjr]

Sounds reasonable to add a serialisation test case for the currently failing scenario (and then implement the fix to make the scenario to not raise an exception).

[kaxil]

@awildturtok Would you still want to work on this?

[o-nikolas]

This was fixed in #21442
Tested manually with provided dag and all seems working.
Closing this one as fixed.