code.grnet.gr Git - aquarium/blob - src/main/scala/gr/grnet/aquarium/store/mongodb/MongoDBStore.scala

   1 /*
   2  * Copyright 2011-2012 GRNET S.A. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or
   5  * without modification, are permitted provided that the following
   6  * conditions are met:
   7  *
   8  *   1. Redistributions of source code must retain the above
   9  *      copyright notice, this list of conditions and the following
  10  *      disclaimer.
  11  *
  12  *   2. Redistributions in binary form must reproduce the above
  13  *      copyright notice, this list of conditions and the following
  14  *      disclaimer in the documentation and/or other materials
  15  *      provided with the distribution.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  18  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  19  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  20  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  21  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  24  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  25  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  27  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  28  * POSSIBILITY OF SUCH DAMAGE.
  29  *
  30  * The views and conclusions contained in the software and
  31  * documentation are those of the authors and should not be
  32  * interpreted as representing official policies, either expressed
  33  * or implied, of GRNET S.A.
  34  */
  35
  36 package gr.grnet.aquarium.store.mongodb
  37
  38 import com.mongodb.util.JSON
  39 import gr.grnet.aquarium.computation.state.UserState.{JsonNames ⇒ UserStateJsonNames}
  40 import gr.grnet.aquarium.util.json.JsonSupport
  41 import collection.mutable.ListBuffer
  42 import gr.grnet.aquarium.event.model.im.IMEventModel
  43 import gr.grnet.aquarium.event.model.im.IMEventModel.{Names ⇒ IMEventNames}
  44 import gr.grnet.aquarium.event.model.resource.ResourceEventModel
  45 import gr.grnet.aquarium.event.model.resource.ResourceEventModel.{Names ⇒ ResourceEventNames}
  46 import gr.grnet.aquarium.store._
  47 import com.mongodb._
  48 import org.bson.types.ObjectId
  49 import gr.grnet.aquarium.util._
  50 import gr.grnet.aquarium.converter.Conversions
  51 import gr.grnet.aquarium.computation.state.UserState
  52 import gr.grnet.aquarium.event.model.ExternalEventModel
  53 import gr.grnet.aquarium.computation.BillingMonthInfo
  54 import gr.grnet.aquarium.policy.PolicyModel
  55 import gr.grnet.aquarium.{Aquarium, AquariumException}
  56
  57 /**
  58  * Mongodb implementation of the various aquarium stores.
  59  *
  60  * @author Christos KK Loverdos <loverdos@gmail.com>
  61  * @author Georgios Gousios <gousiosg@gmail.com>
  62  */
  63 class MongoDBStore(
  64     val aquarium: Aquarium,
  65     val mongo: Mongo,
  66     val database: String,
  67     val username: String,
  68     val password: String)
  69   extends ResourceEventStore
  70   with UserStateStore
  71   with IMEventStore
  72   with PolicyStore
  73   with Loggable {
  74
  75   override type IMEvent = MongoDBIMEvent
  76   override type ResourceEvent = MongoDBResourceEvent
  77   override type Policy = MongoDBPolicy
  78
  79   private[store] lazy val resourceEvents = getCollection(MongoDBStore.RESOURCE_EVENTS_COLLECTION)
  80   private[store] lazy val userStates = getCollection(MongoDBStore.USER_STATES_COLLECTION)
  81   private[store] lazy val imEvents = getCollection(MongoDBStore.IM_EVENTS_COLLECTION)
  82   private[store] lazy val policies = getCollection(MongoDBStore.POLICY_COLLECTION)
  83
  84   private[this] def getCollection(name: String): DBCollection = {
  85     val db = mongo.getDB(database)
  86     //logger.debug("Authenticating to mongo")
  87     if(!db.isAuthenticated && !db.authenticate(username, password.toCharArray)) {
  88       throw new AquariumException("Could not authenticate user %s".format(username))
  89     }
  90     db.getCollection(name)
  91   }
  92
  93   //+ResourceEventStore
  94   def createResourceEventFromOther(event: ResourceEventModel): ResourceEvent = {
  95     MongoDBResourceEvent.fromOther(event, null)
  96   }
  97
  98   def pingResourceEventStore(): Unit = synchronized {
  99     MongoDBStore.ping(mongo)
 100   }
 101
 102   def insertResourceEvent(event: ResourceEventModel) = {
 103     val localEvent = MongoDBResourceEvent.fromOther(event, new ObjectId().toStringMongod)
 104     MongoDBStore.insertObject(localEvent, resourceEvents, MongoDBStore.jsonSupportToDBObject)
 105     localEvent
 106   }
 107
 108   def findResourceEventByID(id: String): Option[ResourceEvent] = {
 109     MongoDBStore.findBy(ResourceEventNames.id, id, resourceEvents, MongoDBResourceEvent.fromDBObject)
 110   }
 111
 112   def findResourceEventsByUserID(userId: String)
 113                                 (sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
 114     val query = new BasicDBObject(ResourceEventNames.userID, userId)
 115
 116     MongoDBStore.runQuery(query, resourceEvents)(MongoDBResourceEvent.fromDBObject)(sortWith)
 117   }
 118
 119   def countOutOfSyncResourceEventsForBillingPeriod(userID: String, startMillis: Long, stopMillis: Long): Long = {
 120     val query = new BasicDBObjectBuilder().
 121       add(ResourceEventModel.Names.userID, userID).
 122       // received within the period
 123       add(ResourceEventModel.Names.receivedMillis, new BasicDBObject("$gte", startMillis)).
 124       add(ResourceEventModel.Names.receivedMillis, new BasicDBObject("$lte", stopMillis)).
 125       // occurred outside the period
 126       add("$or", {
 127         val dbList = new BasicDBList()
 128         dbList.add(0, new BasicDBObject(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$lt", startMillis)))
 129         dbList.add(1, new BasicDBObject(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$gt", stopMillis)))
 130         dbList
 131       }).
 132       get()
 133
 134     resourceEvents.count(query)
 135   }
 136
 137   def foreachResourceEventOccurredInPeriod(
 138       userID: String,
 139       startMillis: Long,
 140       stopMillis: Long
 141   )(f: ResourceEvent ⇒ Unit): Unit = {
 142
 143     val query = new BasicDBObjectBuilder().
 144       add(ResourceEventModel.Names.userID, userID).
 145       add(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$gte", startMillis)).
 146       add(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$lte", stopMillis)).
 147       get()
 148
 149     val sorter = new BasicDBObject(ResourceEventModel.Names.occurredMillis, 1)
 150     val cursor = resourceEvents.find(query).sort(sorter)
 151
 152     withCloseable(cursor) { cursor ⇒
 153       while(cursor.hasNext) {
 154         val nextDBObject = cursor.next()
 155         val nextEvent = MongoDBResourceEvent.fromDBObject(nextDBObject)
 156
 157         f(nextEvent)
 158       }
 159     }
 160   }
 161   //-ResourceEventStore
 162
 163   //+ UserStateStore
 164   def insertUserState(userState: UserState) = {
 165     MongoDBStore.insertObject(
 166       userState.copy(_id = new ObjectId().toString),
 167       userStates,
 168       MongoDBStore.jsonSupportToDBObject
 169     )
 170   }
 171
 172   def findUserStateByUserID(userID: String): Option[UserState] = {
 173     val query = new BasicDBObject(UserStateJsonNames.userID, userID)
 174     val cursor = userStates find query
 175
 176     MongoDBStore.firstResultIfExists(cursor, MongoDBStore.dbObjectToUserState)
 177   }
 178
 179   def findLatestUserStateForFullMonthBilling(userID: String, bmi: BillingMonthInfo): Option[UserState] = {
 180     val query = new BasicDBObjectBuilder().
 181       add(UserState.JsonNames.userID, userID).
 182       add(UserState.JsonNames.isFullBillingMonthState, true).
 183       add(UserState.JsonNames.theFullBillingMonth_year, bmi.year).
 184       add(UserState.JsonNames.theFullBillingMonth_month, bmi.month).
 185       get()
 186
 187     // Descending order, so that the latest comes first
 188     val sorter = new BasicDBObject(UserState.JsonNames.occurredMillis, -1)
 189
 190     val cursor = userStates.find(query).sort(sorter)
 191
 192     MongoDBStore.firstResultIfExists(cursor, MongoDBStore.dbObjectToUserState)
 193   }
 194   //- UserStateStore
 195
 196   //+IMEventStore
 197   def createIMEventFromJson(json: String) = {
 198     MongoDBStore.createIMEventFromJson(json)
 199   }
 200
 201   def createIMEventFromOther(event: IMEventModel) = {
 202     MongoDBStore.createIMEventFromOther(event)
 203   }
 204
 205   def pingIMEventStore(): Unit = {
 206     MongoDBStore.ping(mongo)
 207   }
 208
 209   def insertIMEvent(event: IMEventModel): IMEvent = {
 210     val localEvent = MongoDBIMEvent.fromOther(event, new ObjectId().toStringMongod)
 211     MongoDBStore.insertObject(localEvent, imEvents, MongoDBStore.jsonSupportToDBObject)
 212     localEvent
 213   }
 214
 215   def findIMEventByID(id: String): Option[IMEvent] = {
 216     MongoDBStore.findBy(IMEventNames.id, id, imEvents, MongoDBIMEvent.fromDBObject)
 217   }
 218
 219
 220   /**
 221    * Find the `CREATE` even for the given user. Note that there must be only one such event.
 222    */
 223   def findCreateIMEventByUserID(userID: String): Option[IMEvent] = {
 224     val query = new BasicDBObjectBuilder().
 225       add(IMEventNames.userID, userID).
 226       add(IMEventNames.eventType, IMEventModel.EventTypeNames.create).get()
 227
 228     // Normally one such event is allowed ...
 229     val cursor = imEvents.find(query).sort(new BasicDBObject(IMEventNames.occurredMillis, 1))
 230
 231     MongoDBStore.firstResultIfExists(cursor, MongoDBIMEvent.fromDBObject)
 232   }
 233
 234   def findLatestIMEventByUserID(userID: String): Option[IMEvent] = {
 235     val query = new BasicDBObject(IMEventNames.userID, userID)
 236     val cursor = imEvents.find(query).sort(new BasicDBObject(IMEventNames.occurredMillis, -1))
 237
 238     MongoDBStore.firstResultIfExists(cursor, MongoDBIMEvent.fromDBObject)
 239   }
 240
 241   /**
 242    * Scans events for the given user, sorted by `occurredMillis` in ascending order and runs them through
 243    * the given function `f`.
 244    *
 245    * Any exception is propagated to the caller. The underlying DB resources are properly disposed in any case.
 246    */
 247   def foreachIMEventInOccurrenceOrder(userID: String)(f: (IMEvent) => Unit) = {
 248     val query = new BasicDBObject(IMEventNames.userID, userID)
 249     val cursor = imEvents.find(query).sort(new BasicDBObject(IMEventNames.occurredMillis, 1))
 250
 251     withCloseable(cursor) { cursor ⇒
 252       while(cursor.hasNext) {
 253         val model = MongoDBIMEvent.fromDBObject(cursor.next())
 254         f(model)
 255       }
 256     }
 257   }
 258   //-IMEventStore
 259
 260
 261
 262   //+PolicyStore
 263   def loadPoliciesAfter(after: Long): List[Policy] = {
 264     // FIXME implement
 265     throw new UnsupportedOperationException
 266   }
 267
 268
 269   def findPolicyByID(id: String) = {
 270     // FIXME implement
 271     throw new UnsupportedOperationException
 272   }
 273
 274   /**
 275    * Store an accounting policy.
 276    */
 277   def insertPolicy(policy: PolicyModel): Policy = {
 278     val dbPolicy = MongoDBPolicy.fromOther(policy, new ObjectId().toStringMongod)
 279     MongoDBStore.insertObject(dbPolicy, policies, MongoDBStore.jsonSupportToDBObject)
 280   }
 281   //-PolicyStore
 282 }
 283
 284 object MongoDBStore {
 285   object JsonNames {
 286     final val _id = "_id"
 287   }
 288
 289   /**
 290    * Collection holding the [[gr.grnet.aquarium.event.model.resource.ResourceEventModel]]s.
 291    *
 292    * Resource events are coming from all systems handling billable resources.
 293    */
 294   final val RESOURCE_EVENTS_COLLECTION = "resevents"
 295
 296   /**
 297    * Collection holding the snapshots of [[gr.grnet.aquarium.computation.state.UserState]].
 298    *
 299    * [[gr.grnet.aquarium.computation.state.UserState]] is held internally within
 300    * [[gr.grnet.aquarium.actor.service.user .UserActor]]s.
 301    */
 302   final val USER_STATES_COLLECTION = "userstates"
 303
 304   /**
 305    * Collection holding [[gr.grnet.aquarium.event.model.im.IMEventModel]]s.
 306    *
 307    * User events are coming from the IM module (external).
 308    */
 309   final val IM_EVENTS_COLLECTION = "imevents"
 310
 311   /**
 312    * Collection holding [[gr.grnet.aquarium.policy.PolicyModel]]s.
 313    */
 314   final val POLICY_COLLECTION = "policies"
 315
 316   def dbObjectToUserState(dbObj: DBObject): UserState = {
 317     UserState.fromJson(JSON.serialize(dbObj))
 318   }
 319
 320   def firstResultIfExists[A](cursor: DBCursor, f: DBObject ⇒ A): Option[A] = {
 321     withCloseable(cursor) { cursor ⇒
 322       if(cursor.hasNext) {
 323         Some(f(cursor.next()))
 324       } else {
 325         None
 326       }
 327     }
 328   }
 329
 330   def ping(mongo: Mongo): Unit = synchronized {
 331     // This requires a network roundtrip
 332     mongo.isLocked
 333   }
 334
 335   def findBy[A >: Null <: AnyRef](name: String,
 336                                   value: String,
 337                                   collection: DBCollection,
 338                                   deserializer: (DBObject) => A) : Option[A] = {
 339     val query = new BasicDBObject(name, value)
 340     val cursor = collection find query
 341
 342     withCloseable(cursor) { cursor ⇒
 343       if(cursor.hasNext)
 344         Some(deserializer apply cursor.next)
 345       else
 346         None
 347     }
 348   }
 349
 350   def runQuery[A <: ExternalEventModel](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
 351                                   (deserializer: (DBObject) => A)
 352                                   (sortWith: Option[(A, A) => Boolean]): List[A] = {
 353     val cursor0 = collection find query
 354     val cursor = if(orderBy ne null) {
 355       cursor0 sort orderBy
 356     } else {
 357       cursor0
 358     } // I really know that docs say that it is the same cursor.
 359
 360     if(!cursor.hasNext) {
 361       cursor.close()
 362       Nil
 363     } else {
 364       val buff = new ListBuffer[A]()
 365
 366       while(cursor.hasNext) {
 367         buff += deserializer apply cursor.next
 368       }
 369
 370       cursor.close()
 371
 372       sortWith match {
 373         case Some(sorter) => buff.toList.sortWith(sorter)
 374         case None => buff.toList
 375       }
 376     }
 377   }
 378
 379   def storeUserState(userState: UserState, collection: DBCollection) = {
 380     storeAny[UserState](userState, collection, ResourceEventNames.userID, _.userID, MongoDBStore.jsonSupportToDBObject)
 381   }
 382
 383   def storeAny[A](any: A,
 384                   collection: DBCollection,
 385                   idName: String,
 386                   idValueProvider: (A) => String,
 387                   serializer: (A) => DBObject) : RecordID = {
 388
 389     val dbObject = serializer apply any
 390     val _id = new ObjectId()
 391     dbObject.put("_id", _id)
 392     val writeResult = collection.insert(dbObject, WriteConcern.JOURNAL_SAFE)
 393     writeResult.getLastError().throwOnError()
 394
 395     RecordID(dbObject.get("_id").toString)
 396   }
 397
 398   def insertObject[A <: AnyRef](obj: A, collection: DBCollection, serializer: A ⇒ DBObject) : A = {
 399     collection.insert(serializer apply obj, WriteConcern.JOURNAL_SAFE)
 400     obj
 401   }
 402
 403   def jsonSupportToDBObject(jsonSupport: JsonSupport) = {
 404     Conversions.jsonSupportToDBObject(jsonSupport)
 405   }
 406
 407   final def isLocalIMEvent(event: IMEventModel) = event match {
 408     case _: MongoDBIMEvent ⇒ true
 409     case _ ⇒ false
 410   }
 411
 412   final def createIMEventFromJson(json: String) = {
 413     MongoDBIMEvent.fromJsonString(json)
 414   }
 415
 416   final def createIMEventFromOther(event: IMEventModel) = {
 417     MongoDBIMEvent.fromOther(event, null)
 418   }
 419
 420   final def createIMEventFromJsonBytes(jsonBytes: Array[Byte]) = {
 421     MongoDBIMEvent.fromJsonBytes(jsonBytes)
 422   }
 423 }