Support non-https elasticsearch external links (#16489)

Also provide an example for `frontend` as its not immediately apparent
how to build a short(ish) deep link.

airflow/config_templates/config.yml

@@ -1978,10 +1978,11 @@
       description: |
         Qualified URL for an elasticsearch frontend (like Kibana) with a template argument for log_id
         Code will construct log_id using the log_id template from the argument above.
-        NOTE: The code will prefix the https:// automatically, don't include that here.
+        NOTE: scheme will default to https if one if not provided
       version_added: 1.10.4
       type: string
-      example: ~
+      example: "http://localhost:5601/app/kibana#/discover\
+        ?_a=(columns:!(message),query:(language:kuery,query:'log_id: \"{log_id}\"'),sort:!(log.offset,asc))"
       default: ""
     - name: write_stdout
       description: |

airflow/config_templates/default_airflow.cfg

@@ -985,7 +985,8 @@ end_of_log_mark = end_of_log
 
 # Qualified URL for an elasticsearch frontend (like Kibana) with a template argument for log_id
 # Code will construct log_id using the log_id template from the argument above.
-# NOTE: The code will prefix the https:// automatically, don't include that here.
+# NOTE: scheme will default to https if one if not provided
+# Example: frontend = http://localhost:5601/app/kibana#/discover?_a=(columns:!(message),query:(language:kuery,query:'log_id: "{{log_id}}"'),sort:!(log.offset,asc))
 frontend =
 
 # Write the task logs to the stdout of the worker, rather than the default files

airflow/providers/elasticsearch/log/es_task_handler.py

@@ -336,7 +336,8 @@ def get_external_log_url(self, task_instance: TaskInstance, try_number: int) ->
         :rtype: str
         """
         log_id = self._render_log_id(task_instance, try_number)
-        return 'https://' + self.frontend.format(log_id=quote(log_id))
+        scheme = '' if '://' in self.frontend else 'https://'
+        return scheme + self.frontend.format(log_id=quote(log_id))
 
     @property
     def supports_external_link(self) -> bool:

docs/apache-airflow-providers-elasticsearch/logging.rst

@@ -98,5 +98,5 @@ To enable it, ``airflow.cfg`` must be configured as in the example below. Note t
     [elasticsearch]
     # Qualified URL for an elasticsearch frontend (like Kibana) with a template argument for log_id
     # Code will construct log_id using the log_id template from the argument above.
-    # NOTE: The code will prefix the https:// automatically, don't include that here.
+    # NOTE: scheme will default to https if one if not provided
     frontend = <host_port>/{log_id}

tests/providers/elasticsearch/log/test_es_task_handler.py

@@ -413,6 +413,10 @@ def test_clean_execution_date(self):
             (False, 'localhost:5601/{log_id}', 'https://localhost:5601/' + quote(LOG_ID)),
             # Ignore template if "{log_id}"" is missing in the URL
             (False, 'localhost:5601', 'https://localhost:5601'),
+            # scheme handling
+            (False, 'https://localhost:5601/path/{log_id}', 'https://localhost:5601/path/' + quote(LOG_ID)),
+            (False, 'http://localhost:5601/path/{log_id}', 'http://localhost:5601/path/' + quote(LOG_ID)),
+            (False, 'other://localhost:5601/path/{log_id}', 'other://localhost:5601/path/' + quote(LOG_ID)),
         ]
     )
     def test_get_external_log_url(self, json_format, es_frontend, expected_url):