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

   1 /*
   2  * Copyright 2011 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 gr.grnet.aquarium.util.Loggable
  39 import com.mongodb.util.JSON
  40 import gr.grnet.aquarium.user.UserState
  41 import gr.grnet.aquarium.user.UserState.{JsonNames => UserStateJsonNames}
  42 import gr.grnet.aquarium.util.displayableObjectInfo
  43 import gr.grnet.aquarium.util.json.JsonSupport
  44 import collection.mutable.ListBuffer
  45 import gr.grnet.aquarium.store._
  46 import gr.grnet.aquarium.logic.events.ResourceEvent.{JsonNames => ResourceJsonNames}
  47 import gr.grnet.aquarium.logic.events.UserEvent.{JsonNames => UserEventJsonNames}
  48 import gr.grnet.aquarium.logic.events.WalletEntry.{JsonNames => WalletJsonNames}
  49 import java.util.Date
  50 import com.ckkloverdos.maybe.Maybe
  51 import gr.grnet.aquarium.logic.accounting.Policy
  52 import gr.grnet.aquarium.logic.accounting.dsl.{Timeslot, DSLPolicy, DSLComplexResource}
  53 import gr.grnet.aquarium.logic.events._
  54 import com.mongodb._
  55
  56 /**
  57  * Mongodb implementation of the various aquarium stores.
  58  *
  59  * @author Christos KK Loverdos <loverdos@gmail.com>
  60  * @author Georgios Gousios <gousiosg@gmail.com>
  61  */
  62 class MongoDBStore(
  63     val mongo: Mongo,
  64     val database: String,
  65     val username: String,
  66     val password: String)
  67   extends ResourceEventStore
  68   with UserStateStore
  69   with WalletEntryStore
  70   with UserEventStore
  71   with PolicyStore
  72   with Loggable {
  73
  74   private[store] lazy val resourceEvents = getCollection(MongoDBStore.RESOURCE_EVENTS_COLLECTION)
  75   private[store] lazy val userStates     = getCollection(MongoDBStore.USER_STATES_COLLECTION)
  76   private[store] lazy val userEvents     = getCollection(MongoDBStore.USER_EVENTS_COLLECTION)
  77   private[store] lazy val walletEntries  = getCollection(MongoDBStore.WALLET_ENTRIES_COLLECTION)
  78   private[store] lazy val policies      = getCollection(MongoDBStore.POLICIES_COLLECTION)
  79
  80   private[this] def getCollection(name: String): DBCollection = {
  81     val db = mongo.getDB(database)
  82     //logger.debug("Authenticating to mongo")
  83     if(!db.isAuthenticated && !db.authenticate(username, password.toCharArray)) {
  84       throw new StoreException("Could not authenticate user %s".format(username))
  85     }
  86     db.getCollection(name)
  87   }
  88
  89   private[this] def _sortByTimestampAsc[A <: AquariumEvent](one: A, two: A): Boolean = {
  90     if (one.occurredMillis > two.occurredMillis) false
  91     else if (one.occurredMillis < two.occurredMillis) true
  92     else true
  93   }
  94
  95   private[this] def _sortByTimestampDesc[A <: AquariumEvent](one: A, two: A): Boolean = {
  96     if (one.occurredMillis < two.occurredMillis) false
  97     else if (one.occurredMillis > two.occurredMillis) true
  98     else true
  99   }
 100
 101   //+ResourceEventStore
 102   def storeResourceEvent(event: ResourceEvent): Maybe[RecordID] =
 103     MongoDBStore.storeAquariumEvent(event, resourceEvents)
 104
 105   def findResourceEventById(id: String): Maybe[ResourceEvent] =
 106     MongoDBStore.findById(id, resourceEvents, MongoDBStore.dbObjectToResourceEvent)
 107
 108   def findResourceEventsByUserId(userId: String)
 109                                 (sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
 110     val query = new BasicDBObject(ResourceJsonNames.userId, userId)
 111
 112     MongoDBStore.runQuery(query, resourceEvents)(MongoDBStore.dbObjectToResourceEvent)(sortWith)
 113   }
 114
 115   def findResourceEventsByUserIdAfterTimestamp(userId: String, timestamp: Long): List[ResourceEvent] = {
 116     val query = new BasicDBObject()
 117     query.put(ResourceJsonNames.userId, userId)
 118     query.put(ResourceJsonNames.occurredMillis, new BasicDBObject("$gt", timestamp))
 119
 120     val sort = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
 121
 122     val cursor = resourceEvents.find(query).sort(sort)
 123
 124     try {
 125       val buffer = new scala.collection.mutable.ListBuffer[ResourceEvent]
 126       while(cursor.hasNext) {
 127         buffer += MongoDBStore.dbObjectToResourceEvent(cursor.next())
 128       }
 129       buffer.toList.sortWith(_sortByTimestampAsc)
 130     } finally {
 131       cursor.close()
 132     }
 133   }
 134
 135   def findResourceEventHistory(userId: String, resName: String,
 136                                instid: Option[String], upTo: Long) : List[ResourceEvent] = {
 137     val query = new BasicDBObject()
 138     query.put(ResourceJsonNames.userId, userId)
 139     query.put(ResourceJsonNames.occurredMillis, new BasicDBObject("$lt", upTo))
 140     query.put(ResourceJsonNames.resource, resName)
 141
 142     instid match {
 143       case Some(id) =>
 144         Policy.policy.findResource(resName) match {
 145           case Some(y) => query.put(ResourceJsonNames.details,
 146             new BasicDBObject(y.asInstanceOf[DSLComplexResource].descriminatorField, instid.get))
 147           case None =>
 148         }
 149       case None =>
 150     }
 151
 152     val sort = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
 153     val cursor = resourceEvents.find(query).sort(sort)
 154
 155     try {
 156       val buffer = new scala.collection.mutable.ListBuffer[ResourceEvent]
 157       while(cursor.hasNext) {
 158         buffer += MongoDBStore.dbObjectToResourceEvent(cursor.next())
 159       }
 160       buffer.toList.sortWith(_sortByTimestampAsc)
 161     } finally {
 162       cursor.close()
 163     }
 164   }
 165
 166   def findResourceEventsForReceivedPeriod(userId: String, startTimeMillis: Long, stopTimeMillis: Long): List[ResourceEvent] = {
 167     val query = new BasicDBObject()
 168     query.put(ResourceJsonNames.userId, userId)
 169     query.put(ResourceJsonNames.receivedMillis, new BasicDBObject("$gte", startTimeMillis))
 170     query.put(ResourceJsonNames.receivedMillis, new BasicDBObject("$lte", stopTimeMillis))
 171
 172     // Sort them by increasing order for occurred time
 173     val orderBy = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
 174
 175     MongoDBStore.runQuery[ResourceEvent](query, resourceEvents, orderBy)(MongoDBStore.dbObjectToResourceEvent)(None)
 176   }
 177
 178   def countOutOfSyncEventsForBillingMonth(userId: String, yearOfBillingMonth: Int, billingMonth: Int): Maybe[Long] = {
 179     Maybe {
 180       // FIXME: Implement
 181       0L
 182     }
 183   }
 184
 185   //-ResourceEventStore
 186
 187   //+UserStateStore
 188   def storeUserState(userState: UserState): Maybe[RecordID] =
 189     MongoDBStore.storeUserState(userState, userStates)
 190
 191   def findUserStateByUserId(userId: String): Maybe[UserState] = {
 192     Maybe {
 193       val query = new BasicDBObject(UserStateJsonNames.userId, userId)
 194       val cursor = userStates find query
 195
 196       try {
 197         if(cursor.hasNext)
 198           MongoDBStore.dbObjectToUserState(cursor.next())
 199         else
 200           null
 201       } finally {
 202         cursor.close()
 203       }
 204     }
 205   }
 206
 207   def deleteUserState(userId: String) = {
 208     val query = new BasicDBObject(UserStateJsonNames.userId, userId)
 209     userStates.findAndRemove(query)
 210   }
 211   //-UserStateStore
 212
 213   //+WalletEntryStore
 214   def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] =
 215     MongoDBStore.storeAquariumEvent(entry, walletEntries)
 216
 217   def findWalletEntryById(id: String): Maybe[WalletEntry] =
 218     MongoDBStore.findById[WalletEntry](id, walletEntries, MongoDBStore.dbObjectToWalletEntry)
 219
 220   def findUserWalletEntries(userId: String) = {
 221     // TODO: optimize
 222     findUserWalletEntriesFromTo(userId, new Date(0), new Date(Int.MaxValue))
 223   }
 224
 225   def findUserWalletEntriesFromTo(userId: String, from: Date, to: Date) : List[WalletEntry] = {
 226     val q = new BasicDBObject()
 227     // TODO: Is this the correct way for an AND query?
 228     q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
 229     q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$lt", to.getTime))
 230     q.put(WalletJsonNames.userId, userId)
 231
 232     MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
 233   }
 234
 235   def findWalletEntriesAfter(userId: String, from: Date) : List[WalletEntry] = {
 236     val q = new BasicDBObject()
 237     q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
 238     q.put(WalletJsonNames.userId, userId)
 239
 240     MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
 241   }
 242
 243   def findLatestUserWalletEntries(userId: String) = {
 244     Maybe {
 245       val orderBy = new BasicDBObject(WalletJsonNames.occurredMillis, -1) // -1 is descending order
 246       val cursor = walletEntries.find().sort(orderBy)
 247
 248       try {
 249         val buffer = new scala.collection.mutable.ListBuffer[WalletEntry]
 250         if(cursor.hasNext) {
 251           val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
 252           buffer += walletEntry
 253
 254           var _previousOccurredMillis = walletEntry.occurredMillis
 255           var _ok = true
 256
 257           while(cursor.hasNext && _ok) {
 258             val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
 259             var currentOccurredMillis = walletEntry.occurredMillis
 260             _ok = currentOccurredMillis == _previousOccurredMillis
 261
 262             if(_ok) {
 263               buffer += walletEntry
 264             }
 265           }
 266
 267           buffer.toList
 268         } else {
 269           null
 270         }
 271       } finally {
 272         cursor.close()
 273       }
 274     }
 275   }
 276
 277   def findPreviousEntry(userId: String, resource: String,
 278                         instanceId: String,
 279                         finalized: Option[Boolean]): List[WalletEntry] = {
 280     val q = new BasicDBObject()
 281     q.put(WalletJsonNames.userId, userId)
 282     q.put(WalletJsonNames.resource, resource)
 283     q.put(WalletJsonNames.instanceId, instanceId)
 284     finalized match {
 285       case Some(x) => q.put(WalletJsonNames.finalized, x)
 286       case None =>
 287     }
 288
 289     MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
 290   }
 291   //-WalletEntryStore
 292
 293   //+UserEventStore
 294   def storeUserEvent(event: UserEvent): Maybe[RecordID] =
 295     MongoDBStore.storeAny[UserEvent](event, userEvents, UserEventJsonNames.userId,
 296       _.userId, MongoDBStore.jsonSupportToDBObject)
 297
 298
 299   def findUserEventById(id: String): Maybe[UserEvent] =
 300     MongoDBStore.findById[UserEvent](id, userEvents, MongoDBStore.dbObjectToUserEvent)
 301
 302   def findUserEventsByUserId(userId: String): List[UserEvent] = {
 303     val query = new BasicDBObject(UserEventJsonNames.userId, userId)
 304     MongoDBStore.runQuery(query, userEvents)(MongoDBStore.dbObjectToUserEvent)(Some(_sortByTimestampAsc))
 305   }
 306   //-UserEventStore
 307
 308   //+PolicyStore
 309   def loadPolicies(after: Long): List[PolicyEntry] = {
 310     val query = new BasicDBObject(PolicyEntry.JsonNames.validFrom,
 311       new BasicDBObject("$gt", after))
 312     MongoDBStore.runQuery(query, policies)(MongoDBStore.dbObjectToPolicyEvent)(Some(_sortByTimestampAsc))
 313   }
 314
 315   def storePolicy(policy: PolicyEntry): Maybe[RecordID] = MongoDBStore.storeAquariumEvent(policy, policies)
 316
 317
 318   def updatePolicy(policy: PolicyEntry) = {
 319     //Find the entry
 320     val query = new BasicDBObject(PolicyEntry.JsonNames.id, policy.id)
 321     val policyObject = MongoDBStore.jsonSupportToDBObject(policy)
 322     policies.update(query, policyObject, true, false)
 323   }
 324
 325   def findPolicy(id: String) =
 326     MongoDBStore.findById[PolicyEntry](id, policies, MongoDBStore.dbObjectToPolicyEvent)
 327
 328   //-PolicyStore
 329 }
 330
 331 object MongoDBStore {
 332   object JsonNames {
 333     final val _id = "_id"
 334   }
 335
 336   /**
 337    * Collection holding the [[gr.grnet.aquarium.logic.events.ResourceEvent]]s.
 338    *
 339    * Resource events are coming from all systems handling billable resources.
 340    */
 341   final val RESOURCE_EVENTS_COLLECTION = "resevents"
 342
 343   /**
 344    * Collection holding the snapshots of [[gr.grnet.aquarium.user.UserState]].
 345    *
 346    * [[gr.grnet.aquarium.user.UserState]] is held internally within [[gr.grnet.aquarium.user.actor.UserActor]]s.
 347    */
 348   final val USER_STATES_COLLECTION = "userstates"
 349
 350   /**
 351    * Collection holding [[gr.grnet.aquarium.logic.events.UserEvent]]s.
 352    *
 353    * User events are coming from the IM module (external).
 354    */
 355   final val USER_EVENTS_COLLECTION = "userevents"
 356
 357   /**
 358    * Collection holding [[gr.grnet.aquarium.logic.events.WalletEntry]].
 359    *
 360    * Wallet entries are generated internally in Aquarium.
 361    */
 362   final val WALLET_ENTRIES_COLLECTION = "wallets"
 363
 364   /**
 365    * Collection holding [[gr.grnet.aquarium.logic.accounting.dsl.DSLPolicy]].
 366    */
 367   final val POLICIES_COLLECTION = "policies"
 368
 369   /* TODO: Some of the following methods rely on JSON (de-)serialization).
 370   * A method based on proper object serialization would be much faster.
 371   */
 372   def dbObjectToResourceEvent(dbObject: DBObject): ResourceEvent = {
 373     ResourceEvent.fromJson(JSON.serialize(dbObject))
 374   }
 375
 376   def dbObjectToUserState(dbObj: DBObject): UserState = {
 377     UserState.fromJson(JSON.serialize(dbObj))
 378   }
 379
 380   def dbObjectToWalletEntry(dbObj: DBObject): WalletEntry = {
 381     WalletEntry.fromJson(JSON.serialize(dbObj))
 382   }
 383
 384   def dbObjectToUserEvent(dbObj: DBObject): UserEvent = {
 385     UserEvent.fromJson(JSON.serialize(dbObj))
 386   }
 387
 388   def dbObjectToPolicyEvent(dbObj: DBObject): PolicyEntry = {
 389     PolicyEntry.fromJson(JSON.serialize(dbObj))
 390   }
 391
 392   def findById[A >: Null <: AquariumEvent](id: String, collection: DBCollection, deserializer: (DBObject) => A) : Maybe[A] = Maybe {
 393     val query = new BasicDBObject(ResourceJsonNames.id, id)
 394     val cursor = collection find query
 395
 396     try {
 397       if(cursor.hasNext)
 398         deserializer apply cursor.next
 399       else
 400         null: A // will be transformed to NoVal by the Maybe polymorphic constructor
 401     } finally {
 402       cursor.close()
 403     }
 404   }
 405
 406   def runQuery[A <: AquariumEvent](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
 407                                   (deserializer: (DBObject) => A)
 408                                   (sortWith: Option[(A, A) => Boolean]): List[A] = {
 409     val cursor0 = collection find query
 410     val cursor = if(orderBy ne null) {
 411       cursor0 sort orderBy
 412     } else {
 413       cursor0
 414     } // I really know that docs say that it is the same cursor.
 415
 416     if(!cursor.hasNext) {
 417       cursor.close()
 418       Nil
 419     } else {
 420       val buff = new ListBuffer[A]()
 421
 422       while(cursor.hasNext) {
 423         buff += deserializer apply cursor.next
 424       }
 425
 426       cursor.close()
 427
 428       sortWith match {
 429         case Some(sorter) => buff.toList.sortWith(sorter)
 430         case None => buff.toList
 431       }
 432     }
 433   }
 434
 435   def storeAquariumEvent[A <: AquariumEvent](event: A, collection: DBCollection) : Maybe[RecordID] = {
 436     storeAny[A](event, collection, ResourceJsonNames.id, (e) => e.id, MongoDBStore.jsonSupportToDBObject)
 437   }
 438
 439   def storeUserState(userState: UserState, collection: DBCollection): Maybe[RecordID] = {
 440     storeAny[UserState](userState, collection, ResourceJsonNames.userId, _.userId, MongoDBStore.jsonSupportToDBObject)
 441   }
 442
 443   def storeAny[A](any: A,
 444                   collection: DBCollection,
 445                   idName: String,
 446                   idValueProvider: (A) => String,
 447                   serializer: (A) => DBObject) : Maybe[RecordID] = {
 448     import com.ckkloverdos.maybe.effect
 449
 450     Maybe {
 451       val dbObj = serializer apply any
 452       val writeResult = collection insert dbObj
 453       writeResult.getLastError().throwOnError()
 454
 455       // Get back to retrieve unique id
 456       val cursor = collection.find(new BasicDBObject(idName, idValueProvider(any)))
 457       cursor
 458     } flatMap { cursor ⇒
 459       effect {
 460         if(cursor.hasNext)
 461           RecordID(cursor.next().get(JsonNames._id).toString)
 462         else
 463           throw new StoreException("Could not store %s to %s".format(any, collection))
 464       } {} { cursor.close() }
 465     }
 466   }
 467
 468   def jsonSupportToDBObject(any: JsonSupport): DBObject = {
 469     JSON.parse(any.toJson) match {
 470       case dbObject: DBObject ⇒
 471         dbObject
 472       case _ ⇒
 473         throw new StoreException("Could not transform %s -> %s".format(displayableObjectInfo(any), classOf[DBObject].getName))
 474     }
 475   }
 476 }