Milimetric has uploaded a new change for review.
https://gerrit.wikimedia.org/r/79429
Change subject: re-organizing the return value of report nodes
......................................................................
re-organizing the return value of report nodes
Change-Id: Iea50bfc69841e9efe530fb1d92be27ffa8d20690
---
M tests/fixtures.py
M tests/test_models/test_report_response.py
M wikimetrics/controllers/reports.py
M wikimetrics/models/persistent_report.py
M wikimetrics/models/report_nodes/aggregate_report.py
D wikimetrics/models/report_nodes/concat_metrics_report.py
M wikimetrics/models/report_nodes/multi_project_metric_report.py
M wikimetrics/models/report_nodes/report.py
M wikimetrics/models/report_nodes/run_report.py
M wikimetrics/run.py
10 files changed, 192 insertions(+), 173 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/analytics/wikimetrics
refs/changes/29/79429/1
diff --git a/tests/fixtures.py b/tests/fixtures.py
index 4f3da64..b24a02e 100644
--- a/tests/fixtures.py
+++ b/tests/fixtures.py
@@ -252,25 +252,25 @@
report_created = PersistentReport(
user_id=web_test_user.id,
status=celery.states.PENDING,
- result_key=None,
+ queue_result_key=None,
show_in_ui=True
)
report_started = PersistentReport(
user_id=web_test_user.id,
status=celery.states.STARTED,
- result_key=None,
+ queue_result_key=None,
show_in_ui=True
)
report_started2 = PersistentReport(
user_id=web_test_user.id,
status=celery.states.STARTED,
- result_key=None,
+ queue_result_key=None,
show_in_ui=True
)
report_finished = PersistentReport(
user_id=web_test_user.id,
status=celery.states.SUCCESS,
- result_key=None,
+ queue_result_key=None,
show_in_ui=True
)
self.session.add_all([
diff --git a/tests/test_models/test_report_response.py
b/tests/test_models/test_report_response.py
index 254ba56..f356b62 100644
--- a/tests/test_models/test_report_response.py
+++ b/tests/test_models/test_report_response.py
@@ -1,7 +1,10 @@
-from wikimetrics.models import Cohort, RunReport, MultiProjectMetricReport,
Aggregation
+from nose.tools import assert_equals, assert_true
+from celery.exceptions import SoftTimeLimitExceeded
+from wikimetrics.models import (
+ Cohort, RunReport, MultiProjectMetricReport, Aggregation, PersistentReport
+)
from wikimetrics.metrics import NamespaceEdits
from ..fixtures import QueueDatabaseTest
-from nose.tools import assert_equals, assert_true
class RunReportTest(QueueDatabaseTest):
@@ -20,7 +23,7 @@
},
'metric': {
'name': 'NamespaceEdits',
- 'namespaces': [0,1,2],
+ 'namespaces': [0, 1, 2],
'start_date': '2013-06-01',
'end_date': '2013-09-01',
'individualResults': True,
@@ -31,7 +34,12 @@
},
}]
jr = RunReport(desired_responses, user_id=self.test_user_id)
- results = jr.task.delay(jr).get()[0]
+ results = jr.task.delay(jr).get()
+ result_key = self.session.query(PersistentReport)\
+ .filter(PersistentReport.id == jr.children[0].persistent_id)\
+ .one()\
+ .result_key
+ results = results[result_key]
# TODO: figure out why one of the resulting wiki_user_ids is None here
assert_equals(
results[Aggregation.IND][0][self.test_mediawiki_user_id]['edits'],
@@ -46,7 +54,7 @@
},
'metric': {
'name': 'NamespaceEdits',
- 'namespaces': [0,1,2],
+ 'namespaces': [0, 1, 2],
'start_date': '2013-06-01',
'end_date': '2013-09-01',
'individualResults': True,
@@ -57,7 +65,12 @@
},
}]
jr = RunReport(desired_responses, user_id=self.test_user_id)
- results = jr.task.delay(jr).get()[0]
+ results = jr.task.delay(jr).get()
+ result_key = self.session.query(PersistentReport)\
+ .filter(PersistentReport.id == jr.children[0].persistent_id)\
+ .one()\
+ .result_key
+ results = results[result_key]
assert_equals(
results[Aggregation.IND][0][self.test_mediawiki_user_id]['edits'],
2,
@@ -76,7 +89,7 @@
},
'metric': {
'name': 'BytesAdded',
- 'namespaces': [0,1,2],
+ 'namespaces': [0, 1, 2],
'start_date': '2013-06-01',
'end_date': '2013-09-01',
'individualResults': True,
@@ -87,8 +100,12 @@
},
}]
jr = RunReport(desired_responses, user_id=self.test_user_id)
- results = jr.task.delay(jr).get()[0]
- print(results)
+ results = jr.task.delay(jr).get()
+ result_key = self.session.query(PersistentReport)\
+ .filter(PersistentReport.id == jr.children[0].persistent_id)\
+ .one()\
+ .result_key
+ results = results[result_key]
assert_equals(
results[Aggregation.IND][0][self.test_mediawiki_user_id]['net_sum'],
6,
@@ -101,39 +118,48 @@
# TODO: figure out how to write this test properly,
# basically: how to make sure that the queue can be hamerred with requests
- #def test_lots_of_concurrent_requests(self):
- #desired_responses = [{
- #'name': 'Edits - test',
- #'cohort': {
- #'id': self.test_cohort_id,
- #},
- #'metric': {
- #'name': 'BytesAdded',
- #'namespaces': '0,1,2',
- #'start_date': '2013-06-01',
- #'end_date': '2013-09-01',
- #'individualResults': True,
- #'aggregateResults': True,
- #'aggregateSum': True,
- #'aggregateAverage': False,
- #'aggregateStandardDeviation': False,
- #},
- #}]
- #reports = []
- ## NOTE: you can make this loop as much as you'd like if celery
- ## is allowed enough concurrent workers, set via CELERYD_CONCURRENCY
- #for i in range(6):
- #jr = RunReport(desired_responses, user_id=self.test_user_id)
- #reports.append(jr.task.delay(jr))
+ def test_lots_of_concurrent_requests(self):
+ desired_responses = [{
+ 'name': 'Edits - test',
+ 'cohort': {
+ 'id': self.test_cohort_id,
+ },
+ 'metric': {
+ 'name': 'BytesAdded',
+ 'namespaces': '0,1,2',
+ 'start_date': '2013-06-01',
+ 'end_date': '2013-09-01',
+ 'individualResults': True,
+ 'aggregateResults': True,
+ 'aggregateSum': True,
+ 'aggregateAverage': False,
+ 'aggregateStandardDeviation': False,
+ },
+ }]
+ reports = []
+ # NOTE: you can make this loop as much as you'd like if celery
+ # is allowed enough concurrent workers, set via CELERYD_CONCURRENCY
+ trials = 3
+ for i in range(trials):
+ jr = RunReport(desired_responses, user_id=self.test_user_id)
+ reports.append((jr, jr.task.delay(jr)))
- #successes = 0
- #for report in reports:
- #try:
- #results = report.get()[0]
- #if results[Aggregation.SUM]['positive_only_sum'] == 50:
- #successes += 1
- #except:
- #print('timeout expired for this task')
+ successes = 0
+ for jr, delayed in reports:
+ try:
+ results = delayed.get()
+ result_key = self.session.query(PersistentReport)\
+ .filter(PersistentReport.id ==
jr.children[0].persistent_id)\
+ .one()\
+ .result_key
+ results = results[result_key]
+ if results[Aggregation.SUM]['positive_only_sum'] == 150:
+ successes += 1
+ except SoftTimeLimitExceeded:
+ print('Timeout expired during this task.')
+ except Exception, e:
+ print('An exception occurred during this task.')
+ raise
- #print('Successes: {0}'.format(successes))
- #assert_true(successes > 3, 'at least half of the tasks succeeded')
+ print('Successes: {0}'.format(successes))
+ assert_true(successes == trials, 'all of the trials must succeed')
diff --git a/wikimetrics/controllers/reports.py
b/wikimetrics/controllers/reports.py
index ccce0a0..7f01ee3 100644
--- a/wikimetrics/controllers/reports.py
+++ b/wikimetrics/controllers/reports.py
@@ -62,20 +62,40 @@
return reports_json
[email protected]('/reports/status/<task_id>')
-def report_status(task_id):
- celery_task = Report.task.AsyncResult(task_id)
+def get_celery_task(result_key):
+ """
+ From a unique identifier, gets the celery task and database records
associated.
+
+ Parameters
+ result_key : The unique identifier found in the report database table
+
+ Returns
+ A tuple of the form (celery_task_object, database_report_object)
+ """
+ db_session = db.get_session()
+ pj = db_session.query(PersistentReport)\
+ .filter(PersistentReport.result_key == result_key)\
+ .one()
+
+ celery_task = Report.task.AsyncResult(pj.queue_result_key)
+ db_session.close()
+ return (celery_task, pj)
+
+
[email protected]('/reports/status/<result_key>')
+def report_status(result_key):
+ celery_task, pj = get_celery_task(result_key)
return json_response(status=celery_task.status)
[email protected]('/reports/result/<task_id>.csv')
-def report_result_csv(task_id):
- celery_task = Report.task.AsyncResult(task_id)
[email protected]('/reports/result/<result_key>.csv')
+def report_result_csv(result_key):
+ celery_task, pj = get_celery_task(result_key)
if not celery_task:
- return json_error('no task exists with id: {0}'.format(task_id))
+ return json_error('no task exists with id: {0}'.format(result_key))
if celery_task.ready():
- task_result = celery_task.get()
+ task_result = celery_task.get()[result_key]
csv_io = StringIO()
if task_result:
@@ -130,38 +150,30 @@
return json_response(status=celery_task.status)
[email protected]('/reports/result/<task_id>.json')
-def report_result_json(task_id):
- celery_task = Report.task.AsyncResult(task_id)
[email protected]('/reports/result/<result_key>.json')
+def report_result_json(result_key):
+ celery_task, pj = get_celery_task(result_key)
if not celery_task:
- return json_error('no task exists with id: {0}'.format(task_id))
+ return json_error('no task exists with id: {0}'.format(result_key))
if celery_task.ready():
- task_result = celery_task.get()
-
- # get the parameters from the database
- db_session = db.get_session()
- report = db_session.query(PersistentReport)\
- .filter(PersistentReport.result_key == task_id)\
- .one()
- parameters = report.parameters
- db_session.close()
+ task_result = celery_task.get()[result_key]
return json_response(
result=task_result,
- parameters=json.loads(parameters),
+ parameters=json.loads(pj.parameters),
)
else:
return json_response(status=celery_task.status)
[email protected]('/reports/kill/<task_id>')
-def report_kill(task_id):
[email protected]('/reports/kill/<result_key>')
+def report_kill(result_key):
return 'not implemented'
#db_session = db.get_session()
- #db_report = db_session.query(PersistentReport).get(task_id)
+ #db_report = db_session.query(PersistentReport).get(result_key)
#if not db_report:
- #return json_error('no task exists with id: {0}'.format(task_id))
+ #return json_error('no task exists with id: {0}'.format(result_key))
#celery_task = Report.task.AsyncResult(db_report.result_key)
#app.logger.debug('revoking task: %s', celery_task.id)
#celery_task.revoke()
diff --git a/wikimetrics/models/persistent_report.py
b/wikimetrics/models/persistent_report.py
index f08fa79..c2b86d0 100644
--- a/wikimetrics/models/persistent_report.py
+++ b/wikimetrics/models/persistent_report.py
@@ -18,6 +18,7 @@
id = Column(Integer, primary_key=True)
created = Column(DateTime, default=func.now())
user_id = Column(Integer)
+ queue_result_key = Column(String(50))
result_key = Column(String(50))
status = Column(String(50))
name = Column(String(2000))
@@ -25,16 +26,14 @@
parameters = Column(String(4000))
def update_status(self):
- # if we don't have the result key leave as is (PENDING)
- if self.result_key and self.status not in (celery.states.READY_STATES):
+ # if we don't have the result key leave as is
+ if self.queue_result_key and self.status not in
(celery.states.READY_STATES):
# TODO: inline import. Can't import up above because of circular
reference
from wikimetrics.models.report_nodes import Report
- celery_task = Report.task.AsyncResult(self.result_key)
+ celery_task = Report.task.AsyncResult(self.queue_result_key)
self.status = celery_task.status
existing_session = Session.object_session(self)
if not existing_session:
existing_session = db.get_session()
existing_session.add(self)
existing_session.commit()
- # if the result is still an AsyncResult, leave it as PENDING
- #if isinstance(celery_task.result, AsyncResult):
diff --git a/wikimetrics/models/report_nodes/aggregate_report.py
b/wikimetrics/models/report_nodes/aggregate_report.py
index 990252a..2a5586c 100644
--- a/wikimetrics/models/report_nodes/aggregate_report.py
+++ b/wikimetrics/models/report_nodes/aggregate_report.py
@@ -62,30 +62,33 @@
name=self.name,
)]
- def finish(self, multi_project_results):
+ def finish(self, result_dicts):
aggregated_results = dict()
+ result_values = [r.values() for r in result_dicts]
+ child_results = [result for sublist in result_values for result in
sublist]
if self.aggregate:
if self.aggregate_sum:
aggregated_results[Aggregation.SUM] = self.calculate(
- multi_project_results,
+ child_results,
Aggregation.SUM
)
if self.aggregate_average:
aggregated_results[Aggregation.AVG] = self.calculate(
- multi_project_results,
+ child_results,
Aggregation.AVG
)
if self.aggregate_std_deviation:
aggregated_results[Aggregation.STD] = self.calculate(
- multi_project_results,
+ child_results,
Aggregation.STD
)
if self.individual:
- aggregated_results[Aggregation.IND] = multi_project_results
+ aggregated_results[Aggregation.IND] = child_results
- return aggregated_results
+ result = self.report_result(aggregated_results,
child_results=result_dicts)
+ return result
def calculate(self, list_of_results, type_of_aggregate):
# TODO: terrible redo this
diff --git a/wikimetrics/models/report_nodes/concat_metrics_report.py
b/wikimetrics/models/report_nodes/concat_metrics_report.py
deleted file mode 100644
index 01fe064..0000000
--- a/wikimetrics/models/report_nodes/concat_metrics_report.py
+++ /dev/null
@@ -1,31 +0,0 @@
-import report
-import pprint
-from wikimetrics.configurables import queue
-from metric_report import MetricReport
-
-
-__all__ = ['ConcatMetricsReport']
-
-
-class ConcatMetricsReport(report.ReportNode):
- """
- Report which runs several metrics on the same cohort and then
- joins together the results from each metric into a suitable
- 2-D representation.
- """
-
- def __init__(self, cohort, metrics):
- super(ConcatMetricsReport, self).__init__()
- self.cohort = cohort
- self.metrics = metrics
- # TODO enforce children always have a run @queue.task
- self.children = [MetricReport(cohort, metric) for metric in metrics]
- # TODO self.save()
-
- def finish(query_results):
- # we're done - record result
- for result in query_results:
- pprint.pprint(result)
-
- def __repr__(self):
- return '<ConcatMetricsReport("{0}")>'.format(self.persistent_id)
diff --git a/wikimetrics/models/report_nodes/multi_project_metric_report.py
b/wikimetrics/models/report_nodes/multi_project_metric_report.py
index e0fdac0..e575a87 100644
--- a/wikimetrics/models/report_nodes/multi_project_metric_report.py
+++ b/wikimetrics/models/report_nodes/multi_project_metric_report.py
@@ -32,15 +32,14 @@
# note that user_ids is actually just an iterator
self.children.append(MetricReport(metric, user_ids, project))
- def finish(self, query_results):
- merged = {}
- for res in query_results:
- try:
- merged.update(res)
- except:
- task_logger.error('updating failed: %s', res)
- raise
- return merged
+ def finish(self, metric_results):
+ merged_individual_results = {}
+ # TODO: handle collisions where the same ID is used accross projects
+ for res in metric_results:
+ merged_individual_results.update(res)
+
+ result = self.report_result(merged_individual_results)
+ return result
def __repr__(self):
return '<MultiProjectMetricReport("{0}")>'.format(self.persistent_id)
diff --git a/wikimetrics/models/report_nodes/report.py
b/wikimetrics/models/report_nodes/report.py
index 96a673a..1ff1f40 100644
--- a/wikimetrics/models/report_nodes/report.py
+++ b/wikimetrics/models/report_nodes/report.py
@@ -1,5 +1,6 @@
import time
import celery
+from uuid import uuid4
from celery import group, chord, current_task
from celery.result import AsyncResult
from celery.exceptions import SoftTimeLimitExceeded
@@ -38,39 +39,12 @@
@queue.task()
def queue_task(report):
- if not isinstance(report, ReportNode):
- report.set_status(celery.states.STARTED,
task_id=current_task.request.id)
task_logger.info('running {0} on celery as {1}'.format(
report,
current_task.request.id,
))
- result = report.run()
- return result
-
-
[email protected]()
-def queue_finish_task(results, report):
- """
- This is the task which is executed after all of the child tasks
- in a ReportNode have been executed. It serves as a wrapper to the
- finish() method which actually deals with child task results.
- Note that the signature of this method is a little funny due to
- a hack to get around the way that celery handles instance method tasks.
- The ReportNode instance (report) is specified when the callback
- subtask is created, and the results argument is filled in by celery
- once they have completed. The order is just reversed because celery
- is hardcoded to prepend the results from a chord into the argument list
- specified when creating the subtask.
- """
- report.set_status(celery.states.STARTED, task_id=current_task.request.id)
-
- task_logger.info('finishing {0} on celery as {1}'.format(
- report,
- current_task.request.id,
- ))
- result = report.finish(results)
- return result
+ return report.run()
class Report(object):
@@ -82,7 +56,7 @@
user_id=None,
status=celery.states.PENDING,
name=None,
- result_key=None,
+ queue_result_key=None,
children=[],
parameters='{}'):
@@ -98,11 +72,11 @@
self.status = status
self.name = name
- self.result_key = result_key
+ self.queue_result_key = queue_result_key
self.children = children
# store report to database
- # note that result_key is always empty at this stage
+ # note that queue_result_key is always empty at this stage
pj = PersistentReport(user_id=self.user_id,
status=self.status,
name=self.name,
@@ -117,7 +91,7 @@
def __repr__(self):
return '<Report("{0}")>'.format(self.persistent_id)
- def set_status(self, status, task_id):
+ def set_status(self, status, task_id=None):
"""
helper function for updating database status after celery
task has been started
@@ -125,7 +99,8 @@
db_session = db.get_session()
pj = db_session.query(PersistentReport).get(self.persistent_id)
pj.status = status
- pj.result_key = task_id
+ if task_id:
+ pj.queue_result_key = task_id
db_session.add(pj)
db_session.commit()
db_session.close()
@@ -139,28 +114,67 @@
class ReportNode(Report):
- finish_task = queue_finish_task
-
def run(self):
+ """
+ This specialized version of run first runs all the children, then
+ calls the finish method with the results.
+
+ NOTE: this used to spawn a tree of celery tasks. That sacrificed
parallelism
+ at the user level and gained parallelism at the task level. That was
bad.
+ So now this just runs all the children's run methods, collects the
results,
+ and passes them to the finish method. Deadlocking and celery worker
starvation
+ are *much* less likely now. Thank you Ori :)
+ """
+ self.set_status(celery.states.STARTED, task_id=current_task.request.id)
+ results = []
+
if self.children:
- callback = queue_finish_task.s(self)
- header = [queue_task.s(child) for child in self.children]
- children_then_finish = chord(header)(callback)
try:
- return children_then_finish.get()
+ child_results = [child.run() for child in self.children]
+ results = self.finish(child_results)
except SoftTimeLimitExceeded:
+ self.set_status(celery.states.FAILURE)
task_logger.error('timeout exceeded for {0}'.format(
current_task.request.id
))
raise
- else:
- return []
+
+ self.set_status(celery.states.SUCCESS)
+ return results
def finish(self, results):
"""
- Each ReportNode sublcass should implement this method to
- deal with the results of its child reports"""
+ Each ReportNode sublcass should implement this method to deal with
+ the results of its child reports. As a standard, report_results should
+ be called at the end of ReportNode.finish implementations.
+ """
pass
+
+ def report_result(self, results, child_results=[]):
+ """
+ Creates a unique identifier for this ReportNode, and returns a one
element
+ dictionary with that identifier as the key and its results as the
value.
+ This allows ReportNode results to be merged as the tree of ReportNodes
is
+ evaluated.
+
+ Parameters
+ results : Anything that the ReportNode compiles in its
finish step
+ child_results : The results from a child Report(s) if they
should be
+ preserved. ReportLeaf results and any
ReportNode results
+ that are copied should not be preserved.
+ """
+ self.result_key = str(uuid4())
+ db_session = db.get_session()
+ pj = db_session.query(PersistentReport).get(self.persistent_id)
+ pj.result_key = self.result_key
+ db_session.add(pj)
+ db_session.commit()
+ db_session.close()
+
+ merged = {self.result_key: results}
+ for child_result in child_results:
+ merged.update(child_result)
+ return merged
class ReportLeaf(Report):
diff --git a/wikimetrics/models/report_nodes/run_report.py
b/wikimetrics/models/report_nodes/run_report.py
index cc584c5..8328399 100644
--- a/wikimetrics/models/report_nodes/run_report.py
+++ b/wikimetrics/models/report_nodes/run_report.py
@@ -84,8 +84,9 @@
self.name = ', '.join(metric_names) + ' for ' + ', '.join(cohort_names)
self.children = children
- def finish(self, aggregated_output):
- return aggregated_output
+ def finish(self, aggregated_results):
+ result = self.report_result('Finished',
child_results=aggregated_results)
+ return result
def __repr__(self):
return '<RunReport("{0}")>'.format(self.persistent_id)
diff --git a/wikimetrics/run.py b/wikimetrics/run.py
index ba75d37..d56e569 100644
--- a/wikimetrics/run.py
+++ b/wikimetrics/run.py
@@ -22,10 +22,6 @@
def run_celery():
from configurables import queue
- from .models import ConcatMetricsReport
- from .models import MultiProjectMetricReport
- from .models import MetricReport
- from .models import RunReport
queue.start(argv=['celery', 'worker', '-l', 'DEBUG'])
--
To view, visit https://gerrit.wikimedia.org/r/79429
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Iea50bfc69841e9efe530fb1d92be27ffa8d20690
Gerrit-PatchSet: 1
Gerrit-Project: analytics/wikimetrics
Gerrit-Branch: master
Gerrit-Owner: Milimetric <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
