Add handling for __recv-iconnection and __send-iconnection messaging layer metrics.

[KevinJMao]

New metrics were added in the upcoming Storm v0.10.0 release for the inter-worker messaging layer.

An example of a __send-iconnection DataPoint:

  #<DataPoint [__send-iconnection = 
  { 
  "6701/87f500b6-9fc3-4390-ae75-cff34e68a7e3" ->
    {
      "sent" -> 47, 
      "dest" -> "supervisor1/1.2.3.4:6701",
      "lostOnSend" -> 0,
      "reconnects" -> 0, 
      "pending" -> 0, 
      "src" -> "/1.2.3.3:43314"
    }, 
  "6700/87f500b6-9fc3-4390-ae75-cff34e68a7e3" -> 
    {
      "sent" -> 22,
      "dest" -> "supervisor1/1.2.3.5:6700",
      "lostOnSend" -> 0,
      "reconnects" -> 0,
      "pending" -> 0,
      "src" -> "/1.2.3.4:41344"
    }
  }]>

An example of a __recv-iconnection DataPoint:

#<DataPoint [__recv-iconnection = 
{
enqueued= {
    /1.2.3.4:33230=11,
    /1.2.3.4:33233=24,
    /1.2.3.5:38332=23
  },
pending=[0],
dequeuedMessages=58
}]>

At a minimum, we should be able to see queue sizes and pending messages fed into Graphite.

[miguno]

The __send-iconnection data corresponds to the following code in Storm's Client.java:

    @Override
    public Object getState() {
        LOG.info("Getting metrics for client connection to {}", dstAddressPrefixedName);
        HashMap<String, Object> ret = new HashMap<String, Object>();
        ret.put("reconnects", totalConnectionAttempts.getAndSet(0));
        ret.put("sent", messagesSent.getAndSet(0));
        ret.put("pending", pendingMessages.get());
        ret.put("lostOnSend", messagesLost.getAndSet(0));
        ret.put("dest", dstAddress.toString());
        String src = srcAddressName();
        if (src != null) {
            ret.put("src", src);
        }
        return ret;
    }

The __recv-iconnection data corresponds to the following code in Storm's Server.java:

    public Object getState() {
        LOG.info("Getting metrics for server on port {}", port);
        HashMap<String, Object> ret = new HashMap<String, Object>();
        ret.put("dequeuedMessages", messagesDequeued.getAndSet(0));
        ArrayList<Integer> pending = new ArrayList<Integer>(pendingMessages.length);
        for (AtomicInteger p: pendingMessages) {
            pending.add(p.get());
        }
        ret.put("pending", pending);
        HashMap<String, Integer> enqueued = new HashMap<String, Integer>();
        Iterator<Map.Entry<String, AtomicInteger>> it = messagesEnqueued.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, AtomicInteger> ent = it.next();
            //Yes we can delete something that is not 0 because of races, but that is OK for metrics
            AtomicInteger i = ent.getValue();
            if (i.get() == 0) {
                it.remove();
            } else {
                enqueued.put(ent.getKey(), i.getAndSet(0));
            }
        }
        ret.put("enqueued", enqueued);
        return ret;
    }