Merge 926e23c into a367585

blueflood-core/src/main/java/com/rackspacecloud/blueflood/eventemitter/RollupEventEmitter.java

@@ -20,8 +20,9 @@
 import com.google.common.collect.Lists;
 import com.rackspacecloud.blueflood.concurrent.ThreadPoolBuilder;
 import com.rackspacecloud.blueflood.io.DiscoveryIO;
+import com.rackspacecloud.blueflood.service.CoreConfig;
 import com.rackspacecloud.blueflood.types.BasicRollup;
-import com.rackspacecloud.blueflood.utils.QueryDiscoveryModuleLoader;
+import com.rackspacecloud.blueflood.utils.ModuleLoader;
 import com.rackspacecloud.blueflood.utils.Util;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,7 +31,7 @@
 import java.util.concurrent.*;
 
 public class RollupEventEmitter extends Emitter<RollupEvent> {
-    private static final Logger log = LoggerFactory.getLogger(QueryDiscoveryModuleLoader.class);
+    private static final Logger log = LoggerFactory.getLogger(ModuleLoader.class);
     private static final int numberOfWorkers = 5;
     public static final String ROLLUP_EVENT_NAME = "rollup".intern();
     private static ThreadPoolExecutor eventExecutors;
@@ -56,8 +57,7 @@ public Future emit(final String event, final RollupEvent... eventPayload) {
                 @Override
                 public Future call() {
                     if (Util.shouldUseESForUnits()) {
-                        QueryDiscoveryModuleLoader.loadDiscoveryModule();
-                        final DiscoveryIO discoveryIO = QueryDiscoveryModuleLoader.getDiscoveryInstance();
+                        final DiscoveryIO discoveryIO = (DiscoveryIO) ModuleLoader.getInstance(DiscoveryIO.class, CoreConfig.DISCOVERY_MODULES);
                         // TODO: Sync for now, but we will have to make it async eventually
                         Lists.transform(Arrays.asList(eventPayload), new Function<RollupEvent, RollupEvent>() {
                             @Override

blueflood-core/src/main/java/com/rackspacecloud/blueflood/io/GenericElasticSearchIO.java → blueflood-core/src/main/java/com/rackspacecloud/blueflood/io/EventsIO.java

@@ -19,7 +19,7 @@
 import java.util.List;
 import java.util.Map;
 
-public interface GenericElasticSearchIO {
+public interface EventsIO {
     public void insert(String tenant, List<Map<String, Object>> metrics) throws Exception;
     public List<Map<String, Object>> search(String tenant, Map<String, List<String>> query) throws Exception;
 }

blueflood-core/src/main/java/com/rackspacecloud/blueflood/outputs/handlers/RollupHandler.java

@@ -32,10 +32,7 @@
 import com.rackspacecloud.blueflood.service.Configuration;
 import com.rackspacecloud.blueflood.service.CoreConfig;
 import com.rackspacecloud.blueflood.types.*;
-import com.rackspacecloud.blueflood.utils.Metrics;
-import com.rackspacecloud.blueflood.utils.QueryDiscoveryModuleLoader;
-import com.rackspacecloud.blueflood.utils.TimeValue;
-import com.rackspacecloud.blueflood.utils.Util;
+import com.rackspacecloud.blueflood.utils.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -136,7 +133,8 @@ public Map<Locator, MetricData> getRollupByGranularity(
 
                  @Override
                  public List<SearchResult> call() throws Exception {
-                     DiscoveryIO discoveryIO = QueryDiscoveryModuleLoader.getDiscoveryInstance();
+                     DiscoveryIO discoveryIO = (DiscoveryIO) ModuleLoader.getInstance(DiscoveryIO.class, CoreConfig.DISCOVERY_MODULES);
+
                      if (discoveryIO == null) {
                          log.warn("USE_ES_FOR_UNITS has been set to true, but no discovery module found." +
                                  " Please check your config");

blueflood-core/src/main/java/com/rackspacecloud/blueflood/utils/ModuleLoader.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2015 Rackspace
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.rackspacecloud.blueflood.utils;
+
+import com.rackspacecloud.blueflood.service.Configuration;
+import com.rackspacecloud.blueflood.service.CoreConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
+
+public class ModuleLoader {
+    private static final Logger log = LoggerFactory.getLogger(ModuleLoader.class);
+    private static Map<String, Object> loadedModules = new HashMap<String, Object>();
+
+    public static Object getInstance(Class c, CoreConfig moduleName) {
+
+        Object moduleInstance = loadedModules.get(moduleName.name().toString());
+        if (moduleInstance != null)
+            return moduleInstance;
+        List<String> modules = Configuration.getInstance().getListProperty(moduleName);
+        if (modules.isEmpty())
+            return null;
+        if (!modules.isEmpty() && modules.size() != 1) {
+            throw new RuntimeException("Cannot load service with more than one "+moduleName+" module");
+        }
+
+        String module = modules.get(0);
+        log.info("Loading the module " + module);
+
+        try {
+            ClassLoader loader = c.getClassLoader();
+            Class genericClass = loader.loadClass(module);
+            moduleInstance = genericClass.newInstance();
+            loadedModules.put(moduleName.name().toString(), moduleInstance);
+            log.info("Registering the module " + module);
+        }
+        catch (InstantiationException e) {
+            log.error(String.format("Unable to create instance of %s class for %s", c.getName(), module), e);
+        } catch (IllegalAccessException e) {
+            log.error("Error starting module: " + module, e);
+        } catch (ClassNotFoundException e) {
+            log.error("Unable to locate module: " + module, e);
+        } catch (RuntimeException e) {
+            log.error("Error starting module: " + module, e);
+        } catch (Throwable e) {
+            log.error("Error starting module: " + module, e);
+        }
+
+        return moduleInstance;
+    }
+}

blueflood-elasticsearch/src/main/java/com/rackspacecloud/blueflood/io/EventElasticSearchIO.java

@@ -34,7 +34,7 @@
 import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;
 import static org.elasticsearch.index.query.QueryBuilders.termQuery;
 
-public class EventElasticSearchIO implements GenericElasticSearchIO {
+public class EventElasticSearchIO implements EventsIO {
     private final Timer eventSearchTimer = Metrics.timer(EventElasticSearchIO.class,
             "Search time for events");
     private final Timer eventInsertTimer = Metrics.timer(EventElasticSearchIO.class,

blueflood-elasticsearch/src/test/java/com/rackspacecloud/blueflood/io/EventElasticSearchIOTest.java

@@ -97,21 +97,21 @@ public void testRangeOnlySearch() throws Exception {
         final int eventCountToCapture = TENANT_RANGE_EVENTS_NUM / 2;
         final int secondsDelta = 10;
         DateTime fromDateTime = new DateTime().minusSeconds(RANGE_STEP_IN_SECONDS * eventCountToCapture - secondsDelta);
-        query.put(Event.fromParameterName, Arrays.asList(Long.toString(fromDateTime.getMillis() / 1000)));
+        query.put(Event.fromParameterName, Arrays.asList(Long.toString(fromDateTime.getMillis())));
         List<Map<String, Object>> results = searchIO.search(TENANT_RANGE, query);
         Assert.assertEquals(eventCountToCapture, results.size());
 
         DateTime untilDateTime = new DateTime().minusSeconds(RANGE_STEP_IN_SECONDS * eventCountToCapture - secondsDelta);
         query.clear();
-        query.put(Event.untilParameterName, Arrays.asList(Long.toString(untilDateTime.getMillis() / 1000)));
+        query.put(Event.untilParameterName, Arrays.asList(Long.toString(untilDateTime.getMillis())));
         results = searchIO.search(TENANT_RANGE, query);
         Assert.assertEquals(eventCountToCapture, results.size());
 
         query.clear();
         fromDateTime = new DateTime().minusSeconds(RANGE_STEP_IN_SECONDS * 2 - secondsDelta);
         untilDateTime = new DateTime().minusSeconds(RANGE_STEP_IN_SECONDS - secondsDelta);
-        query.put(Event.fromParameterName, Arrays.asList(Long.toString(fromDateTime.getMillis() / 1000)));
-        query.put(Event.untilParameterName, Arrays.asList(Long.toString(untilDateTime.getMillis() / 1000)));
+        query.put(Event.fromParameterName, Arrays.asList(Long.toString(fromDateTime.getMillis())));
+        query.put(Event.untilParameterName, Arrays.asList(Long.toString(untilDateTime.getMillis())));
         results = searchIO.search(TENANT_RANGE, query);
         Assert.assertEquals(1, results.size());
     }
@@ -139,7 +139,7 @@ private void createTestEvents(final String tenant, int eventCount) throws Except
         for (int i=0; i<eventCount; i++) {
             Event event = new Event();
             event.setWhat(String.format("[%s] %s %d", tenant, "Event title sample", i));
-            event.setWhen(date.getMillis() / 1000);
+            event.setWhen(date.getMillis());
             event.setData(String.format("[%s] %s %d", tenant, "Event data sample", i));
             event.setTags(String.format("[%s] %s %d", tenant, "Event tags sample", i));
 
@@ -155,7 +155,7 @@ private void createRangeEvents(String tenant, int eventCount, int stepInSeconds)
         for (int i=0;i<eventCount; i++) {
             Event event = new Event();
             event.setWhat("1");
-            event.setWhen(date.getMillis() / 1000);
+            event.setWhen(date.getMillis());
             event.setData("2");
             event.setTags("event");
             eventList.add(event.toMap());