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