Test in third-party test

Author: nicktindall

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

@@ -15,27 +15,53 @@
 import com.google.cloud.storage.StorageException;
 
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.SecureSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.Booleans;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.repositories.AbstractThirdPartyRepositoryTestCase;
+import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
 import org.elasticsearch.rest.RestStatus;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 import org.junit.ClassRule;
 
+import java.io.InputStream;
+import java.nio.file.NoSuchFileException;
 import java.util.Base64;
 import java.util.Collection;
 
+import static org.elasticsearch.common.io.Streams.readFully;
+import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
+import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.PROXY_HOST_SETTING;
+import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.PROXY_PORT_SETTING;
+import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.PROXY_TYPE_SETTING;
 import static org.hamcrest.Matchers.blankOrNullString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.startsWith;
 
 public class GoogleCloudStorageThirdPartyTests extends AbstractThirdPartyRepositoryTestCase {
     private static final boolean USE_FIXTURE = Booleans.parseBoolean(System.getProperty("test.google.fixture", "true"));
+    private static final String PROXIED_TEST_REPO = "proxied-test-repo";
+    private static final String PROXIED_CLIENT = "proxied";
 
     @ClassRule
     public static GoogleCloudStorageHttpFixture fixture = new GoogleCloudStorageHttpFixture(USE_FIXTURE, "bucket", "o/oauth2/token");
+    private static WebProxyServer proxyServer;
+
+    @BeforeClass
+    public static void beforeClass() {
+        proxyServer = new WebProxyServer();
+    }
+
+    @AfterClass
+    public static void afterClass() throws Exception {
+        proxyServer.close();
+    }
 
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
@@ -49,8 +75,15 @@ protected Settings nodeSettings() {
         if (USE_FIXTURE) {
             builder.put("gcs.client.default.endpoint", fixture.getAddress());
             builder.put("gcs.client.default.token_uri", fixture.getAddress() + "/o/oauth2/token");
+            builder.put("gcs.client.proxied.endpoint", fixture.getAddress());
+            builder.put("gcs.client.proxied.token_uri", fixture.getAddress() + "/o/oauth2/token");
         }
 
+        // Add a proxied client so we can test resume on fail
+        builder.put(PROXY_HOST_SETTING.getConcreteSettingForNamespace(PROXIED_CLIENT).getKey(), proxyServer.getHost());
+        builder.put(PROXY_PORT_SETTING.getConcreteSettingForNamespace(PROXIED_CLIENT).getKey(), proxyServer.getPort());
+        builder.put(PROXY_TYPE_SETTING.getConcreteSettingForNamespace(PROXIED_CLIENT).getKey(), "http");
+
         return builder.build();
     }
 
@@ -64,17 +97,26 @@ protected SecureSettings credentials() {
         MockSecureSettings secureSettings = new MockSecureSettings();
         if (USE_FIXTURE) {
             secureSettings.setFile("gcs.client.default.credentials_file", TestUtils.createServiceAccount(random()));
+            secureSettings.setFile("gcs.client.proxied.credentials_file", TestUtils.createServiceAccount(random()));
         } else {
             secureSettings.setFile(
                 "gcs.client.default.credentials_file",
                 Base64.getDecoder().decode(System.getProperty("test.google.account"))
             );
+            secureSettings.setFile(
+                "gcs.client.proxied.credentials_file",
+                Base64.getDecoder().decode(System.getProperty("test.google.account"))
+            );
         }
         return secureSettings;
     }
 
     @Override
     protected void createRepository(final String repoName) {
+        createRepository(repoName, "default");
+    }
+
+    private void createRepository(final String repoName, String clientName) {
         AcknowledgedResponse putRepositoryResponse = clusterAdmin().preparePutRepository(
             TEST_REQUEST_TIMEOUT,
             TEST_REQUEST_TIMEOUT,
@@ -85,6 +127,7 @@ protected void createRepository(final String repoName) {
                 Settings.builder()
                     .put("bucket", System.getProperty("test.google.bucket"))
                     .put("base_path", System.getProperty("test.google.base", "/"))
+                    .put("client", clientName)
             )
             .get();
         assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
@@ -95,4 +138,46 @@ public void testReadFromPositionLargerThanBlobLength() {
             e -> asInstanceOf(StorageException.class, e.getCause()).getCode() == RestStatus.REQUESTED_RANGE_NOT_SATISFIED.getStatus()
         );
     }
+
+    public void testResumeAfterUpdate() {
+        createRepository(PROXIED_TEST_REPO, PROXIED_CLIENT);
+
+        // The blob needs to be large enough that it won't be entirely buffered on the first request
+        final int enoughBytesToNotBeEntirelyBuffered = Math.toIntExact(ByteSizeValue.ofMb(10).getBytes());
+
+        final BlobStoreRepository repo = getRepository(PROXIED_TEST_REPO);
+        final String blobKey = randomIdentifier();
+        final byte[] initialValue = randomByteArrayOfLength(enoughBytesToNotBeEntirelyBuffered);
+        executeOnBlobStore(repo, container -> {
+            container.writeBlob(randomPurpose(), blobKey, new BytesArray(initialValue), true);
+
+            try (InputStream inputStream = container.readBlob(randomPurpose(), blobKey)) {
+                // Trigger the first request for the blob, partially read it
+                int read = inputStream.read();
+                assert read != -1;
+
+                // Restart the server (this triggers a retry)
+                proxyServer.restart();
+
+                // Update the file
+                byte[] updatedValue = randomByteArrayOfLength(enoughBytesToNotBeEntirelyBuffered);
+                container.writeBlob(randomPurpose(), blobKey, new BytesArray(updatedValue), false);
+
+                // Read the rest of the stream, it should throw because the contents changed
+                String message = assertThrows(NoSuchFileException.class, () -> readFully(inputStream)).getMessage();
+                assertThat(
+                    message,
+                    startsWith(
+                        "Blob object ["
+                            + container.path().buildAsString()
+                            + blobKey
+                            + "] generation [1] unavailable on resume (contents changed, or object deleted):"
+                    )
+                );
+            } catch (Exception e) {
+                fail(e);
+            }
+            return null;
+        });
+    }
 }

modules/repository-gcs/src/test/java/org/elasticsearch/repositories/gcs/MockHttpProxyServer.java

@@ -26,19 +26,16 @@
  */
 abstract class MockHttpProxyServer implements Closeable {
 
-    private final HttpServer httpServer;
+    private HttpServer httpServer;
 
     MockHttpProxyServer() {
-        httpServer = ServerBootstrap.bootstrap()
-            .setLocalAddress(InetAddress.getLoopbackAddress())
-            .setListenerPort(0)
-            .registerHandler("*", this::handle)
-            .create();
-        try {
-            httpServer.start();
-        } catch (IOException e) {
-            throw new RuntimeException("Unable to start HTTP proxy server", e);
-        }
+        startHttpServer(0);
+    }
+
+    public void restart() {
+        int originalPort = httpServer.getLocalPort();
+        httpServer.shutdown(0, TimeUnit.SECONDS);
+        startHttpServer(originalPort);
     }
 
     public abstract void handle(HttpRequest request, HttpResponse response, HttpContext context) throws HttpException, IOException;
@@ -51,6 +48,19 @@ String getHost() {
         return NetworkAddress.format(httpServer.getInetAddress());
     }
 
+    private void startHttpServer(int portNumber) {
+        httpServer = ServerBootstrap.bootstrap()
+            .setLocalAddress(InetAddress.getLoopbackAddress())
+            .setListenerPort(portNumber)
+            .registerHandler("*", this::handle)
+            .create();
+        try {
+            httpServer.start();
+        } catch (IOException e) {
+            throw new RuntimeException("Unable to start HTTP proxy server", e);
+        }
+    }
+
     @Override
     public void close() throws IOException {
         httpServer.shutdown(10, TimeUnit.SECONDS);

test/framework/src/main/java/org/elasticsearch/repositories/AbstractThirdPartyRepositoryTestCase.java

@@ -465,6 +465,10 @@ private Set<String> listChildren(BlobPath path) {
     }
 
     protected BlobStoreRepository getRepository() {
-        return (BlobStoreRepository) getInstanceFromNode(RepositoriesService.class).repository(TEST_REPO_NAME);
+        return getRepository(TEST_REPO_NAME);
+    }
+
+    protected BlobStoreRepository getRepository(String name) {
+        return (BlobStoreRepository) getInstanceFromNode(RepositoriesService.class).repository(name);
     }
 }