Merge pull request #484 from edx/brian/delete-manifest

Delete manifest file on overwrite.

Author: brianhw

edx/analytics/tasks/common/mapreduce.py

@@ -15,7 +15,7 @@
 import luigi.task
 from luigi import configuration
 
-from edx.analytics.tasks.util.manifest import convert_to_manifest_input_if_necessary
+from edx.analytics.tasks.util.manifest import convert_to_manifest_input_if_necessary, remove_manifest_target_if_exists
 from edx.analytics.tasks.util.url import get_target_from_url, url_path_join
 
 log = logging.getLogger(__name__)
@@ -126,9 +126,15 @@ def _get_engine_parameters_from_targets(self):
             'input_format': input_format,
         }
 
+    @property
+    def manifest_id(self):
+        return str(hash(self)).replace('-', 'n')
+
     def input_hadoop(self):
-        manifest_id = str(hash(self)).replace('-', 'n')
-        return convert_to_manifest_input_if_necessary(manifest_id, super(MapReduceJobTask, self).input_hadoop())
+        return convert_to_manifest_input_if_necessary(self.manifest_id, super(MapReduceJobTask, self).input_hadoop())
+
+    def remove_manifest_target_if_exists(self):
+        return remove_manifest_target_if_exists(self.manifest_id)
 
 
 class MapReduceJobRunner(luigi.contrib.hadoop.HadoopJobRunner):

edx/analytics/tasks/insights/module_engagement.py

@@ -208,6 +208,8 @@ def run(self):
         output_target = self.output()
         if not self.complete() and output_target.exists():
             output_target.remove()
+        if self.overwrite:
+            self.remove_manifest_target_if_exists()
         return super(ModuleEngagementDataTask, self).run()
 
 

edx/analytics/tasks/util/manifest.py

@@ -5,7 +5,7 @@
 import luigi.task
 from luigi import configuration
 
-from edx.analytics.tasks.util.url import get_target_class_from_url, url_path_join
+from edx.analytics.tasks.util.url import get_target_class_from_url, get_target_from_url, url_path_join
 
 CONFIG_SECTION = 'manifest'
 
@@ -27,6 +27,13 @@ def convert_to_manifest_input_if_necessary(manifest_id, targets):
         return targets
 
 
+def get_manifest_file_path(manifest_id):
+    # Construct the manifest file URL from the manifest_id and the configuration
+    base_url = configuration.get_config().get(CONFIG_SECTION, 'path')
+    manifest_file_path = url_path_join(base_url, manifest_id + '.manifest')
+    return manifest_file_path
+
+
 def create_manifest_target(manifest_id, targets):
     # If we are running locally, we need our manifest file to be a local file target, however, if we are running on
     # a real Hadoop cluster, it has to be an HDFS file so that the input format can read it. Luigi makes it a little
@@ -35,8 +42,7 @@ def create_manifest_target(manifest_id, targets):
     # base class at runtime based on the URL of the manifest file.
 
     # Construct the manifest file URL from the manifest_id and the configuration
-    base_url = configuration.get_config().get(CONFIG_SECTION, 'path')
-    manifest_file_path = url_path_join(base_url, manifest_id + '.manifest')
+    manifest_file_path = get_manifest_file_path(manifest_id)
 
     # Figure out the type of target that should be used to write/read the file.
     manifest_file_target_class, init_args, init_kwargs = get_target_class_from_url(manifest_file_path)
@@ -49,6 +55,16 @@ class ManifestInputTarget(ManifestInputTargetMixin, manifest_file_target_class):
     return ManifestInputTarget.from_existing_targets(targets, *init_args, **init_kwargs)
 
 
+def remove_manifest_target_if_exists(manifest_id):
+    """Given an id and configuration, construct a target that can check and remove a manifest file."""
+    manifest_file_path = get_manifest_file_path(manifest_id)
+    # we don't need the mixin in order to check for existence or to remove the manifest file.
+    manifest_target = get_target_from_url(manifest_file_path)
+    if manifest_target.exists():
+        log.info('Removing existing manifest found at %s', manifest_target.path)
+        manifest_target.remove()
+
+
 class ManifestInputTargetMixin(object):
 
     def __init__(self, *args, **kwargs):
@@ -66,7 +82,7 @@ def __init__(self, *args, **kwargs):
     def from_existing_targets(cls, other_targets, *init_args, **init_kwargs):
         manifest_target = cls(*init_args, **init_kwargs)
         if not manifest_target.exists():
-            log.debug('Writing manifest file %s', manifest_target.path)
+            log.info('Writing manifest file %s', manifest_target.path)
             with manifest_target.open('w') as manifest_file:
                 for target in other_targets:
                     manifest_file.write(target.path)

edx/analytics/tasks/util/tests/test_manifest.py

@@ -1,13 +1,17 @@
 """Ensure manifest files are created appropriately."""
 
+import os
+import shutil
+import tempfile
 from unittest import TestCase
 
 import luigi
 from luigi.contrib.hdfs.target import HdfsTarget
 from mock import patch
 
 from edx.analytics.tasks.util.manifest import (
-    ManifestInputTargetMixin, convert_to_manifest_input_if_necessary, create_manifest_target
+    ManifestInputTargetMixin, convert_to_manifest_input_if_necessary, create_manifest_target,
+    remove_manifest_target_if_exists
 )
 from edx.analytics.tasks.util.tests.config import OPTION_REMOVED, with_luigi_config
 from edx.analytics.tasks.util.tests.target import FakeTarget
@@ -81,3 +85,30 @@ def test_negative_threshold(self):
     @with_luigi_config('manifest', 'threshold', OPTION_REMOVED)
     def test_threshold_not_set(self):
         self.assert_no_conversion()
+
+
+temp_rootdir = None
+
+
+class ManifestInputTargetDeletionTest(TestCase):
+    """Ensure manifest files are deleted appropriately."""
+
+    MANIFEST_ID = 'test'
+
+    def setUp(self):
+        temp_rootdir = tempfile.mkdtemp()
+        self.addCleanup(self.cleanup, temp_rootdir)
+
+    def cleanup(self, dirname):
+        """Remove the temp directory only if it exists."""
+        if os.path.exists(dirname):
+            shutil.rmtree(dirname)
+
+    @with_luigi_config(
+        ('manifest', 'path', temp_rootdir),
+    )
+    def test_manifest_file_deletion(self):
+        target = create_manifest_target(self.MANIFEST_ID, [HdfsTarget('s3://foo/bar')])
+        self.assertTrue(target.exists())
+        remove_manifest_target_if_exists(self.MANIFEST_ID)
+        self.assertFalse(target.exists())