Add OpensearchRemoteLogIO.from_config and register opensearch scheme by FrankYang0529 · Pull Request #70295 · apache/airflow

FrankYang0529 · 2026-07-23T11:42:25Z

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'

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 · 2026-08-01T03:19:02Z

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 · 2026-08-01T16:43:31Z

+            "base_log_folder",
+        }
+        io_kwargs = {k: v for k, v in remote_task_handler_kwargs.items() if k not in fth_params}
+        port = conf.get("opensearch", "port", fallback="")


Suggested change

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

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

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.

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.

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

FrankYang0529 · 2026-08-03T06:14:16Z

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 · 2026-08-03T08:12:36Z

+            "base_log_folder",
+        }
+        io_kwargs = {k: v for k, v in remote_task_handler_kwargs.items() if k not in fth_params}
+        port = conf.get("opensearch", "port", fallback="")


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.

Signed-off-by: PoAn Yang <payang@apache.org>

boring-cyborg Bot added area:logging area:providers kind:documentation provider:opensearch labels Jul 23, 2026

FrankYang0529 marked this pull request as ready for review July 23, 2026 13:52

FrankYang0529 force-pushed the airflow-70272 branch from 4d1174b to de1fb45 Compare July 24, 2026 02:47

This was referenced Aug 1, 2026

Add ElasticsearchRemoteLogIO.from_config and register elasticsearch scheme #70525

Merged

Add WasbRemoteLogIO.from_config and register wasb remote logging scheme #70301

Merged

jason810496 reviewed Aug 1, 2026

View reviewed changes

jason810496 mentioned this pull request Aug 1, 2026

Clarify Elasticsearch remote-logging scheme needs Airflow 3.3.0+ #70920

Merged

1 task

FrankYang0529 force-pushed the airflow-70272 branch from de1fb45 to a1053c0 Compare August 3, 2026 06:05

jason810496 reviewed Aug 3, 2026

View reviewed changes

Add OpensearchRemoteLogIO.from_config and register opensearch scheme

21df2b0

Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 force-pushed the airflow-70272 branch from a1053c0 to 21df2b0 Compare August 3, 2026 13:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add OpensearchRemoteLogIO.from_config and register opensearch scheme - #70295

Add OpensearchRemoteLogIO.from_config and register opensearch scheme#70295
FrankYang0529 wants to merge 1 commit into
apache:mainfrom
FrankYang0529:airflow-70272

FrankYang0529 commented Jul 23, 2026 •

edited

Loading

Uh oh!

potiuk commented Aug 1, 2026

Uh oh!

jason810496 left a comment

Uh oh!

Uh oh!

jason810496 Aug 1, 2026

Uh oh!

FrankYang0529 Aug 3, 2026 •

edited

Loading

Uh oh!

jason810496 Aug 3, 2026

Uh oh!

FrankYang0529 Aug 3, 2026

Uh oh!

FrankYang0529 commented Aug 3, 2026

Uh oh!

jason810496 left a comment

Uh oh!

jason810496 Aug 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

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

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

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

Uh oh!

Conversation

FrankYang0529 commented Jul 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why

How

Verify

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

Uh oh!

potiuk commented Aug 1, 2026

Uh oh!

jason810496 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jason810496 Aug 1, 2026

Choose a reason for hiding this comment

Uh oh!

FrankYang0529 Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jason810496 Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

FrankYang0529 Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

FrankYang0529 commented Aug 3, 2026

Uh oh!

jason810496 left a comment

Choose a reason for hiding this comment

Uh oh!

jason810496 Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

FrankYang0529 commented Jul 23, 2026 •

edited

Loading

FrankYang0529 Aug 3, 2026 •

edited

Loading