nominatim/tools/replication.py

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