DB migration 25->24 failed when dropping column

"cinder-manage db sync 24" failed when dropping column cgsnapshot_id
from the snapshots table.

The reason that drop column failed was because of the foreign key
constraint. MySQL cannot drop column until the foreign key constraint
is removed. So the solution is to remove the foreign key first, and
then drop the column. This affects the cgsnapshot_id column in the
snapshots table and the consistencygroup_id column in the volumes table.

Change-Id: I89d5016be450ec91c7a7b1c42803add3f62c88df
Closes-Bug: #1368213

cinder/db/sqlalchemy/migrate_repo/versions/025_add_consistencygroup.py

@@ -13,6 +13,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from migrate import ForeignKeyConstraint
 from sqlalchemy import Boolean, Column, DateTime
 from sqlalchemy import ForeignKey, MetaData, String, Table
 
@@ -109,11 +110,45 @@ def downgrade(migrate_engine):
     meta.bind = migrate_engine
 
     # Drop column from snapshots table
+    if migrate_engine.name == 'mysql':
+        # MySQL cannot drop column cgsnapshot_id until the foreign key
+        # constraint is removed. So remove the foreign key first, and
+        # then drop the column.
+        table = Table('snapshots', meta, autoload=True)
+        ref_table = Table('snapshots', meta, autoload=True)
+        params = {'columns': [table.c['cgsnapshot_id']],
+                  'refcolumns': [ref_table.c['id']],
+                  'name': 'snapshots_ibfk_1'}
+
+        try:
+            fkey = ForeignKeyConstraint(**params)
+            fkey.drop()
+        except Exception:
+            LOG.error(_("Dropping foreign key 'cgsnapshot_id' in "
+                        "the 'snapshots' table failed."))
+
     snapshots = Table('snapshots', meta, autoload=True)
     cgsnapshot_id = snapshots.columns.cgsnapshot_id
     snapshots.drop_column(cgsnapshot_id)
 
     # Drop column from volumes table
+    if migrate_engine.name == 'mysql':
+        # MySQL cannot drop column consistencygroup_id until the foreign
+        # key constraint is removed. So remove the foreign key first,
+        # and then drop the column.
+        table = Table('volumes', meta, autoload=True)
+        ref_table = Table('volumes', meta, autoload=True)
+        params = {'columns': [table.c['consistencygroup_id']],
+                  'refcolumns': [ref_table.c['id']],
+                  'name': 'volumes_ibfk_1'}
+
+        try:
+            fkey = ForeignKeyConstraint(**params)
+            fkey.drop()
+        except Exception:
+            LOG.error(_("Dropping foreign key 'consistencygroup_id' in "
+                        "the 'volumes' table failed."))
+
     volumes = Table('volumes', meta, autoload=True)
     consistencygroup_id = volumes.columns.consistencygroup_id
     volumes.drop_column(consistencygroup_id)

cinder/tests/test_migrations.py

@@ -1218,6 +1218,18 @@ class TestMigrations(test.TestCase):
             self.assertIsInstance(cgsnapshots.c.status.type,
                                   sqlalchemy.types.VARCHAR)
 
+            # Verify foreign keys are created
+            fkey, = volumes.c.consistencygroup_id.foreign_keys
+            self.assertEqual(consistencygroups.c.id, fkey.column)
+            self.assertEqual(1, len(volumes.foreign_keys))
+
+            fkey, = snapshots.c.cgsnapshot_id.foreign_keys
+            self.assertEqual(cgsnapshots.c.id, fkey.column)
+            fkey, = snapshots.c.volume_id.foreign_keys
+            self.assertEqual(volumes.c.id, fkey.column)
+            # 2 foreign keys in Table snapshots
+            self.assertEqual(2, len(snapshots.foreign_keys))
+
             # Downgrade
             migration_api.downgrade(engine, TestMigrations.REPOSITORY, 24)
             metadata = sqlalchemy.schema.MetaData()
@@ -1235,6 +1247,13 @@ class TestMigrations(test.TestCase):
                                          autoload=True)
             self.assertNotIn('cgsnapshot_id', snapshots.c)
 
+            # Verify foreign keys are removed
+            self.assertEqual(0, len(volumes.foreign_keys))
+            self.assertEqual(1, len(snapshots.foreign_keys))
+            # volume_id foreign key is still in Table snapshots
+            fkey, = snapshots.c.volume_id.foreign_keys
+            self.assertEqual(volumes.c.id, fkey.column)
+
             # Test Table cgsnapshots doesn't exist any more
             self.assertFalse(engine.dialect.has_table(engine.connect(),
                                                       "cgsnapshots"))