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.util.json.JsonSupport
  40 import collection.mutable.ListBuffer
  41 import gr.grnet.aquarium.event.model.im.IMEventModel
  42 import gr.grnet.aquarium.event.model.im.IMEventModel.{Names ⇒ IMEventNames}
  43 import gr.grnet.aquarium.event.model.resource.ResourceEventModel
  44 import gr.grnet.aquarium.event.model.resource.ResourceEventModel.{Names ⇒ ResourceEventNames}
  45 import gr.grnet.aquarium.store._
  46 import com.mongodb._
  47 import org.bson.types.ObjectId
  48 import gr.grnet.aquarium.util._
  49 import gr.grnet.aquarium.converter.StdConverters
  50 import gr.grnet.aquarium.event.model.ExternalEventModel
  51 import gr.grnet.aquarium.computation.BillingMonthInfo
  52 import gr.grnet.aquarium.policy.PolicyModel
  53 import gr.grnet.aquarium.{Aquarium, AquariumException}
  54 import collection.immutable.SortedMap
  55 import gr.grnet.aquarium.logic.accounting.dsl.Timeslot
  56 import collection.immutable
  57 import gr.grnet.aquarium.charging.state.UserStateModel
  58
  59 /**
  60  * Mongodb implementation of the various aquarium stores.
  61  *
  62  * @author Christos KK Loverdos <loverdos@gmail.com>
  63  * @author Georgios Gousios <gousiosg@gmail.com>
  64  * @author Prodromos Gerakios <pgerakio@grnet.gr>
  65  */
  66 class MongoDBStore(
  67     val aquarium: Aquarium,
  68     val mongo: Mongo,
  69     val database: String,
  70     val username: String,
  71     val password: String)
  72   extends ResourceEventStore
  73   with UserStateStore
  74   with IMEventStore
  75   with PolicyStore
  76   with Loggable {
  77
  78   override type IMEvent = MongoDBIMEvent
  79   override type ResourceEvent = MongoDBResourceEvent
  80   override type Policy = MongoDBPolicy
  81   override type UserState = MongoDBUserState
  82
  83   private[store] lazy val resourceEvents = getCollection(MongoDBStore.RESOURCE_EVENTS_COLLECTION)
  84   private[store] lazy val userStates = getCollection(MongoDBStore.USER_STATES_COLLECTION)
  85   private[store] lazy val imEvents = getCollection(MongoDBStore.IM_EVENTS_COLLECTION)
  86   private[store] lazy val policies = getCollection(MongoDBStore.POLICY_COLLECTION)
  87
  88   private[this] def getCollection(name: String): DBCollection = {
  89     val db = mongo.getDB(database)
  90     //logger.debug("Authenticating to mongo")
  91     if(!db.isAuthenticated && !db.authenticate(username, password.toCharArray)) {
  92       throw new AquariumException("Could not authenticate user %s".format(username))
  93     }
  94     db.getCollection(name)
  95   }
  96
  97   //+ResourceEventStore
  98   def createResourceEventFromOther(event: ResourceEventModel): ResourceEvent = {
  99     MongoDBResourceEvent.fromOther(event, null)
 100   }
 101
 102   def pingResourceEventStore(): Unit = synchronized {
 103     MongoDBStore.ping(mongo)
 104   }
 105
 106   def insertResourceEvent(event: ResourceEventModel) = {
 107     val localEvent = MongoDBResourceEvent.fromOther(event, new ObjectId().toStringMongod)
 108     MongoDBStore.insertObject(localEvent, resourceEvents, MongoDBStore.jsonSupportToDBObject)
 109     localEvent
 110   }
 111
 112   def findResourceEventByID(id: String): Option[ResourceEvent] = {
 113     MongoDBStore.findBy(ResourceEventNames.id, id, resourceEvents, MongoDBResourceEvent.fromDBObject)
 114   }
 115
 116   def findResourceEventsByUserID(userId: String)
 117                                 (sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
 118     val query = new BasicDBObject(ResourceEventNames.userID, userId)
 119
 120     MongoDBStore.runQuery(query, resourceEvents)(MongoDBResourceEvent.fromDBObject)(sortWith)
 121   }
 122
 123   def countOutOfSyncResourceEventsForBillingPeriod(userID: String, startMillis: Long, stopMillis: Long): Long = {
 124     val query = new BasicDBObjectBuilder().
 125       add(ResourceEventModel.Names.userID, userID).
 126       // received within the period
 127       add(ResourceEventModel.Names.receivedMillis, new BasicDBObject("$gte", startMillis)).
 128       add(ResourceEventModel.Names.receivedMillis, new BasicDBObject("$lte", stopMillis)).
 129       // occurred outside the period
 130       add("$or", {
 131         val dbList = new BasicDBList()
 132         dbList.add(0, new BasicDBObject(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$lt", startMillis)))
 133         dbList.add(1, new BasicDBObject(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$gt", stopMillis)))
 134         dbList
 135       }).
 136       get()
 137
 138     resourceEvents.count(query)
 139   }
 140
 141   def foreachResourceEventOccurredInPeriod(
 142       userID: String,
 143       startMillis: Long,
 144       stopMillis: Long
 145   )(f: ResourceEvent ⇒ Unit): Unit = {
 146
 147     val query = new BasicDBObjectBuilder().
 148       add(ResourceEventModel.Names.userID, userID).
 149       add(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$gte", startMillis)).
 150       add(ResourceEventModel.Names.occurredMillis, new BasicDBObject("$lte", stopMillis)).
 151       get()
 152
 153     val sorter = new BasicDBObject(ResourceEventModel.Names.occurredMillis, 1)
 154     val cursor = resourceEvents.find(query).sort(sorter)
 155
 156     withCloseable(cursor) { cursor ⇒
 157       while(cursor.hasNext) {
 158         val nextDBObject = cursor.next()
 159         val nextEvent = MongoDBResourceEvent.fromDBObject(nextDBObject)
 160
 161         f(nextEvent)
 162       }
 163     }
 164   }
 165   //-ResourceEventStore
 166
 167   //+ UserStateStore
 168   def insertUserState(userState: UserState) = {
 169     MongoDBStore.insertObject(
 170       userState.copy(_id = new ObjectId().toString),
 171       userStates,
 172       MongoDBStore.jsonSupportToDBObject
 173     )
 174   }
 175
 176   def findUserStateByUserID(userID: String): Option[UserState] = {
 177     val query = new BasicDBObject(UserStateModel.Names.userID, userID)
 178     val cursor = userStates find query
 179
 180     MongoDBStore.firstResultIfExists(cursor, MongoDBStore.dbObjectToUserState)
 181   }
 182
 183   def findLatestUserStateForFullMonthBilling(userID: String, bmi: BillingMonthInfo): Option[UserState] = {
 184     val query = new BasicDBObjectBuilder().
 185       add(UserStateModel.Names.userID, userID).
 186       add(UserStateModel.Names.isFullBillingMonth, true).
 187       add(UserStateModel.Names.billingYear, bmi.year).
 188       add(UserStateModel.Names.billingMonth, bmi.month).
 189       get()
 190
 191     // Descending order, so that the latest comes first
 192     val sorter = new BasicDBObject(UserStateModel.Names.occurredMillis, -1)
 193
 194     val cursor = userStates.find(query).sort(sorter)
 195
 196     MongoDBStore.firstResultIfExists(cursor, MongoDBStore.dbObjectToUserState)
 197   }
 198
 199   def createUserStateFromOther(userState: UserStateModel) = {
 200     MongoDBUserState.fromOther(userState, new ObjectId().toStringMongod)
 201   }
 202
 203   /**
 204    * Stores a user state.
 205    */
 206   def insertUserState(userState: UserStateModel): UserState = {
 207     val localUserState = createUserStateFromOther(userState)
 208     MongoDBStore.insertObject(localUserState, userStates, MongoDBStore.jsonSupportToDBObject)
 209   }
 210   //- UserStateStore
 211
 212   //+IMEventStore
 213   def createIMEventFromJson(json: String) = {
 214     MongoDBStore.createIMEventFromJson(json)
 215   }
 216
 217   def createIMEventFromOther(event: IMEventModel) = {
 218     MongoDBStore.createIMEventFromOther(event)
 219   }
 220
 221   def pingIMEventStore(): Unit = {
 222     MongoDBStore.ping(mongo)
 223   }
 224
 225   def insertIMEvent(event: IMEventModel): IMEvent = {
 226     val localEvent = MongoDBIMEvent.fromOther(event, new ObjectId().toStringMongod)
 227     MongoDBStore.insertObject(localEvent, imEvents, MongoDBStore.jsonSupportToDBObject)
 228     localEvent
 229   }
 230
 231   def findIMEventByID(id: String): Option[IMEvent] = {
 232     MongoDBStore.findBy(IMEventNames.id, id, imEvents, MongoDBIMEvent.fromDBObject)
 233   }
 234
 235
 236   /**
 237    * Find the `CREATE` even for the given user. Note that there must be only one such event.
 238    */
 239   def findCreateIMEventByUserID(userID: String): Option[IMEvent] = {
 240     val query = new BasicDBObjectBuilder().
 241       add(IMEventNames.userID, userID).
 242       add(IMEventNames.eventType, IMEventModel.EventTypeNames.create).get()
 243
 244     // Normally one such event is allowed ...
 245     val cursor = imEvents.find(query).sort(new BasicDBObject(IMEventNames.occurredMillis, 1))
 246
 247     MongoDBStore.firstResultIfExists(cursor, MongoDBIMEvent.fromDBObject)
 248   }
 249
 250   def findLatestIMEventByUserID(userID: String): Option[IMEvent] = {
 251     val query = new BasicDBObject(IMEventNames.userID, userID)
 252     val cursor = imEvents.find(query).sort(new BasicDBObject(IMEventNames.occurredMillis, -1))
 253
 254     MongoDBStore.firstResultIfExists(cursor, MongoDBIMEvent.fromDBObject)
 255   }
 256
 257   /**
 258    * Scans events for the given user, sorted by `occurredMillis` in ascending order and runs them through
 259    * the given function `f`.
 260    *
 261    * Any exception is propagated to the caller. The underlying DB resources are properly disposed in any case.
 262    */
 263   def foreachIMEventInOccurrenceOrder(userID: String)(f: (IMEvent) => Unit) = {
 264     val query = new BasicDBObject(IMEventNames.userID, userID)
 265     val cursor = imEvents.find(query).sort(new BasicDBObject(IMEventNames.occurredMillis, 1))
 266
 267     withCloseable(cursor) { cursor ⇒
 268       while(cursor.hasNext) {
 269         val model = MongoDBIMEvent.fromDBObject(cursor.next())
 270         f(model)
 271       }
 272     }
 273   }
 274   //-IMEventStore
 275
 276
 277
 278   //+PolicyStore
 279   /**
 280    * Store an accounting policy.
 281    */
 282   def insertPolicy(policy: PolicyModel): Policy = {
 283     val dbPolicy = MongoDBPolicy.fromOther(policy, new ObjectId().toStringMongod)
 284     MongoDBStore.insertObject(dbPolicy, policies, MongoDBStore.jsonSupportToDBObject)
 285   }
 286
 287   private def emptyMap = immutable.SortedMap[Timeslot,Policy]()
 288
 289   def loadValidPolicyAt(atMillis: Long): Option[Policy] = {
 290     /*var d = new Date(atMillis)
 291     /* sort in reverse order  and return the first that includes this date*/
 292     policies.sortWith({(x,y)=> y.validFrom < x.validFrom}).collectFirst({
 293       case t if(t.validityTimespan.toTimeslot.includes(d)) => t
 294     })*/
 295     throw new UnsupportedOperationException
 296   }
 297
 298   def loadAndSortPoliciesWithin(fromMillis: Long, toMillis: Long): SortedMap[Timeslot, Policy] = {
 299
 300     throw new UnsupportedOperationException
 301   }
 302   //-PolicyStore
 303 }
 304
 305 object MongoDBStore {
 306   object JsonNames {
 307     final val _id = "_id"
 308   }
 309
 310   /**
 311    * Collection holding the [[gr.grnet.aquarium.event.model.resource.ResourceEventModel]]s.
 312    *
 313    * Resource events are coming from all systems handling billable resources.
 314    */
 315   final val RESOURCE_EVENTS_COLLECTION = "resevents"
 316
 317   /**
 318    * Collection holding the snapshots of [[gr.grnet.aquarium.charging.state.UserStateModel]].
 319    *
 320    * [[gr.grnet.aquarium.charging.state.UserStateModel]] is held internally within
 321    * [[gr.grnet.aquarium.actor.service.user.UserActor]]s.
 322    */
 323   final val USER_STATES_COLLECTION = "userstates"
 324
 325   /**
 326    * Collection holding [[gr.grnet.aquarium.event.model.im.IMEventModel]]s.
 327    *
 328    * User events are coming from the IM module (external).
 329    */
 330   final val IM_EVENTS_COLLECTION = "imevents"
 331
 332   /**
 333    * Collection holding [[gr.grnet.aquarium.policy.PolicyModel]]s.
 334    */
 335   final val POLICY_COLLECTION = "policies"
 336
 337   def dbObjectToUserState(dbObj: DBObject): MongoDBUserState = {
 338     MongoDBUserState.fromJSONString(JSON.serialize(dbObj))
 339   }
 340
 341   def firstResultIfExists[A](cursor: DBCursor, f: DBObject ⇒ A): Option[A] = {
 342     withCloseable(cursor) { cursor ⇒
 343       if(cursor.hasNext) {
 344         Some(f(cursor.next()))
 345       } else {
 346         None
 347       }
 348     }
 349   }
 350
 351   def ping(mongo: Mongo): Unit = synchronized {
 352     // This requires a network roundtrip
 353     mongo.isLocked
 354   }
 355
 356   def findBy[A >: Null <: AnyRef](name: String,
 357                                   value: String,
 358                                   collection: DBCollection,
 359                                   deserializer: (DBObject) => A) : Option[A] = {
 360     val query = new BasicDBObject(name, value)
 361     val cursor = collection find query
 362
 363     withCloseable(cursor) { cursor ⇒
 364       if(cursor.hasNext)
 365         Some(deserializer apply cursor.next)
 366       else
 367         None
 368     }
 369   }
 370
 371   def runQuery[A <: ExternalEventModel](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
 372                                   (deserializer: (DBObject) => A)
 373                                   (sortWith: Option[(A, A) => Boolean]): List[A] = {
 374     val cursor0 = collection find query
 375     val cursor = if(orderBy ne null) {
 376       cursor0 sort orderBy
 377     } else {
 378       cursor0
 379     } // I really know that docs say that it is the same cursor.
 380
 381     if(!cursor.hasNext) {
 382       cursor.close()
 383       Nil
 384     } else {
 385       val buff = new ListBuffer[A]()
 386
 387       while(cursor.hasNext) {
 388         buff += deserializer apply cursor.next
 389       }
 390
 391       cursor.close()
 392
 393       sortWith match {
 394         case Some(sorter) => buff.toList.sortWith(sorter)
 395         case None => buff.toList
 396       }
 397     }
 398   }
 399
 400   def insertObject[A <: AnyRef](obj: A, collection: DBCollection, serializer: A ⇒ DBObject) : A = {
 401     collection.insert(serializer apply obj, WriteConcern.JOURNAL_SAFE)
 402     obj
 403   }
 404
 405   def jsonSupportToDBObject(jsonSupport: JsonSupport) = {
 406     StdConverters.AllConverters.convertEx[DBObject](jsonSupport)
 407   }
 408
 409   final def isLocalIMEvent(event: IMEventModel) = event match {
 410     case _: MongoDBIMEvent ⇒ true
 411     case _ ⇒ false
 412   }
 413
 414   final def createIMEventFromJson(json: String) = {
 415     MongoDBIMEvent.fromJsonString(json)
 416   }
 417
 418   final def createIMEventFromOther(event: IMEventModel) = {
 419     MongoDBIMEvent.fromOther(event, new ObjectId().toStringMongod)
 420   }
 421
 422   final def createIMEventFromJsonBytes(jsonBytes: Array[Byte]) = {
 423     MongoDBIMEvent.fromJsonBytes(jsonBytes)
 424   }
 425 }