Sounds good to me. If it breaks anything, we can always just reinstall the mirrormanager2-mirrorlist package to return to the current version. With kind regards, Patrick Uiterwijk Fedora Infra ----- Original Message ----- > As there are more reports of timeouts connected to the mirrorlist > servers I had a closer look at the code. It seems we are hitting the > default value of max_children = 40. This freeze break request tries to > increase this to 80 via a hotfix. See commit message below for further > details. > > Adrian > > > commit 2b8e284e687d097c2cad6cd24dc3db3aa23f837d > Author: Adrian Reber <adrian@xxxxxxxx> > Date: Thu Oct 15 14:03:02 2015 +0000 > > Increase the number of possible child processes > > The mirrorlist-server is the process which has the mirrorlist data > loaded and which is accessed by the public facing > mirrorlist_client.wsgi. The mirrorlist-server uses the > ForkingUnixStreamServer which has a default of max_children = 40. > (https://hg.python.org/cpython/file/2.7/Lib/SocketServer.py#l516) > > Looking at the code of ForkingUnixStreamServer it says at > > https://hg.python.org/cpython/file/2.7/Lib/SocketServer.py#l523 > > # If we're above the max number of children, wait and reap them until > # we go back below threshold. Note that we use waitpid(-1) below to be > # able to collect children in size(<defunct children>) syscalls instead > # of size(<children>): the downside is that this might reap children > # which we didn't spawn, which is why we only resort to this when we're > # above max_children. > > As we are running the wsgi with processes=45 this sounds like it can > lead to situation where it might just hang. > > This increases max_children to 80 and maybe processes could be upped to > 60 or 70. > > Signed-off-by: Adrian Reber <adrian@xxxxxxxx> > > diff --git a/files/hotfix/mirrorlist/mirrorlist_server.py > b/files/hotfix/mirrorlist/mirrorlist_server.py > index 2d98fa0..2b81912 100644 > --- a/files/hotfix/mirrorlist/mirrorlist_server.py > +++ b/files/hotfix/mirrorlist/mirrorlist_server.py > @@ -938,6 +938,7 @@ def sigterm_handler(signum, frame): > > class ForkingUnixStreamServer(ForkingMixIn, UnixStreamServer): > request_queue_size = 300 > + max_children = 80 > def finish_request(self, request, client_address): > signal.signal(signal.SIGHUP, signal.SIG_IGN) > BaseServer.finish_request(self, request, client_address) > diff --git a/roles/mirrormanager/mirrorlist2/tasks/main.yml > b/roles/mirrormanager/mirrorlist2/tasks/main.yml > index 786432f..2e5f5ee 100644 > --- a/roles/mirrormanager/mirrorlist2/tasks/main.yml > +++ b/roles/mirrormanager/mirrorlist2/tasks/main.yml > @@ -73,3 +73,12 @@ > # tags: > # - mirrorlist2 > # - selinux > + > + > +- name: HOTFIX - increase the number of possible child processes > + copy: src="{{ files }}/hotfix/mirrorlist/mirrorlist_server.py" > dest=/usr/share/mirrormanager2/mirrorlist_server.py > + owner=root group=root mode=0755 > + tags: > + - hotfix > + notify: > + - restart mirrorlist-server > > commit d04588274a6e161a36d8dbd9c91ea0fe78de017e > Author: Adrian Reber <adrian@xxxxxxxx> > Date: Thu Oct 15 13:54:53 2015 +0000 > > Original mirrorlist_server.py which needs to be hotfixed > > Signed-off-by: Adrian Reber <adrian@xxxxxxxx> > > diff --git a/files/hotfix/mirrorlist/mirrorlist_server.py > b/files/hotfix/mirrorlist/mirrorlist_server.py > new file mode 100644 > index 0000000..2d98fa0 > --- /dev/null > +++ b/files/hotfix/mirrorlist/mirrorlist_server.py > @@ -0,0 +1,1136 @@ > +#!/usr/bin/python > +# > +# Copyright (c) 2007-2013 Dell, Inc. > +# by Matt Domsch <Matt_Domsch@xxxxxxxx> > +# Licensed under the MIT/X11 license > + > +# standard library modules in alphabetical order > +from collections import defaultdict > +import datetime > +import getopt > +import logging > +import logging.handlers > +import os > +import random > +import cPickle as pickle > +import select > +import signal > +import socket > +from SocketServer import (StreamRequestHandler, ForkingMixIn, > + UnixStreamServer, BaseServer) > +import sys > +from string import zfill, atoi > +import time > +import traceback > + > +try: > + import threading > +except ImportError: > + import dummy_threading as threading > + > +# not-so-standard library modules that this program needs > +from IPy import IP > +import GeoIP > +import radix > +from weighted_shuffle import weighted_shuffle > + > +# can be overridden on the command line > +pidfile = '/var/run/mirrormanager/mirrorlist_server.pid' > +socketfile = '/var/run/mirrormanager/mirrorlist_server.sock' > +cachefile = '/var/lib/mirrormanager/mirrorlist_cache.pkl' > +internet2_netblocks_file = '/var/lib/mirrormanager/i2_netblocks.txt' > +global_netblocks_file = '/var/lib/mirrormanager/global_netblocks.txt' > +logfile = None > +debug = False > +must_die = False > +# at a point in time when we're no longer serving content for versions > +# that don't use yum prioritymethod=fallback > +# (e.g. after Fedora 7 is past end-of-life) > +# then we can set this value to True > +# this only affects results requested using path=... > +# for dirs which aren't repositories (such as iso/) > +# because we don't know the Version associated with that dir here. > +default_ordered_mirrorlist = False > + > +gipv4 = None > +gipv6 = None > + > +# key is strings in tuple (repo.prefix, arch) > +mirrorlist_cache = {} > + > +# key is directory.name, returns keys for mirrorlist_cache > +directory_name_to_mirrorlist = {} > + > +# key is an IPy.IP structure, value is list of host ids > +host_netblock_cache = {} > + > +# key is hostid, value is list of countries to allow > +host_country_allowed_cache = {} > + > +repo_arch_to_directoryname = {} > + > +# redirect from a repo with one name to a repo with another > +repo_redirect = {} > +country_continent_redirect_cache = {} > + > +# our own private copy of country_continents to be edited > +country_continents = GeoIP.country_continents > + > +disabled_repositories = {} > +host_bandwidth_cache = {} > +host_country_cache = {} > +host_max_connections_cache = {} > +file_details_cache = {} > +hcurl_cache = {} > +asn_host_cache = {} > +internet2_tree = radix.Radix() > +global_tree = radix.Radix() > +host_netblocks_tree = radix.Radix() > +netblock_country_tree = radix.Radix() > +location_cache = {} > +netblock_country_cache = {} > + > +## Set up our syslog data. > +syslogger = logging.getLogger('mirrormanager') > +syslogger.setLevel(logging.INFO) > +handler = logging.handlers.SysLogHandler( > + address='/dev/log', > + facility=logging.handlers.SysLogHandler.LOG_LOCAL4) > +syslogger.addHandler(handler) > + > + > +def lookup_ip_asn(tree, ip): > + """ @t is a radix tree > + @ip is an IPy.IP object which may be contained in an entry in l > + """ > + node = tree.search_best(ip.strNormal()) > + if node is None: > + return None > + return node.data['asn'] > + > + > +def uniqueify(seq, idfun=None): > + # order preserving > + if idfun is None: > + def idfun(x): return x > + seen = {} > + result = [] > + for item in seq: > + marker = idfun(item) > + # in old Python versions: > + # if seen.has_key(marker) > + # but in new ones: > + if marker in seen: continue > + seen[marker] = 1 > + result.append(item) > + return result > + > + > +##### Metalink Support ##### > + > +def metalink_header(): > + # fixme add alternate format pubdate when specified > + pubdate = datetime.datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S > GMT") > + doc = '' > + doc += '<?xml version="1.0" encoding="utf-8"?>\n' > + doc += '<metalink version="3.0" xmlns="http://www.metalinker.org/";' > + doc += ' type="dynamic"' > + doc += ' pubdate="%s"' % pubdate > + doc += ' generator="mirrormanager"' > + doc += ' xmlns:mm0="http://fedorahosted.org/mirrormanager";' > + doc += '>\n' > + return doc > + > + > +def metalink_failuredoc(message=None): > + doc = metalink_header() > + if message is not None: > + doc += '<!--\n' > + doc += message + '\n' > + doc += '-->\n' > + doc += '</metalink>\n' > + return doc > + > + > +def metalink_file_not_found(directory, file): > + message = '%s/%s not found or has no metalink' % (directory, file) > + return metalink_failuredoc(message) > + > + > +def metalink(cache, directory, file, hosts_and_urls): > + preference = 100 > + try: > + fdc = file_details_cache[directory] > + detailslist = fdc[file] > + except KeyError: > + return ('metalink', 404, metalink_file_not_found(directory, file)) > + > + def indent(n): > + return ' ' * n * 2 > + > + doc = metalink_header() > + doc += indent(1) + '<files>\n' > + doc += indent(2) + '<file name="%s">\n' % (file) > + y = detailslist[0] > + > + def details(y, indentlevel=2): > + doc = '' > + if y['timestamp'] is not None: > + doc += indent(indentlevel+1) \ > + + '<mm0:timestamp>%s</mm0:timestamp>\n' % y['timestamp'] > + if y['size'] is not None: > + doc += indent(indentlevel+1) + '<size>%s</size>\n' % y['size'] > + doc += indent(indentlevel+1) + '<verification>\n' > + hashes = ('md5', 'sha1', 'sha256', 'sha512') > + for h in hashes: > + if y[h] is not None: > + doc += indent(indentlevel+2) \ > + + '<hash type="%s">%s</hash>\n' % (h, y[h]) > + doc += indent(indentlevel+1) + '</verification>\n' > + return doc > + > + doc += details(y, 2) > + # there can be multiple files > + if len(detailslist) > 1: > + doc += indent(3) + '<mm0:alternates>\n' > + for y in detailslist[1:]: > + doc += indent(4) + '<mm0:alternate>\n' > + doc += details(y,5) > + doc += indent(4) + '</mm0:alternate>\n' > + doc += indent(3) + '</mm0:alternates>\n' > + > + doc += indent(3) + '<resources maxconnections="1">\n' > + for (hostid, hcurls) in hosts_and_urls: > + private = '' > + if hostid not in cache['global']: > + private = 'mm0:private="True"' > + for url in hcurls: > + protocol = url.split(':')[0] > + # FIXME January 2010 > + # adding protocol= here is not part of the Metalink 3.0 spec, > + # but MirrorManager 1.2.6 used it accidentally, as did > + # yum 3.2.20-3 as released in Fedora 8, 9, and 10. After those > + # three are EOL (~January 2010), the extra protocol= can be > + # removed. > + doc += indent(4) + \ > + '<url protocol="%s" type="%s" location="%s" '\ > + 'preference="%s" %s>' % ( > + protocol, protocol, host_country_cache[hostid].upper(), > + preference, private) > + doc += url > + doc += '</url>\n' > + preference = max(preference-1, 1) > + doc += indent(3) + '</resources>\n' > + doc += indent(2) + '</file>\n' > + doc += indent(1) + '</files>\n' > + doc += '</metalink>\n' > + return ('metalink', 200, doc) > + > + > +def tree_lookup(tree, ip, field, maxResults=None): > + # fast lookup in the tree; if present, find all the matching values by > + # deleting the found one and searching again this is safe w/o copying > + # the tree again only because this is the only place the tree is used, > + # and we'll get a new copy of the tree from our parent the next time it > + # fork()s. > + # returns a list of tuples (prefix, data) > + result = [] > + len_data = 0 > + if ip is None: > + return result > + node = tree.search_best(ip.strNormal()) > + while node is not None: > + prefix = node.prefix > + if type(node.data[field]) == list: > + len_data += len(node.data[field]) > + else: > + len_data += 1 > + t = (prefix, node.data[field],) > + result.append(t) > + if maxResults is None or len_data < maxResults: > + tree.delete(prefix) > + node = tree.search_best(ip.strNormal()) > + else: > + break > + return result > + > + > +def trim_by_client_country(s, clientCountry): > + if clientCountry is None: > + return s > + r = s.copy() > + for hostid in s: > + if hostid in host_country_allowed_cache and \ > + clientCountry not in host_country_allowed_cache[hostid]: > + r.remove(hostid) > + return r > + > + > +def shuffle(s): > + l = [] > + for hostid in s: > + item = (host_bandwidth_cache[hostid], hostid) > + l.append(item) > + newlist = weighted_shuffle(l) > + results = [] > + for (bandwidth, hostid) in newlist: > + results.append(hostid) > + return results > + > + > +continents = {} > + > +def handle_country_continent_redirect(): > + new_country_continents = GeoIP.country_continents > + for country, continent in country_continent_redirect_cache.iteritems(): > + new_country_continents[country] = continent > + global country_continents > + country_continents = new_country_continents > + > + > +def setup_continents(): > + new_continents = defaultdict(list) > + handle_country_continent_redirect() > + for c, continent in country_continents.iteritems(): > + new_continents[continent].append(c) > + global continents > + continents = new_continents > + > + > +def do_global(kwargs, cache, clientCountry, header): > + c = trim_by_client_country(cache['global'], clientCountry) > + header += 'country = global ' > + return (header, c) > + > + > +def do_countrylist(kwargs, cache, clientCountry, requested_countries, > header): > + > + def collapse(d): > + """ collapses a dict {key:set(hostids)} into a set of hostids """ > + s = set() > + for country, hostids in d.iteritems(): > + for hostid in hostids: > + s.add(hostid) > + return s > + > + country_cache = {} > + for c in requested_countries: > + if c in cache['byCountry']: > + country_cache[c] = cache['byCountry'][c] > + header += 'country = %s ' % c > + s = collapse(country_cache) > + s = trim_by_client_country(s, clientCountry) > + return (header, s) > + > + > +def get_same_continent_countries(clientCountry, requested_countries): > + result = [] > + for r in requested_countries: > + if r in country_continents: > + requestedCountries = [ > + c.upper() for c in continents[country_continents[r]] > + if c != clientCountry] > + result.extend(requestedCountries) > + result = uniqueify(result) > + return result > + > + > +def do_continent(kwargs, cache, clientCountry, requested_countries, header): > + if len(requested_countries) > 0: > + rc = requested_countries > + else: > + rc = [clientCountry] > + clist = get_same_continent_countries(clientCountry, rc) > + return do_countrylist(kwargs, cache, clientCountry, clist, header) > + > + > +def do_country(kwargs, cache, clientCountry, requested_countries, header): > + if 'GLOBAL' in requested_countries: > + return do_global(kwargs, cache, clientCountry, header) > + return do_countrylist( > + kwargs, cache, clientCountry, requested_countries, header) > + > + > +def do_netblocks(kwargs, cache, header): > + hostresults = set() > + if not kwargs.has_key('netblock') or kwargs['netblock'] == "1": > + tree_results = tree_lookup(host_netblocks_tree, kwargs['IP'], > 'hosts') > + for (prefix, hostids) in tree_results: > + for hostid in hostids: > + if hostid in cache['byHostId']: > + hostresults.add((prefix, hostid,)) > + header += 'Using preferred netblock ' > + return (header, hostresults) > + > + > +def do_internet2(kwargs, cache, clientCountry, header): > + hostresults = set() > + ip = kwargs['IP'] > + if ip is None: > + return (header, hostresults) > + asn = lookup_ip_asn(internet2_tree, ip) > + if asn is not None: > + header += 'Using Internet2 ' > + if clientCountry is not None \ > + and clientCountry in cache['byCountryInternet2']: > + hostresults = cache['byCountryInternet2'][clientCountry] > + hostresults = trim_by_client_country(hostresults, clientCountry) > + return (header, hostresults) > + > + > +def do_asn(kwargs, cache, header): > + hostresults = set() > + ip = kwargs['IP'] > + if ip is None: > + return (header, hostresults) > + asn = lookup_ip_asn(global_tree, ip) > + if asn is not None and asn in asn_host_cache: > + for hostid in asn_host_cache[asn]: > + if hostid in cache['byHostId']: > + hostresults.add(hostid) > + header += 'Using ASN %s ' % asn > + return (header, hostresults) > + > + > +def do_geoip(kwargs, cache, clientCountry, header): > + hostresults = set() > + if clientCountry is not None and clientCountry in cache['byCountry']: > + hostresults = cache['byCountry'][clientCountry] > + header += 'country = %s ' % clientCountry > + hostresults = trim_by_client_country(hostresults, clientCountry) > + return (header, hostresults) > + > + > +def do_location(kwargs, header): > + hostresults = set() > + if 'location' in kwargs and kwargs['location'] in location_cache: > + hostresults = set(location_cache[kwargs['location']]) > + header += "Using location %s " % kwargs['location'] > + return (header, hostresults) > + > + > +def append_path(hosts, cache, file, pathIsDirectory=False): > + """ given a list of hosts, return a list of objects: > + [(hostid, [hcurls]), ... ] > + in the same order, appending file if it's not None""" > + subpath = None > + results = [] > + if 'subpath' in cache: > + subpath = cache['subpath'] > + for hostid in hosts: > + hcurls = [] > + for hcurl_id in cache['byHostId'][hostid]: > + s = hcurl_cache[hcurl_id] > + if subpath is not None: > + s += "/" + subpath > + if file is None and pathIsDirectory: > + s += "/" > + if file is not None: > + if not s.endswith('/'): > + s += "/" > + s += file > + hcurls.append(s) > + results.append((hostid, hcurls)) > + return results > + > + > +def trim_to_preferred_protocols(hosts_and_urls): > + """ remove all but http and ftp URLs, > + and if both http and ftp are offered, > + leave only http. Return [(hostid, url), ...] """ > + results = [] > + try_protocols = ('https', 'http', 'ftp') > + for (hostid, hcurls) in hosts_and_urls: > + protocols = {} > + url = None > + for hcurl in hcurls: > + for p in try_protocols: > + if hcurl.startswith(p+':'): > + protocols[p] = hcurl > + > + for p in try_protocols: > + if p in protocols: > + url = protocols[p] > + break > + > + if url is not None: > + results.append((hostid, url)) > + return results > + > + > +def client_ip_to_country(ip): > + clientCountry = None > + if ip is None: > + return None > + > + # lookup in the cache first > + tree_results = tree_lookup( > + netblock_country_tree, ip, 'country', maxResults=1) > + if len(tree_results) > 0: > + (prefix, clientCountry) = tree_results[0] > + return clientCountry > + > + # attempt IPv6, then IPv6 6to4 as IPv4, then Teredo, then IPv4 > + try: > + if ip.version() == 6: > + if gipv6 is not None: > + clientCountry = gipv6.country_code_by_addr_v6( > + ip.strNormal()) > + if clientCountry is None: > + # Try the IPv6-to-IPv4 translation schemes > + for scheme in (convert_6to4_v4, convert_teredo_v4): > + result = scheme(ip) > + if result is not None: > + ip = result > + break > + if ip.version() == 4 and gipv4 is not None: > + clientCountry = gipv4.country_code_by_addr(ip.strNormal()) > + except: > + pass > + return clientCountry > + > + > +def do_mirrorlist(kwargs): > + global debug > + global logfile > + > + def return_error(kwargs, message='', returncode=200): > + d = dict( > + returncode=returncode, > + message=message, > + resulttype='mirrorlist', > + results=[]) > + if 'metalink' in kwargs and kwargs['metalink']: > + d['resulttype'] = 'metalink' > + d['results'] = metalink_failuredoc(message) > + return d > + > + if not (kwargs.has_key('repo') \ > + and kwargs.has_key('arch')) \ > + and not kwargs.has_key('path'): > + return return_error( > + kwargs, > + message='# either path=, or repo= and arch= must be specified') > + > + file = None > + cache = None > + pathIsDirectory = False > + if kwargs.has_key('path'): > + path = kwargs['path'].strip('/') > + > + # Strip duplicate "//" from the path > + path = path.replace('//', '/') > + > + header = "# path = %s " % (path) > + > + sdir = path.split('/') > + try: > + # path was to a directory > + cache = mirrorlist_cache['/'.join(sdir)] > + pathIsDirectory=True > + except KeyError: > + # path was to a file, try its directory > + file = sdir[-1] > + sdir = sdir[:-1] > + try: > + cache = mirrorlist_cache['/'.join(sdir)] > + except KeyError: > + return return_error( > + kwargs, message=header + 'error: invalid path') > + dir = '/'.join(sdir) > + else: > + if u'source' in kwargs['repo']: > + kwargs['arch'] = u'source' > + repo = repo_redirect.get(kwargs['repo'], kwargs['repo']) > + arch = kwargs['arch'] > + header = "# repo = %s arch = %s " % (repo, arch) > + > + if repo in disabled_repositories: > + return return_error(kwargs, message=header + 'repo disabled') > + try: > + dir = repo_arch_to_directoryname[(repo, arch)] > + if 'metalink' in kwargs and kwargs['metalink']: > + dir += '/repodata' > + file = 'repomd.xml' > + else: > + pathIsDirectory=True > + cache = mirrorlist_cache[dir] > + except KeyError: > + repos = repo_arch_to_directoryname.keys() > + repos.sort() > + repo_information = header + "error: invalid repo or arch\n" > + repo_information += "# following repositories are available:\n" > + for i in repos: > + if i[0] is not None and i[1] is not None: > + repo_information += "# repo=%s&arch=%s\n" % i > + return return_error(kwargs, message=repo_information) > + > + # set kwargs['IP'] exactly once > + try: > + kwargs['IP'] = IP(kwargs['client_ip']) > + except: > + kwargs['IP'] = None > + > + ordered_mirrorlist = cache.get( > + 'ordered_mirrorlist', default_ordered_mirrorlist) > + done = 0 > + location_results = set() > + netblock_results = set() > + asn_results = set() > + internet2_results = set() > + country_results = set() > + geoip_results = set() > + continent_results = set() > + global_results = set() > + > + header, location_results = do_location(kwargs, header) > + > + requested_countries = [] > + if kwargs.has_key('country'): > + requested_countries = uniqueify( > + [c.upper() for c in kwargs['country'].split(',') ]) > + > + # if they specify a country, don't use netblocks or ASN > + if not 'country' in kwargs: > + header, netblock_results = do_netblocks(kwargs, cache, header) > + if len(netblock_results) > 0: > + if not ordered_mirrorlist: > + done=1 > + > + if not done: > + header, asn_results = do_asn(kwargs, cache, header) > + if len(asn_results) + len(netblock_results) >= 3: > + if not ordered_mirrorlist: > + done = 1 > + > + clientCountry = client_ip_to_country(kwargs['IP']) > + > + if clientCountry is None: > + print_client_country = "N/A" > + else: > + print_client_country = clientCountry > + > + if debug and kwargs.has_key('repo') and kwargs.has_key('arch'): > + msg = "IP: %s; DATE: %s; COUNTRY: %s; REPO: %s; ARCH: %s\n" % ( > + (kwargs['IP'] or 'None'), time.strftime("%Y-%m-%d"), > + print_client_country, kwargs['repo'], kwargs['arch']) > + > + sys.stdout.write(msg) > + sys.stdout.flush() > + > + if logfile is not None: > + logfile.write(msg) > + logfile.flush() > + > + if not done: > + header, internet2_results = do_internet2( > + kwargs, cache, clientCountry, header) > + if len(internet2_results) + len(netblock_results) + len(asn_results) > >= 3: > + if not ordered_mirrorlist: > + done = 1 > + > + if not done and 'country' in kwargs: > + header, country_results = do_country( > + kwargs, cache, clientCountry, requested_countries, header) > + if len(country_results) == 0: > + header, continent_results = do_continent( > + kwargs, cache, clientCountry, requested_countries, header) > + done = 1 > + > + if not done: > + header, geoip_results = do_geoip( > + kwargs, cache, clientCountry, header) > + if len(geoip_results) >= 3: > + if not ordered_mirrorlist: > + done = 1 > + > + if not done: > + header, continent_results = do_continent( > + kwargs, cache, clientCountry, [], header) > + if len(geoip_results) + len(continent_results) >= 3: > + done = 1 > + > + if not done: > + header, global_results = do_global( > + kwargs, cache, clientCountry, header) > + > + def _random_shuffle(s): > + l = list(s) > + random.shuffle(l) > + return l > + > + def _ordered_netblocks(s): > + def ipy_len(t): > + (prefix, hostid) = t > + return IP(prefix).len() > + v4_netblocks = [] > + v6_netblocks = [] > + for (prefix, hostid) in s: > + ip = IP(prefix) > + if ip.version() == 4: > + v4_netblocks.append((prefix, hostid)) > + elif ip.version() == 6: > + v6_netblocks.append((prefix, hostid)) > + # mix up the order, as sort will preserve same-key ordering > + random.shuffle(v4_netblocks) > + v4_netblocks.sort(key=ipy_len) > + random.shuffle(v6_netblocks) > + v6_netblocks.sort(key=ipy_len) > + v4_netblocks = [t[1] for t in v4_netblocks] > + v6_netblocks = [t[1] for t in v6_netblocks] > + return v6_netblocks + v4_netblocks > + > + def whereismymirror(result_sets): > + return_string = 'None' > + allhosts = [] > + found = False > + for (l,s,f) in result_sets: > + if len(l) > 0: > + allhosts.extend(f(l)) > + if not found: > + return_string = s > + found = True > + > + allhosts = uniqueify(allhosts) > + return allhosts, return_string > + > + result_sets = [ > + (location_results, "location", _random_shuffle), > + (netblock_results, "netblocks", _ordered_netblocks), > + (asn_results, "asn", _random_shuffle), > + (internet2_results, "I2", _random_shuffle), > + (country_results, "country", shuffle), > + (geoip_results, "geoip", shuffle), > + (continent_results, "continent", shuffle), > + (global_results, "global", shuffle), > + ] > + > + allhosts, where_string = whereismymirror(result_sets) > + try: > + ip_str = kwargs['IP'].strNormal() > + except: > + ip_str = 'Unknown IP' > + log_string = "mirrorlist: %s found its best mirror from %s" % ( > + ip_str, where_string) > + syslogger.info(log_string) > + > + hosts_and_urls = append_path( > + allhosts, cache, file, pathIsDirectory=pathIsDirectory) > + > + if 'metalink' in kwargs and kwargs['metalink']: > + (resulttype, returncode, results)=metalink( > + cache, dir, file, hosts_and_urls) > + d = dict( > + message=None, > + resulttype=resulttype, > + returncode=returncode, > + results=results) > + return d > + > + else: > + host_url_list = trim_to_preferred_protocols(hosts_and_urls) > + d = dict( > + message=header, > + resulttype='mirrorlist', > + returncode=200, > + results=host_url_list) > + return d > + > + > +def setup_cache_tree(cache, field): > + tree = radix.Radix() > + for k, v in cache.iteritems(): > + node = tree.add(k.strNormal()) > + node.data[field] = v > + return tree > + > + > +def setup_netblocks(netblocks_file, asns_wanted=None): > + tree = radix.Radix() > + if netblocks_file is not None: > + try: > + f = open(netblocks_file, 'r') > + except: > + return tree > + for l in f: > + try: > + s = l.split() > + start, mask = s[0].split('/') > + mask = int(mask) > + if mask == 0: continue > + asn = int(s[1]) > + if asns_wanted is None or asn in asns_wanted: > + node = tree.add(s[0]) > + node.data['asn'] = asn > + except: > + pass > + f.close() > + return tree > + > + > +def read_caches(): > + global mirrorlist_cache > + global host_netblock_cache > + global host_country_allowed_cache > + global host_max_connections_cache > + global repo_arch_to_directoryname > + global repo_redirect > + global country_continent_redirect_cache > + global disabled_repositories > + global host_bandwidth_cache > + global host_country_cache > + global file_details_cache > + global hcurl_cache > + global asn_host_cache > + global location_cache > + global netblock_country_cache > + > + data = {} > + try: > + f = open(cachefile, 'r') > + data = pickle.load(f) > + f.close() > + except: > + pass > + > + if 'mirrorlist_cache' in data: > + mirrorlist_cache = data['mirrorlist_cache'] > + if 'host_netblock_cache' in data: > + host_netblock_cache = data['host_netblock_cache'] > + if 'host_country_allowed_cache' in data: > + host_country_allowed_cache = data['host_country_allowed_cache'] > + if 'repo_arch_to_directoryname' in data: > + repo_arch_to_directoryname = data['repo_arch_to_directoryname'] > + if 'repo_redirect_cache' in data: > + repo_redirect = data['repo_redirect_cache'] > + if 'country_continent_redirect_cache' in data: > + country_continent_redirect_cache = data[ > + 'country_continent_redirect_cache'] > + if 'disabled_repositories' in data: > + disabled_repositories = data['disabled_repositories'] > + if 'host_bandwidth_cache' in data: > + host_bandwidth_cache = data['host_bandwidth_cache'] > + if 'host_country_cache' in data: > + host_country_cache = data['host_country_cache'] > + if 'file_details_cache' in data: > + file_details_cache = data['file_details_cache'] > + if 'hcurl_cache' in data: > + hcurl_cache = data['hcurl_cache'] > + if 'asn_host_cache' in data: > + asn_host_cache = data['asn_host_cache'] > + if 'location_cache' in data: > + location_cache = data['location_cache'] > + if 'netblock_country_cache' in data: > + netblock_country_cache = data['netblock_country_cache'] > + if 'host_max_connections_cache' in data: > + host_max_connections_cache = data['host_max_connections_cache'] > + > + setup_continents() > + global internet2_tree > + global global_tree > + global host_netblocks_tree > + global netblock_country_tree > + > + internet2_tree = setup_netblocks(internet2_netblocks_file) > + global_tree = setup_netblocks(global_netblocks_file, asn_host_cache) > + # host_netblocks_tree key is a netblock, value is a list of host IDs > + host_netblocks_tree = setup_cache_tree(host_netblock_cache, 'hosts') > + # netblock_country_tree key is a netblock, value is a single country > string > + netblock_country_tree = setup_cache_tree( > + netblock_country_cache, 'country') > + > + > +def errordoc(metalink, message): > + if metalink: > + doc = metalink_failuredoc(message) > + else: > + doc = message > + return doc > + > + > +class MirrorlistHandler(StreamRequestHandler): > + def handle(self): > + signal.signal(signal.SIGHUP, signal.SIG_IGN) > + random.seed() > + try: > + # read size of incoming pickle > + readlen = 0 > + size = '' > + while readlen < 10: > + size += self.rfile.read(10 - readlen) > + readlen = len(size) > + size = atoi(size) > + > + # read the pickle > + readlen = 0 > + p = '' > + while readlen < size: > + p += self.rfile.read(size - readlen) > + readlen = len(p) > + d = pickle.loads(p) > + self.connection.shutdown(socket.SHUT_RD) > + except: > + pass > + > + try: > + try: > + r = do_mirrorlist(d) > + except: > + raise > + message = r['message'] > + results = r['results'] > + resulttype = r['resulttype'] > + returncode = r['returncode'] > + except Exception, e: > + message=u'# Bad Request %s\n# %s' % (e, d) > + exception_msg = traceback.format_exc(e) > + sys.stderr.write(message+'\n') > + sys.stderr.write(exception_msg) > + sys.stderr.flush() > + returncode = 400 > + results = [] > + resulttype = 'mirrorlist' > + if d['metalink']: > + resulttype = 'metalink' > + results = errordoc(d['metalink'], message) > + > + try: > + p = pickle.dumps({ > + 'message':message, > + 'resulttype':resulttype, > + 'results':results, > + 'returncode':returncode}) > + self.connection.sendall(zfill('%s' % len(p), 10)) > + > + self.connection.sendall(p) > + self.connection.shutdown(socket.SHUT_WR) > + except: > + pass > + > + > +def sighup_handler(signum, frame): > + global logfile > + if logfile is not None: > + name = logfile.name > + logfile.close() > + logfile = open(name, 'a') > + > + # put this in a separate thread so it doesn't block clients > + if threading.active_count() < 2: > + thread = threading.Thread(target=load_databases_and_caches) > + thread.daemon = False > + try: > + thread.start() > + except KeyError: > + # bug fix for handing an exception when unable to delete from > + #_limbo even though it's not in limbo > + # > https://code.google.com/p/googleappengine/source/browse/trunk/python/google/appengine/dist27/threading.py?r=327 > + pass > + > + > +def sigterm_handler(signum, frame): > + global must_die > + signal.signal(signal.SIGHUP, signal.SIG_IGN) > + signal.signal(signal.SIGTERM, signal.SIG_IGN) > + if signum == signal.SIGTERM: > + must_die = True > + > + > +class ForkingUnixStreamServer(ForkingMixIn, UnixStreamServer): > + request_queue_size = 300 > + def finish_request(self, request, client_address): > + signal.signal(signal.SIGHUP, signal.SIG_IGN) > + BaseServer.finish_request(self, request, client_address) > + > + > +def parse_args(): > + global cachefile > + global socketfile > + global internet2_netblocks_file > + global global_netblocks_file > + global debug > + global logfile > + global pidfile > + opts, args = getopt.getopt( > + sys.argv[1:], "c:i:g:p:s:dl:", > + [ > + "cache", "internet2_netblocks", "global_netblocks", > + "pidfile", "socket", "debug", "log=" > + ] > + ) > + for option, argument in opts: > + if option in ("-c", "--cache"): > + cachefile = argument > + if option in ("-i", "--internet2_netblocks"): > + internet2_netblocks_file = argument > + if option in ("-g", "--global_netblocks"): > + global_netblocks_file = argument > + if option in ("-s", "--socket"): > + socketfile = argument > + if option in ("-p", "--pidfile"): > + pidfile = argument > + if option in ("-l", "--log"): > + try: > + logfile = open(argument, 'a') > + except: > + logfile = None > + if option in ("-d", "--debug"): > + debug = True > + > + > +def open_geoip_databases(): > + global gipv4 > + global gipv6 > + try: > + gipv4 = GeoIP.open( > + "/usr/share/GeoIP/GeoIP.dat", GeoIP.GEOIP_STANDARD) > + except: > + gipv4=None > + try: > + gipv6 = GeoIP.open( > + "/usr/share/GeoIP/GeoIPv6.dat", GeoIP.GEOIP_STANDARD) > + except: > + gipv6=None > + > + > +def convert_6to4_v4(ip): > + all_6to4 = IP('2002::/16') > + if ip.version() != 6 or ip not in all_6to4: > + return None > + parts=ip.strNormal().split(':') > + > + ab = int(parts[1],16) > + a = (ab >> 8) & 0xFF > + b = ab & 0xFF > + cd = int(parts[2],16) > + c = (cd >> 8) & 0xFF > + d = cd & 0xFF > + > + v4addr = '%d.%d.%d.%d' % (a,b,c,d) > + return IP(v4addr) > + > + > +def convert_teredo_v4(ip): > + teredo_std = IP('2001::/32') > + teredo_xp = IP('3FFE:831F::/32') > + if ip.version() != 6 or (ip not in teredo_std and ip not in teredo_xp): > + return None > + parts=ip.strNormal().split(':') > + > + ab = int(parts[6],16) > + a = ((ab >> 8) & 0xFF) ^ 0xFF > + b = (ab & 0xFF) ^ 0xFF > + cd = int(parts[7],16) > + c = ((cd >> 8) & 0xFF) ^ 0xFF > + d = (cd & 0xFF) ^ 0xFF > + > + v4addr = '%d.%d.%d.%d' % (a,b,c,d) > + return IP(v4addr) > + > + > +def load_databases_and_caches(*args, **kwargs): > + sys.stderr.write("load_databases_and_caches...") > + sys.stderr.flush() > + open_geoip_databases() > + read_caches() > + sys.stderr.write("done.\n") > + sys.stderr.flush() > + > + > +def remove_pidfile(pidfile): > + os.unlink(pidfile) > + > + > +def create_pidfile_dir(pidfile): > + piddir = os.path.dirname(pidfile) > + if not piddir: > + return > + try: > + os.makedirs(piddir, mode=0755) > + except OSError, err: > + if err.errno == 17: # File exists > + pass > + else: > + raise > + except: > + raise > + > + > +def write_pidfile(pidfile, pid): > + create_pidfile_dir(pidfile) > + f = open(pidfile, 'w') > + f.write(str(pid)+'\n') > + f.close() > + return 0 > + > + > +def manage_pidfile(pidfile): > + """returns 1 if another process is running that is named in pidfile, > + otherwise creates/writes pidfile and returns 0.""" > + pid = os.getpid() > + try: > + f = open(pidfile, 'r') > + except IOError, err: > + if err.errno == 2: # No such file or directory > + return write_pidfile(pidfile, pid) > + return 1 > + > + oldpid=f.read() > + f.close() > + > + # is the oldpid process still running? > + try: > + os.kill(int(oldpid), 0) > + except ValueError: # malformed oldpid > + return write_pidfile(pidfile, pid) > + except OSError, err: > + if err.errno == 3: # No such process > + return write_pidfile(pidfile, pid) > + return 1 > + > + > +def main(): > + global logfile > + global pidfile > + signal.signal(signal.SIGHUP, signal.SIG_IGN) > + parse_args() > + manage_pidfile(pidfile) > + > + oldumask = os.umask(0) > + try: > + os.unlink(socketfile) > + except: > + pass > + > + load_databases_and_caches() > + signal.signal(signal.SIGHUP, sighup_handler) > + # restart interrupted syscalls like select > + signal.siginterrupt(signal.SIGHUP, False) > + ss = ForkingUnixStreamServer(socketfile, MirrorlistHandler) > + > + while not must_die: > + try: > + ss.serve_forever() > + except select.error: > + pass > + > + try: > + os.unlink(socketfile) > + except: > + pass > + > + if logfile is not None: > + try: > + logfile.close() > + except: > + pass > + > + remove_pidfile(pidfile) > + return 0 > + > + > +if __name__ == "__main__": > + try: > + sys.exit(main()) > + except KeyboardInterrupt: > + sys.exit(-1) > > _______________________________________________ > infrastructure mailing list > infrastructure@xxxxxxxxxxxxxxxxxxxxxxx > http://lists.fedoraproject.org/admin/infrastructure@xxxxxxxxxxxxxxxxxxxxxxx > _______________________________________________ infrastructure mailing list infrastructure@xxxxxxxxxxxxxxxxxxxxxxx http://lists.fedoraproject.org/admin/infrastructure@xxxxxxxxxxxxxxxxxxxxxxx
