[HWKMETRICS-613] Update DataAccess.updateMetricExpirationIndex to ret…

…urn Observable. The scheduler is now started after all the recurring jobs are scheduled. And log errors when metrics cannot be deleted.

core/metrics-core-service/src/main/java/org/hawkular/metrics/core/jobs/DeleteExpiredMetrics.java

@@ -23,6 +23,7 @@
 import org.hawkular.metrics.scheduler.api.JobDetails;
 import org.hawkular.metrics.sysconfig.ConfigurationService;
 import org.hawkular.rx.cassandra.driver.RxSession;
+import org.jboss.logging.Logger;
 
 import com.datastax.driver.core.PreparedStatement;
 
@@ -34,6 +35,9 @@
  * @author Stefan Negrea
  */
 public class DeleteExpiredMetrics implements Func1<JobDetails, Completable> {
+
+    private static Logger logger = Logger.getLogger(DeleteExpiredMetrics.class);
+
     public static final String JOB_NAME = "DELETE_EXPIRED_METRICS";
 
     private MetricsService metricsService;
@@ -42,7 +46,6 @@ public class DeleteExpiredMetrics implements Func1<JobDetails, Completable> {
     private PreparedStatement findEligibleTenants;
     private PreparedStatement findEligibleMetrics;
     private PreparedStatement findUnexpiredDataPoints;
-
     private long metricExpirationDelay;
 
     public DeleteExpiredMetrics(MetricsService metricsService, RxSession session,
@@ -108,7 +111,10 @@ public Completable call(JobDetails jobDetails) {
         }
 
         return expirationIndexResults
-                .flatMap(metricId -> metricsService.deleteMetric(metricId))
+                .concatMap(metricId -> metricsService.deleteMetric(metricId))
+                .doOnError(e -> {
+                    logger.error("Failed to delete metric data", e);
+                })
                 .toCompletable();
     }
 }

core/metrics-core-service/src/main/java/org/hawkular/metrics/core/jobs/JobsServiceImpl.java

@@ -95,8 +95,6 @@ public void setScheduler(Scheduler scheduler) {
 
     @Override
     public List<JobDetails> start() {
-        scheduler.start();
-
         List<JobDetails> backgroundJobs = new ArrayList<>();
 
         deleteTenant = new DeleteTenant(session, metricsService);
@@ -120,6 +118,8 @@ public List<JobDetails> start() {
         scheduler.register(DeleteExpiredMetrics.JOB_NAME, deleteExpiredMetrics);
         maybeScheduleMetricExpirationJob(backgroundJobs);
 
+        scheduler.start();
+
         return backgroundJobs;
     }
 

core/metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccess.java

@@ -137,7 +137,7 @@ <T> Observable<ResultSet> deleteAndInsertCompressedGauge(MetricId<T> id, long ti
                                                              CompressedPointContainer cpc,
                                                              long sliceStart, long sliceEnd, int ttl);
 
-    <T> ResultSetFuture updateMetricExpirationIndex(MetricId<T> id, long expirationTime);
+    <T> Observable<ResultSet> updateMetricExpirationIndex(MetricId<T> id, long expirationTime);
 
     <T> Observable<ResultSet> deleteFromMetricExpirationIndex(MetricId<T> id);
 

core/metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccessImpl.java

@@ -1228,8 +1228,8 @@ public Observable<Row> findAllMetricsFromTagsIndex() {
     }
 
     @Override
-    public <T> ResultSetFuture updateMetricExpirationIndex(MetricId<T> id, long expirationTime) {
-        return session.executeAsync(updateMetricExpirationIndex.bind(id.getTenantId(),
+    public <T> Observable<ResultSet> updateMetricExpirationIndex(MetricId<T> id, long expirationTime) {
+        return rxSession.execute(updateMetricExpirationIndex.bind(id.getTenantId(),
                 id.getType().getCode(), id.getName(), new Date(expirationTime)));
     }
 

core/metrics-core-service/src/main/java/org/hawkular/metrics/core/service/MetricsServiceImpl.java

@@ -1036,8 +1036,7 @@ public <T> Observable<Void> updateMetricExpiration(Metric<T> metric) {
                 expiration = DateTimeService.now.get().getMillis() + this.getTTL(metric.getMetricId()) * DAY_TO_MILLIS;
             }
 
-            return ListenableFutureObservable
-                    .from(dataAccess.updateMetricExpirationIndex(metric.getMetricId(), expiration), metricsTasks)
+            return dataAccess.updateMetricExpirationIndex(metric.getMetricId(), expiration)
                     .doOnError(t -> log.error("Failure to update expiration index", t))
                     .flatMap(r -> null);
         }

core/metrics-core-service/src/test/java/org/hawkular/metrics/core/jobs/DeleteExpiredMetricsJobITest.java

@@ -54,8 +54,6 @@
 import com.google.common.collect.ImmutableMap;
 
 import rx.Observable;
-import rx.observable.ListenableFutureObservable;
-import rx.schedulers.Schedulers;
 
 /**
  * Test the job that deletes expired metrics.
@@ -229,9 +227,8 @@ public void testScheduleDeleteExpiredMetricsJob() throws Exception {
                 ImmutableMap.of("x", "2", "y", "3"), 12);
 
         doAction(() -> metricsService.createMetric(c1, true));
-        ListenableFutureObservable
-        .from(dataAccess.updateMetricExpirationIndex(c1.getMetricId(),
-                        DateTimeService.now.get().getMillis() - 3 * 24 * 3600 * 1000L), Schedulers.immediate());
+        dataAccess.updateMetricExpirationIndex(c1.getMetricId(),
+                DateTimeService.now.get().getMillis() - 3 * 24 * 3600 * 1000L).toBlocking();
         doAction(() -> metricsService.createMetric(c2, true));
 
         List<Metric<Long>> metrics = getOnNextEvents(() -> metricsService.findMetrics(tenantId, COUNTER));

core/metrics-core-service/src/test/java/org/hawkular/metrics/core/service/DelegatingDataAccess.java

@@ -275,7 +275,7 @@ public Observable<Row> findAllMetricsFromTagsIndex() {
     }
 
     @Override
-    public <T> ResultSetFuture updateMetricExpirationIndex(MetricId<T> id, long expirationTime) {
+    public <T> Observable<ResultSet> updateMetricExpirationIndex(MetricId<T> id, long expirationTime) {
         return delegate.updateMetricExpirationIndex(id, expirationTime);
     }