Expire AWS credentials with a certain interval

README.md

@@ -1564,6 +1564,19 @@ In this case, the endpoint configuration looks like:
 </endpoint>
 ```
 
+### Expiring AWS credentials
+
+If you want to expire AWS credentials in certain interval, you should specify `refresh_credentials_interval` parameter under `endpoint` section:
+
+```aconf
+<endpoint>
+  url https://CLUSTER_ENDPOINT_URL
+  region eu-west-1
+  # ...
+  refresh_credentials_interval 3h # default is 5h (five hours).
+</endpoint>
+```
+
 ## Troubleshooting
 
 See [Troubleshooting document](README.Troubleshooting.md)

lib/fluent/plugin/out_opensearch.rb

@@ -194,6 +194,7 @@ def initialize(retry_stream)
       config_param :assume_role_session_name, :string, :default => "fluentd"
       config_param :assume_role_web_identity_token_file, :string, :default => nil
       config_param :sts_credentials_region, :string, :default => nil
+      config_param :refresh_credentials_interval, :time, :default => "5h"
     end
 
     config_section :buffer do
@@ -336,6 +337,22 @@ class << self
           }
         end
       end
+      # If AWS credentials is set, consider to expire credentials information forcibly before expired.
+      @credential_mutex = Mutex.new
+      if @endpoint
+        @_aws_credentials = aws_credentials(@endpoint)
+
+        if @endpoint.refresh_credentials_interval
+          timer_execute(:out_opensearch_expire_credentials, @endpoint.refresh_credentials_interval) do
+            log.debug('Recreate the AWS credentials')
+
+            @credential_mutex.synchronize do
+              @_os = nil
+              @_aws_credentials = aws_credentials(@endpoint)
+            end
+          end
+        end
+      end
 
       @serializer_class = nil
       begin
@@ -607,7 +624,7 @@ def client(host = nil, compress_connection = false)
                              :aws_sigv4,
                              service: 'es',
                              region: @endpoint.region,
-                             credentials: aws_credentials(@endpoint),
+                             credentials: @_aws_credentials,
                            )
 
                            f.adapter @http_backend, @backend_options