files: import array
  import datetime
  import itertools
  import logging
  import md5
  import random
  import sys
  import time
  
  from tornado import gen
  from tornado.ioloop import IOLoop
  
  from appscale.datastore import dbconstants, helper_functions
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from appscale.datastore.dbconstants import (
    APP_ENTITY_SCHEMA, ID_KEY_LENGTH, InternalError,
    MAX_TX_DURATION, Timeout
  )
  from appscale.datastore.cassandra_env.entity_id_allocator import EntityIDAllocator
  from appscale.datastore.cassandra_env.entity_id_allocator import ScatteredAllocator
  from appscale.datastore.cassandra_env.utils import deletions_for_entity
  from appscale.datastore.cassandra_env.utils import mutations_for_entity
  from appscale.datastore.taskqueue_client import EnqueueError, TaskQueueClient
  from appscale.datastore.utils import clean_app_id
  from appscale.datastore.utils import encode_entity_table_key
  from appscale.datastore.utils import encode_index_pb
  from appscale.datastore.utils import encode_path_from_filter
  from appscale.datastore.utils import get_composite_index_keys
  from appscale.datastore.utils import get_entity_key
  from appscale.datastore.utils import get_entity_kind
  from appscale.datastore.utils import get_index_key_from_params
  from appscale.datastore.utils import get_kind_key
  from appscale.datastore.utils import group_for_key
  from appscale.datastore.utils import reference_property_to_reference
  from appscale.datastore.utils import UnprocessedQueryCursor
  from appscale.datastore.zkappscale import entity_lock
  from appscale.datastore.zkappscale import zktransaction
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import datastore_errors
  from google.appengine.api.datastore_distributed import _MAX_ACTIONS_PER_TXN
  from google.appengine.datastore import appscale_stub_util
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_index
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import sortable_pb_encoder
  from google.appengine.datastore.datastore_stub_util import IdToCounter
  from google.appengine.datastore.datastore_stub_util import SEQUENTIAL
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext import db
  from google.appengine.ext.db.metadata import Namespace
  from google.net.proto.ProtocolBuffer import ProtocolBufferDecodeError
  
  
  class DatastoreDistributed():
    """ AppScale persistent layer for the datastore API. It is the
        replacement for the AppServers to persist their data into
        a distributed datastore instead of a flat file.
    """
    # Max number of results for a query
    _MAXIMUM_RESULTS = 10000
  
    # The number of entries looked at when doing a composite query
    # It will keep looking at this size window when getting the result
    _MAX_COMPOSITE_WINDOW = 10000
  
    # Maximum amount of filter and orderings allowed within a query
    _MAX_QUERY_COMPONENTS = 63
  
    # For enabling and disabling range inclusivity
    _ENABLE_INCLUSIVITY = True
    _DISABLE_INCLUSIVITY = False
  
    # Delimiter between app names and namespace and the rest of an entity key
    _NAMESPACE_SEPARATOR = dbconstants.KEY_DELIMITER
  
    # Delimiter between parameters in index keys.
    _SEPARATOR = dbconstants.KEY_DELIMITER
  
    # This is the terminating string for range queries
    _TERM_STRING = dbconstants.TERMINATING_STRING
  
    # Smallest possible value that is considered non-null and indexable.
    MIN_INDEX_VALUE = '\x01'
  
    # When assigning the first allocated ID, give this value
    _FIRST_VALID_ALLOCATED_ID = 1
  
    # The key we use to lock for allocating new IDs
    _ALLOCATE_ROOT_KEY = "__allocate__"
  
    # Number of times to retry acquiring a lock for non transactions.
    NON_TRANS_LOCK_RETRY_COUNT = 5
  
    # How long to wait before retrying to grab a lock
    LOCK_RETRY_TIME = .5
  
    # Maximum number of allowed composite indexes any one application can
    # register.
    _MAX_NUM_INDEXES = dbconstants.MAX_NUMBER_OF_COMPOSITE_INDEXES
  
    # The position of the prop name when splitting an index entry by the
    # delimiter.
    PROP_NAME_IN_SINGLE_PROP_INDEX = 3
  
    # The cassandra index column that stores the reference to the entity.
    INDEX_REFERENCE_COLUMN = 'reference'
  
    # The number of entities to fetch at a time when updating indices.
    BATCH_SIZE = 100
  
    def __init__(self, datastore_batch, transaction_manager, zookeeper=None,
                 log_level=logging.INFO, taskqueue_locations=()):
      """
         Constructor.
  
       Args:
         datastore_batch: A reference to the batch datastore interface.
         zookeeper: A reference to the zookeeper interface.
      """
      class_name = self.__class__.__name__
      self.logger = logging.getLogger(class_name)
      self.logger.setLevel(log_level)
  
      assert datastore_batch.valid_data_version_sync()
  
      self.logger.info('Starting {}'.format(class_name))
  
      # datastore accessor used by this class to do datastore operations.
      self.datastore_batch = datastore_batch
  
      # zookeeper instance for accesing ZK functionality.
      self.zookeeper = zookeeper
  
      # Maintain a scattered allocator for each project.
      self.scattered_allocators = {}
  
      # Maintain a sequential allocator for each project.
      self.sequential_allocators = {}
  
      self.taskqueue_client = TaskQueueClient(taskqueue_locations)
      self.transaction_manager = transaction_manager
      self.zookeeper.handle.add_listener(self._zk_state_listener)
  
    def get_limit(self, query):
      """ Returns the limit that should be used for the given query.
  
      Args:
        query: A datastore_pb.Query.
      Returns:
        An int, the limit to be used when accessing the datastore.
      """
      limit = self._MAXIMUM_RESULTS
      if query.has_count():
        limit = min(query.count(), self._MAXIMUM_RESULTS)
  
      if query.has_limit():
        limit = min(query.limit(), limit)
  
      if query.has_offset():
        limit = limit + min(query.offset(), self._MAXIMUM_RESULTS)
      # We can not scan with 0 or less, hence we set it to one.
      if limit <= 0:
        limit = 1
      return limit
  
    @staticmethod
    def __decode_index_str(value, prop_value):
      """ Takes an encoded string and converts it to a PropertyValue.
  
      Args:
        value: An encoded str.
        prop_value: PropertyValue to fill in.
      """
      value = str(value).replace('\x01\x01', '\x00').replace('\x01\x02', '\x01')
      decoded_value = sortable_pb_encoder.Decoder(
        array.array('B', str(value)))
      prop_value.Merge(decoded_value)
  
    @staticmethod
    def validate_app_id(app_id):
      """ Verify that this is the stub for app_id.
  
      Args:
        app_id: An application ID.
      Raises:
        AppScaleBadArg: If the application id is not set.
      """
      if not app_id:
        raise dbconstants.AppScaleBadArg("Application name must be set")
  
    @staticmethod
    def validate_key(key):
      """ Validate this key by checking to see if it has a name or id.
  
      Args:
        key: entity_pb.Reference
      Raises:
        datastore_errors.BadRequestError: if the key is invalid
        TypeError: if key is not of entity_pb.Reference
      """
  
      if not isinstance(key, entity_pb.Reference):
        raise TypeError("Expected type Reference")
  
      DatastoreDistributed.validate_app_id(key.app())
  
      for elem in key.path().element_list():
        if elem.has_id() and elem.has_name():
          raise datastore_errors.BadRequestError(
              'Each key path element should have id or name but not both: {0}' \
              .format(key))
  
    def get_table_prefix(self, data):
      """ Returns the namespace prefix for a query.
  
      Args:
        data: An Entity, Key or Query PB, or an (app_id, ns) tuple.
      Returns:
        A valid table prefix.
      """
      if isinstance(data, entity_pb.EntityProto):
        app_id = clean_app_id(data.key().app())
        namespace = data.key().name_space()
      elif isinstance(data, tuple):
        app_id = data[0]
        namespace = data[1]
      else:
        app_id = clean_app_id(data.app())
        namespace = data.name_space()
  
      return self._SEPARATOR.join([app_id, namespace])
  
    @staticmethod
    def get_ancestor_key_from_ent_key(ent_key):
      """ Get the key string for the ancestor portion of a composite key.
  
      Args:
        ent_key: A string of the entire path of an entity.
      Returns:
        A str of the path of the ancestor.
      """
      ancestor = ""
      tokens = str(ent_key).split(dbconstants.KIND_SEPARATOR)
      # Strip off the empty placeholder and also do not include the last kind.
      for token in tokens[:-2]:
        ancestor += token + dbconstants.KIND_SEPARATOR
      return ancestor
  
    @staticmethod
    def get_composite_index_key(index, entity, position_list=None,
      filters=None):
      """ Creates a key to the composite index table for a given entity
      for a composite cursor.
  
      Keys are built as such:
        app_id/ns/composite_id/ancestor/valuevaluevalue..../entity_key
      Components explained:
      ns: The namespace of the entity.
      composite_id: The composite ID assigned to this index upon creation.
      ancestor: The root ancestor path (only if the query this index is for
        has an ancestor)
      value(s): The string representation of mulitiple properties.
      entity_key: The entity key (full path) used as a means of having a unique
        identifier. This prevents two entities with the same values from
        colliding.
  
      Args:
        index: A datstore_pb.CompositeIndex.
        entity: A entity_pb.EntityProto.
        position_list: A list of datastore_pb.CompiledCursor_Position items.
          Contains values for property items from a cursor.
        filters: A list of datastore_pb.Query_Filters, used to attain equality
          values not present in position_list.
      Returns:
        A string representing a key to the composite table.
      """
      composite_id = index.id()
      definition = index.definition()
      app_id = clean_app_id(entity.key().app())
      name_space = entity.key().name_space()
      ent_key = encode_index_pb(entity.key().path())
      pre_comp_index_key = "{0}{1}{2}{4}{3}{4}".format(app_id,
        DatastoreDistributed._NAMESPACE_SEPARATOR, name_space, composite_id,
        DatastoreDistributed._SEPARATOR)
      if definition.ancestor() == 1:
        ancestor = DatastoreDistributed.get_ancestor_key_from_ent_key(ent_key)
        pre_comp_index_key += "{0}{1}".format(ancestor,
          DatastoreDistributed._SEPARATOR)
  
      value_dict = {}
      for prop in entity.property_list():
        value_dict[prop.name()] = \
          str(encode_index_pb(prop.value()))
  
      # Position list and filters are used if we're creating a composite
      # key for a cursor.
      if position_list:
        for indexvalue in position_list[0].indexvalue_list():
          value_dict[indexvalue.property()] = \
            str(encode_index_pb(indexvalue.value()))
      if filters:
        for filt in filters:
          if filt.op() == datastore_pb.Query_Filter.EQUAL:
            value_dict[filt.property(0).name()] = \
              str(encode_index_pb(filt.property(0).value()))
  
      index_value = ""
      for prop in definition.property_list():
        name = prop.name()
        value = ''
        if name in value_dict:
          value = value_dict[name]
        elif name == "__key__":
          value = ent_key
        else:
          logging.warning("Given entity {0} is missing a property value {1}.".\
            format(entity, prop.name()))
        if prop.direction() == entity_pb.Index_Property.DESCENDING:
          value = helper_functions.reverse_lex(value)
  
        index_value += str(value) + DatastoreDistributed._SEPARATOR
  
      # We append the ent key to have unique keys if entities happen
      # to share the same index values (and ancestor).
      composite_key = "{0}{1}{2}".format(pre_comp_index_key, index_value,
        ent_key)
      return composite_key
  
    @gen.coroutine
    def insert_composite_indexes(self, entities, composite_indexes):
      """ Creates composite indexes for a set of entities.
  
      Args:
        entities: A list entities.
        composite_indexes: A list of datastore_pb.CompositeIndex.
      """
      if not composite_indexes:
        return
      row_keys = []
      row_values = {}
      # Create default composite index for all entities. Here we take each
      # of the properties in one
      for ent in entities:
        for index_def in composite_indexes:
          # Skip any indexes if the kind does not match.
          kind = get_entity_kind(ent.key())
          if index_def.definition().entity_type() != kind:
            continue
  
          # Make sure the entity contains the required entities for the composite
          # definition.
          prop_name_def_list = [index_prop.name() for index_prop in \
            index_def.definition().property_list()]
          all_prop_names_in_ent = [prop.name() for prop in \
            ent.property_list()]
          has_values = True
          for index_prop_name in prop_name_def_list:
            if index_prop_name not in all_prop_names_in_ent:
              has_values = False
            # Special property name which does not show up in the list but
            # is apart of the key of the entity.
            if index_prop_name == "__key__":
              has_values = True
          if not has_values:
            continue
  
          # Get the composite index key.
          composite_index_keys = get_composite_index_keys(index_def, ent)
          row_keys.extend(composite_index_keys)
  
          # Get the reference value for the composite table.
          entity_key = str(encode_index_pb(ent.key().path()))
          prefix = self.get_table_prefix(ent.key())
          reference = "{0}{1}{2}".format(prefix, self._SEPARATOR,  entity_key)
          for composite_key in composite_index_keys:
            row_values[composite_key] = {'reference': reference}
  
      yield self.datastore_batch.batch_put_entity(
        dbconstants.COMPOSITE_TABLE, row_keys,
        dbconstants.COMPOSITE_SCHEMA, row_values)
  
    @gen.coroutine
    def delete_composite_index_metadata(self, app_id, index):
      """ Deletes a index for the given application identifier.
  
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      """
      self.logger.info('Deleting composite index:\n{}'.format(index))
      index_keys = []
      composite_id = str(index.id())
      index_keys.append(self._SEPARATOR.join([app_id, 'index', composite_id]))
      yield self.datastore_batch.batch_delete(
        dbconstants.METADATA_TABLE, index_keys,
        column_names=dbconstants.METADATA_TABLE)
  
    @gen.coroutine
    def create_composite_index(self, app_id, index):
      """ Stores a new index for the given application identifier.
  
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      Returns:
        A unique number representing the composite index ID.
      """
      # Generate a random number based on time of creation.
      rand = int(str(int(time.time())) + str(random.randint(0, 999999)))
      index.set_id(rand)
      encoded_entity = index.Encode()
      row_key = self._SEPARATOR.join([app_id, 'index', str(rand)])
      row_keys = [row_key]
      row_values = {}
      row_values[row_key] = {dbconstants.METADATA_SCHEMA[0]: encoded_entity}
      yield self.datastore_batch.batch_put_entity(
        dbconstants.METADATA_TABLE, row_keys,
        dbconstants.METADATA_SCHEMA, row_values)
      raise gen.Return(rand)
  
    @gen.coroutine
    def update_composite_index(self, app_id, index):
      """ Updates an index for a given app ID.
  
      Args:
        app_id: A string containing the app ID.
        index: An entity_pb.CompositeIndex object.
      """
      self.logger.info('Updating index: {}'.format(index))
      entries_updated = 0
      entity_type = index.definition().entity_type()
  
      # TODO: Adjust prefix based on ancestor.
      prefix = '{app}{delimiter}{entity_type}{kind_separator}'.format(
        app=app_id,
        delimiter=self._SEPARATOR * 2,
        entity_type=entity_type,
        kind_separator=dbconstants.KIND_SEPARATOR,
      )
      start_row = prefix
      end_row = prefix + self._TERM_STRING
      start_inclusive = True
  
      while True:
        # Fetch references from the kind table since entity keys can have a
        # parent prefix.
        references = yield self.datastore_batch.range_query(
          table_name=dbconstants.APP_KIND_TABLE,
          column_names=dbconstants.APP_KIND_SCHEMA,
          start_key=start_row,
          end_key=end_row,
          limit=self.BATCH_SIZE,
          offset=0,
          start_inclusive=start_inclusive,
        )
  
        pb_entities = yield self.__fetch_entities(references)
        entities = [entity_pb.EntityProto(entity) for entity in pb_entities]
  
        yield self.insert_composite_indexes(entities, [index])
        entries_updated += len(entities)
  
        # If we fetched fewer references than we asked for, we're done.
        if len(references) < self.BATCH_SIZE:
          break
  
        start_row = references[-1].keys()[0]
        start_inclusive = self._DISABLE_INCLUSIVITY
  
      self.logger.info('Updated {} index entries.'.format(entries_updated))
  
    @gen.coroutine
    def allocate_size(self, project, size):
      """ Allocates a block of IDs for a project.
  
      Args:
        project: A string specifying the project ID.
        size: An integer specifying the number of IDs to reserve.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      start_id, end_id = yield allocator.allocate_size(size)
      raise gen.Return((start_id, end_id))
  
    @gen.coroutine
    def allocate_max(self, project, max_id):
      """ Reserves all IDs up to the one given.
  
      Args:
        project: A string specifying the project ID.
        max_id: An integer specifying the maximum ID to allocated.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      start_id, end_id = yield allocator.allocate_max(max_id)
      raise gen.Return((start_id, end_id))
  
    @gen.coroutine
    def reserve_ids(self, project_id, ids):
      """ Ensures the given IDs are not re-allocated.
  
      Args:
        project_id: A string specifying the project ID.
        ids: An iterable of integers specifying entity IDs.
      """
      if project_id not in self.sequential_allocators:
        self.sequential_allocators[project_id] = EntityIDAllocator(
          self.datastore_batch.session, project_id)
  
      if project_id not in self.scattered_allocators:
        self.scattered_allocators[project_id] = ScatteredAllocator(
          self.datastore_batch.session, project_id)
  
      for id_ in ids:
        counter, space = IdToCounter(id_)
        if space == SEQUENTIAL:
          allocator = self.sequential_allocators[project_id]
        else:
          allocator = self.scattered_allocators[project_id]
  
        yield allocator.set_min_counter(counter)
  
    @gen.coroutine
    def put_entities(self, app, entities, composite_indexes=()):
      """ Updates indexes of existing entities, inserts new entities and
          indexes for them.
  
      Args:
        app: A string containing the application ID.
        entities: List of entities.
        composite_indexes: A list or tuple of CompositeIndex objects.
      """
      self.logger.debug('Inserting {} entities'.format(len(entities)))
  
      by_group = {}
      for entity in entities:
        group_key = group_for_key(entity.key()).Encode()
        if group_key not in by_group:
          by_group[group_key] = []
        by_group[group_key].append(entity)
  
      for encoded_group_key, entity_list in by_group.iteritems():
        group_key = entity_pb.Reference(encoded_group_key)
  
        txid = self.transaction_manager.create_transaction_id(app, xg=False)
        self.transaction_manager.set_groups(app, txid, [group_key])
  
        # Allow the lock to stick around if there is an issue applying the batch.
        lock = entity_lock.EntityLock(self.zookeeper.handle, [group_key], txid)
        try:
          yield lock.acquire()
        except entity_lock.LockTimeout:
          raise Timeout('Unable to acquire entity group lock')
  
        try:
          entity_keys = [
            get_entity_key(self.get_table_prefix(entity), entity.key().path())
            for entity in entity_list]
          current_values = yield self.datastore_batch.batch_get_entity(
            dbconstants.APP_ENTITY_TABLE, entity_keys, APP_ENTITY_SCHEMA)
  
          batch = []
          entity_changes = []
          for entity in entity_list:
            prefix = self.get_table_prefix(entity)
            entity_key = get_entity_key(prefix, entity.key().path())
  
            current_value = None
            if current_values[entity_key]:
              current_value = entity_pb.EntityProto(
                current_values[entity_key][APP_ENTITY_SCHEMA[0]])
  
            batch.extend(mutations_for_entity(entity, txid, current_value,
                                              composite_indexes))
  
            batch.append({'table': 'group_updates',
                          'key': bytearray(encoded_group_key),
                          'last_update': txid})
  
            entity_changes.append(
              {'key': entity.key(), 'old': current_value, 'new': entity})
          yield self.datastore_batch.batch_mutate(
            app, batch, entity_changes, txid)
          lock.release()
  
        finally:
          # In case of failure entity group lock should stay acquired
          # as transaction groomer will handle it later.
          # But tornado lock must be released.
          lock
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/datastore_distributed.py
files: import array
  import datetime
  import itertools
  import logging
  import md5
  import random
  import sys
  import threading
  import time
  
  from tornado import gen
  from tornado.ioloop import IOLoop
  
  from appscale.datastore import dbconstants, helper_functions
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from appscale.datastore.dbconstants import (
    APP_ENTITY_SCHEMA, ID_KEY_LENGTH, MAX_TX_DURATION, Timeout
  )
  from appscale.datastore.cassandra_env.entity_id_allocator import EntityIDAllocator
  from appscale.datastore.cassandra_env.entity_id_allocator import ScatteredAllocator
  from appscale.datastore.cassandra_env.utils import deletions_for_entity
  from appscale.datastore.cassandra_env.utils import mutations_for_entity
  from appscale.datastore.taskqueue_client import EnqueueError, TaskQueueClient
  from appscale.datastore.utils import clean_app_id
  from appscale.datastore.utils import encode_entity_table_key
  from appscale.datastore.utils import encode_index_pb
  from appscale.datastore.utils import encode_path_from_filter
  from appscale.datastore.utils import get_composite_index_keys
  from appscale.datastore.utils import get_entity_key
  from appscale.datastore.utils import get_entity_kind
  from appscale.datastore.utils import get_index_key_from_params
  from appscale.datastore.utils import get_kind_key
  from appscale.datastore.utils import group_for_key
  from appscale.datastore.utils import reference_property_to_reference
  from appscale.datastore.utils import UnprocessedQueryCursor
  from appscale.datastore.zkappscale import entity_lock
  from appscale.datastore.zkappscale import zktransaction
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.api import datastore_errors
  from google.appengine.api.datastore_distributed import _MAX_ACTIONS_PER_TXN
  from google.appengine.datastore import appscale_stub_util
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_index
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import sortable_pb_encoder
  from google.appengine.datastore.datastore_stub_util import IdToCounter
  from google.appengine.datastore.datastore_stub_util import SEQUENTIAL
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext import db
  from google.appengine.ext.db.metadata import Namespace
  from google.net.proto.ProtocolBuffer import ProtocolBufferDecodeError
  
  
  class DatastoreDistributed():
    """ AppScale persistent layer for the datastore API. It is the
        replacement for the AppServers to persist their data into
        a distributed datastore instead of a flat file.
    """
    # Max number of results for a query
    _MAXIMUM_RESULTS = 10000
  
    # The number of entries looked at when doing a composite query
    # It will keep looking at this size window when getting the result
    _MAX_COMPOSITE_WINDOW = 10000
  
    # Maximum amount of filter and orderings allowed within a query
    _MAX_QUERY_COMPONENTS = 63
  
    # For enabling and disabling range inclusivity
    _ENABLE_INCLUSIVITY = True
    _DISABLE_INCLUSIVITY = False
  
    # Delimiter between app names and namespace and the rest of an entity key
    _NAMESPACE_SEPARATOR = dbconstants.KEY_DELIMITER
  
    # Delimiter between parameters in index keys.
    _SEPARATOR = dbconstants.KEY_DELIMITER
  
    # This is the terminating string for range queries
    _TERM_STRING = dbconstants.TERMINATING_STRING
  
    # Smallest possible value that is considered non-null and indexable.
    MIN_INDEX_VALUE = '\x01'
  
    # When assigning the first allocated ID, give this value
    _FIRST_VALID_ALLOCATED_ID = 1
  
    # The key we use to lock for allocating new IDs
    _ALLOCATE_ROOT_KEY = "__allocate__"
  
    # Number of times to retry acquiring a lock for non transactions.
    NON_TRANS_LOCK_RETRY_COUNT = 5
  
    # How long to wait before retrying to grab a lock
    LOCK_RETRY_TIME = .5
  
    # Maximum number of allowed composite indexes any one application can
    # register.
    _MAX_NUM_INDEXES = dbconstants.MAX_NUMBER_OF_COMPOSITE_INDEXES
  
    # The position of the prop name when splitting an index entry by the
    # delimiter.
    PROP_NAME_IN_SINGLE_PROP_INDEX = 3
  
    # The cassandra index column that stores the reference to the entity.
    INDEX_REFERENCE_COLUMN = 'reference'
  
    # The number of entities to fetch at a time when updating indices.
    BATCH_SIZE = 100
  
    def __init__(self, datastore_batch, transaction_manager, zookeeper=None,
                 log_level=logging.INFO, taskqueue_locations=()):
      """
         Constructor.
  
       Args:
         datastore_batch: A reference to the batch datastore interface.
         zookeeper: A reference to the zookeeper interface.
      """
      class_name = self.__class__.__name__
      self.logger = logging.getLogger(class_name)
      self.logger.setLevel(log_level)
  
      assert datastore_batch.valid_data_version_sync()
  
      self.logger.info('Starting {}'.format(class_name))
  
      # datastore accessor used by this class to do datastore operations.
      self.datastore_batch = datastore_batch
  
      # zookeeper instance for accesing ZK functionality.
      self.zookeeper = zookeeper
  
      # Maintain a scattered allocator for each project.
      self.scattered_allocators = {}
  
      # Maintain a sequential allocator for each project.
      self.sequential_allocators = {}
  
      self.taskqueue_client = TaskQueueClient(taskqueue_locations)
      self.transaction_manager = transaction_manager
      self.zookeeper.handle.add_listener(self._zk_state_listener)
  
    def get_limit(self, query):
      """ Returns the limit that should be used for the given query.
  
      Args:
        query: A datastore_pb.Query.
      Returns:
        An int, the limit to be used when accessing the datastore.
      """
      limit = self._MAXIMUM_RESULTS
      if query.has_count():
        limit = min(query.count(), self._MAXIMUM_RESULTS)
  
      if query.has_limit():
        limit = min(query.limit(), limit)
  
      if query.has_offset():
        limit = limit + min(query.offset(), self._MAXIMUM_RESULTS)
      # We can not scan with 0 or less, hence we set it to one.
      if limit <= 0:
        limit = 1
      return limit
  
    @staticmethod
    def __decode_index_str(value, prop_value):
      """ Takes an encoded string and converts it to a PropertyValue.
  
      Args:
        value: An encoded str.
        prop_value: PropertyValue to fill in.
      """
      value = str(value).replace('\x01\x01', '\x00').replace('\x01\x02', '\x01')
      decoded_value = sortable_pb_encoder.Decoder(
        array.array('B', str(value)))
      prop_value.Merge(decoded_value)
  
    @staticmethod
    def validate_app_id(app_id):
      """ Verify that this is the stub for app_id.
  
      Args:
        app_id: An application ID.
      Raises:
        AppScaleBadArg: If the application id is not set.
      """
      if not app_id:
        raise dbconstants.AppScaleBadArg("Application name must be set")
  
    @staticmethod
    def validate_key(key):
      """ Validate this key by checking to see if it has a name or id.
  
      Args:
        key: entity_pb.Reference
      Raises:
        datastore_errors.BadRequestError: if the key is invalid
        TypeError: if key is not of entity_pb.Reference
      """
  
      if not isinstance(key, entity_pb.Reference):
        raise TypeError("Expected type Reference")
  
      DatastoreDistributed.validate_app_id(key.app())
  
      for elem in key.path().element_list():
        if elem.has_id() and elem.has_name():
          raise datastore_errors.BadRequestError(
              'Each key path element should have id or name but not both: {0}' \
              .format(key))
  
    def get_table_prefix(self, data):
      """ Returns the namespace prefix for a query.
  
      Args:
        data: An Entity, Key or Query PB, or an (app_id, ns) tuple.
      Returns:
        A valid table prefix.
      """
      if isinstance(data, entity_pb.EntityProto):
        app_id = clean_app_id(data.key().app())
        namespace = data.key().name_space()
      elif isinstance(data, tuple):
        app_id = data[0]
        namespace = data[1]
      else:
        app_id = clean_app_id(data.app())
        namespace = data.name_space()
  
      return self._SEPARATOR.join([app_id, namespace])
  
    @staticmethod
    def get_ancestor_key_from_ent_key(ent_key):
      """ Get the key string for the ancestor portion of a composite key.
  
      Args:
        ent_key: A string of the entire path of an entity.
      Returns:
        A str of the path of the ancestor.
      """
      ancestor = ""
      tokens = str(ent_key).split(dbconstants.KIND_SEPARATOR)
      # Strip off the empty placeholder and also do not include the last kind.
      for token in tokens[:-2]:
        ancestor += token + dbconstants.KIND_SEPARATOR
      return ancestor
  
    @staticmethod
    def get_composite_index_key(index, entity, position_list=None,
      filters=None):
      """ Creates a key to the composite index table for a given entity
      for a composite cursor.
  
      Keys are built as such:
        app_id/ns/composite_id/ancestor/valuevaluevalue..../entity_key
      Components explained:
      ns: The namespace of the entity.
      composite_id: The composite ID assigned to this index upon creation.
      ancestor: The root ancestor path (only if the query this index is for
        has an ancestor)
      value(s): The string representation of mulitiple properties.
      entity_key: The entity key (full path) used as a means of having a unique
        identifier. This prevents two entities with the same values from
        colliding.
  
      Args:
        index: A datstore_pb.CompositeIndex.
        entity: A entity_pb.EntityProto.
        position_list: A list of datastore_pb.CompiledCursor_Position items.
          Contains values for property items from a cursor.
        filters: A list of datastore_pb.Query_Filters, used to attain equality
          values not present in position_list.
      Returns:
        A string representing a key to the composite table.
      """
      composite_id = index.id()
      definition = index.definition()
      app_id = clean_app_id(entity.key().app())
      name_space = entity.key().name_space()
      ent_key = encode_index_pb(entity.key().path())
      pre_comp_index_key = "{0}{1}{2}{4}{3}{4}".format(app_id,
        DatastoreDistributed._NAMESPACE_SEPARATOR, name_space, composite_id,
        DatastoreDistributed._SEPARATOR)
      if definition.ancestor() == 1:
        ancestor = DatastoreDistributed.get_ancestor_key_from_ent_key(ent_key)
        pre_comp_index_key += "{0}{1}".format(ancestor,
          DatastoreDistributed._SEPARATOR)
  
      value_dict = {}
      for prop in entity.property_list():
        value_dict[prop.name()] = \
          str(encode_index_pb(prop.value()))
  
      # Position list and filters are used if we're creating a composite
      # key for a cursor.
      if position_list:
        for indexvalue in position_list[0].indexvalue_list():
          value_dict[indexvalue.property()] = \
            str(encode_index_pb(indexvalue.value()))
      if filters:
        for filt in filters:
          if filt.op() == datastore_pb.Query_Filter.EQUAL:
            value_dict[filt.property(0).name()] = \
              str(encode_index_pb(filt.property(0).value()))
  
      index_value = ""
      for prop in definition.property_list():
        name = prop.name()
        value = ''
        if name in value_dict:
          value = value_dict[name]
        elif name == "__key__":
          value = ent_key
        else:
          logging.warning("Given entity {0} is missing a property value {1}.".\
            format(entity, prop.name()))
        if prop.direction() == entity_pb.Index_Property.DESCENDING:
          value = helper_functions.reverse_lex(value)
  
        index_value += str(value) + DatastoreDistributed._SEPARATOR
  
      # We append the ent key to have unique keys if entities happen
      # to share the same index values (and ancestor).
      composite_key = "{0}{1}{2}".format(pre_comp_index_key, index_value,
        ent_key)
      return composite_key
  
    @gen.coroutine
    def insert_composite_indexes(self, entities, composite_indexes):
      """ Creates composite indexes for a set of entities.
  
      Args:
        entities: A list entities.
        composite_indexes: A list of datastore_pb.CompositeIndex.
      """
      if not composite_indexes:
        return
      row_keys = []
      row_values = {}
      # Create default composite index for all entities. Here we take each
      # of the properties in one
      for ent in entities:
        for index_def in composite_indexes:
          # Skip any indexes if the kind does not match.
          kind = get_entity_kind(ent.key())
          if index_def.definition().entity_type() != kind:
            continue
  
          # Make sure the entity contains the required entities for the composite
          # definition.
          prop_name_def_list = [index_prop.name() for index_prop in \
            index_def.definition().property_list()]
          all_prop_names_in_ent = [prop.name() for prop in \
            ent.property_list()]
          has_values = True
          for index_prop_name in prop_name_def_list:
            if index_prop_name not in all_prop_names_in_ent:
              has_values = False
            # Special property name which does not show up in the list but
            # is apart of the key of the entity.
            if index_prop_name == "__key__":
              has_values = True
          if not has_values:
            continue
  
          # Get the composite index key.
          composite_index_keys = get_composite_index_keys(index_def, ent)
          row_keys.extend(composite_index_keys)
  
          # Get the reference value for the composite table.
          entity_key = str(encode_index_pb(ent.key().path()))
          prefix = self.get_table_prefix(ent.key())
          reference = "{0}{1}{2}".format(prefix, self._SEPARATOR,  entity_key)
          for composite_key in composite_index_keys:
            row_values[composite_key] = {'reference': reference}
  
      yield self.datastore_batch.batch_put_entity(
        dbconstants.COMPOSITE_TABLE, row_keys,
        dbconstants.COMPOSITE_SCHEMA, row_values)
  
    @gen.coroutine
    def delete_composite_index_metadata(self, app_id, index):
      """ Deletes a index for the given application identifier.
  
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      """
      self.logger.info('Deleting composite index:\n{}'.format(index))
      index_keys = []
      composite_id = str(index.id())
      index_keys.append(self._SEPARATOR.join([app_id, 'index', composite_id]))
      yield self.datastore_batch.batch_delete(
        dbconstants.METADATA_TABLE, index_keys,
        column_names=dbconstants.METADATA_TABLE)
  
    @gen.coroutine
    def create_composite_index(self, app_id, index):
      """ Stores a new index for the given application identifier.
  
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      Returns:
        A unique number representing the composite index ID.
      """
      # Generate a random number based on time of creation.
      rand = int(str(int(time.time())) + str(random.randint(0, 999999)))
      index.set_id(rand)
      encoded_entity = index.Encode()
      row_key = self._SEPARATOR.join([app_id, 'index', str(rand)])
      row_keys = [row_key]
      row_values = {}
      row_values[row_key] = {dbconstants.METADATA_SCHEMA[0]: encoded_entity}
      yield self.datastore_batch.batch_put_entity(
        dbconstants.METADATA_TABLE, row_keys,
        dbconstants.METADATA_SCHEMA, row_values)
      raise gen.Return(rand)
  
    @gen.coroutine
    def update_composite_index(self, app_id, index):
      """ Updates an index for a given app ID.
  
      Args:
        app_id: A string containing the app ID.
        index: An entity_pb.CompositeIndex object.
      """
      self.logger.info('Updating index: {}'.format(index))
      entries_updated = 0
      entity_type = index.definition().entity_type()
  
      # TODO: Adjust prefix based on ancestor.
      prefix = '{app}{delimiter}{entity_type}{kind_separator}'.format(
        app=app_id,
        delimiter=self._SEPARATOR * 2,
        entity_type=entity_type,
        kind_separator=dbconstants.KIND_SEPARATOR,
      )
      start_row = prefix
      end_row = prefix + self._TERM_STRING
      start_inclusive = True
  
      while True:
        # Fetch references from the kind table since entity keys can have a
        # parent prefix.
        references = yield self.datastore_batch.range_query(
          table_name=dbconstants.APP_KIND_TABLE,
          column_names=dbconstants.APP_KIND_SCHEMA,
          start_key=start_row,
          end_key=end_row,
          limit=self.BATCH_SIZE,
          offset=0,
          start_inclusive=start_inclusive,
        )
  
        pb_entities = yield self.__fetch_entities(references)
        entities = [entity_pb.EntityProto(entity) for entity in pb_entities]
  
        yield self.insert_composite_indexes(entities, [index])
        entries_updated += len(entities)
  
        # If we fetched fewer references than we asked for, we're done.
        if len(references) < self.BATCH_SIZE:
          break
  
        start_row = references[-1].keys()[0]
        start_inclusive = self._DISABLE_INCLUSIVITY
  
      self.logger.info('Updated {} index entries.'.format(entries_updated))
  
    @gen.coroutine
    def allocate_size(self, project, size):
      """ Allocates a block of IDs for a project.
  
      Args:
        project: A string specifying the project ID.
        size: An integer specifying the number of IDs to reserve.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      start_id, end_id = yield allocator.allocate_size(size)
      raise gen.Return((start_id, end_id))
  
    @gen.coroutine
    def allocate_max(self, project, max_id):
      """ Reserves all IDs up to the one given.
  
      Args:
        project: A string specifying the project ID.
        max_id: An integer specifying the maximum ID to allocated.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      start_id, end_id = yield allocator.allocate_max(max_id)
      raise gen.Return((start_id, end_id))
  
    @gen.coroutine
    def reserve_ids(self, project_id, ids):
      """ Ensures the given IDs are not re-allocated.
  
      Args:
        project_id: A string specifying the project ID.
        ids: An iterable of integers specifying entity IDs.
      """
      if project_id not in self.sequential_allocators:
        self.sequential_allocators[project_id] = EntityIDAllocator(
          self.datastore_batch.session, project_id)
  
      if project_id not in self.scattered_allocators:
        self.scattered_allocators[project_id] = ScatteredAllocator(
          self.datastore_batch.session, project_id)
  
      for id_ in ids:
        counter, space = IdToCounter(id_)
        if space == SEQUENTIAL:
          allocator = self.sequential_allocators[project_id]
        else:
          allocator = self.scattered_allocators[project_id]
  
        yield allocator.set_min_counter(counter)
  
    @gen.coroutine
    def put_entities(self, app, entities, composite_indexes=()):
      """ Updates indexes of existing entities, inserts new entities and
          indexes for them.
  
      Args:
        app: A string containing the application ID.
        entities: List of entities.
        composite_indexes: A list or tuple of CompositeIndex objects.
      """
      self.logger.debug('Inserting {} entities'.format(len(entities)))
  
      by_group = {}
      for entity in entities:
        group_key = group_for_key(entity.key()).Encode()
        if group_key not in by_group:
          by_group[group_key] = []
        by_group[group_key].append(entity)
  
      for encoded_group_key, entity_list in by_group.iteritems():
        group_key = entity_pb.Reference(encoded_group_key)
  
        txid = self.transaction_manager.create_transaction_id(app, xg=False)
        self.transaction_manager.set_groups(app, txid, [group_key])
  
        # Allow the lock to stick around if there is an issue applying the batch.
        lock = entity_lock.EntityLock(self.zookeeper.handle, [group_key], txid)
        try:
          yield lock.acquire()
        except entity_lock.LockTimeout:
          raise Timeout('Unable to acquire entity group lock')
  
        try:
          entity_keys = [
            get_entity_key(self.get_table_prefix(entity), entity.key().path())
            for entity in entity_list]
          current_values = yield self.datastore_batch.batch_get_entity(
            dbconstants.APP_ENTITY_TABLE, entity_keys, APP_ENTITY_SCHEMA)
  
          batch = []
          entity_changes = []
          for entity in entity_list:
            prefix = self.get_table_prefix(entity)
            entity_key = get_entity_key(prefix, entity.key().path())
  
            current_value = None
            if current_values[entity_key]:
              current_value = entity_pb.EntityProto(
                current_values[entity_key][APP_ENTITY_SCHEMA[0]])
  
            batch.extend(mutations_for_entity(entity, txid, current_value,
                                              composite_indexes))
  
            batch.append({'table': 'group_updates',
                          'key': bytearray(encoded_group_key),
                          'last_update': txid})
  
            entity_changes.append(
              {'key': entity.key(), 'old': current_value, 'new': entity})
          yield self.datastore_batch.batch_mutate(
            app, batch, entity_changes, txid)
          lock.release()
  
        finally:
          # In case of failure entity group lock should stay acquired
          # as transaction groomer will handle it later.
         
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/datastore_distributed.py
files: # See LICENSE file
  #
  """
  This web service interfaces with the datastore. It takes protocol buffer
  requests from AppServers and responds according to the type of request its
  given (Put, Get, Delete, Query, etc).
  """
  import argparse
  import json
  import logging
  import os
  import sys
  import time
  import tornado.httpserver
  import tornado.web
  
  from appscale.common import appscale_info
  from appscale.common.appscale_info import get_load_balancer_ips
  from appscale.common.async_retrying import retry_data_watch_coroutine
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from kazoo.exceptions import NodeExistsError
  from tornado import gen
  from tornado.httpclient import AsyncHTTPClient
  from tornado.ioloop import IOLoop
  from tornado.options import options
  from .. import dbconstants
  from ..appscale_datastore_batch import DatastoreFactory
  from ..datastore_distributed import DatastoreDistributed
  from ..utils import (clean_app_id,
                       logger,
                       UnprocessedQueryResult)
  from ..zkappscale import zktransaction
  from ..zkappscale.transaction_manager import TransactionManager
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_v4_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.ext.remote_api import remote_api_pb
  
  # Global for accessing the datastore. An instance of DatastoreDistributed.
  datastore_access = None
  
  # A record of active datastore servers.
  datastore_servers = set()
  
  # The ZooKeeper path where this server registers its availability.
  server_node = None
  
  # ZooKeeper global variable for locking
  zookeeper = None
  
  # Determines whether or not to allow datastore writes. Note: After enabling,
  # datastore processes must be restarted and the groomer must be stopped.
  READ_ONLY = False
  
  # Global stats.
  STATS = {}
  
  # The ZooKeeper path where a list of active datastore servers is stored.
  DATASTORE_SERVERS_NODE = '/appscale/datastore/servers'
  
  
  class ClearHandler(tornado.web.RequestHandler):
    """ Defines what to do when the webserver receives a /clear HTTP request. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handles POST requests for clearing datastore server stats. """
      global STATS
      STATS = {}
      self.write({"message": "Statistics for this server cleared."})
      self.finish()
  
  
  class ReadOnlyHandler(tornado.web.RequestHandler):
    """ Handles requests to check or set read-only mode. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handle requests to turn read-only mode on or off. """
      global READ_ONLY
  
      payload = self.request.body
      data = json.loads(payload)
      if 'readOnly' not in data:
        self.set_status(dbconstants.HTTP_BAD_REQUEST)
  
      if data['readOnly']:
        READ_ONLY = True
        message = 'Write operations now disabled.'
      else:
        READ_ONLY = False
        message = 'Write operations now enabled.'
  
      logger.info(message)
      self.write({'message': message})
      self.finish()
  
  
  class ReserveKeysHandler(tornado.web.RequestHandler):
    """ Handles v4 AllocateIds requests from other servers. """
    @gen.coroutine
    def post(self):
      """ Prevents the provided IDs from being re-allocated. """
      project_id = self.request.headers['appdata']
      request = datastore_v4_pb.AllocateIdsRequest(self.request.body)
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      yield datastore_access.reserve_ids(project_id, ids)
  
  
  class MainHandler(tornado.web.RequestHandler):
    """
    Defines what to do when the webserver receives different types of 
    HTTP requests.
    """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    def unknown_request(self, app_id, http_request_data, pb_type):
      """ Function which handles unknown protocol buffers.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer
      Raises:
        Raises exception.
      """
      raise NotImplementedError("Unknown request of operation {0}" \
        .format(pb_type))
  
    @gen.coroutine
    def post(self):
      """ Function which handles POST requests. Data of the request is
          the request from the AppServer in an encoded protocol buffer
          format.
      """
      request = self.request
      http_request_data = request.body
      pb_type = request.headers['protocolbuffertype']
      app_data = request.headers['appdata']
      app_data = app_data.split(':')
  
      if len(app_data) == 4:
        app_id, user_email, nick_name, auth_domain = app_data
        os.environ['AUTH_DOMAIN'] = auth_domain
        os.environ['USER_EMAIL'] = user_email
        os.environ['USER_NICKNAME'] = nick_name
        os.environ['APPLICATION_ID'] = app_id
      elif len(app_data) == 1:
        app_id = app_data[0]
        os.environ['APPLICATION_ID'] = app_id
      else:
        return
  
      # If the application identifier has the HRD string prepened, remove it.
      app_id = clean_app_id(app_id)
  
      if pb_type == "Request":
        yield self.remote_request(app_id, http_request_data,
                                  service_id=request.headers.get('Module'),
                                  version_id=request.headers.get('Version'))
      else:
        self.unknown_request(app_id, http_request_data, pb_type)
  
    @tornado.web.asynchronous
    def get(self):
      """ Handles get request for the web server. Returns that it is currently
          up in json.
      """
      self.write(json.dumps(STATS))
      self.finish()
  
    @gen.coroutine
    def remote_request(self, app_id, http_request_data, service_id, version_id):
      """ Receives a remote request to which it should give the correct 
          response. The http_request_data holds an encoded protocol buffer
          of a certain type. Each type has a particular response type. 
      
      Args:
        app_id: The application ID that is sending this request.
        http_request_data: Encoded protocol buffer.
        service_id: A string specifying the client's service ID.
        version_id: A string specifying the client's version ID.
      """
      apirequest = remote_api_pb.Request()
      apirequest.ParseFromString(http_request_data)
      apiresponse = remote_api_pb.Response()
      response = None
      if not apirequest.has_method():
        apirequest.set_method("NOT_FOUND")
      if not apirequest.has_request():
        apirequest.set_method("NOT_FOUND")
        apirequest.clear_request()
      method = apirequest.method()
      http_request_data = apirequest.request()
      start = time.time()
  
      request_log = method
      if apirequest.has_request_id():
        request_log += ': {}'.format(apirequest.request_id())
      logger.debug(request_log)
  
      if method == "Put":
        response, errcode, errdetail = yield self.put_request(
          app_id, http_request_data)
      elif method == "Get":
        response, errcode, errdetail = yield self.get_request(
          app_id, http_request_data)
      elif method == "Delete":
        response, errcode, errdetail = yield self.delete_request(
          app_id, http_request_data)
      elif method == "RunQuery":
        response, errcode, errdetail = yield self.run_query(http_request_data)
      elif method == "BeginTransaction":
        response, errcode, errdetail = yield self.begin_transaction_request(
          app_id, http_request_data)
      elif method == "Commit":
        response, errcode, errdetail = yield self.commit_transaction_request(
          app_id, http_request_data)
      elif method == "Rollback":
        response, errcode, errdetail = self.rollback_transaction_request(
          app_id, http_request_data)
      elif method == "AllocateIds":
        response, errcode, errdetail = yield self.allocate_ids_request(
          app_id, http_request_data)
      elif method == "CreateIndex":
        response, errcode, errdetail = yield self.create_index_request(
          app_id, http_request_data)
      elif method == "GetIndices":
        response, errcode, errdetail = yield self.get_indices_request(app_id)
      elif method == "UpdateIndex":
        response, errcode, errdetail = self.update_index_request(
          app_id, http_request_data)
      elif method == "DeleteIndex":
        response, errcode, errdetail = yield self.delete_index_request(
          app_id, http_request_data)
      elif method == 'AddActions':
        response, errcode, errdetail = yield self.add_actions_request(
          app_id, http_request_data, service_id, version_id)
      elif method == 'datastore_v4.AllocateIds':
        response, errcode, errdetail = yield self.v4_allocate_ids_request(
          app_id, http_request_data)
      else:
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Unknown datastore message"
  
      time_taken = time.time() - start
      if method in STATS:
        if errcode in STATS[method]:
          prev_req, pre_time = STATS[method][errcode]
          STATS[method][errcode] = prev_req + 1, pre_time + time_taken
        else:
          STATS[method][errcode] = (1, time_taken)
      else:
        STATS[method] = {}
        STATS[method][errcode] = (1, time_taken)
  
      apiresponse.set_response(response)
      if errcode != 0:
        apperror_pb = apiresponse.mutable_application_error()
        apperror_pb.set_code(errcode)
        apperror_pb.set_detail(errdetail)
  
      self.write(apiresponse.Encode())
  
    @gen.coroutine
    def begin_transaction_request(self, app_id, http_request_data):
      """ Handles the intial request to start a transaction. Replies with 
          a unique identifier to handle this transaction in future requests.
    
      Args:
        app_id: The application ID requesting the transaction.
        http_request_data: The encoded request.
      Returns:
        An encoded transaction protocol buffer with a unique handler.
      """
      global datastore_access
      begin_transaction_req_pb = datastore_pb.BeginTransactionRequest(
        http_request_data)
      multiple_eg = False
      if begin_transaction_req_pb.has_allow_multiple_eg():
        multiple_eg = bool(begin_transaction_req_pb.allow_multiple_eg())
  
      if READ_ONLY:
        logger.warning('Unable to begin transaction in read-only mode: {}'.
                       format(begin_transaction_req_pb))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      try:
        handle = yield datastore_access.setup_transaction(app_id, multiple_eg)
      except dbconstants.InternalError as error:
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      except dbconstants.BadRequest as error:
        raise gen.Return(('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except (zktransaction.ZKInternalException,
              dbconstants.AppScaleDBConnectionError) as error:
        logger.exception('Unable to begin transaction')
        raise gen.Return(
          ('', datastore_pb.Error.INTERNAL_ERROR,
           str(error)))
  
      transaction_pb = datastore_pb.Transaction()
      transaction_pb.set_app(app_id)
      transaction_pb.set_handle(handle)
      raise gen.Return((transaction_pb.Encode(), 0, ''))
  
    @gen.coroutine
    def commit_transaction_request(self, app_id, http_request_data):
      """ Handles the commit phase of a transaction.
  
      Args:
        app_id: The application ID requesting the transaction commit.
        http_request_data: The encoded request of datastore_pb.Transaction.
      Returns:
        An encoded protocol buffer commit response.
      """
      global datastore_access
  
      if READ_ONLY:
        transaction_pb = datastore_pb.Transaction(http_request_data)
        logger.warning('Unable to commit in read-only mode: {}'.
          format(transaction_pb))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      result = yield datastore_access.commit_transaction(
        app_id, http_request_data)
      raise gen.Return(result)
  
    def rollback_transaction_request(self, app_id, http_request_data):
      """ Handles the rollback phase of a transaction.
  
      Args:
        app_id: The application ID requesting the rollback.
        http_request_data: The encoded request.
      Returns:
        An encoded protocol buffer void response.
      """
      global datastore_access
  
      if READ_ONLY:
        logger.warning('Unable to rollback in read-only mode: {}'.
          format(http_request_data))
        return ('', datastore_pb.Error.CAPABILITY_DISABLED,
                'Datastore is in read-only mode.')
  
      txn = datastore_pb.Transaction(http_request_data)
      try:
        datastore_access.rollback_transaction(app_id, txn.handle())
      except dbconstants.InternalError as error:
        logger.exception('Unable to rollback transaction')
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except Exception as error:
        logger.exception('Unable to rollback transaction')
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
  
      return api_base_pb.VoidProto().Encode(), 0, ''
  
    @gen.coroutine
    def run_query(self, http_request_data):
      """ High level function for running queries.
  
      Args:
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        Returns an encoded query response.
      """
      global datastore_access
      query = datastore_pb.Query(http_request_data)
      clone_qr_pb = UnprocessedQueryResult()
      try:
        yield datastore_access._dynamic_run_query(query, clone_qr_pb)
      except dbconstants.BadRequest as error:
        raise gen.Return( ('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except zktransaction.ZKBadRequest as error:
        logger.exception(
          'Illegal arguments in transaction during {}'.format(query))
        raise gen.Return(('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except zktransaction.ZKInternalException as error:
        logger.exception('ZKInternalException during {}'.format(query))
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      except zktransaction.ZKTransactionException as error:
        logger.exception('Concurrent transaction during {}'.format(query))
        raise gen.Return(
          ('', datastore_pb.Error.CONCURRENT_TRANSACTION, str(error)))
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception('DB connection error during query')
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      raise gen.Return((clone_qr_pb.Encode(), 0, ''))
  
    @gen.coroutine
    def create_index_request(self, app_id, http_request_data):
      """ High level function for creating composite indexes.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.Integer64Proto()
  
      if READ_ONLY:
        logger.warning('Unable to create in read-only mode: {}'.format(request))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      try:
        index_id = yield datastore_access.create_composite_index(app_id, request)
        response.set_value(index_id)
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception('DB connection error during index creation')
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      raise gen.Return((response.Encode(), 0, ''))
  
    def update_index_request(self, app_id, http_request_data):
      """ High level function for updating a composite index.
  
      Args:
        app_id: A string containing the application ID.
        http_request_data: A string containing the protocol buffer request
          from the AppServer.
      Returns:
         A tuple containing an encoded response, error code, and error details.
      """
      global datastore_access
      index = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to update in read-only mode: {}'.format(index))
        return ('', datastore_pb.Error.CAPABILITY_DISABLED,
                'Datastore is in read-only mode.')
  
      state = index.state()
      if state not in [index.READ_WRITE, index.WRITE_ONLY]:
        state_name = entity_pb.CompositeIndex.State_Name(state)
        error_message = 'Unable to update index because state is {}. '\
          'Index: {}'.format(state_name, index)
        logger.error(error_message)
        return '', datastore_pb.Error.PERMISSION_DENIED, error_message
      else:
        # Updating index in background so we can return a response quickly.
        IOLoop.current().spawn_callback(
          datastore_access.update_composite_index, app_id, index)
  
      return response.Encode(), 0, ''
  
    @gen.coroutine
    def delete_index_request(self, app_id, http_request_data):
      """ Deletes a composite index for a given application.
    
      Args:
        app_id: Name of the application.
        http_request_data: A serialized CompositeIndices item
      Returns:
        A Tuple of an encoded entity_pb.VoidProto, error code, and 
        error explanation.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to delete in read-only mode: {}'.format(request))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      try:
        yield datastore_access.delete_composite_index_metadata(app_id, request)
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception('DB connection error during index deletion')
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
  
      raise gen.Return((response.Encode(), 0, ''))
  
    @gen.coroutine
    def get_indices_request(self, app_id):
      """ Gets the indices of the given application.
  
      Args:
        app_id: Name of the application.
      Returns: 
        A Tuple of an encoded response, error code, and error explanation.
      """
      global datastore_access
      response = datastore_pb.CompositeIndices()
      try:
        indices = yield datastore_access.datastore_batch.get_indices(app_id)
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception(
          'DB connection error while fetching indices for {}'.format(app_id))
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
  
      for index in indices:
        new_index = response.add_index()
        new_index.ParseFromString(index)
  
      raise gen.Return((response.Encode(), 0, ''))
  
    @gen.coroutine
    def allocate_ids_request(self, app_id, http_request_data):
      """ High level function for getting unique identifiers for entities.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      Raises:
         NotImplementedError: when requesting a max id.
      """
      request = datastore_pb.AllocateIdsRequest(http_request_data)
  
      if request.has_max() and request.has_size():
        raise gen.Return(
          ('', datastore_pb.Error.BAD_REQUEST,
           'Both size and max cannot be set.'))
      if not (request.has_max() or request.has_size()):
        raise gen.Return(
          ('', datastore_pb.Error.BAD_REQUEST,
           'Either size or max must be set.'))
  
      if request.has_size():
        coroutine = datastore_access.allocate_size
        args = (app_id, request.size())
      else:
        coroutine = datastore_access.allocate_max
        args = (app_id, request.max())
  
      try:
        start, end = yield coroutine(*args)
      except dbconstants.AppScaleBadArg as error:
        raise gen.Return(('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except dbconstants.AppScaleDBConnectionError as error:
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
  
      response = datastore_pb.AllocateIdsResponse()
      response.set_start(start)
      response.set_end(end)
      raise gen.Return((response.Encode(), 0, ''))
  
    @staticmethod
    @gen.coroutine
    def v4_allocate_ids_request(app_id, http_request_data):
      """ Reserves entity IDs so that they will not be re-allocated.
  
      Args:
        app_id: Name of the application.
        http_request_data: The protocol buffer request from the AppServer.
      Returns:
         Returns an encoded response.
      """
      request = datastore_v4_pb.AllocateIdsRequest(http_request_data)
      response = datastore_v4_pb.AllocateIdsResponse()
  
      if not request.reserve_list():
        raise gen.Return((response.Encode(), datastore_v4_pb.Error.BAD_REQUEST,
                          'Request must include reserve list'))
  
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      yield datastore_access.reserve_ids(app_id, ids)
  
      # Forward request to other datastore servers in order to adjust any blocks
      # they've already allocated.
      client = AsyncHTTPClient()
      headers = {'appdata': app_id}
  
      futures = []
      for server in datastore_servers:
        ip, port = server.split(':')
        port = int(port)
        if ip == options.private_ip and port == options.port:
          continue
  
        url = 'http://{}:{}/reserve-keys'.format(ip, port)
        future = client.fetch(url, method='POST', headers=headers,
                              body=http_request_data)
        futures.append(future)
  
      for future in futures:
        yield future
  
      raise
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/scripts/datastore.py
files: # See LICENSE file
  #
  """
  This web service interfaces with the datastore. It takes protocol buffer
  requests from AppServers and responds according to the type of request its
  given (Put, Get, Delete, Query, etc).
  """
  import argparse
  import json
  import logging
  import os
  import sys
  import time
  import tornado.httpserver
  import tornado.web
  
  from appscale.common import appscale_info
  from appscale.common.appscale_info import get_load_balancer_ips
  from appscale.common.async_retrying import retry_data_watch_coroutine
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from kazoo.exceptions import NodeExistsError
  from tornado import gen
  from tornado.httpclient import AsyncHTTPClient
  from tornado.ioloop import IOLoop
  from tornado.options import options
  from .. import dbconstants
  from ..appscale_datastore_batch import DatastoreFactory
  from ..datastore_distributed import DatastoreDistributed
  from ..utils import (clean_app_id,
                       logger,
                       UnprocessedQueryResult)
  from ..zkappscale import zktransaction
  from ..zkappscale.transaction_manager import TransactionManager
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_v4_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.ext.remote_api import remote_api_pb
  
  # Global for accessing the datastore. An instance of DatastoreDistributed.
  datastore_access = None
  
  # A record of active datastore servers.
  datastore_servers = set()
  
  # The ZooKeeper path where this server registers its availability.
  server_node = None
  
  # ZooKeeper global variable for locking
  zookeeper = None
  
  # Determines whether or not to allow datastore writes. Note: After enabling,
  # datastore processes must be restarted and the groomer must be stopped.
  READ_ONLY = False
  
  # Global stats.
  STATS = {}
  
  # The ZooKeeper path where a list of active datastore servers is stored.
  DATASTORE_SERVERS_NODE = '/appscale/datastore/servers'
  
  
  class ClearHandler(tornado.web.RequestHandler):
    """ Defines what to do when the webserver receives a /clear HTTP request. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handles POST requests for clearing datastore server stats. """
      global STATS
      STATS = {}
      self.write({"message": "Statistics for this server cleared."})
      self.finish()
  
  
  class ReadOnlyHandler(tornado.web.RequestHandler):
    """ Handles requests to check or set read-only mode. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handle requests to turn read-only mode on or off. """
      global READ_ONLY
  
      payload = self.request.body
      data = json.loads(payload)
      if 'readOnly' not in data:
        self.set_status(dbconstants.HTTP_BAD_REQUEST)
  
      if data['readOnly']:
        READ_ONLY = True
        message = 'Write operations now disabled.'
      else:
        READ_ONLY = False
        message = 'Write operations now enabled.'
  
      logger.info(message)
      self.write({'message': message})
      self.finish()
  
  
  class ReserveKeysHandler(tornado.web.RequestHandler):
    """ Handles v4 AllocateIds requests from other servers. """
    @gen.coroutine
    def post(self):
      """ Prevents the provided IDs from being re-allocated. """
      project_id = self.request.headers['appdata']
      request = datastore_v4_pb.AllocateIdsRequest(self.request.body)
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      yield datastore_access.reserve_ids(project_id, ids)
  
  
  class MainHandler(tornado.web.RequestHandler):
    """
    Defines what to do when the webserver receives different types of 
    HTTP requests.
    """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    def unknown_request(self, app_id, http_request_data, pb_type):
      """ Function which handles unknown protocol buffers.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer
      Raises:
        Raises exception.
      """
      raise NotImplementedError("Unknown request of operation {0}" \
        .format(pb_type))
  
    @gen.coroutine
    def post(self):
      """ Function which handles POST requests. Data of the request is
          the request from the AppServer in an encoded protocol buffer
          format.
      """
      request = self.request
      http_request_data = request.body
      pb_type = request.headers['protocolbuffertype']
      app_data = request.headers['appdata']
      app_data = app_data.split(':')
  
      if len(app_data) == 4:
        app_id, user_email, nick_name, auth_domain = app_data
        os.environ['AUTH_DOMAIN'] = auth_domain
        os.environ['USER_EMAIL'] = user_email
        os.environ['USER_NICKNAME'] = nick_name
        os.environ['APPLICATION_ID'] = app_id
      elif len(app_data) == 1:
        app_id = app_data[0]
        os.environ['APPLICATION_ID'] = app_id
      else:
        return
  
      # If the application identifier has the HRD string prepened, remove it.
      app_id = clean_app_id(app_id)
  
      if pb_type == "Request":
        yield self.remote_request(app_id, http_request_data,
                                  service_id=request.headers.get('Module'),
                                  version_id=request.headers.get('Version'))
      else:
        self.unknown_request(app_id, http_request_data, pb_type)
  
    @tornado.web.asynchronous
    def get(self):
      """ Handles get request for the web server. Returns that it is currently
          up in json.
      """
      self.write(json.dumps(STATS))
      self.finish()
  
    @gen.coroutine
    def remote_request(self, app_id, http_request_data, service_id, version_id):
      """ Receives a remote request to which it should give the correct 
          response. The http_request_data holds an encoded protocol buffer
          of a certain type. Each type has a particular response type. 
      
      Args:
        app_id: The application ID that is sending this request.
        http_request_data: Encoded protocol buffer.
        service_id: A string specifying the client's service ID.
        version_id: A string specifying the client's version ID.
      """
      apirequest = remote_api_pb.Request()
      apirequest.ParseFromString(http_request_data)
      apiresponse = remote_api_pb.Response()
      response = None
      if not apirequest.has_method():
        apirequest.set_method("NOT_FOUND")
      if not apirequest.has_request():
        apirequest.set_method("NOT_FOUND")
        apirequest.clear_request()
      method = apirequest.method()
      http_request_data = apirequest.request()
      start = time.time()
  
      request_log = method
      if apirequest.has_request_id():
        request_log += ': {}'.format(apirequest.request_id())
      logger.debug(request_log)
  
      if method == "Put":
        response, errcode, errdetail = yield self.put_request(
          app_id, http_request_data)
      elif method == "Get":
        response, errcode, errdetail = yield self.get_request(
          app_id, http_request_data)
      elif method == "Delete":
        response, errcode, errdetail = yield self.delete_request(
          app_id, http_request_data)
      elif method == "RunQuery":
        response, errcode, errdetail = yield self.run_query(http_request_data)
      elif method == "BeginTransaction":
        response, errcode, errdetail = yield self.begin_transaction_request(
          app_id, http_request_data)
      elif method == "Commit":
        response, errcode, errdetail = yield self.commit_transaction_request(
          app_id, http_request_data)
      elif method == "Rollback":
        response, errcode, errdetail = self.rollback_transaction_request(
          app_id, http_request_data)
      elif method == "AllocateIds":
        response, errcode, errdetail = yield self.allocate_ids_request(
          app_id, http_request_data)
      elif method == "CreateIndex":
        response, errcode, errdetail = yield self.create_index_request(
          app_id, http_request_data)
      elif method == "GetIndices":
        response, errcode, errdetail = yield self.get_indices_request(app_id)
      elif method == "UpdateIndex":
        response, errcode, errdetail = self.update_index_request(
          app_id, http_request_data)
      elif method == "DeleteIndex":
        response, errcode, errdetail = yield self.delete_index_request(
          app_id, http_request_data)
      elif method == 'AddActions':
        response, errcode, errdetail = yield self.add_actions_request(
          app_id, http_request_data, service_id, version_id)
      elif method == 'datastore_v4.AllocateIds':
        response, errcode, errdetail = yield self.v4_allocate_ids_request(
          app_id, http_request_data)
      else:
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Unknown datastore message"
  
      time_taken = time.time() - start
      if method in STATS:
        if errcode in STATS[method]:
          prev_req, pre_time = STATS[method][errcode]
          STATS[method][errcode] = prev_req + 1, pre_time + time_taken
        else:
          STATS[method][errcode] = (1, time_taken)
      else:
        STATS[method] = {}
        STATS[method][errcode] = (1, time_taken)
  
      apiresponse.set_response(response)
      if errcode != 0:
        apperror_pb = apiresponse.mutable_application_error()
        apperror_pb.set_code(errcode)
        apperror_pb.set_detail(errdetail)
  
      self.write(apiresponse.Encode())
  
    @gen.coroutine
    def begin_transaction_request(self, app_id, http_request_data):
      """ Handles the intial request to start a transaction. Replies with 
          a unique identifier to handle this transaction in future requests.
    
      Args:
        app_id: The application ID requesting the transaction.
        http_request_data: The encoded request.
      Returns:
        An encoded transaction protocol buffer with a unique handler.
      """
      global datastore_access
      begin_transaction_req_pb = datastore_pb.BeginTransactionRequest(
        http_request_data)
      multiple_eg = False
      if begin_transaction_req_pb.has_allow_multiple_eg():
        multiple_eg = bool(begin_transaction_req_pb.allow_multiple_eg())
  
      if READ_ONLY:
        logger.warning('Unable to begin transaction in read-only mode: {}'.
                       format(begin_transaction_req_pb))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      try:
        handle = yield datastore_access.setup_transaction(app_id, multiple_eg)
      except dbconstants.InternalError as error:
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      except dbconstants.BadRequest as error:
        raise gen.Return(('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except (zktransaction.ZKInternalException,
              dbconstants.AppScaleDBConnectionError) as error:
        logger.exception('Unable to begin transaction')
        raise gen.Return(
          ('', datastore_pb.Error.INTERNAL_ERROR,
           str(error)))
  
      transaction_pb = datastore_pb.Transaction()
      transaction_pb.set_app(app_id)
      transaction_pb.set_handle(handle)
      raise gen.Return((transaction_pb.Encode(), 0, ''))
  
    @gen.coroutine
    def commit_transaction_request(self, app_id, http_request_data):
      """ Handles the commit phase of a transaction.
  
      Args:
        app_id: The application ID requesting the transaction commit.
        http_request_data: The encoded request of datastore_pb.Transaction.
      Returns:
        An encoded protocol buffer commit response.
      """
      global datastore_access
  
      if READ_ONLY:
        transaction_pb = datastore_pb.Transaction(http_request_data)
        logger.warning('Unable to commit in read-only mode: {}'.
          format(transaction_pb))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      result = yield datastore_access.commit_transaction(
        app_id, http_request_data)
      raise gen.Return(result)
  
    def rollback_transaction_request(self, app_id, http_request_data):
      """ Handles the rollback phase of a transaction.
  
      Args:
        app_id: The application ID requesting the rollback.
        http_request_data: The encoded request.
      Returns:
        An encoded protocol buffer void response.
      """
      global datastore_access
  
      if READ_ONLY:
        logger.warning('Unable to rollback in read-only mode: {}'.
          format(http_request_data))
        return ('', datastore_pb.Error.CAPABILITY_DISABLED,
                'Datastore is in read-only mode.')
  
      try:
        return datastore_access.rollback_transaction(app_id, http_request_data)
      except zktransaction.ZKInternalException as error:
        logger.exception('ZKInternalException during {} for {}'.
          format(http_request_data, app_id))
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except Exception as error:
        logger.exception('Unable to rollback transaction')
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
  
    @gen.coroutine
    def run_query(self, http_request_data):
      """ High level function for running queries.
  
      Args:
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        Returns an encoded query response.
      """
      global datastore_access
      query = datastore_pb.Query(http_request_data)
      clone_qr_pb = UnprocessedQueryResult()
      try:
        yield datastore_access._dynamic_run_query(query, clone_qr_pb)
      except dbconstants.BadRequest as error:
        raise gen.Return( ('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except zktransaction.ZKBadRequest as error:
        logger.exception(
          'Illegal arguments in transaction during {}'.format(query))
        raise gen.Return(('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except zktransaction.ZKInternalException as error:
        logger.exception('ZKInternalException during {}'.format(query))
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      except zktransaction.ZKTransactionException as error:
        logger.exception('Concurrent transaction during {}'.format(query))
        raise gen.Return(
          ('', datastore_pb.Error.CONCURRENT_TRANSACTION, str(error)))
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception('DB connection error during query')
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      raise gen.Return((clone_qr_pb.Encode(), 0, ''))
  
    @gen.coroutine
    def create_index_request(self, app_id, http_request_data):
      """ High level function for creating composite indexes.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.Integer64Proto()
  
      if READ_ONLY:
        logger.warning('Unable to create in read-only mode: {}'.format(request))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      try:
        index_id = yield datastore_access.create_composite_index(app_id, request)
        response.set_value(index_id)
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception('DB connection error during index creation')
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
      raise gen.Return((response.Encode(), 0, ''))
  
    def update_index_request(self, app_id, http_request_data):
      """ High level function for updating a composite index.
  
      Args:
        app_id: A string containing the application ID.
        http_request_data: A string containing the protocol buffer request
          from the AppServer.
      Returns:
         A tuple containing an encoded response, error code, and error details.
      """
      global datastore_access
      index = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to update in read-only mode: {}'.format(index))
        return ('', datastore_pb.Error.CAPABILITY_DISABLED,
                'Datastore is in read-only mode.')
  
      state = index.state()
      if state not in [index.READ_WRITE, index.WRITE_ONLY]:
        state_name = entity_pb.CompositeIndex.State_Name(state)
        error_message = 'Unable to update index because state is {}. '\
          'Index: {}'.format(state_name, index)
        logger.error(error_message)
        return '', datastore_pb.Error.PERMISSION_DENIED, error_message
      else:
        # Updating index in background so we can return a response quickly.
        IOLoop.current().spawn_callback(
          datastore_access.update_composite_index, app_id, index)
  
      return response.Encode(), 0, ''
  
    @gen.coroutine
    def delete_index_request(self, app_id, http_request_data):
      """ Deletes a composite index for a given application.
    
      Args:
        app_id: Name of the application.
        http_request_data: A serialized CompositeIndices item
      Returns:
        A Tuple of an encoded entity_pb.VoidProto, error code, and 
        error explanation.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to delete in read-only mode: {}'.format(request))
        raise gen.Return(
          ('', datastore_pb.Error.CAPABILITY_DISABLED,
           'Datastore is in read-only mode.'))
  
      try:
        yield datastore_access.delete_composite_index_metadata(app_id, request)
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception('DB connection error during index deletion')
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
  
      raise gen.Return((response.Encode(), 0, ''))
  
    @gen.coroutine
    def get_indices_request(self, app_id):
      """ Gets the indices of the given application.
  
      Args:
        app_id: Name of the application.
      Returns: 
        A Tuple of an encoded response, error code, and error explanation.
      """
      global datastore_access
      response = datastore_pb.CompositeIndices()
      try:
        indices = yield datastore_access.datastore_batch.get_indices(app_id)
      except dbconstants.AppScaleDBConnectionError as error:
        logger.exception(
          'DB connection error while fetching indices for {}'.format(app_id))
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
  
      for index in indices:
        new_index = response.add_index()
        new_index.ParseFromString(index)
  
      raise gen.Return((response.Encode(), 0, ''))
  
    @gen.coroutine
    def allocate_ids_request(self, app_id, http_request_data):
      """ High level function for getting unique identifiers for entities.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      Raises:
         NotImplementedError: when requesting a max id.
      """
      request = datastore_pb.AllocateIdsRequest(http_request_data)
  
      if request.has_max() and request.has_size():
        raise gen.Return(
          ('', datastore_pb.Error.BAD_REQUEST,
           'Both size and max cannot be set.'))
      if not (request.has_max() or request.has_size()):
        raise gen.Return(
          ('', datastore_pb.Error.BAD_REQUEST,
           'Either size or max must be set.'))
  
      if request.has_size():
        coroutine = datastore_access.allocate_size
        args = (app_id, request.size())
      else:
        coroutine = datastore_access.allocate_max
        args = (app_id, request.max())
  
      try:
        start, end = yield coroutine(*args)
      except dbconstants.AppScaleBadArg as error:
        raise gen.Return(('', datastore_pb.Error.BAD_REQUEST, str(error)))
      except dbconstants.AppScaleDBConnectionError as error:
        raise gen.Return(('', datastore_pb.Error.INTERNAL_ERROR, str(error)))
  
      response = datastore_pb.AllocateIdsResponse()
      response.set_start(start)
      response.set_end(end)
      raise gen.Return((response.Encode(), 0, ''))
  
    @staticmethod
    @gen.coroutine
    def v4_allocate_ids_request(app_id, http_request_data):
      """ Reserves entity IDs so that they will not be re-allocated.
  
      Args:
        app_id: Name of the application.
        http_request_data: The protocol buffer request from the AppServer.
      Returns:
         Returns an encoded response.
      """
      request = datastore_v4_pb.AllocateIdsRequest(http_request_data)
      response = datastore_v4_pb.AllocateIdsResponse()
  
      if not request.reserve_list():
        raise gen.Return((response.Encode(), datastore_v4_pb.Error.BAD_REQUEST,
                          'Request must include reserve list'))
  
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      yield datastore_access.reserve_ids(app_id, ids)
  
      # Forward request to other datastore servers in order to adjust any blocks
      # they've already allocated.
      client = AsyncHTTPClient()
      headers = {'appdata': app_id}
  
      futures = []
      for server in datastore_servers:
        ip, port = server.split(':')
        port = int(port)
        if ip == options.private_ip and port == options.port:
          continue
  
        url = 'http://{}:{}/reserve-keys'.format(ip, port)
        future = client.fetch(url, method='POST', headers=headers,
                              body=http_request_data)
        futures.append(future)
  
      for future in futures:
        yield future
  
      raise gen.Return((response.Encode(), 0, ''))
  
    @gen.cor
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/scripts/datastore.py
files: #!/usr/bin/env python
  # Programmer: Navraj Chohan <nlake44@gmail.com>
  
  import datetime
  import random
  import sys
  import unittest
  
  from tornado import gen, testing
  
  from appscale.common import appscale_info
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore import dbconstants
  from appscale.datastore import utils
  from appscale.datastore.datastore_distributed import DatastoreDistributed
  from appscale.datastore.dbconstants import APP_ENTITY_SCHEMA
  from appscale.datastore.dbconstants import JOURNAL_SCHEMA
  from appscale.datastore.dbconstants import TOMBSTONE
  from appscale.datastore.cassandra_env.entity_id_allocator import\
    ScatteredAllocator
  
  from appscale.datastore.cassandra_env.cassandra_interface import (
    DatastoreProxy)
  from appscale.datastore.cassandra_env.utils import deletions_for_entity
  from appscale.datastore.cassandra_env.utils import index_deletions
  from appscale.datastore.cassandra_env.utils import mutations_for_entity
  
  from appscale.datastore.utils import (
    get_entity_key,
    get_entity_kind,
    get_index_key_from_params,
    get_index_kv_from_tuple,
    get_kind_key
  )
  
  from appscale.datastore.zkappscale.entity_lock import EntityLock
  from appscale.datastore.zkappscale.zktransaction import ZKTransactionException
  from cassandra.cluster import Cluster
  from flexmock import flexmock
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.ext import db
  
  
  class Item(db.Model):
    name = db.StringProperty(required = True)
  
  
  ASYNC_NONE = gen.Future()
  ASYNC_NONE.set_result(None)
  
  
  class TestDatastoreServer(testing.AsyncTestCase):
    """
    A set of test cases for the datastore server (datastore server v2)
    """
    BASIC_ENTITY = ['guestbook', 'Greeting', 'foo', 'content', 'hello world']
  
    def setUp(self, *args, **kwargs):
      super(TestDatastoreServer, self).setUp(*args, **kwargs)
  
  
  
    def get_zookeeper(self):
      zk_handle = flexmock(handler=flexmock(event_object=lambda: None,
                                            sleep_func=lambda: None,
                                            lock_object=lambda: None))
      zk_handle.should_receive('add_listener')
      zookeeper = flexmock(handle=zk_handle)
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      zookeeper.should_receive("increment_and_get_counter").and_return(0,1000)
      return zookeeper
  
    def test_get_entity_kind(self):
      item = Item(name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_entity_kind(key), "Item")
  
    def test_kind_key(self):
      item = Item(name="Dyan", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:0000000000\x01")
  
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item1)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:Bob\x01")
  
      item2 = Item(key_name="Frank", name="Frank", _app="hello", parent = item1)
      key = db.model_to_protobuf(item2)
      self.assertEquals(get_kind_key("howdy", key.key().path()),
             "howdy\x00Item\x01Item:Bob\x01Item:Frank\x01")
  
    def test_get_entity_key(self):
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(str(get_entity_key("howdy", key.key().path())), "howdy\x00Item:Bob\x01")
  
    def test_validate_key(self):
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      dd.validate_key(key.key())
  
    def test_get_table_prefix(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(dd.get_table_prefix(key), "hello\x00")
  
    def test_get_index_key_from_params(self):
      params = ['a','b','c','d','e']
      self.assertEquals(get_index_key_from_params(params),
                        "a\x00b\x00c\x00d\x00e")
  
    def test_get_index_kv_from_tuple(self):
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      item2 = Item(key_name="Sally", name="Sally", _app="hello")
      key1 = db.model_to_protobuf(item1)
      key2 = db.model_to_protobuf(item2)
      tuples_list = [("a\x00b",key1),("a\x00b",key2)]
      self.assertEquals(get_index_kv_from_tuple(
        tuples_list), (['a\x00b\x00Item\x00name\x00\x9aBob\x01\x01\x00Item:Bob\x01',
        'a\x00b\x00Item:Bob\x01'],
        ['a\x00b\x00Item\x00name\x00\x9aSally\x01\x01\x00Item:Sally\x01',
        'a\x00b\x00Item:Sally\x01']))
  
    def test_get_composite_index_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      dd = flexmock(dd)
  
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto("appid", "kind", "entity_name", "prop1", "value", ns="")
  
      self.assertEquals(
        dd.get_composite_index_key(composite_index, ent),
        "appid\x00\x00123\x00\x9avalue\x01\x01\x00\x00kind:entity_name\x01")
  
    @testing.gen_test
    def test_get_indices(self):
      session = flexmock(default_consistency_level=None)
      cluster = flexmock(connect=lambda keyspace: session)
      flexmock(appscale_info).should_receive('get_db_ips')
      flexmock(Cluster).new_instances(cluster)
      flexmock(DatastoreProxy).should_receive('range_query').and_return({})
      db_batch = DatastoreProxy()
  
      result = yield db_batch.get_indices("appid")
      self.assertEquals(result, [])
  
    @testing.gen_test
    def test_delete_composite_index_metadata(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_delete").and_return(ASYNC_NONE)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      dd = flexmock(dd)
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(1)
      yield dd.delete_composite_index_metadata("appid", composite_index)
  
    @testing.gen_test
    def test_create_composite_index(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(ASYNC_NONE)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      dd = flexmock(dd)
      index = entity_pb.CompositeIndex()
      index.set_app_id("appid")
      index.set_state(2)
      definition = index.mutable_definition()
      definition.set_entity_type("kind")
      definition.set_ancestor(0)
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      yield dd.create_composite_index("appid", index)
      assert index.id() > 0
  
    @testing.gen_test
    def test_insert_composite_indexes(self):
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto(
        "appid", "kind", "entity_name", "prop1", "value", ns="")
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(ASYNC_NONE).once()
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      yield dd.insert_composite_indexes([ent], [composite_index])
  
    @testing.gen_test
    def test_fetch_keys(self):
      entity_proto1 = self.get_new_entity_proto(
        "test", "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      async_get_1 = gen.Future()
      async_get_1.set_result({
        'test\x00blah\x00test_kind:bob\x01': {
          APP_ENTITY_SCHEMA[0]: entity_proto1.Encode(),
          APP_ENTITY_SCHEMA[1]: 1
        }
      })
      async_get_2 = gen.Future()
      async_get_2.set_result({
        'test\x00blah\x00test_kind:bob\x01\x000000000002': {
          JOURNAL_SCHEMA[0]: entity_proto1.Encode()
      }})
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_delete").and_return(ASYNC_NONE)
      db_batch.should_receive("batch_put_entity").and_return(ASYNC_NONE)
      db_batch.should_receive("batch_get_entity").\
        and_return(async_get_1).\
        and_return(async_get_2)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
  
      fetched = yield dd.fetch_keys([entity_proto1.key()])
      self.assertEquals(fetched[0], {
        'test\x00blah\x00test_kind:bob\x01': {
         'txnID': 1, 'entity': entity_proto1.Encode()
        }
      })
      self.assertEqual(fetched[1], ['test\x00blah\x00test_kind:bob\x01'])
  
    @testing.gen_test
    def test_commit_transaction(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      transaction_manager = flexmock(
        delete_transaction_id=lambda project, txid: None)
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive('apply_txn_changes').and_return(ASYNC_NONE)
      commit_request = datastore_pb.Transaction()
      commit_request.set_handle(123)
      commit_request.set_app("aaa")
      http_request = commit_request.Encode()
      result = yield dd.commit_transaction("app_id", http_request)
      self.assertEquals(result, (datastore_pb.CommitResponse().Encode(), 0, ""))
  
    def test_rollback_transcation(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("release_lock").and_return(True)
      zookeeper.should_receive("notify_failed_transaction").and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      commit_request = datastore_pb.Transaction()
      commit_request.set_handle(123)
      commit_request.set_app("aaa")
      self.assertEquals(
        dd.rollback_transaction("app_id", commit_request.handle()), None)
  
    @staticmethod
    def get_new_entity_proto(app_id, kind, entity_name, prop_name, prop_value, ns=""):
      entity_proto = datastore_pb.EntityProto()
  
      reference = entity_proto.mutable_key()
      reference.set_app(app_id)
      reference.set_name_space(ns)
  
      path = reference.mutable_path()
      element = path.add_element()
      element.set_type(kind)
      element.set_name(entity_name)
  
      ent_group = entity_proto.mutable_entity_group()
      eg_element = ent_group.add_element()
      eg_element.set_type(kind)
      eg_element.set_name(entity_name)
  
      prop = entity_proto.add_property()
      prop.set_meaning(datastore_pb.Property.BYTESTRING)
      prop.set_name(prop_name)
      prop.set_multiple(1)
      val = prop.mutable_value()
      val.set_stringvalue(prop_value)
      return entity_proto
  
    @testing.gen_test
    def test_dynamic_put(self):
      db_batch = flexmock(session=flexmock())
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      entity_proto1 = self.get_new_entity_proto(
        "test", "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      entity_key1 = 'test\x00blah\x00test_kind:bob\x01'
      entity_proto2 = self.get_new_entity_proto(
        "test", "test_kind", "nancy", "prop1name", "prop2val", ns="blah")
      entity_key2 = 'test\x00blah\x00test_kind:nancy\x01'
      async_result = gen.Future()
      async_result.set_result({entity_key1: {}, entity_key2: {}})
  
      db_batch.should_receive('batch_get_entity').and_return(async_result)
      db_batch.should_receive('batch_mutate').and_return(ASYNC_NONE)
      transaction_manager = flexmock(
        create_transaction_id=lambda project, xg: 1,
        delete_transaction_id=lambda project, txid: None,
        set_groups=lambda project, txid, groups: None)
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      putreq_pb = datastore_pb.PutRequest()
      putreq_pb.add_entity()
      putreq_pb.mutable_entity(0).MergeFrom(entity_proto1)
      putreq_pb.add_entity()
      putreq_pb.mutable_entity(1).MergeFrom(entity_proto2)
  
      putresp_pb = datastore_pb.PutResponse()
  
      async_true = gen.Future()
      async_true.set_result(True)
      entity_lock = flexmock(EntityLock)
      entity_lock.should_receive('acquire').and_return(async_true)
      entity_lock.should_receive('release')
  
      flexmock(ScatteredAllocator).should_receive('next').\
        and_return(random.randint(1, 500))
  
      yield dd.dynamic_put('test', putreq_pb, putresp_pb)
      self.assertEquals(len(putresp_pb.key_list()), 2)
  
    @testing.gen_test
    def test_put_entities(self):
      app_id = 'test'
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      entity_proto1 = self.get_new_entity_proto(
        app_id, "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      entity_key1 = 'test\x00blah\x00test_kind:bob\x01'
      entity_proto2 = self.get_new_entity_proto(
        app_id, "test_kind", "nancy", "prop1name", "prop2val", ns="blah")
      entity_key2 = 'test\x00blah\x00test_kind:nancy\x01'
      entity_list = [entity_proto1, entity_proto2]
      async_result = gen.Future()
      async_result.set_result({entity_key1: {}, entity_key2: {}})
  
      db_batch.should_receive('batch_get_entity').and_return(async_result)
      db_batch.should_receive('batch_mutate').and_return(ASYNC_NONE)
      transaction_manager = flexmock(
        create_transaction_id=lambda project, xg: 1,
        delete_transaction_id=lambda project, txid: None,
        set_groups=lambda project, txid, groups: None)
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
  
      async_true = gen.Future()
      async_true.set_result(True)
      entity_lock = flexmock(EntityLock)
      entity_lock.should_receive('acquire').and_return(async_true)
      entity_lock.should_receive('release')
  
      yield dd.put_entities(app_id, entity_list)
  
    def test_acquire_locks_for_trans(self):
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(False).once()
      self.assertRaises(TypeError, dd.acquire_locks_for_trans, [1], 1)
  
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(False).and_return(False)
      self.assertRaises(TypeError, dd.acquire_locks_for_trans, [1], 1)
  
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True)
  
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True).and_return(False)
      flexmock(dd).should_receive("get_table_prefix").and_return("prefix").never()
      flexmock(dd).should_receive("get_root_key_from_entity_key").and_return("rootkey").never()
      self.assertEquals({}, dd.acquire_locks_for_trans([], 1))
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("acquire_lock").once()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      entity = flexmock()
      entity.should_receive("app").and_return("appid")
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True).and_return(True)
      flexmock(dd).should_receive("get_root_key_from_entity_key").and_return("rootkey").once()
      self.assertEquals({'rootkey':1}, dd.acquire_locks_for_trans([entity], 1))
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("acquire_lock").once().and_raise(ZKTransactionException)
      zookeeper.should_receive("notify_failed_transaction").once()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      entity = flexmock()
      entity.should_receive("app").and_return("appid")
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True).and_return(True)
      flexmock(dd).should_receive("get_root_key_from_entity_key").and_return("rootkey").once()
      self.assertRaises(ZKTransactionException, dd.acquire_locks_for_trans, [entity], 1)
  
    @testing.gen_test
    def test_delete_entities(self):
      app_id = 'test'
      entity_proto1 = self.get_new_entity_proto(
        app_id, "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      row_key = "test\x00blah\x00test_kind:bob\x01"
      row_values = {row_key: {APP_ENTITY_SCHEMA[0]: entity_proto1.Encode(),
                              APP_ENTITY_SCHEMA[1
  
  [... the rest is omitted ...]
files: AppDB/test/unit/test_datastore_server.py
files: #!/usr/bin/env python
  # Programmer: Navraj Chohan <nlake44@gmail.com>
  
  import datetime
  import random
  import sys
  import unittest
  
  from tornado import gen, testing
  
  from appscale.common import appscale_info
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore import dbconstants
  from appscale.datastore import utils
  from appscale.datastore.datastore_distributed import DatastoreDistributed
  from appscale.datastore.dbconstants import APP_ENTITY_SCHEMA
  from appscale.datastore.dbconstants import JOURNAL_SCHEMA
  from appscale.datastore.dbconstants import TOMBSTONE
  from appscale.datastore.cassandra_env.entity_id_allocator import\
    ScatteredAllocator
  
  from appscale.datastore.cassandra_env.cassandra_interface import (
    DatastoreProxy)
  from appscale.datastore.cassandra_env.utils import deletions_for_entity
  from appscale.datastore.cassandra_env.utils import index_deletions
  from appscale.datastore.cassandra_env.utils import mutations_for_entity
  
  from appscale.datastore.utils import (
    get_entity_key,
    get_entity_kind,
    get_index_key_from_params,
    get_index_kv_from_tuple,
    get_kind_key
  )
  
  from appscale.datastore.zkappscale.entity_lock import EntityLock
  from appscale.datastore.zkappscale.zktransaction import ZKTransactionException
  from cassandra.cluster import Cluster
  from flexmock import flexmock
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.ext import db
  
  
  class Item(db.Model):
    name = db.StringProperty(required = True)
  
  
  ASYNC_NONE = gen.Future()
  ASYNC_NONE.set_result(None)
  
  
  class TestDatastoreServer(testing.AsyncTestCase):
    """
    A set of test cases for the datastore server (datastore server v2)
    """
    BASIC_ENTITY = ['guestbook', 'Greeting', 'foo', 'content', 'hello world']
  
    def setUp(self, *args, **kwargs):
      super(TestDatastoreServer, self).setUp(*args, **kwargs)
  
  
  
    def get_zookeeper(self):
      zk_handle = flexmock(handler=flexmock(event_object=lambda: None,
                                            sleep_func=lambda: None,
                                            lock_object=lambda: None))
      zk_handle.should_receive('add_listener')
      zookeeper = flexmock(handle=zk_handle)
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      zookeeper.should_receive("increment_and_get_counter").and_return(0,1000)
      return zookeeper
  
    def test_get_entity_kind(self):
      item = Item(name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_entity_kind(key), "Item")
  
    def test_kind_key(self):
      item = Item(name="Dyan", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:0000000000\x01")
  
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item1)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:Bob\x01")
  
      item2 = Item(key_name="Frank", name="Frank", _app="hello", parent = item1)
      key = db.model_to_protobuf(item2)
      self.assertEquals(get_kind_key("howdy", key.key().path()),
             "howdy\x00Item\x01Item:Bob\x01Item:Frank\x01")
  
    def test_get_entity_key(self):
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(str(get_entity_key("howdy", key.key().path())), "howdy\x00Item:Bob\x01")
  
    def test_validate_key(self):
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      dd.validate_key(key.key())
  
    def test_get_table_prefix(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(dd.get_table_prefix(key), "hello\x00")
  
    def test_get_index_key_from_params(self):
      params = ['a','b','c','d','e']
      self.assertEquals(get_index_key_from_params(params),
                        "a\x00b\x00c\x00d\x00e")
  
    def test_get_index_kv_from_tuple(self):
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      item2 = Item(key_name="Sally", name="Sally", _app="hello")
      key1 = db.model_to_protobuf(item1)
      key2 = db.model_to_protobuf(item2)
      tuples_list = [("a\x00b",key1),("a\x00b",key2)]
      self.assertEquals(get_index_kv_from_tuple(
        tuples_list), (['a\x00b\x00Item\x00name\x00\x9aBob\x01\x01\x00Item:Bob\x01',
        'a\x00b\x00Item:Bob\x01'],
        ['a\x00b\x00Item\x00name\x00\x9aSally\x01\x01\x00Item:Sally\x01',
        'a\x00b\x00Item:Sally\x01']))
  
    def test_get_composite_index_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      dd = flexmock(dd)
  
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto("appid", "kind", "entity_name", "prop1", "value", ns="")
  
      self.assertEquals(
        dd.get_composite_index_key(composite_index, ent),
        "appid\x00\x00123\x00\x9avalue\x01\x01\x00\x00kind:entity_name\x01")
  
    @testing.gen_test
    def test_get_indices(self):
      session = flexmock(default_consistency_level=None)
      cluster = flexmock(connect=lambda keyspace: session)
      flexmock(appscale_info).should_receive('get_db_ips')
      flexmock(Cluster).new_instances(cluster)
      flexmock(DatastoreProxy).should_receive('range_query').and_return({})
      db_batch = DatastoreProxy()
  
      result = yield db_batch.get_indices("appid")
      self.assertEquals(result, [])
  
    @testing.gen_test
    def test_delete_composite_index_metadata(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_delete").and_return(ASYNC_NONE)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      dd = flexmock(dd)
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(1)
      yield dd.delete_composite_index_metadata("appid", composite_index)
  
    @testing.gen_test
    def test_create_composite_index(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(ASYNC_NONE)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      dd = flexmock(dd)
      index = entity_pb.CompositeIndex()
      index.set_app_id("appid")
      index.set_state(2)
      definition = index.mutable_definition()
      definition.set_entity_type("kind")
      definition.set_ancestor(0)
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      yield dd.create_composite_index("appid", index)
      assert index.id() > 0
  
    @testing.gen_test
    def test_insert_composite_indexes(self):
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto(
        "appid", "kind", "entity_name", "prop1", "value", ns="")
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(ASYNC_NONE).once()
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      yield dd.insert_composite_indexes([ent], [composite_index])
  
    @testing.gen_test
    def test_fetch_keys(self):
      entity_proto1 = self.get_new_entity_proto(
        "test", "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      async_get_1 = gen.Future()
      async_get_1.set_result({
        'test\x00blah\x00test_kind:bob\x01': {
          APP_ENTITY_SCHEMA[0]: entity_proto1.Encode(),
          APP_ENTITY_SCHEMA[1]: 1
        }
      })
      async_get_2 = gen.Future()
      async_get_2.set_result({
        'test\x00blah\x00test_kind:bob\x01\x000000000002': {
          JOURNAL_SCHEMA[0]: entity_proto1.Encode()
      }})
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      db_batch.should_receive("batch_delete").and_return(ASYNC_NONE)
      db_batch.should_receive("batch_put_entity").and_return(ASYNC_NONE)
      db_batch.should_receive("batch_get_entity").\
        and_return(async_get_1).\
        and_return(async_get_2)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
  
      fetched = yield dd.fetch_keys([entity_proto1.key()])
      self.assertEquals(fetched[0], {
        'test\x00blah\x00test_kind:bob\x01': {
         'txnID': 1, 'entity': entity_proto1.Encode()
        }
      })
      self.assertEqual(fetched[1], ['test\x00blah\x00test_kind:bob\x01'])
  
    @testing.gen_test
    def test_commit_transaction(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      transaction_manager = flexmock(
        delete_transaction_id=lambda project, txid: None)
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive('apply_txn_changes').and_return(ASYNC_NONE)
      commit_request = datastore_pb.Transaction()
      commit_request.set_handle(123)
      commit_request.set_app("aaa")
      http_request = commit_request.Encode()
      result = yield dd.commit_transaction("app_id", http_request)
      self.assertEquals(result, (datastore_pb.CommitResponse().Encode(), 0, ""))
  
    def test_rollback_transcation(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("release_lock").and_return(True)
      zookeeper.should_receive("notify_failed_transaction").and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      commit_request = datastore_pb.Transaction()
      commit_request.set_handle(123)
      commit_request.set_app("aaa")
      http_request = commit_request.Encode()
      self.assertEquals(dd.rollback_transaction("app_id", http_request),
                        (api_base_pb.VoidProto().Encode(), 0, ""))
  
    @staticmethod
    def get_new_entity_proto(app_id, kind, entity_name, prop_name, prop_value, ns=""):
      entity_proto = datastore_pb.EntityProto()
  
      reference = entity_proto.mutable_key()
      reference.set_app(app_id)
      reference.set_name_space(ns)
  
      path = reference.mutable_path()
      element = path.add_element()
      element.set_type(kind)
      element.set_name(entity_name)
  
      ent_group = entity_proto.mutable_entity_group()
      eg_element = ent_group.add_element()
      eg_element.set_type(kind)
      eg_element.set_name(entity_name)
  
      prop = entity_proto.add_property()
      prop.set_meaning(datastore_pb.Property.BYTESTRING)
      prop.set_name(prop_name)
      prop.set_multiple(1)
      val = prop.mutable_value()
      val.set_stringvalue(prop_value)
      return entity_proto
  
    @testing.gen_test
    def test_dynamic_put(self):
      db_batch = flexmock(session=flexmock())
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      entity_proto1 = self.get_new_entity_proto(
        "test", "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      entity_key1 = 'test\x00blah\x00test_kind:bob\x01'
      entity_proto2 = self.get_new_entity_proto(
        "test", "test_kind", "nancy", "prop1name", "prop2val", ns="blah")
      entity_key2 = 'test\x00blah\x00test_kind:nancy\x01'
      async_result = gen.Future()
      async_result.set_result({entity_key1: {}, entity_key2: {}})
  
      db_batch.should_receive('batch_get_entity').and_return(async_result)
      db_batch.should_receive('batch_mutate').and_return(ASYNC_NONE)
      transaction_manager = flexmock(
        create_transaction_id=lambda project, xg: 1,
        delete_transaction_id=lambda project, txid: None,
        set_groups=lambda project, txid, groups: None)
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
      putreq_pb = datastore_pb.PutRequest()
      putreq_pb.add_entity()
      putreq_pb.mutable_entity(0).MergeFrom(entity_proto1)
      putreq_pb.add_entity()
      putreq_pb.mutable_entity(1).MergeFrom(entity_proto2)
  
      putresp_pb = datastore_pb.PutResponse()
  
      async_true = gen.Future()
      async_true.set_result(True)
      entity_lock = flexmock(EntityLock)
      entity_lock.should_receive('acquire').and_return(async_true)
      entity_lock.should_receive('release')
  
      flexmock(ScatteredAllocator).should_receive('next').\
        and_return(random.randint(1, 500))
  
      yield dd.dynamic_put('test', putreq_pb, putresp_pb)
      self.assertEquals(len(putresp_pb.key_list()), 2)
  
    @testing.gen_test
    def test_put_entities(self):
      app_id = 'test'
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
  
      entity_proto1 = self.get_new_entity_proto(
        app_id, "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      entity_key1 = 'test\x00blah\x00test_kind:bob\x01'
      entity_proto2 = self.get_new_entity_proto(
        app_id, "test_kind", "nancy", "prop1name", "prop2val", ns="blah")
      entity_key2 = 'test\x00blah\x00test_kind:nancy\x01'
      entity_list = [entity_proto1, entity_proto2]
      async_result = gen.Future()
      async_result.set_result({entity_key1: {}, entity_key2: {}})
  
      db_batch.should_receive('batch_get_entity').and_return(async_result)
      db_batch.should_receive('batch_mutate').and_return(ASYNC_NONE)
      transaction_manager = flexmock(
        create_transaction_id=lambda project, xg: 1,
        delete_transaction_id=lambda project, txid: None,
        set_groups=lambda project, txid, groups: None)
      dd = DatastoreDistributed(db_batch, transaction_manager,
                                self.get_zookeeper())
  
      async_true = gen.Future()
      async_true.set_result(True)
      entity_lock = flexmock(EntityLock)
      entity_lock.should_receive('acquire').and_return(async_true)
      entity_lock.should_receive('release')
  
      yield dd.put_entities(app_id, entity_list)
  
    def test_acquire_locks_for_trans(self):
      zk_client = flexmock()
      zk_client.should_receive('add_listener')
  
      zookeeper = flexmock(handle=zk_client)
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version_sync').and_return(True)
      transaction_manager = flexmock()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(False).once()
      self.assertRaises(TypeError, dd.acquire_locks_for_trans, [1], 1)
  
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(False).and_return(False)
      self.assertRaises(TypeError, dd.acquire_locks_for_trans, [1], 1)
  
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True)
  
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True).and_return(False)
      flexmock(dd).should_receive("get_table_prefix").and_return("prefix").never()
      flexmock(dd).should_receive("get_root_key_from_entity_key").and_return("rootkey").never()
      self.assertEquals({}, dd.acquire_locks_for_trans([], 1))
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("acquire_lock").once()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      entity = flexmock()
      entity.should_receive("app").and_return("appid")
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True).and_return(True)
      flexmock(dd).should_receive("get_root_key_from_entity_key").and_return("rootkey").once()
      self.assertEquals({'rootkey':1}, dd.acquire_locks_for_trans([entity], 1))
  
      zookeeper = flexmock(handle=zk_client)
      zookeeper.should_receive("acquire_lock").once().and_raise(ZKTransactionException)
      zookeeper.should_receive("notify_failed_transaction").once()
      dd = DatastoreDistributed(db_batch, transaction_manager, zookeeper)
      entity = flexmock()
      entity.should_receive("app").and_return("appid")
      flexmock(dd).should_receive("is_instance_wrapper").and_return(True) \
        .and_return(True).and_return(True)
      flexmock(dd).should_receive("get_root_key_from_entity_key").and_return("rootkey").once()
      self.assertRaises(ZKTransactionException, dd.acquire_locks_for_trans, [entity], 1)
  
    @testing.gen_test
    def test_delete_entities(self):
      app_id = 'test'
      entity_proto1 = self.get_new_entity_proto(
        app_id, "test_kind", "bob", "prop1name", "prop1val", ns="blah")
      row_key = "test\x00blah\x00test_kind:bob\x01"
      row_values = {row_key: {APP_
  
  [... the rest is omitted ...]
files: AppDB/test/unit/test_datastore_server.py