Merge pull request #847 from jpkrohling/HAWKULAR-1032-PingerSendingDa…

…taToMetricsViaBus

HAWKULAR-1032 - Pinger now sends data to metrics via bus

dist/src/main/resources/standalone.xsl

@@ -266,6 +266,9 @@
       <jms-topic name="HawkularAvailData" entries="java:/topic/HawkularAvailData"/>
       <jms-topic name="HawkularCommandEvent" entries="java:/topic/HawkularCommandEvent"/>
       <jms-topic name="HawkularAlertsActionsTopic" entries="java:/topic/HawkularAlertsActionsTopic"/>
+      <jms-queue name="hawkular/metrics/gauges/new" entries="java:/queue/hawkular/metrics/gauges/new"/>
+      <jms-queue name="hawkular/metrics/counters/new" entries="java:/queue/hawkular/metrics/counters/new"/>
+      <jms-queue name="hawkular/metrics/availability/new" entries="java:/queue/hawkular/metrics/availability/new"/>
     </xsl:copy>
   </xsl:template>
 

modules/pinger/src/main/java/org/hawkular/component/pinger/MetricDataMessage.java

@@ -17,17 +17,24 @@
 package org.hawkular.component.pinger;
 
 import java.util.List;
+import java.util.Objects;
 
 import org.hawkular.bus.common.AbstractMessage;
 import org.hawkular.metrics.client.common.SingleMetric;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 
 /**
+ * A bus message for messages on HawkularMetricData Topic.
+ *
+ * @author Jay Shaughnessy
  * @author Lucas Ponce
  */
+
 public class MetricDataMessage extends AbstractMessage {
+    // TODO: we now have *3 copies* of this payload... shouldn't we have a place for payloads?
 
+    // the basic message body - it will be exposed to the JSON output
     @JsonInclude
     private MetricData metricData;
 
@@ -46,6 +53,26 @@ public void setMetricData(MetricData metricData) {
         this.metricData = metricData;
     }
 
+    @Override
+    public String toString() {
+        return "MetricDataMessage{" +
+                "metricData=" + metricData +
+                '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        MetricDataMessage that = (MetricDataMessage) o;
+        return Objects.equals(metricData, that.metricData);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(metricData);
+    }
+
     public static class MetricData {
         @JsonInclude
         String tenantId;
@@ -75,5 +102,82 @@ public void setData(List<SingleMetric> data) {
         public String toString() {
             return "MetricData [tenantId=" + tenantId + ", data=" + data + "]";
         }
+
+        @Override public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            MetricData that = (MetricData) o;
+            return Objects.equals(tenantId, that.tenantId) &&
+                    Objects.equals(data, that.data);
+        }
+
+        @Override public int hashCode() {
+            return Objects.hash(tenantId, data);
+        }
+    }
+
+    /**
+     * This is meant to parse out an instance of <code>org.rhq.metrics.client.common.SingleMetric</code>
+     */
+    public static class SingleMetric {
+        @JsonInclude
+        private String source;
+        @JsonInclude
+        private long timestamp;
+        @JsonInclude
+        private double value;
+
+        public SingleMetric() {
+        }
+
+        public SingleMetric(String source, long timestamp, double value) {
+            this.source = source;
+            this.timestamp = timestamp;
+            this.value = value;
+        }
+
+        public String getSource() {
+            return source;
+        }
+
+        public void setSource(String source) {
+            this.source = source;
+        }
+
+        public long getTimestamp() {
+            return timestamp;
+        }
+
+        public void setTimestamp(long timestamp) {
+            this.timestamp = timestamp;
+        }
+
+        public double getValue() {
+            return value;
+        }
+
+        public void setValue(double value) {
+            this.value = value;
+        }
+
+        @Override
+        public String toString() {
+            return "SingleMetric [source=" + source + ", timestamp=" + timestamp + ", value=" + value + "]";
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            SingleMetric that = (SingleMetric) o;
+            return Objects.equals(timestamp, that.timestamp) &&
+                    Objects.equals(value, that.value) &&
+                    Objects.equals(source, that.source);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(source, timestamp, value);
+        }
     }
 }

modules/pinger/src/main/java/org/hawkular/component/pinger/MetricPublisher.java

@@ -16,88 +16,80 @@
  */
 package org.hawkular.component.pinger;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.Arrays;
 
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
 import javax.ejb.Asynchronous;
-import javax.ejb.Stateless;
-
-import org.apache.http.HttpResponse;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.HttpClientBuilder;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
+import javax.ejb.Singleton;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.Queue;
+import javax.jms.TextMessage;
 
 /**
  * Publish metrics data
  *
  * @author Heiko W. Rupp
  */
-@Stateless
+@Singleton
 public class MetricPublisher {
+    @Resource(mappedName = "java:/queue/hawkular/metrics/gauges/new")
+    Queue gaugesQueue;
 
-    private static void addDataItem(List<Map<String, Object>> mMetrics, String resourceId, long timestamp,
-            Number value, String name) {
-        Map<String, Number> dataMap = new HashMap<>(2);
-        dataMap.put("timestamp", timestamp);
-        dataMap.put("value", value);
-        List<Map<String, Number>> data = new ArrayList<>(1);
-        data.add(dataMap);
-        Map<String, Object> outer = new HashMap<>(2);
-        outer.put("id", resourceId + ".status." + name);
-        outer.put("data", data);
-        mMetrics.add(outer);
-    }
+    @Resource(mappedName = "java:/HawkularBusConnectionFactory")
+    private ConnectionFactory connectionFactory;
 
-    private final PingerConfiguration configuration = PingerConfiguration.getInstance();
+    private JMSContext context;
+    private JMSProducer producer;
+
+    @PostConstruct
+    public void createContext() {
+        context = connectionFactory.createContext();
+        producer = context.createProducer();
+    }
 
     /**
-     * Serializes the given {@link PingStatus} and then submits it to Hawkular-metrics service via REST
+     * Submits data from {@link PingStatus} to Metrics via Bus
      *
      * @param status
      *            the {@link PingStatus} to publish
      */
     @Asynchronous
-    public void sendToMetricsViaRest(PingStatus status) {
-
-        List<Map<String, Object>> mMetrics = new ArrayList<>();
-
+    public void publish(PingStatus status) {
         final PingDestination dest = status.getDestination();
         final String resourceId = dest.getResourceId();
         final long timestamp = status.getTimestamp();
-        addDataItem(mMetrics, resourceId, timestamp, status.getDuration(), "duration");
-        addDataItem(mMetrics, resourceId, timestamp, status.getCode(), "code");
+        MetricDataMessage message = new MetricDataMessage();
+        MetricDataMessage.MetricData metricData = new MetricDataMessage.MetricData();
 
-        // Send it to metrics via rest
-        String payload;
-        try {
-            payload = new ObjectMapper().writeValueAsString(mMetrics);
-        } catch (JsonProcessingException e) {
-            Log.LOG.eCouldNotParseMessage(e);
-            return;
-        }
-        HttpClient client = HttpClientBuilder.create().build();
+        MetricDataMessage.SingleMetric durationMetric = new MetricDataMessage.SingleMetric(
+                resourceId,
+                timestamp,
+                status.getDuration()
+        );
 
-        HttpPost request = new HttpPost(configuration.getMetricsBaseUri() + "/gauges/data");
-        request.addHeader("Hawkular-Tenant", status.getDestination().getTenantId());
+        MetricDataMessage.SingleMetric statusCodeMetric = new MetricDataMessage.SingleMetric(
+                resourceId,
+                timestamp,
+                status.getCode()
+        );
 
-        request.setEntity(new StringEntity(payload, ContentType.APPLICATION_JSON));
+        metricData.setTenantId(status.getDestination().getTenantId());
+        metricData.setData(Arrays.asList(durationMetric, statusCodeMetric));
+        message.setMetricData(metricData);
 
+        TextMessage jmsMessage = context.createTextMessage();
+        String json = message.toJSON();
         try {
-            HttpResponse response = client.execute(request);
-            if (response.getStatusLine().getStatusCode() > 399) {
-                Log.LOG.wMetricPostStatus(response.getStatusLine().toString());
-            }
-        } catch (IOException e) {
-            Log.LOG.eMetricsIoException(e);
+            jmsMessage.setText(json);
+            producer.send(gaugesQueue, jmsMessage);
+        } catch (JMSException e) {
+            Log.LOG.eCouldNotSendMessage(e);
         }
+
     }
 
 }

modules/pinger/src/main/java/org/hawkular/component/pinger/PingManager.java

@@ -191,7 +191,7 @@ private void reportResults(List<PingStatus> results) {
         }
 
         for (PingStatus status : results) {
-            metricPublisher.sendToMetricsViaRest(status);
+            metricPublisher.publish(status);
             traitsPublisher.publish(status);
         }
 

modules/pinger/src/main/webapp/WEB-INF/jboss-deployment-structure.xml

@@ -21,6 +21,7 @@
   <deployment>
     <dependencies>
       <module name="org.hawkular.nest" />
+      <module name="org.hawkular.bus"/>
       <module name="org.hawkular.commons.guava" />
       <module name="org.jboss.resteasy.resteasy-jackson2-provider" services="import"/>
       <module name="deployment.hawkular-inventory-dist.war" services="import"/>

modules/pinger/src/test/java/org/hawkular/component/pinger/PingManagerTest.java

@@ -50,7 +50,7 @@ public void testScheduleWork() throws Exception {
         String expectedPoweredBy = "GitHub.com";
 
         ArgumentCaptor<PingStatus> metricsRestStatusCaptor = ArgumentCaptor.forClass(PingStatus.class);
-        Mockito.verify(manager.metricPublisher).sendToMetricsViaRest(metricsRestStatusCaptor.capture());
+        Mockito.verify(manager.metricPublisher).publish(metricsRestStatusCaptor.capture());
         assertStatus(expectedDest, expectedPoweredBy, metricsRestStatusCaptor.getValue());
 
         ArgumentCaptor<PingStatus> traitsStatusCaptor = ArgumentCaptor.forClass(PingStatus.class);

pom.xml

@@ -93,7 +93,7 @@
     <version.org.hawkular.cmdgw>0.10.11.Final-SRC-revision-6858e69c9543587bcc15da618c6cec80b08dd108</version.org.hawkular.cmdgw>
     <version.org.hawkular.commons>0.3.5.Final</version.org.hawkular.commons>
     <version.org.hawkular.inventory>0.13.0.Final-SRC-revision-8030a5aa969e3ff5bab534112ea2b5b91c8088d7</version.org.hawkular.inventory>
-    <version.org.hawkular.metrics>0.13.0-SRC-revision-c9260099383bcaab020b9623dff7baa1a421e73e</version.org.hawkular.metrics>
+    <version.org.hawkular.metrics>0.12.1.Final</version.org.hawkular.metrics>
     <version.org.jboss.weld.se>2.3.2.Final</version.org.jboss.weld.se>
     <version.org.keycloak>1.8.1.Final</version.org.keycloak>
     <version.org.keycloak.secretstore>1.0.12.Final</version.org.keycloak.secretstore>