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