Merge pull request #2684 from cdonati/persistent-group-locks

Allow group locks to persist when commits fail

AppDB/appscale/datastore/cassandra_env/cassandra_interface.py

@@ -20,6 +20,7 @@
 from cassandra.query import ConsistencyLevel
 from cassandra.query import SimpleStatement
 from cassandra.query import ValueSequence
+from .constants import CURRENT_VERSION
 from .large_batch import (FailedBatch,
                           LargeBatch)
 from .retry_policies import (BASIC_RETRIES,
@@ -53,9 +54,6 @@
 # The number of times to retry connecting to Cassandra.
 INITIAL_CONNECT_RETRIES = 20
 
-# The data layout version that the datastore expects.
-EXPECTED_DATA_VERSION = 1.0
-
 # The metadata key for the data layout version.
 VERSION_INFO_KEY = 'version'
 
@@ -96,6 +94,7 @@ class ThriftColumn(object):
 class IndexStates(object):
   """ Possible states for datastore indexes. """
   CLEAN = 'clean'
+  DIRTY = 'dirty'
   SCRUB_IN_PROGRESS = 'scrub_in_progress'
 
 
@@ -780,7 +779,7 @@ def valid_data_version(self):
     except cassandra.InvalidRequest:
       return False
 
-    return version is not None and float(version) == EXPECTED_DATA_VERSION
+    return version is not None and float(version) == CURRENT_VERSION
 
   def group_updates(self, groups):
     """ Fetch the latest transaction IDs for each group.

AppDB/appscale/datastore/cassandra_env/constants.py

@@ -0,0 +1,4 @@
+""" Datastore constants specific to the Cassandra implementation. """
+
+# The current data layout version.
+CURRENT_VERSION = 2.0

AppDB/appscale/datastore/cassandra_env/schema.py

@@ -17,11 +17,9 @@
 from .cassandra_interface import INITIAL_CONNECT_RETRIES
 from .cassandra_interface import KEYSPACE
 from .cassandra_interface import ThriftColumn
+from .constants import CURRENT_VERSION
 from .. import dbconstants
 
-# The data layout version to set after removing the journal table.
-POST_JOURNAL_VERSION = 1.0
-
 # A policy that does not retry statements.
 NO_RETRIES = FallthroughRetryPolicy()
 
@@ -110,6 +108,7 @@ def create_batch_tables(cluster, session):
     time.sleep(SCHEMA_CHANGE_TIMEOUT)
     raise
 
+
 def create_groups_table(session):
   create_table = """
     CREATE TABLE IF NOT EXISTS group_updates (
@@ -180,6 +179,32 @@ def create_entity_ids_table(session):
     raise
 
 
+def current_datastore_version(session):
+  """ Retrieves the existing datastore version value.
+
+  Args:
+    session: A cassandra-driver session.
+  Returns:
+    A float specifying the existing datastore version or None.
+  """
+  key = cassandra_interface.VERSION_INFO_KEY
+  statement = """
+    SELECT {value} FROM "{table}"
+    WHERE {key} = %s
+    AND {column} = %s
+  """.format(
+    value=ThriftColumn.VALUE,
+    table=dbconstants.DATASTORE_METADATA_TABLE,
+    key=ThriftColumn.KEY,
+    column=ThriftColumn.COLUMN_NAME
+  )
+  results = session.execute(statement, (bytearray(key), key))
+  try:
+    return float(results[0].value)
+  except IndexError:
+    return None
+
+
 def prime_cassandra(replication):
   """ Create Cassandra keyspace and initial tables.
 
@@ -268,10 +293,23 @@ def prime_cassandra(replication):
     value=ThriftColumn.VALUE
   )
 
-  if not existing_entities:
+  if existing_entities:
+    current_version = current_datastore_version(session)
+    if current_version == 1.0:
+      # Instruct the groomer to reclean the indexes.
+      parameters = {'key': bytearray(cassandra_interface.INDEX_STATE_KEY),
+                    'column': cassandra_interface.INDEX_STATE_KEY,
+                    'value': bytearray(str(IndexStates.DIRTY))}
+      session.execute(metadata_insert, parameters)
+
+      parameters = {'key': bytearray(cassandra_interface.VERSION_INFO_KEY),
+                    'column': cassandra_interface.VERSION_INFO_KEY,
+                    'value': bytearray(str(CURRENT_VERSION))}
+      session.execute(metadata_insert, parameters)
+  else:
     parameters = {'key': bytearray(cassandra_interface.VERSION_INFO_KEY),
                   'column': cassandra_interface.VERSION_INFO_KEY,
-                  'value': bytearray(str(POST_JOURNAL_VERSION))}
+                  'value': bytearray(str(CURRENT_VERSION))}
     session.execute(metadata_insert, parameters)
 
     # Mark the newly created indexes as clean.
@@ -283,7 +321,7 @@ def prime_cassandra(replication):
   # Indicate that the database has been successfully primed.
   parameters = {'key': bytearray(cassandra_interface.PRIMED_KEY),
                 'column': cassandra_interface.PRIMED_KEY,
-                'value': bytearray('true')}
+                'value': bytearray(str(CURRENT_VERSION))}
   session.execute(metadata_insert, parameters)
   logging.info('Cassandra is primed.')
 
@@ -300,6 +338,7 @@ def primed():
     return False
 
   try:
-    return db_access.get_metadata(cassandra_interface.PRIMED_KEY) == 'true'
+    primed_version = db_access.get_metadata(cassandra_interface.PRIMED_KEY)
+    return primed_version == str(CURRENT_VERSION)
   finally:
     db_access.close()