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