[kenhys]

Before: client() cleared the shared @_os slot and then rebuilt it with ||=, without a lock. With flush_thread_count of 2 or more and a host placeholder, a thread could pass the "is this the same host" check and then pick up the client another thread had just built for a different host, so its chunk went to that cluster instead.

After: the whole check-and-build runs under a mutex, so a thread always gets a client for the host it asked for. The mutex is created in initialize because configure builds a client before it returns.

[cosmo0920]

@kenhys This type of fix also ought to apply for https://github.com/uken/fluent-plugin-elasticsearch?
Once the fix is applied there, we need to release a new version of ES plugin as well.

[kenhys]

@kenhys This type of fix also ought to apply for https://github.com/uken/fluent-plugin-elasticsearch? Once the fix is applied there, we need to release a new version of ES plugin as well.

I've overlooked it, maybe similar fix is needed.

[Watson1978]

fluent-plugin-opensearch/lib/fluent/plugin/out_opensearch.rb

Lines 355 to 365 in 907cb63

	timer_execute(:out_opensearch_expire_credentials, @endpoint.refresh_credentials_interval) do
	log.debug('Recreate the AWS credentials')
	@credential_mutex.synchronize do
	@_os = nil
	begin
	@_aws_credentials = aws_credentials(@endpoint)
	rescue => e
	log.error("Failed to get new AWS credentials: #{e}")
	end
	end

This code runs periodically using a timer.
It assigns a value to @_os here as well, but because a different mutex is used, a timing conflict could cause issues.
Therefore, it needs to be protected by the same mutex.

diff --git a/lib/fluent/plugin/out_opensearch.rb b/lib/fluent/plugin/out_opensearch.rb
index cd3ff78..90e6221 100644
--- a/lib/fluent/plugin/out_opensearch.rb
+++ b/lib/fluent/plugin/out_opensearch.rb
@@ -349,7 +349,6 @@ class << self
         end
       end
       # Credential providers refresh themselves near expiration. This timer recreates them forcibly.
-      @credential_mutex = Mutex.new
       if @endpoint
         @_aws_credentials = aws_credentials(@endpoint)
 
@@ -357,12 +356,14 @@ class << self
           timer_execute(:out_opensearch_expire_credentials, @endpoint.refresh_credentials_interval) do
             log.debug('Recreate the AWS credentials')
 
-            @credential_mutex.synchronize do
-              @_os = nil
-              begin
-                @_aws_credentials = aws_credentials(@endpoint)
-              rescue => e
-                log.error("Failed to get new AWS credentials: #{e}")
+            begin
+              credentials = aws_credentials(@endpoint)
+            rescue => e
+              log.error("Failed to get new AWS credentials: #{e}")
+            else
+              @client_mutex.synchronize do
+                @_aws_credentials = credentials
+                @_os = nil
               end
             end
           end

[Watson1978]

fluent-plugin-opensearch/lib/fluent/plugin/out_opensearch.rb

Line 1140 in 907cb63

@_os = nil if @reconnect_on_error

I think the above assignment also requires the mutex.

[kenhys]

Now follow-up review feedback