Adding support for Dhalion to fetch metrics from multiple metrics providers.

src/main/java/com/microsoft/dhalion/CompositeMetricsProvider.java

@@ -0,0 +1,109 @@
+package com.microsoft.dhalion;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.microsoft.dhalion.api.MetricsProvider;
+import com.microsoft.dhalion.conf.Config;
+import com.microsoft.dhalion.conf.Key;
+import com.microsoft.dhalion.core.Measurement;
+
+import javax.inject.Inject;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class CompositeMetricsProvider implements MetricsProvider {
+
+  private List<MetricsProvider> metricsProviders;
+  private Map<String, MetricsProvider> metricsProviderMap;
+  protected Config sysConfig;
+
+  @Inject
+  public CompositeMetricsProvider(Config sysConf) throws ClassNotFoundException {
+    sysConfig = sysConf;
+    metricsProviders = new ArrayList<>();
+    metricsProviderMap = new HashMap<>();
+    instantiateMetricsProviders();
+  }
+
+  @VisibleForTesting
+  protected void instantiateMetricsProviders() throws ClassNotFoundException {
+    Injector injector = Guice.createInjector(new AbstractModule() {
+      @Override
+      protected void configure() {
+        bind(Config.class).toInstance(sysConfig);
+      }
+    });
+    String metricsProviderClasses = (String) sysConfig.get(Key.METRICS_PROVIDER_CLASS.value());
+    if (!metricsProviderClasses.isEmpty()) {
+      String[] mpClasses = metricsProviderClasses.split(",");
+      for (String mpClass : mpClasses) {
+        Class<MetricsProvider> metricsProviderClass =
+            (Class<MetricsProvider>) this.getClass().getClassLoader().loadClass(mpClass);
+        addMetricsProvider(injector.getInstance(metricsProviderClass));
+      }
+    }
+  }
+
+  protected void addMetricsProvider(MetricsProvider provider) {
+    metricsProviders.add(provider);
+  }
+
+  @Override
+  public void initialize() {
+    for (MetricsProvider metricsProvider : metricsProviders) {
+      metricsProvider.initialize();
+      Set<String> metricTypes = metricsProvider.getMetricTypes();
+      if (metricTypes != null) {
+        for (String metricType : metricsProvider.getMetricTypes()) {
+          metricsProviderMap.put(metricType, metricsProvider);
+        }
+      }
+    }
+  }
+
+  @Override
+  public Collection<Measurement> getMeasurements(Instant startTime,
+      Duration duration, Collection<String> metrics,
+      Collection<String> components) {
+    Collection<Measurement> measurements = new ArrayList<>();
+    for (String metric : metrics) {
+      MetricsProvider provider = metricsProviderMap.get(metric);
+      if (provider != null) {
+        measurements.addAll(provider.getMeasurements(startTime, duration,
+            Collections.singletonList(metric), components));
+      }
+    }
+    return measurements;
+  }
+
+  @Override
+  public Set<String> getMetricTypes() {
+    return metricsProviderMap.keySet();
+  }
+
+  @Override
+  public Collection<Measurement> getMeasurements(Instant startTime,
+      Duration duration, String metric, String component) {
+    MetricsProvider provider = metricsProviderMap.get(metric);
+    if (provider != null) {
+      return provider.getMeasurements(startTime, duration, metric, component);
+    }
+    return null;
+  }
+
+  @Override
+  public void close() {
+    for (MetricsProvider metricsProvider : metricsProviders) {
+      metricsProvider.close();
+    }
+  }
+}

src/main/java/com/microsoft/dhalion/HealthManager.java

@@ -102,19 +102,15 @@ public class HealthManager {
     cb.loadConfig(conf).loadPolicyConf();
     config = cb.build();
 
-    //Read the MetricsProvider class
-    String metricsProviderClass = (String) conf.get(Key.METRICS_PROVIDER_CLASS.value());
-    Class<MetricsProvider> mpClass
-        = (Class<MetricsProvider>) this.getClass().getClassLoader().loadClass(metricsProviderClass);
     injector = injector.createChildInjector(new AbstractModule() {
       @Override
       protected void configure() {
         bind(Config.class).toInstance(config);
-        bind(mpClass).in(Singleton.class);
+        bind(CompositeMetricsProvider.class).in(Singleton.class);
       }
     });
 
-    metricsProvider = injector.getInstance(mpClass);
+    metricsProvider = injector.getInstance(CompositeMetricsProvider.class);
 
     injector = injector.createChildInjector(new AbstractModule() {
       @Override

src/main/java/com/microsoft/dhalion/api/MetricsProvider.java

@@ -12,6 +12,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Set;
 
 /**
  * A {@link MetricsProvider} implements common utility methods to produce {@link Measurement}s. In some cases it will
@@ -44,6 +45,15 @@ public interface MetricsProvider {
     throw new UnsupportedOperationException("This method is not implemented in the metrics provider");
   }
 
+  /**
+   * Returns metric types which are supported by metrics provider.
+   *
+   * @return set of metrics types.
+   */
+  default Set<String> getMetricTypes() {
+    throw new UnsupportedOperationException("This method is not implemented in the metrics provider");
+  }
+
   /**
    * @param startTime metric aggregation window start time, endTime = startTime - duration
    * @param duration  the duration for which the metric was aggregated

src/main/java/com/microsoft/dhalion/examples/CSVMetricsProvider.java

@@ -23,6 +23,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Optional;
+import java.util.Set;
 import java.util.logging.Logger;
 
 /**
@@ -53,6 +54,13 @@ public class CSVMetricsProvider implements MetricsProvider {
     return measurements;
   }
 
+  public Set<String> getMetricTypes() {
+    Set<String> metricTypes = new HashSet<>();
+    metricTypes.add(NodeStat.MEMORY_UTILIZATION);
+    metricTypes.add(NodeStat.CPU_UTILIZATION);
+    return metricTypes;
+  }
+
   private Collection<Measurement> getMeasurements(String metric, Instant startTS, Duration duration, String
       component) {
     Collection<Measurement> metrics = new ArrayList<>();

src/test/java/com/microsoft/dhalion/CompositeMetricsProviderTest.java

@@ -0,0 +1,112 @@
+package com.microsoft.dhalion;
+
+import com.microsoft.dhalion.api.MetricsProvider;
+import com.microsoft.dhalion.conf.Config;
+import com.microsoft.dhalion.conf.Key;
+import com.microsoft.dhalion.core.Measurement;
+import com.microsoft.dhalion.core.MeasurementsTable;
+import com.microsoft.dhalion.examples.CSVMetricsProvider;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+import static com.microsoft.dhalion.core.MeasurementsTable.SortKey.TIME_STAMP;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class CompositeMetricsProviderTest {
+
+  private MetricsProvider provider;
+
+  @Before
+  public void setup() throws ClassNotFoundException {
+    Config conf = mock(Config.class);
+    when(conf.get(Key.DATA_DIR.value())).thenReturn(CompositeMetricsProvider.class.getClassLoader().getResource(".").getFile());
+    provider = new MockCompositeMetricsProviderTest(conf);
+    provider.initialize();
+  }
+
+  @Test
+  public void testGetMetricTypes() {
+    Set<String> metricTypes = provider.getMetricTypes();
+    Assert.assertEquals(metricTypes.size(), 3);
+    Assert.assertTrue(metricTypes.contains("MockMetric"));
+    Assert.assertTrue(metricTypes.contains("Cpu"));
+  }
+
+  @Test
+  public void testGetMeasurements() {
+    Instant startTS = Instant.parse("2018-01-08T01:37:36.934Z");
+    String metric = "Cpu";
+    Duration duration = Duration.ofMinutes(2);
+    String comp = "NodeA";
+
+    Collection<String> metricNames = new ArrayList<>();
+    metricNames.add(metric);
+    Collection<String> components = new ArrayList<>();
+    components.add(comp);
+
+    MeasurementsTable metrics = MeasurementsTable.of(
+        provider.getMeasurements(startTS, duration, metricNames, components));
+    assertEquals(4, metrics.size());
+    assertEquals(4, metrics.type(metric).size());
+    assertEquals(2, metrics.component(comp).instance("1").size());
+    assertEquals(2, metrics.component(comp).instance("3").size());
+
+    Iterator<Measurement> measurements = metrics.component(comp).instance("1").sort(false, TIME_STAMP).get().iterator();
+    assertEquals("2018-01-08T01:36:36.934Z", measurements.next().instant().toString());
+    assertEquals("2018-01-08T01:37:36.934Z", measurements.next().instant().toString());
+
+    metrics = MeasurementsTable.of(
+        provider.getMeasurements(startTS, duration, "MockMetric", "abc"));
+    assertEquals(metrics.size(), 1);
+
+    Assert.assertNull(
+        provider.getMeasurements(startTS, duration, "WrongMetric", "abc"));
+  }
+
+  class MockCompositeMetricsProviderTest extends CompositeMetricsProvider {
+    MockCompositeMetricsProviderTest(Config sysConf) throws ClassNotFoundException {
+      super(sysConf);
+    }
+
+    @Override
+    protected void instantiateMetricsProviders() throws ClassNotFoundException {
+      addMetricsProvider(new CSVMetricsProvider(sysConfig));
+      addMetricsProvider(new MockMetricsProvider());
+    }
+  }
+
+  class MockMetricsProvider implements MetricsProvider {
+    @Override public Collection<Measurement> getMeasurements(Instant startTime,
+        Duration duration, Collection<String> metrics,
+        Collection<String> components) {
+      List<Measurement> measurements = new ArrayList<>();
+      for (String component : components) {
+        for (String metric : metrics) {
+          measurements.addAll(getMeasurements(startTime, duration, metric, component));
+        }
+      }
+      return measurements;
+    }
+
+    @Override public Set<String> getMetricTypes() {
+      return Collections.singleton("MockMetric");
+    }
+
+    @Override public Collection<Measurement> getMeasurements(Instant startTime,
+        Duration duration, String metric, String component) {
+      return Collections.singleton(new Measurement(component, "1", metric, startTime, 100));
+    }
+  }
+}