src/backend/mirrors.py

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