Merge branch 'main' into non-issue/esql-full-text-functions-disjunctions

Author: carlosdelest

docs/changelog/120250.yaml

@@ -0,0 +1,6 @@
+pr: 120250
+summary: "Retry internally when CAS upload is throttled [GCS]"
+area: Snapshot/Restore
+type: enhancement
+issues:
+ - 116546

libs/entitlement/src/main/java/org/elasticsearch/entitlement/runtime/policy/NetworkEntitlement.java

@@ -47,10 +47,10 @@ public NetworkEntitlement(List<String> actionsList) {
         for (String actionString : actionsList) {
             var action = ACTION_MAP.get(actionString);
             if (action == null) {
-                throw new IllegalArgumentException("unknown network action [" + actionString + "]");
+                throw new PolicyValidationException("unknown network action [" + actionString + "]");
             }
             if ((actionsInt & action) == action) {
-                throw new IllegalArgumentException(Strings.format("network action [%s] specified multiple times", actionString));
+                throw new PolicyValidationException(Strings.format("network action [%s] specified multiple times", actionString));
             }
             actionsInt |= action;
         }

libs/entitlement/src/main/java/org/elasticsearch/entitlement/runtime/policy/PolicyParser.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.entitlement.runtime.policy;
 
+import org.elasticsearch.xcontent.XContentLocation;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.yaml.YamlXContent;
@@ -119,6 +120,7 @@ protected Scope parseScope(String scopeName) throws IOException {
     }
 
     protected Entitlement parseEntitlement(String scopeName, String entitlementType) throws IOException {
+        XContentLocation startLocation = policyParser.getTokenLocation();
         Class<?> entitlementClass = EXTERNAL_ENTITLEMENTS.get(entitlementType);
 
         if (entitlementClass == null) {
@@ -170,7 +172,10 @@ protected Entitlement parseEntitlement(String scopeName, String entitlementType)
         try {
             return (Entitlement) entitlementConstructor.newInstance(parameterValues);
         } catch (InvocationTargetException | InstantiationException | IllegalAccessException e) {
-            throw new IllegalStateException("internal error");
+            if (e.getCause() instanceof PolicyValidationException piae) {
+                throw newPolicyParserException(startLocation, scopeName, entitlementType, piae);
+            }
+            throw new IllegalStateException("internal error", e);
         }
     }
 
@@ -191,4 +196,13 @@ protected PolicyParserException newPolicyParserException(String scopeName, Strin
             message
         );
     }
+
+    protected PolicyParserException newPolicyParserException(
+        XContentLocation location,
+        String scopeName,
+        String entitlementType,
+        PolicyValidationException cause
+    ) {
+        return PolicyParserException.newPolicyParserException(location, policyName, scopeName, entitlementType, cause);
+    }
 }

libs/entitlement/src/main/java/org/elasticsearch/entitlement/runtime/policy/PolicyParserException.java

@@ -86,7 +86,36 @@ public static PolicyParserException newPolicyParserException(
         }
     }
 
+    public static PolicyParserException newPolicyParserException(
+        XContentLocation location,
+        String policyName,
+        String scopeName,
+        String entitlementType,
+        PolicyValidationException cause
+    ) {
+        assert (scopeName != null);
+        return new PolicyParserException(
+            "["
+                + location.lineNumber()
+                + ":"
+                + location.columnNumber()
+                + "] policy parsing error for ["
+                + policyName
+                + "] in scope ["
+                + scopeName
+                + "] for entitlement type ["
+                + entitlementType
+                + "]: "
+                + cause.getMessage(),
+            cause
+        );
+    }
+
     private PolicyParserException(String message) {
         super(message);
     }
+
+    private PolicyParserException(String message, PolicyValidationException cause) {
+        super(message, cause);
+    }
 }

libs/entitlement/src/main/java/org/elasticsearch/entitlement/runtime/policy/PolicyValidationException.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.entitlement.runtime.policy;
+
+/**
+ * This exception is used to track validation errors thrown during the construction
+ * of entitlements. By using this instead of other exception types the policy
+ * parser is able to wrap this exception with a line/character number for
+ * additional useful error information.
+ */
+class PolicyValidationException extends RuntimeException {
+
+    PolicyValidationException(String message) {
+        super(message);
+    }
+
+    PolicyValidationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

libs/entitlement/src/test/java/org/elasticsearch/entitlement/runtime/policy/PolicyParserTests.java

@@ -68,6 +68,24 @@ public void testParseNetwork() throws IOException {
         assertEquals(expected, parsedPolicy);
     }
 
+    public void testParseNetworkIllegalAction() throws IOException {
+        var ex = expectThrows(PolicyParserException.class, () -> new PolicyParser(new ByteArrayInputStream("""
+            entitlement-module-name:
+              - network:
+                  actions:
+                    - listen
+                    - doesnotexist
+                    - connect
+            """.getBytes(StandardCharsets.UTF_8)), "test-policy.yaml", false).parsePolicy());
+        assertThat(
+            ex.getMessage(),
+            equalTo(
+                "[2:5] policy parsing error for [test-policy.yaml] in scope [entitlement-module-name] for entitlement type [network]: "
+                    + "unknown network action [doesnotexist]"
+            )
+        );
+    }
+
     public void testParseCreateClassloader() throws IOException {
         Policy parsedPolicy = new PolicyParser(new ByteArrayInputStream("""
             entitlement-module-name:

modules/repository-azure/src/internalClusterTest/java/org/elasticsearch/repositories/azure/AzureBlobStoreRepositoryMetricsTests.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.repositories.azure;
 
+import com.sun.net.httpserver.Headers;
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
 
@@ -21,6 +22,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.http.ResponseInjectingHttpHandler;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.repositories.RepositoriesMetrics;
 import org.elasticsearch.repositories.RepositoriesService;
@@ -46,7 +48,6 @@
 import java.util.stream.IntStream;
 
 import static org.elasticsearch.repositories.azure.AbstractAzureServerTestCase.randomBlobContent;
-import static org.elasticsearch.repositories.azure.ResponseInjectingAzureHttpHandler.createFailNRequestsHandler;
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -60,15 +61,15 @@ public class AzureBlobStoreRepositoryMetricsTests extends AzureBlobStoreReposito
     );
     private static final int MAX_RETRIES = 3;
 
-    private final Queue<ResponseInjectingAzureHttpHandler.RequestHandler> requestHandlers = new ConcurrentLinkedQueue<>();
+    private final Queue<ResponseInjectingHttpHandler.RequestHandler> requestHandlers = new ConcurrentLinkedQueue<>();
 
     @Override
     protected Map<String, HttpHandler> createHttpHandlers() {
         Map<String, HttpHandler> httpHandlers = super.createHttpHandlers();
         assert httpHandlers.size() == 1 : "This assumes there's a single handler";
         return httpHandlers.entrySet()
             .stream()
-            .collect(Collectors.toMap(Map.Entry::getKey, e -> new ResponseInjectingAzureHttpHandler(requestHandlers, e.getValue())));
+            .collect(Collectors.toMap(Map.Entry::getKey, e -> new ResponseInjectingHttpHandler(requestHandlers, e.getValue())));
     }
 
     /**
@@ -106,7 +107,7 @@ public void testThrottleResponsesAreCountedInMetrics() throws IOException {
         // Queue up some throttle responses
         final int numThrottles = randomIntBetween(1, MAX_RETRIES);
         IntStream.range(0, numThrottles)
-            .forEach(i -> requestHandlers.offer(new ResponseInjectingAzureHttpHandler.FixedRequestHandler(RestStatus.TOO_MANY_REQUESTS)));
+            .forEach(i -> requestHandlers.offer(new ResponseInjectingHttpHandler.FixedRequestHandler(RestStatus.TOO_MANY_REQUESTS)));
 
         // Check that the blob exists
         blobContainer.blobExists(purpose, blobName);
@@ -132,11 +133,7 @@ public void testRangeNotSatisfiedAreCountedInMetrics() throws IOException {
 
         // Queue up a range-not-satisfied error
         requestHandlers.offer(
-            new ResponseInjectingAzureHttpHandler.FixedRequestHandler(
-                RestStatus.REQUESTED_RANGE_NOT_SATISFIED,
-                null,
-                GET_BLOB_REQUEST_PREDICATE
-            )
+            new ResponseInjectingHttpHandler.FixedRequestHandler(RestStatus.REQUESTED_RANGE_NOT_SATISFIED, null, GET_BLOB_REQUEST_PREDICATE)
         );
 
         // Attempt to read the blob
@@ -169,7 +166,7 @@ public void testErrorResponsesAreCountedInMetrics() throws IOException {
             if (status == RestStatus.TOO_MANY_REQUESTS) {
                 throttles.incrementAndGet();
             }
-            requestHandlers.offer(new ResponseInjectingAzureHttpHandler.FixedRequestHandler(status));
+            requestHandlers.offer(new ResponseInjectingHttpHandler.FixedRequestHandler(status));
         });
 
         // Check that the blob exists
@@ -265,7 +262,7 @@ public void testBatchDeleteFailure() throws IOException {
         clearMetrics(dataNodeName);
 
         // Handler will fail one or more of the batch requests
-        final ResponseInjectingAzureHttpHandler.RequestHandler failNRequestRequestHandler = createFailNRequestsHandler(failedBatches);
+        final ResponseInjectingHttpHandler.RequestHandler failNRequestRequestHandler = createFailNRequestsHandler(failedBatches);
 
         // Exhaust the retries
         IntStream.range(0, (numberOfBatches - failedBatches) + (failedBatches * (MAX_RETRIES + 1)))
@@ -308,6 +305,35 @@ private MetricsAsserter metricsAsserter(
         return new MetricsAsserter(dataNodeName, operationPurpose, operation, repository);
     }
 
+    /**
+     * Creates a {@link ResponseInjectingHttpHandler.RequestHandler} that will persistently fail the first <code>numberToFail</code>
+     * distinct requests it sees. Any other requests are passed through to the delegate.
+     *
+     * @param numberToFail The number of requests to fail
+     * @return the handler
+     */
+    private static ResponseInjectingHttpHandler.RequestHandler createFailNRequestsHandler(int numberToFail) {
+        final List<String> requestsToFail = new ArrayList<>(numberToFail);
+        return (exchange, delegate) -> {
+            final Headers requestHeaders = exchange.getRequestHeaders();
+            final String requestId = requestHeaders.get("X-ms-client-request-id").get(0);
+            boolean failRequest = false;
+            synchronized (requestsToFail) {
+                if (requestsToFail.contains(requestId)) {
+                    failRequest = true;
+                } else if (requestsToFail.size() < numberToFail) {
+                    requestsToFail.add(requestId);
+                    failRequest = true;
+                }
+            }
+            if (failRequest) {
+                exchange.sendResponseHeaders(500, -1);
+            } else {
+                delegate.handle(exchange);
+            }
+        };
+    }
+
     private class MetricsAsserter {
         private final String dataNodeName;
         private final OperationPurpose purpose;

modules/repository-azure/src/test/java/org/elasticsearch/repositories/azure/AzureBlobContainerStatsTests.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.common.blobstore.OperationPurpose;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.http.ResponseInjectingHttpHandler;
 import org.elasticsearch.rest.RestStatus;
 import org.junit.Before;
 
@@ -34,14 +35,14 @@
 
 public class AzureBlobContainerStatsTests extends AbstractAzureServerTestCase {
 
-    private final Queue<ResponseInjectingAzureHttpHandler.RequestHandler> requestHandlers = new ConcurrentLinkedQueue<>();
+    private final Queue<ResponseInjectingHttpHandler.RequestHandler> requestHandlers = new ConcurrentLinkedQueue<>();
 
     @SuppressForbidden(reason = "use a http server")
     @Before
     public void configureAzureHandler() {
         httpServer.createContext(
             "/",
-            new ResponseInjectingAzureHttpHandler(
+            new ResponseInjectingHttpHandler(
                 requestHandlers,
                 new AzureHttpHandler(ACCOUNT, CONTAINER, null, MockAzureBlobStore.LeaseExpiryPredicate.NEVER_EXPIRE)
             )
@@ -61,7 +62,7 @@ public void testRetriesAndOperationsAreTrackedSeparately() throws IOException {
         for (int i = 0; i < randomIntBetween(10, 50); i++) {
             final boolean triggerRetry = randomBoolean();
             if (triggerRetry) {
-                requestHandlers.offer(new ResponseInjectingAzureHttpHandler.FixedRequestHandler(RestStatus.TOO_MANY_REQUESTS));
+                requestHandlers.offer(new ResponseInjectingHttpHandler.FixedRequestHandler(RestStatus.TOO_MANY_REQUESTS));
             }
             final AzureBlobStore.Operation operation = randomFrom(supportedOperations);
             switch (operation) {

modules/repository-gcs/src/internalClusterTest/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java

@@ -25,6 +25,7 @@
 import org.apache.lucene.util.BytesRefBuilder;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.BackoffPolicy;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStore;
@@ -268,7 +269,8 @@ protected GoogleCloudStorageBlobStore createBlobStore() {
                             metadata.name(),
                             storageService,
                             bigArrays,
-                            randomIntBetween(1, 8) * 1024
+                            randomIntBetween(1, 8) * 1024,
+                            BackoffPolicy.noBackoff()
                         ) {
                             @Override
                             long getLargeBlobThresholdInBytes() {