Hello.

Kubernetes generates a unique pod name.
https://github.com/apache/airflow/blob/736a62f824d9062b52983633528e58c445d8cc56/airflow/kubernetes/pod_generator.py#L434-L458
It would be great if the pod name was available in Airflow after completing the task, so that, for example, we could use it to add [extra links](http://airflow.apache.org/docs/apache-airflow/stable/howto/define_extra_link.html) or use it as an argument in downstream tasks. To do this, we should save this name in XCOM table.

The operator for BigQuery works in a similar way.
https://github.com/apache/airflow/blob/736a62f824d9062b52983633528e58c445d8cc56/airflow/providers/google/cloud/operators/bigquery.py#L730
Thanks to this, we have links to the BigQuery console.
https://github.com/apache/airflow/blob/736a62f824d9062b52983633528e58c445d8cc56/airflow/providers/google/cloud/operators/bigquery.py#L600-L605
https://github.com/apache/airflow/blob/736a62f824d9062b52983633528e58c445d8cc56/airflow/providers/google/cloud/operators/bigquery.py#L57-L86

Best regards,
Kamil Breguła


kubernetes_tests/test_kubernetes_pod_operator.py
'nodeAffinity': {
                'requiredDuringSchedulingIgnoredDuringExecution': {
                    'nodeSelectorTerms': [
                        {
                            'matchExpressions': [
                                {'key': 'beta.kubernetes.io/os', 'operator': 'In', 'values': ['linux']}
                            ]
                        }
                    ]
                }
            }
        }
        k = KubernetesPodOperator(
            namespace='default',
            image="ubuntu:16.04",
            cmds=["bash", "-cx"],
            arguments=["echo 10"],
            labels={"foo": "bar"},
            name="test-" + str(random.randint(0, 1000000)),
            task_id="task" + self.get_current_task_name(),
            in_cluster=False,
            do_xcom_push=False,
            affinity=affinity,
        )
        context = create_context(k)
        k.execute(context=context)
        actual_pod = self.api_client.sanitize_for_serialization(k.pod)
        self.expected_pod['spec']['affinity'] = affinity
        assert self.expected_pod == actual_pod
    def test_port(self):