Cloud Plugin: Gateway should store meta data and indices under the same container, closes #180.

Author: kimchy

.idea/dictionaries/kimchy.xml

@@ -29,8 +29,6 @@
  */
 public class JCloudsUtils {
 
-    public static final String BLOB_CONTAINER_SEP = "-";
-
     public static Iterable<? extends Module> buildModules(Settings settings) {
         return ImmutableList.of(new JCloudsLoggingModule(settings));
     }

.idea/modules/plugins-cloud.iml

@@ -22,7 +22,6 @@
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.cloud.blobstore.CloudBlobStoreService;
-import org.elasticsearch.cloud.jclouds.JCloudsUtils;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.gateway.Gateway;
@@ -48,6 +47,8 @@
 import java.io.IOException;
 import java.util.Set;
 
+import static org.jclouds.blobstore.options.ListContainerOptions.Builder.*;
+
 /**
  * @author kimchy (shay.banon)
  */
@@ -60,9 +61,9 @@ public class CloudGateway extends AbstractLifecycleComponent<Gateway> implements
 
     private final Location location;
 
-    private final SizeValue chunkSize;
+    private final String metaDataDirectory;
 
-    private final String metadataContainer;
+    private final SizeValue chunkSize;
 
     private volatile int currentIndex;
 
@@ -90,17 +91,13 @@ public class CloudGateway extends AbstractLifecycleComponent<Gateway> implements
             }
         }
 
-        String container = componentSettings.get("container");
+        this.container = componentSettings.get("container");
         if (container == null) {
             throw new ElasticSearchIllegalArgumentException("Cloud gateway requires 'container' setting");
         }
-        this.container = container + JCloudsUtils.BLOB_CONTAINER_SEP + clusterName.value();
-
-        this.metadataContainer = this.container + JCloudsUtils.BLOB_CONTAINER_SEP + "metadata";
-
-        logger.debug("Using location [{}], container [{}], metadata_container [{}]", this.location, this.container, metadataContainer);
-
-        blobStoreContext.getBlobStore().createContainerInLocation(this.location, metadataContainer);
+        this.metaDataDirectory = clusterName.value() + "/metadata";
+        logger.debug("Using location [{}], container [{}], metadata_directory [{}]", this.location, this.container, metaDataDirectory);
+        blobStoreContext.getBlobStore().createContainerInLocation(this.location, container);
 
         this.currentIndex = findLatestIndex();
         logger.debug("Latest metadata found at index [" + currentIndex + "]");
@@ -129,7 +126,7 @@ public SizeValue chunkSize() {
 
     @Override public void write(MetaData metaData) throws GatewayException {
         try {
-            String name = "metadata-" + (currentIndex + 1);
+            String name = metaDataDirectory + "/metadata-" + (currentIndex + 1);
 
             BinaryXContentBuilder builder = XContentFactory.contentBinaryBuilder(XContentType.JSON);
             builder.prettyPrint();
@@ -141,14 +138,14 @@ public SizeValue chunkSize() {
             blob.setPayload(new FastByteArrayInputStream(builder.unsafeBytes(), 0, builder.unsafeBytesLength()));
             blob.setContentLength(builder.unsafeBytesLength());
 
-            blobStoreContext.getBlobStore().putBlob(metadataContainer, blob);
+            blobStoreContext.getBlobStore().putBlob(container, blob);
 
             currentIndex++;
 
-            PageSet<? extends StorageMetadata> pageSet = blobStoreContext.getBlobStore().list(metadataContainer);
+            PageSet<? extends StorageMetadata> pageSet = blobStoreContext.getBlobStore().list(container, inDirectory(metaDataDirectory));
             for (StorageMetadata storageMetadata : pageSet) {
-                if (storageMetadata.getName().startsWith("metadata-") && !name.equals(storageMetadata.getName())) {
-                    blobStoreContext.getAsyncBlobStore().removeBlob(metadataContainer, storageMetadata.getName());
+                if (storageMetadata.getName().contains("metadata-") && !name.equals(storageMetadata.getName())) {
+                    blobStoreContext.getAsyncBlobStore().removeBlob(container, storageMetadata.getName());
                 }
             }
         } catch (IOException e) {
@@ -161,7 +158,7 @@ public SizeValue chunkSize() {
             if (currentIndex == -1)
                 return null;
 
-            return readMetaData("metadata-" + currentIndex);
+            return readMetaData(metaDataDirectory + "/metadata-" + currentIndex);
         } catch (GatewayException e) {
             throw e;
         } catch (Exception e) {
@@ -174,26 +171,26 @@ public SizeValue chunkSize() {
     }
 
     @Override public void reset() {
-        PageSet<? extends StorageMetadata> pageSet = blobStoreContext.getBlobStore().list(metadataContainer);
+        PageSet<? extends StorageMetadata> pageSet = blobStoreContext.getBlobStore().list(container, inDirectory(metaDataDirectory));
         for (StorageMetadata storageMetadata : pageSet) {
-            if (storageMetadata.getName().startsWith("metadata-")) {
-                blobStoreContext.getBlobStore().removeBlob(metadataContainer, storageMetadata.getName());
+            if (storageMetadata.getName().contains("metadata-")) {
+                blobStoreContext.getBlobStore().removeBlob(container, storageMetadata.getName());
             }
         }
         currentIndex = -1;
     }
 
     private int findLatestIndex() {
         int index = -1;
-        PageSet<? extends StorageMetadata> pageSet = blobStoreContext.getBlobStore().list(metadataContainer);
+        PageSet<? extends StorageMetadata> pageSet = blobStoreContext.getBlobStore().list(container, inDirectory(metaDataDirectory).maxResults(1000));
         for (StorageMetadata storageMetadata : pageSet) {
             if (logger.isTraceEnabled()) {
                 logger.trace("[findLatestMetadata]: Processing blob [" + storageMetadata.getName() + "]");
             }
-            if (!storageMetadata.getName().startsWith("metadata-")) {
+            if (!storageMetadata.getName().contains("metadata-")) {
                 continue;
             }
-            int fileIndex = Integer.parseInt(storageMetadata.getName().substring(storageMetadata.getName().indexOf('-') + 1));
+            int fileIndex = Integer.parseInt(storageMetadata.getName().substring(storageMetadata.getName().lastIndexOf('-') + 1));
             if (fileIndex >= index) {
                 // try and read the meta data
                 try {
@@ -210,7 +207,7 @@ private int findLatestIndex() {
     private MetaData readMetaData(String name) throws IOException {
         XContentParser parser = null;
         try {
-            Blob blob = blobStoreContext.getBlobStore().getBlob(metadataContainer, name);
+            Blob blob = blobStoreContext.getBlobStore().getBlob(container, name);
             parser = XContentFactory.xContent(XContentType.JSON).createParser(blob.getContent());
             return MetaData.Builder.fromXContent(parser, settings);
         } finally {

plugins/cloud/src/main/java/org/elasticsearch/cloud/jclouds/JCloudsUtils.java

@@ -22,7 +22,7 @@
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.cloud.blobstore.CloudBlobStoreService;
-import org.elasticsearch.cloud.jclouds.JCloudsUtils;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.gateway.Gateway;
 import org.elasticsearch.gateway.cloud.CloudGateway;
 import org.elasticsearch.index.AbstractIndexComponent;
@@ -48,13 +48,15 @@ public class CloudIndexGateway extends AbstractIndexComponent implements IndexGa
 
     private final String indexContainer;
 
+    private final String indexDirectory;
+
     private final Location location;
 
     private final SizeValue chunkSize;
 
     private final BlobStoreContext blobStoreContext;
 
-    @Inject public CloudIndexGateway(Index index, @IndexSettings Settings indexSettings, CloudBlobStoreService blobStoreService, Gateway gateway) {
+    @Inject public CloudIndexGateway(Index index, @IndexSettings Settings indexSettings, ClusterName clusterName, CloudBlobStoreService blobStoreService, Gateway gateway) {
         super(index, indexSettings);
         this.blobStoreContext = blobStoreService.context();
         this.gateway = gateway;
@@ -66,7 +68,7 @@ public class CloudIndexGateway extends AbstractIndexComponent implements IndexGa
         if (gateway instanceof CloudGateway) {
             CloudGateway cloudGateway = (CloudGateway) gateway;
             if (container == null) {
-                container = cloudGateway.container() + JCloudsUtils.BLOB_CONTAINER_SEP + index.name();
+                container = cloudGateway.container();
             }
             if (chunkSize == null) {
                 chunkSize = cloudGateway.chunkSize();
@@ -99,11 +101,10 @@ public class CloudIndexGateway extends AbstractIndexComponent implements IndexGa
             }
         }
         this.indexContainer = container;
+        this.indexDirectory = clusterName.value() + "/" + index.name();
         this.chunkSize = chunkSize;
 
-        logger.debug("Using location [{}], container [{}], chunk_size [{}]", this.location, this.indexContainer, this.chunkSize);
-
-//        blobStoreContext.getBlobStore().createContainerInLocation(this.location, this.indexContainer);
+        logger.debug("Using location [{}], container [{}], index_directory [{}], chunk_size [{}]", this.location, this.indexContainer, this.indexDirectory, this.chunkSize);
     }
 
     public Location indexLocation() {
@@ -114,6 +115,10 @@ public String indexContainer() {
         return this.indexContainer;
     }
 
+    public String indexDirectory() {
+        return this.indexDirectory;
+    }
+
     public SizeValue chunkSize() {
         return this.chunkSize;
     }
@@ -126,6 +131,5 @@ public SizeValue chunkSize() {
         if (!delete) {
             return;
         }
-//        blobStoreContext.getBlobStore().deleteContainer(indexContainer);
     }
 }