Add metric with neighbour info for each owned token range

common/src/main/java/com/zegelin/cassandra/exporter/FactoriesSupplier.java

@@ -115,6 +115,7 @@ public interface CollectorConstructor {
     private final boolean perThreadTimingEnabled;
     private final Set<TableLabels> tableLabels;
     private final Set<String> excludedKeyspaces;
+    private final boolean resolveIP;
     private final Map<TableMetricScope, TableMetricScope.Filter> tableMetricScopeFilters;
 
 
@@ -123,6 +124,7 @@ public FactoriesSupplier(final MetadataFactory metadataFactory, final HarvesterO
         this.perThreadTimingEnabled = options.perThreadTimingEnabled;
         this.tableLabels = options.tableLabels;
         this.excludedKeyspaces = options.excludedKeyspaces;
+        this.resolveIP = options.resolveIP;
 
         this.tableMetricScopeFilters = ImmutableMap.<TableMetricScope, TableMetricScope.Filter>builder()
                 .put(TableMetricScope.NODE, options.nodeMetricsFilter)
@@ -532,7 +534,7 @@ public List<Factory> get() {
         final ImmutableList.Builder<Factory> builder = ImmutableList.builder();
 
         builder.add(FailureDetectorMBeanMetricFamilyCollector.factory(metadataFactory));
-        builder.add(cache(StorageServiceMBeanMetricFamilyCollector.factory(metadataFactory, excludedKeyspaces), 5, TimeUnit.MINUTES));
+        builder.add(cache(StorageServiceMBeanMetricFamilyCollector.factory(metadataFactory, excludedKeyspaces, resolveIP), 5, TimeUnit.MINUTES));
 
         builder.add(MemoryPoolMXBeanMetricFamilyCollector.FACTORY);
         builder.add(GarbageCollectorMXBeanMetricFamilyCollector.FACTORY);

common/src/main/java/com/zegelin/cassandra/exporter/Harvester.java

@@ -131,6 +131,8 @@ public boolean equals(final Object o) {
     private final Set<GlobalLabel> enabledGlobalLabels;
 
     private final boolean collectorTimingEnabled;
+    private final boolean resolveIP;
+
     private final Map<String, Stopwatch> collectionTimes = new ConcurrentHashMap<>();
 
     private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder()
@@ -145,6 +147,7 @@ protected Harvester(final MetadataFactory metadataFactory, final HarvesterOption
         this.exclusions = options.exclusions;
         this.enabledGlobalLabels = options.globalLabels;
         this.collectorTimingEnabled = options.collectorTimingEnabled;
+        this.resolveIP = options.resolveIP;
     }
 
     protected void addCollectorFactory(final MBeanGroupMetricFamilyCollector.Factory factory) {

common/src/main/java/com/zegelin/cassandra/exporter/cli/HarvesterOptions.java

@@ -152,4 +152,8 @@ public void setExcludeSystemTables(final boolean excludeSystemTables) {
 
         excludedKeyspaces.addAll(CASSANDRA_SYSTEM_KEYSPACES);
     }
+
+    @Option(names = "--enable-ip-resolve",
+            description = "Add label with hostname(s) from resolved IP(s)")
+    public boolean resolveIP;
 }

common/src/main/java/com/zegelin/cassandra/exporter/collector/StorageServiceMBeanMetricFamilyCollector.java

@@ -1,7 +1,9 @@
 package com.zegelin.cassandra.exporter.collector;
 
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
 import com.zegelin.cassandra.exporter.MBeanGroupMetricFamilyCollector;
 import com.zegelin.cassandra.exporter.MetadataFactory;
 import com.zegelin.prometheus.domain.GaugeMetricFamily;
@@ -14,41 +16,50 @@
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.nio.file.FileStore;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.HashSet;
+import java.util.StringTokenizer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import static com.zegelin.cassandra.exporter.CassandraObjectNames.STORAGE_SERVICE_MBEAN_NAME;
 
 public class StorageServiceMBeanMetricFamilyCollector extends MBeanGroupMetricFamilyCollector {
     private static final Logger logger = LoggerFactory.getLogger(StorageServiceMBeanMetricFamilyCollector.class);
 
-    public static Factory factory(final MetadataFactory metadataFactory, final Set<String> excludedKeyspaces) {
+    public static Factory factory(final MetadataFactory metadataFactory, final Set<String> excludedKeyspaces, final boolean resolveIP) {
         return mBean -> {
             if (!STORAGE_SERVICE_MBEAN_NAME.apply(mBean.name))
                 return null;
 
-            return new StorageServiceMBeanMetricFamilyCollector((StorageServiceMBean) mBean.object, metadataFactory, excludedKeyspaces);
+            return new StorageServiceMBeanMetricFamilyCollector((StorageServiceMBean) mBean.object, metadataFactory, excludedKeyspaces, resolveIP);
         };
     }
 
     private final StorageServiceMBean storageServiceMBean;
     private final MetadataFactory metadataFactory;
     private final Set<String> excludedKeyspaces;
-
+    private final boolean resolveIP;
 
     private final Map<Labels, FileStore> labeledFileStores;
 
+    private final Pattern tokenRangePattern = Pattern.compile("TokenRange\\(start_token:(-?\\d+), end_token:(-?\\d+)(, )?endpoints:\\[([^\\]]+)\\]");
+
 
     private StorageServiceMBeanMetricFamilyCollector(final StorageServiceMBean storageServiceMBean,
-                                                     final MetadataFactory metadataFactory, final Set<String> excludedKeyspaces) {
+                                                     final MetadataFactory metadataFactory, final Set<String> excludedKeyspaces, final boolean resolveIP) {
         this.storageServiceMBean = storageServiceMBean;
         this.metadataFactory = metadataFactory;
         this.excludedKeyspaces = excludedKeyspaces;
+        this.resolveIP = resolveIP;
 
         // determine the set of FileStores (i.e., mountpoints) for the Cassandra data/CL/cache directories
         // (which can be done once -- changing directories requires a server restart)
@@ -76,6 +87,44 @@ private StorageServiceMBeanMetricFamilyCollector(final StorageServiceMBean stora
         this.labeledFileStores = ImmutableMap.copyOf(labeledFileStores);
     }
 
+    private Labels decodeTokenRange(String tokenRange, String localIP, String keyspace) {
+        HashMap<String, String> m = Maps.newHashMap(metadataFactory.endpointLabels(localIP));
+        HashSet<String> endpoints = new HashSet<String>();
+
+        m.put("keyspace", keyspace);
+
+        // token range example:
+        // TokenRange(start_token:5585272669612250202, end_token:5664918566912044362, endpoints:[172.16.28.48, 172.16.28.166], rpc_endpoints:[172.16.28.48, 172.16.28.166], endpoint_details:[EndpointDetails(host:172.16.28.48, datacenter:eu-west_edge-irl1_profiles-bk, rack:1a), EndpointDetails(host:172.16.28.166, datacenter:eu-west_edge-irl1_profiles-bk, rack:1c)])
+        // see https://github.com/apache/cassandra/blob/trunk/src/java/org/apache/cassandra/service/TokenRange.java
+        Matcher matcher = tokenRangePattern.matcher(tokenRange);
+        if (matcher.find()) {
+            m.put("start_token", matcher.group(1));
+            m.put("end_token", matcher.group(2));
+
+            StringTokenizer st = new StringTokenizer(matcher.group(4), ",");
+            while (st.hasMoreTokens()) {
+                endpoints.add(st.nextToken().trim());
+            }
+
+            if (endpoints.remove(localIP)) {
+                m.put("neighbours_endpoints", String.join(", ", endpoints));
+                if (resolveIP)
+                    m.put("neighbours_hostnames", endpoints.stream().map(e -> {
+                        try {
+                            return InetAddress.getByName(e).getHostName();
+                        } catch (UnknownHostException ex) {
+                            return e;
+                        }
+                    }).collect(Collectors.joining(", ")));
+            } else {
+                m.put("neighbours_endpoints", "");
+                if (resolveIP)
+                    m.put("neighbours_hostnames", "");
+            }
+        }
+        return new Labels(m);
+    }
+
     @Override
     public Stream<MetricFamily> collect() {
         final Stream.Builder<MetricFamily> metricFamilyStreamBuilder = Stream.builder();
@@ -91,6 +140,23 @@ public Stream<MetricFamily> collect() {
             metricFamilyStreamBuilder.add(new GaugeMetricFamily("cassandra_token_ownership_ratio", null, ownershipMetricStream));
         }
 
+        {
+            String localIP = storageServiceMBean.getHostIdToEndpoint().get(storageServiceMBean.getLocalHostId());
+            final Stream<NumericMetric> ownershipMetricStream = metadataFactory.keyspaces().stream()
+                    .filter(keyspace -> !excludedKeyspaces.contains(keyspace))
+                    .flatMap(keyspace -> {
+                        try {
+                            return storageServiceMBean.describeRingJMX(keyspace).stream()
+                                    .map(e -> decodeTokenRange(e, localIP, keyspace))
+                                    .filter(e -> !Strings.isNullOrEmpty(e.get("neighbours_endpoints")))
+                                    .map(e -> new NumericMetric(e, 1.0f));
+                        } catch (IOException e) {
+                            return Stream.empty();
+                        }
+                    });
+            metricFamilyStreamBuilder.add(new GaugeMetricFamily("cassandra_neighbours", null, ownershipMetricStream));
+        }
+
         {
             final Stream<NumericMetric> ownershipMetricStream = metadataFactory.keyspaces().stream()
                     .filter(keyspace -> !excludedKeyspaces.contains(keyspace))