src/backend/mirrors.py

   1 #!/usr/bin/python
   2
   3 import datetime
   4 import logging
   5 import iso3166
   6 import math
   7 import os.path
   8 import random
   9 import socket
  10 import time
  11 import ssl
  12 import tornado.httpclient
  13 import tornado.iostream
  14 import tornado.netutil
  15 import urllib.parse
  16
  17 from . import countries
  18 from .misc import Object
  19 from .decorators import *
  20
  21 class Mirrors(Object):
  22         def _get_mirrors(self, query, *args):
  23                 res = self.db.query(query, *args)
  24
  25                 for row in res:
  26                         yield Mirror(self.backend, row.id, data=row)
  27
  28         def _get_mirror(self, query, *args):
  29                 res = self.db.get(query, *args)
  30
  31                 if res:
  32                         return Mirror(self.backend, res.id, data=res)
  33
  34         def __iter__(self):
  35                 mirrors = self._get_mirrors("SELECT * FROM mirrors \
  36                         WHERE enabled IS TRUE ORDER BY hostname")
  37
  38                 return iter(mirrors)
  39
  40         async def check_all(self):
  41                 for mirror in self:
  42                         with self.db.transaction():
  43                                 await mirror.check()
  44
  45         def get_for_download(self, filename, country_code=None):
  46                 # Try to find a good mirror for this country first
  47                 if country_code:
  48                         zone = countries.get_zone(country_code)
  49
  50                         mirror = self._get_mirror("SELECT mirrors.* FROM mirror_files files \
  51                                 LEFT JOIN mirrors ON files.mirror = mirrors.id \
  52                                 WHERE files.filename = %s \
  53                                         AND mirrors.enabled IS TRUE AND mirrors.state = %s \
  54                                         AND mirrors.country_code = ANY(%s) \
  55                                         ORDER BY RANDOM() LIMIT 1", filename, "UP",
  56                                         countries.get_in_zone(zone))
  57
  58                         if mirror:
  59                                 return mirror
  60
  61                 # Get a random mirror that serves the file
  62                 return self._get_mirror("SELECT mirrors.* FROM mirror_files files \
  63                         LEFT JOIN mirrors ON files.mirror = mirrors.id \
  64                         WHERE files.filename = %s \
  65                                 AND mirrors.enabled IS TRUE AND mirrors.state = %s \
  66                                 ORDER BY RANDOM() LIMIT 1", filename, "UP")
  67
  68         def get_by_hostname(self, hostname):
  69                 return self._get_mirror("SELECT * FROM mirrors \
  70                         WHERE hostname = %s", hostname)
  71
  72         def get_by_countries(self):
  73                 mirrors = {}
  74
  75                 for m in self:
  76                         try:
  77                                 mirrors[m.country].append(m)
  78                         except KeyError:
  79                                 mirrors[m.country] = [m]
  80
  81                 return mirrors
  82
  83
  84 class Mirror(Object):
  85         def init(self, id, data=None):
  86                 self.id   = id
  87                 self.data = data
  88
  89         def __str__(self):
  90                 return self.hostname
  91
  92         def __repr__(self):
  93                 return "<%s %s>" % (self.__class__.__name__, self.url)
  94
  95         def __eq__(self, other):
  96                 if isinstance(other, self.__class__):
  97                         return self.id == other.id
  98
  99         def __lt__(self, other):
 100                 if isinstance(other, self.__class__):
 101                         return self.hostname < other.hostname
 102
 103         @lazy_property
 104         def url(self):
 105                 url = "%s://%s" % ("https" if self.supports_https else "http", self.hostname)
 106
 107                 if not self.path.startswith("/"):
 108                         url += "/"
 109
 110                 url += "%s" % self.path
 111
 112                 if not self.path.endswith("/"):
 113                         url += "/"
 114
 115                 return url
 116
 117         @property
 118         def hostname(self):
 119                 return self.data.hostname
 120
 121         @property
 122         def path(self):
 123                 return self.data.path
 124
 125         @property
 126         def supports_https(self):
 127                 return self.data.supports_https
 128
 129         @property
 130         def address(self):
 131                 for addr in self.addresses4:
 132                         return addr
 133
 134                 for addr in self.addresses6:
 135                         return addr
 136
 137         @property
 138         def owner(self):
 139                 return self.data.owner
 140
 141         @lazy_property
 142         def location(self):
 143                 return self.geoip.get_location(self.address)
 144
 145         @property
 146         def latitude(self):
 147                 if self.location:
 148                         return self.location.latitude
 149
 150         @property
 151         def longitude(self):
 152                 if self.location:
 153                         return self.location.longitude
 154
 155         @lazy_property
 156         def country(self):
 157                 return iso3166.countries.get(self.country_code)
 158
 159         @property
 160         def country_code(self):
 161                 return self.data.country_code
 162
 163         @property
 164         def country_name(self):
 165                 return self.geoip.get_country_name(self.country_code)
 166
 167         @property
 168         def zone(self):
 169                 return countries.get_zone(self.country_name)
 170
 171         @lazy_property
 172         def asn(self):
 173                 return self.geoip.get_asn(self.address)
 174
 175         @property
 176         def filelist(self):
 177                 filelist = self.db.query("SELECT filename FROM mirror_files WHERE mirror=%s ORDER BY filename", self.id)
 178                 return [f.filename for f in filelist]
 179
 180         @property
 181         def prefix(self):
 182                 return ""
 183
 184         def build_url(self, filename):
 185                 return urllib.parse.urljoin(self.url, filename)
 186
 187         @property
 188         def last_update(self):
 189                 return self.data.last_update
 190
 191         @property
 192         def state(self):
 193                 return self.data.state
 194
 195         def set_state(self, state):
 196                 logging.debug("Setting state of %s to %s" % (self.hostname, state))
 197
 198                 if self.state == state:
 199                         return
 200
 201                 self.db.execute("UPDATE mirrors SET state = %s WHERE id = %s", state, self.id)
 202
 203                 # Reload changed settings
 204                 self.data["state"] = state
 205
 206         @property
 207         def enabled(self):
 208                 return self.data.enabled
 209
 210         @property
 211         def disabled(self):
 212                 return not self.enabled
 213
 214         async def check(self):
 215                 logging.debug("Running check for mirror %s" % self.hostname)
 216
 217                 self.db.execute("UPDATE mirrors SET address = %s WHERE id = %s",
 218                         self.address, self.id)
 219
 220                 success = await self.check_timestamp()
 221                 if success:
 222                         await self.check_filelist()
 223
 224         def check_state(self, last_update):
 225                 logging.debug("Checking state of mirror %s" % self.id)
 226
 227                 if not self.enabled:
 228                         self.set_state("DOWN")
 229                         return
 230
 231                 now = datetime.datetime.utcnow()
 232
 233                 time_delta = now - last_update
 234                 time_diff = time_delta.total_seconds()
 235
 236                 time_down = self.settings.get_int("mirrors_time_down", 3*24*60*60)
 237                 if time_diff >= time_down:
 238                         self.set_state("DOWN")
 239                         return
 240
 241                 time_outofsync = self.settings.get_int("mirrors_time_outofsync", 6*60*60)
 242                 if time_diff >= time_outofsync:
 243                         self.set_state("OUTOFSYNC")
 244                         return
 245
 246                 self.set_state("UP")
 247
 248         async def check_timestamp(self):
 249                 http = tornado.httpclient.AsyncHTTPClient()
 250
 251                 try:
 252                         response = await http.fetch(self.url + ".timestamp",
 253                                 headers={ "Pragma" : "no-cache" })
 254                 except tornado.httpclient.HTTPError as e:
 255                         logging.warning("Error getting timestamp from %s: %s" % (self.hostname, e))
 256                         self.set_state("DOWN")
 257                         return False
 258
 259                 except ssl.SSLError as e:
 260                         logging.warning("SSL error when getting timestamp from %s: %s" % (self.hostname, e))
 261                         self.set_state("DOWN")
 262                         return False
 263
 264                 except tornado.iostream.StreamClosedError as e:
 265                         logging.warning("Connection closed unexpectedly for %s: %s" % (self.hostname, e))
 266                         self.set_state("DOWN")
 267                         return False
 268
 269                 except OSError as e:
 270                         logging.warning("Could not connect to %s: %s" % (self.hostname, e))
 271                         self.set_state("DOWN")
 272                         return False
 273
 274                 if response.error:
 275                         logging.warning("Error getting timestamp from %s" % self.hostname)
 276                         self.set_state("DOWN")
 277                         return
 278
 279                 try:
 280                         timestamp = int(response.body.strip())
 281                 except ValueError:
 282                         timestamp = 0
 283
 284                 timestamp = datetime.datetime.utcfromtimestamp(timestamp)
 285
 286                 self.db.execute("UPDATE mirrors SET last_update = %s WHERE id = %s",
 287                         timestamp, self.id)
 288
 289                 # Update state
 290                 self.check_state(timestamp)
 291
 292                 logging.debug("Successfully updated timestamp from %s" % self.hostname)
 293
 294                 return True
 295
 296         async def check_filelist(self):
 297                 # XXX need to remove data from disabled mirrors
 298                 if not self.enabled:
 299                         return
 300
 301                 http = tornado.httpclient.AsyncHTTPClient()
 302
 303                 try:
 304                         response = await http.fetch(self.url + ".filelist",
 305                                 headers={ "Pragma" : "no-cache" })
 306                 except tornado.httpclient.HTTPError as e:
 307                         logging.warning("Error getting filelist from %s: %s" % (self.hostname, e))
 308                         self.set_state("DOWN")
 309                         return
 310
 311                 if response.error:
 312                         logging.debug("Error getting filelist from %s" % self.hostname)
 313                         return
 314
 315                 # Drop the old filelist
 316                 self.db.execute("DELETE FROM mirror_files WHERE mirror = %s", self.id)
 317
 318                 # Add them all again
 319                 for file in response.body.splitlines():
 320                         file = os.path.join(self.prefix, file.decode())
 321
 322                         self.db.execute("INSERT INTO mirror_files(mirror, filename) \
 323                                 VALUES(%s, %s)", self.id, file)
 324
 325                 logging.debug("Successfully updated mirror filelist from %s" % self.hostname)
 326
 327         @property
 328         def development(self):
 329                 return self.data.get("mirrorlist_devel", False)
 330
 331         @property
 332         def mirrorlist(self):
 333                 return self.data.get("mirrorlist", False)
 334
 335         @lazy_property
 336         def addresses(self):
 337                 addrinfo = socket.getaddrinfo(self.hostname, 0, socket.AF_UNSPEC, socket.SOCK_STREAM)
 338
 339                 ret = []
 340                 for family, socktype, proto, canonname, address in addrinfo:
 341                         if family == socket.AF_INET:
 342                                 address, port = address
 343                         elif family == socket.AF_INET6:
 344                                 address, port, flowid, scopeid = address
 345                         ret.append((family, address))
 346
 347                 return ret
 348
 349         @property
 350         def addresses6(self):
 351                 return [address for family, address in self.addresses if family == socket.AF_INET6]
 352
 353         @property
 354         def addresses4(self):
 355                 return [address for family, address in self.addresses if family == socket.AF_INET]