storage_service: Add "getSSTableInfo" command/attribute

Fixes #76

Requires seastar patch:
 json: Make date formatter use RFC8601/RFC3339 format

Requires scylla patch set "Sstabledesc"

Forwards call to REST sstable_info and packs the data
into CompositeData for JMX consumption.

src/main/java/org/apache/cassandra/service/StorageService.java

@@ -42,6 +42,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
 
 import javax.json.JsonArray;
 import javax.json.JsonObject;
@@ -52,14 +53,14 @@ import javax.management.NotificationBroadcaster;
 import javax.management.NotificationBroadcasterSupport;
 import javax.management.NotificationFilter;
 import javax.management.NotificationListener;
+import javax.management.openmbean.CompositeData;
 import javax.management.openmbean.TabularData;
+import javax.ws.rs.core.GenericType;
 import javax.ws.rs.core.MultivaluedHashMap;
 import javax.ws.rs.core.MultivaluedMap;
 
 import org.apache.cassandra.metrics.StorageMetrics;
 import org.apache.cassandra.repair.RepairParallelism;
-import org.glassfish.jersey.client.ClientConfig;
-import org.glassfish.jersey.client.ClientProperties;
 
 import com.google.common.base.Joiner;
 import com.scylladb.jmx.api.APIClient;
@@ -604,7 +605,8 @@ public class StorageService extends MetricsMBean implements StorageServiceMBean,
     }
 
     @Override
-    public void forceKeyspaceCompactionForTokenRange(String keyspaceName, String startToken, String endToken, String... tableNames) throws IOException, ExecutionException, InterruptedException {
+    public void forceKeyspaceCompactionForTokenRange(String keyspaceName, String startToken, String endToken,
+            String... tableNames) throws IOException, ExecutionException, InterruptedException {
         // TODO: actually handle token ranges.
         forceKeyspaceCompaction(keyspaceName, tableNames);
     }
@@ -1298,11 +1300,18 @@ public class StorageService extends MetricsMBean implements StorageServiceMBean,
     }
 
     /**
-     * Same as {@link #rebuild(String)}, but only for specified keyspace and ranges.
+     * Same as {@link #rebuild(String)}, but only for specified keyspace and
+     * ranges.
      *
-     * @param sourceDc Name of DC from which to select sources for streaming or null to pick any node
+     * @param sourceDc
-     * @param keyspace Name of the keyspace which to rebuild or null to rebuild all keyspaces.
+     *            Name of DC from which to select sources for streaming or null
-     * @param tokens Range of tokens to rebuild or null to rebuild all token ranges. In the format of:
+     *            to pick any node
+     * @param keyspace
+     *            Name of the keyspace which to rebuild or null to rebuild all
+     *            keyspaces.
+     * @param tokens
+     *            Range of tokens to rebuild or null to rebuild all token
+     *            ranges. In the format of:
      *            "(start_token_1,end_token_1],(start_token_2,end_token_2],...(start_token_n,end_token_n]"
      */
     @Override
@@ -1704,4 +1713,29 @@ public class StorageService extends MetricsMBean implements StorageServiceMBean,
         log(" resumeBootstrap");
         return false;
     }
+
+    @Override
+    public List<CompositeData> getSSTableInfo(String keyspace, String table) {
+        if (keyspace == null && table != null) {
+            throw new IllegalArgumentException("Missing keyspace name");
+        }
+        MultivaluedMap<String, String> queryParams = null;
+
+        if (keyspace != null) {
+            queryParams = new MultivaluedHashMap<String, String>();
+            queryParams.add("keyspace", keyspace);
+        }
+        if (table != null) {
+            queryParams.add("cf", table);
+        }
+
+        return client.get("/storage_service/sstable_info", queryParams)
+                .get(new GenericType<List<PerTableSSTableInfo>>() {
+                }).stream().map((i) -> i.toCompositeData()).collect(Collectors.toList());
+    }
+    
+    @Override
+    public List<CompositeData> getSSTableInfo() {
+        return getSSTableInfo(null, null);
+    }
 }

src/main/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -35,6 +35,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
 
 import javax.management.NotificationEmitter;
+import javax.management.openmbean.CompositeData;
 import javax.management.openmbean.TabularData;
 
 public interface StorageServiceMBean extends NotificationEmitter {
@@ -872,4 +873,8 @@ public interface StorageServiceMBean extends NotificationEmitter {
      * Sets the hinted handoff throttle in kb per second, per delivery thread.
      */
     public boolean resumeBootstrap();
+    
+    public List<CompositeData> getSSTableInfo(String keyspace, String table);
+
+    public List<CompositeData> getSSTableInfo();
 }