2 * Copyright 2011-2012 GRNET S.A. All rights reserved.
4 * Redistribution and use in source and binary forms, with or
5 * without modification, are permitted provided that the following
8 * 1. Redistributions of source code must retain the above
9 * copyright notice, this list of conditions and the following
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.
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.
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.
36 package gr.grnet.aquarium.store.mongodb
38 import com.mongodb.util.JSON
39 import gr.grnet.aquarium.user.UserState
40 import gr.grnet.aquarium.user.UserState.{JsonNames => UserStateJsonNames}
41 import gr.grnet.aquarium.util.json.JsonSupport
42 import collection.mutable.ListBuffer
43 import gr.grnet.aquarium.event.im.IMEventModel.{Names => IMEventNames}
44 import gr.grnet.aquarium.store._
45 import gr.grnet.aquarium.event.ResourceEvent.{JsonNames => ResourceJsonNames}
46 import gr.grnet.aquarium.event.WalletEntry.{JsonNames => WalletJsonNames}
47 import gr.grnet.aquarium.event.PolicyEntry.{JsonNames => PolicyJsonNames}
49 import gr.grnet.aquarium.logic.accounting.Policy
51 import org.bson.types.ObjectId
52 import gr.grnet.aquarium.event._
53 import com.ckkloverdos.maybe.{NoVal, Maybe}
54 import im.IMEventModel
55 import gr.grnet.aquarium.util._
56 import gr.grnet.aquarium.converter.{Conversions, StdConverters}
59 * Mongodb implementation of the various aquarium stores.
61 * @author Christos KK Loverdos <loverdos@gmail.com>
62 * @author Georgios Gousios <gousiosg@gmail.com>
69 extends ResourceEventStore
76 override type IMEvent = MongoDBIMEvent
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 unparsedIMEvents = getCollection(MongoDBStore.UNPARSED_IM_EVENTS_COLLECTION)
82 private[store] lazy val walletEntries = getCollection(MongoDBStore.WALLET_ENTRIES_COLLECTION)
83 private[store] lazy val policyEntries = getCollection(MongoDBStore.POLICY_ENTRIES_COLLECTION)
85 private[this] def getCollection(name: String): DBCollection = {
86 val db = mongo.getDB(database)
87 //logger.debug("Authenticating to mongo")
88 if(!db.isAuthenticated && !db.authenticate(username, password.toCharArray)) {
89 throw new StoreException("Could not authenticate user %s".format(username))
91 db.getCollection(name)
94 private[this] def _sortByTimestampAsc[A <: AquariumEventModel](one: A, two: A): Boolean = {
95 if (one.occurredMillis > two.occurredMillis) false
96 else if (one.occurredMillis < two.occurredMillis) true
100 private[this] def _sortByTimestampDesc[A <: AquariumEventSkeleton](one: A, two: A): Boolean = {
101 if (one.occurredMillis < two.occurredMillis) false
102 else if (one.occurredMillis > two.occurredMillis) true
106 //+ResourceEventStore
107 def storeResourceEvent(event: ResourceEvent) = {
108 MongoDBStore.storeAny[ResourceEvent](
111 ResourceJsonNames.id,
113 MongoDBStore.jsonSupportToDBObject)
116 def findResourceEventById(id: String): Maybe[ResourceEvent] =
117 MongoDBStore.findById(id, resourceEvents, MongoDBStore.dbObjectToResourceEvent)
119 def findResourceEventsByUserId(userId: String)
120 (sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
121 val query = new BasicDBObject(ResourceJsonNames.userId, userId)
123 MongoDBStore.runQuery(query, resourceEvents)(MongoDBStore.dbObjectToResourceEvent)(sortWith)
126 def findResourceEventsByUserIdAfterTimestamp(userId: String, timestamp: Long): List[ResourceEvent] = {
127 val query = new BasicDBObject()
128 query.put(ResourceJsonNames.userId, userId)
129 query.put(ResourceJsonNames.occurredMillis, new BasicDBObject("$gt", timestamp))
131 val sort = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
133 val cursor = resourceEvents.find(query).sort(sort)
136 val buffer = new scala.collection.mutable.ListBuffer[ResourceEvent]
137 while(cursor.hasNext) {
138 buffer += MongoDBStore.dbObjectToResourceEvent(cursor.next())
140 buffer.toList.sortWith(_sortByTimestampAsc)
146 def findResourceEventHistory(userId: String, resName: String,
147 instid: Option[String], upTo: Long) : List[ResourceEvent] = {
148 val query = new BasicDBObject()
149 query.put(ResourceJsonNames.userId, userId)
150 query.put(ResourceJsonNames.occurredMillis, new BasicDBObject("$lt", upTo))
151 query.put(ResourceJsonNames.resource, resName)
155 Policy.policy.findResource(resName) match {
156 case Some(y) => query.put(ResourceJsonNames.details,
157 new BasicDBObject(y.descriminatorField, instid.get))
163 val sort = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
164 val cursor = resourceEvents.find(query).sort(sort)
167 val buffer = new scala.collection.mutable.ListBuffer[ResourceEvent]
168 while(cursor.hasNext) {
169 buffer += MongoDBStore.dbObjectToResourceEvent(cursor.next())
171 buffer.toList.sortWith(_sortByTimestampAsc)
177 def findResourceEventsForReceivedPeriod(userId: String, startTimeMillis: Long, stopTimeMillis: Long): List[ResourceEvent] = {
178 val query = new BasicDBObject()
179 query.put(ResourceJsonNames.userId, userId)
180 query.put(ResourceJsonNames.receivedMillis, new BasicDBObject("$gte", startTimeMillis))
181 query.put(ResourceJsonNames.receivedMillis, new BasicDBObject("$lte", stopTimeMillis))
183 // Sort them by increasing order for occurred time
184 val orderBy = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
186 MongoDBStore.runQuery[ResourceEvent](query, resourceEvents, orderBy)(MongoDBStore.dbObjectToResourceEvent)(None)
189 def countOutOfSyncEventsForBillingPeriod(userId: String, startMillis: Long, stopMillis: Long): Maybe[Long] = {
196 def findAllRelevantResourceEventsForBillingPeriod(userId: String,
198 stopMillis: Long): List[ResourceEvent] = {
202 //-ResourceEventStore
205 def storeUserState(userState: UserState): Maybe[RecordID] = {
206 MongoDBStore.storeUserState(userState, userStates)
209 def findUserStateByUserId(userId: String): Maybe[UserState] = {
211 val query = new BasicDBObject(UserStateJsonNames.userId, userId)
212 val cursor = userStates find query
216 MongoDBStore.dbObjectToUserState(cursor.next())
225 def findLatestUserStateForEndOfBillingMonth(userId: String,
226 yearOfBillingMonth: Int,
227 billingMonth: Int): Maybe[UserState] = {
228 NoVal // FIXME: implement
231 def deleteUserState(userId: String) = {
232 val query = new BasicDBObject(UserStateJsonNames.userId, userId)
233 userStates.findAndRemove(query)
238 def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] = {
240 MongoDBStore.storeAny[WalletEntry](
243 ResourceJsonNames.id,
245 MongoDBStore.jsonSupportToDBObject)
249 def findWalletEntryById(id: String): Maybe[WalletEntry] =
250 MongoDBStore.findById[WalletEntry](id, walletEntries, MongoDBStore.dbObjectToWalletEntry)
252 def findUserWalletEntries(userId: String) = {
254 findUserWalletEntriesFromTo(userId, new Date(0), new Date(Int.MaxValue))
257 def findUserWalletEntriesFromTo(userId: String, from: Date, to: Date) : List[WalletEntry] = {
258 val q = new BasicDBObject()
259 // TODO: Is this the correct way for an AND query?
260 q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
261 q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$lt", to.getTime))
262 q.put(WalletJsonNames.userId, userId)
264 MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
267 def findWalletEntriesAfter(userId: String, from: Date) : List[WalletEntry] = {
268 val q = new BasicDBObject()
269 q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
270 q.put(WalletJsonNames.userId, userId)
272 MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
275 def findLatestUserWalletEntries(userId: String) = {
277 val orderBy = new BasicDBObject(WalletJsonNames.occurredMillis, -1) // -1 is descending order
278 val cursor = walletEntries.find().sort(orderBy)
281 val buffer = new scala.collection.mutable.ListBuffer[WalletEntry]
283 val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
284 buffer += walletEntry
286 var _previousOccurredMillis = walletEntry.occurredMillis
289 while(cursor.hasNext && _ok) {
290 val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
291 var currentOccurredMillis = walletEntry.occurredMillis
292 _ok = currentOccurredMillis == _previousOccurredMillis
295 buffer += walletEntry
309 def findPreviousEntry(userId: String, resource: String,
311 finalized: Option[Boolean]): List[WalletEntry] = {
312 val q = new BasicDBObject()
313 q.put(WalletJsonNames.userId, userId)
314 q.put(WalletJsonNames.resource, resource)
315 q.put(WalletJsonNames.instanceId, instanceId)
317 case Some(x) => q.put(WalletJsonNames.finalized, x)
321 MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
326 def isLocalIMEvent(event: IMEventModel) = {
327 MongoDBStore.isLocalIMEvent(event)
330 def createIMEventFromJson(json: String) = {
331 MongoDBStore.createIMEventFromJson(json)
334 def createIMEventFromOther(event: IMEventModel) = {
335 MongoDBStore.createIMEventFromOther(event)
338 // def storeUnparsed(json: String): Maybe[RecordID] = {
339 // MongoDBStore.storeJustJson(json, unparsedIMEvents)
342 def insertIMEvent(event: IMEventModel): IMEvent = {
343 val localEvent = MongoDBIMEvent.fromOther(event, new ObjectId())
344 MongoDBStore.insertObject(localEvent, imEvents, MongoDBStore.jsonSupportToDBObject)
348 def findIMEventById(id: String): Maybe[IMEvent] =
349 MongoDBStore.findById[IMEvent](id, imEvents, MongoDBStore.dbObjectToIMEvent)
351 def findIMEventsByUserId(userId: String): List[IMEvent] = {
352 val query = new BasicDBObject(IMEventNames.userID, userId)
353 MongoDBStore.runQuery(query, imEvents)(MongoDBStore.dbObjectToIMEvent)(Some(_sortByTimestampAsc))
358 def loadPolicyEntriesAfter(after: Long): List[PolicyEntry] = {
359 val query = new BasicDBObject(PolicyEntry.JsonNames.validFrom,
360 new BasicDBObject("$gt", after))
361 MongoDBStore.runQuery(query, policyEntries)(MongoDBStore.dbObjectToPolicyEntry)(Some(_sortByTimestampAsc))
364 def storePolicyEntry(policy: PolicyEntry): Maybe[RecordID] = MongoDBStore.storePolicyEntry(policy, policyEntries)
367 def updatePolicyEntry(policy: PolicyEntry) = {
369 val query = new BasicDBObject(PolicyEntry.JsonNames.id, policy.id)
370 val policyObject = MongoDBStore.jsonSupportToDBObject(policy)
371 policyEntries.update(query, policyObject, true, false)
374 def findPolicyEntry(id: String) =
375 MongoDBStore.findById[PolicyEntry](id, policyEntries, MongoDBStore.dbObjectToPolicyEntry)
380 object MongoDBStore {
382 final val _id = "_id"
386 * Collection holding the [[gr.grnet.aquarium.event.ResourceEvent]]s.
388 * Resource events are coming from all systems handling billable resources.
390 final val RESOURCE_EVENTS_COLLECTION = "resevents"
393 * Collection holding the snapshots of [[gr.grnet.aquarium.user.UserState]].
395 * [[gr.grnet.aquarium.user.UserState]] is held internally within [[gr.grnet.aquarium.actor.service.user .UserActor]]s.
397 final val USER_STATES_COLLECTION = "userstates"
400 * Collection holding [[gr.grnet.aquarium.event.im.IMEventModel]]s.
402 * User events are coming from the IM module (external).
404 final val IM_EVENTS_COLLECTION = "imevents"
407 * Collection holding [[gr.grnet.aquarium.event.im.IMEventModel]]s that could not be parsed to normal objects.
409 * We of course assume at least a valid JSON representation.
411 * User events are coming from the IM module (external).
413 final val UNPARSED_IM_EVENTS_COLLECTION = "unparsed_imevents"
416 * Collection holding [[gr.grnet.aquarium.event.WalletEntry]].
418 * Wallet entries are generated internally in Aquarium.
420 final val WALLET_ENTRIES_COLLECTION = "wallets"
423 * Collection holding [[gr.grnet.aquarium.logic.accounting.dsl.DSLPolicy]].
425 // final val POLICIES_COLLECTION = "policies"
428 * Collection holding [[gr.grnet.aquarium.event.PolicyEntry]].
430 final val POLICY_ENTRIES_COLLECTION = "policyEntries"
432 /* TODO: Some of the following methods rely on JSON (de-)serialization).
433 * A method based on proper object serialization would be much faster.
435 def dbObjectToResourceEvent(dbObject: DBObject): ResourceEvent = {
436 ResourceEvent.fromJson(JSON.serialize(dbObject))
439 def dbObjectToUserState(dbObj: DBObject): UserState = {
440 UserState.fromJson(JSON.serialize(dbObj))
443 def dbObjectToWalletEntry(dbObj: DBObject): WalletEntry = {
444 WalletEntry.fromJson(JSON.serialize(dbObj))
447 def dbObjectToIMEvent(dbObj: DBObject): MongoDBIMEvent = {
448 MongoDBIMEvent.fromJson(JSON.serialize(dbObj))
451 def dbObjectToPolicyEntry(dbObj: DBObject): PolicyEntry = {
452 PolicyEntry.fromJson(JSON.serialize(dbObj))
455 def findById[A >: Null <: AnyRef](id: String, collection: DBCollection, deserializer: (DBObject) => A) : Maybe[A] =
457 val query = new BasicDBObject(ResourceJsonNames.id, id)
458 val cursor = collection find query
462 deserializer apply cursor.next
464 null: A // will be transformed to NoVal by the Maybe polymorphic constructor
470 def runQuery[A <: AquariumEventModel](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
471 (deserializer: (DBObject) => A)
472 (sortWith: Option[(A, A) => Boolean]): List[A] = {
473 val cursor0 = collection find query
474 val cursor = if(orderBy ne null) {
478 } // I really know that docs say that it is the same cursor.
480 if(!cursor.hasNext) {
484 val buff = new ListBuffer[A]()
486 while(cursor.hasNext) {
487 buff += deserializer apply cursor.next
493 case Some(sorter) => buff.toList.sortWith(sorter)
494 case None => buff.toList
499 def storeUserState(userState: UserState, collection: DBCollection): Maybe[RecordID] = {
500 Maybe(storeAny[UserState](userState, collection, ResourceJsonNames.userId, _.userId, MongoDBStore.jsonSupportToDBObject))
503 def storePolicyEntry(policyEntry: PolicyEntry, collection: DBCollection): Maybe[RecordID] = {
504 Maybe(storeAny[PolicyEntry](policyEntry, collection, PolicyJsonNames.id, _.id, MongoDBStore.jsonSupportToDBObject))
507 def storeAny[A](any: A,
508 collection: DBCollection,
510 idValueProvider: (A) => String,
511 serializer: (A) => DBObject) : RecordID = {
513 val dbObject = serializer apply any
514 val _id = new ObjectId()
515 dbObject.put("_id", _id)
516 val writeResult = collection.insert(dbObject, WriteConcern.JOURNAL_SAFE)
517 writeResult.getLastError().throwOnError()
519 RecordID(dbObject.get("_id").toString)
522 def insertObject[A <: MongoDBEventModel](obj: A, collection: DBCollection, serializer: (A) => DBObject) : ObjectId = {
523 val dbObject = serializer apply obj
524 val objectId = obj._id match {
526 val _id = new ObjectId()
527 dbObject.put("_id", _id)
534 dbObject.put(JsonNames._id, objectId)
536 val writeResult = collection.insert(dbObject, WriteConcern.JOURNAL_SAFE)
537 writeResult.getLastError().throwOnError()
542 def jsonSupportToDBObject(jsonSupport: JsonSupport) = {
543 Conversions.jsonSupportToDBObject(jsonSupport)
546 final def isLocalIMEvent(event: IMEventModel) = event match {
547 case _: MongoDBIMEvent ⇒ true
551 final def createIMEventFromJson(json: String) = {
552 MongoDBIMEvent.fromJson(json)
555 final def createIMEventFromOther(event: IMEventModel) = {
556 MongoDBIMEvent.fromOther(event)
559 final def createIMEventFromJsonBytes(jsonBytes: Array[Byte]) = {
560 MongoDBIMEvent.fromJsonBytes(jsonBytes)