Add OpensearchRemoteLogIO.from_config and register opensearch scheme

[FrankYang0529]

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. This migrates the opensearch scheme.

How

• Add OpensearchRemoteLogIO.from_config() reading the [opensearch] section — a mirror of the legacy branch, so the user-facing config surface is unchanged.
• Register the opensearch scheme in provider.yaml.

Verify

• uv run --project providers/opensearch pytest providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
• end to end test

# 1. Create config
mkdir -p files/airflow-breeze-config
cat > files/airflow-breeze-config/environment_variables.env <<'EOF'
export AIRFLOW__LOGGING__REMOTE_LOGGING=True
export AIRFLOW__LOGGING__REMOTE_BASE_LOG_FOLDER=opensearch://
export AIRFLOW__OPENSEARCH__HOST=http://opensearch:9200
export AIRFLOW__OPENSEARCH__PORT=9200
export AIRFLOW__OPENSEARCH__WRITE_TO_OS=True
export AIRFLOW__OPENSEARCH__JSON_FORMAT=True
export AIRFLOW__OPENSEARCH__WRITE_STDOUT=False
export AIRFLOW__OPENSEARCH__TARGET_INDEX=airflow-logs
EOF

# 2. Start Airflow
breeze start-airflow --integration opensearch --backend postgres --load-example-dags

# 3. Run Dag
airflow dags unpause example_bash_operator
airflow dags trigger example_bash_operator
airflow dags list-runs example_bash_operator

# 4. Check log
curl -s localhost:9200/airflow-logs/_count ; echo
curl -s 'localhost:9200/airflow-logs/_search?q=log_id:example_bash_operator&size=3'

• closes: Add OpensearchRemoteLogIO.from_config and register opensearch scheme #70272
• related: Provider migration for decoupling remote logging from core #70265

---

Was generative AI tooling used to co-author this PR?

• Yes - Claude Code

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[potiuk]

I reviewed this alongside the four sibling PRs doing the same port for other providers (#70525 elasticsearch, #70301 wasb, #70682 oss, #70549 stackdriver), and this one diverges from the legacy behaviour it is porting.

airflow_local_settings.py splits into two camps. The object-storage backends merge the IO half of remote_task_handler_kwargs:

REMOTE_TASK_LOG = WasbRemoteLogIO(
    **cast("dict[str, Any]", { ... } | _io_kwargs)
)

The two search backends do not. OpensearchRemoteLogIO is constructed with explicit keyword arguments and no | _io_kwargs at all:

REMOTE_TASK_LOG = OpensearchRemoteLogIO(
    host=OPENSEARCH_HOST,
    port=OPENSEARCH_PORT,
    ...
    log_id_template=OPENSEARCH_LOG_ID_TEMPLATE,
)

identical in shape to the ElasticsearchRemoteLogIO branch directly above it. This PR adds the merge anyway, so from_config forwards every non-FileTaskHandler key to OpensearchRemoteLogIO.__init__.

That is a behaviour change rather than a port. remote_task_handler_kwargs is one config knob whose keys are split between the file handler and the IO object, so an opensearch deployment can currently put arbitrary keys there for the file-handler side and nothing reaches the IO class. After this change those keys are passed to the constructor, and anything it does not accept raises TypeError: unexpected keyword argument at startup — on a configuration that works today.

Worth noting #70525 hit the same fork and went the other way, with the reasoning recorded in its docstring:

Unlike the object-storage backends, this does not merge [logging] remote_task_handler_kwargs IO-kwargs, matching the legacy behavior for Elasticsearch.

Two ways forward, and I don't mind which:

• Drop the merge here so the port is faithful and opensearch stays consistent with elasticsearch.
• Keep it, if you think the legacy omission was itself an oversight — but then say so in the PR description and docstring, since it changes startup behaviour for existing deployments and shouldn't ride in as an incidental detail of a port.

Everything else here looks like a clean port of the legacy branch.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[jason810496]

Thanks! I agreed with Jarek's comment overall, we need to follow the existing pattern in

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

Lines 357 to 387 in 182919a

	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,
	)

[jason810496]

Suggested change

	port = conf.get("opensearch", "port", fallback="")
	port = conf.getint("opensearch", "port", fallback=9200)

[FrankYang0529]

I prefer to use port = conf.get("opensearch", "port", fallback=""). The default value for port field is empty string. If we use conf.getint("opensearch", "port", fallback=9200), it doesn't fallback to 9200 port.

airflow/providers/opensearch/provider.yaml

Lines 110 to 116 in 4899e90

	port:
	description: |
	The port number of Opensearch host
	version_added: 1.5.0
	type: integer
	example: ~
	default: ""

It shows error like following:

ImportError: Unable to load logging config from airflow.config_templates.airflow_local_settings.DEFAULT_LOGGING_CONFIG due to: AirflowConfigException:Failed to convert value to int. Please check "PORT" key in "opensearch" section. Current value: "".

The original airflow_local_setting.py path has similar problem. I can create another PR to fix it after we get conclusion here.

[jason810496]

How about using port = conf.get("opensearch", "port", fallback="") or 9200 here and the airflow_local_setting.py.

Yes, let's create another PR to fix the same semantic error in airflow_local_setting.py, thanks.

[FrankYang0529]

Thank you. I use 9200 as default in port=int(port) if port else 9200 and keep port = conf.get("opensearch", "port", fallback="").

[FrankYang0529]

• Drop the merge here so the port is faithful and opensearch stays consistent with elasticsearch.

Thanks @potiuk and @jason810496's review. I prefer to keep behavior consistent with existing pattern.

[jason810496]

Thanks for the update. Nice catch! I didn't know this.

[jason810496]

How about using port = conf.get("opensearch", "port", fallback="") or 9200 here and the airflow_local_setting.py.

Yes, let's create another PR to fix the same semantic error in airflow_local_setting.py, thanks.