Add ElasticsearchRemoteLogIO.from_config and register elasticsearch scheme

[jason810496]

Part of #70265 (related: #67056).

Why

#67056 decoupled remote logging from the hardcoded branches in airflow_local_settings.py: core and
the Task SDK now resolve the handler via ProvidersManager dispatch on the
[logging] remote_base_log_folder URL scheme, instantiating the provider class through a no-arg
from_config() classmethod. This issue migrates the Elasticsearch backend.

Design note (feedback welcome)

Unlike the object-storage backends, Elasticsearch is currently selected by [elasticsearch] host
being set — remote_base_log_folder typically has no scheme for these deployments, so pure scheme
dispatch cannot reach it. The proposal:

• Register an elasticsearch scheme and document setting
  [logging] remote_base_log_folder = elasticsearch:// as the forward-looking configuration.
• Keep the legacy host-based selection working via the transitional fallback in
  airflow_local_settings.py until the chain is removed (tracked by the meta issue), so existing
  configs are unaffected.

What

• Add ElasticsearchRemoteLogIO.from_config() in
  providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py,
  mirroring the legacy branch:
  

  airflow/airflow-core/src/airflow/config_templates/airflow_local_settings.py

  Lines 304 to 347 in 104ad12

  	elif ELASTICSEARCH_HOST:
  	from airflow.providers.elasticsearch.log.es_task_handler import ElasticsearchRemoteLogIO
  	ELASTICSEARCH_WRITE_STDOUT: bool = conf.getboolean("elasticsearch", "WRITE_STDOUT")
  	ELASTICSEARCH_WRITE_TO_ES: bool = conf.getboolean("elasticsearch", "WRITE_TO_ES")
  	ELASTICSEARCH_JSON_FORMAT: bool = conf.getboolean("elasticsearch", "JSON_FORMAT")
  	ELASTICSEARCH_TARGET_INDEX: str = conf.get_mandatory_value("elasticsearch", "TARGET_INDEX")
  	ELASTICSEARCH_HOST_FIELD: str = conf.get_mandatory_value("elasticsearch", "HOST_FIELD")
  	ELASTICSEARCH_OFFSET_FIELD: str = conf.get_mandatory_value("elasticsearch", "OFFSET_FIELD")
  	ELASTICSEARCH_LOG_ID_TEMPLATE: str = conf.get_mandatory_value("elasticsearch", "LOG_ID_TEMPLATE")
  	ELASTICSEARCH_END_OF_LOG_MARK: str = conf.get_mandatory_value("elasticsearch", "END_OF_LOG_MARK")
  	ELASTICSEARCH_FRONTEND: str = conf.get_mandatory_value("elasticsearch", "FRONTEND")
  	ELASTICSEARCH_JSON_FIELDS: str = conf.get_mandatory_value("elasticsearch", "JSON_FIELDS")
  	ELASTICSEARCH_REMOTE_HANDLERS: dict[str, dict[str, str | bool | None]] = {
  	"task": {
  	"class": "airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchTaskHandler",
  	"formatter": "airflow",
  	"base_log_folder": BASE_LOG_FOLDER,
  	"end_of_log_mark": ELASTICSEARCH_END_OF_LOG_MARK,
  	"host": ELASTICSEARCH_HOST,
  	"frontend": ELASTICSEARCH_FRONTEND,
  	"write_stdout": ELASTICSEARCH_WRITE_STDOUT,
  	"write_to_es": ELASTICSEARCH_WRITE_TO_ES,
  	"json_format": ELASTICSEARCH_JSON_FORMAT,
  	"json_fields": ELASTICSEARCH_JSON_FIELDS,
  	"host_field": ELASTICSEARCH_HOST_FIELD,
  	"offset_field": ELASTICSEARCH_OFFSET_FIELD,
  	},
  	}
  	DEFAULT_LOGGING_CONFIG["handlers"].update(ELASTICSEARCH_REMOTE_HANDLERS)
  	REMOTE_TASK_LOG = ElasticsearchRemoteLogIO(
  	host=ELASTICSEARCH_HOST,
  	target_index=ELASTICSEARCH_TARGET_INDEX,
  	write_stdout=ELASTICSEARCH_WRITE_STDOUT,
  	write_to_es=ELASTICSEARCH_WRITE_TO_ES,
  	offset_field=ELASTICSEARCH_OFFSET_FIELD,
  	host_field=ELASTICSEARCH_HOST_FIELD,
  	base_log_folder=BASE_LOG_FOLDER,
  	delete_local_copy=delete_local_copy,
  	json_format=ELASTICSEARCH_JSON_FORMAT,
  	log_id_template=ELASTICSEARCH_LOG_ID_TEMPLATE,
  	)

  
  — reading the [elasticsearch] options the branch reads (host, target_index,
  write_stdout, write_to_es, json_format, log_id_template, host_field,
  offset_field) plus [logging] base_log_folder (with expanduser) and
  [logging] delete_local_logs, so behavior is unchanged for existing configs. Note the legacy
  branch does not merge remote_task_handler_kwargs IO-kwargs for this backend — mirror that
  too.
• Register the elasticsearch scheme under a remote-logging: section in
  providers/elasticsearch/provider.yaml and mirror it in
  providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py.
• Document the remote_base_log_folder = elasticsearch:// form in the provider's logging docs.
• Add tests mirroring TestS3RemoteLogIOFromConfig in
  providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py.
• Verify end to end with a real system test: set up the actual backend yourself (a real
  Elasticsearch cluster or a local one, e.g. via Docker), run a task with remote logging
  enabled, and confirm logs are written and read back through the new dispatch path. Include
  the setup and verification results in the PR description.

Notes

The legacy branch also swaps the task handler class inside DEFAULT_LOGGING_CONFIG; that dict is
kept only for import compatibility (see the comment at its definition in
airflow_local_settings.py), so from_config should not need to replicate it — but please verify
task-log reads work end to end with the new dispatch.

Reference

Merged examples to follow: #69817 (s3), #69816 (cloudwatch). If from_config raises on a bad
config, the shared factory falls back to the legacy path, so this is not a breaking change.

[henry3260]

@yuseok89 Could you drop a comment here so I can assign it to you?

[yuseok89]

Thanks, @henry3260 .
I'd like to work on this.