Merge pull request #2291 from hongwei1/develop

feature/added the hikariCP properties file and new  DataBaseCleanerScheduler

.github/workflows/build_package.yml

@@ -52,6 +52,7 @@ jobs:
 
           echo kafka.akka.timeout = 9 >> obp-api/src/main/resources/props/test.default.props
           echo remotedata.timeout = 10 >> obp-api/src/main/resources/props/test.default.props
+          echo api_instance_id = 1 >> obp-api/src/main/resources/props/test.default.props
 
           echo allow_oauth2_login=true >> obp-api/src/main/resources/props/test.default.props
           echo oauth2.jwk_set.url=https://www.googleapis.com/oauth2/v3/certs >> obp-api/src/main/resources/props/test.default.props

obp-api/src/main/resources/props/sample.props.template

@@ -120,8 +120,12 @@ long_endpoint_timeout = 55000
 #kafka.bootstrap_hosts=localhost:9092
 # WARNING: if this number does not match the partitions in Kafka config, you will SUFFER !
 #kafka.partitions=3
+
 #This is the api instance, we create kafka topic based on this number, each instance should have each own id. use it in load balancing + Kafka setup
+#This is also used for scheduler.
+#OBP will concatenate the UUID string with the value. If we set the value to end with "final", obp will retain that value.
 #api_instance_id=1
+api_instance_id=1_final
 
 ## DEPRECATED
 ## Enable user authentication via kafka

obp-api/src/main/resources/props/test.default.props.template

@@ -15,6 +15,9 @@
 ### if logger.logfile is defined
 #logger.logfile="obp-api.log"
 
+#This is the api instance
+api_instance_id=1_final
+
 #which data connector to use
 
 #connector=rest

obp-api/src/main/scala/bootstrap/liftweb/Boot.scala

@@ -108,7 +108,7 @@ import code.productfee.ProductFee
 import code.products.MappedProduct
 import code.ratelimiting.RateLimiting
 import code.remotedata.RemotedataActors
-import code.scheduler.{DatabaseDriverScheduler, JobScheduler, MetricsArchiveScheduler}
+import code.scheduler.{DatabaseDriverScheduler, JobScheduler, MetricsArchiveScheduler, DataBaseCleanerScheduler}
 import code.scope.{MappedScope, MappedUserScope}
 import code.snippet.{OAuthAuthorisation, OAuthWorkedThanks}
 import code.socialmedia.MappedSocialMedia
@@ -323,8 +323,8 @@ class Boot extends MdcLoggable {
     //see the notes for this method:
     createDefaultBankAndDefaultAccountsIfNotExisting()
 
-    //launch the scheduler to clean the database from the expired tokens and nonces
-    Schedule.schedule(() => OAuthAuthorisation.dataBaseCleaner, 2 minutes)
+    //launch the scheduler to clean the database from the expired tokens and nonces, 1 hour
+    DataBaseCleanerScheduler.start(intervalInSeconds = 60*60)
 
 //    if (Props.devMode || Props.testMode) {
 //      StoredProceduresMockedData.createOrDropMockedPostgresStoredProcedures()
@@ -361,7 +361,7 @@ class Boot extends MdcLoggable {
 //      }
 //    }
 
-//    LiftRules.unloadHooks.append(APIUtil.vendor.closeAllConnections_! _)
+    LiftRules.unloadHooks.append(APIUtil.vendor.closeAllConnections_! _)
 
 //    LiftRules.statelessDispatch.prepend {
 //      case _ if tryo(DB.use(DefaultConnectionIdentifier){ conn => conn}.isClosed).isEmpty =>

obp-api/src/main/scala/code/api/constant/constant.scala

@@ -23,6 +23,16 @@ object Constant extends MdcLoggable {
   final val h2DatabaseDefaultUrlValue = "jdbc:h2:mem:OBPTest_H2_v2.1.214;NON_KEYWORDS=VALUE;DB_CLOSE_DELAY=10"
 
   final val HostName = APIUtil.getPropsValue("hostname").openOrThrowException(ErrorMessages.HostnameNotSpecified)
+
+  final val ApiInstanceId = {
+    val apiInstanceIdFromProps = APIUtil.getPropsValue("api_instance_id").openOrThrowException(ErrorMessages.ApiInstanceIdNotSpecified)
+    if(apiInstanceIdFromProps.endsWith("final")){
+      apiInstanceIdFromProps
+    }else{
+      s"${apiInstanceIdFromProps}_${APIUtil.generateUUID()}"
+    }
+  }
+
   def localIdentityProvider = APIUtil.getPropsValue("local_identity_provider", HostName)
 
   // This is the part before the version. Do not change this default!

obp-api/src/main/scala/code/api/util/APIUtil.scala

@@ -4796,7 +4796,7 @@ object APIUtil extends MdcLoggable with CustomJsonFormats{
     Props.mode match {
       case Props.RunModes.Production | Props.RunModes.Staging | Props.RunModes.Development =>
         new CustomDBVendor(driver,
-          APIUtil.getPropsValue("db.url") openOr "jdbc:h2:lift_proto.db;AUTO_SERVER=TRUE",
+          APIUtil.getPropsValue("db.url") openOr h2DatabaseDefaultUrlValue,
           APIUtil.getPropsValue("db.user"), APIUtil.getPropsValue("db.password"))
       case Props.RunModes.Test =>
         new CustomDBVendor(

obp-api/src/main/scala/code/api/util/ErrorMessages.scala

@@ -51,6 +51,7 @@ object ErrorMessages {
   val MissingPropsValueAtThisInstance = "OBP-00010: Missing props value at this API instance - "
   val NoValidElasticsearchIndicesConfigured = "OBP-00011: No elasticsearch indices are allowed on this instance. Please set es.warehouse.allowed.indices = index1,index2 (or = ALL for all). "
   val CustomerFirehoseNotAllowedOnThisInstance = "OBP-00012: Customer firehose is not allowed on this instance. Please set allow_customer_firehose = true in props files. "
+  val ApiInstanceIdNotSpecified = "OBP-00013: 'api_instance_id' not specified. Please edit your props file."
 
   // Exceptions (OBP-01XXX) ------------------------------------------------>
   val requestTimeout = "OBP-01000: Request Timeout. The OBP API decided to return a timeout. This is probably because a backend service did not respond in time. "

obp-api/src/main/scala/code/kafka/KafkaConfig.scala

@@ -1,5 +1,6 @@
 package code.kafka
 
+import code.api.Constant
 import code.api.util.{APIUtil, ErrorMessages}
 import scala.concurrent.duration.{FiniteDuration, MILLISECONDS}
 
@@ -10,7 +11,7 @@ trait KafkaConfig {
 
   val bootstrapServers = APIUtil.getPropsValue("kafka.bootstrap_hosts")openOr("localhost:9092")
   val groupId = APIUtil.getPropsValue("kafka.group.id").openOr("obp-api")
-  val apiInstanceId = APIUtil.getPropsAsIntValue("api_instance_id").openOr("1") 
+  val apiInstanceId = Constant.ApiInstanceId 
   val partitions = APIUtil.getPropsAsIntValue("kafka.partitions", 10)
 
   val clientId = s"obp.api.$apiInstanceId"

obp-api/src/main/scala/code/scheduler/DataBaseCleanerScheduler.scala

@@ -0,0 +1,80 @@
+package code.scheduler
+
+import code.actorsystem.ObpLookupSystem
+import code.api.Constant
+import code.api.util.APIUtil.generateUUID
+import code.api.util.APIUtil
+import code.nonce.Nonces
+import code.util.Helper.MdcLoggable
+import net.liftweb.common.Full
+import net.liftweb.mapper.{By, By_<=}
+
+import java.util.concurrent.TimeUnit
+import java.util.Date
+import scala.concurrent.duration._
+import code.token.Tokens
+
+
+object DataBaseCleanerScheduler extends MdcLoggable {
+
+  private lazy val actorSystem = ObpLookupSystem.obpLookupSystem
+  implicit lazy val executor = actorSystem.dispatcher
+  private lazy val scheduler = actorSystem.scheduler
+  private val oneDayInMillis: Long = 86400000
+  //in scala DataBaseCleanerScheduler.getClass.getSimpleName ==> DataBaseCleanerScheduler$
+  private val jobName = DataBaseCleanerScheduler.getClass.getSimpleName.replace("$", "")
+  private val apiInstanceId = Constant.ApiInstanceId
+
+  def start(intervalInSeconds: Long): Unit = {
+    logger.info(s"Hello from $jobName.start")
+
+    logger.info(s"--------- Clean up Jobs ---------")
+    logger.info(s"Delete all Jobs created by api_instance_id=$apiInstanceId")
+    JobScheduler.findAll(By(JobScheduler.Name, apiInstanceId)).map { i =>
+      logger.info(s"Job name: ${i.name}, Date: ${i.createdAt}")
+      i
+    }.map(_.delete_!)
+    logger.info(s"Delete all Jobs older than 5 days")
+    val fiveDaysAgo: Date = new Date(new Date().getTime - (oneDayInMillis * 5))
+    JobScheduler.findAll(By_<=(JobScheduler.createdAt, fiveDaysAgo)).map { i =>
+      logger.info(s"Job name: ${i.name}, Date: ${i.createdAt}, api_instance_id: ${apiInstanceId}")
+      i
+    }.map(_.delete_!)
+
+    scheduler.schedule(
+      initialDelay = Duration(intervalInSeconds, TimeUnit.SECONDS),
+      interval = Duration(intervalInSeconds, TimeUnit.SECONDS),
+      runnable = new Runnable {
+        def run(): Unit = {
+          JobScheduler.find(By(JobScheduler.Name, jobName)) match {
+            case Full(job) => // There is an ongoing/hanging job
+              logger.info(s"Cannot start $jobName.start.run due to ongoing job. Job ID: ${job.JobId}")
+            case _ => // Start a new job
+              val uniqueId = generateUUID()
+              val job = JobScheduler.create
+                .JobId(uniqueId)
+                .Name(jobName)
+                .ApiInstanceId(apiInstanceId)
+                .saveMe()
+              logger.info(s"Starting $jobName.Job ID: $uniqueId")
+              deleteExpiredTokensAndNonces()
+              JobScheduler.delete_!(job) // Allow future jobs
+              logger.info(s"End of $jobName.Job ID: $uniqueId")
+          }
+        } 
+      }
+    )
+    logger.info(s"Bye from $jobName.start")
+  }
+
+  def deleteExpiredTokensAndNonces() = {
+    //looks for expired tokens and nonces and deletes them
+    val currentDate = new Date()
+    //delete expired tokens and nonces
+    Tokens.tokens.vend.deleteExpiredTokens(currentDate)
+    Nonces.nonces.vend.deleteExpiredNonces(currentDate)
+  }
+
+
+}
+

obp-api/src/main/scala/code/scheduler/MetricsArchiveScheduler.scala

@@ -2,8 +2,8 @@ package code.scheduler
 
 import java.util.concurrent.TimeUnit
 import java.util.{Calendar, Date}
-
 import code.actorsystem.ObpLookupSystem
+import code.api.Constant
 import code.api.util.APIUtil.generateUUID
 import code.api.util.{APIUtil, OBPLimit, OBPToDate}
 import code.metrics.{APIMetric, APIMetrics, MappedMetric, MetricArchive}
@@ -21,7 +21,7 @@ object MetricsArchiveScheduler extends MdcLoggable {
   private lazy val scheduler = actorSystem.scheduler
   private val oneDayInMillis: Long = 86400000
   private val jobName = "MetricsArchiveScheduler"
-  private val apiInstanceId = APIUtil.getPropsValue("api_instance_id", "NOT_SET")
+  private val apiInstanceId = Constant.ApiInstanceId
 
   def start(intervalInSeconds: Long): Unit = {
     logger.info("Hello from MetricsArchiveScheduler.start")

obp-api/src/main/scala/code/snippet/OAuthAuthorisation.scala

@@ -189,23 +189,4 @@ object OAuthAuthorisation {
     }
 
   }
-
-  //looks for expired tokens and nonces and deletes them
-  def dataBaseCleaner: Unit = {
-    import net.liftweb.util.Schedule
-    Schedule.schedule(dataBaseCleaner _, 1 hour)
-
-    val currentDate = new Date()
-
-    /*
-      As in "wrong timestamp" function, 3 minutes is the timestamp limit where we accept
-      requests. So this function will delete nonces which have a timestamp older than
-      currentDate - 3 minutes
-    */
-    val timeLimit = new Date(currentDate.getTime + 180000)
-
-    //delete expired tokens and nonces
-    Tokens.tokens.vend.deleteExpiredTokens(currentDate)
-    Nonces.nonces.vend.deleteExpiredNonces(currentDate)
-  }
 }

obp-api/src/test/scala/RunMTLSWebApp.scala

@@ -117,7 +117,7 @@ object RunMTLSWebApp extends App with PropsProgrammatically {
   context.setWar(s"${basePath}src/main/webapp")
 
   // rename JSESSIONID, avoid conflict with other project when start two project at local
-  val propsApiInstanceId = APIUtil.getPropsValue("api_instance_id").openOrThrowException("connector props filed `api_instance_id` not set")
+  val propsApiInstanceId = Constant.ApiInstanceId
   context.getSessionHandler.getSessionCookieConfig.setName("JSESSIONID_OBP_API_" + propsApiInstanceId)
 
   server.setHandler(context)

obp-api/src/test/scala/RunTLSWebApp.scala

@@ -118,7 +118,7 @@ object RunTLSWebApp extends App with PropsProgrammatically {
   context.setWar(s"${basePath}src/main/webapp")
 
   // rename JSESSIONID, avoid conflict with other project when start two project at local
-  val propsApiInstanceId = APIUtil.getPropsValue("api_instance_id").openOrThrowException("connector props filed `api_instance_id` not set")
+  val propsApiInstanceId = Constant.ApiInstanceId
   context.getSessionHandler.getSessionCookieConfig.setName("JSESSIONID_OBP_API_" + propsApiInstanceId)
 
   server.setHandler(context)

obp-api/src/test/scala/RunWebApp.scala

@@ -70,7 +70,7 @@ object RunWebApp extends App {
   val basePath = this.getClass.getResource("/").toString .replaceFirst("target[/\\\\].*$", "")
   context.setWar(s"${basePath}src/main/webapp")
   // rename JSESSIONID, avoid conflict with other project when start two project at local
-  val propsApiInstanceId = APIUtil.getPropsValue("api_instance_id").openOrThrowException("connector props filed `api_instance_id` not set")
+  val propsApiInstanceId = code.api.Constant.ApiInstanceId
   context.getSessionHandler.getSessionCookieConfig.setName("JSESSIONID_OBP_API_" + propsApiInstanceId)
   server.setHandler(context)
 

obp-api/src/test/scala/code/bankconnectors/vMay2019/KafkaMappedConnector_vMay2019Test.scala

@@ -66,7 +66,7 @@ class KafkaMappedConnector_vMay2019Test extends KafkaSetup with ServerSetupWithT
       scenario("1st test `getObpConnectorLoopback` method, there no need Adapter message for this method!", kafkaTest) {
         //This method is only used for `kafka` connector, should first set `connector=kafka_vSept2018` in test.default.props. 
         //and also need to set up `api_instance_id` and `remotedata.timeout` field for it.
-        val propsApiInstanceId = APIUtil.getPropsValue("api_instance_id").openOrThrowException("connector props filed `api_instance_id` not set")
+        val propsApiInstanceId = code.api.Constant.ApiInstanceId
         val propsRemotedataTimeout = APIUtil.getPropsValue("remotedata.timeout").openOrThrowException("connector props filed `remotedata.timeout` not set")
 
         PropsConnectorVersion contains ("kafka") should be(true)

obp-api/src/test/scala/code/kafka/KafkaTest.scala

@@ -46,7 +46,7 @@ class KafkaTest extends KafkaSetup with ServerSetupWithTestData {
       scenario("1st test `getObpConnectorLoopback` method, there no need Adapter message for this method!", kafkaTest) {
         //This method is only used for `kafka` connector, should first set `connector=kafka_vSept2018` in test.default.props. 
         //and also need to set up `api_instance_id` and `remotedata.timeout` field for it.
-        val propsApiInstanceId = APIUtil.getPropsValue("api_instance_id").openOrThrowException("connector props filed `api_instance_id` not set")
+        val propsApiInstanceId = code.api.Constant.ApiInstanceId
         val propsRemotedataTimeout = APIUtil.getPropsValue("remotedata.timeout").openOrThrowException("connector props filed `remotedata.timeout` not set")
 
         PropsConnectorVersion contains ("kafka") should be (true)

obp-api/src/test/scala/code/setup/ServerSetup.scala

@@ -53,6 +53,7 @@ trait ServerSetup extends FeatureSpec with SendServerRequests
   setPropsValues("jwt.public_key_rsa" -> "src/test/resources/cert/public_dauth.pem")
   setPropsValues("transactionRequests_supported_types" -> "SEPA,SANDBOX_TAN,FREE_FORM,COUNTERPARTY,ACCOUNT,ACCOUNT_OTP,SIMPLE,CARD")
   setPropsValues("CARD_OTP_INSTRUCTION_TRANSPORT" -> "DUMMY")
+  setPropsValues("api_instance_id" -> "1_final")
 
   val server = TestServer
   def baseRequest = host(server.host, server.port)