src/third_party/chromite/lib/cidb.py

   1 # Copyright 2014 The Chromium OS Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 """Continuous Integration Database Library."""
   6
   7 from __future__ import print_function
   8
   9 import glob
  10 import logging
  11 import os
  12 import re
  13 try:
  14   import sqlalchemy
  15   import sqlalchemy.exc
  16   import sqlalchemy.interfaces
  17   from sqlalchemy import MetaData
  18 except ImportError:
  19   raise AssertionError(
  20       'Unable to import sqlalchemy. Please install this package by running '
  21       '`sudo apt-get install python-sqlalchemy` or similar.')
  22
  23 from chromite.cbuildbot import constants
  24 from chromite.lib import retry_util
  25
  26 CIDB_MIGRATIONS_DIR = os.path.join(constants.CHROMITE_DIR, 'cidb',
  27                                    'migrations')
  28
  29 _RETRYABLE_OPERATIONAL_ERROR_CODES = (
  30     2006,   # Error code 2006 'MySQL server has gone away' indicates that
  31             # the connection used was closed or dropped
  32     2013,   # 'Lost connection to MySQL server during query'
  33             # TODO(akeshet): consider only retrying UPDATE queries against
  34             # this error code, not INSERT queries, since we don't know
  35             # whether the query completed before or after the connection
  36             # lost.
  37     2026,    # 'SSL connection error: unknown error number'
  38 )
  39
  40
  41 def _IsRetryableException(e):
  42   """Determine whether a query should be retried based on exception.
  43
  44   Intended for use as a handler for retry_util.
  45
  46   Args:
  47     e: The exception to be filtered.
  48
  49   Returns:
  50     True if the query should be retried, False otherwise.
  51   """
  52   if isinstance(e, sqlalchemy.exc.OperationalError):
  53     error_code = e.orig.args[0]
  54     if error_code in _RETRYABLE_OPERATIONAL_ERROR_CODES:
  55       return True
  56
  57   return False
  58
  59
  60 class DBException(Exception):
  61   """General exception class for this module."""
  62
  63
  64 class UnsupportedMethodException(DBException):
  65   """Raised when a call is made that the database does not support."""
  66
  67
  68 def minimum_schema(min_version):
  69   """Generate a decorator to specify a minimum schema version for a method.
  70
  71   This decorator should be applied only to instance methods of
  72   SchemaVersionedMySQLConnection objects.
  73   """
  74
  75   def decorator(f):
  76     def wrapper(self, *args, **kwargs):
  77       if self.schema_version < min_version:
  78         raise UnsupportedMethodException()
  79       return f(self, *args, **kwargs)
  80     return wrapper
  81   return decorator
  82
  83
  84 class StrictModeListener(sqlalchemy.interfaces.PoolListener):
  85   """This listener ensures that STRICT_ALL_TABLES for all connections."""
  86   # pylint: disable-msg=W0613
  87   def connect(self, dbapi_con, *args, **kwargs):
  88     cur = dbapi_con.cursor()
  89     cur.execute("SET SESSION sql_mode='STRICT_ALL_TABLES'")
  90     cur.close()
  91
  92
  93 class SchemaVersionedMySQLConnection(object):
  94   """Connection to a database that is aware of its schema version."""
  95
  96   SCHEMA_VERSION_TABLE_NAME = 'schemaVersionTable'
  97   SCHEMA_VERSION_COL = 'schemaVersion'
  98
  99   def __init__(self, db_name, db_migrations_dir, db_credentials_dir):
 100     """SchemaVersionedMySQLConnection constructor.
 101
 102     Args:
 103       db_name: Name of the database to connect to.
 104       db_migrations_dir: Absolute path to directory of migration scripts
 105                          for this database.
 106       db_credentials_dir: Absolute path to directory containing connection
 107                           information to the database. Specifically, this
 108                           directory should contain files names user.txt,
 109                           password.txt, host.txt, client-cert.pem,
 110                           client-key.pem, and server-ca.pem
 111     """
 112     # None, or a sqlalchemy.MetaData instance
 113     self._meta = None
 114
 115     # pid of process on which _engine was created
 116     self._engine_pid = None
 117
 118     self._engine = None
 119
 120     self.db_migrations_dir = db_migrations_dir
 121     self.db_credentials_dir = db_credentials_dir
 122     self.db_name = db_name
 123
 124     with open(os.path.join(db_credentials_dir, 'password.txt')) as f:
 125       password = f.read().strip()
 126     with open(os.path.join(db_credentials_dir, 'host.txt')) as f:
 127       host = f.read().strip()
 128     with open(os.path.join(db_credentials_dir, 'user.txt')) as f:
 129       user = f.read().strip()
 130
 131     cert = os.path.join(db_credentials_dir, 'client-cert.pem')
 132     key = os.path.join(db_credentials_dir, 'client-key.pem')
 133     ca = os.path.join(db_credentials_dir, 'server-ca.pem')
 134     self._ssl_args = {'ssl': {'cert': cert, 'key': key, 'ca': ca}}
 135
 136     connect_url = sqlalchemy.engine.url.URL('mysql', username=user,
 137                                             password=password,
 138                                             host=host)
 139
 140     # Create a temporary engine to connect to the mysql instance, and check if
 141     # a database named |db_name| exists. If not, create one. We use a temporary
 142     # engine here because the real engine will be opened with a default
 143     # database name given by |db_name|.
 144     temp_engine = sqlalchemy.create_engine(connect_url,
 145                                            connect_args=self._ssl_args,
 146                                            listeners=[StrictModeListener()])
 147     databases = temp_engine.execute('SHOW DATABASES').fetchall()
 148     if (db_name,) not in databases:
 149       temp_engine.execute('CREATE DATABASE %s' % db_name)
 150       logging.info('Created database %s', db_name)
 151
 152     temp_engine.dispose()
 153
 154     # Now create the persistent connection to the database named |db_name|.
 155     # If there is a schema version table, read the current schema version
 156     # from it. Otherwise, assume schema_version 0.
 157     self._connect_url = sqlalchemy.engine.url.URL('mysql', username=user,
 158                                                   password=password,
 159                                                   host=host, database=db_name)
 160
 161     self.schema_version = self.QuerySchemaVersion()
 162
 163     logging.info('Created a SchemaVersionedMySQLConnection, '
 164                  'sqlalchemy version %s', sqlalchemy.__version__)
 165
 166   def DropDatabase(self):
 167     """Delete all data and tables from database, and drop database.
 168
 169     Use with caution. All data in database will be deleted. Invalidates
 170     this database connection instance.
 171     """
 172     self._meta = None
 173     self._GetEngine().execute('DROP DATABASE %s' % self.db_name)
 174     self._InvalidateEngine()
 175
 176   def QuerySchemaVersion(self):
 177     """Query the database for its current schema version number.
 178
 179     Returns:
 180       The current schema version from the database's schema version table,
 181       as an integer, or 0 if the table is empty or nonexistent.
 182     """
 183     tables = self._GetEngine().execute('SHOW TABLES').fetchall()
 184     if (self.SCHEMA_VERSION_TABLE_NAME,) in tables:
 185       r = self._GetEngine().execute('SELECT MAX(%s) from %s' %
 186           (self.SCHEMA_VERSION_COL, self.SCHEMA_VERSION_TABLE_NAME))
 187       return r.fetchone()[0] or 0
 188     else:
 189       return 0
 190
 191   def _GetMigrationScripts(self):
 192     """Look for migration scripts and return their versions and paths."
 193
 194     Returns:
 195       A list of (schema_version, script_path) tuples of the migration
 196       scripts for this database, sorted in ascending schema_version order.
 197     """
 198     # Look for migration script files in the migration script directory,
 199     # with names of the form [number]*.sql, and sort these by number.
 200     migration_scripts = glob.glob(os.path.join(self.db_migrations_dir, '*.sql'))
 201     migrations = []
 202     for script in migration_scripts:
 203       match = re.match(r'([0-9]*).*', os.path.basename(script))
 204       if match:
 205         migrations.append((int(match.group(1)), script))
 206
 207     migrations.sort()
 208     return migrations
 209
 210   def ApplySchemaMigrations(self, maxVersion=None):
 211     """Apply pending migration scripts to database, in order.
 212
 213     Args:
 214       maxVersion: The highest version migration script to apply. If
 215                   unspecified, all migrations found will be applied.
 216     """
 217     migrations = self._GetMigrationScripts()
 218
 219     # Execute the migration scripts in order, asserting that each one
 220     # updates the schema version to the expected number. If maxVersion
 221     # is specified stop early.
 222     for (number, script) in migrations:
 223       if maxVersion is not None and number > maxVersion:
 224         break
 225
 226       if number > self.schema_version:
 227         # Invalidate self._meta, then run script and ensure that schema
 228         # version was increased.
 229         self._meta = None
 230         logging.info('Running migration script %s', script)
 231         self.RunQueryScript(script)
 232         self.schema_version = self.QuerySchemaVersion()
 233         if self.schema_version != number:
 234           raise DBException('Migration script %s did not update '
 235                             'schema version to %s as expected. ' % (number,
 236                                                                     script))
 237
 238   def RunQueryScript(self, script_path):
 239     """Run a .sql script file located at |script_path| on the database."""
 240     with open(script_path, 'r') as f:
 241       script = f.read()
 242     queries = [q.strip() for q in script.split(';') if q.strip()]
 243     for q in queries:
 244       self._GetEngine().execute(q)
 245
 246   def _ReflectToMetadata(self):
 247     """Use sqlalchemy reflection to construct MetaData model of database.
 248
 249     If self._meta is already populated, this does nothing.
 250     """
 251     if self._meta is not None:
 252       return
 253     self._meta = MetaData()
 254     self._meta.reflect(bind=self._GetEngine())
 255
 256   def _Insert(self, table, values):
 257     """Create and execute a one-row INSERT query.
 258
 259     Args:
 260       table: Table name to insert to.
 261       values: Dictionary of column values to insert.
 262
 263     Returns:
 264       Integer primary key of the inserted row.
 265     """
 266     self._ReflectToMetadata()
 267     ins = self._meta.tables[table].insert().values(values)
 268     r = self._Execute(ins)
 269     return r.inserted_primary_key[0]
 270
 271   def _InsertMany(self, table, values):
 272     """Create and execute an multi-row INSERT query.
 273
 274     Args:
 275       table: Table name to insert to.
 276       values: A list of value dictionaries to insert multiple rows.
 277
 278     Returns:
 279       The number of inserted rows.
 280     """
 281     # sqlalchemy 0.7 and prior has a bug in which it does not always
 282     # correctly unpack a list of rows to multi-insert if the list contains
 283     # only one item.
 284     if len(values) == 1:
 285       self._Insert(table, values[0])
 286       return 1
 287
 288     self._ReflectToMetadata()
 289     ins = self._meta.tables[table].insert().values(values)
 290     r = self._Execute(ins)
 291     return r.rowcount
 292
 293   def _GetPrimaryKey(self, table):
 294     """Gets the primary key column of |table|.
 295
 296     This function requires that the given table have a 1-column promary key.
 297
 298     Args:
 299       table: Name of table to primary key for.
 300
 301     Returns:
 302       A sqlalchemy.sql.schema.Column representing the primary key column.
 303
 304     Raises:
 305       DBException if the table does not have a single column primary key.
 306    """
 307     self._ReflectToMetadata()
 308     t = self._meta.tables[table]
 309
 310     # TODO(akeshet): between sqlalchemy 0.7 and 0.8, a breaking change was
 311     # made to how t.columns and t.primary_key are stored, and in sqlalchemy
 312     # 0.7 t.columns does not have a .values() method. Hence this clumsy way
 313     # of extracting the primary key column. Currently, our builders have 0.7
 314     # installed. Once we drop support for 0.7, this code can be simply replaced
 315     # by:
 316     # key_columns = t.primary_key.columns.values()
 317     col_names = t.columns.keys()
 318     cols = [t.columns[n] for n in col_names]
 319     key_columns = [c for c in cols if c.primary_key]
 320
 321     if len(key_columns) != 1:
 322       raise DBException('Table %s does not have a 1-column primary '
 323                         'key.' % table)
 324     return key_columns[0]
 325
 326   def _Update(self, table, row_id, values):
 327     """Create and execute an UPDATE query by primary key.
 328
 329     Args:
 330       table: Table name to update.
 331       row_id: Primary key value of row to update.
 332       values: Dictionary of column values to update.
 333
 334     Returns:
 335       The number of rows that were updated (0 or 1).
 336     """
 337     self._ReflectToMetadata()
 338     primary_key = self._GetPrimaryKey(table)
 339     upd = self._meta.tables[table].update().where(primary_key==row_id
 340                                                   ).values(values)
 341     r = self._Execute(upd)
 342     return r.rowcount
 343
 344   def _UpdateWhere(self, table, where, values):
 345     """Create and execute an update query with a custom where clause.
 346
 347     Args:
 348       table: Table name to update.
 349       where: Raw SQL for the where clause, in string form, e.g.
 350              'build_id = 1 and board = "tomato"'
 351       values: dictionary of column values to update.
 352
 353     Returns:
 354       The number of rows that were updated.
 355     """
 356     self._ReflectToMetadata()
 357     upd = self._meta.tables[table].update().where(where)
 358     r = self._Execute(upd, values)
 359     return r.rowcount
 360
 361   def _Select(self, table, row_id, columns):
 362     """Create and execute a one-row one-table SELECT query by primary key.
 363
 364     Args:
 365       table: Table name to select from.
 366       row_id: Primary key value of row to select.
 367       columns: List of column names to select.
 368
 369     Returns:
 370       A column name to column value dict for the row found, if a row was found.
 371       None if no row was.
 372     """
 373     self._ReflectToMetadata()
 374     primary_key = self._GetPrimaryKey(table)
 375     table_m = self._meta.tables[table]
 376     columns_m = [table_m.c[col_name] for col_name in columns]
 377     sel = sqlalchemy.sql.select(columns_m).where(primary_key==row_id)
 378     r = self._Execute(sel).fetchall()
 379     if r:
 380       assert len(r) == 1, 'Query by primary key returned more than 1 row.'
 381       return dict(zip(columns, r[0]))
 382     else:
 383       return None
 384
 385   def _SelectWhere(self, table, where, columns):
 386     """Create and execute a one-table SELECT query with a custom where clause.
 387
 388     Args:
 389       table: Table name to update.
 390       where: Raw SQL for the where clause, in string form, e.g.
 391              'build_id = 1 and board = "tomato"'
 392       columns: List of column names to select.
 393
 394     Returns:
 395       A list of column name to column value dictionaries each representing
 396       a row that was selected.
 397     """
 398     self._ReflectToMetadata()
 399     table_m = self._meta.tables[table]
 400     columns_m = [table_m.c[col_name] for col_name in columns]
 401     sel = sqlalchemy.sql.select(columns_m).where(where)
 402     r = self._Execute(sel)
 403     return [dict(zip(columns, values)) for values in r.fetchall()]
 404
 405   def _Execute(self, query, *args, **kwargs):
 406     """Execute a query using engine, with retires.
 407
 408     This method wraps execution of a query in retries that create a new
 409     engine in case the engine's connection has been dropped.
 410
 411     Args:
 412       query: Query to execute, of type string, or sqlalchemy.Executible,
 413              or other sqlalchemy-executible statement (see sqlalchemy
 414              docs).
 415       *args: Additional args passed along to .execute(...)
 416       **kwargs: Additional args passed along to .execute(...)
 417
 418     Returns:
 419       The result of .execute(...)
 420     """
 421     f = lambda: self._GetEngine().execute(query, *args, **kwargs)
 422     return retry_util.GenericRetry(
 423         handler=_IsRetryableException,
 424         max_retry=4,
 425         sleep=1,
 426         functor=f)
 427
 428   def _GetEngine(self):
 429     """Get the sqlalchemy engine for this process.
 430
 431     This method creates a new sqlalchemy engine if necessary, and
 432     returns an engine that is unique to this process.
 433
 434     Returns:
 435       An sqlalchemy.engine instance for this database.
 436     """
 437     pid = os.getpid()
 438     if pid == self._engine_pid and self._engine:
 439       return self._engine
 440     else:
 441       e = sqlalchemy.create_engine(self._connect_url,
 442                                    connect_args=self._ssl_args,
 443                                    listeners=[StrictModeListener()])
 444       self._engine = e
 445       self._engine_pid = pid
 446       logging.info('Created cidb engine %s@%s for pid %s', e.url.username,
 447                    e.url.host, pid)
 448       return self._engine
 449
 450   def _InvalidateEngine(self):
 451     """Dispose of an sqlalchemy engine."""
 452     try:
 453       pid = os.getpid()
 454       if pid == self._engine_pid and self._engine:
 455         self._engine.dispose()
 456     finally:
 457       self._engine = None
 458       self._meta = None
 459
 460
 461 class CIDBConnection(SchemaVersionedMySQLConnection):
 462   """Connection to a Continuous Integration database."""
 463   def __init__(self, db_credentials_dir):
 464     super(CIDBConnection, self).__init__('cidb', CIDB_MIGRATIONS_DIR,
 465                                          db_credentials_dir)
 466
 467   @minimum_schema(2)
 468   def InsertBuild(self, builder_name, waterfall, build_number,
 469                   build_config, bot_hostname,  master_build_id=None):
 470     """Insert a build row.
 471
 472     Args:
 473       builder_name: buildbot builder name.
 474       waterfall: buildbot waterfall name.
 475       build_number: buildbot build number.
 476       build_config: cbuildbot config of build
 477       bot_hostname: hostname of bot running the build
 478       master_build_id: (Optional) primary key of master build to this build.
 479     """
 480     return self._Insert('buildTable', {'builder_name': builder_name,
 481                                        'buildbot_generation':
 482                                          constants.BUILDBOT_GENERATION,
 483                                        'waterfall': waterfall,
 484                                        'build_number': build_number,
 485                                        'build_config' : build_config,
 486                                        'bot_hostname': bot_hostname,
 487                                        'start_time' :
 488                                            sqlalchemy.func.current_timestamp(),
 489                                        'master_build_id' : master_build_id}
 490                         )
 491
 492   @minimum_schema(3)
 493   def InsertCLActions(self, build_id, cl_actions):
 494     """Insert a list of |cl_actions|.
 495
 496     If |cl_actions| is empty, this function does nothing.
 497
 498     Args:
 499       build_id: primary key of build that performed these actions.
 500       cl_actions: A list of cl_action tuples.
 501
 502     Returns:
 503       Number of actions inserted.
 504     """
 505     if not cl_actions:
 506       return 0
 507
 508     values = []
 509     # TODO(akeshet): Refactor to use either cl action tuples out of the
 510     # metadata dict (as now) OR CLActionTuple objects.
 511     for cl_action in cl_actions:
 512       change_source = 'internal' if cl_action[0]['internal'] else 'external'
 513       change_number = cl_action[0]['gerrit_number']
 514       patch_number = cl_action[0]['patch_number']
 515       action = cl_action[1]
 516       reason = cl_action[3]
 517       values.append({
 518           'build_id' : build_id,
 519           'change_source' : change_source,
 520           'change_number': change_number,
 521           'patch_number' : patch_number,
 522           'action' : action,
 523           'reason' : reason})
 524
 525     return self._InsertMany('clActionTable', values)
 526
 527   @minimum_schema(4)
 528   def InsertBuildStage(self, build_id, stage_name, board, status,
 529                        log_url, duration_seconds, summary):
 530     """Insert a build stage into buildStageTable.
 531
 532     Args:
 533       build_id: id of responsible build
 534       stage_name: name of stage
 535       board: board that stage ran for
 536       status: 'pass' or 'fail'
 537       log_url: URL of stage log
 538       duration_seconds: run time of stage, in seconds
 539       summary: summary message of stage
 540
 541     Returns:
 542       Primary key of inserted stage.
 543     """
 544     return self._Insert('buildStageTable',
 545                         {'build_id': build_id,
 546                          'name': stage_name,
 547                          'board': board,
 548                          'status': status,
 549                          'log_url': log_url,
 550                          'duration_seconds': duration_seconds,
 551                          'summary': summary})
 552
 553   @minimum_schema(4)
 554   def InsertBuildStages(self, stages):
 555     """For testing only. Insert multiple build stages into buildStageTable.
 556
 557     This method allows integration tests to more quickly populate build
 558     stages into the database, from test data. Normal builder operations are
 559     expected to insert build stage rows one at a time, using InsertBuildStage.
 560
 561     Args:
 562       stages: A list of dictionaries, each dictionary containing keys
 563               build_id, name, board, status, log_url, duration_seconds, and
 564               summary.
 565
 566     Returns:
 567       The number of build stage rows inserted.
 568     """
 569     if not stages:
 570       return 0
 571     return self._InsertMany('buildStageTable',
 572                             stages)
 573
 574   @minimum_schema(6)
 575   def InsertBoardPerBuild(self, build_id, board):
 576     """Inserts a board-per-build entry into database.
 577
 578     Args:
 579       build_id: primary key of the build in the buildTable
 580       board: String board name.
 581     """
 582     self._Insert('boardPerBuildTable', {'build_id': build_id,
 583                                         'board': board})
 584
 585   @minimum_schema(7)
 586   def InsertChildConfigPerBuild(self, build_id, child_config):
 587     """Insert a child-config-per-build entry into database.
 588
 589     Args:
 590       build_id: primary key of the build in the buildTable
 591       child_config: String child_config name.
 592     """
 593     self._Insert('childConfigPerBuildTable', {'build_id': build_id,
 594                                               'child_config': child_config})
 595
 596   @minimum_schema(2)
 597   def UpdateMetadata(self, build_id, metadata):
 598     """Update the given metadata row in database.
 599
 600     Args:
 601       build_id: id of row to update.
 602       metadata: CBuildbotMetadata instance to update with.
 603
 604     Returns:
 605       The number of build rows that were updated (0 or 1).
 606     """
 607     d = metadata.GetDict()
 608     versions = d.get('version') or {}
 609     return self._Update('buildTable', build_id,
 610                         {'chrome_version': versions.get('chrome'),
 611                          'milestone_version': versions.get('milestone'),
 612                          'platform_version': versions.get('platform'),
 613                          'full_version': versions.get('full'),
 614                          'sdk_version': d.get('sdk-versions'),
 615                          'toolchain_url': d.get('toolchain-url'),
 616                          'build_type': d.get('build_type')})
 617
 618   @minimum_schema(6)
 619   def UpdateBoardPerBuildMetadata(self, build_id, board, board_metadata):
 620     """Update the given board-per-build metadata.
 621
 622     Args:
 623       build_id: id of the build
 624       board: board to update
 625       board_metadata: per-board metadata dict for this board
 626     """
 627     update_dict = {
 628         'main_firmware_version': board_metadata.get('main-firmware-version'),
 629         'ec_firmware_version': board_metadata.get('ec-firmware-version')
 630         }
 631     return self._UpdateWhere('boardPerBuildTable',
 632         'build_id = %s and board = "%s"' % (build_id, board),
 633         update_dict)
 634
 635
 636   @minimum_schema(11)
 637   def FinishBuild(self, build_id, status=None, status_pickle=None,
 638                   metadata_url=None):
 639     """Update the given build row, marking it as finished.
 640
 641     This should be called once per build, as the last update to the build.
 642     This will also mark the row's final=True.
 643
 644     Args:
 645       build_id: id of row to update.
 646       status: Final build status, one of
 647               manifest_version.BuilderStatus.COMPLETED_STATUSES.
 648       status_pickle: Pickled manifest_version.BuilderStatus.
 649       metadata_url: google storage url to metadata.json file for this build,
 650                     e.g. ('gs://chromeos-image-archive/master-paladin/'
 651                           'R39-6225.0.0-rc1/metadata.json')
 652     """
 653     self._ReflectToMetadata()
 654     # The current timestamp is evaluated on the database, not locally.
 655     current_timestamp = sqlalchemy.func.current_timestamp()
 656     self._Update('buildTable', build_id, {'finish_time' : current_timestamp,
 657                                           'status' : status,
 658                                           'status_pickle' : status_pickle,
 659                                           'metadata_url': metadata_url,
 660                                           'final' : True})
 661
 662   @minimum_schema(2)
 663   def GetBuildStatus(self, build_id):
 664     """Gets the status of the build.
 665
 666     Args:
 667       build_id: build id to fetch.
 668
 669     Returns:
 670       A dictionary with keys (id, build_config, start_time, finish_time,
 671       status), or None if no build with this id was found.
 672     """
 673     return self._Select('buildTable', build_id,
 674                         ['id', 'build_config', 'start_time',
 675                          'finish_time', 'status'])
 676
 677   @minimum_schema(2)
 678   def GetSlaveStatuses(self, master_build_id):
 679     """Gets the statuses of slave builders to given build.
 680
 681     Args:
 682       master_build_id: build id of the master build to fetch the slave
 683                        statuses for.
 684
 685     Returns:
 686       A list containing, for each slave build (row) found, a dictionary
 687       with keys (id, build_config, start_time, finish_time, status).
 688     """
 689     return self._SelectWhere('buildTable',
 690                              'master_build_id = %s' % master_build_id,
 691                              ['id', 'build_config', 'start_time',
 692                               'finish_time', 'status'])
 693
 694   @minimum_schema(11)
 695   def GetActionsForChange(self, change):
 696     """Gets all the actions for the given change.
 697
 698     Note, this includes all patches of the given change.
 699
 700     Args:
 701       change: A GerritChangeTuple or GerritPatchTuple specifing the
 702               change.
 703
 704     Returns:
 705       A list of actions, in timestamp order, each of which is a dict
 706       with keys (id, build_id, action, build_config, change_number,
 707                  patch_number, change_source, timestamp)
 708     """
 709     change_number = int(change.gerrit_number)
 710     change_source = 'internal' if change.internal else 'external'
 711     results = self._Execute(
 712         'SELECT c.id, b.id, action, build_config, change_number, '
 713         'patch_number, change_source, timestamp FROM '
 714         'clActionTable c JOIN buildTable b ON build_id = b.id '
 715         'WHERE change_number = %s AND change_source = "%s"' % (
 716         change_number, change_source)).fetchall()
 717     columns = ['id', 'build_id', 'action', 'build_config', 'change_number',
 718                'patch_number', 'change_source', 'timestamp']
 719     return [dict(zip(columns, values)) for values in results]
 720
 721
 722 class CIDBConnectionFactory(object):
 723   """Factory class used by builders to fetch the appropriate cidb connection"""
 724
 725   # A call to one of the Setup methods below is necessary before using the
 726   # GetCIDBConnectionForBuilder Factory. This ensures that unit tests do not
 727   # accidentally use one of the live database instances.
 728
 729   _ConnectionIsSetup = False
 730   _ConnectionType = None
 731   _ConnectionCredsPath = None
 732   _MockCIDB = None
 733   _CachedCIDB = None
 734
 735   _CONNECTION_TYPE_PROD = 'prod'   # production database
 736   _CONNECTION_TYPE_DEBUG = 'debug' # debug database, used by --debug builds
 737   _CONNECTION_TYPE_MOCK = 'mock'   # mock connection, not backed by database
 738   _CONNECTION_TYPE_NONE = 'none'   # explicitly no connection
 739   _CONNECTION_TYPE_INV = 'invalid' # invalidated connection
 740
 741
 742   @classmethod
 743   def IsCIDBSetup(cls):
 744     """Returns True iff GetCIDBConnectionForBuilder is ready to be called."""
 745     return cls._ConnectionIsSetup
 746
 747   @classmethod
 748   def GetCIDBConnectionType(cls):
 749     """Returns the type of db connection that is set up.
 750
 751     Returns:
 752       One of ('prod', 'debug', 'mock', 'none', 'invalid', None)
 753     """
 754     return cls._ConnectionType
 755
 756   @classmethod
 757   def InvalidateCIDBSetup(cls):
 758     """Invalidate the CIDB connection factory.
 759
 760     This method may be called at any time, even after a setup method. Once
 761     this is called, future calls to GetCIDBConnectionForBuilder will raise
 762     an assertion error.
 763     """
 764     cls._ConnectionType = cls._CONNECTION_TYPE_INV
 765     cls._CachedCIDB = None
 766
 767   @classmethod
 768   def SetupProdCidb(cls):
 769     """Sets up CIDB to use the prod instance of the database.
 770
 771     May be called only once, and may not be called after any other CIDB Setup
 772     method, otherwise it will raise an AssertionError.
 773     """
 774     assert not cls._ConnectionIsSetup, 'CIDB is already set up.'
 775     assert not cls._ConnectionType == cls._CONNECTION_TYPE_MOCK, (
 776         'CIDB set for mock use.')
 777     cls._ConnectionType = cls._CONNECTION_TYPE_PROD
 778     cls._ConnectionCredsPath = constants.CIDB_PROD_BOT_CREDS
 779     cls._ConnectionIsSetup = True
 780
 781
 782   @classmethod
 783   def SetupDebugCidb(cls):
 784     """Sets up CIDB to use the debug instance of the database.
 785
 786     May be called only once, and may not be called after any other CIDB Setup
 787     method, otherwise it will raise an AssertionError.
 788     """
 789     assert not cls._ConnectionIsSetup, 'CIDB is already set up.'
 790     assert not cls._ConnectionType == cls._CONNECTION_TYPE_MOCK, (
 791         'CIDB set for mock use.')
 792     cls._ConnectionType = cls._CONNECTION_TYPE_DEBUG
 793     cls._ConnectionCredsPath = constants.CIDB_DEBUG_BOT_CREDS
 794     cls._ConnectionIsSetup = True
 795
 796
 797   @classmethod
 798   def SetupMockCidb(cls, mock_cidb=None):
 799     """Sets up CIDB to use a mock object. May be called more than once.
 800
 801     Args:
 802       mock_cidb: (optional) The mock cidb object to be returned by
 803                  GetCIDBConnection. If not supplied, then CIDB will be
 804                  considered not set up, but future calls to set up a
 805                  non-(mock or None) connection will fail.
 806     """
 807     if cls._ConnectionIsSetup:
 808       assert cls._ConnectionType == cls._CONNECTION_TYPE_MOCK, (
 809           'A non-mock CIDB is already set up.')
 810     cls._ConnectionType = cls._CONNECTION_TYPE_MOCK
 811     if mock_cidb:
 812       cls._ConnectionIsSetup = True
 813       cls._MockCIDB = mock_cidb
 814
 815
 816   @classmethod
 817   def SetupNoCidb(cls):
 818     """Sets up CIDB to use an explicit None connection.
 819
 820     May be called more than once, or after SetupMockCidb.
 821     """
 822     if cls._ConnectionIsSetup:
 823       assert (cls._ConnectionType == cls._CONNECTION_TYPE_MOCK or
 824               cls._ConnectionType == cls._CONNECTION_TYPE_NONE) , (
 825           'A non-mock CIDB is already set up.')
 826     cls._ConnectionType = cls._CONNECTION_TYPE_NONE
 827     cls._ConnectionIsSetup = True
 828
 829
 830   @classmethod
 831   def GetCIDBConnectionForBuilder(cls):
 832     """Get a CIDBConnection.
 833
 834     A call to one of the CIDB Setup methods must have been made before calling
 835     this factory method.
 836
 837     Returns:
 838       A CIDBConnection instance connected to either the prod or debug
 839       instance of the database, or a mock connection, depending on which
 840       Setup method was called. Returns None if CIDB has been explicitly
 841       set up for that using SetupNoCidb.
 842     """
 843     assert cls._ConnectionIsSetup, 'CIDB has not be set up with a Setup call.'
 844     assert cls._ConnectionType != cls._CONNECTION_TYPE_INV, (
 845         'CIDB Connection factory has been invalidated.')
 846     if cls._ConnectionType == cls._CONNECTION_TYPE_MOCK:
 847       return cls._MockCIDB
 848     elif cls._ConnectionType == cls._CONNECTION_TYPE_NONE:
 849       return None
 850     else:
 851       if cls._CachedCIDB:
 852         return cls._CachedCIDB
 853       try:
 854         cls._CachedCIDB = CIDBConnection(cls._ConnectionCredsPath)
 855       except sqlalchemy.exc.OperationalError as e:
 856         logging.warn('Retrying to create a database connection, due to '
 857                      'exception %s.', e)
 858         cls._CachedCIDB = CIDBConnection(cls._ConnectionCredsPath)
 859       return cls._CachedCIDB
 860
 861   @classmethod
 862   def _ClearCIDBSetup(cls):
 863     """Clears the CIDB Setup state. For testing purposes only."""
 864     cls._ConnectionIsSetup = False
 865     cls._ConnectionType = None
 866     cls._ConnectionCredsPath = None
 867     cls._MockCIDB = None
 868     cls._CachedCIDB = None
 869
 870