Merge branch 'main' into refactor/streaming-metrics

Author: prwhelan

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3ClientSettings.java

@@ -121,28 +121,28 @@ final class S3ClientSettings {
     static final Setting.AffixSetting<TimeValue> READ_TIMEOUT_SETTING = Setting.affixKeySetting(
         PREFIX,
         "read_timeout",
-        key -> Setting.timeSetting(key, TimeValue.timeValueMillis(ClientConfiguration.DEFAULT_SOCKET_TIMEOUT), Property.NodeScope)
+        key -> Setting.timeSetting(key, Defaults.READ_TIMEOUT, Property.NodeScope)
     );
 
     /** The maximum number of concurrent connections to use. */
     static final Setting.AffixSetting<Integer> MAX_CONNECTIONS_SETTING = Setting.affixKeySetting(
         PREFIX,
         "max_connections",
-        key -> Setting.intSetting(key, ClientConfiguration.DEFAULT_MAX_CONNECTIONS, 1, Property.NodeScope)
+        key -> Setting.intSetting(key, Defaults.MAX_CONNECTIONS, 1, Property.NodeScope)
     );
 
     /** The number of retries to use when an s3 request fails. */
     static final Setting.AffixSetting<Integer> MAX_RETRIES_SETTING = Setting.affixKeySetting(
         PREFIX,
         "max_retries",
-        key -> Setting.intSetting(key, ClientConfiguration.DEFAULT_RETRY_POLICY.getMaxErrorRetry(), 0, Property.NodeScope)
+        key -> Setting.intSetting(key, Defaults.RETRY_COUNT, 0, Property.NodeScope)
     );
 
     /** Whether retries should be throttled (ie use backoff). */
     static final Setting.AffixSetting<Boolean> USE_THROTTLE_RETRIES_SETTING = Setting.affixKeySetting(
         PREFIX,
         "use_throttle_retries",
-        key -> Setting.boolSetting(key, ClientConfiguration.DEFAULT_THROTTLE_RETRIES, Property.NodeScope)
+        key -> Setting.boolSetting(key, Defaults.THROTTLE_RETRIES, Property.NodeScope)
     );
 
     /** Whether the s3 client should use path style access. */
@@ -335,7 +335,7 @@ S3ClientSettings refine(Settings repositorySettings) {
 
     /**
      * Load all client settings from the given settings.
-     *
+     * <p>
      * Note this will always at least return a client named "default".
      */
     static Map<String, S3ClientSettings> load(Settings settings) {
@@ -501,4 +501,11 @@ private static <T> T getRepoSettingOrDefault(Setting.AffixSetting<T> setting, Se
         }
         return defaultValue;
     }
+
+    static final class Defaults {
+        static final TimeValue READ_TIMEOUT = TimeValue.timeValueMillis(ClientConfiguration.DEFAULT_SOCKET_TIMEOUT);
+        static final int MAX_CONNECTIONS = ClientConfiguration.DEFAULT_MAX_CONNECTIONS;
+        static final int RETRY_COUNT = ClientConfiguration.DEFAULT_RETRY_POLICY.getMaxErrorRetry();
+        static final boolean THROTTLE_RETRIES = ClientConfiguration.DEFAULT_THROTTLE_RETRIES;
+    }
 }

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Service.java

@@ -30,7 +30,6 @@
 import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClient;
 import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClientBuilder;
 
-import org.apache.http.HttpStatus;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchException;
@@ -44,6 +43,7 @@
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.watcher.FileChangesListener;
 import org.elasticsearch.watcher.FileWatcher;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -115,8 +115,8 @@ class S3Service implements Closeable {
 
     /**
      * Refreshes the settings for the AmazonS3 clients and clears the cache of
-     * existing clients. New clients will be build using these new settings. Old
-     * clients are usable until released. On release they will be destroyed instead
+     * existing clients. New clients will be built using these new settings. Old
+     * clients are usable until released. On release, they will be destroyed instead
      * of being returned to the cache.
      */
     public synchronized void refreshAndClearCache(Map<String, S3ClientSettings> clientsSettings) {
@@ -126,7 +126,7 @@ public synchronized void refreshAndClearCache(Map<String, S3ClientSettings> clie
         this.staticClientSettings = Maps.ofEntries(clientsSettings.entrySet());
         derivedClientSettings = emptyMap();
         assert this.staticClientSettings.containsKey("default") : "always at least have 'default'";
-        // clients are built lazily by {@link client}
+        /* clients are built lazily by {@link #client} */
     }
 
     /**
@@ -341,7 +341,8 @@ public void refresh() {
      * <ul>
      * <li>Reads the location of the web identity token not from AWS_WEB_IDENTITY_TOKEN_FILE, but from a symlink
      * in the plugin directory, so we don't need to create a hardcoded read file permission for the plugin.</li>
-     * <li>Supports customization of the STS endpoint via a system property, so we can test it against a test fixture.</li>
+     * <li>Supports customization of the STS (Security Token Service) endpoint via a system property, so we can
+     * test it against a test fixture.</li>
      * <li>Supports gracefully shutting down the provider and the STS client.</li>
      * </ul>
      */
@@ -384,7 +385,7 @@ static class CustomWebIdentityTokenCredentialsProvider implements AWSCredentials
             if (roleArn == null) {
                 LOGGER.warn(
                     "Unable to use a web identity token for authentication. The AWS_WEB_IDENTITY_TOKEN_FILE environment "
-                        + "variable is set, but either AWS_ROLE_ARN is missing"
+                        + "variable is set, but AWS_ROLE_ARN is missing"
                 );
                 return;
             }
@@ -528,7 +529,7 @@ interface JvmEnvironment {
                 return true;
             }
             if (exception instanceof AmazonServiceException ase) {
-                return ase.getStatusCode() == HttpStatus.SC_FORBIDDEN && "InvalidAccessKeyId".equals(ase.getErrorCode());
+                return ase.getStatusCode() == RestStatus.FORBIDDEN.getStatus() && "InvalidAccessKeyId".equals(ase.getErrorCode());
             }
             return false;
         })

server/src/main/java/org/elasticsearch/transport/OutboundHandler.java

@@ -31,7 +31,6 @@
 import org.elasticsearch.common.recycler.Recycler;
 import org.elasticsearch.common.transport.NetworkExceptionHelper;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
-import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
@@ -116,6 +115,7 @@ void sendRequest(
         assert assertValidTransportVersion(transportVersion);
         sendMessage(
             channel,
+            MessageDirection.REQUEST,
             action,
             request,
             requestId,
@@ -148,7 +148,8 @@ void sendResponse(
         try {
             sendMessage(
                 channel,
-                null,
+                MessageDirection.RESPONSE,
+                action,
                 response,
                 requestId,
                 isHandshake,
@@ -190,7 +191,8 @@ void sendErrorResponse(
         try {
             sendMessage(
                 channel,
-                null,
+                MessageDirection.RESPONSE_ERROR,
+                action,
                 msg,
                 requestId,
                 false,
@@ -206,29 +208,36 @@ void sendErrorResponse(
         }
     }
 
+    public enum MessageDirection {
+        REQUEST,
+        RESPONSE,
+        RESPONSE_ERROR
+    }
+
     private void sendMessage(
         TcpChannel channel,
-        @Nullable String requestAction,
+        MessageDirection messageDirection,
+        String action,
         Writeable writeable,
         long requestId,
         boolean isHandshake,
-        Compression.Scheme compressionScheme,
+        Compression.Scheme possibleCompressionScheme,
         TransportVersion version,
         ResponseStatsConsumer responseStatsConsumer,
         Releasable onAfter
     ) throws IOException {
-        compressionScheme = writeable instanceof BytesTransportRequest ? null : compressionScheme;
+        assert action != null;
+        final var compressionScheme = writeable instanceof BytesTransportRequest ? null : possibleCompressionScheme;
         final BytesReference message;
         boolean serializeSuccess = false;
-        final boolean isError = writeable instanceof RemoteTransportException;
         final RecyclerBytesStreamOutput byteStreamOutput = new RecyclerBytesStreamOutput(recycler);
         try {
             message = serialize(
-                requestAction,
+                messageDirection,
+                action,
                 requestId,
                 isHandshake,
                 version,
-                isError,
                 compressionScheme,
                 writeable,
                 threadPool.getThreadContext(),
@@ -244,14 +253,23 @@ private void sendMessage(
             }
         }
         responseStatsConsumer.addResponseStats(message.length());
-        final var responseType = writeable.getClass();
-        final boolean compress = compressionScheme != null;
+        final var messageType = writeable.getClass();
         internalSend(
             channel,
             message,
-            requestAction == null
-                ? () -> "Response{" + requestId + "}{" + isError + "}{" + compress + "}{" + isHandshake + "}{" + responseType + "}"
-                : () -> "Request{" + requestAction + "}{" + requestId + "}{" + isError + "}{" + compress + "}{" + isHandshake + "}",
+            () -> (messageDirection == MessageDirection.REQUEST ? "Request{" : "Response{")
+                + action
+                + "}{id="
+                + requestId
+                + "}{err="
+                + (messageDirection == MessageDirection.RESPONSE_ERROR)
+                + "}{cs="
+                + compressionScheme
+                + "}{hs="
+                + isHandshake
+                + "}{t="
+                + messageType
+                + "}",
             ActionListener.releasing(
                 message instanceof ReleasableBytesReference r
                     ? Releasables.wrap(byteStreamOutput, onAfter, r)
@@ -262,38 +280,39 @@ private void sendMessage(
 
     // public for tests
     public static BytesReference serialize(
-        @Nullable String requestAction,
+        MessageDirection messageDirection,
+        String action,
         long requestId,
         boolean isHandshake,
         TransportVersion version,
-        boolean isError,
         Compression.Scheme compressionScheme,
         Writeable writeable,
         ThreadContext threadContext,
         RecyclerBytesStreamOutput byteStreamOutput
     ) throws IOException {
+        assert action != null;
         assert byteStreamOutput.position() == 0;
         byteStreamOutput.setTransportVersion(version);
         byteStreamOutput.skip(TcpHeader.HEADER_SIZE);
         threadContext.writeTo(byteStreamOutput);
-        if (requestAction != null) {
+        if (messageDirection == MessageDirection.REQUEST) {
             if (version.before(TransportVersions.V_8_0_0)) {
                 // empty features array
                 byteStreamOutput.writeStringArray(Strings.EMPTY_ARRAY);
             }
-            byteStreamOutput.writeString(requestAction);
+            byteStreamOutput.writeString(action);
         }
 
         final int variableHeaderLength = Math.toIntExact(byteStreamOutput.position() - TcpHeader.HEADER_SIZE);
         BytesReference message = serializeMessageBody(writeable, compressionScheme, version, byteStreamOutput);
         byte status = 0;
-        if (requestAction == null) {
+        if (messageDirection != MessageDirection.REQUEST) {
             status = TransportStatus.setResponse(status);
         }
         if (isHandshake) {
             status = TransportStatus.setHandshake(status);
         }
-        if (isError) {
+        if (messageDirection == MessageDirection.RESPONSE_ERROR) {
             status = TransportStatus.setError(status);
         }
         if (compressionScheme != null) {
@@ -316,6 +335,8 @@ private static BytesReference serializeMessageBody(
         try {
             stream.setTransportVersion(version);
             if (writeable instanceof BytesTransportRequest bRequest) {
+                assert stream == byteStreamOutput;
+                assert compressionScheme == null;
                 bRequest.writeThin(stream);
                 zeroCopyBuffer = bRequest.bytes;
             } else if (writeable instanceof RemoteTransportException remoteTransportException) {