OY-4784 Use ytlFetchActor for nightly sync, use correct tunniste

Author: msiukola

src/main/scala/fi/vm/sade/hakurekisteri/integration/ytl/YtlFetchActor.scala

@@ -14,16 +14,18 @@ import fi.vm.sade.properties.OphProperties
 import org.apache.commons.io.IOUtils
 import scalaz.concurrent.Task
 import support.TypedActorRef
-import scala.concurrent.duration._
 
-import java.util.UUID
+import java.time.LocalDate
+import scala.concurrent.duration._
+import java.util.{Date, UUID}
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
 import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.util.{Failure, Success, Try}
 
 case class YtlSyncHaku(hakuOid: String, tunniste: String)
 
 case class YtlSyncAllHaut(tunniste: String)
+case class YtlSyncAllHautNightly(tunniste: String)
 case class YtlSyncSingle(personOid: String, tunniste: String)
 case class ActiveKkHakuOids(hakuOids: Set[String])
 case class YtlFetchActorRef(actor: ActorRef) extends TypedActorRef
@@ -40,15 +42,37 @@ class YtlFetchActor(
 
   val activeKKHakuOids = new AtomicReference[Set[String]](Set.empty)
 
+  //val lastSyncStart = new AtomicReference[Option[LocalDate]](None)
+  val lastSyncStart = new AtomicReference[Long](0)
+  val minIntervalBetween = 1000 * 60 * 22 //At least 22 hours between nightly syncs
+
   implicit val ec: ExecutionContext = ExecutorUtil.createExecutor(
     config.integrations.asyncOperationThreadPoolSize,
     getClass.getSimpleName
   )
 
   def setAktiivisetKKHaut(hakuOids: Set[String]): Unit = activeKKHakuOids.set(hakuOids)
   override def receive: Receive = {
+    case ah: YtlSyncAllHautNightly =>
+      val tunniste = ah.tunniste
+      val now = System.currentTimeMillis()
+      val lss = lastSyncStart.get()
+      val timeToStartNewSync = (lss + minIntervalBetween) < now
+      if (timeToStartNewSync) {
+        log.info(s"Starting nightly sync for all hakus. Previous run was $lss")
+        lastSyncStart.set(now)
+        val resultF = syncAllOneHakuAtATime(tunniste)
+        resultF.onComplete {
+          case Success(_) =>
+            log.info(s"($tunniste) Nightly sync for all hakus success!")
+          case Failure(t) =>
+            log.error(t, s"($tunniste) Nightly sync for all hakus failed...")
+        }
+      } else {
+        log.warning(s"Not starting nightly sync for all hakus as the previous run was on $lss")
+      }
     case ah: YtlSyncAllHaut =>
-      val tunniste = "manual_sync_for_all_hakus_" + System.currentTimeMillis()
+      val tunniste = ah.tunniste
       val resultF = syncAllOneHakuAtATime(tunniste)
       resultF.onComplete {
         case Success(_) =>
@@ -58,7 +82,7 @@ class YtlFetchActor(
       }
       sender ! tunniste
     case s: YtlSyncHaku =>
-      val tunniste = System.currentTimeMillis() + "_manual_sync_for_haku_" + s.hakuOid
+      val tunniste = s.tunniste
       val resultF = fetchAndHandleHakemuksetForSingleHakuF(hakuOid = s.hakuOid, s.tunniste)
       resultF.onComplete {
         case Success(_) =>
@@ -69,7 +93,7 @@ class YtlFetchActor(
       log.info(s"Ytl-sync käynnistetty haulle ${s.hakuOid} tunnisteella $tunniste")
       resultF pipeTo sender
     case s: YtlSyncSingle =>
-      val tunniste = System.currentTimeMillis() + "_manual_sync_for_person_" + s.personOid
+      val tunniste = s.tunniste
       val resultF = syncSingle(s.personOid)
       resultF.onComplete {
         case Success(_) =>

src/main/scala/fi/vm/sade/hakurekisteri/integration/ytl/YtlRerunPolicy.scala

@@ -10,35 +10,14 @@ import org.slf4j.LoggerFactory
 object YtlRerunPolicy {
   private val logger = LoggerFactory.getLogger(YtlRerunPolicy.getClass)
 
-  def rerunPolicy(expression: String, ytlIntegration: YtlIntegration): () => Unit = {
+  def rerunPolicy(expression: String, ytlIntegrationActor: YtlFetchActorRef): () => Unit = {
     def nextTimestamp(expression: String, d: Date) = new SimpleDateFormat("dd.MM.yyyy HH:mm")
       .format(new CronExpression(expression).getNextValidTimeAfter(d))
     logger.info(s"First YTL fetch at '${nextTimestamp(expression, new Date())}'")
 
     () => {
-      val fetchStatus = ytlIntegration.AtomicStatus.getLastStatus
-      val isRunning = fetchStatus.exists(_.inProgress)
-      if (isRunning) {
-        logger.info(
-          s"Scheduled to make YTL fetch but fetch is already running! Next try will be ${nextTimestamp(expression, new Date())}"
-        )
-      } else {
-        val isYesterday =
-          fetchStatus.exists(status => !DateUtils.isSameDay(status.start, new Date()))
-        val isSucceeded = !(fetchStatus.flatMap(_.hasFailures).getOrElse(true))
-        if ((isSucceeded && isYesterday) || (!isSucceeded)) {
-          logger.info(
-            s"Starting new YTL fetch because: last run was yesterday=$isYesterday and that run succeeded=$isSucceeded"
-          )
-          ytlIntegration.syncAll()
-        } else {
-          logger.info(
-            s"Scheduled to make YTL fetch but not running because: " +
-              s"last run was yesterday=$isYesterday and that run succeeded=$isSucceeded! " +
-              s"Next try will be ${nextTimestamp(expression, new Date())}"
-          )
-        }
-      }
+      logger.info("Calling YtlFetchActor to start nightly YLT sync")
+      ytlIntegrationActor.actor ! YtlSyncAllHautNightly("Nightly YTL sync")
     }
   }
 

src/main/scala/fi/vm/sade/hakurekisteri/web/integration/ytl/YtlResource.scala

@@ -62,7 +62,7 @@ class YtlResource(ytlIntegration: YtlIntegration, ytlFetchActor: YtlFetchActorRe
     shouldBeAdmin
     logger.info("Fetching YTL data for everybody")
     audit.log(auditUser, YTLSyncForAll, new Target.Builder().build, Changes.EMPTY)
-    val tunniste = "manual_sync_for_all_hakus_" + System.currentTimeMillis()
+    val tunniste = "manual_sync_for_all_hakus" + System.currentTimeMillis()
     ytlFetchActor.actor ! YtlSyncAllHaut(tunniste)
     Accepted(s"YTL sync started, tunniste $tunniste")
   }
@@ -83,12 +83,14 @@ class YtlResource(ytlIntegration: YtlIntegration, ytlFetchActor: YtlFetchActorRe
     logger.info(s"Fetching YTL data for person OID $personOid")
     audit.log(auditUser, YTLSyncForPerson, AuditUtil.targetFromParams(params).build, Changes.EMPTY)
     try {
-      val resultF = ytlFetchActor.actor ? YtlSyncSingle(personOid, tunniste = "sync") recoverWith {
-        case t: Throwable =>
-          logger.error(t, s"Error while ytl-syncing $personOid")
-          Future.failed(
-            new RuntimeException(s"Error while ytl-syncing $personOid: ${t.getMessage}")
-          )
+      val resultF = ytlFetchActor.actor ? YtlSyncSingle(
+        personOid,
+        tunniste = s"manual_sync_for_person_${personOid}"
+      ) recoverWith { case t: Throwable =>
+        logger.error(t, s"Error while ytl-syncing $personOid")
+        Future.failed(
+          new RuntimeException(s"Error while ytl-syncing $personOid: ${t.getMessage}")
+        )
       }
       logger.info(s"Waiting for result for YTL data for person OID $personOid")
       val result = Await.result(resultF, 30.seconds)

src/main/scala/support/Integrations.scala

@@ -584,7 +584,7 @@ class BaseIntegrations(rekisterit: Registers, system: ActorSystem, config: Confi
 
   val ytlSyncAllEnabled = OphUrlProperties.getProperty("ytl.http.syncAllEnabled").toBoolean
   val syncAllCronExpression = OphUrlProperties.getProperty("ytl.http.syncAllCronJob")
-  val rerunSync = YtlRerunPolicy.rerunPolicy(syncAllCronExpression, ytlIntegration)
+  val rerunSync = YtlRerunPolicy.rerunPolicy(syncAllCronExpression, ytlFetchActor)
   if (ytlSyncAllEnabled) {
     quartzScheduler.scheduleJob(
       lambdaJob(rerunSync),