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