**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
"""
    ElasticsearchTaskHandler is a python log handler that
    reads logs from Elasticsearch. Note logs are not directly
    indexed into Elasticsearch. Instead, it flushes logs
    into local files. Additional software setup is required
    to index the log into Elasticsearch, such as using
    Filebeat and Logstash.
    To efficiently query and sort Elasticsearch results, we assume each
    log message has a field `log_id` consists of ti primary keys:
    `log_id = {dag_id}-{task_id}-{execution_date}-{try_number}`
    Log messages with specific log_id are sorted based on `offset`,
    which is a unique integer indicates log message's order.
    Timestamp here are unreliable because multiple log messages
    might have the same timestamp.
    """
    PAGE = 0
    MAX_LINE_PER_PAGE = 1000
    LOG_NAME = 'Elasticsearch'
    def __init__(