Queue info REST end-point
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/202d5bed Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/202d5bed Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/202d5bed Branch: refs/heads/master Commit: 202d5beda527e54852c86439c9c65d6ab2157ad8 Parents: e328584 Author: Dave Johnson <snoopd...@apache.org> Authored: Wed Oct 5 17:42:42 2016 -0400 Committer: Dave Johnson <snoopd...@apache.org> Committed: Wed Oct 5 17:42:42 2016 -0400 ---------------------------------------------------------------------- .../rest/system/QueueSystemResource.java | 78 ++++++++++++++++++++ 1 file changed, 78 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/202d5bed/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java ---------------------------------------------------------------------- diff --git a/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java b/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java index 15bdb34..86cd387 100644 --- a/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java +++ b/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java @@ -19,9 +19,18 @@ */ package org.apache.usergrid.rest.system; +import com.codahale.metrics.Timer; import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.usergrid.corepersistence.asyncevents.AsyncEventService; +import org.apache.usergrid.persistence.qakka.MetricsService; +import org.apache.usergrid.persistence.qakka.QakkaFig; +import org.apache.usergrid.persistence.qakka.core.Queue; +import org.apache.usergrid.persistence.qakka.core.QueueManager; +import org.apache.usergrid.persistence.qakka.core.QueueMessageManager; +import org.apache.usergrid.persistence.qakka.core.impl.InMemoryQueue; +import org.apache.usergrid.persistence.qakka.core.impl.QueueManagerImpl; +import org.apache.usergrid.persistence.qakka.core.impl.QueueMessageManagerImpl; import org.apache.usergrid.rest.AbstractContextResource; import org.apache.usergrid.rest.ApiResponse; import org.apache.usergrid.rest.security.annotations.RequireSystemAccess; @@ -32,6 +41,10 @@ import org.springframework.stereotype.Component; import javax.ws.rs.*; import javax.ws.rs.core.MediaType; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.text.DecimalFormat; +import java.util.*; /** * retrieves queue stats @@ -47,6 +60,7 @@ public class QueueSystemResource extends AbstractContextResource { public QueueSystemResource(){logger.info("queue resource initialized");} + /** * Return queue depth of this Usergrid instance in JSON format. * @@ -74,4 +88,68 @@ public class QueueSystemResource extends AbstractContextResource { return response; } + + @GET + @RequireSystemAccess + @Path("info") + public ApiResponse getQueueInfo( + @QueryParam("callback") @DefaultValue("callback") String callback ) { + + ApiResponse response = createApiResponse(); + response.setAction( "get queue info" ); + + MetricsService metricsService = injector.getInstance( MetricsService.class ); + + final DecimalFormat format = new DecimalFormat("##.###"); + final long nano = 1000000000; + + Map<String, Object> info = new HashMap<String, Object>() {{ + put( "name", "Queue Info" ); + try { + put( "host", InetAddress.getLocalHost().getHostName() ); + } catch (UnknownHostException e) { + put( "host", "unknown" ); + } + SortedSet<String> names = metricsService.getMetricRegistry().getNames(); + for (String name : names) { + Timer t = metricsService.getMetricRegistry().timer( name ); + put( name, new HashMap<String, Object>() {{ + put( "count", "" + t.getCount() ); + put( "mean_rate", "" + format.format( t.getMeanRate() ) ); + put( "one_minute_rate", "" + format.format( t.getOneMinuteRate() ) ); + put( "five_minute_rate", "" + format.format( t.getFiveMinuteRate() ) ); + put( "mean (s)", "" + format.format( t.getSnapshot().getMean() / nano ) ); + put( "min (s)", "" + format.format( (double) t.getSnapshot().getMin() / nano ) ); + put( "max (s)", "" + format.format( (double) t.getSnapshot().getMax() / nano ) ); + }} ); + } + }}; + + QueueManager queueManager = injector.getInstance( QueueManagerImpl.class ); + QueueMessageManager queueMessageManager = injector.getInstance( QueueMessageManagerImpl.class ); + InMemoryQueue inMemoryQueue = injector.getInstance( InMemoryQueue.class ); + + List queues = new ArrayList(); + final List<String> listOfQueues = queueManager.getListOfQueues(); + for ( String queueName : listOfQueues ) { + + Map<String, Object> queueInfo = new HashMap<>(); + + queueInfo.put("name", queueName ); + queueInfo.put("depth", queueMessageManager.getQueueDepth( queueName )); + queueInfo.put("inmemory", inMemoryQueue.size( queueName )); + + UUID newest = inMemoryQueue.getNewest( queueName ); + queueInfo.put("since", newest == null ? "null" : newest.timestamp()); + + queues.add( queueInfo ); + } + + info.put("queues", queues); + + response.setProperty( "data", info ); + + return response; + } + }
