Continue S3 -> Cassandra shredding, add logging, mark things TODO.

Author: Tom May

plugins/cassandra/src/main/java/org/elasticsearch/cassandra/blobstore/AbstractCassandraBlobContainer.java

@@ -19,14 +19,13 @@
 
 package org.elasticsearch.cassandra.blobstore;
 
-//import com.amazonaws.services.s3.model.ObjectListing;
-//import com.amazonaws.services.s3.model.S3Object;
-//import com.amazonaws.services.s3.model.S3ObjectSummary;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
 import org.elasticsearch.common.blobstore.support.PlainBlobMetaData;
 import org.elasticsearch.common.collect.ImmutableMap;
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.common.logging.Loggers;
 
 import javax.annotation.Nullable;
 import java.io.IOException;
@@ -37,31 +36,35 @@
  */
 public class AbstractCassandraBlobContainer extends AbstractBlobContainer {
 
+    protected final ESLogger logger = Loggers.getLogger(getClass());
+
     protected final CassandraBlobStore blobStore;
 
-    protected final String keyPath;
+    protected final String keyPath; // XXX
 
     public AbstractCassandraBlobContainer(BlobPath path, CassandraBlobStore blobStore) {
         super(path);
         this.blobStore = blobStore;
         this.keyPath = path.buildAsString("/") + "/";
+        logger.debug("AbstractCassandraBlobContainer path={}", path);
     }
 
     @Override public boolean blobExists(String blobName) {
         try {
-            //XXX blobStore.client().getObjectMetadata(blobStore.bucket(), buildKey(blobName));
+            logger.debug("TODO blobExists blobName={}", blobName);
             return true;
         } catch (Exception e) {
             return false;
         }
     }
 
     @Override public boolean deleteBlob(String blobName) throws IOException {
-        //XXX blobStore.client().deleteObject(blobStore.bucket(), buildKey(blobName));
+        logger.debug("TODO deleteBlob blobName={}", blobName);
         return true;
     }
 
     @Override public void readBlob(final String blobName, final ReadBlobListener listener) {
+        logger.debug("TODO readBlob blobName={}", blobName);
         /* XXX
         blobStore.executor().execute(new Runnable() {
             @Override public void run() {
@@ -94,6 +97,7 @@ public AbstractCassandraBlobContainer(BlobPath path, CassandraBlobStore blobStor
     }
 
     @Override public ImmutableMap<String, BlobMetaData> listBlobsByPrefix(@Nullable String blobNamePrefix) throws IOException {
+        logger.debug("TODO listBlobsByPrefix blobNamePrefix={}", blobNamePrefix);
         ImmutableMap.Builder<String, BlobMetaData> blobsBuilder = ImmutableMap.builder();
         /* XXX
         ObjectListing prevListing = null;

plugins/cassandra/src/main/java/org/elasticsearch/cassandra/blobstore/CassandraBlobStore.java

@@ -19,13 +19,12 @@
 
 package org.elasticsearch.cassandra.blobstore;
 
-//import com.amazonaws.services.s3.AmazonS3;
-//import com.amazonaws.services.s3.model.ObjectListing;
-//import com.amazonaws.services.s3.model.S3ObjectSummary;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStore;
 import org.elasticsearch.common.blobstore.ImmutableBlobContainer;
 import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -38,62 +37,32 @@
  */
 public class CassandraBlobStore extends AbstractComponent implements BlobStore {
 
-    //XXX private final AmazonS3 client;
+    private final ESLogger logger = Loggers.getLogger(getClass());
 
-    //XXX private final String bucket;
+    private final Executor executor;
 
-    //XXX private final String region;
+    private final int bufferSizeInBytes; // XXX
 
-    //XXX private final Executor executor;
-
-    //XXX private final int bufferSizeInBytes;
-
-    public CassandraBlobStore(Settings settings, /*AmazonS3 client, */ String bucket, @Nullable String region, Executor executor) {
+    // XXX executor is a java.util.concurrent.ThreadPoolExecutor
+    public CassandraBlobStore(Settings settings, Executor executor) {
         super(settings);
-        /* XXX
-        this.client = client;
-        this.bucket = bucket;
-        this.region = region;
+
         this.executor = executor;
 
         this.bufferSizeInBytes = (int) settings.getAsBytesSize("buffer_size", new ByteSizeValue(100, ByteSizeUnit.KB)).bytes();
 
-        if (!client.doesBucketExist(bucket)) {
-            if (region != null) {
-                client.createBucket(bucket, region);
-            } else {
-                client.createBucket(bucket);
-            }
-        }
-        */
+        logger.debug("CassandraBlobStore executor={}, bufferSizeInBytes={}", executor, bufferSizeInBytes);
     }
 
     @Override public String toString() {
-    /* XXX
-        return (region == null ? "" : region + "/") + bucket;
-    */
-        return null;
-    }
-
-    /* XXX
-    public AmazonS3 client() {
-        return client;
-    }
-    */
-
-    /* XXX
-    public String bucket() {
-        return bucket;
+        return "cassandra"; // XXX
     }
-    */
 
-    /* XXX
     public Executor executor() {
         return executor;
     }
-    */
 
-    /*
+    /* XXX
     public int bufferSizeInBytes() {
         return bufferSizeInBytes;
     }
@@ -104,7 +73,8 @@ public int bufferSizeInBytes() {
     }
 
     @Override public void delete(BlobPath path) {
-        /* XXX
+        logger.debug("TODO delete path={}", path);
+        /* XXX TODO
         ObjectListing prevListing = null;
         while (true) {
             ObjectListing list;

plugins/cassandra/src/main/java/org/elasticsearch/cassandra/blobstore/CassandraImmutableBlobContainer.java

@@ -19,8 +19,6 @@
 
 package org.elasticsearch.cassandra.blobstore;
 
-//import com.amazonaws.services.s3.model.ObjectMetadata;
-//import com.amazonaws.services.s3.model.PutObjectResult;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.ImmutableBlobContainer;
 import org.elasticsearch.common.blobstore.support.BlobStores;
@@ -38,20 +36,17 @@ public CassandraImmutableBlobContainer(BlobPath path, CassandraBlobStore blobSto
     }
 
     @Override public void writeBlob(final String blobName, final InputStream is, final long sizeInBytes, final WriterListener listener) {
-        /* XXX
         blobStore.executor().execute(new Runnable() {
             @Override public void run() {
                 try {
-                    ObjectMetadata md = new ObjectMetadata();
-                    md.setContentLength(sizeInBytes);
-                    PutObjectResult objectResult = blobStore.client().putObject(blobStore.bucket(), buildKey(blobName), is, md);
+                    // XXX TODO
+                    logger.debug("TODO writeBlob blobName={}, sizeInBytes={}, is={}", blobName, sizeInBytes, is);
                     listener.onCompleted();
                 } catch (Exception e) {
                     listener.onFailure(e);
                 }
             }
         });
-        */
     }
 
     @Override public void writeBlob(String blobName, InputStream is, long sizeInBytes) throws IOException {

plugins/cassandra/src/main/java/org/elasticsearch/gateway/cassandra/CassandraGateway.java

@@ -21,8 +21,7 @@
 
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.ElasticSearchIllegalArgumentException;
-//import org.elasticsearch.cloud.aws.AwsS3Service;
-//import org.elasticsearch.cloud.aws.blobstore.S3BlobStore;
+import org.elasticsearch.cassandra.blobstore.CassandraBlobStore;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.metadata.MetaDataCreateIndexService;
@@ -43,22 +42,14 @@
 public class CassandraGateway extends BlobStoreGateway {
 
     @Inject public CassandraGateway(Settings settings, ClusterService clusterService, MetaDataCreateIndexService createIndexService,
-    ClusterName clusterName, ThreadPool threadPool /* XXX , AwsS3Service s3Service*/) throws IOException {
+    ClusterName clusterName, ThreadPool threadPool) throws IOException {
         super(settings, clusterService, createIndexService);
 
-        /* XXX
-        String bucket = componentSettings.get("bucket");
-        if (bucket == null) {
-            throw new ElasticSearchIllegalArgumentException("No bucket defined for s3 gateway");
-        }
-
-        String region = componentSettings.get("region");
         ByteSizeValue chunkSize = componentSettings.getAsBytesSize("chunk_size", new ByteSizeValue(100, ByteSizeUnit.MB));
 
-        logger.debug("using bucket [{}], region [{}], chunk_size [{}]", bucket, region, chunkSize);
+        logger.debug("using chunk_size [{}]", chunkSize);
 
-        initialize(new S3BlobStore(settings, s3Service.client(), bucket, region, threadPool.cached()), clusterName, chunkSize);
-        */
+        initialize(new CassandraBlobStore(settings, threadPool.cached()), clusterName, chunkSize);
     }
 
     @Override public void close() throws ElasticSearchException {