Fix bulk import parsing bug and limit concurrent client connections (#…

…973) * test (webapi): Add test for #971. * fix (webapi): Block when making HTTP client connections (#972). * fix (StringFormatter): Fix #971. - Remove NIE-INE test data. * style (ResourcesV1R2RSpec): Remove extra comma. * docs (release-notes): Update release notes. * build (webapi): raise release version to v1.7.1 * feature (webapi): Make triplestore and Sipi connection timeouts configurable. - Improve error handling on timeout. * test (bulk import): Add test using knoraXmlImport__seqnum.
dasch-swiss · Aug 22, 2018 · 7d37918 · 7d37918
1 parent d0dbe6f
commit 7d37918
Show file tree

Hide file tree

Showing 12 changed files with 246 additions and 118 deletions.
diff --git a/docs/DocsBuild.sbt b/docs/DocsBuild.sbt
@@ -16,7 +16,7 @@ lazy val docs = (project in file(".")).
     ).
     settings(
         // Set version string
-        version in ParadoxSite := "v1.7.0",
+        version in ParadoxSite := "v1.7.1",
 
         // Ghpages settings
         ghpagesNoJekyll := true,

diff --git a/docs/src/paradox/00-release-notes/v1.7.0.md b/docs/src/paradox/00-release-notes/v1.7.0.md
@@ -1,28 +1,36 @@
-v1.7.0 Release Notes
-====================
+# v1.7.x Release Notes
 
 See the
 [release](https://github.com/dhlab-basel/Knora/releases/tag/v1.7.0) and closed tickets on the
 [v1.7.0 milestone](https://github.com/dhlab-basel/Knora/milestone/11) on Github.
 
-Required changes to existing data:
-----------------------------------
+## v1.7.0
+
+### Required changes to existing data:
 
 - To use the inferred Gravsearch predicate `knora-api:standoffTagHasStartAncestor`,
   you must recreate your repository with the updated `KnoraRules.pie`.
 
-New features:
--------------
+### New features:
+
 
 - Gravsearch queries can now match standoff markup (@github[#910](#910)).
 - Add Graphdb-Free initialization scripts for local and docker installation (@github[#955](#955)).
 - Create temp dirs at startup (@github[#951](#951))
 - Update versions of monitoring tools (@github[#951](#951))
 
 
-Bugfixes:
----------
+### Bugfixes:
 
 - timeout or java.lang.OutOfMemoryError when using /v1/resources/xmlimportschemas/ for some ontologies (@github[#944](#944))
 - Timeout cleanup (@github[#951](#951))
 - Add separate dispatchers (@github[#945](#945))
+- "Property not found"-problem when using seqnum during bulk-import (@github[#971](#971))
+- Exceeded configured max-open-requests value (@github[#972](#972))
+
+## v1.7.1
+
+### Bugfixes:
+
+- "Property not found"-problem when using seqnum during bulk-import (@github[#971](#971))
+- Exceeded configured max-open-requests value (@github[#972](#972))
diff --git a/docs/src/paradox/00-release-notes/v1.8.0.md b/docs/src/paradox/00-release-notes/v1.8.0.md
@@ -1,15 +1,13 @@
-v1.8.0 Release Notes (not released yet)
-=======================================
+# v1.8.0 Release Notes (not released yet)
+
 
 See the
 [release](https://github.com/dhlab-basel/Knora/releases/tag/v1.8.0) and closed tickets on the
-[v1.7.0 milestone](https://github.com/dhlab-basel/Knora/milestone/12) on Github.
+[v1.8.0 milestone](https://github.com/dhlab-basel/Knora/milestone/12) on Github.
+
 
-Required changes to existing data:
-----------------------------------
+## Required changes to existing data:
 
-New features:
--------------
+## New features:
 
-Bugfixes:
----------
+## Bugfixes:
diff --git a/webapi/WebapiBuild.sbt b/webapi/WebapiBuild.sbt
@@ -10,7 +10,7 @@ connectInput in run := true
 lazy val webApiCommonSettings = Seq(
     organization := "org.knora",
     name := "webapi",
-    version := "1.7.0",
+    version := "1.7.1",
     scalaVersion := "2.12.4"
 )
 

diff --git a/webapi/_test_data/ontologies/anything-onto.ttl b/webapi/_test_data/ontologies/anything-onto.ttl
@@ -574,6 +574,23 @@
                                "maxlength=255" .
 
 
+:ThingWithSeqnum rdf:type owl:Class ;
+
+    rdfs:subClassOf :Thing ,
+        [
+           rdf:type owl:Restriction ;
+           owl:onProperty knora-base:seqnum ;
+           owl:minCardinality "0"^^xsd:nonNegativeInteger ;
+           salsah-gui:guiOrder "100"^^xsd:nonNegativeInteger
+        ] ;
+
+    rdfs:comment """Diese Resource-Klasse beschreibt ein Ding mit einer Sequenznummer"""@de ;
+
+    rdfs:label "Ding mit Sequenznummer"@de ,
+               "Chose avec numéro de séquence"@fr ,
+               "Cosa con numero di sequenza"@it ,
+               "Thing with sequence number"@en .
+
 
 :ThingPicture rdf:type owl:Class ;
 

diff --git a/webapi/src/main/resources/application.conf b/webapi/src/main/resources/application.conf
@@ -391,6 +391,8 @@ app {
         internal-host = "localhost"
         internal-port = 1024
 
+        timeout = 5 seconds
+
         // relevant for the client, i.e. browser
         external-protocol = "http"
         external-host = "localhost"
@@ -504,6 +506,12 @@ app {
 
         host = "localhost"
 
+        // timeout for triplestore queries
+        query-timeout = 10 seconds
+
+        // timeout for tripelstore updates
+        update-timeout = 15 minutes
+
         graphdb {
             port = 7200
             repository-name = "knora-test"

diff --git a/webapi/src/main/scala/org/knora/webapi/Settings.scala b/webapi/src/main/scala/org/knora/webapi/Settings.scala
@@ -87,6 +87,8 @@ class SettingsImpl(config: Config) extends Extension {
     val internalSipiPort: Int = config.getInt("app.sipi.internal-port")
     val internalSipiBaseUrl: String = internalSipiProtocol + "://" + internalSipiHost + (if (internalSipiPort != 80) ":" + internalSipiPort else "")
 
+    val sipiTimeout: FiniteDuration = getFiniteDuration("app.sipi.timeout", config)
+
     val externalSipiProtocol: String = config.getString("app.sipi.external-protocol")
     val externalSipiHost: String = config.getString("app.sipi.external-host")
     val externalSipiPort: Int = config.getInt("app.sipi.external-port")
@@ -132,6 +134,9 @@ class SettingsImpl(config: Config) extends Extension {
     val triplestoreType: String = config.getString("app.triplestore.dbtype")
     val triplestoreHost: String = config.getString("app.triplestore.host")
 
+    val triplestoreQueryTimeout: FiniteDuration = getFiniteDuration("app.triplestore.query-timeout", config)
+    val triplestoreUpdateTimeout: FiniteDuration = getFiniteDuration("app.triplestore.update-timeout", config)
+
     val triplestoreUseHttps: Boolean = config.getBoolean("app.triplestore.use-https")
 
     val triplestorePort: Int = triplestoreType match {

diff --git a/webapi/src/main/scala/org/knora/webapi/responders/v1/SipiResponderV1.scala b/webapi/src/main/scala/org/knora/webapi/responders/v1/SipiResponderV1.scala
@@ -38,16 +38,17 @@ import org.knora.webapi.util.ActorUtil._
 import org.knora.webapi.util.PermissionUtilADM
 import spray.json._
 
-import scala.concurrent.Future
+import scala.concurrent.{Await, Future, TimeoutException}
 import scala.concurrent.duration._
+import scala.util.{Failure, Try}
 
 /**
   * Responds to requests for information about binary representations of resources, and returns responses in Knora API
   * v1 format.
   */
 class SipiResponderV1 extends Responder {
 
-    implicit val materializer = ActorMaterializer()
+    implicit private val materializer: ActorMaterializer = ActorMaterializer()
 
     // Converts SPARQL query results to ApiValueV1 objects.
     val valueUtilV1 = new ValueUtilV1(settings)
@@ -57,10 +58,10 @@ class SipiResponderV1 extends Responder {
       * [[Status.Failure]]. If a serious error occurs (i.e. an error that isn't the client's fault), this
       * method first returns `Failure` to the sender, then throws an exception.
       */
-    def receive = {
+    def receive: PartialFunction[Any, Unit] = {
         case SipiFileInfoGetRequestV1(fileValueIri, userProfile) => future2Message(sender(), getFileInfoForSipiV1(fileValueIri, userProfile), log)
-        case convertPathRequest: SipiResponderConversionPathRequestV1 => future2Message(sender(), convertPathV1(convertPathRequest), log)
-        case convertFileRequest: SipiResponderConversionFileRequestV1 => future2Message(sender(), convertFileV1(convertFileRequest), log)
+        case convertPathRequest: SipiResponderConversionPathRequestV1 => try2message(sender(), convertPathV1(convertPathRequest), log)
+        case convertFileRequest: SipiResponderConversionFileRequestV1 => try2message(sender(), convertFileV1(convertFileRequest), log)
         case other => handleUnexpectedMessage(sender(), other, log, this.getClass.getName)
     }
 
@@ -87,7 +88,7 @@ class SipiResponderV1 extends Responder {
 
             // check that only one file value was found (by grouping by file value IRI)
             groupedByResourceIri = rows.groupBy {
-                (row: VariableResultsRow) =>
+                row: VariableResultsRow =>
                     row.rowMap("fileValue")
             }
             _ = if (groupedByResourceIri.size > 1) throw InconsistentTriplestoreDataException(s"filename $filename is referred to from more than one file value")
@@ -112,7 +113,7 @@ class SipiResponderV1 extends Responder {
       * @param conversionRequest the information about the file (uploaded by Knora).
       * @return a [[SipiResponderConversionResponseV1]] representing the file values to be added to the triplestore.
       */
-    private def convertPathV1(conversionRequest: SipiResponderConversionPathRequestV1): Future[SipiResponderConversionResponseV1] = {
+    private def convertPathV1(conversionRequest: SipiResponderConversionPathRequestV1): Try[SipiResponderConversionResponseV1] = {
         val url = s"${settings.internalSipiImageConversionUrl}/${settings.sipiPathConversionRoute}"
 
         callSipiConvertRoute(url, conversionRequest)
@@ -126,7 +127,7 @@ class SipiResponderV1 extends Responder {
       * @param conversionRequest the information about the file (managed by Sipi).
       * @return a [[SipiResponderConversionResponseV1]] representing the file values to be added to the triplestore.
       */
-    private def convertFileV1(conversionRequest: SipiResponderConversionFileRequestV1): Future[SipiResponderConversionResponseV1] = {
+    private def convertFileV1(conversionRequest: SipiResponderConversionFileRequestV1): Try[SipiResponderConversionResponseV1] = {
         val url = s"${settings.internalSipiImageConversionUrl}/${settings.sipiFileConversionRoute}"
 
         callSipiConvertRoute(url, conversionRequest)
@@ -141,7 +142,7 @@ class SipiResponderV1 extends Responder {
       * @param conversionRequest the message holding the information to make the request.
       * @return a [[SipiResponderConversionResponseV1]].
       */
-    private def callSipiConvertRoute(url: String, conversionRequest: SipiResponderConversionRequestV1): Future[SipiResponderConversionResponseV1] = {
+    private def callSipiConvertRoute(url: String, conversionRequest: SipiResponderConversionRequestV1): Try[SipiResponderConversionResponseV1] = {
 
         val conversionResultFuture: Future[HttpResponse] = for {
             request <- Marshal(FormData(conversionRequest.toFormData())).to[RequestEntity]
@@ -167,45 +168,68 @@ class SipiResponderV1 extends Responder {
                 throw SipiException(message = s"Unknown error: ${err.toString}", e = err, log = log)
         }
 
+        // Block until Sipi responds, to ensure that the number of concurrent connections to Sipi will never be greater
+        // than the value of akka.actor.deployment./responderManager/sipiRouterV1.nr-of-instances
+        val conversionResultTry: Try[HttpResponse] = try {
+            Await.ready(recoveredConversionResultFuture, settings.sipiTimeout)
+            recoveredConversionResultFuture.value.get
+        } catch {
+            case timeoutEx: TimeoutException => Failure(TriplestoreConnectionException(s"Connection to Sipi timed out after ${settings.sipiTimeout}", timeoutEx, log))
+        }
+
         for {
-            conversionResultResponse <- recoveredConversionResultFuture
+            conversionResultResponse: HttpResponse <- conversionResultTry
 
             httpStatusCode: StatusCode = conversionResultResponse.status
             statusInt: Int = httpStatusCode.intValue / 100
 
             /* get json from response body */
             responseAsJson: JsValue <- statusInt match {
-                case 2 => conversionResultResponse.entity.toStrict(5.seconds).map(
-                    (strict: Strict) =>
-                        try {
-                            strict.data.decodeString("UTF-8").parseJson
-                        } catch {
-                            // the Sipi response message could not be parsed correctly
-                            case e: spray.json.JsonParser.ParsingException => throw SipiException(message = "JSON response returned by Sipi is not valid JSON", e = e, log = log)
-
-                            case all: Exception => throw SipiException(message = "JSON response returned by Sipi is not valid JSON", e = all, log = log)
-                        }
-                ) // returns a Future(Map(...))
-                case 4 =>
-                    // Bad Request: it is the user's responsibility
-                    val errMessage: Future[SipiErrorConversionResponse] = conversionResultResponse.entity.toStrict(5.seconds).map(
-                        (strict: Strict) =>
-                            try {
-                                strict.data.decodeString("UTF-8").parseJson.convertTo[SipiErrorConversionResponse]
-                            } catch {
-                                // the Sipi error message could not be parsed correctly
-                                case e: spray.json.JsonParser.ParsingException => throw SipiException(message = "JSON error response returned by Sipi is invalid, it cannot be turned into a SipiErrorConversionResponse", e = e, log = log)
-
-                                case all: Exception => throw SipiException(message = "JSON error response returned by Sipi is not valid JSON", e = all, log = log)
+                case 2 | 4 | 5 =>
+                    val strictEntityFuture: Future[Strict] = conversionResultResponse.entity.toStrict(5.seconds)
+                    Await.ready(strictEntityFuture, 5.seconds)
+                    val strictEntityTry = strictEntityFuture.value.get
+
+                    statusInt match {
+                        case 2 =>
+                            // Success
+
+                            strictEntityTry.map {
+                                strict: Strict =>
+                                    try {
+                                        strict.data.decodeString("UTF-8").parseJson
+                                    } catch {
+                                        // the Sipi response message could not be parsed correctly
+                                        case e: spray.json.JsonParser.ParsingException => throw SipiException(message = "JSON response returned by Sipi is not valid JSON", e = e, log = log)
+
+                                        case all: Exception => throw SipiException(message = "JSON response returned by Sipi is not valid JSON", e = all, log = log)
+                                    }
+                            }
+
+                        case 4 =>
+                            // Bad Request: it is the user's responsibility
+
+                            val errMessage: Try[SipiErrorConversionResponse] = strictEntityTry.map {
+                                strict: Strict =>
+                                    try {
+                                        strict.data.decodeString("UTF-8").parseJson.convertTo[SipiErrorConversionResponse]
+                                    } catch {
+                                        // the Sipi error message could not be parsed correctly
+                                        case e: spray.json.JsonParser.ParsingException => throw SipiException(message = "JSON error response returned by Sipi is invalid, it cannot be turned into a SipiErrorConversionResponse", e = e, log = log)
+
+                                        case all: Exception => throw SipiException(message = "JSON error response returned by Sipi is not valid JSON", e = all, log = log)
+                                    }
                             }
-                    )
-
-                    // most probably the user sent invalid data which caused a Sipi error
-                    errMessage.map(errMsg => throw BadRequestException(s"Sipi returned a non successful HTTP status code $httpStatusCode: $errMsg"))
-                case 5 =>
-                    // Internal Server Error: not the user's fault
-                    val errString: Future[String] = conversionResultResponse.entity.toStrict(5.seconds).map(_.data.decodeString("UTF-8"))
-                    errString.map(errStr => throw SipiException(s"Sipi reported an internal server error $httpStatusCode - $errStr"))
+
+                            // most probably the user sent invalid data which caused a Sipi error
+                            errMessage.map(errMsg => throw BadRequestException(s"Sipi returned a non successful HTTP status code $httpStatusCode: $errMsg"))
+
+                        case 5 =>
+                            // Internal Server Error: not the user's fault
+                            val errString: Try[String] = strictEntityTry.map(_.data.decodeString("UTF-8"))
+                            errString.map(errStr => throw SipiException(s"Sipi reported an internal server error $httpStatusCode - $errStr"))
+                    }
+
                 case _ => throw SipiException(s"Sipi returned $httpStatusCode!")
             }