Feature/94 log user extraction (#123)

* Track user_id in Extraction log.

* Default to anonymous user when creating an Extraction event.

* Updated changelog. Fixes [#94].

Author: lmarini

CHANGELOG.md

@@ -4,9 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
-## Unreleased
+## [Unreleased]
 
 ### Added
+- Track user_id with every extraction event. [#94](https://github.com/clowder-framework/clowder/issues/94)
 - Added a new storage report at `GET api/reports/storage/spaces/:id` for auditing user storage usage on a space basis.
 - The file and dataset metrics reports also have support for since and until ISO8601 date parameters.
 

app/models/Extraction.scala

@@ -19,7 +19,9 @@ case class Extraction(
   extractor_id: String,
   status: String = "N/A",
   start: Date,
-  end: Option[Date])
+  end: Option[Date],
+  user_id: UUID = User.anonymous.id
+)
 
 /**
  * Currently running extractor name

app/services/RabbitmqPlugin.scala

@@ -436,14 +436,14 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     * @param file_id the UUID of file
     * @param extractor_id the extractor queue name to be submitted
     */
-  def postSubmissionEven(file_id: UUID, extractor_id: String): (UUID, Option[UUID]) = {
+  def postSubmissionEvent(file_id: UUID, extractor_id: String, user_id: UUID): (UUID, Option[UUID]) = {
     val extractions: ExtractionService = DI.injector.getInstance(classOf[ExtractionService])
 
     import java.text.SimpleDateFormat
     val dateFormatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssX")
     val submittedDateConvert = new java.util.Date()
     val job_id = Some(UUID.generate())
-    extractions.insert(Extraction(UUID.generate(), file_id, job_id, extractor_id, "SUBMITTED", submittedDateConvert, None)) match {
+    extractions.insert(Extraction(UUID.generate(), file_id, job_id, extractor_id, "SUBMITTED", submittedDateConvert, None, user_id)) match {
       case Some(objectid) => (UUID(objectid.toString), job_id)
       case None => (UUID(""), job_id)
     }
@@ -477,6 +477,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     val routingKey = exchange + "." + "file." + contentTypeToRoutingKey(file.contentType)
     val extraInfo = Map("filename" -> file.filename)
     val apiKey = requestAPIKey.getOrElse(globalAPIKey)
+    val user = userService.findByKey(apiKey).getOrElse(User.anonymous)
     val sourceExtra = JsObject((Seq("filename" -> JsString(file.filename))))
     Logger.debug(s"Sending message $routingKey from $host with extraInfo $extraInfo")
     var jobId: Option[UUID] = None
@@ -487,7 +488,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
 
           val notifies = getEmailNotificationEmailList(requestAPIKey)
 
-          val (id, job_id) = postSubmissionEven(file.id, queue)
+          val (id, job_id) = postSubmissionEvent(file.id, queue, user.id)
 
           val msg = ExtractorMessage(id, file.id, job_id, notifies, file.id, host, queue, extraInfo, file.length.toString,
             d.id, "", apiKey, routingKey, source, "created", None)
@@ -513,11 +514,12 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     val routingKey = exchange + "." + "file." + contentTypeToRoutingKey(file.contentType)
     val extraInfo = Map("filename" -> file.filename)
     val apiKey = requestAPIKey.getOrElse(globalAPIKey)
+    val user = userService.findByKey(apiKey).getOrElse(User.anonymous)
     Logger.debug(s"Sending message $routingKey from $host with extraInfo $extraInfo")
     val sourceExtra = JsObject((Seq("filename" -> JsString(file.filename))))
     val source = Entity(ResourceRef(ResourceRef.file, file.id), Some(file.contentType), sourceExtra)
     val notifies = getEmailNotificationEmailList(requestAPIKey)
-    val (id, job_id) = postSubmissionEven(file.id, routingKey)
+    val (id, job_id) = postSubmissionEvent(file.id, routingKey, user.id)
     val msg = ExtractorMessage(id, file.id, job_id, notifies, file.id, host, routingKey, extraInfo, file.length.toString, null,
       "", apiKey, routingKey, source, "created", None)
     extractWorkQueue(msg)
@@ -535,14 +537,15 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     Logger.debug(s"Sending message $routingKey from $host")
     val queues = getQueues(dataset, routingKey, file.contentType)
     val apiKey = requestAPIKey.getOrElse(globalAPIKey)
+    val user = userService.findByKey(apiKey).getOrElse(User.anonymous)
     val sourceExtra = JsObject((Seq("filename" -> JsString(file.filename))))
     queues.foreach{ extractorId =>
       val source = Entity(ResourceRef(ResourceRef.file, file.id), Some(file.contentType), sourceExtra)
       val target = Entity(ResourceRef(ResourceRef.dataset, dataset.id), None, JsObject(Seq.empty))
 
       val notifies = getEmailNotificationEmailList(requestAPIKey)
 
-      val (id, job_id) = postSubmissionEven(file.id, extractorId)
+      val (id, job_id) = postSubmissionEvent(file.id, extractorId, user.id)
 
       val msg = ExtractorMessage(id, file.id, job_id, notifies, file.id, host, extractorId, Map.empty, file.length.toString, dataset.id,
         "", apiKey, routingKey, source, "added", Some(target))
@@ -562,14 +565,15 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     Logger.debug(s"Sending message $routingKey from $host")
     val queues = getQueues(dataset, routingKey, "")
     val apiKey = requestAPIKey.getOrElse(globalAPIKey)
+    val user = userService.findByKey(apiKey).getOrElse(User.anonymous)
     val sourceExtra = JsObject((Seq("filenames" -> JsArray(filelist.map(f=>JsString(f.filename)).toSeq))))
     val msgExtra = Map("filenames" -> filelist.map(f=>f.filename))
     queues.foreach{ extractorId =>
       val source = Entity(ResourceRef(ResourceRef.dataset, dataset.id), None, sourceExtra)
 
       val notifies = getEmailNotificationEmailList(requestAPIKey)
 
-      val (id, job_id) = postSubmissionEven(dataset.id, extractorId)
+      val (id, job_id) = postSubmissionEvent(dataset.id, extractorId, user.id)
 
       var totalsize: Long = 0
       filelist.map(f => totalsize += f.length)
@@ -590,13 +594,14 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     Logger.debug(s"Sending message $routingKey from $host")
     val queues = getQueues(dataset, routingKey, file.contentType)
     val apiKey = requestAPIKey.getOrElse(globalAPIKey)
+    val user = userService.findByKey(apiKey).getOrElse(User.anonymous)
     val sourceExtra = JsObject((Seq("filename" -> JsString(file.filename))))
     queues.foreach{ extractorId =>
       val source = Entity(ResourceRef(ResourceRef.file, file.id), Some(file.contentType), sourceExtra)
       val target = Entity(ResourceRef(ResourceRef.dataset, dataset.id), None, JsObject(Seq.empty))
 
       val notifies = getEmailNotificationEmailList(requestAPIKey)
-      val (id, job_id) = postSubmissionEven(file.id, extractorId)
+      val (id, job_id) = postSubmissionEvent(file.id, extractorId, user.id)
       val msg = ExtractorMessage(id, file.id, job_id, notifies, file.id, host, extractorId, Map.empty, file.length.toString, dataset.id,
         "", apiKey, routingKey, source, "removed", Some(target))
       extractWorkQueue(msg)
@@ -621,7 +626,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     val source = Entity(ResourceRef(ResourceRef.file, originalId), Some(file.contentType), sourceExtra)
 
     val notifies = getEmailNotificationEmailList(requestAPIKey)
-    val (id, job_id) = postSubmissionEven(file.id, queue)
+    val (id, job_id) = postSubmissionEvent(file.id, queue, user.getOrElse(User.anonymous).id)
     val msg = ExtractorMessage(id, file.id, job_id, notifies, file.id, host, queue, extraInfo, file.length.toString, datasetId,
       "", apiKey, "extractors." + queue, source, "submitted", None)
     extractWorkQueue(msg)
@@ -644,7 +649,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
 
     val notifies = getEmailNotificationEmailList(requestAPIKey)
 
-    val (id, job_id) = postSubmissionEven(datasetId, queue)
+    val (id, job_id) = postSubmissionEvent(datasetId, queue, user.getOrElse(User.anonymous).id)
     val msg = ExtractorMessage(id, datasetId, job_id, notifies, datasetId, host, queue, extraInfo, 0.toString, datasetId,
       "", apiKey, "extractors." + queue, source, "submitted", None)
     extractWorkQueue(msg)
@@ -676,7 +681,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
               val source = Entity(ResourceRef(ResourceRef.metadata, metadataId), None, JsObject(Seq.empty))
               val target = Entity(resourceRef, None, JsObject(Seq.empty))
 
-              val (id, job_id) = postSubmissionEven(resourceRef.id, extractorId)
+              val (id, job_id) = postSubmissionEvent(resourceRef.id, extractorId, user.getOrElse(User.anonymous).id)
               val msg = ExtractorMessage(id, resourceRef.id, job_id, notifies, resourceRef.id, host, extractorId, extraInfo, 0.toString, resourceRef.id,
                 "", apiKey, routingKey, source, "added", Some(target))
               extractWorkQueue(msg)
@@ -693,7 +698,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
             val source = Entity(ResourceRef(ResourceRef.metadata, metadataId), None, JsObject(Seq.empty))
             val target = Entity(resourceRef, None, JsObject(Seq.empty))
 
-            val (id, job_id) = postSubmissionEven(resourceRef.id, extractorId)
+            val (id, job_id) = postSubmissionEvent(resourceRef.id, extractorId, user.getOrElse(User.anonymous).id)
             val msg = ExtractorMessage(id, resourceRef.id, job_id, notifies, resourceRef.id, host, extractorId, extraInfo, 0.toString, null,
               "", apiKey, routingKey, source, "added", Some(target))
             extractWorkQueue(msg)
@@ -728,7 +733,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
               val source = Entity(ResourceRef(ResourceRef.metadata, metadataId), None, JsObject(Seq.empty))
               val target = Entity(resourceRef, None, JsObject(Seq.empty))
 
-              val (id, job_id) = postSubmissionEven(resourceRef.id, extractorId)
+              val (id, job_id) = postSubmissionEvent(resourceRef.id, extractorId, user.getOrElse(User.anonymous).id)
               val msg = ExtractorMessage(id, resourceRef.id, job_id, notifies, resourceRef.id, host, extractorId, extraInfo, 0.toString, resourceRef.id,
                 "", apiKey, routingKey, source, "removed", Some(target))
               extractWorkQueue(msg)
@@ -745,7 +750,7 @@ class RabbitmqPlugin(application: Application) extends Plugin {
             val source = Entity(ResourceRef(ResourceRef.metadata, metadataId), None, JsObject(Seq.empty))
             val target = Entity(resourceRef, None, JsObject(Seq.empty))
 
-            val (id, job_id) = postSubmissionEven(resourceRef.id, extractorId)
+            val (id, job_id) = postSubmissionEvent(resourceRef.id, extractorId, user.getOrElse(User.anonymous).id)
             val msg = ExtractorMessage(id, resourceRef.id, job_id, notifies, resourceRef.id, host, extractorId, extraInfo, 0.toString, null,
               "", apiKey, routingKey, source, "removed", Some(target))
             extractWorkQueue(msg)
@@ -768,12 +773,13 @@ class RabbitmqPlugin(application: Application) extends Plugin {
     //since the thumbnail extractor during processing will need to upload to correct mongo collection.
     val routingKey = exchange +".query." + contentType.replace("/", ".")
     val apiKey = requestAPIKey.getOrElse(globalAPIKey)
+    val user = userService.findByKey(apiKey).getOrElse(User.anonymous)
     Logger.debug(s"Sending message $routingKey from $host")
 
     val notifies = getEmailNotificationEmailList(requestAPIKey)
     val source = Entity(ResourceRef(ResourceRef.file, tempFileId), Some(contentType), JsObject(Seq.empty))
 
-    val (id, job_id) = postSubmissionEven(tempFileId, routingKey)
+    val (id, job_id) = postSubmissionEvent(tempFileId, routingKey, user.id)
     val msg = ExtractorMessage(id, tempFileId, job_id, notifies, tempFileId, host, routingKey, Map.empty[String, Any], length, null,
       "", apiKey, routingKey, source, "created", None)
     extractWorkQueue(msg)
@@ -788,12 +794,13 @@ class RabbitmqPlugin(application: Application) extends Plugin {
   def submitSectionPreviewManually(preview: Preview, sectionId: UUID, host: String, requestAPIKey: Option[String]): Unit = {
     val routingKey = exchange + ".index."+ contentTypeToRoutingKey(preview.contentType)
     val apiKey = requestAPIKey.getOrElse(globalAPIKey)
+    val user = userService.findByKey(apiKey).getOrElse(User.anonymous)
     val extraInfo = Map("section_id"->sectionId)
     val source = Entity(ResourceRef(ResourceRef.preview, preview.id), None, JsObject(Seq.empty))
     val target = Entity(ResourceRef(ResourceRef.section, sectionId), None, JsObject(Seq.empty))
     val notifies = getEmailNotificationEmailList(requestAPIKey)
 
-    val (id, job_id) = postSubmissionEven(sectionId, routingKey)
+    val (id, job_id) = postSubmissionEvent(sectionId, routingKey, user.id)
     val msg = ExtractorMessage(id, sectionId, job_id, notifies, sectionId, host, routingKey, extraInfo, 0.toString, null,
       "", apiKey, routingKey, source, "added", Some(target))
     extractWorkQueue(msg)
@@ -961,24 +968,27 @@ class RabbitmqPlugin(application: Application) extends Plugin {
   * @param connection                     the connection to the rabbitmq
   * @param cancellationDownloadQueueName  the queue name of the cancellation downloaded queue
   */
-class PendingRequestCancellationActor(exchange: String, connection: Option[Connection], cancellationDownloadQueueName: String, cancellationSearchTimeout: Long) extends Actor {
+class PendingRequestCancellationActor(exchange: String, connection: Option[Connection], cancellationDownloadQueueName: String,
+                                      cancellationSearchTimeout: Long) extends Actor {
   val configuration = play.api.Play.configuration
   val CancellationSearchNumLimits: Integer = configuration.getString("submission.cancellation.search.numlimits").getOrElse("100").toInt
   def receive = {
     case CancellationMessage(id, queueName, msg_id) => {
       val extractions: ExtractionService = DI.injector.getInstance(classOf[ExtractionService])
       val dateFormatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssX")
       var startDate = new java.util.Date()
+      var user_id =  User.anonymous.id
       val job_id: Option[UUID] = extractions.get(msg_id) match {
         case Some(extraction) => {
+          user_id = extraction.user_id
           extraction.job_id
         }
         case None => {
           Logger.warn("Failed to lookup jobId.. no extraction message found with id=" + msg_id)
           None
         }
       }
-      extractions.insert(Extraction(UUID.generate(), id, job_id, queueName, "Cancel Requested", startDate, None))
+      extractions.insert(Extraction(UUID.generate(), id, job_id, queueName, "Cancel Requested", startDate, None, user_id))
 
       val channel: Channel = connection.get.createChannel()
       //1. connect to the target rabbitmq queue
@@ -1046,9 +1056,9 @@ class PendingRequestCancellationActor(exchange: String, connection: Option[Conne
       // update extraction event
       startDate = new java.util.Date()
       if(foundCancellationRequest) {
-        extractions.insert(Extraction(UUID.generate(), id, job_id, queueName, "Cancel Success", startDate, None))
+        extractions.insert(Extraction(UUID.generate(), id, job_id, queueName, "Cancel Success", startDate, None, user_id))
       } else {
-        extractions.insert(Extraction(UUID.generate(), id, job_id, queueName, "Cancel Failed", startDate, None))
+        extractions.insert(Extraction(UUID.generate(), id, job_id, queueName, "Cancel Failed", startDate, None, user_id))
       }
 
       try {
@@ -1194,6 +1204,7 @@ class EventFilter(channel: Channel, queue: String) extends Actor {
       Logger.debug("Received extractor status: " + statusBody)
       val json = Json.parse(statusBody)
       val file_id = UUID((json \ "file_id").as[String])
+      val user_id = UUID((json \ "user_id").as[String])
       val job_id: Option[UUID] = (json \ "job_id").asOpt[String] match {
         case Some(jid) => { Some(UUID(jid)) }
         case None => { None }
@@ -1209,11 +1220,11 @@ class EventFilter(channel: Channel, queue: String) extends Actor {
       //       other detailed status updates to logs when we start implementing
       //       distributed logging
       if (updatedStatus.contains("DONE")) {
-        extractions.insert(Extraction(UUID.generate(), file_id, job_id, extractor_id, "DONE", startDate.get, None))
+        extractions.insert(Extraction(UUID.generate(), file_id, job_id, extractor_id, "DONE", startDate.get, None, user_id))
       } else {
         val commKey = "key=" + play.Play.application().configuration().getString("commKey")
         val parsed_status = status.replace(commKey, "key=secretKey")
-        extractions.insert(Extraction(UUID.generate(), file_id, job_id, extractor_id, parsed_status, startDate.get, None))
+        extractions.insert(Extraction(UUID.generate(), file_id, job_id, extractor_id, parsed_status, startDate.get, None, user_id))
       }
       Logger.debug("updatedStatus=" + updatedStatus + " status=" + status + " startDate=" + startDate)
       models.ExtractionInfoSetUp.updateDTSRequests(file_id, extractor_id)