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 gr.grnet.aquarium.logic.accounting.Policy
  51 import gr.grnet.aquarium.logic.accounting.dsl.{Timeslot, DSLPolicy, DSLComplexResource}
  52 import gr.grnet.aquarium.logic.events._
  53 import com.mongodb._
  54 import com.ckkloverdos.maybe.{NoVal, Maybe}
  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 countOutOfSyncEventsForBillingPeriod(userId: String, startMillis: Long, stopMillis: Long): 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
 192   def findUserStateByUserId(userId: String): Maybe[UserState] = {
 193     Maybe {
 194       val query = new BasicDBObject(UserStateJsonNames.userId, userId)
 195       val cursor = userStates find query
 196
 197       try {
 198         if(cursor.hasNext)
 199           MongoDBStore.dbObjectToUserState(cursor.next())
 200         else
 201           null
 202       } finally {
 203         cursor.close()
 204       }
 205     }
 206   }
 207
 208   def findLatestUserStateForEndOfBillingMonth(userId: String,
 209                                               yearOfBillingMonth: Int,
 210                                               billingMonth: Int): Maybe[UserState] = {
 211     NoVal // FIXME: implement
 212   }
 213
 214   def deleteUserState(userId: String) = {
 215     val query = new BasicDBObject(UserStateJsonNames.userId, userId)
 216     userStates.findAndRemove(query)
 217   }
 218   //- UserStateStore
 219
 220   //+WalletEntryStore
 221   def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] =
 222     MongoDBStore.storeAquariumEvent(entry, walletEntries)
 223
 224   def findWalletEntryById(id: String): Maybe[WalletEntry] =
 225     MongoDBStore.findById[WalletEntry](id, walletEntries, MongoDBStore.dbObjectToWalletEntry)
 226
 227   def findUserWalletEntries(userId: String) = {
 228     // TODO: optimize
 229     findUserWalletEntriesFromTo(userId, new Date(0), new Date(Int.MaxValue))
 230   }
 231
 232   def findUserWalletEntriesFromTo(userId: String, from: Date, to: Date) : List[WalletEntry] = {
 233     val q = new BasicDBObject()
 234     // TODO: Is this the correct way for an AND query?
 235     q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
 236     q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$lt", to.getTime))
 237     q.put(WalletJsonNames.userId, userId)
 238
 239     MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
 240   }
 241
 242   def findWalletEntriesAfter(userId: String, from: Date) : List[WalletEntry] = {
 243     val q = new BasicDBObject()
 244     q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
 245     q.put(WalletJsonNames.userId, userId)
 246
 247     MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
 248   }
 249
 250   def findLatestUserWalletEntries(userId: String) = {
 251     Maybe {
 252       val orderBy = new BasicDBObject(WalletJsonNames.occurredMillis, -1) // -1 is descending order
 253       val cursor = walletEntries.find().sort(orderBy)
 254
 255       try {
 256         val buffer = new scala.collection.mutable.ListBuffer[WalletEntry]
 257         if(cursor.hasNext) {
 258           val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
 259           buffer += walletEntry
 260
 261           var _previousOccurredMillis = walletEntry.occurredMillis
 262           var _ok = true
 263
 264           while(cursor.hasNext && _ok) {
 265             val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
 266             var currentOccurredMillis = walletEntry.occurredMillis
 267             _ok = currentOccurredMillis == _previousOccurredMillis
 268
 269             if(_ok) {
 270               buffer += walletEntry
 271             }
 272           }
 273
 274           buffer.toList
 275         } else {
 276           null
 277         }
 278       } finally {
 279         cursor.close()
 280       }
 281     }
 282   }
 283
 284   def findPreviousEntry(userId: String, resource: String,
 285                         instanceId: String,
 286                         finalized: Option[Boolean]): List[WalletEntry] = {
 287     val q = new BasicDBObject()
 288     q.put(WalletJsonNames.userId, userId)
 289     q.put(WalletJsonNames.resource, resource)
 290     q.put(WalletJsonNames.instanceId, instanceId)
 291     finalized match {
 292       case Some(x) => q.put(WalletJsonNames.finalized, x)
 293       case None =>
 294     }
 295
 296     MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
 297   }
 298   //-WalletEntryStore
 299
 300   //+UserEventStore
 301   def storeUserEvent(event: UserEvent): Maybe[RecordID] =
 302     MongoDBStore.storeAny[UserEvent](event, userEvents, UserEventJsonNames.userId,
 303       _.userId, MongoDBStore.jsonSupportToDBObject)
 304
 305
 306   def findUserEventById(id: String): Maybe[UserEvent] =
 307     MongoDBStore.findById[UserEvent](id, userEvents, MongoDBStore.dbObjectToUserEvent)
 308
 309   def findUserEventsByUserId(userId: String): List[UserEvent] = {
 310     val query = new BasicDBObject(UserEventJsonNames.userId, userId)
 311     MongoDBStore.runQuery(query, userEvents)(MongoDBStore.dbObjectToUserEvent)(Some(_sortByTimestampAsc))
 312   }
 313   //-UserEventStore
 314
 315   //+PolicyStore
 316   def loadPolicies(after: Long): List[PolicyEntry] = {
 317     val query = new BasicDBObject(PolicyEntry.JsonNames.validFrom,
 318       new BasicDBObject("$gt", after))
 319     MongoDBStore.runQuery(query, policies)(MongoDBStore.dbObjectToPolicyEvent)(Some(_sortByTimestampAsc))
 320   }
 321
 322   def storePolicy(policy: PolicyEntry): Maybe[RecordID] = MongoDBStore.storeAquariumEvent(policy, policies)
 323
 324
 325   def updatePolicy(policy: PolicyEntry) = {
 326     //Find the entry
 327     val query = new BasicDBObject(PolicyEntry.JsonNames.id, policy.id)
 328     val policyObject = MongoDBStore.jsonSupportToDBObject(policy)
 329     policies.update(query, policyObject, true, false)
 330   }
 331   //-PolicyStore
 332 }
 333
 334 object MongoDBStore {
 335   object JsonNames {
 336     final val _id = "_id"
 337   }
 338
 339   /**
 340    * Collection holding the [[gr.grnet.aquarium.logic.events.ResourceEvent]]s.
 341    *
 342    * Resource events are coming from all systems handling billable resources.
 343    */
 344   final val RESOURCE_EVENTS_COLLECTION = "resevents"
 345
 346   /**
 347    * Collection holding the snapshots of [[gr.grnet.aquarium.user.UserState]].
 348    *
 349    * [[gr.grnet.aquarium.user.UserState]] is held internally within [[gr.grnet.aquarium.user.actor.UserActor]]s.
 350    */
 351   final val USER_STATES_COLLECTION = "userstates"
 352
 353   /**
 354    * Collection holding [[gr.grnet.aquarium.logic.events.UserEvent]]s.
 355    *
 356    * User events are coming from the IM module (external).
 357    */
 358   final val USER_EVENTS_COLLECTION = "userevents"
 359
 360   /**
 361    * Collection holding [[gr.grnet.aquarium.logic.events.WalletEntry]].
 362    *
 363    * Wallet entries are generated internally in Aquarium.
 364    */
 365   final val WALLET_ENTRIES_COLLECTION = "wallets"
 366
 367   /**
 368    * Collection holding [[gr.grnet.aquarium.logic.accounting.dsl.DSLPolicy]].
 369    */
 370   final val POLICIES_COLLECTION = "policies"
 371
 372   /* TODO: Some of the following methods rely on JSON (de-)serialization).
 373   * A method based on proper object serialization would be much faster.
 374   */
 375   def dbObjectToResourceEvent(dbObject: DBObject): ResourceEvent = {
 376     ResourceEvent.fromJson(JSON.serialize(dbObject))
 377   }
 378
 379   def dbObjectToUserState(dbObj: DBObject): UserState = {
 380     UserState.fromJson(JSON.serialize(dbObj))
 381   }
 382
 383   def dbObjectToWalletEntry(dbObj: DBObject): WalletEntry = {
 384     WalletEntry.fromJson(JSON.serialize(dbObj))
 385   }
 386
 387   def dbObjectToUserEvent(dbObj: DBObject): UserEvent = {
 388     UserEvent.fromJson(JSON.serialize(dbObj))
 389   }
 390
 391   def dbObjectToPolicyEvent(dbObj: DBObject): PolicyEntry = {
 392     PolicyEntry.fromJson(JSON.serialize(dbObj))
 393   }
 394
 395   def findById[A >: Null <: AquariumEvent](id: String, collection: DBCollection, deserializer: (DBObject) => A) : Maybe[A] = Maybe {
 396     val query = new BasicDBObject(ResourceJsonNames.id, id)
 397     val cursor = collection find query
 398
 399     try {
 400       if(cursor.hasNext)
 401         deserializer apply cursor.next
 402       else
 403         null: A // will be transformed to NoVal by the Maybe polymorphic constructor
 404     } finally {
 405       cursor.close()
 406     }
 407   }
 408
 409   def runQuery[A <: AquariumEvent](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
 410                                   (deserializer: (DBObject) => A)
 411                                   (sortWith: Option[(A, A) => Boolean]): List[A] = {
 412     val cursor0 = collection find query
 413     val cursor = if(orderBy ne null) {
 414       cursor0 sort orderBy
 415     } else {
 416       cursor0
 417     } // I really know that docs say that it is the same cursor.
 418
 419     if(!cursor.hasNext) {
 420       cursor.close()
 421       Nil
 422     } else {
 423       val buff = new ListBuffer[A]()
 424
 425       while(cursor.hasNext) {
 426         buff += deserializer apply cursor.next
 427       }
 428
 429       cursor.close()
 430
 431       sortWith match {
 432         case Some(sorter) => buff.toList.sortWith(sorter)
 433         case None => buff.toList
 434       }
 435     }
 436   }
 437
 438   def storeAquariumEvent[A <: AquariumEvent](event: A, collection: DBCollection) : Maybe[RecordID] = {
 439     storeAny[A](event, collection, ResourceJsonNames.id, (e) => e.id, MongoDBStore.jsonSupportToDBObject)
 440   }
 441
 442   def storeUserState(userState: UserState, collection: DBCollection): Maybe[RecordID] = {
 443     storeAny[UserState](userState, collection, ResourceJsonNames.userId, _.userId, MongoDBStore.jsonSupportToDBObject)
 444   }
 445
 446   def storeAny[A](any: A,
 447                   collection: DBCollection,
 448                   idName: String,
 449                   idValueProvider: (A) => String,
 450                   serializer: (A) => DBObject) : Maybe[RecordID] = {
 451     import com.ckkloverdos.maybe.effect
 452
 453     Maybe {
 454       val dbObj = serializer apply any
 455       val writeResult = collection insert dbObj
 456       writeResult.getLastError().throwOnError()
 457
 458       // Get back to retrieve unique id
 459       val cursor = collection.find(new BasicDBObject(idName, idValueProvider(any)))
 460       cursor
 461     } flatMap { cursor ⇒
 462       effect {
 463         if(cursor.hasNext)
 464           RecordID(cursor.next().get(JsonNames._id).toString)
 465         else
 466           throw new StoreException("Could not store %s to %s".format(any, collection))
 467       } {} { cursor.close() }
 468     }
 469   }
 470
 471   def jsonSupportToDBObject(any: JsonSupport): DBObject = {
 472     JSON.parse(any.toJson) match {
 473       case dbObject: DBObject ⇒
 474         dbObject
 475       case _ ⇒
 476         throw new StoreException("Could not transform %s -> %s".format(displayableObjectInfo(any), classOf[DBObject].getName))
 477     }
 478   }
 479 }