Merge pull request cfpb#1207 from jmarin/2016-lar-loader

2016 lar loader

Author: Nick Grippin

Documents/2016_Data_load.md

@@ -0,0 +1,33 @@
+# 2016 HMDA DATA LOAD
+
+The 2016 LAR data is necessary for some macro edits that will make comparisons between the file submitted and the previous year's data.
+
+In order to load the data, the 2017 panel should be loaded already. The LAR Loader will iterate through a file directory where the `HMDA` files reside,
+and send their contents to the `HMDA Platform`. The `HMDA` files must be in the pipe delimited format, with `.txt` extension and contain both Transmittal Sheet
+and Loan Application Register data, as per the [2017 Filing Instruction Guide](https://www.consumerfinance.gov/data-research/hmda/static/for-filers/2017/2017-HMDA-FIG.pdf)
+
+
+## Loading LAR data
+
+By default, the loader will use `localhost` as the host where the HMDA Platform cluster is running. For remote systems, use the `HMDA_CLUSTER_HOST` environment variable.
+
+When starting the cluster, take note of the port that Akka uses to communicate (i.e. 2551). This value needs to be set in an environment variable.
+The persistence cluster role needs to be active, and its dependencies (`Zookeeper`, `Cassandra`) running and properly configured.
+
+```shell
+$ export HMDA_CLUSTER_PORT=2551
+$ sbt
+> project loader
+> run /tmp/sample_lar 2016
+```
+
+This will connect to the running cluster and send the contents of each file in the directory to the appropriate Actor responsible for storing `LARs` in the 2016 period.
+
+## Troubleshooting
+
+* In cases where there is no connection, make sure that the `HmdaSupervisor` and `HmdaQuerySupervisor` are receiving messages from the cluster client. Both of these actors
+need to have the Cluster Receptionist enabled. For more information please consult the current Akka documentation on [Cluster Client](https://doc.akka.io/docs/akka/current/scala/cluster-client.html)
+
+* When loading large number of files, it might be necessary to increase the `hmda.persistent-actor-timeout` in the `Persistence-model` project's configuration file (or by passing the relevant
+runtime flag in a deployed application).
+

api/src/main/scala/hmda/api/tcp/admin/InstitutionAdminTcpApi.scala

@@ -46,7 +46,7 @@ class InstitutionAdminTcpApi(supervisor: ActorRef) extends TcpApi with FlowUtils
       .via(byte2StringFlow)
       .map(x => InstitutionParser(x))
       .mapAsync(parallelism = buffer)(i => createInstitution(i))
-      .mapAsync(parallelism = buffer)(i => createFiling(i))
+      .mapAsync(parallelism = buffer)(i => createFilings(i))
       .map(e => ByteString(e.toString))
 
   override val tcp: Future[Tcp.ServerBinding] = Tcp().bindAndHandle(
@@ -67,13 +67,17 @@ class InstitutionAdminTcpApi(supervisor: ActorRef) extends TcpApi with FlowUtils
     } yield i
   }
 
-  private def createFiling(institution: Institution): Future[Filing] = {
+  private def createFilings(institution: Institution): Future[Filing] = {
     val fFilingPersistence = (supervisor ? FindFilings(FilingPersistence.name, institution.id)).mapTo[ActorRef]
     for {
       actor <- fFilingPersistence
       f <- (actor ? CreateFiling(Filing(institution.activityYear.toString, institution.id)))
         .mapTo[Option[Filing]]
         .map(x => x.getOrElse(Filing()))
+
+      _ <- (actor ? CreateFiling(Filing((institution.activityYear - 1).toString, institution.id)))
+        .mapTo[Option[Filing]]
+        .map(x => x.getOrElse(Filing()))
     } yield f
   }
 

api/src/test/resources/application.conf

@@ -4,6 +4,14 @@ akka {
   testKit.filter-leeway = 10s
   http.parsing.max-content-length = 1G
   http.server.default-host-header = "cfpb.gov"
+  actor {
+    provider = cluster
+  }
+  remote {
+    netty.tcp {
+      port = 0
+    }
+  }
 }
 
 hmda {

build.sbt

@@ -147,6 +147,7 @@ lazy val loader = (project in file("loader"))
     )
   ).dependsOn(parserJVM % "compile->compile;test->test")
   .dependsOn(apiModel % "compile->compile;test->test")
+  .dependsOn(query)
 
 lazy val persistenceModel = (project in file("persistence-model"))
   .settings(hmdaBuildSettings:_*)

cluster/src/main/resources/application.conf

@@ -1,6 +1,7 @@
 akka {
   loggers = ["akka.event.slf4j.Slf4jLogger"]
   loglevel = "DEBUG"
+  loglevel = ${?HMDA_LOGLEVEL}
   logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
   http.parsing.max-content-length = 1G
   http.server.default-host-header = "cfpb.gov"
@@ -39,7 +40,7 @@ akka {
 
   }
 
-  extensions = ["de.heikoseeberger.constructr.ConstructrExtension"]
+  extensions = ["de.heikoseeberger.constructr.ConstructrExtension", "akka.cluster.client.ClusterClientReceptionist"]
 
 }
 
@@ -64,6 +65,7 @@ clustering {
   name = "hmda"
   name = ${?HMDA_CLUSTER_NAME}
   ip = "127.0.0.1"
+  ip = ${?HMDA_CLUSTER_IP}
   port = 0
   port = ${?APP_PORT}
 }

loader/src/main/resources/application.conf

@@ -1,12 +1,18 @@
 akka {
-  loglevel = "INFO"
+  loglevel = "DEBUG"
   loglevel = ${?HMDA_LOGLEVEL}
   actor.warn-about-java-serializer-usage = off
+  actor.provider = cluster
+
+  extensions = ["akka.cluster.client.ClusterClientReceptionist"]
 }
 
 hmda {
+  clusterName = "hmda"
+  clusterName = ${?HMDA_CLUSTER_NAME}
   adminUrl = "http://0.0.0.0:8081"
   adminUrl = ${?HMDA_HTTP_ADMIN_URL}
+  actorTimeout = 5
   httpTimeout = 5
   panel {
     tcp {
@@ -16,5 +22,12 @@ hmda {
       port = ${?HMDA_PANEL_LOADER_PORT}
     }
   }
+  lar {
+    host = "127.0.0.1"
+    host = ${?HMDA_CLUSTER_HOST}
+    port = 0
+    port = ${?HMDA_CLUSTER_PORT}
+    parallelism = 5
+  }
 }
 

loader/src/main/scala/hmda/loader/lar/HmdaBatchLarLoader.scala

@@ -0,0 +1,140 @@
+package hmda.loader.lar
+
+import java.io.File
+import java.time.Instant
+
+import akka.pattern.ask
+import akka.actor.{ ActorPath, ActorRef, ActorSystem }
+import akka.cluster.client.{ ClusterClient, ClusterClientSettings }
+import akka.stream.{ ActorMaterializer, IOResult }
+import akka.stream.scaladsl.{ FileIO, Sink, Source }
+import akka.util.{ ByteString, Timeout }
+import hmda.api.util.FlowUtils
+import hmda.model.fi.{ Created, Filing, Submission }
+import hmda.persistence.HmdaSupervisor.{ FindFilings, FindHmdaFiling, FindProcessingActor, FindSubmissions }
+import hmda.persistence.institutions.FilingPersistence.CreateFiling
+import hmda.persistence.institutions.SubmissionPersistence.CreateSubmission
+import hmda.persistence.institutions.{ FilingPersistence, SubmissionPersistence }
+import hmda.persistence.processing.HmdaRawFile.AddLine
+import hmda.persistence.processing.ProcessingMessages.{ CompleteUpload, Persisted, StartUpload }
+import hmda.persistence.processing.SubmissionManager
+import hmda.persistence.messages.CommonMessages._
+import hmda.persistence.processing.SubmissionManager.AddFileName
+import hmda.query.HmdaQuerySupervisor.FindHmdaFilingView
+import org.slf4j.LoggerFactory
+
+import scala.concurrent.duration._
+import scala.concurrent.{ ExecutionContext, Future }
+import scala.util.{ Failure, Success }
+
+object HmdaBatchLarLoader extends FlowUtils {
+
+  override implicit val system: ActorSystem = ActorSystem("hmda-cluster-client")
+  override implicit val materializer: ActorMaterializer = ActorMaterializer()
+  override implicit val ec: ExecutionContext = system.dispatcher
+
+  val hmdaClusterName = config.getString("hmda.clusterName")
+  val hmdaClusterIP = config.getString("hmda.lar.host")
+  val hmdaClusterPort = config.getInt("hmda.lar.port")
+  val actorTimeout = config.getInt("hmda.actorTimeout")
+  val flowParallelism = config.getInt("hmda.lar.parallelism")
+
+  implicit val timeout = Timeout(actorTimeout.seconds)
+
+  val log = LoggerFactory.getLogger("hmda-lar-loader")
+
+  val initialContacts = Set(
+    ActorPath.fromString(s"akka.tcp://$hmdaClusterName@$hmdaClusterIP:$hmdaClusterPort/system/receptionist")
+  )
+
+  val settings = ClusterClientSettings(system)
+    .withInitialContacts(initialContacts)
+
+  val clusterClient = system.actorOf(ClusterClient.props(settings), "hmda-lar-loader")
+
+  def main(args: Array[String]): Unit = {
+
+    if (args.length < 2) {
+      exitSys(log, "Please provide a directory containing files with LAR data and period to process", 1)
+    }
+
+    val path = new File(args(0))
+    if (!path.isDirectory) {
+      exitSys(log, "Argument must be the full path to a directory containing files with LAR data", 2)
+    }
+
+    val period = args(1)
+
+    val fileList = path.listFiles().toSet
+    val fileNames = fileList.map(file => file.getName).filter(name => name.endsWith(".txt"))
+
+    fileNames.foreach(fileName => processLars(new File(s"$path/$fileName"), s"$fileName", s"${fileName.substring(0, fileName.indexOf("."))}", period))
+
+  }
+
+  private def processLars(file: File, fileName: String, institutionId: String, period: String) = {
+    val uploadTimestamp = Instant.now.toEpochMilli
+    val source = FileIO.fromPath(file.toPath)
+
+    val fSubmissionsActor = (clusterClient ? ClusterClient
+      .Send(
+        "/user/supervisor/singleton",
+        FindSubmissions(SubmissionPersistence.name, institutionId, period),
+        localAffinity = true
+      )).mapTo[ActorRef]
+
+    (clusterClient ? ClusterClient.Send("/user/supervisor/singleton", FindHmdaFiling(period), localAffinity = true)).mapTo[ActorRef]
+    (clusterClient ? ClusterClient.Send("/user/query-supervisor", FindHmdaFilingView(period), localAffinity = true)).mapTo[ActorRef]
+
+    val fUploadSubmission = for {
+      s <- fSubmissionsActor
+      fSubmission <- (s ? CreateSubmission).mapTo[Option[Submission]]
+      submission = fSubmission.getOrElse(Submission())
+    } yield (submission, submission.status == Created)
+
+    fUploadSubmission.onComplete {
+      case Success((submission, true)) =>
+        val findSubmissionManager = FindProcessingActor(SubmissionManager.name, submission.id)
+        val findFilingPersistence = FindFilings(FilingPersistence.name, institutionId)
+        val fProcessingActor = (clusterClient ? ClusterClient.Send("/user/supervisor/singleton", findSubmissionManager, localAffinity = true)).mapTo[ActorRef]
+        val fFilingPersistence = (clusterClient ? ClusterClient.Send("/user/supervisor/singleton", findFilingPersistence, localAffinity = true)).mapTo[ActorRef]
+
+        for {
+          f <- fFilingPersistence
+          p <- fProcessingActor
+          _ <- (f ? CreateFiling(Filing(period, institutionId))).mapTo[Option[Filing]]
+        } yield {
+          uploadData(p, uploadTimestamp, fileName, submission, source)
+        }
+
+      case Success((_, false)) =>
+        log.error("submission not available for upload")
+        sys.exit(0)
+
+      case Failure(error) =>
+        log.error(error.getLocalizedMessage)
+        sys.exit(0)
+
+    }
+  }
+
+  private def uploadData(processingActor: ActorRef, uploadTimestamp: Long, fileName: String, submission: Submission, source: Source[ByteString, Future[IOResult]]): Unit = {
+    processingActor ! AddFileName(fileName)
+    processingActor ! StartUpload
+    val uploadedF = source
+      .via(framing)
+      .map(_.utf8String)
+      .mapAsync(parallelism = flowParallelism)(line => (processingActor ? AddLine(uploadTimestamp, line)).mapTo[Persisted.type])
+      .runWith(Sink.ignore)
+
+    uploadedF.onComplete {
+      case Success(_) =>
+        processingActor ! CompleteUpload
+
+      case Failure(error) =>
+        processingActor ! Shutdown
+        log.error(error.getLocalizedMessage)
+    }
+  }
+
+}

persistence/src/main/scala/hmda/persistence/HmdaSupervisor.scala

@@ -1,6 +1,7 @@
 package hmda.persistence
 
 import akka.actor.{ ActorRef, ActorSystem, Props }
+import akka.cluster.client.ClusterClientReceptionist
 import hmda.model.fi.SubmissionId
 import hmda.persistence.institutions.{ FilingPersistence, InstitutionPersistence, SubmissionPersistence }
 import hmda.persistence.model.HmdaSupervisorActor
@@ -25,6 +26,8 @@ class HmdaSupervisor(validationStats: ActorRef) extends HmdaSupervisorActor {
 
   import HmdaSupervisor._
 
+  ClusterClientReceptionist(context.system).registerService(self)
+
   override def receive: Receive = super.receive orElse {
 
     case FindHmdaFiling(filingPeriod) =>

persistence/src/test/resources/application.conf

@@ -4,6 +4,15 @@ akka {
   log-dead-letters = off
   log-dead-letters-during-shutdown = off
   actor.warn-about-java-serializer-usage = off
+
+  actor {
+    provider = cluster
+  }
+  remote {
+    netty.tcp {
+      port = 0
+    }
+  }
 }
 
 akka.persistence.journal.plugin = "inmemory-journal"

query/src/main/scala/hmda/query/HmdaQuerySupervisor.scala

@@ -1,12 +1,14 @@
 package hmda.query
 
 import akka.actor.{ ActorRef, ActorSystem, Props }
+import akka.cluster.client.ClusterClientReceptionist
 import akka.util.Timeout
 import hmda.persistence.model.HmdaSupervisorActor
 import hmda.query.view.filing.HmdaFilingView
 import hmda.query.view.institutions.InstitutionView
 import hmda.persistence.PersistenceConfig._
 import hmda.persistence.messages.CommonMessages._
+
 import scala.concurrent.duration._
 
 object HmdaQuerySupervisor {
@@ -28,6 +30,8 @@ class HmdaQuerySupervisor extends HmdaSupervisorActor {
   implicit val timeout = Timeout(duration.seconds)
   implicit val ec = context.dispatcher
 
+  ClusterClientReceptionist(context.system).registerService(self)
+
   override def receive: Receive = super.receive orElse {
     case FindHmdaFilingView(period) =>
       sender() ! findHmdaFilingView(period)

query/src/test/resources/application.conf

@@ -8,6 +8,12 @@ akka {
   persistence.snapshot-store.plugin = "inmemory-snapshot-store"
   actor {
     timeout = 5
+    provider = cluster
+  }
+  remote {
+    netty.tcp {
+      port = 0
+    }
   }
 }