python-swift 2.7.0-0ubuntu2 / usr / lib / python2.7 / dist-packages / swift / account / backend.py

# Copyright (c) 2010-2012 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Pluggable Back-end for Account Server
"""

from uuid import uuid4
import time
import six.moves.cPickle as pickle

import sqlite3

from swift.common.utils import Timestamp
from swift.common.db import DatabaseBroker, utf8encode

DATADIR = 'accounts'


POLICY_STAT_TRIGGER_SCRIPT = """
    CREATE TRIGGER container_insert_ps AFTER INSERT ON container
    BEGIN
        INSERT OR IGNORE INTO policy_stat
            (storage_policy_index, container_count, object_count, bytes_used)
            VALUES (new.storage_policy_index, 0, 0, 0);
        UPDATE policy_stat
        SET container_count = container_count + (1 - new.deleted),
            object_count = object_count + new.object_count,
            bytes_used = bytes_used + new.bytes_used
        WHERE storage_policy_index = new.storage_policy_index;
    END;
    CREATE TRIGGER container_delete_ps AFTER DELETE ON container
    BEGIN
        UPDATE policy_stat
        SET container_count = container_count - (1 - old.deleted),
            object_count = object_count - old.object_count,
            bytes_used = bytes_used - old.bytes_used
        WHERE storage_policy_index = old.storage_policy_index;
    END;

"""


class AccountBroker(DatabaseBroker):
    """Encapsulates working with an account database."""
    db_type = 'account'
    db_contains_type = 'container'
    db_reclaim_timestamp = 'delete_timestamp'

    def _initialize(self, conn, put_timestamp, **kwargs):
        """
        Create a brand new account database (tables, indices, triggers, etc.)

        :param conn: DB connection object
        :param put_timestamp: put timestamp
        """
        if not self.account:
            raise ValueError(
                'Attempting to create a new database with no account set')
        self.create_container_table(conn)
        self.create_account_stat_table(conn, put_timestamp)
        self.create_policy_stat_table(conn)

    def create_container_table(self, conn):
        """
        Create container table which is specific to the account DB.

        :param conn: DB connection object
        """
        conn.executescript("""
            CREATE TABLE container (
                ROWID INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT,
                put_timestamp TEXT,
                delete_timestamp TEXT,
                object_count INTEGER,
                bytes_used INTEGER,
                deleted INTEGER DEFAULT 0,
                storage_policy_index INTEGER DEFAULT 0
            );

            CREATE INDEX ix_container_deleted_name ON
                container (deleted, name);

            CREATE TRIGGER container_insert AFTER INSERT ON container
            BEGIN
                UPDATE account_stat
                SET container_count = container_count + (1 - new.deleted),
                    object_count = object_count + new.object_count,
                    bytes_used = bytes_used + new.bytes_used,
                    hash = chexor(hash, new.name,
                                  new.put_timestamp || '-' ||
                                    new.delete_timestamp || '-' ||
                                    new.object_count || '-' || new.bytes_used);
            END;

            CREATE TRIGGER container_update BEFORE UPDATE ON container
            BEGIN
                SELECT RAISE(FAIL, 'UPDATE not allowed; DELETE and INSERT');
            END;


            CREATE TRIGGER container_delete AFTER DELETE ON container
            BEGIN
                UPDATE account_stat
                SET container_count = container_count - (1 - old.deleted),
                    object_count = object_count - old.object_count,
                    bytes_used = bytes_used - old.bytes_used,
                    hash = chexor(hash, old.name,
                                  old.put_timestamp || '-' ||
                                    old.delete_timestamp || '-' ||
                                    old.object_count || '-' || old.bytes_used);
            END;
        """ + POLICY_STAT_TRIGGER_SCRIPT)

    def create_account_stat_table(self, conn, put_timestamp):
        """
        Create account_stat table which is specific to the account DB.
        Not a part of Pluggable Back-ends, internal to the baseline code.

        :param conn: DB connection object
        :param put_timestamp: put timestamp
        """
        conn.executescript("""
            CREATE TABLE account_stat (
                account TEXT,
                created_at TEXT,
                put_timestamp TEXT DEFAULT '0',
                delete_timestamp TEXT DEFAULT '0',
                container_count INTEGER,
                object_count INTEGER DEFAULT 0,
                bytes_used INTEGER DEFAULT 0,
                hash TEXT default '00000000000000000000000000000000',
                id TEXT,
                status TEXT DEFAULT '',
                status_changed_at TEXT DEFAULT '0',
                metadata TEXT DEFAULT ''
            );

            INSERT INTO account_stat (container_count) VALUES (0);
        """)

        conn.execute('''
            UPDATE account_stat SET account = ?, created_at = ?, id = ?,
                   put_timestamp = ?, status_changed_at = ?
            ''', (self.account, Timestamp(time.time()).internal, str(uuid4()),
                  put_timestamp, put_timestamp))

    def create_policy_stat_table(self, conn):
        """
        Create policy_stat table which is specific to the account DB.
        Not a part of Pluggable Back-ends, internal to the baseline code.

        :param conn: DB connection object
        """
        conn.executescript("""
            CREATE TABLE policy_stat (
                storage_policy_index INTEGER PRIMARY KEY,
                container_count INTEGER DEFAULT 0,
                object_count INTEGER DEFAULT 0,
                bytes_used INTEGER DEFAULT 0
            );
            INSERT OR IGNORE INTO policy_stat (
                storage_policy_index, container_count, object_count,
                bytes_used
            )
            SELECT 0, container_count, object_count, bytes_used
            FROM account_stat
            WHERE container_count > 0;
        """)

    def get_db_version(self, conn):
        if self._db_version == -1:
            self._db_version = 0
            for row in conn.execute('''
                    SELECT name FROM sqlite_master
                    WHERE name = 'ix_container_deleted_name' '''):
                self._db_version = 1
        return self._db_version

    def _delete_db(self, conn, timestamp, force=False):
        """
        Mark the DB as deleted.

        :param conn: DB connection object
        :param timestamp: timestamp to mark as deleted
        """
        conn.execute("""
            UPDATE account_stat
            SET delete_timestamp = ?,
                status = 'DELETED',
                status_changed_at = ?
            WHERE delete_timestamp < ? """, (timestamp, timestamp, timestamp))

    def _commit_puts_load(self, item_list, entry):
        """See :func:`swift.common.db.DatabaseBroker._commit_puts_load`"""
        loaded = pickle.loads(entry.decode('base64'))
        # check to see if the update includes policy_index or not
        (name, put_timestamp, delete_timestamp, object_count, bytes_used,
         deleted) = loaded[:6]
        if len(loaded) > 6:
            storage_policy_index = loaded[6]
        else:
            # legacy support during upgrade until first non legacy storage
            # policy is defined
            storage_policy_index = 0
        item_list.append(
            {'name': name,
             'put_timestamp': put_timestamp,
             'delete_timestamp': delete_timestamp,
             'object_count': object_count,
             'bytes_used': bytes_used,
             'deleted': deleted,
             'storage_policy_index': storage_policy_index})

    def empty(self):
        """
        Check if the account DB is empty.

        :returns: True if the database has no active containers.
        """
        self._commit_puts_stale_ok()
        with self.get() as conn:
            row = conn.execute(
                'SELECT container_count from account_stat').fetchone()
            return (row[0] == 0)

    def make_tuple_for_pickle(self, record):
        return (record['name'], record['put_timestamp'],
                record['delete_timestamp'], record['object_count'],
                record['bytes_used'], record['deleted'],
                record['storage_policy_index'])

    def put_container(self, name, put_timestamp, delete_timestamp,
                      object_count, bytes_used, storage_policy_index):
        """
        Create a container with the given attributes.

        :param name: name of the container to create
        :param put_timestamp: put_timestamp of the container to create
        :param delete_timestamp: delete_timestamp of the container to create
        :param object_count: number of objects in the container
        :param bytes_used: number of bytes used by the container
        :param storage_policy_index:  the storage policy for this container
        """
        if delete_timestamp > put_timestamp and \
                object_count in (None, '', 0, '0'):
            deleted = 1
        else:
            deleted = 0
        record = {'name': name, 'put_timestamp': put_timestamp,
                  'delete_timestamp': delete_timestamp,
                  'object_count': object_count,
                  'bytes_used': bytes_used,
                  'deleted': deleted,
                  'storage_policy_index': storage_policy_index}
        self.put_record(record)

    def _is_deleted_info(self, status, container_count, delete_timestamp,
                         put_timestamp):
        """
        Apply delete logic to database info.

        :returns: True if the DB is considered to be deleted, False otherwise
        """
        return status == 'DELETED' or (
            container_count in (None, '', 0, '0') and
            Timestamp(delete_timestamp) > Timestamp(put_timestamp))

    def _is_deleted(self, conn):
        """
        Check account_stat table and evaluate info.

        :param conn: database conn

        :returns: True if the DB is considered to be deleted, False otherwise
        """
        info = conn.execute('''
            SELECT put_timestamp, delete_timestamp, container_count, status
            FROM account_stat''').fetchone()
        return self._is_deleted_info(**info)

    def is_status_deleted(self):
        """Only returns true if the status field is set to DELETED."""
        with self.get() as conn:
            row = conn.execute('''
                SELECT put_timestamp, delete_timestamp, status
                FROM account_stat''').fetchone()
            return row['status'] == "DELETED" or (
                row['delete_timestamp'] > row['put_timestamp'])

    def get_policy_stats(self, do_migrations=False):
        """
        Get global policy stats for the account.

        :param do_migrations: boolean, if True the policy stat dicts will
                              always include the 'container_count' key;
                              otherwise it may be omitted on legacy databases
                              until they are migrated.

        :returns: dict of policy stats where the key is the policy index and
                  the value is a dictionary like {'object_count': M,
                  'bytes_used': N, 'container_count': L}
        """
        columns = [
            'storage_policy_index',
            'container_count',
            'object_count',
            'bytes_used',
        ]

        def run_query():
            return (conn.execute('''
                SELECT %s
                FROM policy_stat
                ''' % ', '.join(columns)).fetchall())

        self._commit_puts_stale_ok()
        info = []
        with self.get() as conn:
            try:
                info = run_query()
            except sqlite3.OperationalError as err:
                if "no such column: container_count" in str(err):
                    if do_migrations:
                        self._migrate_add_container_count(conn)
                    else:
                        columns.remove('container_count')
                    info = run_query()
                elif "no such table: policy_stat" not in str(err):
                    raise

        policy_stats = {}
        for row in info:
            stats = dict(row)
            key = stats.pop('storage_policy_index')
            policy_stats[key] = stats
        return policy_stats

    def get_info(self):
        """
        Get global data for the account.

        :returns: dict with keys: account, created_at, put_timestamp,
                  delete_timestamp, status_changed_at, container_count,
                  object_count, bytes_used, hash, id
        """
        self._commit_puts_stale_ok()
        with self.get() as conn:
            return dict(conn.execute('''
                SELECT account, created_at,  put_timestamp, delete_timestamp,
                       status_changed_at, container_count, object_count,
                       bytes_used, hash, id
                FROM account_stat
            ''').fetchone())

    def list_containers_iter(self, limit, marker, end_marker, prefix,
                             delimiter, reverse=False):
        """
        Get a list of containers sorted by name starting at marker onward, up
        to limit entries. Entries will begin with the prefix and will not have
        the delimiter after the prefix.

        :param limit: maximum number of entries to get
        :param marker: marker query
        :param end_marker: end marker query
        :param prefix: prefix query
        :param delimiter: delimiter for query
        :param reverse: reverse the result order.

        :returns: list of tuples of (name, object_count, bytes_used, 0)
        """
        delim_force_gte = False
        (marker, end_marker, prefix, delimiter) = utf8encode(
            marker, end_marker, prefix, delimiter)
        if reverse:
            # Reverse the markers if we are reversing the listing.
            marker, end_marker = end_marker, marker
        self._commit_puts_stale_ok()
        if delimiter and not prefix:
            prefix = ''
        if prefix:
            end_prefix = prefix[:-1] + chr(ord(prefix[-1]) + 1)
        orig_marker = marker
        with self.get() as conn:
            results = []
            while len(results) < limit:
                query = """
                    SELECT name, object_count, bytes_used, 0
                    FROM container
                    WHERE """
                query_args = []
                if end_marker and (not prefix or end_marker < end_prefix):
                    query += ' name < ? AND'
                    query_args.append(end_marker)
                elif prefix:
                    query += ' name < ? AND'
                    query_args.append(end_prefix)

                if delim_force_gte:
                    query += ' name >= ? AND'
                    query_args.append(marker)
                    # Always set back to False
                    delim_force_gte = False
                elif marker and marker >= prefix:
                    query += ' name > ? AND'
                    query_args.append(marker)
                elif prefix:
                    query += ' name >= ? AND'
                    query_args.append(prefix)
                if self.get_db_version(conn) < 1:
                    query += ' +deleted = 0'
                else:
                    query += ' deleted = 0'
                query += ' ORDER BY name %s LIMIT ?' % \
                         ('DESC' if reverse else '')
                query_args.append(limit - len(results))
                curs = conn.execute(query, query_args)
                curs.row_factory = None

                # Delimiters without a prefix is ignored, further if there
                # is no delimiter then we can simply return the result as
                # prefixes are now handled in the SQL statement.
                if prefix is None or not delimiter:
                    return [r for r in curs]

                # We have a delimiter and a prefix (possibly empty string) to
                # handle
                rowcount = 0
                for row in curs:
                    rowcount += 1
                    name = row[0]
                    if reverse:
                        end_marker = name
                    else:
                        marker = name

                    if len(results) >= limit:
                        curs.close()
                        return results
                    end = name.find(delimiter, len(prefix))
                    if end > 0:
                        if reverse:
                            end_marker = name[:end + 1]
                        else:
                            marker = name[:end] + chr(ord(delimiter) + 1)
                            # we want result to be inclusive of delim+1
                            delim_force_gte = True
                        dir_name = name[:end + 1]
                        if dir_name != orig_marker:
                            results.append([dir_name, 0, 0, 1])
                        curs.close()
                        break
                    results.append(row)
                if not rowcount:
                    break
            return results

    def merge_items(self, item_list, source=None):
        """
        Merge items into the container table.

        :param item_list: list of dictionaries of {'name', 'put_timestamp',
                          'delete_timestamp', 'object_count', 'bytes_used',
                          'deleted', 'storage_policy_index'}
        :param source: if defined, update incoming_sync with the source
        """
        def _really_merge_items(conn):
            max_rowid = -1
            curs = conn.cursor()
            for rec in item_list:
                rec.setdefault('storage_policy_index', 0)  # legacy
                record = [rec['name'], rec['put_timestamp'],
                          rec['delete_timestamp'], rec['object_count'],
                          rec['bytes_used'], rec['deleted'],
                          rec['storage_policy_index']]
                query = '''
                    SELECT name, put_timestamp, delete_timestamp,
                           object_count, bytes_used, deleted,
                           storage_policy_index
                    FROM container WHERE name = ?
                '''
                if self.get_db_version(conn) >= 1:
                    query += ' AND deleted IN (0, 1)'
                curs_row = curs.execute(query, (rec['name'],))
                curs_row.row_factory = None
                row = curs_row.fetchone()
                if row:
                    row = list(row)
                    for i in range(5):
                        if record[i] is None and row[i] is not None:
                            record[i] = row[i]
                    if row[1] > record[1]:  # Keep newest put_timestamp
                        record[1] = row[1]
                    if row[2] > record[2]:  # Keep newest delete_timestamp
                        record[2] = row[2]
                    # If deleted, mark as such
                    if record[2] > record[1] and \
                            record[3] in (None, '', 0, '0'):
                        record[5] = 1
                    else:
                        record[5] = 0
                curs.execute('''
                    DELETE FROM container WHERE name = ? AND
                                                deleted IN (0, 1)
                ''', (record[0],))
                curs.execute('''
                    INSERT INTO container (name, put_timestamp,
                        delete_timestamp, object_count, bytes_used,
                        deleted, storage_policy_index)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                ''', record)
                if source:
                    max_rowid = max(max_rowid, rec['ROWID'])
            if source:
                try:
                    curs.execute('''
                        INSERT INTO incoming_sync (sync_point, remote_id)
                        VALUES (?, ?)
                    ''', (max_rowid, source))
                except sqlite3.IntegrityError:
                    curs.execute('''
                        UPDATE incoming_sync
                        SET sync_point=max(?, sync_point)
                        WHERE remote_id=?
                    ''', (max_rowid, source))
            conn.commit()

        with self.get() as conn:
            # create the policy stat table if needed and add spi to container
            try:
                _really_merge_items(conn)
            except sqlite3.OperationalError as err:
                if 'no such column: storage_policy_index' not in str(err):
                    raise
                self._migrate_add_storage_policy_index(conn)
                _really_merge_items(conn)

    def _migrate_add_container_count(self, conn):
        """
        Add the container_count column to the 'policy_stat' table and
        update it

        :param conn: DB connection object
        """
        # add the container_count column
        curs = conn.cursor()
        curs.executescript('''
            DROP TRIGGER container_delete_ps;
            DROP TRIGGER container_insert_ps;
            ALTER TABLE policy_stat
            ADD COLUMN container_count INTEGER DEFAULT 0;
        ''' + POLICY_STAT_TRIGGER_SCRIPT)

        # keep the simple case simple, if there's only one entry in the
        # policy_stat table we just copy the total container count from the
        # account_stat table

        # if that triggers an update then the where changes <> 0 *would* exist
        # and the insert or replace from the count subqueries won't execute

        curs.executescript("""
        UPDATE policy_stat
        SET container_count = (
            SELECT container_count
            FROM account_stat)
        WHERE (
            SELECT COUNT(storage_policy_index)
            FROM policy_stat
        ) <= 1;

        INSERT OR REPLACE INTO policy_stat (
            storage_policy_index,
            container_count,
            object_count,
            bytes_used
        )
        SELECT p.storage_policy_index,
               c.count,
               p.object_count,
               p.bytes_used
        FROM (
            SELECT storage_policy_index,
                   COUNT(*) as count
            FROM container
            WHERE deleted = 0
            GROUP BY storage_policy_index
        ) c
        JOIN policy_stat p
        ON p.storage_policy_index = c.storage_policy_index
        WHERE NOT EXISTS(
            SELECT changes() as change
            FROM policy_stat
            WHERE change <> 0
        );
        """)
        conn.commit()

    def _migrate_add_storage_policy_index(self, conn):
        """
        Add the storage_policy_index column to the 'container' table and
        set up triggers, creating the policy_stat table if needed.

        :param conn: DB connection object
        """
        try:
            self.create_policy_stat_table(conn)
        except sqlite3.OperationalError as err:
            if 'table policy_stat already exists' not in str(err):
                raise
        conn.executescript('''
            ALTER TABLE container
            ADD COLUMN storage_policy_index INTEGER DEFAULT 0;
        ''' + POLICY_STAT_TRIGGER_SCRIPT)