**Apache Airflow version**:
apache/airflow:1.10.11

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`):
v1.16.11-gke.5

**Environment**:
GKE


**What happened**:
Webserver doesn't fetch logs for tasks from elasticsearch

**What you expected to happen**:
task logs will be displayed in the webserver UI

It seems like the webserver is trying to query task logs by the `log_id` field:
https://github.com/apache/airflow/blob/1.10.11/airflow/utils/log/es_task_handler.py#L175

this field is missing from all log lines (which are written to stdout) using the KubernetesExecutor. Example log line:
`{"asctime": null, "filename": "standard_task_runner.py", "lineno": 77, "levelname": "INFO", "message": "Running: ['airflow', 'run', 'hello_world', 'hello_task_3', '2020-08-19T14:26:07.226064+00:00', '--job_id', '158', '--pool', 'default_pool', '--raw', '-sd', '/opt/airflow/dags/repo/dags/hello_world.py', '--cfg_path', '/tmp/tmpt7lafkaf']", "dag_id": "hello_world", "task_id": "hello_task_3", "execution_date": "2020_08_19T14_26_07_226064", "try_number": "1"}`


**How to reproduce it**:
this is the relevant configuration we have, scheduler and webserver running separately and tasks run using KubernetsExecutor (all in the same cluster/namespace):
```
AIRFLOW__CORE__LOGGING_LEVEL: INFO
AIRFLOW__CORE__REMOTE_LOGGING: "True"
AIRFLOW__ELASTICSEARCH__HOST: http://elasticsearch.logging:9200
AIRFLOW__ELASTICSEARCH__JSON_FORMAT: "True"
AIRFLOW__ELASTICSEARCH__WRITE_STDOUT: "True"
```

we are using fluentd (https://github.com/fluent/fluentd-kubernetes-daemonset) to forward log lines to elasticsearch, all task logs are written to stdout + elasticsearch as expected.
airflow/providers/elasticsearch/log/es_task_handler.py
:param log_id_template: log id template
        :param host: Elasticsearch host name
        """
        es_kwargs = es_kwargs or {}
        super().__init__(base_log_folder, filename_template)
        self.closed = False
        self.log_id_template, self.log_id_jinja_template = parse_template_string(log_id_template)
        self.client = elasticsearch.Elasticsearch([host], **es_kwargs)
        self.frontend = frontend
        self.mark_end_on_close = True
        self.end_of_log_mark = end_of_log_mark
        self.write_stdout = write_stdout
        self.json_format = json_format
        self.json_fields = [label.strip() for label in json_fields.split(",")]
        self.handler = None
        self.context_set = False
    def _render_log_id(self, ti: TaskInstance, try_number: int) -> str:
        if self.log_id_jinja_template:
            jinja_context = ti.get_template_context()
            jinja_context['try_number'] = try_number
            return self.log_id_jinja_template.render(**jinja_context)
        if self.json_format:
            execution_date = self._clean_execution_date(ti.execution_date)
        else:
            execution_date = ti.execution_date.isoformat()
        return self.log_id_template.format(
            dag_id=ti.dag_id, task_id=ti.task_id, execution_date=execution_date, try_number=try_number
        )
    @staticmethod
    def _clean_execution_date(execution_date: datetime) -> str: