diff --git a/docs/ref/settings.rst b/docs/ref/settings.rst index 77b998b3c..ae67516d7 100644 --- a/docs/ref/settings.rst +++ b/docs/ref/settings.rst @@ -81,98 +81,6 @@ Default: ``True`` Whether to split HTTP cache storage in several dirs for performance. -.. setting:: CLUSTER_LOGDIR - -CLUSTER_LOGDIR --------------- - -Default: ``''`` (empty string) - -The directory to use for cluster logging. - -.. setting:: CLUSTER_MASTER_CACHEFILE - -CLUSTER_MASTER_CACHEFILE ------------------------- - -Default: ``''`` - -The file to use for storing the state of the cluster master, before shotting -down. And also used for restoring the state on start up. If not set, state -won't be persisted. - -.. setting:: CLUSTER_MASTER_ENABLED - -CLUSTER_MASTER_ENABLED ------------------------- - -Default: ``False`` - -A boolean which specifies whether to enabled the cluster master. - -.. setting:: CLUSTER_MASTER_NODES - -CLUSTER_MASTER_NODES --------------------- - -Default: ``{}`` - -A dict which defines the nodes of the cluster. The keys are the node/worker -names and the values are the worker URLs. - -Example:: - - CLUSTER_MASTER_NODES = { - 'local': 'localhost:8789', - 'remote': 'someworker.example.com:8789', - } - -.. setting:: CLUSTER_MASTER_POLL_INTERVAL - -CLUSTER_MASTER_POLL_INTERVAL ----------------------------- - -Default: ``60`` - -The amount of time (in secs) that the master should wait before polling the -workers. - -.. setting:: CLUSTER_MASTER_PORT - -CLUSTER_MASTER_PORT -------------------- - -Default: ``8790`` - -The port where the cluster master will listen. - -.. setting:: CLUSTER_WORKER_ENABLED - -CLUSTER_WORKER_ENABLED ------------------------- - -Default: ``False`` - -A boolean which specifies whether to enabled the cluster master. - -.. setting:: CLUSTER_WORKER_MAXPROC - -CLUSTER_WORKER_MAXPROC ------------------------- - -Default: ``4`` - -The maximum number of process that the cluster worker will be allowed to spawn. - -.. setting:: CLUSTER_WORKER_PORT - -CLUSTER_WORKER_PORT -------------------- - -Default: ``8789`` - -The port where the cluster worker will listen. - .. setting:: COMMANDS_MODULE COMMANDS_MODULE diff --git a/scrapy/conf/default_settings.py b/scrapy/conf/default_settings.py index 1162eb094..ac0113ef3 100644 --- a/scrapy/conf/default_settings.py +++ b/scrapy/conf/default_settings.py @@ -23,18 +23,6 @@ BOT_VERSION = '1.0' CLOSEDOMAIN_TIMEOUT = 0 CLOSEDOMAIN_ITEMPASSED = 0 -CLUSTER_LOGDIR = '' - -CLUSTER_MASTER_PORT = 8790 -CLUSTER_MASTER_ENABLED = 0 -CLUSTER_MASTER_POLL_INTERVAL = 60 -CLUSTER_MASTER_NODES = {} -CLUSTER_MASTER_STATEFILE = "" - -CLUSTER_WORKER_ENABLED = 0 -CLUSTER_WORKER_MAXPROC = 4 -CLUSTER_WORKER_PORT = 8789 - COMMANDS_MODULE = '' COMMANDS_SETTINGS_MODULE = '' @@ -199,6 +187,3 @@ WEBCONSOLE_ENABLED = True WEBCONSOLE_PORT = 6080 WEBCONSOLE_LOGFILE = None -# this setting is used by the cluster master to pass additional settings to -# workers at connection time -GLOBAL_CLUSTER_SETTINGS = [] diff --git a/scrapy/contrib/cluster/__init__.py b/scrapy/contrib/cluster/__init__.py deleted file mode 100644 index 2c6267ad8..000000000 --- a/scrapy/contrib/cluster/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from scrapy.contrib.cluster.worker.manager import ClusterWorker -from scrapy.contrib.cluster.master.web import ClusterMasterWeb -from scrapy.contrib.cluster.crawler.manager import ClusterCrawler diff --git a/scrapy/contrib/cluster/crawler/__init__.py b/scrapy/contrib/cluster/crawler/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/scrapy/contrib/cluster/crawler/manager.py b/scrapy/contrib/cluster/crawler/manager.py deleted file mode 100644 index 097206855..000000000 --- a/scrapy/contrib/cluster/crawler/manager.py +++ /dev/null @@ -1,41 +0,0 @@ -import os - -from twisted.spread import pb -from twisted.internet import reactor - -from scrapy.conf import settings -from scrapy import log -from scrapy.core.manager import scrapymanager -from scrapy.core.exceptions import NotConfigured - -class ClusterCrawlerBroker(pb.Referenceable): - """ClusterCrawlerBroker is the class that's used for communication between - the cluster worker and the crawling proces""" - - def __init__(self, crawler, remote): - self.__remote = remote - self.__crawler = crawler - deferred = self.__remote.callRemote("register_crawler", os.getpid(), self) - deferred.addCallbacks(callback=lambda x: None, errback=lambda reason: log.msg(reason, log.ERROR)) - - def remote_stop(self): - scrapymanager.stop() - -class ClusterCrawler(object): - """ClusterCrawler is an extension that instances a ClusterCrawlerBroker - which is used to control a crawling process from the cluster worker. It - also registers that broker to the local cluster worker""" - - def __init__(self): - if not settings.getbool('CLUSTER_CRAWLER_ENABLED'): - raise NotConfigured - - self.worker = None - - factory = pb.PBClientFactory() - reactor.connectTCP("localhost", settings.getint('CLUSTER_WORKER_PORT'), factory) - d = factory.getRootObject() - def _set_worker(obj): - self.worker = ClusterCrawlerBroker(self, obj) - d.addCallbacks(callback=_set_worker, errback=lambda reason: log.msg(reason, log.ERROR)) - diff --git a/scrapy/contrib/cluster/hooks/__init__.py b/scrapy/contrib/cluster/hooks/__init__.py deleted file mode 100644 index a66f00650..000000000 --- a/scrapy/contrib/cluster/hooks/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ -""" -This module contains pre-run hooks that can be attached to scrapy workers. - -Pre-run hooks must be callable objects (ie. functions) which implement this -interface: - -pre_hook(domain, spider_settings) - -domain is the domain to be scraped -spider_settings is the settings to use to scrape it - -Values returned from the pre-run hooks will be ignored. -""" diff --git a/scrapy/contrib/cluster/hooks/svn.py b/scrapy/contrib/cluster/hooks/svn.py deleted file mode 100644 index b3f7a6d6c..000000000 --- a/scrapy/contrib/cluster/hooks/svn.py +++ /dev/null @@ -1,23 +0,0 @@ -""" -This module contains hooks for updating code via svn. Useful for running before -starting to crawl a domain, for example to update spider code -""" - -import pysvn - -from scrapy.conf import settings -from scrapy import log - -SVN_DIR = settings['SVN_DIR'] -SVN_USER = settings['SVN_USER'] -SVN_PASS = settings['SVN_PASS'] - -def svnup(domain, spider_settings): - c = pysvn.Client() - c.callback_get_login = lambda x,y,z: (True, SVN_USER, SVN_PASS, False) - try: - r = c.update(SVN_DIR) - log.msg("ClusterWorker: SVN code updated to revision %s (triggered by domain %s)" % \ - (r[0].number, domain), level=log.DEBUG) - except pysvn.ClientError, e: - log.msg("ClusterWorker: unable to update svn code - %s" % e, level=log.WARNING) diff --git a/scrapy/contrib/cluster/master/__init__.py b/scrapy/contrib/cluster/master/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/scrapy/contrib/cluster/master/manager.py b/scrapy/contrib/cluster/master/manager.py deleted file mode 100644 index 6166fb421..000000000 --- a/scrapy/contrib/cluster/master/manager.py +++ /dev/null @@ -1,390 +0,0 @@ -from __future__ import with_statement - -import datetime -import cPickle as pickle - -from scrapy.xlib.pydispatch import dispatcher -from twisted.spread import pb -from twisted.internet import reactor - -from scrapy.core import signals -from scrapy import log -from scrapy.core.engine import scrapyengine -from scrapy.core.exceptions import NotConfigured -from scrapy.contrib.cluster.worker.manager import ResponseCode -from scrapy.conf import settings - -def my_import(name): - mod = __import__(name) - components = name.split('.') - for comp in components[1:]: - mod = getattr(mod, comp) - return mod - -class ClusterNodeBroker(pb.Referenceable): - - def __init__(self, worker, name, master): - self.unsafeTracebacks = True - self._worker = worker - self.alive = False - self.name = name - self.master = master - self.available = True - try: - deferred = self._worker.callRemote("set_master", self) - except pb.DeadReferenceError: - self._set_status(None) - log.msg("ClusterMaster: Lost connection to node %s." % self.name, log.ERROR) - else: - def _eb(failure): - self._logfailure("Error while setting master to worker node", failure) - deferred.addCallbacks(callback=self._set_status, errback=_eb) - - def status_as_dict(self, verbosity=1): - if verbosity == 0: - return - status = {"alive": self.alive} - if self.alive: - if verbosity == 1: - # dont show spider settings - status["running"] = [] - for proc in self.running: - proccopy = proc.copy() - del proccopy["settings"] - status["running"].append(proccopy) - elif verbosity == 2: - status["running"] = self.running - status["maxproc"] = self.maxproc - status["freeslots"] = self.maxproc - len(self.running) - status["available"] = self.available - status["starttime"] = self.starttime - status["timestamp"] = self.timestamp - status["loadavg"] = self.loadavg - return status - - def update_status(self): - """Update status from this worker. This is called periodically.""" - try: - deferred = self._worker.callRemote("status") - except pb.DeadReferenceError: - self._set_status(None) - log.msg("ClusterMaster: Lost connection to worker=%s." % self.name, log.ERROR) - else: - def _eb(failure): - self._logfailure("Error while updating status", failure) - deferred.addCallbacks(callback=self._set_status, errback=_eb) - - def stop(self, domain): - try: - deferred = self._worker.callRemote("stop", domain) - except pb.DeadReferenceError: - self._set_status(None) - log.msg("ClusterMaster: Lost connection to worker=%s." % self.name, log.ERROR) - else: - def _eb(failure): - self._logfailure("Error while stopping domain=%s" % domain, failure) - deferred.addCallbacks(callback=self._set_status, errback=_eb) - - def run(self, domain_info): - """Run the given domain. - - domain_info is a dict of keys: - domain - the domain to run - settings - the settings to use - priority - the priority to use - """ - - domain = domain_info['domain'] - priority = domain_info['priority'] - spider_settings = domain_info['settings'] - dsettings = self.master.compute_final_spider_settings(domain, spider_settings) - - def _run_errback(failure): - self._logfailure("Error while running domain=%s" % domain, failure) - self.master.loading.remove(domain) - newprio = priority - 1 # increase priority for reschedule - self.master.reschedule([domain], spider_settings, newprio, - reason="error while try to run it") - - def _run_callback(status): - if status['callresponse'][0] == ResponseCode.NO_FREE_SLOT: - log.msg("ClusterMaster: No available slots at worker=%s when trying to run domain=%s" - % (self.name, domain), log.WARNING) - self.master.loading.remove(domain) - newprio = priority - 1 # increase priority for rerunning asap - self.master.reschedule([domain], spider_settings, newprio, - reason="no available slots at worker=%s" % self.name) - elif status['callresponse'][0] == ResponseCode.DOMAIN_ALREADY_RUNNING: - log.msg("ClusterMaster: Already running domain=%s at worker=%s" % - (domain, self.name), log.WARNING) - self.master.loading.remove(domain) - self.master.reschedule([domain], spider_settings, priority, - reason="domain already running at worker=%s" % self.name) - - try: - log.msg("ClusterMaster: Running domain=%s at worker=%s" % (domain, self.name), log.DEBUG) - deferred = self._worker.callRemote("run", domain, dsettings) - except pb.DeadReferenceError: - self._set_status(None) - log.msg("ClusterMaster: Lost connection to worker=%s." % self.name, log.ERROR) - else: - deferred.addCallbacks(callback=_run_callback, errback=_run_errback) - - def remote_update(self, worker_status, domain, domain_status): - """Called remotely form worker when domains finish to update status""" - self._set_status(worker_status) - stats = self.master.statistics - dstats = stats["domains"] - - if domain in self.master.loading and domain_status == "running": - self.master.loading.remove(domain) - dstats["running"].add(domain) - elif domain_status in ("done", "terminated"): - dstats["running"].remove(domain) - dstats["scraped"][domain] = dstats["scraped"].get(domain, 0) + 1 - stats["scraped_count"] = stats.get("scraped_count", 0) + 1 - if domain in dstats["lost"]: - dstats["lost"].remove(domain) - - log.msg("ClusterMaster: Changed status to <%s> for domain=%s at worker=%s" % - (domain_status, domain, self.name)) - - def _logfailure(self, msg, failure): - log.msg("ClusterMaster: %s (worker=%s)\n%s" % (msg, self.name, failure), log.ERROR) - - def _set_status(self, status): - if not status: - self.alive = False - else: - self.alive = True - self.running = status['running'] - self.maxproc = status['maxproc'] - self.starttime = status['starttime'] - self.timestamp = status['timestamp'] - self.loadavg = status['loadavg'] - self.logdir = status['logdir'] - free_slots = self.maxproc - len(self.running) - - # load domains by one, so to mix up better the domain loading between nodes. - # The next one in the same node will be loaded - # when there is no loading domain or in the next status update. - # This way also we load the nodes softly - if self.available and free_slots > 0 and self.master.pending: - pending = self.master.pending.pop(0) - # If domain already running in some node, reschedule with same - # priority (so it will be run later) - if pending['domain'] in self.master.running or pending['domain'] in self.master.loading: - self.master.reschedule([pending['domain']], pending['settings'], - pending['priority'], reason="domain already running in other worker") - else: - self.run(pending) - self.master.loading.append(pending['domain']) - - -class ScrapyPBClientFactory(pb.PBClientFactory): - - noisy = False - - def __init__(self, master, nodename): - pb.PBClientFactory.__init__(self) - self.unsafeTracebacks = True - self.master = master - self.nodename = nodename - - def clientConnectionLost(self, *args, **kargs): - pb.PBClientFactory.clientConnectionLost(self, *args, **kargs) - self.master.remove_node(self.nodename) - log.msg("ClusterMaster: Lost connection to worker=%s. Node removed" % self.nodename) - - -class ClusterMaster(object): - - def __init__(self): - - if not settings.getbool('CLUSTER_MASTER_ENABLED'): - raise NotConfigured - - self.statefile = settings['CLUSTER_MASTER_STATEFILE'] - if not self.statefile: - raise NotConfigured("ClusterMaster: Missing CLUSTER_MASTER_STATEFILE setting") - - # import groups settings - if settings.getbool('GROUPSETTINGS_ENABLED'): - self.get_spider_groupsettings = my_import(settings["GROUPSETTINGS_MODULE"]).get_spider_groupsettings - else: - self.get_spider_groupsettings = lambda x: {} - - # load pending domains - try: - statefile = open(self.statefile, "r") - self.pending = pickle.load(statefile) - log.msg("ClusterMaster: Loaded state from %s" % self.statefile) - except IOError: - self.pending = [] - - self.loading = [] - self.nodes = {} - self.nodesconf = settings.get('CLUSTER_MASTER_NODES', {}) - self.start_time = datetime.datetime.utcnow() - - # for more info about statistics see self.update_nodes() and ClusterNodeBroker.remote_update() - self.statistics = {"domains": {"running": set(), "scraped": {}, "lost_count": {}, "lost": set()}, "scraped_count": 0 } - self.global_settings = {} - - # load cluster global settings - for sname in settings.getlist('GLOBAL_CLUSTER_SETTINGS'): - self.global_settings[sname] = settings[sname] - - dispatcher.connect(self._engine_started, signal=signals.engine_started) - dispatcher.connect(self._engine_stopped, signal=signals.engine_stopped) - - def load_nodes(self): - """Loads nodes listed in CLUSTER_MASTER_NODES setting""" - for name, hostport in self.nodesconf.iteritems(): - self.load_node(name, hostport) - - def load_node(self, name, hostport): - """Creates the remote reference for a worker node""" - server, port = hostport.split(":") - port = int(port) - log.msg("ClusterMaster: Connecting to worker=%s (%s)..." % (name, hostport)) - factory = ScrapyPBClientFactory(self, name) - try: - reactor.connectTCP(server, port, factory) - except Exception, err: - log.msg("ClusterMaster: Could not connect to worker=%s (%s): %s" % - (name, hostport, err), log.ERROR) - else: - def _eb(failure): - log.msg("ClusterMaster: Could not connect to worker=%s (%s): %s" % - (name, hostport, failure.value), log.ERROR) - - d = factory.getRootObject() - d.addCallbacks(callback=lambda obj: self.add_node(obj, name), errback=_eb) - - def update_nodes(self): - """Update worker nodes statistics""" - for name, hostport in self.nodesconf.iteritems(): - if name in self.nodes and self.nodes[name].alive: - self.nodes[name].update_status() - else: - self.load_node(name, hostport) - - dstats = self.statistics["domains"] - real_running = set(self.running.keys()) - lost = dstats["running"].difference(real_running) - for domain in lost: - dstats["lost_count"][domain] = dstats["lost_count"].get(domain, 0) + 1 - dstats["lost"] = dstats["lost"].union(lost) - - def add_node(self, cworker, name): - """Add node given its node""" - node = ClusterNodeBroker(cworker, name, self) - self.nodes[name] = node - log.msg("ClusterMaster: Added worker=%s" % name) - - def remove_node(self, nodename): - del self.nodes[nodename] - - def disable_node(self, name): - self.nodes[name].available = False - - def enable_node(self, name): - self.nodes[name].available = True - - def _schedule(self, domains, spider_settings=None, priority=20): - """Private method which performs the schedule of the given domains, - with the given priority. Used for both scheduling and rescheduling.""" - insert_pos = len([p for p in self.pending if p['priority'] <= priority]) - for domain in domains: - pd = self.get_first_pending(domain) - if pd: # domain already pending, so just change priority if new is higher - if priority < pd['priority']: - self.pending.remove(pd) - pd['priority'] = priority - self.pending.insert(insert_pos, pd) - else: - self.pending.insert(insert_pos, - {'domain': domain, 'settings': spider_settings, 'priority': priority}) - - def compute_final_spider_settings(self, domain, spider_settings=None): - """Return merged dictionary with final settings to run spider""" - final = dict(self.get_spider_groupsettings(domain) or {}) - final.update(self.global_settings) - final.update(spider_settings or {}) - return final - - def schedule(self, domains, spider_settings=None, priority=20): - """Schedule the given domains, with the given priority""" - self._schedule(domains, spider_settings, priority) - log.msg("clustermaster: Scheduled domains=%s with priority=%s" % - (','.join(domains), priority), log.DEBUG) - - def reschedule(self, domains, spider_settings=None, priority=20, reason=None): - """Reschedule the given domains, with the given priority""" - self._schedule(domains, spider_settings, priority) - log.msg("clustermaster: Rescheduled domains=%s with priority=%s reason='%s'" % - (','.join(domains), priority, reason), log.DEBUG) - - - def stop(self, domains): - """Stop the given domains""" - to_stop = {} - for domain in domains: - node = self.running.get(domain, None) - if node: - if node.name not in to_stop: - to_stop[node.name] = [] - to_stop[node.name].append(domain) - - for nodename, domains in to_stop.iteritems(): - for domain in domains: - self.nodes[nodename].stop(domain) - - def remove(self, domains): - """Remove all scheduled instances of the given domains (if they haven't - started yet). Otherwise use stop() to stop running domains""" - - self.pending = [p for p in self.pending if p['domain'] not in domains] - - def discard(self, domains): - """Stop and remove all running and pending instances of the given - domains""" - self.remove(domains) - self.stop(domains) - - @property - def running(self): - """Return dict of running domains as domain -> node""" - d = {} - for node in self.nodes.itervalues(): - for proc in node.running: - d[proc['domain']] = node - return d - - def get_first_pending(self, domain): - """Return first pending instance of a given domain""" - for p in self.pending: - if domain == p['domain']: - return p - - def get_pending(self, verbosity=1): - if verbosity == 1: - pending = [] - for p in self.pending: - pp = p.copy() - del pp["settings"] - pending.append(pp) - return pending - elif verbosity == 2: - return self.pending - return - - def _engine_started(self): - self.load_nodes() - scrapyengine.addtask(self.update_nodes, settings.getint('CLUSTER_MASTER_POLL_INTERVAL', 60)) - - def _engine_stopped(self): - with open(self.statefile, "w") as f: - pickle.dump(self.pending, f) - log.msg("ClusterMaster: Saved state in %s" % self.statefile) diff --git a/scrapy/contrib/cluster/master/web.py b/scrapy/contrib/cluster/master/web.py deleted file mode 100644 index 24114d442..000000000 --- a/scrapy/contrib/cluster/master/web.py +++ /dev/null @@ -1,239 +0,0 @@ -import datetime - -from scrapy.xlib.pydispatch import dispatcher - -from scrapy.spider import spiders -from scrapy.management.web import banner, webconsole_discover_module -from scrapy.contrib.cluster.master.manager import ClusterMaster -from scrapy.utils.serialization import serialize - -class ClusterMasterWeb(ClusterMaster): - webconsole_id = 'cluster_master' - webconsole_name = 'Cluster master' - - def __init__(self): - ClusterMaster.__init__(self) - - dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module) - - def webconsole_render(self, wc_request): - changes = "" - if wc_request.path == '/cluster_master/nodes/': - return self.render_nodes(wc_request) - elif wc_request.path == '/cluster_master/domains/': - return self.render_domains(wc_request) - elif wc_request.path == '/cluster_master/ws/': - return self.webconsole_control(wc_request, ws=True) - elif wc_request.args: - changes = self.webconsole_control(wc_request) - - s = self.render_header() - - s += "

Home

\n" - - s += "\n" - s += "\n" - for node in self.nodes.itervalues(): - #chkbox = "" % domain if node.status in ["up", "idle"] else " " - nodelink = "%s" % (node.name, node.name) - chkbox = " " - loadavg = "%.2f %.2f %.2f" % node.loadavg - s += "\n" % \ - (chkbox, nodelink, node.available, len(node.running), node.maxproc, loadavg) - s += "
 NameAvailableRunningLoad.avg
%s%s%s%d/%d%s
\n" - - s += "\n" - s += "\n" - - return str(s) - - def webconsole_control(self, wc_request, ws=False): - args = wc_request.args - if "updatenodes" in args: - self.update_nodes() - if ws: - return self.ws_status(wc_request) - - if "schedule" in args: - if ws: - sep = "," - domains = args["schedule"][0].split(sep) - else: - sep = "\r" - domains = args["schedule"] - priority = int(args.get("priority", [20])[0]) - - # spider settings - slist = args.get("settings", [""])[0].split(sep) - spider_settings = {} - for s in slist: - try: - k, v = s.strip().split("=") - except ValueError: - pass - else: - spider_settings[k] = v - - self.schedule(domains, spider_settings, priority) - if ws: - return self.ws_status(wc_request, verbosity=0) - - if "stop" in args: - if ws: - domains = args["stop"][0].split(",") - else: - domains=args["stop"] - self.stop(domains) - if ws: - return self.ws_status(wc_request) - - if "remove" in args: - if ws: - domains = args["remove"][0].split(",") - else: - domains=args["remove"] - self.remove(domains) - if ws: - return self.ws_status(wc_request) - if "disable_node" in args: - self.disable_node(args["disable_node"][0]) - if ws: - return self.ws_status(wc_request) - if "enable_node" in args: - self.enable_node(args["enable_node"][0]) - if ws: - return self.ws_status(wc_request) - if "statistics" in args: - if ws: - return self.ws_statistics(wc_request) - - if ws: - return self.ws_status(wc_request) - else: - return "" - - def render_nodes(self, wc_request): - if wc_request.args: - self.webconsole_control(wc_request) - - now = datetime.datetime.utcnow() - - s = self.render_header() - for node in self.nodes.itervalues(): - if node.available: - s += "

%s

\n" % (node.name, node.name) - - s += "

Running domains

\n" - if node.running: - s += "
\n" - s += "\n" - s += "\n" - for proc in node.running: - chkbox = "" % proc['domain'] if proc['status'] == "running" else " " - start_time = proc.get('starttime', None) - elapsed = now - start_time if start_time else None - s += "\n" % \ - (chkbox, proc['pid'], proc['domain'], proc['status'], elapsed, proc['logfile']) - s += "
 PIDDomainStatusRunning timeLog file
%s%s%s%s%s%s
\n" - s += "\n" % node.name - s += "

\n" % node.name - s += "
\n" - else: - s += "

No running domains on %s

\n" % node.name - - return str(s) - - def render_domains(self, wc_request): - if wc_request.args: - self.webconsole_control(wc_request) - - enabled_domains = set(spiders.asdict().keys()) - print "Enabled domains: %s" % len(enabled_domains) - inactive_domains = enabled_domains - set(self.running.keys() + [p['domain'] for p in self.pending]) - - s = self.render_header() - - s += "

Schedule domains

\n" - - s += "Inactive domains (not running or pending)
" - s += "
\n" - s += "\n" - s += "
\n" - - s += "Priority:
\n" - s += "%s" % 20 - s += "
\n" - - # spider settings - s += "Overrided spider settings:
\n" - s += "\n" - s += "
\n" - - s += "

\n" - s += "
\n" - - s += "

Domains

\n" - - s += "\n" - s += "\n" - s += self._domains_table(self.running, 'running') - s += "
DomainStatusNode
\n" - - # pending domains - s += "

Pending domains

\n" - if self.pending: - s += "
\n" - s += "\n" - s += "

\n" - s += "
\n" - else: - s += "

No pending domains

\n" - - return str(s) - - def render_header(self): - s = banner(self) - s += "

Nav: " - s += "Home | " - s += "Domains | " - s += "Nodes (update)" - s += "

" - return s - - def _domains_table(self, dict_, status): - s = "" - for domain, node in dict_.iteritems(): - s += "%s%s%s\n" % (domain, status, node.name) - return s - - def webconsole_discover_module(self): - return self - - def ws_status(self, wc_request, verbosity=1): - format = wc_request.args['format'][0] if 'format' in wc_request.args else 'json' - verbosity = int(wc_request.args['verbosity'][0]) if 'verbosity' in wc_request.args else verbosity - wc_request.setHeader('content-type', 'text/plain') - status = {} - nodes_status = {} - if verbosity > 0: - for d, n in self.nodes.iteritems(): - nodes_status[d] = n.status_as_dict(verbosity) - status["nodes"] = nodes_status - status["pending"] = self.get_pending(verbosity) - status["loading"] = self.loading - content = serialize(status, format) - return content - return "" - - def ws_statistics(self, wc_request): - format = wc_request.args['format'][0] if 'format' in wc_request.args else 'json' - content = serialize(self.statistics, format) - return content diff --git a/scrapy/contrib/cluster/master/ws_api.txt b/scrapy/contrib/cluster/master/ws_api.txt deleted file mode 100644 index c47626a08..000000000 --- a/scrapy/contrib/cluster/master/ws_api.txt +++ /dev/null @@ -1,51 +0,0 @@ -Cluster Webservice API -====================== - -The webservice API is available at - -http://server:port/cluster_master/ws/ - -With no parameters, webservice returns the cluster status. - -Query parameters -================ - - - `format`: the answer format. By default, format=json. Other formats: pprint, pickle. - - - `schedule`: schedules a comma separated list of domains. Schedule function takes optional parameters: - - "priority": sets the queue priority for the specified domains (an integer). The default is setted by "DEFAULT_PRIORITY" - setting (20 if not given). A lower priority number implies more priority. - - "settings": run settings for the specified domains. This is a comma separated list of = pairs. By default it is empty. - - - `remove`: removes from pending list a comma separated list of domains. - - - `stop`: stops comma separated list of domains (they have to be running in some node) - - - `disable_node`: disables a node so no more domains will be loaded in it until enabled again (but it will finish to run the running domains) - - - `enable_node`: revert the state setted by 'disable_node' - - - `verbosity`: sets the output verbosity level (1 is the default minimal, 2 includes domain settings, 0 disables output) - - - `statistics`: shows the pending/running/scraped/lost statistics - -Examples: ---------- - -1) Schedule argos.co.uk, diy.com, littlewoodsdirect.com spiders, with priority=0, and settings UNAVAILABLES_NOTIFY=2 and UNAVAILABLES_DAYS_BACK=3. Answer with pprint format - - http://localhost:8080/cluster_master/ws/?format=pprint&schedule=argos.co.uk,diy.com,littlewoodsdirect.com&priority=0&settings=UNAVAILABLES_NOTIFY=2,UNAVAILABLES_DAYS_BACK=3 - -2) Get status with pprint format: - - http://localhost:8080/cluster_master/ws/?format=pprint - -3) Remove from pending lists domains argos.co.uk and diy.com. Answer with pprint format: - - http://localhost:8080/cluster_master/ws/?remove=argos.co.uk,diy.com - -4) Stop running domain littlewoodsdirect.com: - - http://localhost:8080/cluster_master/ws/?stop=littlewoodsdirect.com \ No newline at end of file diff --git a/scrapy/contrib/cluster/tools/scrapy-cluster-ctl.py b/scrapy/contrib/cluster/tools/scrapy-cluster-ctl.py deleted file mode 100755 index f3ca78055..000000000 --- a/scrapy/contrib/cluster/tools/scrapy-cluster-ctl.py +++ /dev/null @@ -1,80 +0,0 @@ -#!/usr/bin/env python -""" -Cluster control script -""" - -from optparse import OptionParser -import urllib - -def main(): - parser = OptionParser(usage="Usage: scrapy-cluster-ctl.py [domain [domain [...]]] [options]" ) - parser.add_option("--disablenode", dest="disable_node", help="Disable given node (by name) so it will no accept more run requests.") - parser.add_option("--enablenode", dest="enable_node", help="Enable given node (by name) so it will accept again run requests") - parser.add_option("--format", dest="format", help="Output format. Default: pprint.", default="pprint") - parser.add_option("--list", metavar="FILE", dest="list", help="Specify a file from where to read domains, one per line.") - parser.add_option("--now", action="store_true", dest="now", help="Schedule domains to run with priority now.") - parser.add_option("--output", metavar="FILE", dest="output", help="Output file. If not given, output to stdout.") - parser.add_option("--port", dest="port", type="int", help="Cluster master port. Default: 8060.", default=8060) - parser.add_option("--remove", dest="remove", action="store_true", help="Remove from schedule domains given as args.") - parser.add_option("--schedule", dest="schedule", action="store_true", help="Schedule domains given as args.") - parser.add_option("--server", dest="server", help="Cluster master server name. Default: localhost.", default="localhost") - parser.add_option("--status", dest="status", action="store_true", help="Print cluster master status and quit.") - parser.add_option("--statistics", dest="statistics", action="store_true", help="Print cluster statistics") - parser.add_option("--stop", dest="stop", action="store_true", help="Stops a running domain.") - parser.add_option("--verbosity", dest="verbosity", type="int", help="Sets the report status verbosity.") - (opts, args) = parser.parse_args() - - output = "" - domains = [] - urlstring = "http://%s:%s/cluster_master/ws/" % (opts.server, opts.port) - post = {"format":opts.format} - if isinstance(opts.verbosity, int): - post["verbosity"] = opts.verbosity - - if args: - domains = ",".join(args) - elif opts.list: - try: - domainlist = [] - for d in open(opts.list, "r").readlines(): - domainlist.append(d.strip()) - domains = ",".join(domainlist) - except IOError: - print "Can't open file %s" % opts.list - - if opts.status: - pass - elif opts.statistics: - post["statistics"] = True - elif opts.schedule and domains: - post["schedule"] = domains - if opts.now: - post["priority"] = "0" - post["settings"] = "UNAVAILABLES_NOTIFY=2" - elif opts.remove and domains: - post["remove"] = domains - elif opts.stop and domains: - post["stop"] = domains - elif opts.disable_node: - post["disable_node"] = opts.disable_node - elif opts.enable_node: - post["enable_node"] = opts.enable_node - else: - parser.print_help() - return - - f = urllib.urlopen(urlstring, urllib.urlencode(post)) - output=f.read() - if not output: - return - if not opts.output: - print output - else: - try: - open(opts.output, "w").write(output) - except IOError: - open("/tmp/scrapy-cluster-schedule.tmp", "w").write(output) - print "Could not open file %s for writing. Output dumped to /tmp/scrapy-cluster-schedule.tmp instead." % opts.output - -if __name__ == '__main__': - main() diff --git a/scrapy/contrib/cluster/tools/test-worker.py b/scrapy/contrib/cluster/tools/test-worker.py deleted file mode 100755 index d0c143006..000000000 --- a/scrapy/contrib/cluster/tools/test-worker.py +++ /dev/null @@ -1,24 +0,0 @@ -#!/usr/bin/python2.5 - -import sys -import pprint - -from twisted.spread import pb -from twisted.internet import reactor - -factory = pb.PBClientFactory() -reactor.connectTCP("localhost", 8789, factory) -d = factory.getRootObject() - -sys.argv.pop(0) - -if not sys.argv or sys.argv[0] == '--status': - d.addCallback(lambda object: object.callRemote("status")) -elif sys.argv[0] == "--stop": - d.addCallback(lambda object: object.callRemote("stop", sys.argv[1])) -elif sys.argv[0] == "--run": - d.addCallback(lambda object: object.callRemote("run", sys.argv[1])) - -d.addCallbacks(callback=pprint.pprint, errback=lambda reason:'error: ' + str(reason.value)) -d.addCallback(lambda _: reactor.stop()) -reactor.run() diff --git a/scrapy/contrib/cluster/worker/__init__.py b/scrapy/contrib/cluster/worker/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/scrapy/contrib/cluster/worker/manager.py b/scrapy/contrib/cluster/worker/manager.py deleted file mode 100644 index 58b478273..000000000 --- a/scrapy/contrib/cluster/worker/manager.py +++ /dev/null @@ -1,208 +0,0 @@ -import os -import sys -import time -import datetime -import cPickle as pickle - -from twisted.internet import protocol, reactor -from twisted.internet.error import ProcessDone -from twisted.spread import pb - -from scrapy import log -from scrapy.core.engine import scrapyengine -from scrapy.core.exceptions import NotConfigured -from scrapy.utils.misc import load_object, gzip_file -from scrapy.conf import settings - - -class ScrapyProcessProtocol(protocol.ProcessProtocol): - - def __init__(self, worker, domain, logfile=None, spider_settings=None): - self.worker = worker - self.domain = domain - self.logfile = logfile - self.start_time = datetime.datetime.utcnow() - self.status = "starting" - self.pid = -1 - self.env = {} - # We preserve the original settings format for info purposes (avoid - # lots of unnecesary "SCRAPY_") - self.scrapy_settings = spider_settings or {} - self.scrapy_settings.update({'LOGFILE': self.logfile, - 'CLUSTER_WORKER_ENABLED': 0, - 'CLUSTER_CRAWLER_ENABLED': 1, - 'WEBCONSOLE_ENABLED': 0}) - pickled_settings = pickle.dumps(self.scrapy_settings) - self.env["SCRAPY_PICKLED_SETTINGS_TO_OVERRIDE"] = pickled_settings - # we nee to pass the worker python path to the crawling process so it - # knows where to find the local_scrapy_settings - self.env["PYTHONPATH"] = ":".join(sys.path) - - def __str__(self): - return "" % (self.domain, self.pid, self.status) - - def info(self): - """Return this scrapy process info as a dict. - - The keys are: - - domain: - the domain being crawled - pid: - the pid of this process - status: - the status of this process (starting, running) - settings: - the scrapy settings overrided for this process by the worker - logfile: - the log file being used - starttime: - the start time of this process as a UTC datetime object - """ - return {"domain": self.domain, - "pid": self.pid, - "status": self.status, - "settings": self.scrapy_settings, - "logfile": self.logfile, - "starttime": self.start_time} - - def connectionMade(self): - self.pid = self.transport.pid - log.msg("ClusterWorker: started domain=%s pid=%d log=%s" % (self.domain, self.pid, self.logfile)) - self.transport.closeStdin() - self.status = "running" - self.worker.update_master(self.domain, "running") - - def processEnded(self, status): - if settings.getbool('CLUSTER_WORKER_GZIP_LOGS'): - try: - self.logfile = gzip_file(self.logfile) - except Exception, e: - log.msg("failed to compress %s exception=%s (domain=%s, pid=%s)" % (self.logfile, e, self.domain, self.pid)) - - if isinstance(status.value, ProcessDone): - st = "done" - er = "" - else: - st = "terminated" - er = ", error=%s" % str(status.value) - log.msg("ClusterWorker: finished domain=%s status=%s pid=%d log=%s%s" % (self.domain, st, self.pid, self.logfile, er)) - del self.worker.running[self.domain] - self.worker.crawlers.pop(self.pid, None) - self.worker.update_master(self.domain, st) - -class ClusterWorker(pb.Root): - - def __init__(self): - if not settings.getbool('CLUSTER_WORKER_ENABLED'): - raise NotConfigured - - self.maxproc = settings.getint('CLUSTER_WORKER_MAXPROC') - self.logdir = settings['CLUSTER_LOGDIR'] - self.running = {} # dict of domain->ScrapyProcessControl - self.crawlers = {} # dict of pid->scrapy process remote pb connection - self.starttime = datetime.datetime.utcnow() - self.prerun_hooks = [load_object(f) for f in settings.getlist('CLUSTER_WORKER_PRERUN_HOOKS', [])] - port = settings.getint('CLUSTER_WORKER_PORT') - factory = pb.PBServerFactory(self, unsafeTracebacks=True) - scrapyengine.listenTCP(port, factory) - log.msg("Using sys.path: %s" % repr(sys.path), level=log.DEBUG) - - def status(self, rcode=0, rstring=None): - """Return the status of this worker as dict. - - The keys of the dict are: - - running: - list of dicts of processes running by this worker. for information - about the dict see ScrapyProcessControl.status() - starttime: - the start time of this worker as a UTC datetime object - timestamp: - the current timestamp as a UTC datetime object - maxproc: - the maximum number of processes supported by this worker - loadavg: - the load average of this worker. see os.getloadavg() - logdir: - the log directory used by this worker - callresponse: - response to the request performed. only available when there was a request - """ - - status = {} - status["running"] = [self.running[k].info() for k in self.running.keys()] - status["starttime"] = self.starttime - status["timestamp"] = datetime.datetime.utcnow() - status["maxproc"] = self.maxproc - status["loadavg"] = os.getloadavg() - status["logdir"] = self.logdir - status["callresponse"] = (rcode, rstring) if rstring else (0, "No request") - return status - - def update_master(self, domain, domain_status): - try: - deferred = self._master.callRemote("update", self.status(), domain, domain_status) - except pb.DeadReferenceError: - self._master = None - log.msg("Lost connection to master", log.ERROR) - else: - def _eb(failure): - log.msg("Error received from ClusterMaster\n%s" % failure, level=log.ERROR) - deferred.addErrback(_eb) - - def remote_set_master(self, master): - """Set the master for this worker""" - log.msg("ClusterWorker: ClusterMaster connected from %s:%s" % master.broker.transport.client) - self._master = master - return self.status() - - def remote_stop(self, domain): - """Stop a running domain""" - if domain in self.running: - proc = self.running[domain] - log.msg("ClusterWorker: Sending shutdown signal to domain=%s pid=%d" % (domain, proc.pid)) - d = self.crawlers[proc.pid].callRemote("stop") - def _close(): - proc.status = "closing" - d.addCallbacks(callback=_close, errback=lambda reason: log.msg(reason, log.ERROR)) - return self.status(ResponseCode.DOMAIN_STOPPED, "Stopped process %s" % proc) - else: - return self.status(ResponseCode.DOMAIN_NOT_RUNNING, "%s: domain not running" % domain) - - def remote_status(self): - """Return worker status as a dict. For infomation about the keys see - the the status() method""" - return self.status() - - def remote_run(self, domain, spider_settings=None): - """Start scraping the given domain by spawning a process""" - if len(self.running) < self.maxproc: - if not domain in self.running: - logfile = os.path.join(self.logdir, domain, time.strftime("%FT%T.log")) - if not os.path.exists(os.path.dirname(logfile)): - os.makedirs(os.path.dirname(logfile)) - - for prerun_hook in self.prerun_hooks: - prerun_hook(domain, spider_settings) - - scrapy_proc = ScrapyProcessProtocol(self, domain, logfile, spider_settings) - args = [sys.executable, sys.argv[0], 'crawl', domain] - self.running[domain] = scrapy_proc - reactor.spawnProcess(scrapy_proc, sys.executable, args=args, env=scrapy_proc.env) - return self.status(ResponseCode.DOMAIN_STARTED, "Started process %s" % scrapy_proc) - else: - return self.status(ResponseCode.DOMAIN_ALREADY_RUNNING, "Domain %s already running" % domain ) - else: - return self.status(ResponseCode.NO_FREE_SLOT, "No free slot to run another domain") - - def remote_register_crawler(self, pid, crawler): - """Register the crawler to the list of crawlers managed by this worker""" - self.crawlers[pid] = crawler - -class ResponseCode(object): - DOMAIN_STARTED = 1 - DOMAIN_STOPPED = 2 - DOMAIN_ALREADY_RUNNING = 3 - DOMAIN_NOT_RUNNING = 4 - NO_FREE_SLOT = 5 diff --git a/scrapy/tests/test_engine.py b/scrapy/tests/test_engine.py index c2970c90f..7658324b4 100644 --- a/scrapy/tests/test_engine.py +++ b/scrapy/tests/test_engine.py @@ -105,11 +105,6 @@ class EngineTest(unittest.TestCase): if not session.wasrun: session.run() - # disable extensions that cause problems with tests (probably - # because they leave the reactor in an unclean state) - from scrapy.conf import settings - settings.overrides['CLUSTER_MANAGER_ENABLED'] = 0 - def test_spider_locator(self): """ Check the spider is loaded and located properly via the SpiderLocator