[HWKMETRICS-153] adding endpoint for fetching rates

The bulk of the changes/effort for this commit were for initializing the task
scheduling service and getting a test in place. There is still work to be done
with the task service initialization, but this is a good first step.

api/metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/MetricsServiceLifecycle.java

@@ -51,6 +51,7 @@
 import org.hawkular.metrics.api.jaxrs.config.ConfigurationProperty;
 import org.hawkular.metrics.api.jaxrs.util.Eager;
 import org.hawkular.metrics.core.api.MetricsService;
+import org.hawkular.metrics.core.impl.GenerateRate;
 import org.hawkular.metrics.core.impl.MetricsServiceImpl;
 import org.hawkular.metrics.core.impl.TaskTypes;
 import org.hawkular.metrics.schema.SchemaManager;
@@ -175,11 +176,19 @@ private void startMetricsService() {
             return;
         }
         try {
+            // When this class was first introduced, I said that the schema management
+            // should stay in MetricsServiceImpl, and now I am putting it back here. OK, so
+            // I deserve some criticism; however, I still think it should be done that way.
+            // I made this change temporarily because the schema for metrics and for the
+            // task scheduling service are declared and created in the same place. That
+            // will change at some point though because the task scheduling service will
+            // probably move to the hawkular-commons repo.
             initSchema();
             initTaskService();
 
             metricsService = new MetricsServiceImpl();
             metricsService.setTaskService(taskService);
+            taskService.subscribe(TaskTypes.COMPUTE_RATE, new GenerateRate(metricsService));
 
             // TODO Set up a managed metric registry
             // We want a managed registry that can be shared by the JAX-RS endpoint and the core. Then we can expose

api/metrics-api-jaxrs/src/main/java/org/hawkular/metrics/api/jaxrs/handler/CounterHandler.java

@@ -53,6 +53,7 @@
 import org.hawkular.metrics.api.jaxrs.handler.observer.ResultSetObserver;
 import org.hawkular.metrics.api.jaxrs.model.Counter;
 import org.hawkular.metrics.api.jaxrs.model.CounterDataPoint;
+import org.hawkular.metrics.api.jaxrs.model.GaugeDataPoint;
 import org.hawkular.metrics.api.jaxrs.request.MetricDefinition;
 import org.hawkular.metrics.api.jaxrs.util.ApiUtils;
 import org.hawkular.metrics.core.api.Metric;
@@ -203,4 +204,36 @@ public void findCounterData(
                         });
     }
 
+    @GET
+    @Path("/{id}/rate")
+    @ApiOperation(value = "Retrieve counter rate data points.", response = List.class)
+    @ApiResponses(value = {
+            @ApiResponse(code = 200, message = "Successfully fetched metric data."),
+            @ApiResponse(code = 204, message = "No metric data was found."),
+            @ApiResponse(code = 400, message = "start or end parameter is invalid.",
+                         response = ApiError.class),
+            @ApiResponse(code = 500, message = "Unexpected error occurred while fetching metric data.",
+                         response = ApiError.class) })
+    public void findRate(
+        @Suspended AsyncResponse asyncResponse,
+        @PathParam("id") String id,
+        @ApiParam(value = "Defaults to now - 8 hours") @QueryParam("start") final Long start,
+        @ApiParam(value = "Defaults to now") @QueryParam("end") final Long end) {
+
+        long now = System.currentTimeMillis();
+        long startTime = start == null ? now - EIGHT_HOURS : start;
+        long endTime = end == null ? now : end;
+
+        metricsService.findRateData(tenantId, new MetricId(id), startTime, endTime)
+                .map(GaugeDataPoint::new)
+                .toList()
+                .map(ApiUtils::collectionToResponse)
+                .subscribe(
+                        asyncResponse::resume,
+                        t -> {
+                            logger.warn("Failed to fetch counter rate data", t);
+                            ApiUtils.serverError(t);
+                        });
+    }
+
 }

rest-tests/pom.xml

@@ -33,6 +33,7 @@
 
   <properties>
     <cassandra.keyspace>hawkular_metrics_rest_tests</cassandra.keyspace>
+    <scheduler.units>seconds</scheduler.units>
   </properties>
 
   <dependencies>
@@ -232,6 +233,7 @@
                     <javaOpt>-Dcassandra.keyspace=${cassandra.keyspace}</javaOpt>
                     <javaOpt>-Dcassandra.resetdb</javaOpt>
                     <javaOpt>-Dhawkular.metrics.waitForService</javaOpt>
+                    <javaOpt>-Dhawkular.scheduler.time-units=${scheduler.units}</javaOpt>
                     <javaOpt>-Xdebug</javaOpt>
                     <javaOpt>-Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=8787</javaOpt>
                   </javaOpts>

rest-tests/src/test/groovy/org/hawkular/metrics/rest/CountersITest.groovy

@@ -15,13 +15,13 @@
  * limitations under the License.
  */
 package org.hawkular.metrics.rest
-
+import org.hawkular.metrics.core.impl.DateTimeService
 import org.joda.time.DateTime
+import org.joda.time.Duration
 import org.junit.Test
 
 import static org.joda.time.DateTime.now
 import static org.junit.Assert.assertEquals
-
 /**
  *
  */
@@ -249,4 +249,52 @@ class CountersITest extends RESTTest {
     assertEquals(204, response.status)
   }
 
+  @Test
+  void findRate() {
+    String tenantId = nextTenantId()
+    String counter = "C1"
+    DateTimeService dateTimeService = new DateTimeService()
+    DateTime start = dateTimeService.getTimeSlice(now(), Duration.standardSeconds(5))
+
+    def response = hawkularMetrics.post(
+        path: "counters",
+        headers: [(tenantHeaderName): tenantId],
+        body: [id: counter]
+    )
+    assertEquals(201, response.status)
+
+    response = hawkularMetrics.post(
+        path: "counters/$counter/data",
+        headers: [(tenantHeaderName): tenantId],
+        body: [
+            [timestamp: start.millis, value: 100],
+            [timestamp: start.plusSeconds(1).millis, value :200]
+        ]
+    )
+    assertEquals(200, response.status)
+
+    while (now().isBefore(start.plusSeconds(10))) {
+      Thread.sleep(100)
+    }
+
+    response = hawkularMetrics.get(
+        path: "counters/$counter/rate",
+        headers: [(tenantHeaderName): tenantId],
+        query: [start: now().minusSeconds(20).millis, end: now().millis]
+    )
+    assertEquals(200, response.status)
+
+    def expectedData = [
+        [timestamp: start.millis, value: (200 / (start.plusSeconds(5).millis - start.millis)) * 1000]
+    ]
+    assertRateEquals(expectedData, response.data)
+  }
+
+  static void assertRateEquals(def expected, def actual) {
+    expected.eachWithIndex { dataPoint, i ->
+      assertEquals("The timestamp does not match the expected value", dataPoint.timestamp, actual[i].timestamp)
+      assertDoubleEquals(dataPoint.value, actual[i].value)
+    }
+  }
+
 }

task-queue/src/main/java/org/hawkular/metrics/tasks/api/TaskService.java

@@ -18,6 +18,8 @@
 
 import org.joda.time.DateTime;
 import rx.Observable;
+import rx.Subscription;
+import rx.functions.Action1;
 
 /**
  * The primary API for task scheduling and execution. See {@link TaskServiceBuilder} for details on creating and
@@ -65,4 +67,6 @@ public interface TaskService {
      * @return The task with its {@link Task#getTimeSlice() scheduled execution time} set
      */
     Observable<Task> scheduleTask(DateTime time, Task task);
+
+    Subscription subscribe(TaskType taskType, Action1<? super Task> onNext);
 }

task-queue/src/main/java/org/hawkular/metrics/tasks/impl/TaskServiceImpl.java

@@ -137,6 +137,7 @@ public TaskServiceImpl(RxSession session, Queries queries, LeaseService leaseSer
      * </p>
      */
     public void setTimeUnit(TimeUnit timeUnit) {
+        logger.info("Using time unit of {}", timeUnit);
         switch (timeUnit) {
             case SECONDS:
                 this.timeUnit = TimeUnit.SECONDS;
@@ -190,6 +191,7 @@ public Subscription subscribe(TaskType taskType, final Action1<? super Task> onN
         return subject.subscribe(onNext, onError, onComplete);
     }
 
+    @Override
     public Subscription subscribe(TaskType taskType, Action1<? super Task> onNext) {
         PublishSubject<Task> subject = subjects.get(taskType);
         if (subject == null) {
@@ -316,6 +318,8 @@ void executeTasks(DateTime timeSlice) {
      * @param taskType
      */
     private void executeTasks(DateTime timeSlice, TaskType taskType) {
+        logger.info("Executing tasks for time slice {}", timeSlice);
+
         // I know, I know. We should not have to used CountDownLatch with RxJava. It is
         // left over from the original implementation and was/is used to ensure tasks of
         // one type finish executing before we start executing tasks of the next type.