webapp/backend/mirrors.py

   1 #!/usr/bin/python
   2
   3 import logging
   4 import math
   5 import os.path
   6 import random
   7 import socket
   8 import time
   9 import tornado.httpclient
  10
  11 from databases import Databases
  12 from geoip import GeoIP
  13 from memcached import Memcached
  14 from misc import Singleton
  15
  16 class Downloads(object):
  17         __metaclass__ = Singleton
  18
  19         @property
  20         def db(self):
  21                 return Databases().webapp
  22
  23         @property
  24         def mirrors(self):
  25                 return Mirrors()
  26
  27         @property
  28         def total(self):
  29                 ret = self.db.get("SELECT COUNT(*) AS total FROM log_download")
  30
  31                 return ret.total
  32
  33         @property
  34         def today(self):
  35                 ret = self.db.get("SELECT COUNT(*) AS today FROM log_download WHERE date >= NOW() - 1000000")
  36
  37                 return ret.today
  38
  39         @property
  40         def yesterday(self):
  41                 ret = self.db.get("SELECT COUNT(*) AS yesterday FROM log_download WHERE DATE(date) = DATE(NOW())-1")
  42
  43                 return ret.yesterday
  44
  45         @property
  46         def daily_map(self):
  47                 ret = self.db.query("SELECT DATE(date) AS date, COUNT(*) AS downloads FROM log_download"
  48                         " WHERE DATE(date) BETWEEN DATE(NOW()) - 31 AND DATE(NOW()) GROUP BY DATE(date)")
  49
  50                 return ret
  51
  52         def get_countries(self, duration="all"):
  53                 query = "SELECT country_code, count(country_code) AS count FROM log_download"
  54
  55                 if duration == "today":
  56                         query += " WHERE date >= NOW() - 1000000"
  57
  58                 query += " GROUP BY country_code ORDER BY count DESC"
  59
  60                 results = self.db.query(query)
  61                 ret = {}
  62
  63                 count = sum([o.count for o in results])
  64                 for res in results:
  65                         ret[res.country_code] = float(res.count) / count
  66
  67                 return ret
  68
  69         def get_mirror_load(self, duration="all"):
  70                 query = "SELECT mirror, COUNT(mirror) AS count FROM log_download"
  71
  72                 if duration == "today":
  73                         query += " WHERE date >= NOW() - 1000000"
  74
  75                 query += " GROUP BY mirror ORDER BY count DESC"
  76
  77                 results = self.db.query(query)
  78                 ret = {}
  79
  80                 count = sum([o.count for o in results])
  81                 for res in results:
  82                         mirror = self.mirrors.get(res.mirror)
  83                         ret[mirror.hostname] = float(res.count) / count
  84
  85                 return ret
  86
  87
  88 class Mirrors(object):
  89         __metaclass__ = Singleton
  90
  91         @property
  92         def db(self):
  93                 return Databases().webapp
  94
  95         @property
  96         def memcached(self):
  97                 return Memcached()
  98
  99         def list(self):
 100                 return [Mirror(m.id) for m in self.db.query("SELECT id FROM mirrors WHERE disabled = 'N' ORDER BY state,hostname")]
 101
 102         def check_all(self):
 103                 for mirror in self.list():
 104                         mirror.check()
 105
 106         def get(self, id):
 107                 return Mirror(id)
 108
 109         def get_all(self):
 110                 return MirrorSet(self.list())
 111
 112         def get_by_hostname(self, hostname):
 113                 mirror = self.db.get("SELECT id FROM mirrors WHERE hostname=%s", hostname)
 114
 115                 return Mirror(mirror.id)
 116
 117         def get_with_file(self, filename, country=None):
 118                 # XXX quick and dirty solution - needs a performance boost
 119                 mirror_ids = [m.mirror for m in self.db.query("SELECT mirror FROM mirror_files WHERE filename=%s", filename)]
 120
 121                 #if country:
 122                 #       # Sort out all mirrors that are not preferred to the given country
 123                 #       for mirror in self.get_for_country(country):
 124                 #               if not mirror.id in mirror_ids:
 125                 #                       mirror_ids.remove(mirror.id)
 126
 127                 mirrors = []
 128                 for mirror_id in mirror_ids:
 129                         mirror = self.get(mirror_id)
 130                         if not mirror.state == "UP":
 131                                 continue
 132                         mirrors.append(mirror)
 133
 134                 logging.debug("%s" % mirrors)
 135
 136                 return mirrors
 137
 138         def get_for_country(self, country):
 139                 # XXX need option for random order
 140                 mirrors = self.db.query("SELECT id FROM mirrors WHERE prefer_for_countries LIKE %s", country)
 141
 142                 for mirror in mirrors:
 143                         yield self.get(mirror.id)
 144
 145         def get_for_location(self, addr):
 146                 distance = 10
 147
 148                 mirrors = []
 149                 all_mirrors = self.list()
 150
 151                 location = GeoIP().get_all(addr)
 152                 if not location:
 153                         return None
 154
 155                 while all_mirrors and len(mirrors) <= 2 and distance <= 270:
 156                         for mirror in all_mirrors:
 157                                 if mirror.distance_to(location) <= distance:
 158                                         mirrors.append(mirror)
 159                                         all_mirrors.remove(mirror)
 160
 161                         distance *= 1.2
 162
 163                 return mirrors
 164
 165         def get_all_files(self):
 166                 files = []
 167
 168                 for mirror in self.list():
 169                         if not mirror.state == "UP":
 170                                 continue
 171
 172                         for file in mirror.filelist:
 173                                 if not file in files:
 174                                         files.append(file)
 175
 176                 return files
 177
 178
 179 class MirrorSet(object):
 180         def __init__(self, mirrors):
 181                 self._mirrors = mirrors
 182
 183         def __add__(self, other):
 184                 mirrors = []
 185
 186                 for mirror in self._mirrors + other._mirrors:
 187                         if mirror in mirrors:
 188                                 continue
 189
 190                         mirrors.append(mirror)
 191
 192                 return MirrorSet(mirrors)
 193
 194         def __sub__(self, other):
 195                 mirrors = self._mirrors[:]
 196
 197                 for mirror in other._mirrors:
 198                         if mirror in mirrors:
 199                                 mirrors.remove(mirror)
 200
 201                 return MirrorSet(mirrors)
 202
 203         def __iter__(self):
 204                 return iter(self._mirrors)
 205
 206         def __len__(self):
 207                 return len(self._mirrors)
 208
 209         def __str__(self):
 210                 return "<MirrorSet %s>" % ", ".join([m.hostname for m in self._mirrors])
 211
 212         @property
 213         def db(self):
 214                 return Mirrors().db
 215
 216         def get_with_file(self, filename):
 217                 with_file = [m.mirror for m in self.db.query("SELECT mirror FROM mirror_files WHERE filename=%s", filename)]
 218
 219                 mirrors = []
 220                 for mirror in self._mirrors:
 221                         if mirror.id in with_file:
 222                                 mirrors.append(mirror)
 223
 224                 return MirrorSet(mirrors)
 225
 226         def get_random(self):
 227                 mirrors = []
 228                 for mirror in self._mirrors:
 229                         for i in range(0, mirror.priority):
 230                                 mirrors.append(mirror)
 231
 232                 return random.choice(mirrors)
 233
 234         def get_for_country(self, country):
 235                 mirrors = []
 236
 237                 for mirror in self._mirrors:
 238                         if country in mirror.prefer_for_countries:
 239                                 mirrors.append(mirror)
 240
 241                 return MirrorSet(mirrors)
 242
 243         def get_for_location(self, addr):
 244                 distance = 10
 245
 246                 mirrors = []
 247
 248                 location = GeoIP().get_all(addr)
 249                 if location:
 250                         while len(mirrors) <= 2 and distance <= 270:
 251                                 for mirror in self._mirrors:
 252                                         if mirror in mirrors:
 253                                                 continue
 254
 255                                         if mirror.distance_to(location) <= distance:
 256                                                 mirrors.append(mirror)
 257
 258                                 distance *= 1.2
 259
 260                 return MirrorSet(mirrors)
 261
 262         def get_with_state(self, state):
 263                 mirrors = []
 264
 265                 for mirror in self._mirrors:
 266                         if mirror.state == state:
 267                                 mirrors.append(mirror)
 268
 269                 return MirrorSet(mirrors)
 270
 271
 272 class Mirror(object):
 273         def __init__(self, id):
 274                 self.id = id
 275
 276                 self.reload()
 277
 278                 self.__location = None
 279                 self.__country_name = None
 280
 281         def __repr__(self):
 282                 return "<%s %s>" % (self.__class__.__name__, self.url)
 283
 284         def __cmp__(self, other):
 285                 return cmp(self.id, other.id)
 286
 287         @property
 288         def db(self):
 289                 return Databases().webapp
 290
 291         def reload(self, force=False):
 292                 memcached = Memcached()
 293                 mem_id = "mirror-%s" % self.id
 294
 295                 if force:
 296                         memcached.delete(mem_id)
 297
 298                 self._info = memcached.get(mem_id)
 299                 if not self._info:
 300                         self._info = self.db.get("SELECT * FROM mirrors WHERE id=%s", self.id)
 301                         self._info["url"] = self.generate_url()
 302
 303                         memcached.set(mem_id, self._info, 60)
 304
 305         def generate_url(self):
 306                 url = "http://%s" % self.hostname
 307                 if not self.path.startswith("/"):
 308                         url += "/"
 309                 url += "%s" % self.path
 310                 if not self.path.endswith("/"):
 311                         url += "/"
 312                 return url
 313
 314         def __getattr__(self, key):
 315                 try:
 316                         return self._info[key]
 317                 except KeyError:
 318                         raise AttributeError(key)
 319
 320         @property
 321         def address(self):
 322                 return socket.gethostbyname(self.hostname)
 323
 324         @property
 325         def location(self):
 326                 if self.__location is None:
 327                         self.__location = GeoIP().get_all(self.address)
 328
 329                 return self.__location
 330
 331         @property
 332         def latitude(self):
 333                 return self.location.latitude
 334
 335         @property
 336         def longitude(self):
 337                 return self.location.longitude
 338
 339         @property
 340         def coordinates(self):
 341                 return (self.latitude, self.longitude)
 342
 343         @property
 344         def coordiante_str(self):
 345                 coordinates = []
 346
 347                 for i in self.coordinates:
 348                         coordinates.append("%s" % i)
 349
 350                 return ",".join(coordinates)
 351
 352         @property
 353         def country_code(self):
 354                 return self.location.country_code.lower() or "unknown"
 355
 356         @property
 357         def country_name(self):
 358                 if self.__country_name is None:
 359                         self.__country_name = GeoIP().get_country_name(self.country_code)
 360
 361                 return self.__country_name
 362
 363         @property
 364         def city(self):
 365                 if self._info["city"]:
 366                         return self._info["city"]
 367
 368                 return self.location.city
 369
 370         @property
 371         def location_str(self):
 372                 s = self.country_name
 373                 if self.city:
 374                         s = "%s, %s" % (self.city, s)
 375
 376                 return s
 377
 378         @property
 379         def filelist(self):
 380                 filelist = self.db.query("SELECT filename FROM mirror_files WHERE mirror=%s ORDER BY filename", self.id)
 381                 return [f.filename for f in filelist]
 382
 383         @property
 384         def prefix(self):
 385                 if self.type.startswith("pakfire"):
 386                         return self.type
 387
 388                 return ""
 389
 390         def set_state(self, state):
 391                 logging.info("Setting state of %s to %s" % (self.hostname, state))
 392
 393                 if self.state == state:
 394                         return
 395
 396                 self.db.execute("UPDATE mirrors SET state=%s WHERE id=%s",
 397                         state, self.id)
 398
 399                 # Reload changed settings
 400                 self.reload(force=True)
 401
 402         def check(self):
 403                 logging.info("Running check for mirror %s" % self.hostname)
 404
 405                 self.check_timestamp()
 406                 self.check_filelist()
 407
 408         def check_state(self):
 409                 logging.debug("Checking state of mirror %s" % self.id)
 410
 411                 if self.disabled == "Y":
 412                         self.set_state("DOWN")
 413
 414                 time_diff = time.time() - self.last_update
 415                 if time_diff > 3*24*60*60: # XXX get this into Settings
 416                         self.set_state("DOWN")
 417                 elif time_diff > 6*60*60:
 418                         self.set_state("OUTOFSYNC")
 419                 else:
 420                         self.set_state("UP")
 421
 422         def check_timestamp(self):
 423                 if self.releases == "N":
 424                         return
 425
 426                 http = tornado.httpclient.AsyncHTTPClient()
 427
 428                 http.fetch(self.url + ".timestamp",
 429                         headers={ "Pragma" : "no-cache" },
 430                         callback=self.__check_timestamp_response)
 431
 432         def __check_timestamp_response(self, response):
 433                 if response.error:
 434                         logging.debug("Error getting timestamp from %s" % self.hostname)
 435                         self.set_state("DOWN")
 436                         return
 437
 438                 try:
 439                         timestamp = int(response.body.strip())
 440                 except ValueError:
 441                         timestamp = 0
 442
 443                 self.db.execute("UPDATE mirrors SET last_update=%s WHERE id=%s",
 444                         timestamp, self.id)
 445
 446                 # Reload changed settings
 447                 self.reload(force=True)
 448
 449                 self.check_state()
 450
 451                 logging.info("Successfully updated timestamp from %s" % self.hostname)
 452
 453         def check_filelist(self):
 454                 # XXX need to remove data from disabled mirrors
 455                 if self.releases == "N" or self.disabled == "Y" or self.type != "full":
 456                         return
 457
 458                 http = tornado.httpclient.AsyncHTTPClient()
 459
 460                 http.fetch(self.url + ".filelist",
 461                         headers={ "Pragma" : "no-cache" },
 462                         callback=self.__check_filelist_response)
 463
 464         def __check_filelist_response(self, response):
 465                 if response.error:
 466                         logging.debug("Error getting timestamp from %s" % self.hostname)
 467                         return
 468
 469                 files = self.filelist
 470
 471                 for file in response.body.splitlines():
 472                         file = os.path.join(self.prefix, file)
 473
 474                         if file in files:
 475                                 files.remove(file)
 476                                 continue
 477
 478                         self.db.execute("INSERT INTO mirror_files(mirror, filename) VALUES(%s, %s)",
 479                                 self.id, file)
 480
 481                 for file in files:
 482                         self.db.execute("DELETE FROM mirror_files WHERE mirror=%s AND filename=%s",
 483                                 self.id, file)
 484
 485                 logging.info("Successfully updated mirror filelist from %s" % self.hostname)
 486
 487         @property
 488         def prefer_for_countries(self):
 489                 countries = self._info.get("prefer_for_countries", "")
 490                 if countries:
 491                         return sorted(countries.split(", "))
 492
 493                 return []
 494
 495         @property
 496         def prefer_for_countries_names(self):
 497                 return sorted([GeoIP().get_country_name(c) for c in self.prefer_for_countries])
 498
 499         def distance_to(self, location, ignore_preference=False):
 500                 if not location:
 501                         return 0
 502
 503                 if not ignore_preference and location.country_code.lower() in self.prefer_for_countries:
 504                         return 0
 505
 506                 distance_vector = (
 507                         self.latitude - location.latitude,
 508                         self.longitude - location.longitude
 509                 )
 510
 511                 distance = 0
 512                 for i in distance_vector:
 513                         distance += i**2
 514
 515                 return math.sqrt(distance)
 516
 517         def traffic(self, since):
 518                 # XXX needs to be done better
 519
 520                 files = {}
 521                 for entry in self.db.query("SELECT filename, filesize FROM files"):
 522                         files[entry.filename] = entry.filesize
 523
 524                 query = "SELECT COUNT(filename) as count, filename FROM log_download WHERE mirror = %s"
 525                 query += " AND date >= %s GROUP BY filename"
 526
 527                 traffic = 0
 528                 for entry in self.db.query(query, self.id, since):
 529                         if files.has_key(entry.filename):
 530                                 traffic += entry.count * files[entry.filename]
 531
 532                 return traffic
 533
 534         @property
 535         def priority(self):
 536                 return self._info.get("priority", 10)
 537