Notification Framework Update

README.md

@@ -224,6 +224,10 @@ Now you can disable root by setting `api.root.enabled` to `false` in `/etc/horiz
     - detect if a pattern is updated with service that has userInput w/o default values, and give warning
     - Consider changing all creates to POST, and update (via put/patch) return codes to 200
 
+## Changes in 2.7.0
+
+- Issue 277: Notification Framework Updates
+
 ## Changes in 2.6.0
 
 - Fixed issue 262 - get icp cluster name once at the beginning

src/main/resources/version.txt

@@ -1 +1 @@
-2.6.0
+2.7.0

src/main/scala/com/horizon/exchangeapi/OrgsRoutes.scala

@@ -528,7 +528,7 @@ trait OrgsRoutes extends JacksonSupport with AuthenticationSupport {
     } // end of exchAuth
   }
 
-  def buildResourceChangesResponse(orgList: scala.Seq[(Int, String, String, String, String, String, String, String)], ibmList: scala.Seq[(Int, String, String, String, String, String, String, String)], maxResp : Int): ResourceChangesRespObject ={
+  def buildResourceChangesResponse(orgList: scala.Seq[ResourceChangeRow], ibmList: scala.Seq[ResourceChangeRow], maxResp : Int): ResourceChangesRespObject ={
     val exchangeVersion = ExchangeApi.adminVersion()
     val inputList = List(orgList, ibmList)
     val changesList = ListBuffer[ChangeEntry]()
@@ -539,33 +539,21 @@ trait OrgsRoutes extends JacksonSupport with AuthenticationSupport {
       for(input <- inputList) { //this for loop should only ever be of size 2
         val changesMap = scala.collection.mutable.Map[String, ChangeEntry]() //using a Map allows us to avoid having a loop in a loop when searching the map for the resource id
         for( entry <- input) {
-          /*
-          Example of what entry might look like
-            {
-              "_1":167,   --> changeId
-              "_2":"org2",    --> orgID
-              "_3":"resourcetest",    --> id
-              "_4":"node",    --> category
-              "_5":"false",   --> public
-              "_6":"node",    --> resource
-              "_7":"created/modified",  --> operation
-              "_8":"2019-12-12T19:28:05.309Z[UTC]",   --> lastUpdated
-            }
-           */
-          val resChange = ResourceChangesInnerObject(entry._1, entry._8)
-          if(changesMap.isDefinedAt(entry._3+"_"+entry._6)){  // using the map allows for better searching and entry
-            if(changesMap(entry._3+"_"+entry._6).resourceChanges.last.changeId < entry._1){
-              // the entry we are looking at actually happened later than the last entry in resourceChanges
-              // doing this check by changeId on the off chance two changes happen at the exact same time changeId tells which one is most updated
-              changesMap(entry._3+"_"+entry._6).addToResourceChanges(resChange) // add the changeId and lastUpdated to the list of recent changes
-              changesMap(entry._3+"_"+entry._6).setOperation(entry._7) // update the most recent operation performed
-            }
-          } else{
-            val resChangeListBuffer = ListBuffer[ResourceChangesInnerObject](resChange)
-            changesMap(entry._3+"_"+entry._6) = ChangeEntry(entry._2, entry._6, entry._3, entry._7, resChangeListBuffer)
+          val resChange = ResourceChangesInnerObject(entry.changeId, entry.lastUpdated)
+          changesMap.get(entry.id+"_"+entry.resource) match { // using the map allows for better searching and entry
+            case Some(change) =>
+              if(change.resourceChanges.last.changeId < entry.changeId){
+                // the entry we are looking at actually happened later than the last entry in resourceChanges
+                // doing this check by changeId on the off chance two changes happen at the exact same time changeId tells which one is most updated
+                change.addToResourceChanges(resChange) // add the changeId and lastUpdated to the list of recent changes
+                change.setOperation(entry.operation) // update the most recent operation performed
+              }
+            case None => // add the change to the changesMap
+              val resChangeListBuffer = ListBuffer[ResourceChangesInnerObject](resChange)
+              changesMap.put(entry.id+"_"+entry.resource, ChangeEntry(entry.orgId, entry.resource, entry.id, entry.operation, resChangeListBuffer))
           }
           //check maxChangeIdOfQuery
-          if (entry._1 > maxChangeIdOfQuery) {maxChangeIdOfQuery = entry._1}
+          if (entry.changeId > maxChangeIdOfQuery) {maxChangeIdOfQuery = entry.changeId}
         }
         // convert changesMap to ListBuffer[ChangeEntry]
         breakable {
@@ -611,16 +599,19 @@ trait OrgsRoutes extends JacksonSupport with AuthenticationSupport {
           // Variables to help with building the query
           val lastTime = reqBody.lastUpdated.getOrElse(ApiTime.beginningUTC)
           //perf: reduce these 2 db queries to 1 db query
-          val qOrg = for {
-            r <- ResourceChangesTQ.rows.filter(_.orgId === orgId).filter(_.lastUpdated >= lastTime).filter(_.changeId >= reqBody.changeId)
-          } yield (r.changeId, r.orgId, r.id, r.category, r.public, r.resource, r.operation, r.lastUpdated)
-
+          var qOrgQuery = ResourceChangesTQ.rows.filter(_.orgId === orgId).filter(_.lastUpdated >= lastTime).filter(_.changeId >= reqBody.changeId)
+          ident match {
+            case _: INode =>
+              qOrgQuery = qOrgQuery.filter(u => (u.category === "node" && u.id === ident.getIdentity) || u.category =!= "node")
+            case _ => ;
+          }
+          val qOrg = for { r <- qOrgQuery } yield r
           val qPublic = for {
             r <- ResourceChangesTQ.rows.filter(_.orgId =!= orgId).filter(_.public === "true").filter(_.lastUpdated >= lastTime).filter(_.changeId >= reqBody.changeId)
-          } yield (r.changeId, r.orgId, r.id, r.category, r.public, r.resource, r.operation, r.lastUpdated)
+          } yield r
 
-          var qOrgResp : scala.Seq[(Int, String, String, String, String, String, String, String)] = null
-          var qPublicResp : scala.Seq[(Int, String, String, String, String, String, String, String)] = null
+          var qOrgResp : scala.Seq[ResourceChangeRow] = null
+          var qPublicResp : scala.Seq[ResourceChangeRow] = null
 
           db.run(qOrg.result.asTry.flatMap({
             case Success(qOrgResult) =>

src/test/scala/exchangeapi/NodesSuite.scala

@@ -107,6 +107,13 @@ class NodesSuite extends FunSuite {
   val ibmService = "TestIBMService"
   val maxRecords = 10000
   val secondsAgo = 120
+  val svcBase = "svc9920"
+  val svcDoc = "http://" + svcBase
+  val svcUrl = "" + svcBase
+  val svcVersion = "1.0.0"
+  val svcArch = "arm"
+  val service = svcBase + "_" + svcVersion + "_" + svcArch
+  val orgservice = authpref+service
 
   implicit val formats = DefaultFormats // Brings in default date formats etc.
 
@@ -1721,6 +1728,27 @@ class NodesSuite extends FunSuite {
     assert(parsedBody.changes.exists(y => {(y.id == nodeId3) && (y.operation == ResourceChangeConfig.DELETED) && (y.resource == "node")}))
   }
 
+  test("POST /orgs/"+orgid+"/services - add "+service+" as user that requires a service") {
+    val input = PostPutServiceRequest(svcBase+" arm", None, public = false, Some(svcDoc), svcUrl, svcVersion, svcArch, "multiple", None, None, Some(List(Map("name" -> "foo"))), "{\"services\":{}}","a",None)
+    val response = Http(URL+"/services").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(USERAUTH).asString
+    info("code: "+response.code+", response.body: "+response.body)
+    assert(response.code === HttpCode.POST_OK.intValue)
+    val respObj = parse(response.body).extract[ApiResponse]
+    assert(respObj.msg.contains("service '"+orgservice+"' created"))
+  }
+
+  test("POST /orgs/"+orgid+"/changes - verify " + nodeId + " doesn't see changes from other nodes but still sees normal changes") {
+    val time = ApiTime.pastUTC(secondsAgo)
+    val input = ResourceChangesRequest(0, Some(time), maxRecords, None)
+    val response = Http(URL+"/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(NODEAUTH).asString
+    info("code: "+response.code)
+    assert(response.code === HttpCode.POST_OK.intValue)
+    assert(!response.body.isEmpty)
+    val parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+    assert(!parsedBody.changes.exists(y => {y.id == nodeId3}))
+    assert(parsedBody.changes.exists(y => {(y.orgId == orgid) && (y.id == service) && (y.operation == ResourceChangeConfig.CREATED) && (y.resource == "service")}))
+  }
+
   test("PUT /orgs/"+orgid+"/nodes/"+nodeId+"/agreements/9952 - Try to add a 3rd agreement with low maxAgreements") {
     if (runningLocally) {     // changing limits via POST /admin/config does not work in multi-node mode
       // Get the current config value so we can restore it afterward