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