1
0

synapse_port_db 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # Copyright 2015, 2016 OpenMarket Ltd
  4. # Copyright 2018 New Vector Ltd
  5. #
  6. # Licensed under the Apache License, Version 2.0 (the "License");
  7. # you may not use this file except in compliance with the License.
  8. # You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import argparse
  18. import curses
  19. import logging
  20. import sys
  21. import time
  22. import traceback
  23. from six import string_types
  24. import yaml
  25. from twisted.enterprise import adbapi
  26. from twisted.internet import defer, reactor
  27. from synapse.storage._base import LoggingTransaction, SQLBaseStore
  28. from synapse.storage.engines import create_engine
  29. from synapse.storage.prepare_database import prepare_database
  30. logger = logging.getLogger("synapse_port_db")
  31. BOOLEAN_COLUMNS = {
  32. "events": ["processed", "outlier", "contains_url"],
  33. "rooms": ["is_public"],
  34. "event_edges": ["is_state"],
  35. "presence_list": ["accepted"],
  36. "presence_stream": ["currently_active"],
  37. "public_room_list_stream": ["visibility"],
  38. "device_lists_outbound_pokes": ["sent"],
  39. "users_who_share_rooms": ["share_private"],
  40. "groups": ["is_public"],
  41. "group_rooms": ["is_public"],
  42. "group_users": ["is_public", "is_admin"],
  43. "group_summary_rooms": ["is_public"],
  44. "group_room_categories": ["is_public"],
  45. "group_summary_users": ["is_public"],
  46. "group_roles": ["is_public"],
  47. "local_group_membership": ["is_publicised", "is_admin"],
  48. "e2e_room_keys": ["is_verified"],
  49. "account_validity": ["email_sent"],
  50. }
  51. APPEND_ONLY_TABLES = [
  52. "event_reference_hashes",
  53. "events",
  54. "event_json",
  55. "state_events",
  56. "room_memberships",
  57. "topics",
  58. "room_names",
  59. "rooms",
  60. "local_media_repository",
  61. "local_media_repository_thumbnails",
  62. "remote_media_cache",
  63. "remote_media_cache_thumbnails",
  64. "redactions",
  65. "event_edges",
  66. "event_auth",
  67. "received_transactions",
  68. "sent_transactions",
  69. "transaction_id_to_pdu",
  70. "users",
  71. "state_groups",
  72. "state_groups_state",
  73. "event_to_state_groups",
  74. "rejections",
  75. "event_search",
  76. "presence_stream",
  77. "push_rules_stream",
  78. "ex_outlier_stream",
  79. "cache_invalidation_stream",
  80. "public_room_list_stream",
  81. "state_group_edges",
  82. "stream_ordering_to_exterm",
  83. ]
  84. end_error_exec_info = None
  85. class Store(object):
  86. """This object is used to pull out some of the convenience API from the
  87. Storage layer.
  88. *All* database interactions should go through this object.
  89. """
  90. def __init__(self, db_pool, engine):
  91. self.db_pool = db_pool
  92. self.database_engine = engine
  93. _simple_insert_txn = SQLBaseStore.__dict__["_simple_insert_txn"]
  94. _simple_insert = SQLBaseStore.__dict__["_simple_insert"]
  95. _simple_select_onecol_txn = SQLBaseStore.__dict__["_simple_select_onecol_txn"]
  96. _simple_select_onecol = SQLBaseStore.__dict__["_simple_select_onecol"]
  97. _simple_select_one = SQLBaseStore.__dict__["_simple_select_one"]
  98. _simple_select_one_txn = SQLBaseStore.__dict__["_simple_select_one_txn"]
  99. _simple_select_one_onecol = SQLBaseStore.__dict__["_simple_select_one_onecol"]
  100. _simple_select_one_onecol_txn = SQLBaseStore.__dict__[
  101. "_simple_select_one_onecol_txn"
  102. ]
  103. _simple_update_one = SQLBaseStore.__dict__["_simple_update_one"]
  104. _simple_update_one_txn = SQLBaseStore.__dict__["_simple_update_one_txn"]
  105. _simple_update_txn = SQLBaseStore.__dict__["_simple_update_txn"]
  106. def runInteraction(self, desc, func, *args, **kwargs):
  107. def r(conn):
  108. try:
  109. i = 0
  110. N = 5
  111. while True:
  112. try:
  113. txn = conn.cursor()
  114. return func(
  115. LoggingTransaction(txn, desc, self.database_engine, [], []),
  116. *args,
  117. **kwargs
  118. )
  119. except self.database_engine.module.DatabaseError as e:
  120. if self.database_engine.is_deadlock(e):
  121. logger.warn("[TXN DEADLOCK] {%s} %d/%d", desc, i, N)
  122. if i < N:
  123. i += 1
  124. conn.rollback()
  125. continue
  126. raise
  127. except Exception as e:
  128. logger.debug("[TXN FAIL] {%s} %s", desc, e)
  129. raise
  130. return self.db_pool.runWithConnection(r)
  131. def execute(self, f, *args, **kwargs):
  132. return self.runInteraction(f.__name__, f, *args, **kwargs)
  133. def execute_sql(self, sql, *args):
  134. def r(txn):
  135. txn.execute(sql, args)
  136. return txn.fetchall()
  137. return self.runInteraction("execute_sql", r)
  138. def insert_many_txn(self, txn, table, headers, rows):
  139. sql = "INSERT INTO %s (%s) VALUES (%s)" % (
  140. table,
  141. ", ".join(k for k in headers),
  142. ", ".join("%s" for _ in headers),
  143. )
  144. try:
  145. txn.executemany(sql, rows)
  146. except Exception:
  147. logger.exception("Failed to insert: %s", table)
  148. raise
  149. class Porter(object):
  150. def __init__(self, **kwargs):
  151. self.__dict__.update(kwargs)
  152. @defer.inlineCallbacks
  153. def setup_table(self, table):
  154. if table in APPEND_ONLY_TABLES:
  155. # It's safe to just carry on inserting.
  156. row = yield self.postgres_store._simple_select_one(
  157. table="port_from_sqlite3",
  158. keyvalues={"table_name": table},
  159. retcols=("forward_rowid", "backward_rowid"),
  160. allow_none=True,
  161. )
  162. total_to_port = None
  163. if row is None:
  164. if table == "sent_transactions":
  165. forward_chunk, already_ported, total_to_port = (
  166. yield self._setup_sent_transactions()
  167. )
  168. backward_chunk = 0
  169. else:
  170. yield self.postgres_store._simple_insert(
  171. table="port_from_sqlite3",
  172. values={
  173. "table_name": table,
  174. "forward_rowid": 1,
  175. "backward_rowid": 0,
  176. },
  177. )
  178. forward_chunk = 1
  179. backward_chunk = 0
  180. already_ported = 0
  181. else:
  182. forward_chunk = row["forward_rowid"]
  183. backward_chunk = row["backward_rowid"]
  184. if total_to_port is None:
  185. already_ported, total_to_port = yield self._get_total_count_to_port(
  186. table, forward_chunk, backward_chunk
  187. )
  188. else:
  189. def delete_all(txn):
  190. txn.execute(
  191. "DELETE FROM port_from_sqlite3 WHERE table_name = %s", (table,)
  192. )
  193. txn.execute("TRUNCATE %s CASCADE" % (table,))
  194. yield self.postgres_store.execute(delete_all)
  195. yield self.postgres_store._simple_insert(
  196. table="port_from_sqlite3",
  197. values={"table_name": table, "forward_rowid": 1, "backward_rowid": 0},
  198. )
  199. forward_chunk = 1
  200. backward_chunk = 0
  201. already_ported, total_to_port = yield self._get_total_count_to_port(
  202. table, forward_chunk, backward_chunk
  203. )
  204. defer.returnValue(
  205. (table, already_ported, total_to_port, forward_chunk, backward_chunk)
  206. )
  207. @defer.inlineCallbacks
  208. def handle_table(
  209. self, table, postgres_size, table_size, forward_chunk, backward_chunk
  210. ):
  211. logger.info(
  212. "Table %s: %i/%i (rows %i-%i) already ported",
  213. table,
  214. postgres_size,
  215. table_size,
  216. backward_chunk + 1,
  217. forward_chunk - 1,
  218. )
  219. if not table_size:
  220. return
  221. self.progress.add_table(table, postgres_size, table_size)
  222. if table == "event_search":
  223. yield self.handle_search_table(
  224. postgres_size, table_size, forward_chunk, backward_chunk
  225. )
  226. return
  227. if table in (
  228. "user_directory",
  229. "user_directory_search",
  230. "users_who_share_rooms",
  231. "users_in_pubic_room",
  232. ):
  233. # We don't port these tables, as they're a faff and we can regenreate
  234. # them anyway.
  235. self.progress.update(table, table_size) # Mark table as done
  236. return
  237. if table == "user_directory_stream_pos":
  238. # We need to make sure there is a single row, `(X, null), as that is
  239. # what synapse expects to be there.
  240. yield self.postgres_store._simple_insert(
  241. table=table, values={"stream_id": None}
  242. )
  243. self.progress.update(table, table_size) # Mark table as done
  244. return
  245. forward_select = (
  246. "SELECT rowid, * FROM %s WHERE rowid >= ? ORDER BY rowid LIMIT ?" % (table,)
  247. )
  248. backward_select = (
  249. "SELECT rowid, * FROM %s WHERE rowid <= ? ORDER BY rowid LIMIT ?" % (table,)
  250. )
  251. do_forward = [True]
  252. do_backward = [True]
  253. while True:
  254. def r(txn):
  255. forward_rows = []
  256. backward_rows = []
  257. if do_forward[0]:
  258. txn.execute(forward_select, (forward_chunk, self.batch_size))
  259. forward_rows = txn.fetchall()
  260. if not forward_rows:
  261. do_forward[0] = False
  262. if do_backward[0]:
  263. txn.execute(backward_select, (backward_chunk, self.batch_size))
  264. backward_rows = txn.fetchall()
  265. if not backward_rows:
  266. do_backward[0] = False
  267. if forward_rows or backward_rows:
  268. headers = [column[0] for column in txn.description]
  269. else:
  270. headers = None
  271. return headers, forward_rows, backward_rows
  272. headers, frows, brows = yield self.sqlite_store.runInteraction("select", r)
  273. if frows or brows:
  274. if frows:
  275. forward_chunk = max(row[0] for row in frows) + 1
  276. if brows:
  277. backward_chunk = min(row[0] for row in brows) - 1
  278. rows = frows + brows
  279. rows = self._convert_rows(table, headers, rows)
  280. def insert(txn):
  281. self.postgres_store.insert_many_txn(txn, table, headers[1:], rows)
  282. self.postgres_store._simple_update_one_txn(
  283. txn,
  284. table="port_from_sqlite3",
  285. keyvalues={"table_name": table},
  286. updatevalues={
  287. "forward_rowid": forward_chunk,
  288. "backward_rowid": backward_chunk,
  289. },
  290. )
  291. yield self.postgres_store.execute(insert)
  292. postgres_size += len(rows)
  293. self.progress.update(table, postgres_size)
  294. else:
  295. return
  296. @defer.inlineCallbacks
  297. def handle_search_table(
  298. self, postgres_size, table_size, forward_chunk, backward_chunk
  299. ):
  300. select = (
  301. "SELECT es.rowid, es.*, e.origin_server_ts, e.stream_ordering"
  302. " FROM event_search as es"
  303. " INNER JOIN events AS e USING (event_id, room_id)"
  304. " WHERE es.rowid >= ?"
  305. " ORDER BY es.rowid LIMIT ?"
  306. )
  307. while True:
  308. def r(txn):
  309. txn.execute(select, (forward_chunk, self.batch_size))
  310. rows = txn.fetchall()
  311. headers = [column[0] for column in txn.description]
  312. return headers, rows
  313. headers, rows = yield self.sqlite_store.runInteraction("select", r)
  314. if rows:
  315. forward_chunk = rows[-1][0] + 1
  316. # We have to treat event_search differently since it has a
  317. # different structure in the two different databases.
  318. def insert(txn):
  319. sql = (
  320. "INSERT INTO event_search (event_id, room_id, key,"
  321. " sender, vector, origin_server_ts, stream_ordering)"
  322. " VALUES (?,?,?,?,to_tsvector('english', ?),?,?)"
  323. )
  324. rows_dict = []
  325. for row in rows:
  326. d = dict(zip(headers, row))
  327. if "\0" in d['value']:
  328. logger.warn('dropping search row %s', d)
  329. else:
  330. rows_dict.append(d)
  331. txn.executemany(
  332. sql,
  333. [
  334. (
  335. row["event_id"],
  336. row["room_id"],
  337. row["key"],
  338. row["sender"],
  339. row["value"],
  340. row["origin_server_ts"],
  341. row["stream_ordering"],
  342. )
  343. for row in rows_dict
  344. ],
  345. )
  346. self.postgres_store._simple_update_one_txn(
  347. txn,
  348. table="port_from_sqlite3",
  349. keyvalues={"table_name": "event_search"},
  350. updatevalues={
  351. "forward_rowid": forward_chunk,
  352. "backward_rowid": backward_chunk,
  353. },
  354. )
  355. yield self.postgres_store.execute(insert)
  356. postgres_size += len(rows)
  357. self.progress.update("event_search", postgres_size)
  358. else:
  359. return
  360. def setup_db(self, db_config, database_engine):
  361. db_conn = database_engine.module.connect(
  362. **{
  363. k: v
  364. for k, v in db_config.get("args", {}).items()
  365. if not k.startswith("cp_")
  366. }
  367. )
  368. prepare_database(db_conn, database_engine, config=None)
  369. db_conn.commit()
  370. @defer.inlineCallbacks
  371. def run(self):
  372. try:
  373. sqlite_db_pool = adbapi.ConnectionPool(
  374. self.sqlite_config["name"], **self.sqlite_config["args"]
  375. )
  376. postgres_db_pool = adbapi.ConnectionPool(
  377. self.postgres_config["name"], **self.postgres_config["args"]
  378. )
  379. sqlite_engine = create_engine(sqlite_config)
  380. postgres_engine = create_engine(postgres_config)
  381. self.sqlite_store = Store(sqlite_db_pool, sqlite_engine)
  382. self.postgres_store = Store(postgres_db_pool, postgres_engine)
  383. yield self.postgres_store.execute(postgres_engine.check_database)
  384. # Step 1. Set up databases.
  385. self.progress.set_state("Preparing SQLite3")
  386. self.setup_db(sqlite_config, sqlite_engine)
  387. self.progress.set_state("Preparing PostgreSQL")
  388. self.setup_db(postgres_config, postgres_engine)
  389. self.progress.set_state("Creating port tables")
  390. def create_port_table(txn):
  391. txn.execute(
  392. "CREATE TABLE IF NOT EXISTS port_from_sqlite3 ("
  393. " table_name varchar(100) NOT NULL UNIQUE,"
  394. " forward_rowid bigint NOT NULL,"
  395. " backward_rowid bigint NOT NULL"
  396. ")"
  397. )
  398. # The old port script created a table with just a "rowid" column.
  399. # We want people to be able to rerun this script from an old port
  400. # so that they can pick up any missing events that were not
  401. # ported across.
  402. def alter_table(txn):
  403. txn.execute(
  404. "ALTER TABLE IF EXISTS port_from_sqlite3"
  405. " RENAME rowid TO forward_rowid"
  406. )
  407. txn.execute(
  408. "ALTER TABLE IF EXISTS port_from_sqlite3"
  409. " ADD backward_rowid bigint NOT NULL DEFAULT 0"
  410. )
  411. try:
  412. yield self.postgres_store.runInteraction("alter_table", alter_table)
  413. except Exception:
  414. # On Error Resume Next
  415. pass
  416. yield self.postgres_store.runInteraction(
  417. "create_port_table", create_port_table
  418. )
  419. # Step 2. Get tables.
  420. self.progress.set_state("Fetching tables")
  421. sqlite_tables = yield self.sqlite_store._simple_select_onecol(
  422. table="sqlite_master", keyvalues={"type": "table"}, retcol="name"
  423. )
  424. postgres_tables = yield self.postgres_store._simple_select_onecol(
  425. table="information_schema.tables",
  426. keyvalues={},
  427. retcol="distinct table_name",
  428. )
  429. tables = set(sqlite_tables) & set(postgres_tables)
  430. logger.info("Found %d tables", len(tables))
  431. # Step 3. Figure out what still needs copying
  432. self.progress.set_state("Checking on port progress")
  433. setup_res = yield defer.gatherResults(
  434. [
  435. self.setup_table(table)
  436. for table in tables
  437. if table not in ["schema_version", "applied_schema_deltas"]
  438. and not table.startswith("sqlite_")
  439. ],
  440. consumeErrors=True,
  441. )
  442. # Step 4. Do the copying.
  443. self.progress.set_state("Copying to postgres")
  444. yield defer.gatherResults(
  445. [self.handle_table(*res) for res in setup_res], consumeErrors=True
  446. )
  447. # Step 5. Do final post-processing
  448. yield self._setup_state_group_id_seq()
  449. self.progress.done()
  450. except Exception:
  451. global end_error_exec_info
  452. end_error_exec_info = sys.exc_info()
  453. logger.exception("")
  454. finally:
  455. reactor.stop()
  456. def _convert_rows(self, table, headers, rows):
  457. bool_col_names = BOOLEAN_COLUMNS.get(table, [])
  458. bool_cols = [i for i, h in enumerate(headers) if h in bool_col_names]
  459. class BadValueException(Exception):
  460. pass
  461. def conv(j, col):
  462. if j in bool_cols:
  463. return bool(col)
  464. elif isinstance(col, string_types) and "\0" in col:
  465. logger.warn(
  466. "DROPPING ROW: NUL value in table %s col %s: %r",
  467. table,
  468. headers[j],
  469. col,
  470. )
  471. raise BadValueException()
  472. return col
  473. outrows = []
  474. for i, row in enumerate(rows):
  475. try:
  476. outrows.append(
  477. tuple(conv(j, col) for j, col in enumerate(row) if j > 0)
  478. )
  479. except BadValueException:
  480. pass
  481. return outrows
  482. @defer.inlineCallbacks
  483. def _setup_sent_transactions(self):
  484. # Only save things from the last day
  485. yesterday = int(time.time() * 1000) - 86400000
  486. # And save the max transaction id from each destination
  487. select = (
  488. "SELECT rowid, * FROM sent_transactions WHERE rowid IN ("
  489. "SELECT max(rowid) FROM sent_transactions"
  490. " GROUP BY destination"
  491. ")"
  492. )
  493. def r(txn):
  494. txn.execute(select)
  495. rows = txn.fetchall()
  496. headers = [column[0] for column in txn.description]
  497. ts_ind = headers.index('ts')
  498. return headers, [r for r in rows if r[ts_ind] < yesterday]
  499. headers, rows = yield self.sqlite_store.runInteraction("select", r)
  500. rows = self._convert_rows("sent_transactions", headers, rows)
  501. inserted_rows = len(rows)
  502. if inserted_rows:
  503. max_inserted_rowid = max(r[0] for r in rows)
  504. def insert(txn):
  505. self.postgres_store.insert_many_txn(
  506. txn, "sent_transactions", headers[1:], rows
  507. )
  508. yield self.postgres_store.execute(insert)
  509. else:
  510. max_inserted_rowid = 0
  511. def get_start_id(txn):
  512. txn.execute(
  513. "SELECT rowid FROM sent_transactions WHERE ts >= ?"
  514. " ORDER BY rowid ASC LIMIT 1",
  515. (yesterday,),
  516. )
  517. rows = txn.fetchall()
  518. if rows:
  519. return rows[0][0]
  520. else:
  521. return 1
  522. next_chunk = yield self.sqlite_store.execute(get_start_id)
  523. next_chunk = max(max_inserted_rowid + 1, next_chunk)
  524. yield self.postgres_store._simple_insert(
  525. table="port_from_sqlite3",
  526. values={
  527. "table_name": "sent_transactions",
  528. "forward_rowid": next_chunk,
  529. "backward_rowid": 0,
  530. },
  531. )
  532. def get_sent_table_size(txn):
  533. txn.execute(
  534. "SELECT count(*) FROM sent_transactions" " WHERE ts >= ?", (yesterday,)
  535. )
  536. size, = txn.fetchone()
  537. return int(size)
  538. remaining_count = yield self.sqlite_store.execute(get_sent_table_size)
  539. total_count = remaining_count + inserted_rows
  540. defer.returnValue((next_chunk, inserted_rows, total_count))
  541. @defer.inlineCallbacks
  542. def _get_remaining_count_to_port(self, table, forward_chunk, backward_chunk):
  543. frows = yield self.sqlite_store.execute_sql(
  544. "SELECT count(*) FROM %s WHERE rowid >= ?" % (table,), forward_chunk
  545. )
  546. brows = yield self.sqlite_store.execute_sql(
  547. "SELECT count(*) FROM %s WHERE rowid <= ?" % (table,), backward_chunk
  548. )
  549. defer.returnValue(frows[0][0] + brows[0][0])
  550. @defer.inlineCallbacks
  551. def _get_already_ported_count(self, table):
  552. rows = yield self.postgres_store.execute_sql(
  553. "SELECT count(*) FROM %s" % (table,)
  554. )
  555. defer.returnValue(rows[0][0])
  556. @defer.inlineCallbacks
  557. def _get_total_count_to_port(self, table, forward_chunk, backward_chunk):
  558. remaining, done = yield defer.gatherResults(
  559. [
  560. self._get_remaining_count_to_port(table, forward_chunk, backward_chunk),
  561. self._get_already_ported_count(table),
  562. ],
  563. consumeErrors=True,
  564. )
  565. remaining = int(remaining) if remaining else 0
  566. done = int(done) if done else 0
  567. defer.returnValue((done, remaining + done))
  568. def _setup_state_group_id_seq(self):
  569. def r(txn):
  570. txn.execute("SELECT MAX(id) FROM state_groups")
  571. next_id = txn.fetchone()[0] + 1
  572. txn.execute("ALTER SEQUENCE state_group_id_seq RESTART WITH %s", (next_id,))
  573. return self.postgres_store.runInteraction("setup_state_group_id_seq", r)
  574. ##############################################
  575. # The following is simply UI stuff
  576. ##############################################
  577. class Progress(object):
  578. """Used to report progress of the port
  579. """
  580. def __init__(self):
  581. self.tables = {}
  582. self.start_time = int(time.time())
  583. def add_table(self, table, cur, size):
  584. self.tables[table] = {
  585. "start": cur,
  586. "num_done": cur,
  587. "total": size,
  588. "perc": int(cur * 100 / size),
  589. }
  590. def update(self, table, num_done):
  591. data = self.tables[table]
  592. data["num_done"] = num_done
  593. data["perc"] = int(num_done * 100 / data["total"])
  594. def done(self):
  595. pass
  596. class CursesProgress(Progress):
  597. """Reports progress to a curses window
  598. """
  599. def __init__(self, stdscr):
  600. self.stdscr = stdscr
  601. curses.use_default_colors()
  602. curses.curs_set(0)
  603. curses.init_pair(1, curses.COLOR_RED, -1)
  604. curses.init_pair(2, curses.COLOR_GREEN, -1)
  605. self.last_update = 0
  606. self.finished = False
  607. self.total_processed = 0
  608. self.total_remaining = 0
  609. super(CursesProgress, self).__init__()
  610. def update(self, table, num_done):
  611. super(CursesProgress, self).update(table, num_done)
  612. self.total_processed = 0
  613. self.total_remaining = 0
  614. for table, data in self.tables.items():
  615. self.total_processed += data["num_done"] - data["start"]
  616. self.total_remaining += data["total"] - data["num_done"]
  617. self.render()
  618. def render(self, force=False):
  619. now = time.time()
  620. if not force and now - self.last_update < 0.2:
  621. # reactor.callLater(1, self.render)
  622. return
  623. self.stdscr.clear()
  624. rows, cols = self.stdscr.getmaxyx()
  625. duration = int(now) - int(self.start_time)
  626. minutes, seconds = divmod(duration, 60)
  627. duration_str = '%02dm %02ds' % (minutes, seconds)
  628. if self.finished:
  629. status = "Time spent: %s (Done!)" % (duration_str,)
  630. else:
  631. if self.total_processed > 0:
  632. left = float(self.total_remaining) / self.total_processed
  633. est_remaining = (int(now) - self.start_time) * left
  634. est_remaining_str = '%02dm %02ds remaining' % divmod(est_remaining, 60)
  635. else:
  636. est_remaining_str = "Unknown"
  637. status = "Time spent: %s (est. remaining: %s)" % (
  638. duration_str,
  639. est_remaining_str,
  640. )
  641. self.stdscr.addstr(0, 0, status, curses.A_BOLD)
  642. max_len = max([len(t) for t in self.tables.keys()])
  643. left_margin = 5
  644. middle_space = 1
  645. items = self.tables.items()
  646. items = sorted(items, key=lambda i: (i[1]["perc"], i[0]))
  647. for i, (table, data) in enumerate(items):
  648. if i + 2 >= rows:
  649. break
  650. perc = data["perc"]
  651. color = curses.color_pair(2) if perc == 100 else curses.color_pair(1)
  652. self.stdscr.addstr(
  653. i + 2, left_margin + max_len - len(table), table, curses.A_BOLD | color
  654. )
  655. size = 20
  656. progress = "[%s%s]" % (
  657. "#" * int(perc * size / 100),
  658. " " * (size - int(perc * size / 100)),
  659. )
  660. self.stdscr.addstr(
  661. i + 2,
  662. left_margin + max_len + middle_space,
  663. "%s %3d%% (%d/%d)" % (progress, perc, data["num_done"], data["total"]),
  664. )
  665. if self.finished:
  666. self.stdscr.addstr(rows - 1, 0, "Press any key to exit...")
  667. self.stdscr.refresh()
  668. self.last_update = time.time()
  669. def done(self):
  670. self.finished = True
  671. self.render(True)
  672. self.stdscr.getch()
  673. def set_state(self, state):
  674. self.stdscr.clear()
  675. self.stdscr.addstr(0, 0, state + "...", curses.A_BOLD)
  676. self.stdscr.refresh()
  677. class TerminalProgress(Progress):
  678. """Just prints progress to the terminal
  679. """
  680. def update(self, table, num_done):
  681. super(TerminalProgress, self).update(table, num_done)
  682. data = self.tables[table]
  683. print(
  684. "%s: %d%% (%d/%d)" % (table, data["perc"], data["num_done"], data["total"])
  685. )
  686. def set_state(self, state):
  687. print(state + "...")
  688. ##############################################
  689. ##############################################
  690. if __name__ == "__main__":
  691. parser = argparse.ArgumentParser(
  692. description="A script to port an existing synapse SQLite database to"
  693. " a new PostgreSQL database."
  694. )
  695. parser.add_argument("-v", action='store_true')
  696. parser.add_argument(
  697. "--sqlite-database",
  698. required=True,
  699. help="The snapshot of the SQLite database file. This must not be"
  700. " currently used by a running synapse server",
  701. )
  702. parser.add_argument(
  703. "--postgres-config",
  704. type=argparse.FileType('r'),
  705. required=True,
  706. help="The database config file for the PostgreSQL database",
  707. )
  708. parser.add_argument(
  709. "--curses", action='store_true', help="display a curses based progress UI"
  710. )
  711. parser.add_argument(
  712. "--batch-size",
  713. type=int,
  714. default=1000,
  715. help="The number of rows to select from the SQLite table each"
  716. " iteration [default=1000]",
  717. )
  718. args = parser.parse_args()
  719. logging_config = {
  720. "level": logging.DEBUG if args.v else logging.INFO,
  721. "format": "%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s",
  722. }
  723. if args.curses:
  724. logging_config["filename"] = "port-synapse.log"
  725. logging.basicConfig(**logging_config)
  726. sqlite_config = {
  727. "name": "sqlite3",
  728. "args": {
  729. "database": args.sqlite_database,
  730. "cp_min": 1,
  731. "cp_max": 1,
  732. "check_same_thread": False,
  733. },
  734. }
  735. postgres_config = yaml.safe_load(args.postgres_config)
  736. if "database" in postgres_config:
  737. postgres_config = postgres_config["database"]
  738. if "name" not in postgres_config:
  739. sys.stderr.write("Malformed database config: no 'name'")
  740. sys.exit(2)
  741. if postgres_config["name"] != "psycopg2":
  742. sys.stderr.write("Database must use 'psycopg2' connector.")
  743. sys.exit(3)
  744. def start(stdscr=None):
  745. if stdscr:
  746. progress = CursesProgress(stdscr)
  747. else:
  748. progress = TerminalProgress()
  749. porter = Porter(
  750. sqlite_config=sqlite_config,
  751. postgres_config=postgres_config,
  752. progress=progress,
  753. batch_size=args.batch_size,
  754. )
  755. reactor.callWhenRunning(porter.run)
  756. reactor.run()
  757. if args.curses:
  758. curses.wrapper(start)
  759. else:
  760. start()
  761. if end_error_exec_info:
  762. exc_type, exc_value, exc_traceback = end_error_exec_info
  763. traceback.print_exception(exc_type, exc_value, exc_traceback)