files: import datetime
  import logging
  import os
  import random
  import re
  import sys
  import threading
  import time
  
  import appscale_datastore_batch
  import dbconstants
  import utils
  
  from appscale.common import appscale_info
  from appscale.common import constants
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.common.unpackaged import DASHBOARD_DIR
  from appscale.taskqueue.distributed_tq import TaskName
  from .cassandra_env import cassandra_interface
  from .datastore_distributed import DatastoreDistributed
  from .utils import get_composite_indexes_rows
  from .zkappscale import zktransaction as zk
  from .zkappscale.entity_lock import EntityLock
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import apiproxy_stub_map
  from google.appengine.api import datastore_distributed
  from google.appengine.api.memcache import memcache_distributed
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore.datastore_query import Cursor
  from google.appengine.ext import db
  from google.appengine.ext.db import stats
  from google.appengine.ext.db import metadata
  from google.appengine.api import datastore_errors
  
  sys.path.append(os.path.join(DASHBOARD_DIR, 'lib'))
  from dashboard_logs import RequestLogLine
  
  
  class DatastoreGroomer(threading.Thread):
    """ Scans the entire database for each application. """
  
    # The amount of seconds between polling to get the groomer lock.
    # Each datastore server does this poll, so it happens the number
    # of datastore servers within this lock period.
    LOCK_POLL_PERIOD = 4 * 60 * 60 # <- 4 hours
  
    # Retry sleep on datastore error in seconds.
    DB_ERROR_PERIOD = 30
  
    # The number of entities retrieved in a datastore request.
    BATCH_SIZE = 100
  
    # Any kind that is of __*__ is private and should not have stats.
    PRIVATE_KINDS = '__(.*)__'
  
    # Any kind that is of _*_ is protected and should not have stats.
    PROTECTED_KINDS = '_(.*)_'
  
    # The amount of time in seconds before we want to clean up task name holders.
    TASK_NAME_TIMEOUT = 24 * 60 * 60
  
    # The amount of time before logs are considered too old.
    LOG_STORAGE_TIMEOUT = 24 * 60 * 60 * 7
  
    # Do not generate stats for AppScale internal apps.
    APPSCALE_APPLICATIONS = ['apichecker', 'appscaledashboard']
  
    # A sentinel value to signify that this app does not have composite indexes.
    NO_COMPOSITES = "NO_COMPS_INDEXES_HERE"
  
    # The path in ZooKeeper where the groomer state is stored.
    GROOMER_STATE_PATH = '/appscale/groomer_state'
  
    # The characters used to separate values when storing the groomer state.
    GROOMER_STATE_DELIMITER = '||'
  
    # The ID for the task to clean up entities.
    CLEAN_ENTITIES_TASK = 'entities'
  
    # The ID for the task to clean up ascending indices.
    CLEAN_ASC_INDICES_TASK = 'asc-indices'
  
    # The ID for the task to clean up descending indices.
    CLEAN_DSC_INDICES_TASK = 'dsc-indices'
  
    # The ID for the task to clean up kind indices.
    CLEAN_KIND_INDICES_TASK = 'kind-indices'
  
    # The ID for the task to clean up old logs.
    CLEAN_LOGS_TASK = 'logs'
  
    # The ID for the task to clean up old tasks.
    CLEAN_TASKS_TASK = 'tasks'
  
    # Log progress every time this many seconds have passed.
    LOG_PROGRESS_FREQUENCY = 60 * 5
  
    def __init__(self, zoo_keeper, table_name, ds_path):
      """ Constructor.
  
      Args:
        zk: ZooKeeper client.
        table_name: The database used (ie, cassandra)
        ds_path: The connection path to the datastore_server.
      """
      logging.info("Logging started")
  
      threading.Thread.__init__(self)
      self.zoo_keeper = zoo_keeper
      self.table_name = table_name
      self.db_access = None
      self.ds_access = None
      self.datastore_path = ds_path
      self.stats = {}
      self.namespace_info = {}
      self.num_deletes = 0
      self.composite_index_cache = {}
      self.entities_checked = 0
      self.journal_entries_cleaned = 0
      self.index_entries_checked = 0
      self.index_entries_delete_failures = 0
      self.index_entries_cleaned = 0
      self.last_logged = time.time()
      self.groomer_state = []
  
    def stop(self):
      """ Stops the groomer thread. """
      self.zoo_keeper.close()
  
    def run(self):
      """ Starts the main loop of the groomer thread. """
      while True:
  
        logging.debug("Trying to get groomer lock.")
        if self.get_groomer_lock():
          logging.info("Got the groomer lock.")
          self.run_groomer()
          try:
            self.zoo_keeper.release_lock_with_path(zk.DS_GROOM_LOCK_PATH)
          except zk.ZKTransactionException, zk_exception:
            logging.error("Unable to release zk lock {0}.".\
              format(str(zk_exception)))
          except zk.ZKInternalException, zk_exception:
            logging.error("Unable to release zk lock {0}.".\
              format(str(zk_exception)))
        else:
          logging.info("Did not get the groomer lock.")
        sleep_time = random.randint(1, self.LOCK_POLL_PERIOD)
        logging.info('Sleeping for {:.1f} minutes.'.format(sleep_time/60.0))
        time.sleep(sleep_time)
  
    def get_groomer_lock(self):
      """ Tries to acquire the lock to the datastore groomer.
  
      Returns:
        True on success, False otherwise.
      """
      return self.zoo_keeper.get_lock_with_path(zk.DS_GROOM_LOCK_PATH)
  
    def get_entity_batch(self, last_key):
      """ Gets a batch of entites to operate on.
  
      Args:
        last_key: The last key from a previous query.
      Returns:
        A list of entities.
      """
      return self.db_access.range_query(dbconstants.APP_ENTITY_TABLE,
        dbconstants.APP_ENTITY_SCHEMA, last_key, "", self.BATCH_SIZE,
        start_inclusive=False)
  
    def reset_statistics(self):
      """ Reinitializes statistics. """
      self.stats = {}
      self.namespace_info = {}
      self.num_deletes = 0
      self.journal_entries_cleaned = 0
  
    def hard_delete_row(self, row_key):
      """ Does a hard delete on a given row key to the entity
          table.
  
      Args:
        row_key: A str representing the row key to delete.
      Returns:
        True on success, False otherwise.
      """
      try:
        self.db_access.batch_delete(dbconstants.APP_ENTITY_TABLE,
          [row_key])
      except dbconstants.AppScaleDBConnectionError, db_error:
        logging.error("Error hard deleting key {0}-->{1}".format(
          row_key, db_error))
        return False
      except Exception, exception:
        logging.error("Caught unexcepted exception {0}".format(exception))
        return False
  
      return True
  
    def load_composite_cache(self, app_id):
      """ Load the composite index cache for an application ID.
  
      Args:
        app_id: A str, the application ID.
      Returns:
        True if the application has composites. False otherwise.
      """
      start_key = dbconstants.KEY_DELIMITER.join([app_id, 'index', ''])
      end_key = dbconstants.KEY_DELIMITER.join(
        [app_id, 'index', dbconstants.TERMINATING_STRING])
  
      results = self.db_access.range_query(dbconstants.METADATA_TABLE,
        dbconstants.METADATA_TABLE, start_key, end_key,
        dbconstants.MAX_NUMBER_OF_COMPOSITE_INDEXES)
      list_result = []
      for list_item in results:
        for _, value in list_item.iteritems():
          list_result.append(value['data'])
  
      self.composite_index_cache[app_id] = self.NO_COMPOSITES
      kind_index_dictionary = {}
      for index in list_result:
        new_index = entity_pb.CompositeIndex()
        new_index.ParseFromString(index)
        kind = new_index.definition().entity_type()
        if kind in kind_index_dictionary:
          kind_index_dictionary[kind].append(new_index)
        else:
          kind_index_dictionary[kind] = [new_index]
      if kind_index_dictionary:
        self.composite_index_cache[app_id] = kind_index_dictionary
        return True
  
      return False
  
    def fetch_entity_dict_for_references(self, references):
      """ Fetches a dictionary of valid entities for a list of references.
  
      Args:
        references: A list of index references to entities.
      Returns:
        A dictionary of validated entities.
      """
      keys = []
      for item in references:
        keys.append(item.values()[0][self.ds_access.INDEX_REFERENCE_COLUMN])
      keys = list(set(keys))
      entities = self.db_access.batch_get_entity(dbconstants.APP_ENTITY_TABLE,
        keys, dbconstants.APP_ENTITY_SCHEMA)
  
      # The datastore needs to know the app ID. The indices could be scattered
      # across apps.
      entities_by_app = {}
      for key in entities:
        app = key.split(self.ds_access._SEPARATOR)[0]
        if app not in entities_by_app:
          entities_by_app[app] = {}
        entities_by_app[app][key] = entities[key]
  
      entities = {}
      for app in entities_by_app:
        app_entities = entities_by_app[app]
        for key in keys:
          if key not in app_entities:
            continue
          if dbconstants.APP_ENTITY_SCHEMA[0] not in app_entities[key]:
            continue
          entities[key] = app_entities[key][dbconstants.APP_ENTITY_SCHEMA[0]]
      return entities
  
    def guess_group_from_table_key(self, entity_key):
      """ Construct a group reference based on an entity key.
  
      Args:
        entity_key: A string specifying an entity table key.
      Returns:
        An entity_pb.Reference object specifying the entity group.
      """
      project_id, namespace, path = entity_key.split(dbconstants.KEY_DELIMITER)
  
      group = entity_pb.Reference()
      group.set_app(project_id)
      if namespace:
        group.set_name_space(namespace)
  
      mutable_path = group.mutable_path()
      first_element = mutable_path.add_element()
      kind, id_ = path.split(dbconstants.KIND_SEPARATOR)[0].split(':')
      first_element.set_type(kind)
  
      # At this point, there's no way to tell if the ID was originally a name,
      # so this is a guess.
      try:
        first_element.set_id(int(id_))
      except ValueError:
        first_element.set_name(id_)
  
      return group
  
    def lock_and_delete_indexes(self, references, direction, entity_key):
      """ For a list of index entries that have the same entity, lock the entity
      and delete the indexes.
  
      Since another process can update an entity after we've determined that
      an index entry is invalid, we need to re-check the index entries after
      locking their entity key.
  
      Args:
        references: A list of references to an entity.
        direction: The direction of the index.
        entity_key: A string containing the entity key.
      """
      if direction == datastore_pb.Query_Order.ASCENDING:
        table_name = dbconstants.ASC_PROPERTY_TABLE
      else:
        table_name = dbconstants.DSC_PROPERTY_TABLE
  
      group_key = self.guess_group_from_table_key(entity_key)
      entity_lock = EntityLock(self.zoo_keeper.handle, [group_key])
      with entity_lock:
        entities = self.fetch_entity_dict_for_references(references)
  
        refs_to_delete = []
        for reference in references:
          index_elements = reference.keys()[0].split(self.ds_access._SEPARATOR)
          prop = index_elements[self.ds_access.PROP_NAME_IN_SINGLE_PROP_INDEX]
          if not self.ds_access._DatastoreDistributed__valid_index_entry(
            reference, entities, direction, prop):
            refs_to_delete.append(reference.keys()[0])
  
        logging.debug('Removing {} indexes starting with {}'.
          format(len(refs_to_delete), [refs_to_delete[0]]))
        try:
          self.db_access.batch_delete(table_name, refs_to_delete,
            column_names=dbconstants.PROPERTY_SCHEMA)
          self.index_entries_cleaned += len(refs_to_delete)
        except Exception:
          logging.exception('Unable to delete indexes')
          self.index_entries_delete_failures += 1
  
    def lock_and_delete_kind_index(self, reference):
      """ For a list of index entries that have the same entity, lock the entity
      and delete the indexes.
  
      Since another process can update an entity after we've determined that
      an index entry is invalid, we need to re-check the index entries after
      locking their entity key.
  
      Args:
        reference: A dictionary containing a kind reference.
      """
      table_name = dbconstants.APP_KIND_TABLE
      entity_key = reference.values()[0].values()[0]
  
      group_key = self.guess_group_from_table_key(entity_key)
      entity_lock = EntityLock(self.zoo_keeper.handle, [group_key])
      with entity_lock:
        entities = self.fetch_entity_dict_for_references([reference])
        if entity_key not in entities:
          index_to_delete = reference.keys()[0]
          logging.debug('Removing {}'.format([index_to_delete]))
          try:
            self.db_access.batch_delete(table_name, [index_to_delete],
              column_names=dbconstants.APP_KIND_SCHEMA)
            self.index_entries_cleaned += 1
          except dbconstants.AppScaleDBConnectionError:
            logging.exception('Unable to delete index.')
            self.index_entries_delete_failures += 1
  
    def clean_up_indexes(self, direction):
      """ Deletes invalid single property index entries.
  
      This is needed because we do not delete index entries when updating or
      deleting entities. With time, this results in queries taking an increasing
      amount of time.
  
      Args:
        direction: The direction of the index.
      """
      if direction == datastore_pb.Query_Order.ASCENDING:
        table_name = dbconstants.ASC_PROPERTY_TABLE
        task_id = self.CLEAN_ASC_INDICES_TASK
      else:
        table_name = dbconstants.DSC_PROPERTY_TABLE
        task_id = self.CLEAN_DSC_INDICES_TASK
  
      # If we have state information beyond what function to use,
      # load the last seen start key.
      if len(self.groomer_state) > 1 and self.groomer_state[0] == task_id:
        start_key = self.groomer_state[1]
      else:
        start_key = ''
      end_key = dbconstants.TERMINATING_STRING
  
      # Indicate that an index scrub has started after the journal was removed.
      if direction == datastore_pb.Query_Order.ASCENDING and not start_key:
        index_state = self.db_access.get_metadata(
          cassandra_interface.INDEX_STATE_KEY)
        if index_state is None:
          self.db_access.set_metadata(
            cassandra_interface.INDEX_STATE_KEY,
            cassandra_interface.IndexStates.SCRUB_IN_PROGRESS)
  
      while True:
        references = self.db_access.range_query(
          table_name=table_name,
          column_names=dbconstants.PROPERTY_SCHEMA,
          start_key=start_key,
          end_key=end_key,
          limit=self.BATCH_SIZE,
          start_inclusive=False,
        )
        if len(references) == 0:
          break
  
        self.index_entries_checked += len(references)
        if time.time() > self.last_logged + self.LOG_PROGRESS_FREQUENCY:
          logging.info('Checked {} index entries'
            .format(self.index_entries_checked))
          self.last_logged = time.time()
        first_ref = references[0].keys()[0]
        logging.debug('Fetched {} total refs, starting with {}, direction: {}'
          .format(self.index_entries_checked, [first_ref], direction))
  
        last_start_key = start_key
        start_key = references[-1].keys()[0]
        if start_key == last_start_key:
          raise dbconstants.AppScaleDBError(
            'An infinite loop was detected while fetching references.')
  
        entities = self.fetch_entity_dict_for_references(references)
  
        # Group invalid references by entity key so we can minimize locks.
        invalid_refs = {}
        for reference in references:
          prop_name = reference.keys()[0].split(self.ds_access._SEPARATOR)[3]
          if not self.ds_access._DatastoreDistributed__valid_index_entry(
            reference, entities, direction, prop_name):
            entity_key = reference.values()[0][self.ds_access.INDEX_REFERENCE_COLUMN]
            if entity_key not in invalid_refs:
              invalid_refs[entity_key] = []
            invalid_refs[entity_key].append(reference)
  
        for entity_key in invalid_refs:
          self.lock_and_delete_indexes(invalid_refs[entity_key], direction, entity_key)
        self.update_groomer_state([task_id, start_key])
  
    def clean_up_kind
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/groomer.py
files: import datetime
  import logging
  import os
  import random
  import re
  import sys
  import threading
  import time
  
  import appscale_datastore_batch
  import dbconstants
  import utils
  
  from appscale.common import appscale_info
  from appscale.common import constants
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.common.unpackaged import DASHBOARD_DIR
  from appscale.taskqueue.distributed_tq import TaskName
  from .cassandra_env import cassandra_interface
  from .datastore_distributed import DatastoreDistributed
  from .utils import get_composite_indexes_rows
  from .zkappscale import zktransaction as zk
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import apiproxy_stub_map
  from google.appengine.api import datastore_distributed
  from google.appengine.api.memcache import memcache_distributed
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore.datastore_query import Cursor
  from google.appengine.ext import db
  from google.appengine.ext.db import stats
  from google.appengine.ext.db import metadata
  from google.appengine.api import datastore_errors
  
  sys.path.append(os.path.join(DASHBOARD_DIR, 'lib'))
  from dashboard_logs import RequestLogLine
  
  
  class DatastoreGroomer(threading.Thread):
    """ Scans the entire database for each application. """
  
    # The amount of seconds between polling to get the groomer lock.
    # Each datastore server does this poll, so it happens the number
    # of datastore servers within this lock period.
    LOCK_POLL_PERIOD = 4 * 60 * 60 # <- 4 hours
  
    # Retry sleep on datastore error in seconds.
    DB_ERROR_PERIOD = 30
  
    # The number of entities retrieved in a datastore request.
    BATCH_SIZE = 100
  
    # Any kind that is of __*__ is private and should not have stats.
    PRIVATE_KINDS = '__(.*)__'
  
    # Any kind that is of _*_ is protected and should not have stats.
    PROTECTED_KINDS = '_(.*)_'
  
    # The amount of time in seconds before we want to clean up task name holders.
    TASK_NAME_TIMEOUT = 24 * 60 * 60
  
    # The amount of time before logs are considered too old.
    LOG_STORAGE_TIMEOUT = 24 * 60 * 60 * 7
  
    # Do not generate stats for AppScale internal apps.
    APPSCALE_APPLICATIONS = ['apichecker', 'appscaledashboard']
  
    # A sentinel value to signify that this app does not have composite indexes.
    NO_COMPOSITES = "NO_COMPS_INDEXES_HERE"
  
    # The path in ZooKeeper where the groomer state is stored.
    GROOMER_STATE_PATH = '/appscale/groomer_state'
  
    # The characters used to separate values when storing the groomer state.
    GROOMER_STATE_DELIMITER = '||'
  
    # The ID for the task to clean up entities.
    CLEAN_ENTITIES_TASK = 'entities'
  
    # The ID for the task to clean up ascending indices.
    CLEAN_ASC_INDICES_TASK = 'asc-indices'
  
    # The ID for the task to clean up descending indices.
    CLEAN_DSC_INDICES_TASK = 'dsc-indices'
  
    # The ID for the task to clean up kind indices.
    CLEAN_KIND_INDICES_TASK = 'kind-indices'
  
    # The ID for the task to clean up old logs.
    CLEAN_LOGS_TASK = 'logs'
  
    # The ID for the task to clean up old tasks.
    CLEAN_TASKS_TASK = 'tasks'
  
    # Log progress every time this many seconds have passed.
    LOG_PROGRESS_FREQUENCY = 60 * 5
  
    def __init__(self, zoo_keeper, table_name, ds_path):
      """ Constructor.
  
      Args:
        zk: ZooKeeper client.
        table_name: The database used (ie, cassandra)
        ds_path: The connection path to the datastore_server.
      """
      logging.info("Logging started")
  
      threading.Thread.__init__(self)
      self.zoo_keeper = zoo_keeper
      self.table_name = table_name
      self.db_access = None
      self.ds_access = None
      self.datastore_path = ds_path
      self.stats = {}
      self.namespace_info = {}
      self.num_deletes = 0
      self.composite_index_cache = {}
      self.entities_checked = 0
      self.journal_entries_cleaned = 0
      self.index_entries_checked = 0
      self.index_entries_delete_failures = 0
      self.index_entries_cleaned = 0
      self.last_logged = time.time()
      self.groomer_state = []
  
    def stop(self):
      """ Stops the groomer thread. """
      self.zoo_keeper.close()
  
    def run(self):
      """ Starts the main loop of the groomer thread. """
      while True:
  
        logging.debug("Trying to get groomer lock.")
        if self.get_groomer_lock():
          logging.info("Got the groomer lock.")
          self.run_groomer()
          try:
            self.zoo_keeper.release_lock_with_path(zk.DS_GROOM_LOCK_PATH)
          except zk.ZKTransactionException, zk_exception:
            logging.error("Unable to release zk lock {0}.".\
              format(str(zk_exception)))
          except zk.ZKInternalException, zk_exception:
            logging.error("Unable to release zk lock {0}.".\
              format(str(zk_exception)))
        else:
          logging.info("Did not get the groomer lock.")
        sleep_time = random.randint(1, self.LOCK_POLL_PERIOD)
        logging.info('Sleeping for {:.1f} minutes.'.format(sleep_time/60.0))
        time.sleep(sleep_time)
  
    def get_groomer_lock(self):
      """ Tries to acquire the lock to the datastore groomer.
  
      Returns:
        True on success, False otherwise.
      """
      return self.zoo_keeper.get_lock_with_path(zk.DS_GROOM_LOCK_PATH)
  
    def get_entity_batch(self, last_key):
      """ Gets a batch of entites to operate on.
  
      Args:
        last_key: The last key from a previous query.
      Returns:
        A list of entities.
      """
      return self.db_access.range_query(dbconstants.APP_ENTITY_TABLE,
        dbconstants.APP_ENTITY_SCHEMA, last_key, "", self.BATCH_SIZE,
        start_inclusive=False)
  
    def reset_statistics(self):
      """ Reinitializes statistics. """
      self.stats = {}
      self.namespace_info = {}
      self.num_deletes = 0
      self.journal_entries_cleaned = 0
  
    def hard_delete_row(self, row_key):
      """ Does a hard delete on a given row key to the entity
          table.
  
      Args:
        row_key: A str representing the row key to delete.
      Returns:
        True on success, False otherwise.
      """
      try:
        self.db_access.batch_delete(dbconstants.APP_ENTITY_TABLE,
          [row_key])
      except dbconstants.AppScaleDBConnectionError, db_error:
        logging.error("Error hard deleting key {0}-->{1}".format(
          row_key, db_error))
        return False
      except Exception, exception:
        logging.error("Caught unexcepted exception {0}".format(exception))
        return False
  
      return True
  
    def load_composite_cache(self, app_id):
      """ Load the composite index cache for an application ID.
  
      Args:
        app_id: A str, the application ID.
      Returns:
        True if the application has composites. False otherwise.
      """
      start_key = dbconstants.KEY_DELIMITER.join([app_id, 'index', ''])
      end_key = dbconstants.KEY_DELIMITER.join(
        [app_id, 'index', dbconstants.TERMINATING_STRING])
  
      results = self.db_access.range_query(dbconstants.METADATA_TABLE,
        dbconstants.METADATA_TABLE, start_key, end_key,
        dbconstants.MAX_NUMBER_OF_COMPOSITE_INDEXES)
      list_result = []
      for list_item in results:
        for _, value in list_item.iteritems():
          list_result.append(value['data'])
  
      self.composite_index_cache[app_id] = self.NO_COMPOSITES
      kind_index_dictionary = {}
      for index in list_result:
        new_index = entity_pb.CompositeIndex()
        new_index.ParseFromString(index)
        kind = new_index.definition().entity_type()
        if kind in kind_index_dictionary:
          kind_index_dictionary[kind].append(new_index)
        else:
          kind_index_dictionary[kind] = [new_index]
      if kind_index_dictionary:
        self.composite_index_cache[app_id] = kind_index_dictionary
        return True
  
      return False
  
    def acquire_lock_for_key(self, app_id, key, retries, retry_time):
      """ Acquires a lock for a given entity key.
  
      Args:
        app_id: The application ID.
        key: A string containing an entity key.
        retries: An integer specifying the number of times to retry.
        retry_time: How many seconds to wait before each retry.
      Returns:
        A transaction ID.
      Raises:
        ZKTransactionException if unable to acquire a lock from ZooKeeper.
      """
      root_key = key.split(dbconstants.KIND_SEPARATOR)[0]
      root_key += dbconstants.KIND_SEPARATOR
  
      txn_id = self.zoo_keeper.get_transaction_id(app_id, is_xg=False)
      try:
        self.zoo_keeper.acquire_lock(app_id, txn_id, root_key)
      except zk.ZKTransactionException as zkte:
        logging.warning('Concurrent transaction exception for app id {} with '
          'info {}'.format(app_id, str(zkte)))
        if retries > 0:
          logging.info('Trying again to acquire lock info {} with retry #{}'
            .format(str(zkte), retries))
          time.sleep(retry_time)
          return self.acquire_lock_for_key(
            app_id=app_id,
            key=key,
            retries=retries-1,
            retry_time=retry_time
          )
        self.zoo_keeper.notify_failed_transaction(app_id, txn_id)
        raise zkte
      return txn_id
  
    def release_lock_for_key(self, app_id, key, txn_id, retries, retry_time):
      """ Releases a lock for a given entity key.
  
      Args:
        app_id: The application ID.
        key: A string containing an entity key.
        txn_id: A transaction ID.
        retries: An integer specifying the number of times to retry.
        retry_time: How many seconds to wait before each retry.
      """
      root_key = key.split(dbconstants.KIND_SEPARATOR)[0]
      root_key += dbconstants.KIND_SEPARATOR
  
      try:
        self.zoo_keeper.release_lock(app_id, txn_id)
      except zk.ZKTransactionException as zkte:
        logging.warning(str(zkte))
        if retries > 0:
          logging.info('Trying again to release lock {} with retry #{}'.
            format(txn_id, retries))
          time.sleep(retry_time)
          self.release_lock_for_key(
            app_id=app_id,
            key=key,
            txn_id=txn_id,
            retries=retries-1,
            retry_time=retry_time
          )
        else:
          self.zoo_keeper.notify_failed_transaction(app_id, txn_id)
  
    def fetch_entity_dict_for_references(self, references):
      """ Fetches a dictionary of valid entities for a list of references.
  
      Args:
        references: A list of index references to entities.
      Returns:
        A dictionary of validated entities.
      """
      keys = []
      for item in references:
        keys.append(item.values()[0][self.ds_access.INDEX_REFERENCE_COLUMN])
      keys = list(set(keys))
      entities = self.db_access.batch_get_entity(dbconstants.APP_ENTITY_TABLE,
        keys, dbconstants.APP_ENTITY_SCHEMA)
  
      # The datastore needs to know the app ID. The indices could be scattered
      # across apps.
      entities_by_app = {}
      for key in entities:
        app = key.split(self.ds_access._SEPARATOR)[0]
        if app not in entities_by_app:
          entities_by_app[app] = {}
        entities_by_app[app][key] = entities[key]
  
      entities = {}
      for app in entities_by_app:
        app_entities = entities_by_app[app]
        for key in keys:
          if key not in app_entities:
            continue
          if dbconstants.APP_ENTITY_SCHEMA[0] not in app_entities[key]:
            continue
          entities[key] = app_entities[key][dbconstants.APP_ENTITY_SCHEMA[0]]
      return entities
  
    def lock_and_delete_indexes(self, references, direction, entity_key):
      """ For a list of index entries that have the same entity, lock the entity
      and delete the indexes.
  
      Since another process can update an entity after we've determined that
      an index entry is invalid, we need to re-check the index entries after
      locking their entity key.
  
      Args:
        references: A list of references to an entity.
        direction: The direction of the index.
        entity_key: A string containing the entity key.
      """
      if direction == datastore_pb.Query_Order.ASCENDING:
        table_name = dbconstants.ASC_PROPERTY_TABLE
      else:
        table_name = dbconstants.DSC_PROPERTY_TABLE
  
      app = entity_key.split(self.ds_access._SEPARATOR)[0]
      try:
        txn_id = self.acquire_lock_for_key(
          app_id=app,
          key=entity_key,
          retries=self.ds_access.NON_TRANS_LOCK_RETRY_COUNT,
          retry_time=self.ds_access.LOCK_RETRY_TIME
        )
      except zk.ZKTransactionException:
        self.index_entries_delete_failures += 1
        return
  
      entities = self.fetch_entity_dict_for_references(references)
  
      refs_to_delete = []
      for reference in references:
        index_elements = reference.keys()[0].split(self.ds_access._SEPARATOR)
        prop_name = index_elements[self.ds_access.PROP_NAME_IN_SINGLE_PROP_INDEX]
        if not self.ds_access._DatastoreDistributed__valid_index_entry(
          reference, entities, direction, prop_name):
          refs_to_delete.append(reference.keys()[0])
  
      logging.debug('Removing {} indexes starting with {}'.
        format(len(refs_to_delete), [refs_to_delete[0]]))
      try:
        self.db_access.batch_delete(table_name, refs_to_delete,
          column_names=dbconstants.PROPERTY_SCHEMA)
        self.index_entries_cleaned += len(refs_to_delete)
      except Exception:
        logging.exception('Unable to delete indexes')
        self.index_entries_delete_failures += 1
  
      self.release_lock_for_key(
        app_id=app,
        key=entity_key,
        txn_id=txn_id,
        retries=self.ds_access.NON_TRANS_LOCK_RETRY_COUNT,
        retry_time=self.ds_access.LOCK_RETRY_TIME
      )
  
    def lock_and_delete_kind_index(self, reference):
      """ For a list of index entries that have the same entity, lock the entity
      and delete the indexes.
  
      Since another process can update an entity after we've determined that
      an index entry is invalid, we need to re-check the index entries after
      locking their entity key.
  
      Args:
        reference: A dictionary containing a kind reference.
      """
      table_name = dbconstants.APP_KIND_TABLE
      entity_key = reference.values()[0].values()[0]
      app = entity_key.split(self.ds_access._SEPARATOR)[0]
      try:
        txn_id = self.acquire_lock_for_key(
          app_id=app,
          key=entity_key,
          retries=self.ds_access.NON_TRANS_LOCK_RETRY_COUNT,
          retry_time=self.ds_access.LOCK_RETRY_TIME
        )
      except zk.ZKTransactionException:
        self.index_entries_delete_failures += 1
        return
  
      entities = self.fetch_entity_dict_for_references([reference])
      if entity_key not in entities:
        index_to_delete = reference.keys()[0]
        logging.debug('Removing {}'.format([index_to_delete]))
        try:
          self.db_access.batch_delete(table_name, [index_to_delete],
            column_names=dbconstants.APP_KIND_SCHEMA)
          self.index_entries_cleaned += 1
        except dbconstants.AppScaleDBConnectionError:
          logging.exception('Unable to delete index.')
          self.index_entries_delete_failures += 1
  
      self.release_lock_for_key(
        app_id=app,
        key=entity_key,
        txn_id=txn_id,
        retries=self.ds_access.NON_TRANS_LOCK_RETRY_COUNT,
        retry_time=self.ds_access.LOCK_RETRY_TIME
      )
  
    def clean_up_indexes(self, direction):
      """ Deletes invalid single property index entries.
  
      This is needed because we do not delete index entries when updating or
      deleting entities. With time, this results in queries taking an increasing
      amount of time.
  
      Args:
        direction: The direction of the index.
      """
      if direction == datastore_pb.Query_Order.ASCENDING:
        table_name = db
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/groomer.py