Merge pull request #88 from fluent/handle-opensearch-serverless

out_opensearch: Provide service_name choices for handling serverless
fluent · Nov 24, 2022 · 5a95b8f · 5a95b8f
2 parents f364806 + e8da9d2
commit 5a95b8f
Show file tree

Hide file tree

Showing 3 changed files with 48 additions and 1 deletion.
diff --git a/README.md b/README.md
@@ -1577,6 +1577,19 @@ If you want to expire AWS credentials in certain interval, you should specify `r
 </endpoint>
 ```
 
+### Use OpenSearch Serverless
+
+If you want to use Serverless version of OpenSearch service, you have to specify `aoss` in `aws_service_name` under `endpoint` section:
+
+```aconf
+<endpoint>
+  url https://CLUSTER_ENDPOINT_URL
+  region us-east-2
+  # ...
+  aws_service_name aoss # default is es that is for AWS OpenSearch Service not Serverless.
+</endpoint>
+```
+
 ## Troubleshooting
 
 See [Troubleshooting document](README.Troubleshooting.md)

diff --git a/lib/fluent/plugin/out_opensearch.rb b/lib/fluent/plugin/out_opensearch.rb
@@ -195,6 +195,7 @@ def initialize(retry_stream)
       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"
+      config_param :aws_service_name, :enum, list: [:es, :aoss], :default => :es
     end
 
     config_section :buffer do
@@ -622,7 +623,7 @@ def client(host = nil, compress_connection = false)
                          lambda do |f|
                            f.request(
                              :aws_sigv4,
-                             service: 'es',
+                             service: @endpoint.aws_service_name.to_s,
                              region: @endpoint.region,
                              credentials: @_aws_credentials,
                            )

diff --git a/test/plugin/test_out_opensearch.rb b/test/plugin/test_out_opensearch.rb
@@ -315,6 +315,39 @@ def test_configure
     assert_equal "fluentd", instance.endpoint.assume_role_session_name
     assert_nil instance.endpoint.assume_role_web_identity_token_file
     assert_nil instance.endpoint.sts_credentials_region
+    assert_equal :es, instance.endpoint.aws_service_name
+  end
+
+  data("OpenSearch Service" => [:es, 'es'],
+       "OpenSearch Serverless" => [:aoss, 'aoss'])
+  test 'configure endpoint section w/ aws_service_name' do |data|
+    expected, conf = data
+    config = Fluent::Config::Element.new(
+      'ROOT', '', {
+        '@type' => 'opensearch',
+      }, [
+        Fluent::Config::Element.new('endpoint', '', {
+                                      'url' => "https://search-opensearch.aws.example.com/",
+                                      'region' => "local",
+                                      'access_key_id' => 'YOUR_AWESOME_KEY',
+                                      'secret_access_key' => 'YOUR_AWESOME_SECRET',
+                                      'aws_service_name' => conf,
+                                    }, []),
+        Fluent::Config::Element.new('buffer', 'tag', {}, [])
+
+      ])
+    instance = driver(config).instance
+
+    assert_equal "https://search-opensearch.aws.example.com", instance.endpoint.url
+    assert_equal "local", instance.endpoint.region
+    assert_equal "YOUR_AWESOME_KEY", instance.endpoint.access_key_id
+    assert_equal "YOUR_AWESOME_SECRET", instance.endpoint.secret_access_key
+    assert_nil instance.endpoint.assume_role_arn
+    assert_nil instance.endpoint.ecs_container_credentials_relative_uri
+    assert_equal "fluentd", instance.endpoint.assume_role_session_name
+    assert_nil instance.endpoint.assume_role_web_identity_token_file
+    assert_nil instance.endpoint.sts_credentials_region
+    assert_equal expected, instance.endpoint.aws_service_name
   end
 
   test 'configure compression' do