]> git.openstreetmap.org Git - nominatim.git/blob - nominatim/tools/replication.py
Merge remote-tracking branch 'upstream/master'
[nominatim.git] / nominatim / tools / replication.py
1 """
2 Functions for updating a database from a replication source.
3 """
4 import datetime as dt
5 from enum import Enum
6 import logging
7 import time
8
9 from ..db import status
10 from .exec_utils import run_osm2pgsql
11 from ..errors import UsageError
12
13 try:
14     from osmium.replication.server import ReplicationServer
15     from osmium import WriteHandler
16 except ModuleNotFoundError as exc:
17     logging.getLogger().fatal("pyosmium not installed. Replication functions not available.\n"
18                               "To install pyosmium via pip: pip3 install osmium")
19     raise UsageError("replication tools not available") from exc
20
21 LOG = logging.getLogger()
22
23 def init_replication(conn, base_url):
24     """ Set up replication for the server at the given base URL.
25     """
26     LOG.info("Using replication source: %s", base_url)
27     date = status.compute_database_date(conn)
28
29     # margin of error to make sure we get all data
30     date -= dt.timedelta(hours=3)
31
32     repl = ReplicationServer(base_url)
33
34     seq = repl.timestamp_to_sequence(date)
35
36     if seq is None:
37         LOG.fatal("Cannot reach the configured replication service '%s'.\n"
38                   "Does the URL point to a directory containing OSM update data?",
39                   base_url)
40         raise UsageError("Failed to reach replication service")
41
42     status.set_status(conn, date=date, seq=seq)
43
44     LOG.warning("Updates intialised at sequence %s (%s)", seq, date)
45
46
47 def check_for_updates(conn, base_url):
48     """ Check if new data is available from the replication service at the
49         given base URL.
50     """
51     _, seq, _ = status.get_status(conn)
52
53     if seq is None:
54         LOG.error("Replication not set up. "
55                   "Please run 'nominatim replication --init' first.")
56         return 254
57
58     state = ReplicationServer(base_url).get_state_info()
59
60     if state is None:
61         LOG.error("Cannot get state for URL %s.", base_url)
62         return 253
63
64     if state.sequence <= seq:
65         LOG.warning("Database is up to date.")
66         return 2
67
68     LOG.warning("New data available (%i => %i).", seq, state.sequence)
69     return 0
70
71 class UpdateState(Enum):
72     """ Possible states after an update has run.
73     """
74
75     UP_TO_DATE = 0
76     MORE_PENDING = 2
77     NO_CHANGES = 3
78
79
80 def update(conn, options):
81     """ Update database from the next batch of data. Returns the state of
82         updates according to `UpdateState`.
83     """
84     startdate, startseq, indexed = status.get_status(conn)
85
86     if startseq is None:
87         LOG.error("Replication not set up. "
88                   "Please run 'nominatim replication --init' first.")
89         raise UsageError("Replication not set up.")
90
91     if not indexed and options['indexed_only']:
92         LOG.info("Skipping update. There is data that needs indexing.")
93         return UpdateState.MORE_PENDING
94
95     last_since_update = dt.datetime.now(dt.timezone.utc) - startdate
96     update_interval = dt.timedelta(seconds=options['update_interval'])
97     if last_since_update < update_interval:
98         duration = (update_interval - last_since_update).seconds
99         LOG.warning("Sleeping for %s sec before next update.", duration)
100         time.sleep(duration)
101
102     if options['import_file'].exists():
103         options['import_file'].unlink()
104
105     # Read updates into file.
106     repl = ReplicationServer(options['base_url'])
107
108     outhandler = WriteHandler(str(options['import_file']))
109     endseq = repl.apply_diffs(outhandler, startseq + 1,
110                               max_size=options['max_diff_size'] * 1024)
111     outhandler.close()
112
113     if endseq is None:
114         return UpdateState.NO_CHANGES
115
116     # Consume updates with osm2pgsql.
117     options['append'] = True
118     options['disable_jit'] = conn.server_version_tuple() >= (11, 0)
119     run_osm2pgsql(options)
120
121     # Write the current status to the file
122     endstate = repl.get_state_info(endseq)
123     status.set_status(conn, endstate.timestamp if endstate else None,
124                       seq=endseq, indexed=False)
125
126     return UpdateState.UP_TO_DATE