import gr.grnet.aquarium.util.Loggable
import com.mongodb.util.JSON
import gr.grnet.aquarium.user.UserState
+import gr.grnet.aquarium.user.UserState.{JsonNames => UserStateJsonNames}
import gr.grnet.aquarium.util.displayableObjectInfo
import gr.grnet.aquarium.util.json.JsonSupport
import collection.mutable.ListBuffer
import gr.grnet.aquarium.store._
-import gr.grnet.aquarium.logic.events.{WalletEntry, ResourceEvent, AquariumEvent}
-import gr.grnet.aquarium.logic.events.ResourceEvent.JsonNames
+import gr.grnet.aquarium.logic.events.ResourceEvent.{JsonNames => ResourceJsonNames}
+import gr.grnet.aquarium.logic.events.UserEvent.{JsonNames => UserEventJsonNames}
+import gr.grnet.aquarium.logic.events.WalletEntry.{JsonNames => WalletJsonNames}
import java.util.Date
+import com.ckkloverdos.maybe.Maybe
+import gr.grnet.aquarium.logic.accounting.Policy
+import gr.grnet.aquarium.logic.accounting.dsl.{Timeslot, DSLPolicy, DSLComplexResource}
+import gr.grnet.aquarium.logic.events._
import com.mongodb._
-import com.ckkloverdos.maybe.{Failed, Just, Maybe}
/**
* Mongodb implementation of the various aquarium stores.
val database: String,
val username: String,
val password: String)
- extends ResourceEventStore with UserStore with WalletStore with Loggable {
-
- private[store] lazy val rcevents: DBCollection = getCollection(MongoDBStore.RESOURCE_EVENTS_COLLECTION)
- private[store] lazy val users: DBCollection = getCollection(MongoDBStore.USERS_COLLECTION)
- private[store] lazy val imevents: DBCollection = getCollection(MongoDBStore.IM_EVENTS_COLLECTION)
- private[store] lazy val wallets: DBCollection = getCollection(MongoDBStore.IM_WALLETS)
+ extends ResourceEventStore
+ with UserStateStore
+ with WalletEntryStore
+ with UserEventStore
+ with PolicyStore
+ with Loggable {
+
+ private[store] lazy val resourceEvents = getCollection(MongoDBStore.RESOURCE_EVENTS_COLLECTION)
+ private[store] lazy val userStates = getCollection(MongoDBStore.USER_STATES_COLLECTION)
+ private[store] lazy val userEvents = getCollection(MongoDBStore.USER_EVENTS_COLLECTION)
+ private[store] lazy val walletEntries = getCollection(MongoDBStore.WALLET_ENTRIES_COLLECTION)
+ private[store] lazy val policies = getCollection(MongoDBStore.POLICIES_COLLECTION)
private[this] def getCollection(name: String): DBCollection = {
val db = mongo.getDB(database)
- if(!db.authenticate(username, password.toCharArray)) {
+ //logger.debug("Authenticating to mongo")
+ if(!db.isAuthenticated && !db.authenticate(username, password.toCharArray)) {
throw new StoreException("Could not authenticate user %s".format(username))
}
db.getCollection(name)
}
- /* TODO: Some of the following methods rely on JSON (de-)serialization).
- * A method based on proper object serialization would be much faster.
- */
-
- private[this] def _makeDBObject(any: JsonSupport): DBObject = {
- JSON.parse(any.toJson) match {
- case dbObject: DBObject ⇒
- dbObject
- case _ ⇒
- throw new StoreException("Could not transform %s -> %s".format(displayableObjectInfo(any), classOf[DBObject].getName))
- }
- }
-
- private[this] def _insertObject(collection: DBCollection, obj: JsonSupport): DBObject = {
- val dbObj = _makeDBObject(obj)
- collection insert dbObj
- dbObj
- }
-
- private[this] def _checkWasInserted(collection: DBCollection, obj: JsonSupport, idName: String, id: String): String = {
- val cursor = collection.find(new BasicDBObject(idName, id))
- if (!cursor.hasNext) {
- val errMsg = "Failed to _store %s".format(displayableObjectInfo(obj))
- logger.error(errMsg)
- throw new StoreException(errMsg)
- }
-
- val retval = cursor.next.get("_id").toString
- cursor.close()
- retval
- }
-
- private[this] def _store[A <: AquariumEvent](entry: A, col: DBCollection) : Maybe[RecordID] = {
- try {
- // Store
- val dbObj = _makeDBObject(entry)
- col.insert(dbObj)
-
- // Get back to retrieve unique id
- val cursor = col.find(new BasicDBObject(JsonNames.id, entry.id))
-
- if (!cursor.hasNext) {
- cursor.close()
- logger.error("Failed to _store entry: %s".format(entry))
- return Failed(new StoreException("Failed to _store entry: %s".format(entry)))
- }
-
- val retval = Just(RecordID(cursor.next.get(JsonNames._id).toString))
- cursor.close()
- retval
- } catch {
- case m: MongoException =>
- logger.error("Unknown Mongo error: %s".format(m)); Failed(m)
- }
- }
-
private[this] def _sortByTimestampAsc[A <: AquariumEvent](one: A, two: A): Boolean = {
if (one.occurredMillis > two.occurredMillis) false
else if (one.occurredMillis < two.occurredMillis) true
}
//+ResourceEventStore
- def storeResourceEvent(event: ResourceEvent): Maybe[RecordID] = _store(event, rcevents)
+ def storeResourceEvent(event: ResourceEvent): Maybe[RecordID] =
+ MongoDBStore.storeAquariumEvent(event, resourceEvents)
- def findResourceEventById(id: String): Maybe[ResourceEvent] = MongoDBStore.findById(id, rcevents, MongoDBStore.dbObjectToResourceEvent)
+ def findResourceEventById(id: String): Maybe[ResourceEvent] =
+ MongoDBStore.findById(id, resourceEvents, MongoDBStore.dbObjectToResourceEvent)
def findResourceEventsByUserId(userId: String)
(sortWith: Option[(ResourceEvent, ResourceEvent) => Boolean]): List[ResourceEvent] = {
- val query = new BasicDBObject(JsonNames.userId, userId)
+ val query = new BasicDBObject(ResourceJsonNames.userId, userId)
- MongoDBStore.runQuery(query, rcevents)(MongoDBStore.dbObjectToResourceEvent)(sortWith)
+ MongoDBStore.runQuery(query, resourceEvents)(MongoDBStore.dbObjectToResourceEvent)(sortWith)
}
def findResourceEventsByUserIdAfterTimestamp(userId: String, timestamp: Long): List[ResourceEvent] = {
val query = new BasicDBObject()
- query.put(JsonNames.userId, userId)
- query.put(JsonNames.timestamp, new BasicDBObject("$gte", timestamp))
+ query.put(ResourceJsonNames.userId, userId)
+ query.put(ResourceJsonNames.occurredMillis, new BasicDBObject("$gt", timestamp))
- val sort = new BasicDBObject(JsonNames.timestamp, 1)
+ val sort = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
- val cursor = rcevents.find(query).sort(sort)
+ val cursor = resourceEvents.find(query).sort(sort)
try {
val buffer = new scala.collection.mutable.ListBuffer[ResourceEvent]
while(cursor.hasNext) {
buffer += MongoDBStore.dbObjectToResourceEvent(cursor.next())
}
- buffer.toList
+ buffer.toList.sortWith(_sortByTimestampAsc)
+ } finally {
+ cursor.close()
+ }
+ }
+
+ def findResourceEventHistory(userId: String, resName: String,
+ instid: Option[String], upTo: Long) : List[ResourceEvent] = {
+ val query = new BasicDBObject()
+ query.put(ResourceJsonNames.userId, userId)
+ query.put(ResourceJsonNames.occurredMillis, new BasicDBObject("$lt", upTo))
+ query.put(ResourceJsonNames.resource, resName)
+
+ instid match {
+ case Some(id) =>
+ Policy.policy.findResource(resName) match {
+ case Some(y) => query.put(ResourceJsonNames.details,
+ new BasicDBObject(y.asInstanceOf[DSLComplexResource].descriminatorField, instid.get))
+ case None =>
+ }
+ case None =>
+ }
+
+ val sort = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
+ val cursor = resourceEvents.find(query).sort(sort)
+
+ try {
+ val buffer = new scala.collection.mutable.ListBuffer[ResourceEvent]
+ while(cursor.hasNext) {
+ buffer += MongoDBStore.dbObjectToResourceEvent(cursor.next())
+ }
+ buffer.toList.sortWith(_sortByTimestampAsc)
} finally {
cursor.close()
}
}
- //-ResourceEventStore
- //+UserStore
- def storeUserState(userState: UserState): Maybe[RecordID] = {
+ def findResourceEventsForReceivedPeriod(userId: String, startTimeMillis: Long, stopTimeMillis: Long): List[ResourceEvent] = {
+ val query = new BasicDBObject()
+ query.put(ResourceJsonNames.userId, userId)
+ query.put(ResourceJsonNames.receivedMillis, new BasicDBObject("$gte", startTimeMillis))
+ query.put(ResourceJsonNames.receivedMillis, new BasicDBObject("$lte", stopTimeMillis))
+
+ // Sort them by increasing order for occurred time
+ val orderBy = new BasicDBObject(ResourceJsonNames.occurredMillis, 1)
+
+ MongoDBStore.runQuery[ResourceEvent](query, resourceEvents, orderBy)(MongoDBStore.dbObjectToResourceEvent)(None)
+ }
+
+ def countOutOfSyncEventsForBillingMonth(userId: String, yearOfBillingMonth: Int, billingMonth: Int): Maybe[Long] = {
Maybe {
- val dbObj = _insertObject(users, userState)
- val id = _checkWasInserted(users, userState, JsonNames.userId, userState.userId)
- RecordID(id)
+ // FIXME: Implement
+ 0L
}
}
+ //-ResourceEventStore
+
+ //+UserStateStore
+ def storeUserState(userState: UserState): Maybe[RecordID] =
+ MongoDBStore.storeUserState(userState, userStates)
+
def findUserStateByUserId(userId: String): Maybe[UserState] = {
Maybe {
- val query = new BasicDBObject(JsonNames.userId, userId)
- val cursor = rcevents find query
+ val query = new BasicDBObject(UserStateJsonNames.userId, userId)
+ val cursor = userStates find query
try {
if(cursor.hasNext)
}
}
}
- //-UserStore
- //+WalletStore
- def store(entry: WalletEntry): Maybe[RecordID] = _store(entry, wallets)
+ def deleteUserState(userId: String) = {
+ val query = new BasicDBObject(UserStateJsonNames.userId, userId)
+ userStates.findAndRemove(query)
+ }
+ //-UserStateStore
- def findEntryById(id: String): Maybe[WalletEntry] = MongoDBStore.findById[WalletEntry](id, wallets, MongoDBStore.dbObjectToWalletEntry)
+ //+WalletEntryStore
+ def storeWalletEntry(entry: WalletEntry): Maybe[RecordID] =
+ MongoDBStore.storeAquariumEvent(entry, walletEntries)
- def findAllUserEntries(userId: String) = findUserEntriesFromTo(userId, new Date(0), new Date(Int.MaxValue))
+ def findWalletEntryById(id: String): Maybe[WalletEntry] =
+ MongoDBStore.findById[WalletEntry](id, walletEntries, MongoDBStore.dbObjectToWalletEntry)
- def findUserEntriesFromTo(userId: String, from: Date, to: Date) : List[WalletEntry] = {
+ def findUserWalletEntries(userId: String) = {
+ // TODO: optimize
+ findUserWalletEntriesFromTo(userId, new Date(0), new Date(Int.MaxValue))
+ }
+
+ def findUserWalletEntriesFromTo(userId: String, from: Date, to: Date) : List[WalletEntry] = {
val q = new BasicDBObject()
// TODO: Is this the correct way for an AND query?
- q.put(JsonNames.timestamp, new BasicDBObject("$gt", from.getTime))
- q.put(JsonNames.timestamp, new BasicDBObject("$lt", to.getTime))
- q.put(JsonNames.userId, userId)
+ q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
+ q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$lt", to.getTime))
+ q.put(WalletJsonNames.userId, userId)
+
+ MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
+ }
+
+ def findWalletEntriesAfter(userId: String, from: Date) : List[WalletEntry] = {
+ val q = new BasicDBObject()
+ q.put(WalletJsonNames.occurredMillis, new BasicDBObject("$gt", from.getTime))
+ q.put(WalletJsonNames.userId, userId)
+
+ MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
+ }
+
+ def findLatestUserWalletEntries(userId: String) = {
+ Maybe {
+ val orderBy = new BasicDBObject(WalletJsonNames.occurredMillis, -1) // -1 is descending order
+ val cursor = walletEntries.find().sort(orderBy)
+
+ try {
+ val buffer = new scala.collection.mutable.ListBuffer[WalletEntry]
+ if(cursor.hasNext) {
+ val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
+ buffer += walletEntry
+
+ var _previousOccurredMillis = walletEntry.occurredMillis
+ var _ok = true
+
+ while(cursor.hasNext && _ok) {
+ val walletEntry = MongoDBStore.dbObjectToWalletEntry(cursor.next())
+ var currentOccurredMillis = walletEntry.occurredMillis
+ _ok = currentOccurredMillis == _previousOccurredMillis
+
+ if(_ok) {
+ buffer += walletEntry
+ }
+ }
+
+ buffer.toList
+ } else {
+ null
+ }
+ } finally {
+ cursor.close()
+ }
+ }
+ }
+
+ def findPreviousEntry(userId: String, resource: String,
+ instanceId: String,
+ finalized: Option[Boolean]): List[WalletEntry] = {
+ val q = new BasicDBObject()
+ q.put(WalletJsonNames.userId, userId)
+ q.put(WalletJsonNames.resource, resource)
+ q.put(WalletJsonNames.instanceId, instanceId)
+ finalized match {
+ case Some(x) => q.put(WalletJsonNames.finalized, x)
+ case None =>
+ }
+
+ MongoDBStore.runQuery[WalletEntry](q, walletEntries)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
+ }
+ //-WalletEntryStore
+
+ //+UserEventStore
+ def storeUserEvent(event: UserEvent): Maybe[RecordID] =
+ MongoDBStore.storeAny[UserEvent](event, userEvents, UserEventJsonNames.userId,
+ _.userId, MongoDBStore.jsonSupportToDBObject)
+
+
+ def findUserEventById(id: String): Maybe[UserEvent] =
+ MongoDBStore.findById[UserEvent](id, userEvents, MongoDBStore.dbObjectToUserEvent)
- MongoDBStore.runQuery[WalletEntry](q, wallets)(MongoDBStore.dbObjectToWalletEntry)(Some(_sortByTimestampAsc))
+ def findUserEventsByUserId(userId: String): List[UserEvent] = {
+ val query = new BasicDBObject(UserEventJsonNames.userId, userId)
+ MongoDBStore.runQuery(query, userEvents)(MongoDBStore.dbObjectToUserEvent)(Some(_sortByTimestampAsc))
}
- //-WalletStore
+ //-UserEventStore
+
+ //+PolicyStore
+ def loadPolicies(after: Long): List[PolicyEntry] = {
+ val query = new BasicDBObject(PolicyEntry.JsonNames.validFrom,
+ new BasicDBObject("$gt", after))
+ MongoDBStore.runQuery(query, policies)(MongoDBStore.dbObjectToPolicyEvent)(Some(_sortByTimestampAsc))
+ }
+
+ def storePolicy(policy: PolicyEntry): Maybe[RecordID] = MongoDBStore.storeAquariumEvent(policy, policies)
+
+
+ def updatePolicy(policy: PolicyEntry) = {
+ //Find the entry
+ val query = new BasicDBObject(PolicyEntry.JsonNames.id, policy.id)
+ val policyObject = MongoDBStore.jsonSupportToDBObject(policy)
+ policies.update(query, policyObject, true, false)
+ }
+ //-PolicyStore
}
object MongoDBStore {
- def RESOURCE_EVENTS_COLLECTION = "rcevents"
- def USERS_COLLECTION = "users"
- def IM_EVENTS_COLLECTION = "imevents"
- def IM_WALLETS = "wallets"
+ object JsonNames {
+ final val _id = "_id"
+ }
+
+ /**
+ * Collection holding the [[gr.grnet.aquarium.logic.events.ResourceEvent]]s.
+ *
+ * Resource events are coming from all systems handling billable resources.
+ */
+ final val RESOURCE_EVENTS_COLLECTION = "resevents"
+
+ /**
+ * Collection holding the snapshots of [[gr.grnet.aquarium.user.UserState]].
+ *
+ * [[gr.grnet.aquarium.user.UserState]] is held internally within [[gr.grnet.aquarium.user.actor.UserActor]]s.
+ */
+ final val USER_STATES_COLLECTION = "userstates"
+
+ /**
+ * Collection holding [[gr.grnet.aquarium.logic.events.UserEvent]]s.
+ *
+ * User events are coming from the IM module (external).
+ */
+ final val USER_EVENTS_COLLECTION = "userevents"
+
+ /**
+ * Collection holding [[gr.grnet.aquarium.logic.events.WalletEntry]].
+ *
+ * Wallet entries are generated internally in Aquarium.
+ */
+ final val WALLET_ENTRIES_COLLECTION = "wallets"
+
+ /**
+ * Collection holding [[gr.grnet.aquarium.logic.accounting.dsl.DSLPolicy]].
+ */
+ final val POLICIES_COLLECTION = "policies"
+ /* TODO: Some of the following methods rely on JSON (de-)serialization).
+ * A method based on proper object serialization would be much faster.
+ */
def dbObjectToResourceEvent(dbObject: DBObject): ResourceEvent = {
ResourceEvent.fromJson(JSON.serialize(dbObject))
}
WalletEntry.fromJson(JSON.serialize(dbObj))
}
+ def dbObjectToUserEvent(dbObj: DBObject): UserEvent = {
+ UserEvent.fromJson(JSON.serialize(dbObj))
+ }
+
+ def dbObjectToPolicyEvent(dbObj: DBObject): PolicyEntry = {
+ PolicyEntry.fromJson(JSON.serialize(dbObj))
+ }
+
def findById[A >: Null <: AquariumEvent](id: String, collection: DBCollection, deserializer: (DBObject) => A) : Maybe[A] = Maybe {
- val query = new BasicDBObject(JsonNames.id, id)
+ val query = new BasicDBObject(ResourceJsonNames.id, id)
val cursor = collection find query
try {
}
}
- def runQuery[A <: AquariumEvent](query: BasicDBObject, collection: DBCollection)
+ def runQuery[A <: AquariumEvent](query: DBObject, collection: DBCollection, orderBy: DBObject = null)
(deserializer: (DBObject) => A)
(sortWith: Option[(A, A) => Boolean]): List[A] = {
- val cur = collection find query
- if(!cur.hasNext) {
- cur.close()
+ val cursor0 = collection find query
+ val cursor = if(orderBy ne null) {
+ cursor0 sort orderBy
+ } else {
+ cursor0
+ } // I really know that docs say that it is the same cursor.
+
+ if(!cursor.hasNext) {
+ cursor.close()
Nil
} else {
val buff = new ListBuffer[A]()
- while(cur.hasNext) {
- buff += deserializer apply cur.next
+ while(cursor.hasNext) {
+ buff += deserializer apply cursor.next
}
- cur.close()
+ cursor.close()
sortWith match {
case Some(sorter) => buff.toList.sortWith(sorter)
}
}
}
+
+ def storeAquariumEvent[A <: AquariumEvent](event: A, collection: DBCollection) : Maybe[RecordID] = {
+ storeAny[A](event, collection, ResourceJsonNames.id, (e) => e.id, MongoDBStore.jsonSupportToDBObject)
+ }
+
+ def storeUserState(userState: UserState, collection: DBCollection): Maybe[RecordID] = {
+ storeAny[UserState](userState, collection, ResourceJsonNames.userId, _.userId, MongoDBStore.jsonSupportToDBObject)
+ }
+
+ def storeAny[A](any: A,
+ collection: DBCollection,
+ idName: String,
+ idValueProvider: (A) => String,
+ serializer: (A) => DBObject) : Maybe[RecordID] = {
+ import com.ckkloverdos.maybe.effect
+
+ Maybe {
+ val dbObj = serializer apply any
+ val writeResult = collection insert dbObj
+ writeResult.getLastError().throwOnError()
+
+ // Get back to retrieve unique id
+ val cursor = collection.find(new BasicDBObject(idName, idValueProvider(any)))
+ cursor
+ } flatMap { cursor ⇒
+ effect {
+ if(cursor.hasNext)
+ RecordID(cursor.next().get(JsonNames._id).toString)
+ else
+ throw new StoreException("Could not store %s to %s".format(any, collection))
+ } {} { cursor.close() }
+ }
+ }
+
+ def jsonSupportToDBObject(any: JsonSupport): DBObject = {
+ JSON.parse(any.toJson) match {
+ case dbObject: DBObject ⇒
+ dbObject
+ case _ ⇒
+ throw new StoreException("Could not transform %s -> %s".format(displayableObjectInfo(any), classOf[DBObject].getName))
+ }
+ }
}
projects / aquarium / blobdiff