[C++][Python] Support per-filesystem S3 User-Agent

cpp/src/arrow/filesystem/s3fs.cc

@@ -489,6 +489,7 @@ bool S3Options::Equals(const S3Options& other) const {
           connect_timeout == other.connect_timeout &&
           request_timeout == other.request_timeout &&
           endpoint_override == other.endpoint_override && scheme == other.scheme &&
+          user_agent == other.user_agent &&
           role_arn == other.role_arn && session_name == other.session_name &&
           external_id == other.external_id && load_frequency == other.load_frequency &&
           proxy_options.Equals(other.proxy_options) &&
@@ -1161,6 +1162,9 @@ class ClientBuilder {
           static_cast<long>(ceil(options_.connect_timeout * 1000));  // NOLINT runtime/int
     }
 
+    if (!options_.user_agent.empty()) {
+      client_config_.userAgent = ToAwsString(options_.user_agent);
+    }
     client_config_.endpointOverride = ToAwsString(options_.endpoint_override);
     if (options_.scheme == "http") {
       client_config_.scheme = Aws::Http::Scheme::HTTP;

cpp/src/arrow/filesystem/s3fs.h

@@ -130,6 +130,10 @@ struct ARROW_S3_EXPORT S3Options {
   /// S3 connection transport, default "https"
   std::string scheme = "https";
 
+  /// Optional User-Agent override for this filesystem's S3 requests.
+  /// If empty, use the AWS SDK default.
+  std::string user_agent;
+
   /// ARN of role to assume
   std::string role_arn;
   /// Optional identifier for an assumed role session.

python/pyarrow/_s3fs.pyx

@@ -263,6 +263,10 @@ cdef class S3FileSystem(FileSystem):
         If true, then virtual addressing is always enabled.
         If false, then virtual addressing is only enabled if `endpoint_override` is empty.
         This can be used for non-AWS backends that only support virtual hosted-style access.
+    user_agent : str, default None
+        Override the User-Agent for this filesystem's S3 requests.
+        If None or empty, use the AWS SDK default. This setting does not
+        affect other filesystem instances.
     tls_ca_file_path : str, default None
         If set, this should be the path of a file containing TLS certificates
         in PEM format which will be used for TLS verification.
@@ -293,7 +297,8 @@ cdef class S3FileSystem(FileSystem):
                  check_directory_existence_before_creation=False,
                  retry_strategy: S3RetryStrategy = AwsStandardS3RetryStrategy(
                      max_attempts=3),
-                 force_virtual_addressing=False, tls_ca_file_path=None):
+                 force_virtual_addressing=False, tls_ca_file_path=None,
+                 user_agent=None):
         cdef:
             optional[CS3Options] options
             shared_ptr[CS3FileSystem] wrapped
@@ -407,6 +412,8 @@ cdef class S3FileSystem(FileSystem):
         options.value().allow_bucket_creation = allow_bucket_creation
         options.value().allow_bucket_deletion = allow_bucket_deletion
         options.value().check_directory_existence_before_creation = check_directory_existence_before_creation
+        if user_agent is not None:
+            options.value().user_agent = tobytes(user_agent)
         options.value().force_virtual_addressing = force_virtual_addressing
 
         if isinstance(retry_strategy, AwsStandardS3RetryStrategy):
@@ -478,6 +485,7 @@ cdef class S3FileSystem(FileSystem):
                                    opts.proxy_options.username),
                                'password': frombytes(
                                    opts.proxy_options.password)},
+                user_agent=frombytes(opts.user_agent),
                 force_virtual_addressing=opts.force_virtual_addressing,
                 tls_ca_file_path=frombytes(opts.tls_ca_file_path),
             ),)

python/pyarrow/includes/libarrow_fs.pxd

@@ -159,6 +159,7 @@ cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
         c_bool allow_bucket_creation
         c_bool allow_bucket_deletion
         c_bool check_directory_existence_before_creation
+        c_string user_agent
         c_bool force_virtual_addressing
         c_string tls_ca_file_path
         shared_ptr[const CKeyValueMetadata] default_metadata

python/pyarrow/tests/test_fs.py

@@ -1725,6 +1725,64 @@ def test_filesystem_from_path_object(path):
     assert path == p.resolve().absolute().as_posix()
 
 
+@pytest.mark.s3
+@pytest.mark.parametrize("pickle_roundtrip", [False, True])
+def test_s3fs_user_agent(pickle_roundtrip, pickle_module):
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+    from threading import Thread
+    from pyarrow.fs import S3FileSystem
+
+    user_agents = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_HEAD(self):
+            user_agents.append(self.headers.get("User-Agent"))
+            self.send_response(200)
+            self.send_header("Content-Length", "3")
+            self.end_headers()
+
+        def do_GET(self):
+            self.do_HEAD()
+            self.wfile.write(b"abc")
+
+        def log_message(self, *args):
+            pass
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        options = dict(anonymous=True, region="us-east-1", scheme="http",
+                       endpoint_override=f"127.0.0.1:{server.server_port}")
+        first = S3FileSystem(**options, user_agent="first-app/1.0")
+        second = S3FileSystem(**options, user_agent="second-app/2.0")
+        default = S3FileSystem(**options)
+        assert first != second
+        assert first != default
+        assert S3FileSystem(**options, user_agent="") == default
+        if pickle_roundtrip:
+            restored = pickle_module.loads(pickle_module.dumps(first))
+            assert restored == first
+            first = restored
+        for filesystem, expected in [(first, "first-app/1.0"),
+                                     (second, "second-app/2.0"),
+                                     (first, "first-app/1.0"),
+                                     (default, None)]:
+            user_agents.clear()
+            with filesystem.open_input_file("bucket/object") as stream:
+                assert stream.read() == b"abc"
+            assert user_agents
+            if expected:
+                assert all(value == expected for value in user_agents)
+            else:
+                assert all("first-app" not in value and "second-app" not in value
+                           for value in user_agents)
+    finally:
+        server.shutdown()
+        thread.join()
+        server.server_close()
+
+
 @pytest.mark.s3
 def test_filesystem_from_uri_s3(s3_server):
     from pyarrow.fs import S3FileSystem