From 5403fd3d9f4189745f00769eef1bf49c999fc183 Mon Sep 17 00:00:00 2001 From: olveyra Date: Wed, 16 Jul 2008 19:17:46 +0000 Subject: [PATCH] - added disable_node and enable_node functions - removed unused imports - autoreload of lost nodes - some code improvements and fixes --HG-- extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%4071 --- .../contrib/pbcluster/master/manager.py | 90 ++++++++++++------- .../scrapy/contrib/pbcluster/master/web.py | 10 ++- .../contrib/pbcluster/master/ws_api.txt | 4 + 3 files changed, 70 insertions(+), 34 deletions(-) diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py b/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py index ae274291d..b65b56bd6 100644 --- a/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py +++ b/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py @@ -1,14 +1,10 @@ -import urlparse -import urllib -import bisect import sys -import pickle, socket +import pickle from pydispatch import dispatcher from twisted.spread import pb from twisted.internet import reactor -from twisted.python import util from scrapy.core import log, signals from scrapy.core.engine import scrapyengine @@ -37,13 +33,25 @@ class Node: self._set_status(status) self.name = name self.master = master + self.available = True + @property + def status_as_dict(self): + status = {"alive": self.alive} + if self.alive: + status["running"] = self.running + status["maxproc"] = self.maxproc + status["available"] = self.available + status["starttime"] = self.starttime + status["timestamp"] = self.timestamp + status["loadavg"] = self.loadavg + return status + def _set_status(self, status): - self.status_as_dict = status if not status: - self.available = False + self.alive = False else: - self.available = True + self.alive = True self.running = status['running'] self.maxproc = status['maxproc'] self.starttime = status['starttime'] @@ -52,16 +60,16 @@ class Node: 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. Th - if 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 will be moved to run later) - if pending['domain'] in self.master.running or pending['domain'] in self.master.loading: - self.master.schedule([pending['domain']], pending['settings'], pending['priority']) - else: - self.run(pending) - self.master.loading.append(pending['domain']) + #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.alive and 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 will be moved to run later) + if pending['domain'] in self.master.running or pending['domain'] in self.master.loading: + self.master.schedule([pending['domain']], pending['settings'], pending['priority']) + else: + self.run(pending) + self.master.loading.append(pending['domain']) def get_status(self): try: @@ -83,6 +91,12 @@ class Node: def run(self, pending): + def _run_errback(reason): + log.msg(reason, log.ERROR) + self.master.loading.remove(pending['domain']) + self.master.schedule([pending['domain']], pending['settings'], pending['priority'] - 1) + log.msg("Domain %s rescheduled: lost connection to node." % pending['domain'], log.WARNING) + def _run_callback(status): self.master.loading.remove(pending['domain']) if status['callresponse'][0] == 1: @@ -102,7 +116,7 @@ class Node: self._set_status(None) log.msg("Lost connection to node %s." % (self.name), log.ERROR) else: - deferred.addCallbacks(callback=_run_callback, errback=lambda reason:log.msg(reason, log.ERROR)) + deferred.addCallbacks(callback=_run_callback, errback=_run_errback) class ClusterMaster(pb.Root): @@ -151,22 +165,26 @@ class ClusterMaster(pb.Root): d = _factory.getRootObject() d.addCallbacks(callback=lambda obj: self.add_node(obj, _name), errback=_errback) - if name not in self.nodes: - server, port = url.split(":") - port = eval(port) - log.msg("Connecting to cluster worker %s..." % name) - log.msg("Server: %s, Port: %s" % (server, port)) - factory = pb.PBClientFactory() - try: - reactor.connectTCP(server, port, factory) - except Exception, err: - log.msg("Could not connect to node %s in %s: %s." % (name, url, reason), log.ERROR) - else: - _make_callback(factory, name, url) + server, port = url.split(":") + port = eval(port) + log.msg("Connecting to cluster worker %s..." % name) + log.msg("Server: %s, Port: %s" % (server, port)) + factory = pb.PBClientFactory() + try: + reactor.connectTCP(server, port, factory) + except Exception, err: + log.msg("Could not connect to node %s in %s: %s." % (name, url, reason), log.ERROR) + else: + _make_callback(factory, name, url) def update_nodes(self): - for node in self.nodes.itervalues(): - node.get_status() + for name, url in settings.get('CLUSTER_MASTER_NODES', {}).iteritems(): + if name in self.nodes and self.nodes[name].alive: + log.msg("Updating node. name: %s, url: %s" % (name, url) ) + self.nodes[name].get_status() + else: + log.msg("Reloading node. name: %s, url: %s" % (name, url) ) + self.load_node(name, url) def add_node(self, cworker, name): """Add node given its node""" @@ -175,6 +193,12 @@ class ClusterMaster(pb.Root): self.nodes[name] = node log.msg("Added cluster worker %s" % name) + def disable_node(self, name): + self.nodes[name].available = False + + def enable_node(self, name): + self.nodes[name].available = True + def remove_node(self, nodename): raise NotImplemented diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py b/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py index 3227fa3e0..92ac9f83c 100644 --- a/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py +++ b/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py @@ -95,7 +95,15 @@ class ClusterMasterWeb(ClusterMaster): 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 ws: return self.ws_status(wc_request) else: diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt b/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt index dc8042b63..f85c549e4 100644 --- a/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt +++ b/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt @@ -22,6 +22,10 @@ Query parameters: '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' + Examples: 1) Schedule argos.co.uk, diy.com, littlewoodsdirect.com spiders, with priority=PRIORITY_NOW, and settings UNAVAILABLES_NOTIFY=2 and UNAVAILABLES_DAYS_BACK=3. Answer with pprint format