From c961cd7736050ace8dfb18663ef678c64f2156a5 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 27 Aug 2015 13:01:38 +0100 Subject: Clean up scripts/ --- docs/postgres.rst | 11 +- scripts/database-prepare-for-0.0.1.sh | 21 - scripts/database-prepare-for-0.5.0.sh | 21 - scripts/port_from_sqlite_to_postgres.py | 760 -------------------------------- scripts/synapse_port_db | 760 ++++++++++++++++++++++++++++++++ scripts/upgrade_db_to_v0.6.0.py | 331 -------------- 6 files changed, 764 insertions(+), 1140 deletions(-) delete mode 100755 scripts/database-prepare-for-0.0.1.sh delete mode 100755 scripts/database-prepare-for-0.5.0.sh delete mode 100755 scripts/port_from_sqlite_to_postgres.py create mode 100755 scripts/synapse_port_db delete mode 100755 scripts/upgrade_db_to_v0.6.0.py diff --git a/docs/postgres.rst b/docs/postgres.rst index 19d8391115..017b3c47e7 100644 --- a/docs/postgres.rst +++ b/docs/postgres.rst @@ -55,9 +55,8 @@ Porting from SQLite Overview ~~~~~~~~ -The script ``port_from_sqlite_to_postgres.py`` allows porting an existing -synapse server backed by SQLite to using PostgreSQL. This is done in as a two -phase process: +The script ``synapse_port_db`` allows porting an existing synapse server +backed by SQLite to using PostgreSQL. This is done in as a two phase process: 1. Copy the existing SQLite database to a separate location (while the server is down) and running the port script against that offline database. @@ -86,8 +85,7 @@ Assuming your new config file (as described in the section *Synapse config*) is named ``homeserver-postgres.yaml`` and the SQLite snapshot is at ``homeserver.db.snapshot`` then simply run:: - python scripts/port_from_sqlite_to_postgres.py \ - --sqlite-database homeserver.db.snapshot \ + synapse_port_db --sqlite-database homeserver.db.snapshot \ --postgres-config homeserver-postgres.yaml The flag ``--curses`` displays a coloured curses progress UI. @@ -100,8 +98,7 @@ To complete the conversion shut down the synapse server and run the port script one last time, e.g. if the SQLite database is at ``homeserver.db`` run:: - python scripts/port_from_sqlite_to_postgres.py \ - --sqlite-database homeserver.db \ + synapse_port_db.py --sqlite-database homeserver.db \ --postgres-config database_config.yaml Once that has completed, change the synapse config to point at the PostgreSQL diff --git a/scripts/database-prepare-for-0.0.1.sh b/scripts/database-prepare-for-0.0.1.sh deleted file mode 100755 index 43d759a5cd..0000000000 --- a/scripts/database-prepare-for-0.0.1.sh +++ /dev/null @@ -1,21 +0,0 @@ -#!/bin/bash - -# This is will prepare a synapse database for running with v0.0.1 of synapse. -# It will store all the user information, but will *delete* all messages and -# room data. - -set -e - -cp "$1" "$1.bak" - -DUMP=$(sqlite3 "$1" << 'EOF' -.dump users -.dump access_tokens -.dump presence -.dump profiles -EOF -) - -rm "$1" - -sqlite3 "$1" <<< "$DUMP" diff --git a/scripts/database-prepare-for-0.5.0.sh b/scripts/database-prepare-for-0.5.0.sh deleted file mode 100755 index e824cb583e..0000000000 --- a/scripts/database-prepare-for-0.5.0.sh +++ /dev/null @@ -1,21 +0,0 @@ -#!/bin/bash - -# This is will prepare a synapse database for running with v0.5.0 of synapse. -# It will store all the user information, but will *delete* all messages and -# room data. - -set -e - -cp "$1" "$1.bak" - -DUMP=$(sqlite3 "$1" << 'EOF' -.dump users -.dump access_tokens -.dump presence -.dump profiles -EOF -) - -rm "$1" - -sqlite3 "$1" <<< "$DUMP" diff --git a/scripts/port_from_sqlite_to_postgres.py b/scripts/port_from_sqlite_to_postgres.py deleted file mode 100755 index e7ed4c309b..0000000000 --- a/scripts/port_from_sqlite_to_postgres.py +++ /dev/null @@ -1,760 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -# Copyright 2015 OpenMarket Ltd -# -# 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. - -from twisted.internet import defer, reactor -from twisted.enterprise import adbapi - -from synapse.storage._base import LoggingTransaction, SQLBaseStore -from synapse.storage.engines import create_engine - -import argparse -import curses -import logging -import sys -import time -import traceback -import yaml - - -logger = logging.getLogger("port_from_sqlite_to_postgres") - - -BOOLEAN_COLUMNS = { - "events": ["processed", "outlier"], - "rooms": ["is_public"], - "event_edges": ["is_state"], - "presence_list": ["accepted"], -} - - -APPEND_ONLY_TABLES = [ - "event_content_hashes", - "event_reference_hashes", - "event_signatures", - "event_edge_hashes", - "events", - "event_json", - "state_events", - "room_memberships", - "feedback", - "topics", - "room_names", - "rooms", - "local_media_repository", - "local_media_repository_thumbnails", - "remote_media_cache", - "remote_media_cache_thumbnails", - "redactions", - "event_edges", - "event_auth", - "received_transactions", - "sent_transactions", - "transaction_id_to_pdu", - "users", - "state_groups", - "state_groups_state", - "event_to_state_groups", - "rejections", -] - - -end_error_exec_info = None - - -class Store(object): - """This object is used to pull out some of the convenience API from the - Storage layer. - - *All* database interactions should go through this object. - """ - def __init__(self, db_pool, engine): - self.db_pool = db_pool - self.database_engine = engine - - _simple_insert_txn = SQLBaseStore.__dict__["_simple_insert_txn"] - _simple_insert = SQLBaseStore.__dict__["_simple_insert"] - - _simple_select_onecol_txn = SQLBaseStore.__dict__["_simple_select_onecol_txn"] - _simple_select_onecol = SQLBaseStore.__dict__["_simple_select_onecol"] - _simple_select_one_onecol = SQLBaseStore.__dict__["_simple_select_one_onecol"] - _simple_select_one_onecol_txn = SQLBaseStore.__dict__["_simple_select_one_onecol_txn"] - - _simple_update_one = SQLBaseStore.__dict__["_simple_update_one"] - _simple_update_one_txn = SQLBaseStore.__dict__["_simple_update_one_txn"] - - _execute_and_decode = SQLBaseStore.__dict__["_execute_and_decode"] - - def runInteraction(self, desc, func, *args, **kwargs): - def r(conn): - try: - i = 0 - N = 5 - while True: - try: - txn = conn.cursor() - return func( - LoggingTransaction(txn, desc, self.database_engine, []), - *args, **kwargs - ) - except self.database_engine.module.DatabaseError as e: - if self.database_engine.is_deadlock(e): - logger.warn("[TXN DEADLOCK] {%s} %d/%d", desc, i, N) - if i < N: - i += 1 - conn.rollback() - continue - raise - except Exception as e: - logger.debug("[TXN FAIL] {%s} %s", desc, e) - raise - - return self.db_pool.runWithConnection(r) - - def execute(self, f, *args, **kwargs): - return self.runInteraction(f.__name__, f, *args, **kwargs) - - def execute_sql(self, sql, *args): - def r(txn): - txn.execute(sql, args) - return txn.fetchall() - return self.runInteraction("execute_sql", r) - - def insert_many_txn(self, txn, table, headers, rows): - sql = "INSERT INTO %s (%s) VALUES (%s)" % ( - table, - ", ".join(k for k in headers), - ", ".join("%s" for _ in headers) - ) - - try: - txn.executemany(sql, rows) - except: - logger.exception( - "Failed to insert: %s", - table, - ) - raise - - -class Porter(object): - def __init__(self, **kwargs): - self.__dict__.update(kwargs) - - @defer.inlineCallbacks - def setup_table(self, table): - if table in APPEND_ONLY_TABLES: - # It's safe to just carry on inserting. - next_chunk = yield self.postgres_store._simple_select_one_onecol( - table="port_from_sqlite3", - keyvalues={"table_name": table}, - retcol="rowid", - allow_none=True, - ) - - total_to_port = None - if next_chunk is None: - if table == "sent_transactions": - next_chunk, already_ported, total_to_port = ( - yield self._setup_sent_transactions() - ) - else: - yield self.postgres_store._simple_insert( - table="port_from_sqlite3", - values={"table_name": table, "rowid": 1} - ) - - next_chunk = 1 - already_ported = 0 - - if total_to_port is None: - already_ported, total_to_port = yield self._get_total_count_to_port( - table, next_chunk - ) - else: - def delete_all(txn): - txn.execute( - "DELETE FROM port_from_sqlite3 WHERE table_name = %s", - (table,) - ) - txn.execute("TRUNCATE %s CASCADE" % (table,)) - - yield self.postgres_store.execute(delete_all) - - yield self.postgres_store._simple_insert( - table="port_from_sqlite3", - values={"table_name": table, "rowid": 0} - ) - - next_chunk = 1 - - already_ported, total_to_port = yield self._get_total_count_to_port( - table, next_chunk - ) - - defer.returnValue((table, already_ported, total_to_port, next_chunk)) - - @defer.inlineCallbacks - def handle_table(self, table, postgres_size, table_size, next_chunk): - if not table_size: - return - - self.progress.add_table(table, postgres_size, table_size) - - select = ( - "SELECT rowid, * FROM %s WHERE rowid >= ? ORDER BY rowid LIMIT ?" - % (table,) - ) - - while True: - def r(txn): - txn.execute(select, (next_chunk, self.batch_size,)) - rows = txn.fetchall() - headers = [column[0] for column in txn.description] - - return headers, rows - - headers, rows = yield self.sqlite_store.runInteraction("select", r) - - if rows: - next_chunk = rows[-1][0] + 1 - - self._convert_rows(table, headers, rows) - - def insert(txn): - self.postgres_store.insert_many_txn( - txn, table, headers[1:], rows - ) - - self.postgres_store._simple_update_one_txn( - txn, - table="port_from_sqlite3", - keyvalues={"table_name": table}, - updatevalues={"rowid": next_chunk}, - ) - - yield self.postgres_store.execute(insert) - - postgres_size += len(rows) - - self.progress.update(table, postgres_size) - else: - return - - def setup_db(self, db_config, database_engine): - db_conn = database_engine.module.connect( - **{ - k: v for k, v in db_config.get("args", {}).items() - if not k.startswith("cp_") - } - ) - - database_engine.prepare_database(db_conn) - - db_conn.commit() - - @defer.inlineCallbacks - def run(self): - try: - sqlite_db_pool = adbapi.ConnectionPool( - self.sqlite_config["name"], - **self.sqlite_config["args"] - ) - - postgres_db_pool = adbapi.ConnectionPool( - self.postgres_config["name"], - **self.postgres_config["args"] - ) - - sqlite_engine = create_engine("sqlite3") - postgres_engine = create_engine("psycopg2") - - self.sqlite_store = Store(sqlite_db_pool, sqlite_engine) - self.postgres_store = Store(postgres_db_pool, postgres_engine) - - yield self.postgres_store.execute( - postgres_engine.check_database - ) - - # Step 1. Set up databases. - self.progress.set_state("Preparing SQLite3") - self.setup_db(sqlite_config, sqlite_engine) - - self.progress.set_state("Preparing PostgreSQL") - self.setup_db(postgres_config, postgres_engine) - - # Step 2. Get tables. - self.progress.set_state("Fetching tables") - sqlite_tables = yield self.sqlite_store._simple_select_onecol( - table="sqlite_master", - keyvalues={ - "type": "table", - }, - retcol="name", - ) - - postgres_tables = yield self.postgres_store._simple_select_onecol( - table="information_schema.tables", - keyvalues={ - "table_schema": "public", - }, - retcol="distinct table_name", - ) - - tables = set(sqlite_tables) & set(postgres_tables) - - self.progress.set_state("Creating tables") - - logger.info("Found %d tables", len(tables)) - - def create_port_table(txn): - txn.execute( - "CREATE TABLE port_from_sqlite3 (" - " table_name varchar(100) NOT NULL UNIQUE," - " rowid bigint NOT NULL" - ")" - ) - - try: - yield self.postgres_store.runInteraction( - "create_port_table", create_port_table - ) - except Exception as e: - logger.info("Failed to create port table: %s", e) - - self.progress.set_state("Setting up") - - # Set up tables. - setup_res = yield defer.gatherResults( - [ - self.setup_table(table) - for table in tables - if table not in ["schema_version", "applied_schema_deltas"] - and not table.startswith("sqlite_") - ], - consumeErrors=True, - ) - - # Process tables. - yield defer.gatherResults( - [ - self.handle_table(*res) - for res in setup_res - ], - consumeErrors=True, - ) - - self.progress.done() - except: - global end_error_exec_info - end_error_exec_info = sys.exc_info() - logger.exception("") - finally: - reactor.stop() - - def _convert_rows(self, table, headers, rows): - bool_col_names = BOOLEAN_COLUMNS.get(table, []) - - bool_cols = [ - i for i, h in enumerate(headers) if h in bool_col_names - ] - - def conv(j, col): - if j in bool_cols: - return bool(col) - return col - - for i, row in enumerate(rows): - rows[i] = tuple( - conv(j, col) - for j, col in enumerate(row) - if j > 0 - ) - - @defer.inlineCallbacks - def _setup_sent_transactions(self): - # Only save things from the last day - yesterday = int(time.time()*1000) - 86400000 - - # And save the max transaction id from each destination - select = ( - "SELECT rowid, * FROM sent_transactions WHERE rowid IN (" - "SELECT max(rowid) FROM sent_transactions" - " GROUP BY destination" - ")" - ) - - def r(txn): - txn.execute(select) - rows = txn.fetchall() - headers = [column[0] for column in txn.description] - - ts_ind = headers.index('ts') - - return headers, [r for r in rows if r[ts_ind] < yesterday] - - headers, rows = yield self.sqlite_store.runInteraction( - "select", r, - ) - - self._convert_rows("sent_transactions", headers, rows) - - inserted_rows = len(rows) - max_inserted_rowid = max(r[0] for r in rows) - - def insert(txn): - self.postgres_store.insert_many_txn( - txn, "sent_transactions", headers[1:], rows - ) - - yield self.postgres_store.execute(insert) - - def get_start_id(txn): - txn.execute( - "SELECT rowid FROM sent_transactions WHERE ts >= ?" - " ORDER BY rowid ASC LIMIT 1", - (yesterday,) - ) - - rows = txn.fetchall() - if rows: - return rows[0][0] - else: - return 1 - - next_chunk = yield self.sqlite_store.execute(get_start_id) - next_chunk = max(max_inserted_rowid + 1, next_chunk) - - yield self.postgres_store._simple_insert( - table="port_from_sqlite3", - values={"table_name": "sent_transactions", "rowid": next_chunk} - ) - - def get_sent_table_size(txn): - txn.execute( - "SELECT count(*) FROM sent_transactions" - " WHERE ts >= ?", - (yesterday,) - ) - size, = txn.fetchone() - return int(size) - - remaining_count = yield self.sqlite_store.execute( - get_sent_table_size - ) - - total_count = remaining_count + inserted_rows - - defer.returnValue((next_chunk, inserted_rows, total_count)) - - @defer.inlineCallbacks - def _get_remaining_count_to_port(self, table, next_chunk): - rows = yield self.sqlite_store.execute_sql( - "SELECT count(*) FROM %s WHERE rowid >= ?" % (table,), - next_chunk, - ) - - defer.returnValue(rows[0][0]) - - @defer.inlineCallbacks - def _get_already_ported_count(self, table): - rows = yield self.postgres_store.execute_sql( - "SELECT count(*) FROM %s" % (table,), - ) - - defer.returnValue(rows[0][0]) - - @defer.inlineCallbacks - def _get_total_count_to_port(self, table, next_chunk): - remaining, done = yield defer.gatherResults( - [ - self._get_remaining_count_to_port(table, next_chunk), - self._get_already_ported_count(table), - ], - consumeErrors=True, - ) - - remaining = int(remaining) if remaining else 0 - done = int(done) if done else 0 - - defer.returnValue((done, remaining + done)) - - -############################################## -###### The following is simply UI stuff ###### -############################################## - - -class Progress(object): - """Used to report progress of the port - """ - def __init__(self): - self.tables = {} - - self.start_time = int(time.time()) - - def add_table(self, table, cur, size): - self.tables[table] = { - "start": cur, - "num_done": cur, - "total": size, - "perc": int(cur * 100 / size), - } - - def update(self, table, num_done): - data = self.tables[table] - data["num_done"] = num_done - data["perc"] = int(num_done * 100 / data["total"]) - - def done(self): - pass - - -class CursesProgress(Progress): - """Reports progress to a curses window - """ - def __init__(self, stdscr): - self.stdscr = stdscr - - curses.use_default_colors() - curses.curs_set(0) - - curses.init_pair(1, curses.COLOR_RED, -1) - curses.init_pair(2, curses.COLOR_GREEN, -1) - - self.last_update = 0 - - self.finished = False - - self.total_processed = 0 - self.total_remaining = 0 - - super(CursesProgress, self).__init__() - - def update(self, table, num_done): - super(CursesProgress, self).update(table, num_done) - - self.total_processed = 0 - self.total_remaining = 0 - for table, data in self.tables.items(): - self.total_processed += data["num_done"] - data["start"] - self.total_remaining += data["total"] - data["num_done"] - - self.render() - - def render(self, force=False): - now = time.time() - - if not force and now - self.last_update < 0.2: - # reactor.callLater(1, self.render) - return - - self.stdscr.clear() - - rows, cols = self.stdscr.getmaxyx() - - duration = int(now) - int(self.start_time) - - minutes, seconds = divmod(duration, 60) - duration_str = '%02dm %02ds' % (minutes, seconds,) - - if self.finished: - status = "Time spent: %s (Done!)" % (duration_str,) - else: - - if self.total_processed > 0: - left = float(self.total_remaining) / self.total_processed - - est_remaining = (int(now) - self.start_time) * left - est_remaining_str = '%02dm %02ds remaining' % divmod(est_remaining, 60) - else: - est_remaining_str = "Unknown" - status = ( - "Time spent: %s (est. remaining: %s)" - % (duration_str, est_remaining_str,) - ) - - self.stdscr.addstr( - 0, 0, - status, - curses.A_BOLD, - ) - - max_len = max([len(t) for t in self.tables.keys()]) - - left_margin = 5 - middle_space = 1 - - items = self.tables.items() - items.sort( - key=lambda i: (i[1]["perc"], i[0]), - ) - - for i, (table, data) in enumerate(items): - if i + 2 >= rows: - break - - perc = data["perc"] - - color = curses.color_pair(2) if perc == 100 else curses.color_pair(1) - - self.stdscr.addstr( - i+2, left_margin + max_len - len(table), - table, - curses.A_BOLD | color, - ) - - size = 20 - - progress = "[%s%s]" % ( - "#" * int(perc*size/100), - " " * (size - int(perc*size/100)), - ) - - self.stdscr.addstr( - i+2, left_margin + max_len + middle_space, - "%s %3d%% (%d/%d)" % (progress, perc, data["num_done"], data["total"]), - ) - - if self.finished: - self.stdscr.addstr( - rows-1, 0, - "Press any key to exit...", - ) - - self.stdscr.refresh() - self.last_update = time.time() - - def done(self): - self.finished = True - self.render(True) - self.stdscr.getch() - - def set_state(self, state): - self.stdscr.clear() - self.stdscr.addstr( - 0, 0, - state + "...", - curses.A_BOLD, - ) - self.stdscr.refresh() - - -class TerminalProgress(Progress): - """Just prints progress to the terminal - """ - def update(self, table, num_done): - super(TerminalProgress, self).update(table, num_done) - - data = self.tables[table] - - print "%s: %d%% (%d/%d)" % ( - table, data["perc"], - data["num_done"], data["total"], - ) - - def set_state(self, state): - print state + "..." - - -############################################## -############################################## - - -if __name__ == "__main__": - parser = argparse.ArgumentParser( - description="A script to port an existing synapse SQLite database to" - " a new PostgreSQL database." - ) - parser.add_argument("-v", action='store_true') - parser.add_argument( - "--sqlite-database", required=True, - help="The snapshot of the SQLite database file. This must not be" - " currently used by a running synapse server" - ) - parser.add_argument( - "--postgres-config", type=argparse.FileType('r'), required=True, - help="The database config file for the PostgreSQL database" - ) - parser.add_argument( - "--curses", action='store_true', - help="display a curses based progress UI" - ) - - parser.add_argument( - "--batch-size", type=int, default=1000, - help="The number of rows to select from the SQLite table each" - " iteration [default=1000]", - ) - - args = parser.parse_args() - - logging_config = { - "level": logging.DEBUG if args.v else logging.INFO, - "format": "%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s" - } - - if args.curses: - logging_config["filename"] = "port-synapse.log" - - logging.basicConfig(**logging_config) - - sqlite_config = { - "name": "sqlite3", - "args": { - "database": args.sqlite_database, - "cp_min": 1, - "cp_max": 1, - "check_same_thread": False, - }, - } - - postgres_config = yaml.safe_load(args.postgres_config) - - if "database" in postgres_config: - postgres_config = postgres_config["database"] - - if "name" not in postgres_config: - sys.stderr.write("Malformed database config: no 'name'") - sys.exit(2) - if postgres_config["name"] != "psycopg2": - sys.stderr.write("Database must use 'psycopg2' connector.") - sys.exit(3) - - def start(stdscr=None): - if stdscr: - progress = CursesProgress(stdscr) - else: - progress = TerminalProgress() - - porter = Porter( - sqlite_config=sqlite_config, - postgres_config=postgres_config, - progress=progress, - batch_size=args.batch_size, - ) - - reactor.callWhenRunning(porter.run) - - reactor.run() - - if args.curses: - curses.wrapper(start) - else: - start() - - if end_error_exec_info: - exc_type, exc_value, exc_traceback = end_error_exec_info - traceback.print_exception(exc_type, exc_value, exc_traceback) diff --git a/scripts/synapse_port_db b/scripts/synapse_port_db new file mode 100755 index 0000000000..e7ed4c309b --- /dev/null +++ b/scripts/synapse_port_db @@ -0,0 +1,760 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# 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. + +from twisted.internet import defer, reactor +from twisted.enterprise import adbapi + +from synapse.storage._base import LoggingTransaction, SQLBaseStore +from synapse.storage.engines import create_engine + +import argparse +import curses +import logging +import sys +import time +import traceback +import yaml + + +logger = logging.getLogger("port_from_sqlite_to_postgres") + + +BOOLEAN_COLUMNS = { + "events": ["processed", "outlier"], + "rooms": ["is_public"], + "event_edges": ["is_state"], + "presence_list": ["accepted"], +} + + +APPEND_ONLY_TABLES = [ + "event_content_hashes", + "event_reference_hashes", + "event_signatures", + "event_edge_hashes", + "events", + "event_json", + "state_events", + "room_memberships", + "feedback", + "topics", + "room_names", + "rooms", + "local_media_repository", + "local_media_repository_thumbnails", + "remote_media_cache", + "remote_media_cache_thumbnails", + "redactions", + "event_edges", + "event_auth", + "received_transactions", + "sent_transactions", + "transaction_id_to_pdu", + "users", + "state_groups", + "state_groups_state", + "event_to_state_groups", + "rejections", +] + + +end_error_exec_info = None + + +class Store(object): + """This object is used to pull out some of the convenience API from the + Storage layer. + + *All* database interactions should go through this object. + """ + def __init__(self, db_pool, engine): + self.db_pool = db_pool + self.database_engine = engine + + _simple_insert_txn = SQLBaseStore.__dict__["_simple_insert_txn"] + _simple_insert = SQLBaseStore.__dict__["_simple_insert"] + + _simple_select_onecol_txn = SQLBaseStore.__dict__["_simple_select_onecol_txn"] + _simple_select_onecol = SQLBaseStore.__dict__["_simple_select_onecol"] + _simple_select_one_onecol = SQLBaseStore.__dict__["_simple_select_one_onecol"] + _simple_select_one_onecol_txn = SQLBaseStore.__dict__["_simple_select_one_onecol_txn"] + + _simple_update_one = SQLBaseStore.__dict__["_simple_update_one"] + _simple_update_one_txn = SQLBaseStore.__dict__["_simple_update_one_txn"] + + _execute_and_decode = SQLBaseStore.__dict__["_execute_and_decode"] + + def runInteraction(self, desc, func, *args, **kwargs): + def r(conn): + try: + i = 0 + N = 5 + while True: + try: + txn = conn.cursor() + return func( + LoggingTransaction(txn, desc, self.database_engine, []), + *args, **kwargs + ) + except self.database_engine.module.DatabaseError as e: + if self.database_engine.is_deadlock(e): + logger.warn("[TXN DEADLOCK] {%s} %d/%d", desc, i, N) + if i < N: + i += 1 + conn.rollback() + continue + raise + except Exception as e: + logger.debug("[TXN FAIL] {%s} %s", desc, e) + raise + + return self.db_pool.runWithConnection(r) + + def execute(self, f, *args, **kwargs): + return self.runInteraction(f.__name__, f, *args, **kwargs) + + def execute_sql(self, sql, *args): + def r(txn): + txn.execute(sql, args) + return txn.fetchall() + return self.runInteraction("execute_sql", r) + + def insert_many_txn(self, txn, table, headers, rows): + sql = "INSERT INTO %s (%s) VALUES (%s)" % ( + table, + ", ".join(k for k in headers), + ", ".join("%s" for _ in headers) + ) + + try: + txn.executemany(sql, rows) + except: + logger.exception( + "Failed to insert: %s", + table, + ) + raise + + +class Porter(object): + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + @defer.inlineCallbacks + def setup_table(self, table): + if table in APPEND_ONLY_TABLES: + # It's safe to just carry on inserting. + next_chunk = yield self.postgres_store._simple_select_one_onecol( + table="port_from_sqlite3", + keyvalues={"table_name": table}, + retcol="rowid", + allow_none=True, + ) + + total_to_port = None + if next_chunk is None: + if table == "sent_transactions": + next_chunk, already_ported, total_to_port = ( + yield self._setup_sent_transactions() + ) + else: + yield self.postgres_store._simple_insert( + table="port_from_sqlite3", + values={"table_name": table, "rowid": 1} + ) + + next_chunk = 1 + already_ported = 0 + + if total_to_port is None: + already_ported, total_to_port = yield self._get_total_count_to_port( + table, next_chunk + ) + else: + def delete_all(txn): + txn.execute( + "DELETE FROM port_from_sqlite3 WHERE table_name = %s", + (table,) + ) + txn.execute("TRUNCATE %s CASCADE" % (table,)) + + yield self.postgres_store.execute(delete_all) + + yield self.postgres_store._simple_insert( + table="port_from_sqlite3", + values={"table_name": table, "rowid": 0} + ) + + next_chunk = 1 + + already_ported, total_to_port = yield self._get_total_count_to_port( + table, next_chunk + ) + + defer.returnValue((table, already_ported, total_to_port, next_chunk)) + + @defer.inlineCallbacks + def handle_table(self, table, postgres_size, table_size, next_chunk): + if not table_size: + return + + self.progress.add_table(table, postgres_size, table_size) + + select = ( + "SELECT rowid, * FROM %s WHERE rowid >= ? ORDER BY rowid LIMIT ?" + % (table,) + ) + + while True: + def r(txn): + txn.execute(select, (next_chunk, self.batch_size,)) + rows = txn.fetchall() + headers = [column[0] for column in txn.description] + + return headers, rows + + headers, rows = yield self.sqlite_store.runInteraction("select", r) + + if rows: + next_chunk = rows[-1][0] + 1 + + self._convert_rows(table, headers, rows) + + def insert(txn): + self.postgres_store.insert_many_txn( + txn, table, headers[1:], rows + ) + + self.postgres_store._simple_update_one_txn( + txn, + table="port_from_sqlite3", + keyvalues={"table_name": table}, + updatevalues={"rowid": next_chunk}, + ) + + yield self.postgres_store.execute(insert) + + postgres_size += len(rows) + + self.progress.update(table, postgres_size) + else: + return + + def setup_db(self, db_config, database_engine): + db_conn = database_engine.module.connect( + **{ + k: v for k, v in db_config.get("args", {}).items() + if not k.startswith("cp_") + } + ) + + database_engine.prepare_database(db_conn) + + db_conn.commit() + + @defer.inlineCallbacks + def run(self): + try: + sqlite_db_pool = adbapi.ConnectionPool( + self.sqlite_config["name"], + **self.sqlite_config["args"] + ) + + postgres_db_pool = adbapi.ConnectionPool( + self.postgres_config["name"], + **self.postgres_config["args"] + ) + + sqlite_engine = create_engine("sqlite3") + postgres_engine = create_engine("psycopg2") + + self.sqlite_store = Store(sqlite_db_pool, sqlite_engine) + self.postgres_store = Store(postgres_db_pool, postgres_engine) + + yield self.postgres_store.execute( + postgres_engine.check_database + ) + + # Step 1. Set up databases. + self.progress.set_state("Preparing SQLite3") + self.setup_db(sqlite_config, sqlite_engine) + + self.progress.set_state("Preparing PostgreSQL") + self.setup_db(postgres_config, postgres_engine) + + # Step 2. Get tables. + self.progress.set_state("Fetching tables") + sqlite_tables = yield self.sqlite_store._simple_select_onecol( + table="sqlite_master", + keyvalues={ + "type": "table", + }, + retcol="name", + ) + + postgres_tables = yield self.postgres_store._simple_select_onecol( + table="information_schema.tables", + keyvalues={ + "table_schema": "public", + }, + retcol="distinct table_name", + ) + + tables = set(sqlite_tables) & set(postgres_tables) + + self.progress.set_state("Creating tables") + + logger.info("Found %d tables", len(tables)) + + def create_port_table(txn): + txn.execute( + "CREATE TABLE port_from_sqlite3 (" + " table_name varchar(100) NOT NULL UNIQUE," + " rowid bigint NOT NULL" + ")" + ) + + try: + yield self.postgres_store.runInteraction( + "create_port_table", create_port_table + ) + except Exception as e: + logger.info("Failed to create port table: %s", e) + + self.progress.set_state("Setting up") + + # Set up tables. + setup_res = yield defer.gatherResults( + [ + self.setup_table(table) + for table in tables + if table not in ["schema_version", "applied_schema_deltas"] + and not table.startswith("sqlite_") + ], + consumeErrors=True, + ) + + # Process tables. + yield defer.gatherResults( + [ + self.handle_table(*res) + for res in setup_res + ], + consumeErrors=True, + ) + + self.progress.done() + except: + global end_error_exec_info + end_error_exec_info = sys.exc_info() + logger.exception("") + finally: + reactor.stop() + + def _convert_rows(self, table, headers, rows): + bool_col_names = BOOLEAN_COLUMNS.get(table, []) + + bool_cols = [ + i for i, h in enumerate(headers) if h in bool_col_names + ] + + def conv(j, col): + if j in bool_cols: + return bool(col) + return col + + for i, row in enumerate(rows): + rows[i] = tuple( + conv(j, col) + for j, col in enumerate(row) + if j > 0 + ) + + @defer.inlineCallbacks + def _setup_sent_transactions(self): + # Only save things from the last day + yesterday = int(time.time()*1000) - 86400000 + + # And save the max transaction id from each destination + select = ( + "SELECT rowid, * FROM sent_transactions WHERE rowid IN (" + "SELECT max(rowid) FROM sent_transactions" + " GROUP BY destination" + ")" + ) + + def r(txn): + txn.execute(select) + rows = txn.fetchall() + headers = [column[0] for column in txn.description] + + ts_ind = headers.index('ts') + + return headers, [r for r in rows if r[ts_ind] < yesterday] + + headers, rows = yield self.sqlite_store.runInteraction( + "select", r, + ) + + self._convert_rows("sent_transactions", headers, rows) + + inserted_rows = len(rows) + max_inserted_rowid = max(r[0] for r in rows) + + def insert(txn): + self.postgres_store.insert_many_txn( + txn, "sent_transactions", headers[1:], rows + ) + + yield self.postgres_store.execute(insert) + + def get_start_id(txn): + txn.execute( + "SELECT rowid FROM sent_transactions WHERE ts >= ?" + " ORDER BY rowid ASC LIMIT 1", + (yesterday,) + ) + + rows = txn.fetchall() + if rows: + return rows[0][0] + else: + return 1 + + next_chunk = yield self.sqlite_store.execute(get_start_id) + next_chunk = max(max_inserted_rowid + 1, next_chunk) + + yield self.postgres_store._simple_insert( + table="port_from_sqlite3", + values={"table_name": "sent_transactions", "rowid": next_chunk} + ) + + def get_sent_table_size(txn): + txn.execute( + "SELECT count(*) FROM sent_transactions" + " WHERE ts >= ?", + (yesterday,) + ) + size, = txn.fetchone() + return int(size) + + remaining_count = yield self.sqlite_store.execute( + get_sent_table_size + ) + + total_count = remaining_count + inserted_rows + + defer.returnValue((next_chunk, inserted_rows, total_count)) + + @defer.inlineCallbacks + def _get_remaining_count_to_port(self, table, next_chunk): + rows = yield self.sqlite_store.execute_sql( + "SELECT count(*) FROM %s WHERE rowid >= ?" % (table,), + next_chunk, + ) + + defer.returnValue(rows[0][0]) + + @defer.inlineCallbacks + def _get_already_ported_count(self, table): + rows = yield self.postgres_store.execute_sql( + "SELECT count(*) FROM %s" % (table,), + ) + + defer.returnValue(rows[0][0]) + + @defer.inlineCallbacks + def _get_total_count_to_port(self, table, next_chunk): + remaining, done = yield defer.gatherResults( + [ + self._get_remaining_count_to_port(table, next_chunk), + self._get_already_ported_count(table), + ], + consumeErrors=True, + ) + + remaining = int(remaining) if remaining else 0 + done = int(done) if done else 0 + + defer.returnValue((done, remaining + done)) + + +############################################## +###### The following is simply UI stuff ###### +############################################## + + +class Progress(object): + """Used to report progress of the port + """ + def __init__(self): + self.tables = {} + + self.start_time = int(time.time()) + + def add_table(self, table, cur, size): + self.tables[table] = { + "start": cur, + "num_done": cur, + "total": size, + "perc": int(cur * 100 / size), + } + + def update(self, table, num_done): + data = self.tables[table] + data["num_done"] = num_done + data["perc"] = int(num_done * 100 / data["total"]) + + def done(self): + pass + + +class CursesProgress(Progress): + """Reports progress to a curses window + """ + def __init__(self, stdscr): + self.stdscr = stdscr + + curses.use_default_colors() + curses.curs_set(0) + + curses.init_pair(1, curses.COLOR_RED, -1) + curses.init_pair(2, curses.COLOR_GREEN, -1) + + self.last_update = 0 + + self.finished = False + + self.total_processed = 0 + self.total_remaining = 0 + + super(CursesProgress, self).__init__() + + def update(self, table, num_done): + super(CursesProgress, self).update(table, num_done) + + self.total_processed = 0 + self.total_remaining = 0 + for table, data in self.tables.items(): + self.total_processed += data["num_done"] - data["start"] + self.total_remaining += data["total"] - data["num_done"] + + self.render() + + def render(self, force=False): + now = time.time() + + if not force and now - self.last_update < 0.2: + # reactor.callLater(1, self.render) + return + + self.stdscr.clear() + + rows, cols = self.stdscr.getmaxyx() + + duration = int(now) - int(self.start_time) + + minutes, seconds = divmod(duration, 60) + duration_str = '%02dm %02ds' % (minutes, seconds,) + + if self.finished: + status = "Time spent: %s (Done!)" % (duration_str,) + else: + + if self.total_processed > 0: + left = float(self.total_remaining) / self.total_processed + + est_remaining = (int(now) - self.start_time) * left + est_remaining_str = '%02dm %02ds remaining' % divmod(est_remaining, 60) + else: + est_remaining_str = "Unknown" + status = ( + "Time spent: %s (est. remaining: %s)" + % (duration_str, est_remaining_str,) + ) + + self.stdscr.addstr( + 0, 0, + status, + curses.A_BOLD, + ) + + max_len = max([len(t) for t in self.tables.keys()]) + + left_margin = 5 + middle_space = 1 + + items = self.tables.items() + items.sort( + key=lambda i: (i[1]["perc"], i[0]), + ) + + for i, (table, data) in enumerate(items): + if i + 2 >= rows: + break + + perc = data["perc"] + + color = curses.color_pair(2) if perc == 100 else curses.color_pair(1) + + self.stdscr.addstr( + i+2, left_margin + max_len - len(table), + table, + curses.A_BOLD | color, + ) + + size = 20 + + progress = "[%s%s]" % ( + "#" * int(perc*size/100), + " " * (size - int(perc*size/100)), + ) + + self.stdscr.addstr( + i+2, left_margin + max_len + middle_space, + "%s %3d%% (%d/%d)" % (progress, perc, data["num_done"], data["total"]), + ) + + if self.finished: + self.stdscr.addstr( + rows-1, 0, + "Press any key to exit...", + ) + + self.stdscr.refresh() + self.last_update = time.time() + + def done(self): + self.finished = True + self.render(True) + self.stdscr.getch() + + def set_state(self, state): + self.stdscr.clear() + self.stdscr.addstr( + 0, 0, + state + "...", + curses.A_BOLD, + ) + self.stdscr.refresh() + + +class TerminalProgress(Progress): + """Just prints progress to the terminal + """ + def update(self, table, num_done): + super(TerminalProgress, self).update(table, num_done) + + data = self.tables[table] + + print "%s: %d%% (%d/%d)" % ( + table, data["perc"], + data["num_done"], data["total"], + ) + + def set_state(self, state): + print state + "..." + + +############################################## +############################################## + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="A script to port an existing synapse SQLite database to" + " a new PostgreSQL database." + ) + parser.add_argument("-v", action='store_true') + parser.add_argument( + "--sqlite-database", required=True, + help="The snapshot of the SQLite database file. This must not be" + " currently used by a running synapse server" + ) + parser.add_argument( + "--postgres-config", type=argparse.FileType('r'), required=True, + help="The database config file for the PostgreSQL database" + ) + parser.add_argument( + "--curses", action='store_true', + help="display a curses based progress UI" + ) + + parser.add_argument( + "--batch-size", type=int, default=1000, + help="The number of rows to select from the SQLite table each" + " iteration [default=1000]", + ) + + args = parser.parse_args() + + logging_config = { + "level": logging.DEBUG if args.v else logging.INFO, + "format": "%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s" + } + + if args.curses: + logging_config["filename"] = "port-synapse.log" + + logging.basicConfig(**logging_config) + + sqlite_config = { + "name": "sqlite3", + "args": { + "database": args.sqlite_database, + "cp_min": 1, + "cp_max": 1, + "check_same_thread": False, + }, + } + + postgres_config = yaml.safe_load(args.postgres_config) + + if "database" in postgres_config: + postgres_config = postgres_config["database"] + + if "name" not in postgres_config: + sys.stderr.write("Malformed database config: no 'name'") + sys.exit(2) + if postgres_config["name"] != "psycopg2": + sys.stderr.write("Database must use 'psycopg2' connector.") + sys.exit(3) + + def start(stdscr=None): + if stdscr: + progress = CursesProgress(stdscr) + else: + progress = TerminalProgress() + + porter = Porter( + sqlite_config=sqlite_config, + postgres_config=postgres_config, + progress=progress, + batch_size=args.batch_size, + ) + + reactor.callWhenRunning(porter.run) + + reactor.run() + + if args.curses: + curses.wrapper(start) + else: + start() + + if end_error_exec_info: + exc_type, exc_value, exc_traceback = end_error_exec_info + traceback.print_exception(exc_type, exc_value, exc_traceback) diff --git a/scripts/upgrade_db_to_v0.6.0.py b/scripts/upgrade_db_to_v0.6.0.py deleted file mode 100755 index f466ca510e..0000000000 --- a/scripts/upgrade_db_to_v0.6.0.py +++ /dev/null @@ -1,331 +0,0 @@ -#!/usr/bin/env python -from synapse.storage import SCHEMA_VERSION, read_schema -from synapse.storage._base import SQLBaseStore -from synapse.storage.signatures import SignatureStore -from synapse.storage.event_federation import EventFederationStore - -from syutil.base64util import encode_base64, decode_base64 - -from synapse.crypto.event_signing import compute_event_signature - -from synapse.events.builder import EventBuilder -from synapse.events.utils import prune_event - -from synapse.crypto.event_signing import check_event_content_hash - -from syutil.crypto.jsonsign import ( - verify_signed_json, SignatureVerifyException, -) -from syutil.crypto.signing_key import decode_verify_key_bytes - -from syutil.jsonutil import encode_canonical_json - -import argparse -# import dns.resolver -import hashlib -import httplib -import json -import sqlite3 -import syutil -import urllib2 - - -delta_sql = """ -CREATE TABLE IF NOT EXISTS event_json( - event_id TEXT NOT NULL, - room_id TEXT NOT NULL, - internal_metadata NOT NULL, - json BLOB NOT NULL, - CONSTRAINT ev_j_uniq UNIQUE (event_id) -); - -CREATE INDEX IF NOT EXISTS event_json_id ON event_json(event_id); -CREATE INDEX IF NOT EXISTS event_json_room_id ON event_json(room_id); - -PRAGMA user_version = 10; -""" - - -class Store(object): - _get_event_signatures_txn = SignatureStore.__dict__["_get_event_signatures_txn"] - _get_event_content_hashes_txn = SignatureStore.__dict__["_get_event_content_hashes_txn"] - _get_event_reference_hashes_txn = SignatureStore.__dict__["_get_event_reference_hashes_txn"] - _get_prev_event_hashes_txn = SignatureStore.__dict__["_get_prev_event_hashes_txn"] - _get_prev_events_and_state = EventFederationStore.__dict__["_get_prev_events_and_state"] - _get_auth_events = EventFederationStore.__dict__["_get_auth_events"] - cursor_to_dict = SQLBaseStore.__dict__["cursor_to_dict"] - _simple_select_onecol_txn = SQLBaseStore.__dict__["_simple_select_onecol_txn"] - _simple_select_list_txn = SQLBaseStore.__dict__["_simple_select_list_txn"] - _simple_insert_txn = SQLBaseStore.__dict__["_simple_insert_txn"] - - def _generate_event_json(self, txn, rows): - events = [] - for row in rows: - d = dict(row) - - d.pop("stream_ordering", None) - d.pop("topological_ordering", None) - d.pop("processed", None) - - if "origin_server_ts" not in d: - d["origin_server_ts"] = d.pop("ts", 0) - else: - d.pop("ts", 0) - - d.pop("prev_state", None) - d.update(json.loads(d.pop("unrecognized_keys"))) - - d["sender"] = d.pop("user_id") - - d["content"] = json.loads(d["content"]) - - if "age_ts" not in d: - # For compatibility - d["age_ts"] = d.get("origin_server_ts", 0) - - d.setdefault("unsigned", {})["age_ts"] = d.pop("age_ts") - - outlier = d.pop("outlier", False) - - # d.pop("membership", None) - - d.pop("state_hash", None) - - d.pop("replaces_state", None) - - b = EventBuilder(d) - b.internal_metadata.outlier = outlier - - events.append(b) - - for i, ev in enumerate(events): - signatures = self._get_event_signatures_txn( - txn, ev.event_id, - ) - - ev.signatures = { - n: { - k: encode_base64(v) for k, v in s.items() - } - for n, s in signatures.items() - } - - hashes = self._get_event_content_hashes_txn( - txn, ev.event_id, - ) - - ev.hashes = { - k: encode_base64(v) for k, v in hashes.items() - } - - prevs = self._get_prev_events_and_state(txn, ev.event_id) - - ev.prev_events = [ - (e_id, h) - for e_id, h, is_state in prevs - if is_state == 0 - ] - - # ev.auth_events = self._get_auth_events(txn, ev.event_id) - - hashes = dict(ev.auth_events) - - for e_id, hash in ev.prev_events: - if e_id in hashes and not hash: - hash.update(hashes[e_id]) - # - # if hasattr(ev, "state_key"): - # ev.prev_state = [ - # (e_id, h) - # for e_id, h, is_state in prevs - # if is_state == 1 - # ] - - return [e.build() for e in events] - - -store = Store() - - -# def get_key(server_name): -# print "Getting keys for: %s" % (server_name,) -# targets = [] -# if ":" in server_name: -# target, port = server_name.split(":") -# targets.append((target, int(port))) -# try: -# answers = dns.resolver.query("_matrix._tcp." + server_name, "SRV") -# for srv in answers: -# targets.append((srv.target, srv.port)) -# except dns.resolver.NXDOMAIN: -# targets.append((server_name, 8448)) -# except: -# print "Failed to lookup keys for %s" % (server_name,) -# return {} -# -# for target, port in targets: -# url = "https://%s:%i/_matrix/key/v1" % (target, port) -# try: -# keys = json.load(urllib2.urlopen(url, timeout=2)) -# verify_keys = {} -# for key_id, key_base64 in keys["verify_keys"].items(): -# verify_key = decode_verify_key_bytes( -# key_id, decode_base64(key_base64) -# ) -# verify_signed_json(keys, server_name, verify_key) -# verify_keys[key_id] = verify_key -# print "Got keys for: %s" % (server_name,) -# return verify_keys -# except urllib2.URLError: -# pass -# except urllib2.HTTPError: -# pass -# except httplib.HTTPException: -# pass -# -# print "Failed to get keys for %s" % (server_name,) -# return {} - - -def reinsert_events(cursor, server_name, signing_key): - print "Running delta: v10" - - cursor.executescript(delta_sql) - - cursor.execute( - "SELECT * FROM events ORDER BY rowid ASC" - ) - - print "Getting events..." - - rows = store.cursor_to_dict(cursor) - - events = store._generate_event_json(cursor, rows) - - print "Got events from DB." - - algorithms = { - "sha256": hashlib.sha256, - } - - key_id = "%s:%s" % (signing_key.alg, signing_key.version) - verify_key = signing_key.verify_key - verify_key.alg = signing_key.alg - verify_key.version = signing_key.version - - server_keys = { - server_name: { - key_id: verify_key - } - } - - i = 0 - N = len(events) - - for event in events: - if i % 100 == 0: - print "Processed: %d/%d events" % (i,N,) - i += 1 - - # for alg_name in event.hashes: - # if check_event_content_hash(event, algorithms[alg_name]): - # pass - # else: - # pass - # print "FAIL content hash %s %s" % (alg_name, event.event_id, ) - - have_own_correctly_signed = False - for host, sigs in event.signatures.items(): - pruned = prune_event(event) - - for key_id in sigs: - if host not in server_keys: - server_keys[host] = {} # get_key(host) - if key_id in server_keys[host]: - try: - verify_signed_json( - pruned.get_pdu_json(), - host, - server_keys[host][key_id] - ) - - if host == server_name: - have_own_correctly_signed = True - except SignatureVerifyException: - print "FAIL signature check %s %s" % ( - key_id, event.event_id - ) - - # TODO: Re sign with our own server key - if not have_own_correctly_signed: - sigs = compute_event_signature(event, server_name, signing_key) - event.signatures.update(sigs) - - pruned = prune_event(event) - - for key_id in event.signatures[server_name]: - verify_signed_json( - pruned.get_pdu_json(), - server_name, - server_keys[server_name][key_id] - ) - - event_json = encode_canonical_json( - event.get_dict() - ).decode("UTF-8") - - metadata_json = encode_canonical_json( - event.internal_metadata.get_dict() - ).decode("UTF-8") - - store._simple_insert_txn( - cursor, - table="event_json", - values={ - "event_id": event.event_id, - "room_id": event.room_id, - "internal_metadata": metadata_json, - "json": event_json, - }, - or_replace=True, - ) - - -def main(database, server_name, signing_key): - conn = sqlite3.connect(database) - cursor = conn.cursor() - - # Do other deltas: - cursor.execute("PRAGMA user_version") - row = cursor.fetchone() - - if row and row[0]: - user_version = row[0] - # Run every version since after the current version. - for v in range(user_version + 1, 10): - print "Running delta: %d" % (v,) - sql_script = read_schema("delta/v%d" % (v,)) - cursor.executescript(sql_script) - - reinsert_events(cursor, server_name, signing_key) - - conn.commit() - - print "Success!" - - -if __name__ == "__main__": - parser = argparse.ArgumentParser() - - parser.add_argument("database") - parser.add_argument("server_name") - parser.add_argument( - "signing_key", type=argparse.FileType('r'), - ) - args = parser.parse_args() - - signing_key = syutil.crypto.signing_key.read_signing_keys( - args.signing_key - ) - - main(args.database, args.server_name, signing_key[0]) -- cgit 1.4.1 From b442217d918a50ae57a932a955fd577240607e92 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 28 Aug 2015 10:37:17 +0100 Subject: Actually add config path --- synapse/config/_base.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/synapse/config/_base.py b/synapse/config/_base.py index d01235d31f..1a6784a714 100644 --- a/synapse/config/_base.py +++ b/synapse/config/_base.py @@ -182,6 +182,8 @@ class Config(object): ) % (entry_path, ) continue + files.add(config_path) + config_files.extend(sorted(files)) else: config_files.append(config_path) -- cgit 1.4.1 From 3e4de64bc9bc5ff1645f917e7c6deacf29fb868b Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 1 Sep 2015 09:46:42 +0100 Subject: Remove spurious .py from docs --- docs/postgres.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/postgres.rst b/docs/postgres.rst index 017b3c47e7..b5027fefb0 100644 --- a/docs/postgres.rst +++ b/docs/postgres.rst @@ -98,7 +98,7 @@ To complete the conversion shut down the synapse server and run the port script one last time, e.g. if the SQLite database is at ``homeserver.db`` run:: - synapse_port_db.py --sqlite-database homeserver.db \ + synapse_port_db --sqlite-database homeserver.db \ --postgres-config database_config.yaml Once that has completed, change the synapse config to point at the PostgreSQL -- cgit 1.4.1