Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: [email protected]
  Slack: https://s.apache.org/airflow-slack

Added test for the change.

Cool. Running tests and hopefully will include it in the next provider's release.

How can we template resources? Its not a string. Its a list object of V1ResourceRequirements.

Maybe im missing something here but I would love to see an actual test that template this field.

Do we mean to something like?

    resources = [
        V1ResourceRequirements(
            requests={"cpu": "{{ dag_run.conf[\'request_cpu\'] }}", "memory": "8G"},
            limits={"cpu": "16000m", "memory": "128G"},
        )
    ]

@eladkal yes. This is the expected behaviour.

I tried locally with kind-cluster and the templating didn't take. @nitinmuteja can you supply a usage example for it?
this is what I tried using the example_kubernetes_executor.py from the example_dags in the repo:

....
  with DAG(
        dag_id='example_kubernetes_executor',
        schedule_interval=None,
        start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
        catchup=False,
        tags=['example3'],
        params={
            'mem': Param(default='512Mi',type='string', description="memory"),
        }
....
 # Use k8s_client.V1ResourceRequirements to define resource limits
        k8s_resource_requirements = k8s.V1ResourceRequirements(
            requests={'memory': "{{ params['mem'] }}"}, limits={'memory': '512Mi'}
        )

@nitinmuteja It would be great if you can add a test that verify the templating similar to:

@nitinmuteja -> last call before preparing newr round of providers - do you have time/capability to add the test as @eladkal asked?

I can add today.

Cool. I can hold on then :)

@talnagar I have added test cases for the requirements.

@potiuk please check the changes. Created a subclass from the k8s class and added class attributes to make limits and requests as template attributes.

I'm not really on board with having our own V1ResourceRequirements - we intentionally moved away from this pattern for a reason.

I agree

Agree - this one will wait move to next round of providers.

I'm not really on board with having our own V1ResourceRequirements - we intentionally moved away from this pattern for a reason.

@jedcunningham we are just inheriting the object and will continue using the object. The only advantage it provides us with is that the props can be templated.

I'm not really on board with having our own V1ResourceRequirements - we intentionally moved away from this pattern for a reason.

I haven't tried it, but would this pattern work instead (making this a documentation issue)?

resources = k8s.V1ResourceRequirements(requests={'memory': '100Mi'})
resources.template_fields = ('requests')

KPO(
...,
resources=resources
)

This one does not work as the external V1ResourceRequirements does not have template_fields class attribute. I can only think of Inheritance as a stable solution for this. Let me know if you folks can think of another solution.

Suggestion: you can modify _render_nested_template_fields in KubernetesPodOperator and add the following code, right after the custom code that is handling k8s.V1EnvVar:

if id(content) not in seen_oids and isinstance(content, k8s.V1ResourceRequirements):
    seen_oids.add(id(content))
    self._do_render_template_fields(content, ("limits", "requests"), context, jinja_env, seen_oids)
    return

In addition to previously mentioned by @lior1990,

class CustomKubernetesPodOperator(KubernetesPodOperator):
    template_fields: Sequence[str] = KubernetesPodOperator.template_fields + ('k8s_resources', )

    def _render_nested_template_fields(
        self,
        content: Any,
        context: 'Context',
        jinja_env: "jinja2.Environment",
        seen_oids: set,
    ) -> None:
        if id(content) not in seen_oids and isinstance(content, k8s.V1ResourceRequirements):
            seen_oids.add(id(content))
            self._do_render_template_fields(content, ("limits", "requests"), context, jinja_env, seen_oids)
            return

        super()._render_nested_template_fields(content, context, jinja_env, seen_oids)

In KubernetesPodOperator, resources allocated to k8s_resources

self.k8s_resources = convert_resources(resources) if resources else {}

Instead of resources, k8s_resources should be added to template_fields.

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.