Merge 'Implement sstable_info command' from Calle

"Fixes #76

Implements JMX level call for "sstable_info" REST api command.

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

Requires scylla patch set "Implement sstable_info API command (info on sstables)"

Forwards call to REST sstable_info and packs the data
into CompositeData for JMX consumption."
* 'sstabledesc' of git://github.com/elcallio/scylla-jmx:
  storage_service: Add "getSSTableInfo" command/attribute
  service: Add objects for deserializing sstable_info json
  scylla-apiclient: Add Date json serializer helper
  APIClient: Add jackson JSON serializer support to client object
  apiclient/pom.xml: Add jackson JSON support libs for REST client

scylla-apiclient/pom.xml

@@ -60,6 +60,21 @@
             <artifactId>activation</artifactId>
             <version>1.1</version>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <version>2.9.9</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.9.9</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.jaxrs</groupId>
+            <artifactId>jackson-jaxrs-json-provider</artifactId>
+            <version>2.9.9</version>
+        </dependency>
     </dependencies>
 
     <build>

scylla-apiclient/src/main/java/com/scylladb/jmx/api/APIClient.java

@@ -38,6 +38,7 @@ import javax.ws.rs.core.Response;
 
 import org.glassfish.jersey.client.ClientConfig;
 
+import com.fasterxml.jackson.jaxrs.json.JacksonJaxbJsonProvider;
 import com.scylladb.jmx.utils.SnapshotDetailsTabularData;
 
 public class APIClient {
@@ -78,9 +79,12 @@ public class APIClient {
     private static final Logger logger = Logger.getLogger(APIClient.class.getName());
 
     private final APIConfig config;
+    private final ClientConfig clientConfig;
 
     public APIClient(APIConfig config) {
         this.config = config;
+        this.clientConfig = new ClientConfig();
+        clientConfig.register(new JacksonJaxbJsonProvider());
     }
 
     private String getBaseUrl() {
@@ -88,7 +92,7 @@ public class APIClient {
     }
 
     public Invocation.Builder get(String path, MultivaluedMap<String, String> queryParams) {
-        Client client = ClientBuilder.newClient(new ClientConfig());
+        Client client = ClientBuilder.newClient(clientConfig);
         WebTarget webTarget = client.target(getBaseUrl()).path(path);
         if (queryParams != null) {
             for (Entry<String, List<String>> qp : queryParams.entrySet()) {

src/main/java/com/scylladb/jmx/utils/DateXmlAdapter.java

@@ -0,0 +1,19 @@
+package com.scylladb.jmx.utils;
+
+import java.time.Instant;
+import java.util.Date;
+
+import javax.xml.bind.annotation.adapters.XmlAdapter;
+
+public class DateXmlAdapter extends XmlAdapter<String, Date> {
+    @Override
+    public String marshal(Date v) throws Exception {
+        return Instant.ofEpochMilli(v.getTime()).toString();
+    }
+
+    @Override
+    public Date unmarshal(String v) throws Exception {
+        return new Date(Instant.parse(v).toEpochMilli());
+    }
+
+}

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

@@ -0,0 +1,66 @@
+package org.apache.cassandra.service;
+
+import static com.sun.jmx.mbeanserver.MXBeanMappingFactory.DEFAULT;
+
+import java.io.InvalidObjectException;
+import java.util.List;
+
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.OpenDataException;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import com.sun.jmx.mbeanserver.MXBeanMapping;
+
+@SuppressWarnings("restriction")
+@XmlRootElement
+public class PerTableSSTableInfo {
+    private static final MXBeanMapping mxBeanMapping;
+
+    static {
+        try {
+            mxBeanMapping = DEFAULT.mappingForType(PerTableSSTableInfo.class, DEFAULT);
+        } catch (OpenDataException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private String keyspace;
+    private List<SSTableInfo> sstables;
+    private String table;
+
+    public String getTable() {
+        return table;
+    }
+
+    public void setTable(String table) {
+        this.table = table;
+    }
+
+    public String getKeyspace() {
+        return keyspace;
+    }
+
+    public void setKeyspace(String keyspace) {
+        this.keyspace = keyspace;
+    }
+
+    public List<SSTableInfo> getSSTables() {
+        return sstables;
+    }
+
+    public void setSSTableInfos(List<SSTableInfo> sstableInfos) {
+        this.sstables = sstableInfos;
+    }
+
+    public CompositeData toCompositeData() {
+        try {
+            return (CompositeData) mxBeanMapping.toOpenValue(this);
+        } catch (OpenDataException e) {
+            throw new Error(e); // should not reach.
+        }
+    }
+
+    public static PerTableSSTableInfo from(CompositeData data) throws InvalidObjectException {
+        return (PerTableSSTableInfo) mxBeanMapping.fromOpenValue(data);
+    }
+}

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

@@ -0,0 +1,143 @@
+package org.apache.cassandra.service;
+
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.scylladb.jmx.utils.DateXmlAdapter;
+
+public class SSTableInfo {
+    private long size;
+
+    @JsonProperty("data_size")
+    private long dataSize;
+
+    @JsonProperty("index_size")
+    private long indexSize;
+
+    @JsonProperty("filter_size")
+    private long filterSize;
+
+    @XmlJavaTypeAdapter(type = Date.class, value = DateXmlAdapter.class)
+    private Date timestamp;
+
+    private long generation;
+
+    private long level;
+
+    private String version;
+
+    private Map<String, String> properties;
+
+    public void setProperties(Map<String, String> properties) {
+        this.properties = properties;
+    }
+
+    @JsonIgnore
+    private Map<String, Map<String, String>> extendedProperties;
+
+    public String getVersion() {
+        return version;
+    }
+
+    public void setVersion(String version) {
+        this.version = version;
+    }
+
+    public long getSize() {
+        return size;
+    }
+
+    public void setSize(long size) {
+        this.size = size;
+    }
+
+    public long getDataSize() {
+        return dataSize;
+    }
+
+    public void setDataSize(long dataSize) {
+        this.dataSize = dataSize;
+    }
+
+    public long getIndexSize() {
+        return indexSize;
+    }
+
+    public void setIndexSize(long indexSize) {
+        this.indexSize = indexSize;
+    }
+
+    public long getFilterSize() {
+        return filterSize;
+    }
+
+    public void setFilterSize(long filterSize) {
+        this.filterSize = filterSize;
+    }
+
+    public Date getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(Date timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public long getGeneration() {
+        return generation;
+    }
+
+    public void setGeneration(long generation) {
+        this.generation = generation;
+    }
+
+    public long getLevel() {
+        return level;
+    }
+
+    public void setLevel(long level) {
+        this.level = level;
+    }
+
+    public Map<String, String> getProperties() {
+        return properties;
+    }
+
+    public Map<String, Map<String, String>> getExtendedProperties() {
+        return extendedProperties;
+    }
+
+    public void setExtendedProperties(Map<String, Map<String, String>> extendedProperties) {
+        this.extendedProperties = extendedProperties;
+    }
+
+    @JsonProperty("extended_properties")
+    private void unpackNested(List<Map<String, Object>> properties) {
+        Map<String, Map<String, String>> result = new HashMap<String, Map<String, String>>();
+
+        for (Map<String, Object> map : properties) {
+            Object name = map.get("group");
+            if (name != null) {
+                Map<String, String> dst = new HashMap<>();
+                List<?> value = (List<?>) map.get("attributes");
+                for (Object v : value) {
+                    Map<?, ?> subMap = (Map<?, ?>) v;
+                    dst.put(String.valueOf(subMap.get("key")), String.valueOf(subMap.get("value")));
+                }
+                result.put(String.valueOf(name), dst);
+            } else {
+                for (Map.Entry<String, Object> e : map.entrySet()) {
+                    result.put(e.getKey(), Collections.singletonMap(String.valueOf(e.getValue()), ""));
+                }
+            }
+        }
+        extendedProperties = result;
+    }
+}

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;
@@ -492,12 +493,12 @@ public class StorageService extends MetricsMBean implements StorageServiceMBean,
     @Override
     public void takeSnapshot(String tag, Map<String, String> options, String... keyspaceNames) throws IOException {
         log(" takeSnapshot(String tag, String... keyspaceNames) throws IOException");
-        MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();        
+        MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
         APIClient.set_query_param(queryParams, "tag", tag);
-        
+
         if (keyspaceNames.length == 1 && keyspaceNames[0].indexOf('.') != -1) {
             String[] parts = keyspaceNames[0].split("\\.");
-            keyspaceNames = new String[] { parts[0] };            
+            keyspaceNames = new String[] { parts[0] };
             APIClient.set_query_param(queryParams, "cf", parts[1]);
         }
         APIClient.set_query_param(queryParams, "kn", APIClient.join(keyspaceNames));
@@ -603,8 +604,9 @@ public class StorageService extends MetricsMBean implements StorageServiceMBean,
         client.post("/storage_service/keyspace_compaction/" + keyspaceName, queryParams);
     }
 
-    @Override 
-    public void forceKeyspaceCompactionForTokenRange(String keyspaceName, String startToken, String endToken, String... tableNames) throws IOException, ExecutionException, InterruptedException {
+    @Override
+    public void forceKeyspaceCompactionForTokenRange(String keyspaceName, String startToken, String endToken,
+            String... tableNames) throws IOException, ExecutionException, InterruptedException {
         // TODO: actually handle token ranges.
         forceKeyspaceCompaction(keyspaceName, tableNames);
     }
@@ -869,7 +871,7 @@ public class StorageService extends MetricsMBean implements StorageServiceMBean,
     @Deprecated
     public int forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters,
             Collection<String> hosts, boolean primaryRange, boolean repairedAt, String... columnFamilies)
-                    throws IOException {
+            throws IOException {
         log(" forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts,  boolean primaryRange, boolean repairedAt, String... columnFamilies) throws IOException");
         return repairRangeAsync(null, null, keyspace, isSequential, dataCenters, hosts, primaryRange, repairedAt,
                 columnFamilies);
@@ -1298,12 +1300,19 @@ 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 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]"
+     * @param sourceDc
+     *            Name of DC from which to select sources for streaming or null
+     *            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
     public void rebuild(String sourceDc, String keyspace, String tokens, String specificSources) {
@@ -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();
 }