Andal.LND/edx-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

only dump courses to neo4j if they've been updated since the last time they were dumped

Browse Source 
improvements to the command line interface for caching
This commit is contained in:
Adam Palay
2016-09-20 12:27:49 -04:00
parent 01a0f6d6ee
commit da58ad2a2b
13 changed files with 299 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
openedx/core/djangoapps/coursegraph/__init__.py Normal file
View File

20
openedx/core/djangoapps/coursegraph/apps.py Normal file
View File

@@ -0,0 +1,20 @@
"""
Coursegraph Application Configuration
Signal handlers are connected here.
"""
from django.apps import AppConfig
class CoursegraphConfig(AppConfig):
"""
AppConfig for courseware app
"""
name = 'openedx.core.djangoapps.coursegraph'
def ready(self):
"""
Import signals on startup
"""
from openedx.core.djangoapps.coursegraph import signals # pylint: disable=unused-variable

0
openedx/core/djangoapps/coursegraph/management/__init__.py Normal file
View File

0
openedx/core/djangoapps/coursegraph/management/commands/__init__.py Normal file
View File

354
openedx/core/djangoapps/coursegraph/management/commands/dump_to_neo4j.py Normal file
View File

@@ -0,0 +1,354 @@
"""
This file contains a management command for exporting the modulestore to
neo4j, a graph database.
"""
from __future__ import unicode_literals, print_function
import logging
from django.core.management.base import BaseCommand
from django.utils import six
from opaque_keys.edx.keys import CourseKey
from py2neo import Graph, Node, Relationship, authenticate
from py2neo.compat import integer, string, unicode as neo4j_unicode
from request_cache.middleware import RequestCache
from xmodule.modulestore.django import modulestore
from openedx.core.djangoapps.coursegraph.utils import (
CommandLastRunCache,
CourseLastPublishedCache,
)
log = logging.getLogger(__name__)
# When testing locally, neo4j's bolt logger was noisy, so we'll only have it
# emit logs if there's an error.
bolt_log = logging.getLogger('neo4j.bolt') # pylint: disable=invalid-name
bolt_log.setLevel(logging.ERROR)
ITERABLE_NEO4J_TYPES = (tuple, list, set, frozenset)
PRIMITIVE_NEO4J_TYPES = (integer, string, neo4j_unicode, float, bool)
COMMAND_LAST_RUN_CACHE = CommandLastRunCache()
COURSE_LAST_PUBLISHED_CACHE = CourseLastPublishedCache()
class ModuleStoreSerializer(object):
"""
Class with functionality to serialize a modulestore into subgraphs,
one graph per course.
"""
def __init__(self, courses=None):
"""
Sets the object's course_keys attribute from the `courses` parameter.
If that parameter isn't furnished, loads all course_keys from the
modulestore.
:param courses: string serialization of course keys
"""
if courses:
course_keys = [CourseKey.from_string(course.strip()) for course in courses]
else:
course_keys = [
course.id for course in modulestore().get_course_summaries()
]
self.course_keys = course_keys
@staticmethod
def serialize_item(item):
"""
Args:
item: an XBlock
Returns:
fields: a dictionary of an XBlock's field names and values
label: the name of the XBlock's type (i.e. 'course'
or 'problem')
"""
# convert all fields to a dict and filter out parent and children field
fields = dict(
(field, field_value.read_from(item))
for (field, field_value) in six.iteritems(item.fields)
if field not in ['parent', 'children']
)
course_key = item.scope_ids.usage_id.course_key
# set or reset some defaults
fields['edited_on'] = six.text_type(getattr(item, 'edited_on', ''))
fields['display_name'] = item.display_name_with_default
fields['org'] = course_key.org
fields['course'] = course_key.course
fields['run'] = course_key.run
fields['course_key'] = six.text_type(course_key)
label = item.scope_ids.block_type
# prune some fields
if label == 'course':
if 'checklists' in fields:
del fields['checklists']
return fields, label
def serialize_course(self, course_id):
"""
Args:
course_id: CourseKey of the course we want to serialize
Returns:
nodes: a list of py2neo Node objects
relationships: a list of py2neo Relationships objects
Serializes a course into Nodes and Relationships
"""
# create a location to node mapping we'll need later for
# writing relationships
location_to_node = {}
items = modulestore().get_items(course_id)
# create nodes
nodes = []
for item in items:
fields, label = self.serialize_item(item)
for field_name, value in six.iteritems(fields):
fields[field_name] = self.coerce_types(value)
node = Node(label, 'item', **fields)
nodes.append(node)
location_to_node[item.location] = node
# create relationships
relationships = []
for item in items:
for child_loc in item.get_children():
parent_node = location_to_node.get(item.location)
child_node = location_to_node.get(child_loc.location)
if parent_node is not None and child_node is not None:
relationship = Relationship(parent_node, "PARENT_OF", child_node)
relationships.append(relationship)
return nodes, relationships
@staticmethod
def coerce_types(value):
"""
Args:
value: the value of an xblock's field
Returns: either the value, a text version of the value, or, if the
value is iterable, the value with each element being converted to text
"""
coerced_value = value
if isinstance(value, ITERABLE_NEO4J_TYPES):
coerced_value = []
for element in value:
coerced_value.append(six.text_type(element))
# convert coerced_value back to its original type
coerced_value = type(value)(coerced_value)
# if it's not one of the types that neo4j accepts,
# just convert it to text
elif not isinstance(value, PRIMITIVE_NEO4J_TYPES):
coerced_value = six.text_type(value)
return coerced_value
@staticmethod
def add_to_transaction(neo4j_entities, transaction):
"""
Args:
neo4j_entities: a list of Nodes or Relationships
transaction: a neo4j transaction
"""
for entity in neo4j_entities:
transaction.create(entity)
@staticmethod
def should_dump_course(course_key):
"""
Only dump the course if it's been changed since the last time it's been
dumped.
:param course_key: a CourseKey object.
:return: bool. Whether or not this course should be dumped to neo4j.
"""
last_this_command_was_run = COMMAND_LAST_RUN_CACHE.get(course_key)
last_course_had_published_event = COURSE_LAST_PUBLISHED_CACHE.get(
course_key
)
# if we have no record of this course being serialized, serialize it
if last_this_command_was_run is None:
return True
# if we've serialized the course recently and we have no published
# events, we can skip re-serializing it
if last_this_command_was_run and last_course_had_published_event is None:
return False
# otherwise, serialize if the command was run before the course's last
# published event
return last_this_command_was_run < last_course_had_published_event
def dump_courses_to_neo4j(self, graph, override_cache=False):
"""
Parameters
----------
graph: py2neo graph object
override_cache: serialize the courses even if they'be been recently
serialized
Returns two lists: one of the courses that were successfully written
to neo4j, and one of courses that were not.
-------
"""
total_number_of_courses = len(self.course_keys)
successful_courses = []
unsuccessful_courses = []
for index, course_key in enumerate(self.course_keys):
# first, clear the request cache to prevent memory leaks
RequestCache.clear_request_cache()
log.info(
"Now exporting %s to neo4j: course %d of %d total courses",
course_key,
index + 1,
total_number_of_courses,
)
if not (override_cache or self.should_dump_course(course_key)):
log.info("skipping dumping %s, since it hasn't changed", course_key)
continue
nodes, relationships = self.serialize_course(course_key)
log.info(
"%d nodes and %d relationships in %s",
len(nodes),
len(relationships),
course_key,
)
transaction = graph.begin()
course_string = six.text_type(course_key)
try:
# first, delete existing course
transaction.run(
"MATCH (n:item) WHERE n.course_key='{}' DETACH DELETE n".format(
course_string
)
)
# now, re-add it
self.add_to_transaction(nodes, transaction)
self.add_to_transaction(relationships, transaction)
transaction.commit()
except Exception: # pylint: disable=broad-except
log.exception(
"Error trying to dump course %s to neo4j, rolling back",
course_string
)
transaction.rollback()
unsuccessful_courses.append(course_string)
else:
COMMAND_LAST_RUN_CACHE.set(course_key)
successful_courses.append(course_string)
return successful_courses, unsuccessful_courses
class Command(BaseCommand):
"""
Command to dump modulestore data to neo4j
Takes the following named arguments:
host: the host of the neo4j server
https_port: the port on the neo4j server that accepts https requests
http_port: the port on the neo4j server that accepts http requests
secure: if set, connects to server over https, otherwise uses http
user: the username for the neo4j user
password: the user's password
courses: list of course key strings to serialize. If not specified, all
courses in the modulestore are serialized.
override: if true, dump all--or all specified--courses, regardless of when
they were last dumped. If false, or not set, only dump those courses that
were updated since the last time the command was run.
Example usage:
python manage.py lms dump_to_neo4j --host localhost --https_port 7473 \
--secure --user user --password password --settings=aws
"""
def add_arguments(self, parser):
parser.add_argument('--host', type=unicode)
parser.add_argument('--https_port', type=int, default=7473)
parser.add_argument('--http_port', type=int, default=7474)
parser.add_argument('--secure', action='store_true')
parser.add_argument('--user', type=unicode)
parser.add_argument('--password', type=unicode)
parser.add_argument('--courses', type=unicode, nargs='*')
parser.add_argument(
'--override',
action='store_true',
help='dump all--or all specified--courses, ignoring cache',
)
def handle(self, *args, **options): # pylint: disable=unused-argument
"""
Iterates through each course, serializes them into graphs, and saves
those graphs to neo4j.
"""
host = options['host']
https_port = options['https_port']
http_port = options['http_port']
secure = options['secure']
neo4j_user = options['user']
neo4j_password = options['password']
authenticate(
"{host}:{port}".format(host=host, port=https_port if secure else http_port),
neo4j_user,
neo4j_password,
)
graph = Graph(
bolt=True,
password=neo4j_password,
user=neo4j_user,
https_port=https_port,
http_port=http_port,
host=host,
secure=secure,
)
mss = ModuleStoreSerializer(options['courses'])
successful_courses, unsuccessful_courses = mss.dump_courses_to_neo4j(
graph, override_cache=options['override']
)
if not successful_courses and not unsuccessful_courses:
print("No courses exported to neo4j at all!")
return
if successful_courses:
print(
"These courses exported to neo4j successfully:\n\t" +
"\n\t".join(successful_courses)
)
else:
print("No courses exported to neo4j successfully.")
if unsuccessful_courses:
print(
"These courses did not export to neo4j successfully:\n\t" +
"\n\t".join(unsuccessful_courses)
)
else:
print("All courses exported to neo4j successfully.")

0
openedx/core/djangoapps/coursegraph/management/commands/tests/__init__.py Normal file
View File

278
openedx/core/djangoapps/coursegraph/management/commands/tests/test_dump_to_neo4j.py Normal file
View File

@@ -0,0 +1,278 @@
# coding=utf-8
"""
Tests for the dump_to_neo4j management command.
"""
from __future__ import unicode_literals
from datetime import datetime
import ddt
import mock
from django.core.management import call_command
from django.utils import six
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from openedx.core.djangoapps.coursegraph.management.commands.dump_to_neo4j import (
ModuleStoreSerializer,
ITERABLE_NEO4J_TYPES,
)
from openedx.core.djangoapps.coursegraph.signals import _listen_for_course_publish
class TestDumpToNeo4jCommandBase(SharedModuleStoreTestCase):
"""
Base class for the test suites in this file. Sets up a couple courses.
"""
@classmethod
def setUpClass(cls):
super(TestDumpToNeo4jCommandBase, cls).setUpClass()
cls.course = CourseFactory.create()
cls.chapter = ItemFactory.create(parent=cls.course, category='chapter')
cls.sequential = ItemFactory.create(parent=cls.chapter, category='sequential')
cls.vertical = ItemFactory.create(parent=cls.sequential, category='vertical')
cls.html = ItemFactory.create(parent=cls.vertical, category='html')
cls.problem = ItemFactory.create(parent=cls.vertical, category='problem')
cls.video = ItemFactory.create(parent=cls.vertical, category='video')
cls.video2 = ItemFactory.create(parent=cls.vertical, category='video')
cls.course2 = CourseFactory.create()
cls.course_strings = [six.text_type(cls.course.id), six.text_type(cls.course2.id)]
@ddt.ddt
class TestDumpToNeo4jCommand(TestDumpToNeo4jCommandBase):
"""
Tests for the dump to neo4j management command
"""
@mock.patch('openedx.core.djangoapps.coursegraph.management.commands.dump_to_neo4j.Graph')
@ddt.data(1, 2)
def test_dump_specific_courses(self, number_of_courses, mock_graph_class):
"""
Test that you can specify which courses you want to dump.
"""
mock_graph = mock_graph_class.return_value
mock_transaction = mock.Mock()
mock_graph.begin.return_value = mock_transaction
call_command(
'dump_to_neo4j',
courses=self.course_strings[:number_of_courses],
host='mock_host',
http_port=7474,
user='mock_user',
password='mock_password',
)
self.assertEqual(mock_graph.begin.call_count, number_of_courses)
self.assertEqual(mock_transaction.commit.call_count, number_of_courses)
self.assertEqual(mock_transaction.commit.rollback.call_count, 0)
@mock.patch('openedx.core.djangoapps.coursegraph.management.commands.dump_to_neo4j.Graph')
def test_dump_all_courses(self, mock_graph_class):
"""
Test if you don't specify which courses to dump, then you'll dump
all of them.
"""
mock_graph = mock_graph_class.return_value
mock_transaction = mock.Mock()
mock_graph.begin.return_value = mock_transaction
call_command(
'dump_to_neo4j',
host='mock_host',
http_port=7474,
user='mock_user',
password='mock_password',
)
self.assertEqual(mock_graph.begin.call_count, 2)
self.assertEqual(mock_transaction.commit.call_count, 2)
self.assertEqual(mock_transaction.commit.rollback.call_count, 0)
@ddt.ddt
class TestModuleStoreSerializer(TestDumpToNeo4jCommandBase):
"""
Tests for the ModuleStoreSerializer
"""
@classmethod
def setUpClass(cls):
"""Any ModuleStore course/content operations can go here."""
super(TestModuleStoreSerializer, cls).setUpClass()
cls.mss = ModuleStoreSerializer()
def test_serialize_item(self):
"""
Tests the serialize_item method.
"""
fields, label = self.mss.serialize_item(self.course)
self.assertEqual(label, "course")
self.assertIn("edited_on", fields.keys())
self.assertIn("display_name", fields.keys())
self.assertIn("org", fields.keys())
self.assertIn("course", fields.keys())
self.assertIn("run", fields.keys())
self.assertIn("course_key", fields.keys())
self.assertNotIn("checklist", fields.keys())
def test_serialize_course(self):
"""
Tests the serialize_course method.
"""
nodes, relationships = self.mss.serialize_course(
self.course.id
)
self.assertEqual(len(nodes), 9)
self.assertEqual(len(relationships), 7)
@ddt.data(*ITERABLE_NEO4J_TYPES)
def test_coerce_types_iterable(self, iterable_type):
"""
Tests the coerce_types helper method for iterable types
"""
example_iterable = iterable_type([object, object, object])
# each element in the iterable is not unicode:
self.assertFalse(any(isinstance(tab, six.text_type) for tab in example_iterable))
# but after they are coerced, they are:
coerced = ModuleStoreSerializer().coerce_types(example_iterable)
self.assertTrue(all(isinstance(tab, six.text_type) for tab in coerced))
# finally, make sure we haven't changed the type:
self.assertEqual(type(coerced), iterable_type)
@ddt.data(
(1, 1),
(object, "<type 'object'>"),
(1.5, 1.5),
("úñîçø∂é", "úñîçø∂é"),
(b"plain string", b"plain string"),
(True, True),
(None, "None"),
)
@ddt.unpack
def test_coerce_types_base(self, original_value, coerced_expected):
"""
Tests the coerce_types helper for the neo4j base types
"""
coerced_value = self.mss.coerce_types(original_value)
self.assertEqual(coerced_value, coerced_expected)
def test_dump_to_neo4j(self):
"""
Tests the dump_to_neo4j method works against a mock
py2neo Graph
"""
mock_graph = mock.Mock()
mock_transaction = mock.Mock()
mock_graph.begin.return_value = mock_transaction
successful, unsuccessful = self.mss.dump_courses_to_neo4j(mock_graph)
self.assertEqual(mock_graph.begin.call_count, 2)
self.assertEqual(mock_transaction.commit.call_count, 2)
self.assertEqual(mock_transaction.rollback.call_count, 0)
# 7 nodes + 9 relationships from the first course
# 2 nodes and no relationships from the second
self.assertEqual(mock_transaction.create.call_count, 18)
self.assertEqual(mock_transaction.run.call_count, 2)
self.assertEqual(len(unsuccessful), 0)
self.assertItemsEqual(successful, self.course_strings)
def test_dump_to_neo4j_rollback(self):
"""
Tests that the the dump_to_neo4j method handles the case where there's
an exception trying to write to the neo4j database.
"""
mock_graph = mock.Mock()
mock_transaction = mock.Mock()
mock_graph.begin.return_value = mock_transaction
mock_transaction.run.side_effect = ValueError('Something went wrong!')
successful, unsuccessful = self.mss.dump_courses_to_neo4j(mock_graph)
self.assertEqual(mock_graph.begin.call_count, 2)
self.assertEqual(mock_transaction.commit.call_count, 0)
self.assertEqual(mock_transaction.rollback.call_count, 2)
self.assertEqual(len(successful), 0)
self.assertItemsEqual(unsuccessful, self.course_strings)
@ddt.data(
(True, 2),
(False, 0),
)
@ddt.unpack
def test_dump_to_neo4j_cache(self, override_cache, expected_number_courses):
"""
Tests the caching mechanism and override to make sure we only publish
recently updated courses.
"""
mock_graph = mock.Mock()
# run once to warm the cache
successful, unsuccessful = self.mss.dump_courses_to_neo4j(mock_graph)
self.assertEqual(len(successful + unsuccessful), len(self.course_strings))
# when run the second time, only dump courses if the cache override
# is enabled
successful, unsuccessful = self.mss.dump_courses_to_neo4j(
mock_graph, override_cache=override_cache
)
self.assertEqual(len(successful + unsuccessful), expected_number_courses)
def test_dump_to_neo4j_published(self):
"""
Tests that we only dump those courses that have been published after
the last time the command was been run.
"""
mock_graph = mock.Mock()
# run once to warm the cache
successful, unsuccessful = self.mss.dump_courses_to_neo4j(mock_graph)
self.assertEqual(len(successful + unsuccessful), len(self.course_strings))
# simulate one of the courses being published
_listen_for_course_publish(None, self.course.id)
# make sure only the published course was dumped
successful, unsuccessful = self.mss.dump_courses_to_neo4j(mock_graph)
self.assertEqual(len(unsuccessful), 0)
self.assertEqual(len(successful), 1)
self.assertEqual(successful[0], unicode(self.course.id))
@ddt.data(
(datetime(2016, 3, 30), datetime(2016, 3, 31), True),
(datetime(2016, 3, 31), datetime(2016, 3, 30), False),
(datetime(2016, 3, 31), None, False),
(None, datetime(2016, 3, 30), True),
(None, None, True),
)
@ddt.unpack
@mock.patch('openedx.core.djangoapps.coursegraph.management.commands.dump_to_neo4j.COMMAND_LAST_RUN_CACHE')
@mock.patch('openedx.core.djangoapps.coursegraph.management.commands.dump_to_neo4j.COURSE_LAST_PUBLISHED_CACHE')
def test_should_dump_course(
self,
last_command_run,
last_course_published,
should_dump,
mock_course_last_published_cache,
mock_command_last_run_cache,
):
"""
Tests whether a course should be dumped given the last time it was
dumped and the last time it was published.
"""
mock_command_last_run_cache.get.return_value = last_command_run
mock_course_last_published_cache.get.return_value = last_course_published
mock_course_key = mock.Mock
self.assertEqual(
self.mss.should_dump_course(mock_course_key),
should_dump
)

15
openedx/core/djangoapps/coursegraph/signals.py Normal file
View File

@@ -0,0 +1,15 @@
"""
Signal handlers for the CourseGraph application
"""
from django.dispatch.dispatcher import receiver
from xmodule.modulestore.django import SignalHandler
from openedx.core.djangoapps.coursegraph.utils import CourseLastPublishedCache
@receiver(SignalHandler.course_published)
def _listen_for_course_publish(sender, course_key, **kwargs): # pylint: disable=unused-argument
"""
Register when the course was published on a course publish event
"""
CourseLastPublishedCache().set(course_key)

0
openedx/core/djangoapps/coursegraph/tests/__init__.py Normal file
View File

27
openedx/core/djangoapps/coursegraph/tests/test_signals.py Normal file
View File

@@ -0,0 +1,27 @@
"""
Tests for coursegraph's signal handler on course publish
"""
from __future__ import unicode_literals
from opaque_keys.edx.keys import CourseKey
from openedx.core.djangoapps.coursegraph.signals import _listen_for_course_publish
from openedx.core.djangoapps.coursegraph.utils import CourseLastPublishedCache
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
class TestCourseGraphSignalHandler(CacheIsolationTestCase):
"""
Tests for the course publish course handler
"""
ENABLED_CACHES = ['default']
def test_cache_set_on_course_publish(self):
"""
Tests that the last published cache is set on course publish
"""
course_key = CourseKey.from_string('course-v1:org+course+run')
last_published_cache = CourseLastPublishedCache()
self.assertIsNone(last_published_cache.get(course_key))
_listen_for_course_publish(None, course_key)
self.assertIsNotNone(last_published_cache.get(course_key))

52
openedx/core/djangoapps/coursegraph/utils.py Normal file
View File

@@ -0,0 +1,52 @@
"""
Helpers for the CourseGraph app
"""
from django.core.cache import cache
from django.utils import timezone
class TimeRecordingCacheBase(object):
"""
A base class for caching the current time for some key.
"""
# cache_prefix should be defined in children classes
cache_prefix = None
_cache = cache
def _key(self, course_key):
"""
Make a cache key from the prefix and a course_key
:param course_key: CourseKey object
:return: a cache key
"""
return self.cache_prefix + unicode(course_key)
def get(self, course_key):
"""
Gets the time value associated with the CourseKey.
:param course_key: a CourseKey object.
:return: the time the key was last set.
"""
return self._cache.get(self._key(course_key))
def set(self, course_key):
"""
Sets the current time for a CourseKey key.
:param course_key: a CourseKey object.
"""
return self._cache.set(self._key(course_key), timezone.now())
class CourseLastPublishedCache(TimeRecordingCacheBase):
"""
Used to record the last time that a course had a publish event run on it.
"""
cache_prefix = u'course_last_published'
class CommandLastRunCache(TimeRecordingCacheBase):
"""
Used to record the last time that the dump_to_neo4j command was run on a
course.
"""
cache_prefix = u'dump_to_neo4j_command_last_run'