synapse_port_db 36 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # Copyright 2015, 2016 OpenMarket Ltd
  4. # Copyright 2018 New Vector Ltd
  5. # Copyright 2019 The Matrix.org Foundation C.I.C.
  6. #
  7. # Licensed under the Apache License, Version 2.0 (the "License");
  8. # you may not use this file except in compliance with the License.
  9. # You may obtain a copy of the License at
  10. #
  11. # http://www.apache.org/licenses/LICENSE-2.0
  12. #
  13. # Unless required by applicable law or agreed to in writing, software
  14. # distributed under the License is distributed on an "AS IS" BASIS,
  15. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16. # See the License for the specific language governing permissions and
  17. # limitations under the License.
  18. import argparse
  19. import curses
  20. import logging
  21. import sys
  22. import time
  23. import traceback
  24. from typing import Optional
  25. import yaml
  26. from twisted.internet import defer, reactor
  27. import synapse
  28. from synapse.config.database import DatabaseConnectionConfig
  29. from synapse.config.homeserver import HomeServerConfig
  30. from synapse.logging.context import (
  31. LoggingContext,
  32. make_deferred_yieldable,
  33. run_in_background,
  34. )
  35. from synapse.storage.database import DatabasePool, make_conn
  36. from synapse.storage.databases.main.client_ips import ClientIpBackgroundUpdateStore
  37. from synapse.storage.databases.main.deviceinbox import DeviceInboxBackgroundUpdateStore
  38. from synapse.storage.databases.main.devices import DeviceBackgroundUpdateStore
  39. from synapse.storage.databases.main.events_bg_updates import (
  40. EventsBackgroundUpdatesStore,
  41. )
  42. from synapse.storage.databases.main.media_repository import (
  43. MediaRepositoryBackgroundUpdateStore,
  44. )
  45. from synapse.storage.databases.main.registration import (
  46. RegistrationBackgroundUpdateStore,
  47. find_max_generated_user_id_localpart,
  48. )
  49. from synapse.storage.databases.main.room import RoomBackgroundUpdateStore
  50. from synapse.storage.databases.main.roommember import RoomMemberBackgroundUpdateStore
  51. from synapse.storage.databases.main.search import SearchBackgroundUpdateStore
  52. from synapse.storage.databases.main.state import MainStateBackgroundUpdateStore
  53. from synapse.storage.databases.main.stats import StatsStore
  54. from synapse.storage.databases.main.user_directory import (
  55. UserDirectoryBackgroundUpdateStore,
  56. )
  57. from synapse.storage.databases.state.bg_updates import StateBackgroundUpdateStore
  58. from synapse.storage.engines import create_engine
  59. from synapse.storage.prepare_database import prepare_database
  60. from synapse.util import Clock
  61. from synapse.util.versionstring import get_version_string
  62. logger = logging.getLogger("synapse_port_db")
  63. BOOLEAN_COLUMNS = {
  64. "events": ["processed", "outlier", "contains_url"],
  65. "rooms": ["is_public"],
  66. "event_edges": ["is_state"],
  67. "presence_list": ["accepted"],
  68. "presence_stream": ["currently_active"],
  69. "public_room_list_stream": ["visibility"],
  70. "devices": ["hidden"],
  71. "device_lists_outbound_pokes": ["sent"],
  72. "users_who_share_rooms": ["share_private"],
  73. "groups": ["is_public"],
  74. "group_rooms": ["is_public"],
  75. "group_users": ["is_public", "is_admin"],
  76. "group_summary_rooms": ["is_public"],
  77. "group_room_categories": ["is_public"],
  78. "group_summary_users": ["is_public"],
  79. "group_roles": ["is_public"],
  80. "local_group_membership": ["is_publicised", "is_admin"],
  81. "e2e_room_keys": ["is_verified"],
  82. "account_validity": ["email_sent"],
  83. "redactions": ["have_censored"],
  84. "room_stats_state": ["is_federatable"],
  85. "local_media_repository": ["safe_from_quarantine"],
  86. "users": ["shadow_banned"],
  87. "e2e_fallback_keys_json": ["used"],
  88. }
  89. APPEND_ONLY_TABLES = [
  90. "event_reference_hashes",
  91. "events",
  92. "event_json",
  93. "state_events",
  94. "room_memberships",
  95. "topics",
  96. "room_names",
  97. "rooms",
  98. "local_media_repository",
  99. "local_media_repository_thumbnails",
  100. "remote_media_cache",
  101. "remote_media_cache_thumbnails",
  102. "redactions",
  103. "event_edges",
  104. "event_auth",
  105. "received_transactions",
  106. "sent_transactions",
  107. "transaction_id_to_pdu",
  108. "users",
  109. "state_groups",
  110. "state_groups_state",
  111. "event_to_state_groups",
  112. "rejections",
  113. "event_search",
  114. "presence_stream",
  115. "push_rules_stream",
  116. "ex_outlier_stream",
  117. "cache_invalidation_stream_by_instance",
  118. "public_room_list_stream",
  119. "state_group_edges",
  120. "stream_ordering_to_exterm",
  121. ]
  122. IGNORED_TABLES = {
  123. # We don't port these tables, as they're a faff and we can regenerate
  124. # them anyway.
  125. "user_directory",
  126. "user_directory_search",
  127. "user_directory_search_content",
  128. "user_directory_search_docsize",
  129. "user_directory_search_segdir",
  130. "user_directory_search_segments",
  131. "user_directory_search_stat",
  132. "user_directory_search_pos",
  133. "users_who_share_private_rooms",
  134. "users_in_public_room",
  135. # UI auth sessions have foreign keys so additional care needs to be taken,
  136. # the sessions are transient anyway, so ignore them.
  137. "ui_auth_sessions",
  138. "ui_auth_sessions_credentials",
  139. "ui_auth_sessions_ips",
  140. }
  141. # Error returned by the run function. Used at the top-level part of the script to
  142. # handle errors and return codes.
  143. end_error = None # type: Optional[str]
  144. # The exec_info for the error, if any. If error is defined but not exec_info the script
  145. # will show only the error message without the stacktrace, if exec_info is defined but
  146. # not the error then the script will show nothing outside of what's printed in the run
  147. # function. If both are defined, the script will print both the error and the stacktrace.
  148. end_error_exec_info = None
  149. class Store(
  150. ClientIpBackgroundUpdateStore,
  151. DeviceInboxBackgroundUpdateStore,
  152. DeviceBackgroundUpdateStore,
  153. EventsBackgroundUpdatesStore,
  154. MediaRepositoryBackgroundUpdateStore,
  155. RegistrationBackgroundUpdateStore,
  156. RoomBackgroundUpdateStore,
  157. RoomMemberBackgroundUpdateStore,
  158. SearchBackgroundUpdateStore,
  159. StateBackgroundUpdateStore,
  160. MainStateBackgroundUpdateStore,
  161. UserDirectoryBackgroundUpdateStore,
  162. StatsStore,
  163. ):
  164. def execute(self, f, *args, **kwargs):
  165. return self.db_pool.runInteraction(f.__name__, f, *args, **kwargs)
  166. def execute_sql(self, sql, *args):
  167. def r(txn):
  168. txn.execute(sql, args)
  169. return txn.fetchall()
  170. return self.db_pool.runInteraction("execute_sql", r)
  171. def insert_many_txn(self, txn, table, headers, rows):
  172. sql = "INSERT INTO %s (%s) VALUES (%s)" % (
  173. table,
  174. ", ".join(k for k in headers),
  175. ", ".join("%s" for _ in headers),
  176. )
  177. try:
  178. txn.executemany(sql, rows)
  179. except Exception:
  180. logger.exception("Failed to insert: %s", table)
  181. raise
  182. def set_room_is_public(self, room_id, is_public):
  183. raise Exception(
  184. "Attempt to set room_is_public during port_db: database not empty?"
  185. )
  186. class MockHomeserver:
  187. def __init__(self, config):
  188. self.clock = Clock(reactor)
  189. self.config = config
  190. self.hostname = config.server_name
  191. self.version_string = "Synapse/" + get_version_string(synapse)
  192. def get_clock(self):
  193. return self.clock
  194. def get_reactor(self):
  195. return reactor
  196. def get_instance_name(self):
  197. return "master"
  198. class Porter(object):
  199. def __init__(self, **kwargs):
  200. self.__dict__.update(kwargs)
  201. async def setup_table(self, table):
  202. if table in APPEND_ONLY_TABLES:
  203. # It's safe to just carry on inserting.
  204. row = await self.postgres_store.db_pool.simple_select_one(
  205. table="port_from_sqlite3",
  206. keyvalues={"table_name": table},
  207. retcols=("forward_rowid", "backward_rowid"),
  208. allow_none=True,
  209. )
  210. total_to_port = None
  211. if row is None:
  212. if table == "sent_transactions":
  213. (
  214. forward_chunk,
  215. already_ported,
  216. total_to_port,
  217. ) = await self._setup_sent_transactions()
  218. backward_chunk = 0
  219. else:
  220. await self.postgres_store.db_pool.simple_insert(
  221. table="port_from_sqlite3",
  222. values={
  223. "table_name": table,
  224. "forward_rowid": 1,
  225. "backward_rowid": 0,
  226. },
  227. )
  228. forward_chunk = 1
  229. backward_chunk = 0
  230. already_ported = 0
  231. else:
  232. forward_chunk = row["forward_rowid"]
  233. backward_chunk = row["backward_rowid"]
  234. if total_to_port is None:
  235. already_ported, total_to_port = await self._get_total_count_to_port(
  236. table, forward_chunk, backward_chunk
  237. )
  238. else:
  239. def delete_all(txn):
  240. txn.execute(
  241. "DELETE FROM port_from_sqlite3 WHERE table_name = %s", (table,)
  242. )
  243. txn.execute("TRUNCATE %s CASCADE" % (table,))
  244. await self.postgres_store.execute(delete_all)
  245. await self.postgres_store.db_pool.simple_insert(
  246. table="port_from_sqlite3",
  247. values={"table_name": table, "forward_rowid": 1, "backward_rowid": 0},
  248. )
  249. forward_chunk = 1
  250. backward_chunk = 0
  251. already_ported, total_to_port = await self._get_total_count_to_port(
  252. table, forward_chunk, backward_chunk
  253. )
  254. return table, already_ported, total_to_port, forward_chunk, backward_chunk
  255. async def handle_table(
  256. self, table, postgres_size, table_size, forward_chunk, backward_chunk
  257. ):
  258. logger.info(
  259. "Table %s: %i/%i (rows %i-%i) already ported",
  260. table,
  261. postgres_size,
  262. table_size,
  263. backward_chunk + 1,
  264. forward_chunk - 1,
  265. )
  266. if not table_size:
  267. return
  268. self.progress.add_table(table, postgres_size, table_size)
  269. if table == "event_search":
  270. await self.handle_search_table(
  271. postgres_size, table_size, forward_chunk, backward_chunk
  272. )
  273. return
  274. if table in IGNORED_TABLES:
  275. self.progress.update(table, table_size) # Mark table as done
  276. return
  277. if table == "user_directory_stream_pos":
  278. # We need to make sure there is a single row, `(X, null), as that is
  279. # what synapse expects to be there.
  280. await self.postgres_store.db_pool.simple_insert(
  281. table=table, values={"stream_id": None}
  282. )
  283. self.progress.update(table, table_size) # Mark table as done
  284. return
  285. forward_select = (
  286. "SELECT rowid, * FROM %s WHERE rowid >= ? ORDER BY rowid LIMIT ?" % (table,)
  287. )
  288. backward_select = (
  289. "SELECT rowid, * FROM %s WHERE rowid <= ? ORDER BY rowid LIMIT ?" % (table,)
  290. )
  291. do_forward = [True]
  292. do_backward = [True]
  293. while True:
  294. def r(txn):
  295. forward_rows = []
  296. backward_rows = []
  297. if do_forward[0]:
  298. txn.execute(forward_select, (forward_chunk, self.batch_size))
  299. forward_rows = txn.fetchall()
  300. if not forward_rows:
  301. do_forward[0] = False
  302. if do_backward[0]:
  303. txn.execute(backward_select, (backward_chunk, self.batch_size))
  304. backward_rows = txn.fetchall()
  305. if not backward_rows:
  306. do_backward[0] = False
  307. if forward_rows or backward_rows:
  308. headers = [column[0] for column in txn.description]
  309. else:
  310. headers = None
  311. return headers, forward_rows, backward_rows
  312. headers, frows, brows = await self.sqlite_store.db_pool.runInteraction(
  313. "select", r
  314. )
  315. if frows or brows:
  316. if frows:
  317. forward_chunk = max(row[0] for row in frows) + 1
  318. if brows:
  319. backward_chunk = min(row[0] for row in brows) - 1
  320. rows = frows + brows
  321. rows = self._convert_rows(table, headers, rows)
  322. def insert(txn):
  323. self.postgres_store.insert_many_txn(txn, table, headers[1:], rows)
  324. self.postgres_store.db_pool.simple_update_one_txn(
  325. txn,
  326. table="port_from_sqlite3",
  327. keyvalues={"table_name": table},
  328. updatevalues={
  329. "forward_rowid": forward_chunk,
  330. "backward_rowid": backward_chunk,
  331. },
  332. )
  333. await self.postgres_store.execute(insert)
  334. postgres_size += len(rows)
  335. self.progress.update(table, postgres_size)
  336. else:
  337. return
  338. async def handle_search_table(
  339. self, postgres_size, table_size, forward_chunk, backward_chunk
  340. ):
  341. select = (
  342. "SELECT es.rowid, es.*, e.origin_server_ts, e.stream_ordering"
  343. " FROM event_search as es"
  344. " INNER JOIN events AS e USING (event_id, room_id)"
  345. " WHERE es.rowid >= ?"
  346. " ORDER BY es.rowid LIMIT ?"
  347. )
  348. while True:
  349. def r(txn):
  350. txn.execute(select, (forward_chunk, self.batch_size))
  351. rows = txn.fetchall()
  352. headers = [column[0] for column in txn.description]
  353. return headers, rows
  354. headers, rows = await self.sqlite_store.db_pool.runInteraction("select", r)
  355. if rows:
  356. forward_chunk = rows[-1][0] + 1
  357. # We have to treat event_search differently since it has a
  358. # different structure in the two different databases.
  359. def insert(txn):
  360. sql = (
  361. "INSERT INTO event_search (event_id, room_id, key,"
  362. " sender, vector, origin_server_ts, stream_ordering)"
  363. " VALUES (?,?,?,?,to_tsvector('english', ?),?,?)"
  364. )
  365. rows_dict = []
  366. for row in rows:
  367. d = dict(zip(headers, row))
  368. if "\0" in d["value"]:
  369. logger.warning("dropping search row %s", d)
  370. else:
  371. rows_dict.append(d)
  372. txn.executemany(
  373. sql,
  374. [
  375. (
  376. row["event_id"],
  377. row["room_id"],
  378. row["key"],
  379. row["sender"],
  380. row["value"],
  381. row["origin_server_ts"],
  382. row["stream_ordering"],
  383. )
  384. for row in rows_dict
  385. ],
  386. )
  387. self.postgres_store.db_pool.simple_update_one_txn(
  388. txn,
  389. table="port_from_sqlite3",
  390. keyvalues={"table_name": "event_search"},
  391. updatevalues={
  392. "forward_rowid": forward_chunk,
  393. "backward_rowid": backward_chunk,
  394. },
  395. )
  396. await self.postgres_store.execute(insert)
  397. postgres_size += len(rows)
  398. self.progress.update("event_search", postgres_size)
  399. else:
  400. return
  401. def build_db_store(
  402. self, db_config: DatabaseConnectionConfig, allow_outdated_version: bool = False,
  403. ):
  404. """Builds and returns a database store using the provided configuration.
  405. Args:
  406. db_config: The database configuration
  407. allow_outdated_version: True to suppress errors about the database server
  408. version being too old to run a complete synapse
  409. Returns:
  410. The built Store object.
  411. """
  412. self.progress.set_state("Preparing %s" % db_config.config["name"])
  413. engine = create_engine(db_config.config)
  414. hs = MockHomeserver(self.hs_config)
  415. with make_conn(db_config, engine, "portdb") as db_conn:
  416. engine.check_database(
  417. db_conn, allow_outdated_version=allow_outdated_version
  418. )
  419. prepare_database(db_conn, engine, config=self.hs_config)
  420. store = Store(DatabasePool(hs, db_config, engine), db_conn, hs)
  421. db_conn.commit()
  422. return store
  423. async def run_background_updates_on_postgres(self):
  424. # Manually apply all background updates on the PostgreSQL database.
  425. postgres_ready = (
  426. await self.postgres_store.db_pool.updates.has_completed_background_updates()
  427. )
  428. if not postgres_ready:
  429. # Only say that we're running background updates when there are background
  430. # updates to run.
  431. self.progress.set_state("Running background updates on PostgreSQL")
  432. while not postgres_ready:
  433. await self.postgres_store.db_pool.updates.do_next_background_update(100)
  434. postgres_ready = await (
  435. self.postgres_store.db_pool.updates.has_completed_background_updates()
  436. )
  437. async def run(self):
  438. """Ports the SQLite database to a PostgreSQL database.
  439. When a fatal error is met, its message is assigned to the global "end_error"
  440. variable. When this error comes with a stacktrace, its exec_info is assigned to
  441. the global "end_error_exec_info" variable.
  442. """
  443. global end_error
  444. try:
  445. # we allow people to port away from outdated versions of sqlite.
  446. self.sqlite_store = self.build_db_store(
  447. DatabaseConnectionConfig("master-sqlite", self.sqlite_config),
  448. allow_outdated_version=True,
  449. )
  450. # Check if all background updates are done, abort if not.
  451. updates_complete = (
  452. await self.sqlite_store.db_pool.updates.has_completed_background_updates()
  453. )
  454. if not updates_complete:
  455. end_error = (
  456. "Pending background updates exist in the SQLite3 database."
  457. " Please start Synapse again and wait until every update has finished"
  458. " before running this script.\n"
  459. )
  460. return
  461. self.postgres_store = self.build_db_store(
  462. self.hs_config.get_single_database()
  463. )
  464. await self.run_background_updates_on_postgres()
  465. self.progress.set_state("Creating port tables")
  466. def create_port_table(txn):
  467. txn.execute(
  468. "CREATE TABLE IF NOT EXISTS port_from_sqlite3 ("
  469. " table_name varchar(100) NOT NULL UNIQUE,"
  470. " forward_rowid bigint NOT NULL,"
  471. " backward_rowid bigint NOT NULL"
  472. ")"
  473. )
  474. # The old port script created a table with just a "rowid" column.
  475. # We want people to be able to rerun this script from an old port
  476. # so that they can pick up any missing events that were not
  477. # ported across.
  478. def alter_table(txn):
  479. txn.execute(
  480. "ALTER TABLE IF EXISTS port_from_sqlite3"
  481. " RENAME rowid TO forward_rowid"
  482. )
  483. txn.execute(
  484. "ALTER TABLE IF EXISTS port_from_sqlite3"
  485. " ADD backward_rowid bigint NOT NULL DEFAULT 0"
  486. )
  487. try:
  488. await self.postgres_store.db_pool.runInteraction(
  489. "alter_table", alter_table
  490. )
  491. except Exception:
  492. # On Error Resume Next
  493. pass
  494. await self.postgres_store.db_pool.runInteraction(
  495. "create_port_table", create_port_table
  496. )
  497. # Step 2. Get tables.
  498. self.progress.set_state("Fetching tables")
  499. sqlite_tables = await self.sqlite_store.db_pool.simple_select_onecol(
  500. table="sqlite_master", keyvalues={"type": "table"}, retcol="name"
  501. )
  502. postgres_tables = await self.postgres_store.db_pool.simple_select_onecol(
  503. table="information_schema.tables",
  504. keyvalues={},
  505. retcol="distinct table_name",
  506. )
  507. tables = set(sqlite_tables) & set(postgres_tables)
  508. logger.info("Found %d tables", len(tables))
  509. # Step 3. Figure out what still needs copying
  510. self.progress.set_state("Checking on port progress")
  511. setup_res = await make_deferred_yieldable(
  512. defer.gatherResults(
  513. [
  514. run_in_background(self.setup_table, table)
  515. for table in tables
  516. if table not in ["schema_version", "applied_schema_deltas"]
  517. and not table.startswith("sqlite_")
  518. ],
  519. consumeErrors=True,
  520. )
  521. )
  522. # Step 4. Do the copying.
  523. self.progress.set_state("Copying to postgres")
  524. await make_deferred_yieldable(
  525. defer.gatherResults(
  526. [run_in_background(self.handle_table, *res) for res in setup_res],
  527. consumeErrors=True,
  528. )
  529. )
  530. # Step 5. Set up sequences
  531. self.progress.set_state("Setting up sequence generators")
  532. await self._setup_state_group_id_seq()
  533. await self._setup_user_id_seq()
  534. await self._setup_events_stream_seqs()
  535. self.progress.done()
  536. except Exception as e:
  537. global end_error_exec_info
  538. end_error = str(e)
  539. end_error_exec_info = sys.exc_info()
  540. logger.exception("")
  541. finally:
  542. reactor.stop()
  543. def _convert_rows(self, table, headers, rows):
  544. bool_col_names = BOOLEAN_COLUMNS.get(table, [])
  545. bool_cols = [i for i, h in enumerate(headers) if h in bool_col_names]
  546. class BadValueException(Exception):
  547. pass
  548. def conv(j, col):
  549. if j in bool_cols:
  550. return bool(col)
  551. if isinstance(col, bytes):
  552. return bytearray(col)
  553. elif isinstance(col, str) and "\0" in col:
  554. logger.warning(
  555. "DROPPING ROW: NUL value in table %s col %s: %r",
  556. table,
  557. headers[j],
  558. col,
  559. )
  560. raise BadValueException()
  561. return col
  562. outrows = []
  563. for i, row in enumerate(rows):
  564. try:
  565. outrows.append(
  566. tuple(conv(j, col) for j, col in enumerate(row) if j > 0)
  567. )
  568. except BadValueException:
  569. pass
  570. return outrows
  571. async def _setup_sent_transactions(self):
  572. # Only save things from the last day
  573. yesterday = int(time.time() * 1000) - 86400000
  574. # And save the max transaction id from each destination
  575. select = (
  576. "SELECT rowid, * FROM sent_transactions WHERE rowid IN ("
  577. "SELECT max(rowid) FROM sent_transactions"
  578. " GROUP BY destination"
  579. ")"
  580. )
  581. def r(txn):
  582. txn.execute(select)
  583. rows = txn.fetchall()
  584. headers = [column[0] for column in txn.description]
  585. ts_ind = headers.index("ts")
  586. return headers, [r for r in rows if r[ts_ind] < yesterday]
  587. headers, rows = await self.sqlite_store.db_pool.runInteraction("select", r)
  588. rows = self._convert_rows("sent_transactions", headers, rows)
  589. inserted_rows = len(rows)
  590. if inserted_rows:
  591. max_inserted_rowid = max(r[0] for r in rows)
  592. def insert(txn):
  593. self.postgres_store.insert_many_txn(
  594. txn, "sent_transactions", headers[1:], rows
  595. )
  596. await self.postgres_store.execute(insert)
  597. else:
  598. max_inserted_rowid = 0
  599. def get_start_id(txn):
  600. txn.execute(
  601. "SELECT rowid FROM sent_transactions WHERE ts >= ?"
  602. " ORDER BY rowid ASC LIMIT 1",
  603. (yesterday,),
  604. )
  605. rows = txn.fetchall()
  606. if rows:
  607. return rows[0][0]
  608. else:
  609. return 1
  610. next_chunk = await self.sqlite_store.execute(get_start_id)
  611. next_chunk = max(max_inserted_rowid + 1, next_chunk)
  612. await self.postgres_store.db_pool.simple_insert(
  613. table="port_from_sqlite3",
  614. values={
  615. "table_name": "sent_transactions",
  616. "forward_rowid": next_chunk,
  617. "backward_rowid": 0,
  618. },
  619. )
  620. def get_sent_table_size(txn):
  621. txn.execute(
  622. "SELECT count(*) FROM sent_transactions" " WHERE ts >= ?", (yesterday,)
  623. )
  624. (size,) = txn.fetchone()
  625. return int(size)
  626. remaining_count = await self.sqlite_store.execute(get_sent_table_size)
  627. total_count = remaining_count + inserted_rows
  628. return next_chunk, inserted_rows, total_count
  629. async def _get_remaining_count_to_port(self, table, forward_chunk, backward_chunk):
  630. frows = await self.sqlite_store.execute_sql(
  631. "SELECT count(*) FROM %s WHERE rowid >= ?" % (table,), forward_chunk
  632. )
  633. brows = await self.sqlite_store.execute_sql(
  634. "SELECT count(*) FROM %s WHERE rowid <= ?" % (table,), backward_chunk
  635. )
  636. return frows[0][0] + brows[0][0]
  637. async def _get_already_ported_count(self, table):
  638. rows = await self.postgres_store.execute_sql(
  639. "SELECT count(*) FROM %s" % (table,)
  640. )
  641. return rows[0][0]
  642. async def _get_total_count_to_port(self, table, forward_chunk, backward_chunk):
  643. remaining, done = await make_deferred_yieldable(
  644. defer.gatherResults(
  645. [
  646. run_in_background(
  647. self._get_remaining_count_to_port,
  648. table,
  649. forward_chunk,
  650. backward_chunk,
  651. ),
  652. run_in_background(self._get_already_ported_count, table),
  653. ],
  654. )
  655. )
  656. remaining = int(remaining) if remaining else 0
  657. done = int(done) if done else 0
  658. return done, remaining + done
  659. def _setup_state_group_id_seq(self):
  660. def r(txn):
  661. txn.execute("SELECT MAX(id) FROM state_groups")
  662. curr_id = txn.fetchone()[0]
  663. if not curr_id:
  664. return
  665. next_id = curr_id + 1
  666. txn.execute("ALTER SEQUENCE state_group_id_seq RESTART WITH %s", (next_id,))
  667. return self.postgres_store.db_pool.runInteraction("setup_state_group_id_seq", r)
  668. def _setup_user_id_seq(self):
  669. def r(txn):
  670. next_id = find_max_generated_user_id_localpart(txn) + 1
  671. txn.execute("ALTER SEQUENCE user_id_seq RESTART WITH %s", (next_id,))
  672. return self.postgres_store.db_pool.runInteraction("setup_user_id_seq", r)
  673. def _setup_events_stream_seqs(self):
  674. def r(txn):
  675. txn.execute("SELECT MAX(stream_ordering) FROM events")
  676. curr_id = txn.fetchone()[0]
  677. if curr_id:
  678. next_id = curr_id + 1
  679. txn.execute(
  680. "ALTER SEQUENCE events_stream_seq RESTART WITH %s", (next_id,)
  681. )
  682. txn.execute("SELECT -MIN(stream_ordering) FROM events")
  683. curr_id = txn.fetchone()[0]
  684. if curr_id:
  685. next_id = curr_id + 1
  686. txn.execute(
  687. "ALTER SEQUENCE events_backfill_stream_seq RESTART WITH %s",
  688. (next_id,),
  689. )
  690. return self.postgres_store.db_pool.runInteraction(
  691. "_setup_events_stream_seqs", r
  692. )
  693. ##############################################
  694. # The following is simply UI stuff
  695. ##############################################
  696. class Progress(object):
  697. """Used to report progress of the port
  698. """
  699. def __init__(self):
  700. self.tables = {}
  701. self.start_time = int(time.time())
  702. def add_table(self, table, cur, size):
  703. self.tables[table] = {
  704. "start": cur,
  705. "num_done": cur,
  706. "total": size,
  707. "perc": int(cur * 100 / size),
  708. }
  709. def update(self, table, num_done):
  710. data = self.tables[table]
  711. data["num_done"] = num_done
  712. data["perc"] = int(num_done * 100 / data["total"])
  713. def done(self):
  714. pass
  715. class CursesProgress(Progress):
  716. """Reports progress to a curses window
  717. """
  718. def __init__(self, stdscr):
  719. self.stdscr = stdscr
  720. curses.use_default_colors()
  721. curses.curs_set(0)
  722. curses.init_pair(1, curses.COLOR_RED, -1)
  723. curses.init_pair(2, curses.COLOR_GREEN, -1)
  724. self.last_update = 0
  725. self.finished = False
  726. self.total_processed = 0
  727. self.total_remaining = 0
  728. super(CursesProgress, self).__init__()
  729. def update(self, table, num_done):
  730. super(CursesProgress, self).update(table, num_done)
  731. self.total_processed = 0
  732. self.total_remaining = 0
  733. for table, data in self.tables.items():
  734. self.total_processed += data["num_done"] - data["start"]
  735. self.total_remaining += data["total"] - data["num_done"]
  736. self.render()
  737. def render(self, force=False):
  738. now = time.time()
  739. if not force and now - self.last_update < 0.2:
  740. # reactor.callLater(1, self.render)
  741. return
  742. self.stdscr.clear()
  743. rows, cols = self.stdscr.getmaxyx()
  744. duration = int(now) - int(self.start_time)
  745. minutes, seconds = divmod(duration, 60)
  746. duration_str = "%02dm %02ds" % (minutes, seconds)
  747. if self.finished:
  748. status = "Time spent: %s (Done!)" % (duration_str,)
  749. else:
  750. if self.total_processed > 0:
  751. left = float(self.total_remaining) / self.total_processed
  752. est_remaining = (int(now) - self.start_time) * left
  753. est_remaining_str = "%02dm %02ds remaining" % divmod(est_remaining, 60)
  754. else:
  755. est_remaining_str = "Unknown"
  756. status = "Time spent: %s (est. remaining: %s)" % (
  757. duration_str,
  758. est_remaining_str,
  759. )
  760. self.stdscr.addstr(0, 0, status, curses.A_BOLD)
  761. max_len = max([len(t) for t in self.tables.keys()])
  762. left_margin = 5
  763. middle_space = 1
  764. items = self.tables.items()
  765. items = sorted(items, key=lambda i: (i[1]["perc"], i[0]))
  766. for i, (table, data) in enumerate(items):
  767. if i + 2 >= rows:
  768. break
  769. perc = data["perc"]
  770. color = curses.color_pair(2) if perc == 100 else curses.color_pair(1)
  771. self.stdscr.addstr(
  772. i + 2, left_margin + max_len - len(table), table, curses.A_BOLD | color
  773. )
  774. size = 20
  775. progress = "[%s%s]" % (
  776. "#" * int(perc * size / 100),
  777. " " * (size - int(perc * size / 100)),
  778. )
  779. self.stdscr.addstr(
  780. i + 2,
  781. left_margin + max_len + middle_space,
  782. "%s %3d%% (%d/%d)" % (progress, perc, data["num_done"], data["total"]),
  783. )
  784. if self.finished:
  785. self.stdscr.addstr(rows - 1, 0, "Press any key to exit...")
  786. self.stdscr.refresh()
  787. self.last_update = time.time()
  788. def done(self):
  789. self.finished = True
  790. self.render(True)
  791. self.stdscr.getch()
  792. def set_state(self, state):
  793. self.stdscr.clear()
  794. self.stdscr.addstr(0, 0, state + "...", curses.A_BOLD)
  795. self.stdscr.refresh()
  796. class TerminalProgress(Progress):
  797. """Just prints progress to the terminal
  798. """
  799. def update(self, table, num_done):
  800. super(TerminalProgress, self).update(table, num_done)
  801. data = self.tables[table]
  802. print(
  803. "%s: %d%% (%d/%d)" % (table, data["perc"], data["num_done"], data["total"])
  804. )
  805. def set_state(self, state):
  806. print(state + "...")
  807. ##############################################
  808. ##############################################
  809. if __name__ == "__main__":
  810. parser = argparse.ArgumentParser(
  811. description="A script to port an existing synapse SQLite database to"
  812. " a new PostgreSQL database."
  813. )
  814. parser.add_argument("-v", action="store_true")
  815. parser.add_argument(
  816. "--sqlite-database",
  817. required=True,
  818. help="The snapshot of the SQLite database file. This must not be"
  819. " currently used by a running synapse server",
  820. )
  821. parser.add_argument(
  822. "--postgres-config",
  823. type=argparse.FileType("r"),
  824. required=True,
  825. help="The database config file for the PostgreSQL database",
  826. )
  827. parser.add_argument(
  828. "--curses", action="store_true", help="display a curses based progress UI"
  829. )
  830. parser.add_argument(
  831. "--batch-size",
  832. type=int,
  833. default=1000,
  834. help="The number of rows to select from the SQLite table each"
  835. " iteration [default=1000]",
  836. )
  837. args = parser.parse_args()
  838. logging_config = {
  839. "level": logging.DEBUG if args.v else logging.INFO,
  840. "format": "%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s",
  841. }
  842. if args.curses:
  843. logging_config["filename"] = "port-synapse.log"
  844. logging.basicConfig(**logging_config)
  845. sqlite_config = {
  846. "name": "sqlite3",
  847. "args": {
  848. "database": args.sqlite_database,
  849. "cp_min": 1,
  850. "cp_max": 1,
  851. "check_same_thread": False,
  852. },
  853. }
  854. hs_config = yaml.safe_load(args.postgres_config)
  855. if "database" not in hs_config:
  856. sys.stderr.write("The configuration file must have a 'database' section.\n")
  857. sys.exit(4)
  858. postgres_config = hs_config["database"]
  859. if "name" not in postgres_config:
  860. sys.stderr.write("Malformed database config: no 'name'\n")
  861. sys.exit(2)
  862. if postgres_config["name"] != "psycopg2":
  863. sys.stderr.write("Database must use the 'psycopg2' connector.\n")
  864. sys.exit(3)
  865. config = HomeServerConfig()
  866. config.parse_config_dict(hs_config, "", "")
  867. def start(stdscr=None):
  868. if stdscr:
  869. progress = CursesProgress(stdscr)
  870. else:
  871. progress = TerminalProgress()
  872. porter = Porter(
  873. sqlite_config=sqlite_config,
  874. progress=progress,
  875. batch_size=args.batch_size,
  876. hs_config=config,
  877. )
  878. @defer.inlineCallbacks
  879. def run():
  880. with LoggingContext("synapse_port_db_run"):
  881. yield defer.ensureDeferred(porter.run())
  882. reactor.callWhenRunning(run)
  883. reactor.run()
  884. if args.curses:
  885. curses.wrapper(start)
  886. else:
  887. start()
  888. if end_error:
  889. if end_error_exec_info:
  890. exc_type, exc_value, exc_traceback = end_error_exec_info
  891. traceback.print_exception(exc_type, exc_value, exc_traceback)
  892. sys.stderr.write(end_error)
  893. sys.exit(5)