Skip to content

Add OpensearchRemoteLogIO.from_config and register opensearch scheme #70272

Description

@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 OpenSearch backend.

Design note (feedback welcome)

Unlike the object-storage backends, OpenSearch is currently selected by [opensearch] host being
set — remote_base_log_folder typically has no scheme for these deployments, so pure scheme
dispatch cannot reach it. The proposal (same as the sibling Elasticsearch issue #70271):

  • Register an opensearch scheme and document setting
    [logging] remote_base_log_folder = opensearch:// 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 OpensearchRemoteLogIO.from_config() in
    providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py, mirroring the
    legacy branch:
    elif OPENSEARCH_HOST:
    from airflow.providers.opensearch.log.os_task_handler import OpensearchRemoteLogIO
    OPENSEARCH_PORT = conf.getint("opensearch", "PORT", fallback=9200)
    OPENSEARCH_USERNAME: str = conf.get_mandatory_value("opensearch", "USERNAME")
    OPENSEARCH_PASSWORD: str = conf.get_mandatory_value("opensearch", "PASSWORD")
    OPENSEARCH_WRITE_STDOUT: bool = conf.getboolean("opensearch", "WRITE_STDOUT")
    OPENSEARCH_WRITE_TO_OS: bool = conf.getboolean("opensearch", "WRITE_TO_OS")
    OPENSEARCH_JSON_FORMAT: bool = conf.getboolean("opensearch", "JSON_FORMAT")
    OPENSEARCH_TARGET_INDEX: str = conf.get_mandatory_value("opensearch", "TARGET_INDEX")
    OPENSEARCH_HOST_FIELD: str = conf.get_mandatory_value("opensearch", "HOST_FIELD")
    OPENSEARCH_OFFSET_FIELD: str = conf.get_mandatory_value("opensearch", "OFFSET_FIELD")
    OPENSEARCH_LOG_ID_TEMPLATE: str = conf.get("opensearch", "LOG_ID_TEMPLATE", fallback="") or (
    "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}"
    )
    REMOTE_TASK_LOG = OpensearchRemoteLogIO(
    host=OPENSEARCH_HOST,
    port=OPENSEARCH_PORT,
    username=OPENSEARCH_USERNAME,
    password=OPENSEARCH_PASSWORD,
    target_index=OPENSEARCH_TARGET_INDEX,
    write_stdout=OPENSEARCH_WRITE_STDOUT,
    write_to_opensearch=OPENSEARCH_WRITE_TO_OS,
    offset_field=OPENSEARCH_OFFSET_FIELD,
    host_field=OPENSEARCH_HOST_FIELD,
    base_log_folder=BASE_LOG_FOLDER,
    delete_local_copy=delete_local_copy,
    json_format=OPENSEARCH_JSON_FORMAT,
    log_id_template=OPENSEARCH_LOG_ID_TEMPLATE,
    )

    — reading the [opensearch] options the branch reads (host, port, username,
    password, target_index, write_stdout, write_to_os, json_format, host_field,
    offset_field, and log_id_template with its
    {dag_id}-{task_id}-{run_id}-{map_index}-{try_number} fallback) 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 opensearch scheme under a remote-logging: section in
    providers/opensearch/provider.yaml and mirror it in
    providers/opensearch/src/airflow/providers/opensearch/get_provider_info.py.
  • Document the remote_base_log_folder = opensearch:// 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
    OpenSearch 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.

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.

Metadata

Metadata

Assignees

Type

No type

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions