[people/shoehn/ipfire.org.git] / www / webapp / backend / mirrors.py

#!/usr/bin/python

import logging
import math
import os.path
import random
import socket
import time
import tornado.httpclient

from databases import Databases
from geoip import GeoIP
from memcached import Memcached
from misc import Singleton

class Mirrors(object):
	__metaclass__ = Singleton

	@property
	def db(self):
		return Databases().webapp

	@property
	def memcached(self):
		return Memcached()

	def list(self):
		return [Mirror(m.id) for m in self.db.query("SELECT id FROM mirrors ORDER BY state")]

	def check_all(self):
		for mirror in self.list():
			mirror.check()

	def get(self, id):
		return Mirror(id)

	def get_all(self):
		return MirrorSet(self.list())

	def get_by_hostname(self, hostname):
		mirror = self.db.get("SELECT id FROM mirrors WHERE hostname=%s", hostname)

		return Mirror(mirror.id)

	def get_with_file(self, filename, country=None):
		# XXX quick and dirty solution - needs a performance boost
		mirror_ids = [m.mirror for m in self.db.query("SELECT mirror FROM mirror_files WHERE filename=%s", filename)]

		#if country:
		#	# Sort out all mirrors that are not preferred to the given country
		#	for mirror in self.get_for_country(country):
		#		if not mirror.id in mirror_ids:
		#			mirror_ids.remove(mirror.id)

		mirrors = []
		for mirror_id in mirror_ids:
			mirror = self.get(mirror_id)
			if not mirror.state == "UP":
				continue
			mirrors.append(mirror)

		logging.debug("%s" % mirrors)

		return mirrors

	def get_for_country(self, country):
		# XXX need option for random order
		mirrors = self.db.query("SELECT id FROM mirrors WHERE prefer_for_countries LIKE %s", country)

		for mirror in mirrors:
			yield self.get(mirror.id)

	def get_for_location(self, addr):
		distance = 10

		mirrors = []
		all_mirrors = self.list()

		while all_mirrors and len(mirrors) <= 2 and distance <= 270:
			for mirror in all_mirrors:
				if mirror.distance_to(addr) <= distance:
					mirrors.append(mirror)
					all_mirrors.remove(mirror)

			distance *= 1.2

		return mirrors

	def get_all_files(self):
		files = []

		for mirror in self.list():
			if not mirror.state == "UP":
				continue

			for file in mirror.filelist:
				if not file in files:
					files.append(file)

		return files


class MirrorSet(object):
	def __init__(self, mirrors):
		self._mirrors = mirrors

	def __add__(self, other):
		mirrors = []

		for mirror in self._mirrors + other._mirrors:
			if mirror in mirrors:
				continue

			mirrors.append(mirror)

		return MirrorSet(mirrors)

	def __sub__(self, other):
		mirrors = self._mirrors[:]

		for mirror in other._mirrors:
			if mirror in mirrors:
				mirrors.remove(mirror)

		return MirrorSet(mirrors)

	def __iter__(self):
		return iter(self._mirrors)

	def __len__(self):
		return len(self._mirrors)

	def __str__(self):
		return "<MirrorSet %s>" % ", ".join([m.hostname for m in self._mirrors])

	@property
	def db(self):
		return Mirrors().db

	def get_with_file(self, filename):
		with_file = [m.mirror for m in self.db.query("SELECT mirror FROM mirror_files WHERE filename=%s", filename)]

		mirrors = []
		for mirror in self._mirrors:
			if mirror.id in with_file:
				mirrors.append(mirror)

		return MirrorSet(mirrors)

	def get_random(self):
		mirrors = []
		for mirror in self._mirrors:
			for i in range(0, mirror.priority):
				mirrors.append(mirror)

		return random.choice(mirrors)

	def get_for_country(self, country):
		mirrors = []

		for mirror in self._mirrors:
			if country in mirror.prefer_for_countries:
				mirrors.append(mirror)

		return MirrorSet(mirrors)

	def get_for_location(self, addr):
		distance = 10

		mirrors = []

		while len(mirrors) <= 2 and distance <= 270:
			for mirror in self._mirrors:
				if mirror in mirrors:
					continue

				if mirror.distance_to(addr) <= distance:
					mirrors.append(mirror)

			distance *= 1.2

		return MirrorSet(mirrors)

	def get_with_state(self, state):
		mirrors = []

		for mirror in self._mirrors:
			if mirror.state == state:
				mirrors.append(mirror)

		return MirrorSet(mirrors)


class Mirror(object):
	def __init__(self, id):
		self.id = id

		self.reload()

	def __repr__(self):
		return "<%s %s>" % (self.__class__.__name__, self.url)

	def __cmp__(self, other):
		return cmp(self.id, other.id)

	@property
	def db(self):
		return Databases().webapp

	def reload(self, force=False):
		memcached = Memcached()
		mem_id = "mirror-%s" % self.id

		if force:
			memcached.delete(mem_id)

		self._info = memcached.get(mem_id)
		if not self._info:
			self._info = self.db.get("SELECT * FROM mirrors WHERE id=%s", self.id)
			self._info["url"] = self.generate_url()

			memcached.set(mem_id, self._info, 60)

	def generate_url(self):
		url = "http://%s" % self.hostname
		if not self.path.startswith("/"):
			url += "/"
		url += "%s" % self.path
		if not self.path.endswith("/"):
			url += "/"
		return url

	def __getattr__(self, key):
		try:
			return self._info[key]
		except KeyError:
			raise AttributeError(key)

	@property
	def address(self):
		return socket.gethostbyname(self.hostname)

	@property
	def location(self):
		if not hasattr(self, "__location"):
			self.__location = GeoIP().get_all(self.address)

		return self.__location

	@property
	def latitude(self):
		return self.location.latitude

	@property
	def longitude(self):
		return self.location.longitude

	@property
	def coordinates(self):
		return (self.latitude, self.longitude)

	@property
	def coordiante_str(self):
		coordinates = []

		for i in self.coordinates:
			coordinates.append("%s" % i)

		return ",".join(coordinates)

	@property
	def country_code(self):
		return GeoIP().get_country(self.address).lower() or "unknown"

	@property
	def country_name(self):
		return GeoIP().get_country_name(self.country_code)

	@property
	def city(self):
		if self._info["city"]:
			return self._info["city"]

		return self.location.city

	@property
	def location_str(self):
		s = self.country_name
		if self.city:
			s = "%s, %s" % (self.city, s)

		return s

	@property
	def filelist(self):
		filelist = self.db.query("SELECT filename FROM mirror_files WHERE mirror=%s ORDER BY filename", self.id)
		return [f.filename for f in filelist]

	@property
	def prefix(self):
		if self.type.startswith("pakfire"):
			return self.type

		return ""

	def set_state(self, state):
		logging.info("Setting state of %s to %s" % (self.hostname, state))

		if self.state == state:
			return

		self.db.execute("UPDATE mirrors SET state=%s WHERE id=%s",
			state, self.id)

		# Reload changed settings
		self.reload(force=True)

	def check(self):
		logging.info("Running check for mirror %s" % self.hostname)

		self.check_timestamp()
		self.check_filelist()

	def check_state(self):
		logging.debug("Checking state of mirror %s" % self.id)

		if self.disabled == "Y":
			self.set_state("DOWN")

		time_diff = time.time() - self.last_update
		if time_diff > 3*24*60*60: # XXX get this into Settings
			self.set_state("DOWN")
		elif time_diff > 6*60*60:
			self.set_state("OUTOFSYNC")
		else:
			self.set_state("UP")

	def check_timestamp(self):
		if self.releases == "N":
			return

		http = tornado.httpclient.AsyncHTTPClient()

		http.fetch(self.url + ".timestamp",
			headers={ "Pragma" : "no-cache" },
			callback=self.__check_timestamp_response)

	def __check_timestamp_response(self, response):
		if response.error:
			logging.debug("Error getting timestamp from %s" % self.hostname)
			return

		try:
			timestamp = int(response.body.strip())
		except ValueError:
			timestamp = 0

		self.db.execute("UPDATE mirrors SET last_update=%s WHERE id=%s",
			timestamp, self.id)

		# Reload changed settings
		self.reload(force=True)

		self.check_state()

		logging.info("Successfully updated timestamp from %s" % self.hostname)

	def check_filelist(self):
		# XXX need to remove data from disabled mirrors
		if self.releases == "N" or self.disabled == "Y" or self.type != "full":
			return

		http = tornado.httpclient.AsyncHTTPClient()

		http.fetch(self.url + ".filelist",
			headers={ "Pragma" : "no-cache" },
			callback=self.__check_filelist_response)

	def __check_filelist_response(self, response):
		if response.error:
			logging.debug("Error getting timestamp from %s" % self.hostname)
			return

		files = self.filelist

		for file in response.body.splitlines():
			file = os.path.join(self.prefix, file)

			if file in files:
				files.remove(file)
				continue

			self.db.execute("INSERT INTO mirror_files(mirror, filename) VALUES(%s, %s)",
				self.id, file)

		for file in files:
			self.db.execute("DELETE FROM mirror_files WHERE mirror=%s AND filename=%s",
				self.id, file)

		logging.info("Successfully updated mirror filelist from %s" % self.hostname)

	@property
	def prefer_for_countries(self):
		countries = self._info.get("prefer_for_countries", "")
		if countries:
			return sorted(countries.split(", "))

		return []

	@property
	def prefer_for_countries_names(self):
		return sorted([GeoIP().get_country_name(c) for c in self.prefer_for_countries])

	def distance_to(self, addr):
		location = GeoIP().get_all(addr)
		if not location:
			return 0

		if location.country_code.lower() in self.prefer_for_countries:
			return 0

		distance_vector = (
			self.latitude - location.latitude,
			self.longitude - location.longitude
		)

		distance = 0
		for i in distance_vector:
			distance += i**2

		return math.sqrt(distance)

	def traffic(self, since):
		# XXX needs to be done better

		files = {}
		for entry in self.db.query("SELECT filename, filesize FROM files"):
			files[entry.filename] = entry.filesize

		query = "SELECT COUNT(filename) as count, filename FROM log_download WHERE mirror = %s"
		query += " AND date >= %s GROUP BY filename"

		traffic = 0
		for entry in self.db.query(query, self.id, since):
			if files.has_key(entry.filename):
				traffic += entry.count * files[entry.filename]

		return traffic

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