diff options
Diffstat (limited to 'opendc-web/opendc-web-api/opendc')
| -rw-r--r-- | opendc-web/opendc-web-api/opendc/api/jobs.py | 105 | ||||
| -rw-r--r-- | opendc-web/opendc-web-api/opendc/database.py | 8 | ||||
| -rw-r--r-- | opendc-web/opendc-web-api/opendc/models/scenario.py | 52 |
3 files changed, 134 insertions, 31 deletions
diff --git a/opendc-web/opendc-web-api/opendc/api/jobs.py b/opendc-web/opendc-web-api/opendc/api/jobs.py new file mode 100644 index 00000000..5feaea16 --- /dev/null +++ b/opendc-web/opendc-web-api/opendc/api/jobs.py @@ -0,0 +1,105 @@ +# Copyright (c) 2021 AtLarge Research +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +from flask import request +from flask_restful import Resource +from marshmallow import fields, Schema, validate +from werkzeug.exceptions import BadRequest, Conflict + +from opendc.exts import requires_auth +from opendc.models.scenario import Scenario + + +def convert_to_job(scenario): + """Convert a scenario to a job. + """ + return JobSchema().dump({ + '_id': scenario['_id'], + 'scenarioId': scenario['_id'], + 'state': scenario['simulation']['state'], + 'heartbeat': scenario['simulation'].get('heartbeat', None), + 'results': scenario.get('results', {}) + }) + + +class JobSchema(Schema): + """ + Schema representing a simulation job. + """ + _id = fields.String(dump_only=True) + scenarioId = fields.String(dump_only=True) + state = fields.String(required=True, + validate=validate.OneOf(["QUEUED", "CLAIMED", "RUNNING", "FINISHED", "FAILED"])) + heartbeat = fields.DateTime() + results = fields.Dict() + + +class JobList(Resource): + """ + Resource representing the list of available jobs. + """ + method_decorators = [requires_auth] + + def get(self): + """Get all available jobs.""" + jobs = Scenario.get_jobs() + data = list(map(convert_to_job, jobs.obj)) + return {'data': data} + + +class Job(Resource): + """ + Resource representing a single job. + """ + method_decorators = [requires_auth] + + def get(self, job_id): + """Get the details of a single job.""" + job = Scenario.from_id(job_id) + job.check_exists() + data = convert_to_job(job.obj) + return {'data': data} + + def post(self, job_id): + """Update the details of a single job.""" + action = JobSchema(only=('state', 'results')).load(request.json) + + job = Scenario.from_id(job_id) + job.check_exists() + + old_state = job.obj['simulation']['state'] + new_state = action['state'] + + if old_state == new_state: + data = job.update_state(new_state) + elif (old_state, new_state) == ('QUEUED', 'CLAIMED'): + data = job.update_state('CLAIMED') + elif (old_state, new_state) == ('CLAIMED', 'RUNNING'): + data = job.update_state('RUNNING') + elif (old_state, new_state) == ('RUNNING', 'FINISHED'): + data = job.update_state('FINISHED', results=action.get('results', None)) + elif old_state in ('CLAIMED', 'RUNNING') and new_state == 'FAILED': + data = job.update_state('FAILED') + else: + raise BadRequest('Invalid state transition') + + if not data: + raise Conflict('State conflict') + + return {'data': convert_to_job(data)} diff --git a/opendc-web/opendc-web-api/opendc/database.py b/opendc-web/opendc-web-api/opendc/database.py index 37fd1a4d..dd6367f2 100644 --- a/opendc-web/opendc-web-api/opendc/database.py +++ b/opendc-web/opendc-web-api/opendc/database.py @@ -20,7 +20,7 @@ import urllib.parse -from pymongo import MongoClient +from pymongo import MongoClient, ReturnDocument DATETIME_STRING_FORMAT = '%Y-%m-%dT%H:%M:%S' CONNECTION_POOL = None @@ -76,6 +76,12 @@ class Database: """Updates an existing object.""" return getattr(self.opendc_db, collection).update({'_id': _id}, obj) + def fetch_and_update(self, query, update, collection): + """Updates an existing object.""" + return getattr(self.opendc_db, collection).find_one_and_update(query, + update, + return_document=ReturnDocument.AFTER) + def delete_one(self, query, collection): """Deletes one object matching the given query. diff --git a/opendc-web/opendc-web-api/opendc/models/scenario.py b/opendc-web/opendc-web-api/opendc/models/scenario.py index 658d790e..0fb6c453 100644 --- a/opendc-web/opendc-web-api/opendc/models/scenario.py +++ b/opendc-web/opendc-web-api/opendc/models/scenario.py @@ -1,15 +1,12 @@ +from datetime import datetime + from marshmallow import Schema, fields + +from opendc.exts import db from opendc.models.model import Model from opendc.models.portfolio import Portfolio -class SimulationSchema(Schema): - """ - Simulation details. - """ - state = fields.String() - - class TraceSchema(Schema): """ Schema for specifying the trace of a scenario. @@ -34,27 +31,6 @@ class OperationalSchema(Schema): schedulerName = fields.String() -class ResultSchema(Schema): - """ - Schema representing the simulation results. - """ - max_num_deployed_images = fields.List(fields.Number()) - max_cpu_demand = fields.List(fields.Number()) - max_cpu_usage = fields.List(fields.Number()) - mean_num_deployed_images = fields.List(fields.Number()) - total_failure_slices = fields.List(fields.Number()) - total_failure_vm_slices = fields.List(fields.Number()) - total_granted_burst = fields.List(fields.Number()) - total_interfered_burst = fields.List(fields.Number()) - total_overcommitted_burst = fields.List(fields.Number()) - total_power_draw = fields.List(fields.Number()) - total_requested_burst = fields.List(fields.Number()) - total_vms_failed = fields.List(fields.Number()) - total_vms_finished = fields.List(fields.Number()) - total_vms_queued = fields.List(fields.Number()) - total_vms_submitted = fields.List(fields.Number()) - - class ScenarioSchema(Schema): """ Schema representing a scenario. @@ -62,11 +38,9 @@ class ScenarioSchema(Schema): _id = fields.String(dump_only=True) portfolioId = fields.String() name = fields.String(required=True) - simulation = fields.Nested(SimulationSchema) trace = fields.Nested(TraceSchema) topology = fields.Nested(TopologySchema) operational = fields.Nested(OperationalSchema) - results = fields.Nested(ResultSchema, dump_only=True) class Scenario(Model): @@ -84,3 +58,21 @@ class Scenario(Model): """ portfolio = Portfolio.from_id(self.obj['portfolioId']) portfolio.check_user_access(user_id, edit_access) + + @classmethod + def get_jobs(cls): + """Obtain the scenarios that have been queued. + """ + return cls(db.fetch_all({'simulation.state': 'QUEUED'}, cls.collection_name)) + + def update_state(self, new_state, results=None): + """Atomically update the state of the Scenario. + """ + update = {'$set': {'simulation.state': new_state, 'simulation.heartbeat': datetime.now()}} + if results: + update['$set']['results'] = results + return db.fetch_and_update( + query={'_id': self.obj['_id'], 'simulation.state': self.obj['simulation']['state']}, + update=update, + collection=self.collection_name + ) |