Merge pull request cfpb#1109 from jmarin/cluster-roles

Cluster roles

Author: schbetsy

.travis.yml

@@ -8,7 +8,7 @@ cache:
 
 language: scala
 
-scala: 2.12.1
+scala: 2.12.3
 
 jdk:
 - oraclejdk8

README.md

@@ -181,7 +181,7 @@ $ sbt
 
 ```shell
 
-> project api
+> project cluster
 > clean
 > ~re-start
 ```

api/src/main/resources/application-dev.conf

@@ -5,36 +5,6 @@ akka {
   logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
   http.parsing.max-content-length = 1G
   http.server.default-host-header = "cfpb.gov"
-
-  actor {
-    provider = cluster
-  }
-
-  remote {
-    log-remote-lifecycle-events = error
-    netty.tcp {
-      hostname = ${clustering.ip}
-      port = ${clustering.port}
-    }
-  }
-
-  cluster {
-    failure-detector.threshold = 12 //Increase value for AWS environments
-    metrics.enabled = off
-  }
-
-  extensions = ["de.heikoseeberger.constructr.ConstructrExtension"]
-
-}
-
-constructr.coordination.nodes = [${hmda.zookeeperHost}":"${hmda.zookeeperPort}]
-
-clustering {
-  name = "hmda"
-  name = ${?HMDA_CLUSTER_NAME}
-  ip = "127.0.0.1"
-  port = 0
-  port = ${?APP_PORT}
 }
 
 hmda {
@@ -59,10 +29,6 @@ hmda {
   }
   isDemo = true
   actor-flow-parallelism = 4
-  zookeeperHost = "192.168.99.100"
-  zookeeperHost = ${?ZOOKEEPER_HOST}
-  zookeeperPort = 2181
-  zookeeperPort = ${?ZOOKEEPER_PORT}
 }
 
 

api/src/main/resources/application.conf

@@ -5,46 +5,6 @@ akka {
   logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
   http.parsing.max-content-length = 1G
   http.server.default-host-header = "cfpb.gov"
-
-  actor {
-    provider = cluster
-  }
-
-  remote {
-    log-remote-lifecycle-events = error
-    netty.tcp {
-      hostname = ${clustering.ip}
-      port = ${clustering.port}
-    }
-  }
-
-  cluster {
-    failure-detector.threshold = 12 //Increase value for AWS environments
-    metrics.enabled = off
-
-    retry-unsuccessful-join-after = 20s
-
-    http {
-      management {
-        hostname = "127.0.0.1"
-        port = 9999
-      }
-    }
-
-  }
-
-  extensions = ["de.heikoseeberger.constructr.ConstructrExtension"]
-
-}
-
-constructr.coordination.nodes = [${hmda.zookeeperHost}":"${hmda.zookeeperPort}]
-
-clustering {
-  name = "hmda"
-  name = ${?HMDA_CLUSTER_NAME}
-  ip = "127.0.0.1"
-  port = 0
-  port = ${?APP_PORT}
 }
 
 hmda {
@@ -69,10 +29,6 @@ hmda {
   }
   isDemo = false
   isDemo = ${?HMDA_IS_DEMO}
-  zookeeperHost = "192.168.99.100"
-  zookeeperHost = ${?ZOOKEEPER_HOST}
-  zookeeperPort = 2181
-  zookeeperPort = ${?ZOOKEEPER_PORT}
 }
 
 api-dispatcher {

api/src/main/scala/hmda/api/HmdaAdminApi.scala

@@ -1,6 +1,6 @@
 package hmda.api
 
-import akka.actor.{ ActorSystem, Props }
+import akka.actor.{ ActorSystem, Props, ActorRef }
 import akka.event.Logging
 import akka.pattern.pipe
 import akka.http.scaladsl.Http
@@ -16,10 +16,10 @@ import scala.concurrent.duration._
 import scala.concurrent.{ ExecutionContext, Future }
 
 object HmdaAdminApi {
-  def props(): Props = Props(new HmdaAdminApi)
+  def props(supervisor: ActorRef, querySupervisor: ActorRef): Props = Props(new HmdaAdminApi(supervisor, querySupervisor))
 }
 
-class HmdaAdminApi extends HttpApi with BaseHttpApi with InstitutionAdminHttpApi {
+class HmdaAdminApi(supervisor: ActorRef, querySupervisor: ActorRef) extends HttpApi with BaseHttpApi with InstitutionAdminHttpApi {
 
   val config = ConfigFactory.load()
 
@@ -36,7 +36,7 @@ class HmdaAdminApi extends HttpApi with BaseHttpApi with InstitutionAdminHttpApi
   override implicit val ec: ExecutionContext = context.dispatcher
   override val log = Logging(system, getClass)
 
-  override val paths: Route = routes(s"$name") ~ institutionAdminRoutes
+  override val paths: Route = routes(s"$name") ~ institutionAdminRoutes(supervisor, querySupervisor)
 
   override val http: Future[ServerBinding] = Http(system).bindAndHandle(
     paths,

api/src/main/scala/hmda/api/HmdaFilingApi.scala

@@ -1,6 +1,6 @@
 package hmda.api
 
-import akka.actor.{ ActorSystem, Props }
+import akka.actor.{ ActorRef, ActorSystem, Props }
 import akka.event.Logging
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.Http.ServerBinding
@@ -10,15 +10,17 @@ import akka.pattern.pipe
 import akka.stream.ActorMaterializer
 import akka.util.Timeout
 import hmda.api.http.{ BaseHttpApi, HmdaCustomDirectives, InstitutionsHttpApi, LarHttpApi }
-import hmda.api.HmdaConfig._
+
 import scala.concurrent.duration._
 import scala.concurrent.{ ExecutionContext, Future }
+import HmdaConfig._
 
 object HmdaFilingApi {
-  def props(): Props = Props(new HmdaFilingApi)
+  def props(supervisor: ActorRef, querySupervisor: ActorRef, validationStats: ActorRef): Props =
+    Props(new HmdaFilingApi(supervisor, querySupervisor, validationStats))
 }
 
-class HmdaFilingApi
+class HmdaFilingApi(supervisor: ActorRef, querySupervisor: ActorRef, validationStats: ActorRef)
     extends HttpApi
     with BaseHttpApi
     with LarHttpApi
@@ -40,7 +42,7 @@ class HmdaFilingApi
   implicit val ec: ExecutionContext = context.dispatcher
   override val log = Logging(system, getClass)
 
-  val paths: Route = routes(s"$name") ~ larRoutes ~ institutionsRoutes
+  val paths: Route = routes(s"$name") ~ institutionsRoutes(supervisor, querySupervisor, validationStats) ~ larRoutes(supervisor)
 
   override val http: Future[ServerBinding] = Http(system).bindAndHandle(
     paths,

api/src/main/scala/hmda/api/HmdaPlatform.scala

@@ -2,7 +2,7 @@ package hmda.api
 
 import akka.actor.{ ActorRef, ActorSystem, Props }
 import akka.event.Logging
-import akka.pattern.{ pipe, ask }
+import akka.pattern.{ ask, pipe }
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.Http.ServerBinding
 import akka.http.scaladsl.server.Route
@@ -14,7 +14,6 @@ import hmda.api.http.public.{ InstitutionSearchPaths, PublicHttpApi }
 import hmda.persistence.model.HmdaSupervisorActor.FindActorByName
 import hmda.query.view.institutions.InstitutionView
 import akka.http.scaladsl.server.Directives._
-
 import scala.concurrent.duration._
 import scala.concurrent.{ ExecutionContext, Future }
 

api/src/main/scala/hmda/api/HmdaPublicApi.scala

@@ -1,5 +1,6 @@
 package hmda.api.http
 
+import akka.actor.ActorRef
 import akka.actor.ActorSystem
 import akka.event.LoggingAdapter
 import akka.http.scaladsl.server.Directives._
@@ -31,29 +32,29 @@ trait InstitutionsHttpApi
 
   implicit val timeout: Timeout
 
-  val institutionsRoutes =
+  def institutionsRoutes(supervisor: ActorRef, querySupervisor: ActorRef, validationStats: ActorRef) =
     extractExecutionContext { executor =>
       implicit val ec: ExecutionContext = executor
       encodeResponse {
         headerAuthorize {
-          institutionsPath ~
+          institutionsPath(querySupervisor) ~
             pathPrefix("institutions" / Segment) { instId =>
               institutionAuthorize(instId) {
-                institutionByIdPath(instId) ~
-                  filingByPeriodPath(instId) ~
-                  submissionPath(instId) ~
-                  submissionLatestPath(instId) ~
-                  uploadPath(instId) ~
-                  submissionEditsPath(instId) ~
-                  submissionParseErrorsPath(instId) ~
-                  submissionEditCsvPath(instId) ~
-                  submissionSingleEditPath(instId) ~
-                  editFailureDetailsPath(instId) ~
-                  verifyEditsPath(instId) ~
-                  submissionIrsPath(instId) ~
-                  submissionIrsCsvPath(instId) ~
-                  submissionSignPath(instId) ~
-                  submissionSummaryPath(instId)
+                institutionByIdPath(supervisor, querySupervisor, instId) ~
+                  filingByPeriodPath(supervisor, instId) ~
+                  submissionPath(supervisor, instId) ~
+                  submissionLatestPath(supervisor, instId) ~
+                  uploadPath(supervisor, querySupervisor, instId) ~
+                  submissionEditsPath(supervisor, querySupervisor, instId) ~
+                  submissionParseErrorsPath(supervisor, querySupervisor, instId) ~
+                  submissionEditCsvPath(supervisor, querySupervisor, instId) ~
+                  submissionSingleEditPath(supervisor, querySupervisor, instId) ~
+                  editFailureDetailsPath(supervisor, querySupervisor, instId) ~
+                  verifyEditsPath(supervisor, querySupervisor, instId) ~
+                  submissionIrsPath(supervisor, querySupervisor, validationStats, instId) ~
+                  submissionIrsCsvPath(supervisor, querySupervisor, validationStats, instId) ~
+                  submissionSignPath(supervisor, querySupervisor, instId) ~
+                  submissionSummaryPath(supervisor, querySupervisor, instId)
               }
             }
         }