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 nominatim.db import status
  10 from nominatim.tools.exec_utils import run_osm2pgsql
  11 from nominatim.errors import UsageError
  12
  13 try:
  14     from osmium.replication.server import ReplicationServer
  15     from osmium import WriteHandler
  16 except ImportError 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