diff --git a/apps/meta-app/src/main/resources/reference.conf b/apps/meta-app/src/main/resources/reference.conf index 372b0c89f..c05f7ad60 100644 --- a/apps/meta-app/src/main/resources/reference.conf +++ b/apps/meta-app/src/main/resources/reference.conf @@ -1,21 +1,25 @@ shrine { metaData { ping = "pong" } + + messagequeue { + receiveWaitTime = "15 seconds" + } } //todo typesafe config precedence seems to do the right thing, but I haven't found the rules that say this reference.conf should override others // todo go with the actor system, whcih we will pull out later akka { loglevel = INFO // log-config-on-start = on loggers = ["akka.event.slf4j.Slf4jLogger"] // logging-filter = "akka.event.slf4j.Slf4jLoggingFilter" // Toggles whether the threads created by this ActorSystem should be daemons or not daemonic = on } spray.servlet { boot-class = "net.shrine.metadata.Boot" request-timeout = 30s } diff --git a/apps/meta-app/src/main/scala/net/shrine/metadata/QepReceiver.scala b/apps/meta-app/src/main/scala/net/shrine/metadata/QepReceiver.scala index 91d8ee6ef..e5f287a77 100644 --- a/apps/meta-app/src/main/scala/net/shrine/metadata/QepReceiver.scala +++ b/apps/meta-app/src/main/scala/net/shrine/metadata/QepReceiver.scala @@ -1,222 +1,222 @@ package net.shrine.metadata import java.util.concurrent.atomic.AtomicBoolean import javax.servlet.{ServletContextEvent, ServletContextListener} import com.typesafe.config.Config import net.shrine.config.ConfigExtensions import net.shrine.hornetqclient.CouldNotCreateQueueButOKToRetryException import net.shrine.log.Log import net.shrine.messagequeueservice.protocol.Envelope import net.shrine.messagequeueservice.{Message, MessageQueueService, Queue} import net.shrine.problem.{AbstractProblem, ProblemSources} import net.shrine.protocol.{AggregatedRunQueryResponse, QueryResult, ResultOutputType, ResultOutputTypes} import net.shrine.qep.querydb.{QepQueryDb, QueryResultRow} import net.shrine.source.ConfigSource import net.shrine.status.protocol.IncrementalQueryResult -import scala.concurrent.duration.Duration +import scala.concurrent.duration.{Duration, FiniteDuration} import scala.util.control.NonFatal import scala.util.{Failure, Success, Try} /** * Receives messages and writes the result to the QEP's cache * * @author david * @since 8/18/17 */ object QepReceiver { val config: Config = ConfigSource.config val nodeName = config.getString("shrine.humanReadableNodeName") + val pollDuration = config.get("shrine.messagequeue.receiveWaitTime",Duration(_)) + //create a daemon thread that long-polls for messages forever - val runner = QepReceiverRunner(nodeName) + val runner = QepReceiverRunner(nodeName,pollDuration) val pollingThread = new Thread(runner,s"${getClass.getSimpleName} poller") pollingThread.setDaemon(true) - pollingThread.setUncaughtExceptionHandler(QepReceiverUncaughtExceptionHandler) + pollingThread.setUncaughtExceptionHandler(QepReceiverUncaughtExceptionHandler) def start(): Unit = { pollingThread.start() Log.debug(s"Started the QepReceiver thread for $nodeName") } def stop(): Unit = { runner.stop() } - case class QepReceiverRunner(nodeName:String) extends Runnable { + case class QepReceiverRunner(nodeName:String,pollDuration:Duration) extends Runnable { val keepGoing = new AtomicBoolean(true) def stop(): Unit = { keepGoing.set(false) Log.debug(s"${this.getClass.getSimpleName} keepGoing set to ${keepGoing.get()}. Will stop asking for messages after the current request.") } - val pollDuration = Duration("15 seconds") //todo from config SHRINE-2198 - val breakdownTypes: Set[ResultOutputType] = ConfigSource.config.getOptionConfigured("shrine.breakdownResultOutputTypes", ResultOutputTypes.fromConfig).getOrElse(Set.empty) override def run(): Unit = { val queue = createQueue(nodeName) while (keepGoing.get()) { //forever try { //todo only ask to receive a message if there are incomplete queries SHRINE-2196 Log.debug("About to call receive.") receiveAMessage(queue) Log.debug("Called receive.") } catch { case NonFatal(x) => ExceptionWhileReceivingMessage(queue,x) //pass-through to blow up the thread, receive no more results, do something dramatic in UncaughtExceptionHandler. case x => Log.error("Fatal exception while receiving a message", x) throw x } } Log.debug(s"QepReceiverRunner will stop. keepGoing is ${keepGoing.get()}") } def receiveAMessage(queue:Queue): Unit = { val maybeMessage: Try[Option[Message]] = MessageQueueService.service.receive(queue, pollDuration) //todo make pollDuration configurable (and testable) SHRINE-2198 maybeMessage.transform({m => m.map(interpretAMessage(_,queue)).getOrElse(Success()) },{x => x match { case NonFatal(nfx) => ExceptionWhileReceivingMessage(queue,x) case _ => //pass through } Failure(x) }) } def interpretAMessage(message: Message,queue: Queue): Try[Unit] = { val unit = () Log.debug(s"Received a message from $queue of $message") val envelopeJson = message.contents Envelope.fromJson(envelopeJson). flatMap{ case e:Envelope => e.checkVersionExactMatch case notE => Failure(new IllegalArgumentException(s"Not an expected message Envelope but a ${notE.getClass}")) }. flatMap { case Envelope(contentsType, contents, shrineVersion) if contentsType == AggregatedRunQueryResponse.getClass.getSimpleName => AggregatedRunQueryResponse.fromXmlString(breakdownTypes)(contents).flatMap{ rqr => QepQueryDb.db.insertQueryResult(rqr.queryId, rqr.results.head) Log.debug(s"Inserted result from ${rqr.results.head.description} for query ${rqr.queryId}") Success(unit) } case Envelope(contentsType, contents, shrineVersion) if contentsType == IncrementalQueryResult.incrementalQueryResultsEnvelopeContentsType => val changeDate = System.currentTimeMillis() IncrementalQueryResult.seqFromJson(contents).flatMap { iqrs: Seq[IncrementalQueryResult] => val rows = iqrs.map(iqr => QueryResultRow( resultId = -1L, networkQueryId = iqr.networkQueryId, instanceId = -1L, adapterNode = iqr.adapterNodeName, resultType = None, size = 0L, startDate = None, endDate = None, status = QueryResult.StatusType.valueOf(iqr.statusTypeName).get, statusMessage = Some(iqr.statusMessage), changeDate = changeDate )) QepQueryDb.db.insertQueryResultRows(rows) Log.debug(s"Inserted incremental results $iqrs") Success(unit) } case e:Envelope => Failure(UnexpectedMessageContentsTypeException(e,queue)) case _ => Failure(new IllegalArgumentException(s"Received something other than an envelope from this queue: $envelopeJson")) }.transform({ s => message.complete() Success(unit) },{ x => x match { case NonFatal(nfx) => QepReceiverCouldNotDecodeMessage(envelopeJson,queue,x) case throwable => throw throwable//blow something up } message.complete() //complete anyway. Can't be interpreted, so we don't want to see it again Failure(x) }) } def createQueue(nodeName:String):Queue = { //Either come back with the right exception to try again, or a Queue def tryToCreateQueue():Try[Queue] = MessageQueueService.service.createQueueIfAbsent(nodeName) def keepGoing(attempt:Try[Queue]):Try[Boolean] = attempt.transform({queue => Success(false)}, { case okIsh: CouldNotCreateQueueButOKToRetryException => Success(true) case x => Failure(x) }) //todo for fun figure out how to do this without the var. maybe a Stream ? SHRINE-2211 var lastAttempt:Try[Queue] = tryToCreateQueue() while(keepGoing(lastAttempt).get) { Log.debug(s"Last attempt to create a queue resulted in $lastAttempt. Sleeping $pollDuration before next attempt") Thread.sleep(pollDuration.toMillis) lastAttempt = tryToCreateQueue() } Log.info(s"Finishing createQueue with $lastAttempt") lastAttempt.get } } } object QepReceiverUncaughtExceptionHandler extends Thread.UncaughtExceptionHandler { override def uncaughtException(thread: Thread, throwable: Throwable): Unit = QepReceiverThreadEndedByThrowable(thread,throwable) } class QueueReceiverContextListener extends ServletContextListener { override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = { QepReceiver.start() } override def contextDestroyed(servletContextEvent: ServletContextEvent): Unit = { QepReceiver.stop() } } case class UnexpectedMessageContentsTypeException(envelope: Envelope, queue: Queue) extends Exception(s"Could not interpret message with contents type of ${envelope.contentsType} from queue ${queue.name} from shrine version ${envelope.shrineVersion}") case class ExceptionWhileReceivingMessage(queue:Queue, x:Throwable) extends AbstractProblem(ProblemSources.Qep) { override val throwable = Some(x) override def summary: String = s"The QEP encountered an exception while trying to receive a message from $queue" override def description: String = s"The QEP encountered an exception while trying to receive a message from $queue on ${Thread.currentThread().getName}: ${x.getMessage}" } case class QepReceiverCouldNotDecodeMessage(messageString:String,queue:Queue, x:Throwable) extends AbstractProblem(ProblemSources.Qep) { override val throwable = Some(x) override def summary: String = s"The QEP could not decode a message from $queue" override def description: String = s"""The QEP encountered an exception while trying to decode a message from $queue on ${Thread.currentThread().getName}: |${x.getMessage} |$messageString""".stripMargin } case class QepReceiverThreadEndedByThrowable(thread:Thread, x:Throwable) extends AbstractProblem(ProblemSources.Qep) { override val throwable = Some(x) override def summary: String = s"The Qep Receiver's thread stopped because of an uncaught exception." override def description: String = s"""The Qep Receiver's thread ${thread.getName} stopped because of an uncaught exception""" } \ No newline at end of file diff --git a/messagequeue/messagequeueservice/src/main/resources/reference.conf b/messagequeue/messagequeueservice/src/main/resources/reference.conf index eb4525a52..572408a0a 100644 --- a/messagequeue/messagequeueservice/src/main/resources/reference.conf +++ b/messagequeue/messagequeueservice/src/main/resources/reference.conf @@ -1,5 +1,5 @@ shrine { messagequeue { - implementation = "net.shrine.hornetqclient.HornetQMomWebClient" //Fully-qualified class name of the MessaageQueueService to use //todo when there are two real implemetnations, maybe leave this out + implementation = "net.shrine.hornetqclient.HornetQMomWebClient" //Fully-qualified class name of the MessaageQueueService to use //todo when there are two real implemetnations, maybe leave this default out } } \ No newline at end of file diff --git a/shrine-setup/src/main/resources/shrine.conf b/shrine-setup/src/main/resources/shrine.conf index 79ab08192..2dc95a555 100644 --- a/shrine-setup/src/main/resources/shrine.conf +++ b/shrine-setup/src/main/resources/shrine.conf @@ -1,358 +1,361 @@ shrine { metaData { ping = "pong" } pmEndpoint { // url = "http://shrine-dev1.catalyst/i2b2/services/PMService/getServices" //use your i2b2 pm url } ontEndpoint { // url = "http://shrine-dev1.catalyst/i2b2/rest/OntologyService/" //use your i2b2 ontology url } hiveCredentials { //use your i2b2 hive credentials // domain = "i2b2demo" // username = "demo" // password = "examplePassword" // crcProjectId = "Demo" // ontProjectId = "SHRINE" } messagequeue { hornetq { serverUrl = "https://localhost:6443/shrine-metadata/mom" } // If you intend for your node to serve as this SHRINE network's messaging hub, // then set shrine.messagequeue.hornetQWebApi.enabled to true in your shrine.conf. // You do not want to do this unless you are the hub admin hornetQWebApi { // enabled = false } + messagequeue { +// receiveWaitTime = "15 seconds" + } } breakdownResultOutputTypes { //use breakdown values appropriate for your shrine network // PATIENT_AGE_COUNT_XML { // description = "Age patient breakdown" // } // PATIENT_RACE_COUNT_XML { // description = "Race patient breakdown" // } // PATIENT_VITALSTATUS_COUNT_XML { // description = "Vital Status patient breakdown" // } // PATIENT_GENDER_COUNT_XML { // description = "Gender patient breakdown" // } } queryEntryPoint { // create = true //false for no qep // audit { // collectQepAudit = true //false to not use the 1.20 audit db tables // database { // dataSourceFrom = "JNDI" //Can be JNDI or testDataSource . Use testDataSource for tests, JNDI everywhere else // jndiDataSourceName = "java:comp/env/jdbc/qepAuditDB" //or leave out for tests // slickProfileClassName = "slick.driver.MySQLDriver$" // Can be // slick.driver.H2Driver$ // slick.driver.MySQLDriver$ // slick.driver.PostgresDriver$ // slick.driver.SQLServerDriver$ // slick.driver.JdbcDriver$ // freeslick.OracleProfile$ // freeslick.MSSQLServerProfile$ // // (Yes, with the $ on the end) // For testing without JNDI // testDataSource { //typical test settings for unit tests //driverClassName = "org.h2.Driver" //url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1" //H2 embedded in-memory for unit tests //url = "jdbc:h2:~/stewardTest.h2" //H2 embedded on disk at ~/test // } // timeout = 30 //time to wait before db gives up, in seconds. // createTablesOnStart = false //for testing with H2 in memory, when not running unit tests. Set to false normally // } // } // trustModelIsHub = true // true by default, false for P2P networks. // authenticationType = "pm" //can be none, pm, or ecommons // authorizationType = "shrine-steward" //can be none, shrine-steward, or hms-steward //hms-steward config // sheriffEndpoint { // url = "http://localhost:8080/shrine-hms-authorization/queryAuthorization" // timeout { // seconds = 1 // } // } // sheriffCredentials { // username = "sheriffUsername" // password = "sheriffPassword" // } //shrine-steward config // shrineSteward { // qepUserName = "qep" // qepPassword = "trustme" // stewardBaseUrl = "https://localhost:6443" // } // includeAggregateResults = false // // maxQueryWaitTime { // minutes = 5 //must be longer than the hub's maxQueryWaitTime // } // broadcasterServiceEndpoint { // url = "http://example.com/shrine/rest/broadcaster/broadcast" //url for the hub // acceptAllCerts = true // timeout { // seconds = 1 // } // } } hub { // create = false //change to true to start a hub maxQueryWaitTime { // minutes = 4.5 //Needs to be longer than the adapter's maxQueryWaitTime, but shorter than the qep's } // downstreamNodes { //Add your downstream nodes here // shrine-dev2 = "https://shrine-dev2.catalyst:6443/shrine/rest/adapter/requests" // } shouldQuerySelf = false //true if there is an adapter at the hub , or just add a loopback address to downstreamNodes } adapter { // create = true by default. False to not create an adapter. // audit { // collectAdapterAudit = true by default. False to not fill in the audit database // database { // dataSourceFrom = "JNDI" //Can be JNDI or testDataSource . Use testDataSource for tests, JNDI everywhere else // jndiDataSourceName = "java:comp/env/jdbc/adapterAuditDB" //or leave out for tests // slickProfileClassName = "slick.driver.MySQLDriver$" // Can be // slick.driver.H2Driver$ // slick.driver.MySQLDriver$ // slick.driver.PostgresDriver$ // slick.driver.SQLServerDriver$ // slick.driver.JdbcDriver$ // freeslick.OracleProfile$ // freeslick.MSSQLServerProfile$ // // (Yes, with the $ on the end) // For testing without JNDI // testDataSource { //typical test settings for unit tests //driverClassName = "org.h2.Driver" //url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1" //H2 embedded in-memory for unit tests //url = "jdbc:h2:~/stewardTest.h2" //H2 embedded on disk at ~/test // } // createTablesOnStart = false //for testing with H2 in memory, when not running unit tests. Set to false normally // } // obfuscation { // binSize = 5 by default //Round to the nearest binSize. Use 1 for no effect (to match SHRINE 1.21 and earlier). // sigma = 6.5 by default //Noise to inject. Use 0 for no effect. (Use 1.33 to match SHRINE 1.21 and earlier). // clamp = 10 by default //Maximum ammount of noise to inject. (Use 3 to match SHRINE 1.21 and earlier). // } // adapterLockoutAttemptsThreshold = 0 by default // Number of allowed queries with the same actual result that can exist before a researcher is locked out of the adapter. In 1.24 the lockout code and this config value will be removed // botDefense { // countsAndMilliseconds = [ //to turn off, use an empty json list // {count = 10, milliseconds = 60000}, //allow up to 10 queries in one minute by default // {count = 200, milliseconds = 36000000} //allow up to 4 queries in 10 hours by default // ] // } crcEndpoint { //must be filled in url = "http://shrine-dev1.catalyst/i2b2/services/QueryToolService/" } setSizeObfuscation = true //must be set. false turns off obfuscation adapterMappingsFileName = "AdapterMappings.xml" maxSignatureAge { minutes = 5 //must be longer than the hub's maxQueryWaitTime } immediatelyRunIncomingQueries = true //false to queue them //delayResponse = "0 seconds" //time to delay before responding to a query. Should be 0 except for testing in shrine-qa } networkStatusQuery = "\\\\SHRINE\\SHRINE\\Demographics\\Gender\\Male\\" humanReadableNodeName = "shrine-dev1" shrineDatabaseType = "mysql" keystore { file = "/opt/shrine/shrine.keystore" password = "changeit" privateKeyAlias = "shrine-dev1.catalyst" keyStoreType = "JKS" caCertAliases = [ "shrine-dev-ca" ] } problem { // problemHandler = "net.shrine.problem.LogAndDatabaseProblemHandler$" Can be other specialized problemHandler implementations // database { // dataSourceFrom = "JNDI" //Can be JNDI or testDataSource . Use testDataSource for tests, JNDI everywhere else // jndiDataSourceName = "java:comp/env/jdbc/problemDB" // slickProfileClassName = "slick.driver.MySQLDriver$" // Can be // slick.driver.H2Driver$ // slick.driver.MySQLDriver$ // slick.driver.PostgresDriver$ // slick.driver.SQLServerDriver$ // slick.driver.JdbcDriver$ // freeslick.OracleProfile$ // freeslick.MSSQLServerProfile$ // // (Yes, with the $ on the end) // For testing without JNDI // testDataSource { //typical test settings for unit tests //driverClassName = "org.h2.Driver" //url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1" //H2 embedded in-memory for unit tests //url = "jdbc:h2:~/stewardTest.h2" //H2 embedded on disk at ~/test // } // createTablesOnStart = false //for testing with H2 in memory, when not running unit tests. Set to false normally // } } dashboard { // gruntWatch = false //false for production, true for mvn tomcat7:run . Allows the client javascript and html files to be loaded via gruntWatch . // happyBaseUrl = "https://localhost:6443/shrine/rest/happy" If the shine servlet is running on a different machime from the dashboard, change this URL to match // statusBaseUrl = "https://localhost:6443/shrine/rest/internalstatus" If the shine servlet is running on a different machime from the dashboard, change this URL to match // } // status { //permittedHostOfOrigin = "localhost" //If absent then get the host name via java.net.InetAddress.getLocalHost.getHostName . Override to control } //Get the older squerl-basd databases through JNDI (inside of tomcant, using tomcat's db connection pool) or directly via a db config here (for testing squerylDataSource { // database { // dataSourceFrom = "JNDI" //Can be JNDI or testDataSource . Use testDataSource for tests, JNDI everywhere else // jndiDataSourceName = "java:comp/env/jdbc/shrineDB" //or leave out for tests // } } authenticate { // realm = "SHRINE Researcher API" //todo figure out what this means. SHRINE-1978 usersource { // domain = "i2b2demo" //you must provide your own domain } } steward { // createTopicsMode = Pending //Can be Pending, Approved, or TopcisIgnoredJustLog. Pending by default //Pending - new topics start in the Pending state; researchers must wait for the Steward to approve them //Approved - new topics start in the Approved state; researchers can use them immediately //TopicsIgnoredJustLog - all queries are logged and approved; researchers don't need to create topics emailDataSteward { // sendAuditEmails = true //false to turn off the whole works of emailing the data steward // interval = "1 day" //Audit researchers daily // timeAfterMidnight = "6 hours" //Audit researchers at 6 am. If the interval is less than 1 day then this delay is ignored. // maxQueryCountBetweenAudits = 30 //If a researcher runs more than this many queries since the last audit audit her // minTimeBetweenAudits = "30 days" //If a researcher runs at least one query, audit those queries if this much time has passed //You must provide the email address of the shrine node system admin, to handle bounces and invalid addresses //from = "shrine-admin@example.com" //You must provide the email address of the data steward //to = "shrine-steward@example.com" // subject = "Audit SHRINE researchers" //The baseUrl for the data steward to be substituted in to email text. Must be supplied if it is used in the email text. //stewardBaseUrl = "https://example.com:8443/steward/" //Text to use for the email audit. // AUDIT_LINES will be replaced by a researcherLine for each researcher to audit. // STEWARD_BASE_URL will be replaced by the value in stewardBaseUrl if available. // emailBody = """Please audit the following users at STEWARD_BASE_URL at your earliest convinience: // //AUDIT_LINES""" //note that this can be a multiline message //Text to use per researcher to audit. //FULLNAME, USERNAME, COUNT and LAST_AUDIT_DATE will be replaced with appropriate text. // researcherLine = "FULLNAME (USERNAME) has run COUNT queries since LAST_AUDIT_DATE." } // database { // dataSourceFrom = "JNDI" //Can be JNDI or testDataSource . Use testDataSource for tests, JNDI everywhere else // jndiDataSourceName = "java:comp/env/jdbc/stewardDB" //or leave out for tests // slickProfileClassName = "slick.driver.MySQLDriver$" // Can be // slick.driver.H2Driver$ // slick.driver.MySQLDriver$ // slick.driver.PostgresDriver$ // slick.driver.SQLServerDriver$ // slick.driver.JdbcDriver$ // freeslick.OracleProfile$ // freeslick.MSSQLServerProfile$ // // (Yes, with the $ on the end) // For testing without JNDI // testDataSource { //typical test settings for unit tests //driverClassName = "org.h2.Driver" //url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1" //H2 embedded in-memory for unit tests //url = "jdbc:h2:~/stewardTest.h2" //H2 embedded on disk at ~/test // } // createTablesOnStart = false // true for testing with H2 in memory, when not running unit tests. Set to false normally // } // gruntWatch = false //false for production, true for mvn tomcat7:run . Allows the client javascript and html files to be loaded via gruntWatch . } email { //add javax mail properties from https://www.tutorialspoint.com/javamail_api/javamail_api_smtp_servers.htm here // javaxmail { // mail { // smtp { //for postfix on localhost // host = localhost // port = 25 //for AWS SES - See http://docs.aws.amazon.com/ses/latest/DeveloperGuide/send-using-smtp-java.html // host = email-smtp.us-east-1.amazonaws.com // port = 25 // transport.protocol = smtps // auth = true // starttls.enable = true // starttls.required = true // } // } // } //Must be set for AWS SES. See http://docs.aws.amazon.com/ses/latest/DeveloperGuide/send-using-smtp-java.html // authenticator { // username = yourUsername // password = yourPassword // } } } //Default settings for akka //akka { // loglevel = INFO // log-config-on-start = on // loggers = ["akka.event.slf4j.Slf4jLogger"] // Toggles whether the threads created by this ActorSystem should be daemons or not. Use daemonic inside of tomcat to support shutdown // daemonic = on //} //You'll see these settings for spray, baked into the .war files. //spray.servlet { // boot-class = "net.shrine.dashboard.net.shrine.metadata.Boot" //Don't change this one. It'll start the wrong (or no) application if you change it. // request-timeout = 30s //}