Sfoglia il codice sorgente

HUE-8508 [core] Add command to cleanup old data in tables

Chris Conner 7 anni fa
parent
commit
e77d34b

+ 0 - 66
desktop/core/src/desktop/management/commands/clean_history_docs.py

@@ -1,66 +0,0 @@
-#!/usr/bin/env python
-# Licensed to Cloudera, Inc. under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  Cloudera, Inc. licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import logging
-from datetime import datetime,  timedelta
-
-from django.core.management.base import BaseCommand
-
-from desktop.models import Document2
-
-
-LOG = logging.getLogger(__name__)
-
-DEFAULT_EXPIRY_DAYS = 120
-
-
-class Command(BaseCommand):
-  """
-  Clean up (delete) history documents without a parent or that are older than N number of days.
-
-  e.g.
-  build/env/bin/hue clean_history_docs 30
-  0 history documents deleted.
-  """
-  args = '<age_in_days> (default is %s)' % DEFAULT_EXPIRY_DAYS
-  help = 'Delete history documents older than %s days.' % DEFAULT_EXPIRY_DAYS
-
-  def handle(self, *args, **options):
-    count = 0
-    days = int(args[0]) if len(args) >= 1 else DEFAULT_EXPIRY_DAYS
-
-    # Clean up orphan history documents (excluding query docs)
-    orphans = Document2.objects.exclude(type__startswith='query-').filter(is_history=True).filter(dependents=None)
-
-    if orphans.count() > 0:
-      count += orphans.count()
-      self.stdout.write('Deleting %d orphan history documents...' % orphans.count())
-      orphans.delete()
-    else:
-      self.stdout.write('No orphan history documents found.')
-
-    # Clean up old history documents
-    old_history_docs = Document2.objects.filter(is_history=True).filter(last_modified__lte=datetime.today() - timedelta(days=days))
-
-    if old_history_docs.count() > 0:
-      count += old_history_docs.count()
-      self.stdout.write('Deleting %d history documents older than %d days...' % (old_history_docs.count(), days))
-      old_history_docs.delete()
-    else:
-      self.stdout.write('No history documents older than %d days found.' % days)
-
-    self.stdout.write('%d total history documents deleted.' % count)

+ 155 - 0
desktop/core/src/desktop/management/commands/desktop_document_cleanup.py

@@ -0,0 +1,155 @@
+
+# adapted from django-extensions (http://code.google.com/p/django-command-extensions/)
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import time
+
+from importlib import import_module
+
+from django.conf import settings
+from django.core.management.base import BaseCommand, CommandError
+from django.utils.translation import ugettext_lazy as _t, ugettext as _
+from beeswax.models import SavedQuery
+from beeswax.models import Session
+from datetime import date, timedelta
+from oozie.models import Workflow
+from django.db.utils import DatabaseError
+import desktop.conf
+from desktop.models import Document2
+import logging
+import logging.handlers
+
+import desktop.conf
+
+LOG = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Handler for purging old Query History, Workflow documents and Session data
+    """
+
+    try:
+        from optparse import make_option
+        option_list = BaseCommand.option_list + (
+            make_option("--keep-days", help=_t("Number of days of history data to keep."),
+                    action="store",
+                    type=int,
+                    default=30),
+        )
+
+    except AttributeError, e:
+        baseoption_test = 'BaseCommand' in str(e) and 'option_list' in str(e)
+        if baseoption_test:
+            def add_arguments(self, parser):
+                parser.add_argument("--keep-days", help=_t("Number of days of history data to keep."),
+                    action="store",
+                    type=int,
+                    default=30)
+        else:
+            LOG.exception(str(e))
+            sys.exit(1)
+
+
+    def objectCleanup(self, objClass, filterType, filterValue, dateField):
+        errorCount = 0
+        checkCount = 0
+        resets = 0
+        deleteRecords = self.deleteRecordsBase
+
+        totalObjects = objClass.objects.filter(**{ '%s' % filterType: filterValue, '%s__lte' % dateField: self.timeDeltaObj, })\
+                                                .values_list("id", flat=True)
+        LOG.info("Looping through %s objects. %s objects to be deleted." % (objClass.__name__, totalObjects.count()))
+        while totalObjects.count():
+            if deleteRecords < 30 and resets < self.resetMax:
+                checkCount += 1
+            if checkCount == self.resetCount:
+                deleteRecords = self.deleteRecordsBase
+                resets += 1
+                checkCount = 0
+            LOG.info("%s objects left: %s" % (objClass.__name__, totalObjects.count()))
+            deleteObjects = objClass.objects.filter(**{ '%s' % filterType: filterValue, '%s__lte' % dateField: self.timeDeltaObj, })\
+                                                    .values_list("id", flat=True)[:deleteRecords]
+            try:
+                objClass.objects.filter(pk__in=list(deleteObjects)).delete()
+                errorCount = 0
+            except DatabaseError, e:
+                LOG.info("Non Fatal Exception: %s: %s" % (e.__class__.__name__, e))
+                errorCount += 1
+                if errorCount > 9 and deleteRecords == 1:
+                    raise
+                if deleteRecords > 100:
+                    deleteRecords = max(deleteRecords - 100, 1)
+                else:
+                    deleteRecords = max(deleteRecords - 10, 1)
+                LOG.info("Decreasing max delete records to: %s" % deleteRecords)
+            totalObjects = objClass.objects.filter(**{'%s' % filterType: filterValue, '%s__lte' % dateField: self.timeDeltaObj, })\
+                                                    .values_list("id", flat=True)
+
+
+    def handle(self, *args, **options):
+
+
+        self.keepDays = options['keep_days']
+        self.timeDeltaObj = date.today() - timedelta(days=self.keepDays)
+        self.resetCount = 15
+        self.resetMax = 5
+        self.deleteRecordsBase = 999  #number of documents to delete in a batch
+                                      #to avoid Non Fatal Exception: DatabaseError: too many SQL variables
+
+        LOG.warn("HUE_CONF_DIR: %s" % os.environ['HUE_CONF_DIR'])
+        LOG.info("DB Engine: %s" % desktop.conf.DATABASE.ENGINE.get())
+        LOG.info("DB Name: %s" % desktop.conf.DATABASE.NAME.get())
+        LOG.info("DB User: %s" % desktop.conf.DATABASE.USER.get())
+        LOG.info("DB Host: %s" % desktop.conf.DATABASE.HOST.get())
+        LOG.info("DB Port: %s" % str(desktop.conf.DATABASE.PORT.get()))
+        LOG.info("Cleaning up anything in the Hue tables django_session, oozie*, desktop* and beeswax* older than %s old" % self.keepDays)
+
+        start = time.time()
+
+        #Clean out Hive / Impala Query History
+        self.objectCleanup(SavedQuery, 'is_auto', True, 'mtime')
+
+        #Clear out old Hive/Impala sessions
+        self.objectCleanup(Session, 'status_code__gte', -10000, 'last_used')
+
+        #Clean out Trashed Workflows
+        self.objectCleanup(Workflow, 'is_trashed', True, 'last_modified')
+
+        #Clean out Workflows without a name
+        self.objectCleanup(Workflow, 'name', '', 'last_modified')
+
+        #Clean out history Doc2 objects
+        self.objectCleanup(Document2, 'is_history', True, 'last_modified')
+
+        #Clean out expired sessions
+        LOG.debug("Cleaning out expired sessions from django_session table")
+
+        engine = import_module(settings.SESSION_ENGINE)
+        try:
+            engine.SessionStore.clear_expired()
+        except NotImplementedError:
+            LOG.error("Session engine '%s' doesn't support clearing "
+                            "expired sessions.\n" % settings.SESSION_ENGINE)
+
+
+        end = time.time()
+        elapsed = (end - start)
+        LOG.debug("Total time elapsed (seconds): %.2f" % elapsed)
+
+