Enable perioding resuming to fetch logs for KPOA

[dstandish]

This PR modifies Async KPO to periodicall resume the sync portion of the task to fetch and emit the latest logs before deferring again.

To control this behavior a new param is added: logging_interval. If None then Async KPO operates as it did before, never resuming the sync task until the pod is done running. If logging_interval is given a value (integer seconds) then the trigger will resume every logging_interval seconds to fetch latest logs and defer again (until pod is done).

Here's a task example you can test this with:

test_kubernetes_pod = KubernetesPodOperatorAsync(
    namespace='airflow',
    image='python:3.8-slim',
    cmds=[
        "bash",
        "-cx",
        (
            "i=0; "
            "while [ $i -ne 30 ]; "
            "do i=$(($i+1)); "
            "echo $i; "
            "sleep 1; "
            "done; "
            "mkdir -p /airflow/xcom/; "
            "echo '{\"message\": \"good afternoon!\"}' > /airflow/xcom/return.json"
        ),
    ],
    name="test_kubernetes_pod",
    in_cluster=False,
    task_id="test_kubernetes_pod",
    logging_interval=10,
    dag=dag,
    is_delete_operator_pod=False,
    # do_xcom_push=True,
)

[dstandish]

@kaxil this is ready for review but not for merge because it needs to wait on another k8s provider release
tests passing locally now though

[kaxil]

This looks great!! Have you tested in on Astro Cloud or locally via chart?

[dstandish]

This looks great!! Have you tested in on Astro Cloud or locally via chart?

Only locally... it should be same now that we have gotten that service account issue resolved... but i can work on that if you think it's important to do.

Related note, are we gonna OSS the istio stuff?

[kaxil]

This looks great!! Have you tested in on Astro Cloud or locally via chart?

Only locally... it should be same now that we have gotten that service account issue resolved... but i can work on that if you think it's important to do.

@phanikumv Can you assign this to someone in the team to test too please?

Related note, are we gonna OSS the istio stuff?

Most-likely yes, in a way that is reusable for the OSS community.

[phanikumv]

@pankajastro please test this once merged to main as part of our integration test

[dstandish]

this is on hold until next provider release

[raphaelauv]

@dstandish apache-airflow-providers-cncf-kubernetes 4.0.0 is out 👍

[kaxil]

apache-airflow-providers-cncf-kubernetes 4.0.0 depends on apache-airflow>=2.3.0

https://app.circleci.com/pipelines/github/astronomer/astronomer-providers/1165/workflows/dfda1e9e-426b-437e-8825-7c9e4bc4ccce/jobs/3125
😮

[kaxil]

https:/apache/airflow/blob/0cd8833df74f4b0498026c4103bab130e1fc1068/airflow/providers/cncf/kubernetes/provider.yaml#L45

I don't think this was intentional, we don't need to depend on Airflow 2.3.0 @dstandish - Let's fix this in OSS

Oh well, I forgot about this -> apache/airflow@6db30f3

[raphaelauv]

@kaxil since astronomer-providers is not related to the KubernetesExecutor

it could be based on

kubernetes==21.7.0
apache-airflow-providers-cncf-kubernetes==4.0.0 --no-deps

apache/airflow#22836 (comment)

[kaxil]

@kaxil since astronomer-providers is not related to the KubernetesExecutor

it could be based on

kubernetes==21.7.0
apache-airflow-providers-cncf-kubernetes==4.0.0 --no-deps

apache/airflow#22836 (comment)

Even if we do that, all the users of this provider will have to do the same to install it :(

Let me think more about it and see if it is better to release a new version of Kubernetes Provider on Apache Airflow with just the logging related changes without the k8s client change.

[kaxil]

@dstandish Now that 2.3 is out and new provider version is out, can we resurrect this PR?

[phanikumv]

@pankajastro will fix the mypy issue and get this merged to main

[codecov]

Codecov Report

Merging #139 (6bfd526) into main (0249916) will increase coverage by 0.23%.
The diff coverage is 97.77%.

@@            Coverage Diff             @@
##             main     #139      +/-   ##
==========================================
+ Coverage   96.79%   97.03%   +0.23%     
==========================================
  Files          56       56              
  Lines        2906     2938      +32     
==========================================
+ Hits         2813     2851      +38     
+ Misses         93       87       -6     

Impacted Files	Coverage Δ
...viders/cncf/kubernetes/operators/kubernetes_pod.py	93.65% <96.66%> (+8.28%)	⬆️
...oviders/cncf/kubernetes/triggers/wait_container.py	98.63% <100.00%> (+6.56%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 0249916...6bfd526. Read the comment docs.

[pankajastro]

@pankajastro will fix the mypy issue and get this merged to main

Mypy issue has been fixed. but test coverage is less than the threshold. I'll fix that too.

[pankajastro]

@pankajastro will fix the mypy issue and get this merged to main

The unit test coverage is passing now.

[phanikumv]

Good job @pankajastro on getting the mypy fixes done quickly.

[pankajastro]

Tested. LGTM.