1
0

synapse_port_db 31 KB

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