Add Feat/gateway cluster monitoring

gateway-ha/src/main/java/com/lyft/data/gateway/ha/clustermonitor/HealthChecker.java

@@ -1,11 +1,17 @@
 package com.lyft.data.gateway.ha.clustermonitor;
 
+import com.lyft.data.gateway.ha.config.DataStoreConfiguration;
 import com.lyft.data.gateway.ha.notifier.Notifier;
+import com.lyft.data.gateway.ha.persistence.JdbcConnectionManager;
+import com.lyft.data.gateway.ha.router.GatewayBackendStateManager.GatewayBackendState;
+import com.lyft.data.gateway.ha.router.HaGatewayBackendStateManager;
 import java.util.List;
 
+
 public class HealthChecker implements PrestoClusterStatsObserver {
   private static final int MAX_THRESHOLD_QUEUED_QUERY_COUNT = 100;
   private Notifier notifier;
+  private HaGatewayBackendStateManager haGatewayBackendStateManager;
 
   public HealthChecker(Notifier notifier) {
     this.notifier = notifier;
@@ -24,24 +30,26 @@ public void observe(List<ClusterStats> clustersStats) {
           notifyForNoWorkers(clusterStats);
         }
       }
+      GatewayBackendState backend = new GatewayBackendState();
+      backend.setName(clusterStats.getClusterId());
+      backend.setHealth(clusterStats.isHealthy());
+      backend.setWorkerCount(clusterStats.getNumWorkerNodes());
+      GatewayBackendState updated = haGatewayBackendStateManager.addBackend(backend);
     }
   }
-
   private void notifyUnhealthyCluster(ClusterStats clusterStats) {
     notifier.sendNotification(String.format("%s - Cluster unhealthy",
-        clusterStats.getClusterId()),
-        clusterStats.toString());
+      clusterStats.getClusterId()),
+      clusterStats.toString());
   }
 
   private void notifyForTooManyQueuedQueries(ClusterStats clusterStats) {
     notifier.sendNotification(String.format("%s - Too many queued queries",
-        clusterStats.getClusterId()), clusterStats.toString());
+      clusterStats.getClusterId()), clusterStats.toString());
   }
 
   private void notifyForNoWorkers(ClusterStats clusterStats) {
     notifier.sendNotification(String.format("%s - Number of workers",
-        clusterStats.getClusterId()), clusterStats.toString());
+      clusterStats.getClusterId()), clusterStats.toString());
   }
-
-
 }

gateway-ha/src/main/java/com/lyft/data/gateway/ha/module/HaGatewayProviderModule.java

@@ -9,6 +9,8 @@
 import com.lyft.data.gateway.ha.handler.QueryIdCachingProxyHandler;
 import com.lyft.data.gateway.ha.persistence.JdbcConnectionManager;
 import com.lyft.data.gateway.ha.router.GatewayBackendManager;
+import com.lyft.data.gateway.ha.router.GatewayBackendStateManager;
+import com.lyft.data.gateway.ha.router.HaGatewayBackendStateManager;
 import com.lyft.data.gateway.ha.router.HaGatewayManager;
 import com.lyft.data.gateway.ha.router.HaQueryHistoryManager;
 import com.lyft.data.gateway.ha.router.HaRoutingManager;
@@ -25,6 +27,7 @@ public class HaGatewayProviderModule extends AppModule<HaGatewayConfiguration, E
   private final QueryHistoryManager queryHistoryManager;
   private final RoutingManager routingManager;
   private final JdbcConnectionManager connectionManager;
+	private final GatewayBackendStateManager gatewayBackendStateManager;
 
   public HaGatewayProviderModule(HaGatewayConfiguration configuration, Environment environment) {
     super(configuration, environment);
@@ -33,6 +36,7 @@ public HaGatewayProviderModule(HaGatewayConfiguration configuration, Environment
     queryHistoryManager = new HaQueryHistoryManager(connectionManager);
     routingManager =
         new HaRoutingManager(gatewayBackendManager, (HaQueryHistoryManager) queryHistoryManager);
+		gatewayBackendStateManager = new HaGatewayBackendStateManager(connectionManager);
   }
 
   protected ProxyHandler getProxyHandler() {
@@ -84,10 +88,16 @@ public RoutingManager getRoutingManager() {
     return this.routingManager;
   }
 
+	@Provides
+	@Singleton
+	public GatewayBackendStateManager getGatewayBackendStateManager() {
+		return this.gatewayBackendStateManager;
+	}
+
   @Provides
   @Singleton
   public JdbcConnectionManager getConnectionManager() {
     return this.connectionManager;
   }
 
-}
+}

gateway-ha/src/main/java/com/lyft/data/gateway/ha/persistence/dao/GatewayBackendStateHistory.java

@@ -0,0 +1,39 @@
+package com.lyft.data.gateway.ha.persistence.dao;
+
+import com.lyft.data.gateway.ha.router.GatewayBackendStateManager.GatewayBackendState;
+import java.util.ArrayList;
+import java.util.List;
+import org.javalite.activejdbc.Model;
+import org.javalite.activejdbc.annotations.Cached;
+import org.javalite.activejdbc.annotations.IdName;
+import org.javalite.activejdbc.annotations.Table;
+
+@Table("gateway_backend_state")
+@IdName("name")
+@Cached
+public class GatewayBackendStateHistory extends Model {
+  //TODO: Create an active jdbc model along with a function to create and update entries
+  private static final String name = "name";
+  private static final String health = "health";
+  private static final String workerCount = "worker_count";
+
+  public static List<GatewayBackendState> upcast(List<GatewayBackendStateHistory>
+      gatewayBackendStateHistoryList) {
+    List<GatewayBackendState> gatewayBackendStates = new ArrayList<>();
+    for (GatewayBackendStateHistory dao : gatewayBackendStateHistoryList) {
+      GatewayBackendState gatewayBackendState = new GatewayBackendState();
+      gatewayBackendState.setName(dao.getString(name));
+      gatewayBackendState.setHealth(dao.getBoolean(health));
+      gatewayBackendState.setWorkerCount(dao.getInteger(workerCount));
+    }
+    return gatewayBackendStates;
+  }
+
+  public static void create(GatewayBackendStateHistory model,
+      GatewayBackendState gatewayBackendState) {
+    model.set(name, gatewayBackendState.getName());
+    model.set(health, gatewayBackendState.getHealth());
+    model.set(workerCount, gatewayBackendState.getWorkerCount());
+    model.insert();
+  }
+}

gateway-ha/src/main/java/com/lyft/data/gateway/ha/resource/EntityEditorResource.java

@@ -6,6 +6,7 @@
 import com.google.inject.Inject;
 import com.lyft.data.gateway.ha.config.ProxyBackendConfiguration;
 import com.lyft.data.gateway.ha.router.GatewayBackendManager;
+import com.lyft.data.gateway.ha.router.GatewayBackendStateManager;
 import io.dropwizard.views.View;
 
 import java.io.IOException;
@@ -31,6 +32,7 @@
 public class EntityEditorResource {
 
   @Inject private GatewayBackendManager gatewayBackendManager;
+	@Inject private GatewayBackendStateManager gatewayBackendStateManager;
   public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
   @GET
@@ -46,7 +48,8 @@ protected EntityView(String templateName) {
   }
 
   private enum EntityType {
-    GATEWAY_BACKEND
+    GATEWAY_BACKEND,
+		GATEWAY_BACKEND_STATE
   }
 
   @GET
@@ -87,6 +90,8 @@ public Response getAllEntitiesForType(@PathParam("entityType") String entityType
     switch (entityType) {
       case GATEWAY_BACKEND:
         return Response.ok(gatewayBackendManager.getAllBackends()).build();
+			case GATEWAY_BACKEND_STATE:
+				return Response.ok(gatewayBackendStateManager.getAllBackendStates()).build();
       default:
     }
     return Response.ok(ImmutableList.of()).build();

gateway-ha/src/main/java/com/lyft/data/gateway/ha/router/GatewayBackendStateManager.java

@@ -0,0 +1,24 @@
+package com.lyft.data.gateway.ha.router;
+
+//import com.lyft.data.gateway.ha.persistence.dao.GatewayBackendStateHistory;
+import java.util.List;
+import lombok.Data;
+import lombok.ToString;
+
+public interface GatewayBackendStateManager {
+
+  List<GatewayBackendState> getAllBackendStates();
+
+  GatewayBackendState addBackend(GatewayBackendState backend);
+
+  @Data
+  @ToString
+  class GatewayBackendState {
+
+    private String name;
+    private Boolean health;
+    private int workerCount;
+
+  }
+
+}

gateway-ha/src/main/java/com/lyft/data/gateway/ha/router/HaGatewayBackendStateManager.java

@@ -0,0 +1,41 @@
+package com.lyft.data.gateway.ha.router;
+
+import com.lyft.data.gateway.ha.persistence.JdbcConnectionManager;
+import com.lyft.data.gateway.ha.persistence.dao.GatewayBackendStateHistory;
+
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class HaGatewayBackendStateManager implements GatewayBackendStateManager {
+
+  private JdbcConnectionManager connectionManager;
+
+  public HaGatewayBackendStateManager(JdbcConnectionManager connectionManager) {
+    this.connectionManager = connectionManager;
+  }
+
+  @Override
+  public List<GatewayBackendState> getAllBackendStates() {
+    //TODO implement method to obtain all cluster states
+    try {
+      connectionManager.open();
+      List<GatewayBackendStateHistory> proxyBackendList = GatewayBackendStateHistory.findAll();
+      return GatewayBackendStateHistory.upcast(proxyBackendList);
+    } finally {
+      connectionManager.close();
+    }
+  }
+
+  @Override
+  public GatewayBackendState addBackend(GatewayBackendState backend) {
+    try {
+      connectionManager.open();
+      GatewayBackendStateHistory.create(new GatewayBackendStateHistory(), backend);
+    } finally {
+      connectionManager.close();
+    }
+
+    return backend;
+  }
+}