mirror of https://github.com/scrapy/scrapy.git
scrapy cluster: added missing docstrings to important methods, fixed some bugs
--HG-- extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%40324
This commit is contained in:
Pablo Hoffman
parent
5349030b31
commit
150f4ed815
|
|
@ -8,21 +8,24 @@ from scrapy import log
|
|||
from scrapy.core.manager import scrapymanager
|
||||
from scrapy.core.exceptions import NotConfigured
|
||||
|
||||
class Broker(pb.Referenceable):
|
||||
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
|
||||
try:
|
||||
deferred = self.__remote.callRemote("register_crawler", os.getpid(), self)
|
||||
except pb.DeadReferenceError:
|
||||
self._set_status(None)
|
||||
log.msg("Lost connection to node %s." % (self.name), log.ERROR)
|
||||
else:
|
||||
deferred.addCallbacks(callback=lambda x: None, errback=lambda reason: log.msg(reason, log.ERROR))
|
||||
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:
|
||||
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
|
||||
|
|
@ -33,6 +36,6 @@ class ClusterCrawler:
|
|||
reactor.connectTCP("localhost", settings.getint('CLUSTER_WORKER_PORT'), factory)
|
||||
d = factory.getRootObject()
|
||||
def _set_worker(obj):
|
||||
self.worker = Broker(self, obj)
|
||||
self.worker = ClusterCrawlerBroker(self, obj)
|
||||
d.addCallbacks(callback=_set_worker, errback=lambda reason: log.msg(reason, log.ERROR))
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
import sys, datetime
|
||||
import pickle
|
||||
from __future__ import with_statement
|
||||
|
||||
import datetime
|
||||
import cPickle as pickle
|
||||
|
||||
from pydispatch import dispatcher
|
||||
|
||||
from twisted.spread import pb
|
||||
from twisted.internet import reactor
|
||||
|
||||
|
|
@ -21,7 +22,8 @@ def my_import(name):
|
|||
mod = getattr(mod, comp)
|
||||
return mod
|
||||
|
||||
class Broker(pb.Referenceable):
|
||||
class ClusterMasterBroker(pb.Referenceable):
|
||||
|
||||
def __init__(self, remote, name, master):
|
||||
self.__remote = remote
|
||||
self.alive = False
|
||||
|
|
@ -42,7 +44,7 @@ class Broker(pb.Referenceable):
|
|||
status = {"alive": self.alive}
|
||||
if self.alive:
|
||||
if verbosity == 1:
|
||||
#dont show spider settings
|
||||
# dont show spider settings
|
||||
status["running"] = []
|
||||
for proc in self.running:
|
||||
proccopy = proc.copy()
|
||||
|
|
@ -143,6 +145,9 @@ class Broker(pb.Referenceable):
|
|||
self.master.statistics["domains"]["lost"].remove(domain)
|
||||
|
||||
class ScrapyPBClientFactory(pb.PBClientFactory):
|
||||
|
||||
noisy = False
|
||||
|
||||
def __init__(self, master, nodename):
|
||||
pb.PBClientFactory.__init__(self)
|
||||
self.master = master
|
||||
|
|
@ -151,32 +156,35 @@ class ScrapyPBClientFactory(pb.PBClientFactory):
|
|||
def clientConnectionLost(self, *args, **kargs):
|
||||
pb.PBClientFactory.clientConnectionLost(self, *args, **kargs)
|
||||
del self.master.nodes[self.nodename]
|
||||
log.msg("Removed node %s." % self.nodename )
|
||||
log.msg("Lost connection to %s. Node removed" % self.nodename )
|
||||
|
||||
class ClusterMaster:
|
||||
class ClusterMaster(object):
|
||||
|
||||
def __init__(self):
|
||||
|
||||
if not (settings.getbool('CLUSTER_MASTER_ENABLED')):
|
||||
if not settings.getbool('CLUSTER_MASTER_ENABLED'):
|
||||
raise NotConfigured
|
||||
if not settings['CLUSTER_MASTER_STATEFILE']:
|
||||
raise NotConfigured("ClusterMaster: Missing CLUSTER_MASTER_STATEFILE setting")
|
||||
|
||||
#import groups settings
|
||||
# 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
|
||||
# load pending domains
|
||||
try:
|
||||
self.pending = pickle.load( open(settings["CLUSTER_MASTER_CACHEFILE"], "r") )
|
||||
statefile = open(settings["CLUSTER_MASTER_STATEFILE"], "r")
|
||||
self.pending = pickle.load(statefile)
|
||||
except IOError:
|
||||
self.pending = []
|
||||
self.loading = []
|
||||
self.nodes = {}
|
||||
self.start_time = datetime.datetime.utcnow()
|
||||
#on how statistics works, see self.update_nodes() and Broker.remote_update()
|
||||
# for more info about statistics see self.update_nodes() and ClusterMasterBroker.remote_update()
|
||||
self.statistics = {"domains": {"running": set(), "scraped": {}, "lost_count": {}, "lost": set()}, "scraped_count": 0 }
|
||||
self.global_settings = {}
|
||||
#load cluster global settings
|
||||
# load cluster global settings
|
||||
for sname in settings.getlist('GLOBAL_CLUSTER_SETTINGS'):
|
||||
self.global_settings[sname] = settings[sname]
|
||||
|
||||
|
|
@ -185,20 +193,12 @@ class ClusterMaster:
|
|||
|
||||
def load_nodes(self):
|
||||
"""Loads nodes listed in CLUSTER_MASTER_NODES setting"""
|
||||
for name, url in settings.get('CLUSTER_MASTER_NODES', {}).iteritems():
|
||||
self.load_node(name, url)
|
||||
for name, hostport in settings.get('CLUSTER_MASTER_NODES', {}).iteritems():
|
||||
self.load_node(name, hostport)
|
||||
|
||||
def load_node(self, name, url):
|
||||
"""Creates the remote reference for each worker node"""
|
||||
def _make_callback(_factory, _name, _url):
|
||||
|
||||
def _errback(_reason):
|
||||
log.msg("Could not get remote node %s in %s: %s." % (_name, _url, _reason), log.ERROR)
|
||||
|
||||
d = _factory.getRootObject()
|
||||
d.addCallbacks(callback=lambda obj: self.add_node(obj, _name), errback=_errback)
|
||||
|
||||
server, port = url.split(":")
|
||||
def load_node(self, name, hostport):
|
||||
"""Creates the remote reference for a worker node"""
|
||||
server, port = hostport.split(":")
|
||||
port = int(port)
|
||||
log.msg("Connecting to cluster worker %s..." % name)
|
||||
log.msg("Server: %s, Port: %s" % (server, port))
|
||||
|
|
@ -206,18 +206,23 @@ class ClusterMaster:
|
|||
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)
|
||||
log.msg("Could not connect to node %s in %s: %s." % (name, hostport, err), log.ERROR)
|
||||
else:
|
||||
_make_callback(factory, name, url)
|
||||
def _errback(_reason):
|
||||
log.msg("Could not connect to remote node %s (%s): %s." % (name, hostport, _reason), log.ERROR)
|
||||
|
||||
d = factory.getRootObject()
|
||||
d.addCallbacks(callback=lambda obj: self.add_node(obj, name), errback=_errback)
|
||||
|
||||
def update_nodes(self):
|
||||
for name, url in settings.get('CLUSTER_MASTER_NODES', {}).iteritems():
|
||||
"""Update worker nodes statistics"""
|
||||
for name, hostport 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) )
|
||||
log.msg("Updating node. name: %s, host: %s" % (name, hostport) )
|
||||
self.nodes[name].update_status()
|
||||
else:
|
||||
log.msg("Reloading node. name: %s, url: %s" % (name, url) )
|
||||
self.load_node(name, url)
|
||||
log.msg("Reloading node. name: %s, host: %s" % (name, hostport) )
|
||||
self.load_node(name, hostport)
|
||||
|
||||
real_running = set(self.running.keys())
|
||||
lost = self.statistics["domains"]["running"].difference(real_running)
|
||||
|
|
@ -227,7 +232,7 @@ class ClusterMaster:
|
|||
|
||||
def add_node(self, cworker, name):
|
||||
"""Add node given its node"""
|
||||
node = Broker(cworker, name, self)
|
||||
node = ClusterMasterBroker(cworker, name, self)
|
||||
self.nodes[name] = node
|
||||
log.msg("Added cluster worker %s" % name)
|
||||
|
||||
|
|
@ -241,6 +246,7 @@ class ClusterMaster:
|
|||
raise NotImplemented
|
||||
|
||||
def schedule(self, domains, spider_settings=None, priority=DEFAULT_PRIORITY):
|
||||
"""Schedule the domains passed"""
|
||||
i = 0
|
||||
for p in self.pending:
|
||||
if p['priority'] <= priority:
|
||||
|
|
@ -261,6 +267,7 @@ class ClusterMaster:
|
|||
self.pending.insert(i, {'domain': domain, 'settings': final_spider_settings, 'priority': priority})
|
||||
|
||||
def stop(self, domains):
|
||||
"""Stop the given domains"""
|
||||
to_stop = {}
|
||||
for domain in domains:
|
||||
node = self.running.get(domain, None)
|
||||
|
|
@ -325,6 +332,8 @@ class ClusterMaster:
|
|||
def _engine_started(self):
|
||||
self.load_nodes()
|
||||
scrapyengine.addtask(self.update_nodes, settings.getint('CLUSTER_MASTER_POLL_INTERVAL'))
|
||||
|
||||
def _engine_stopped(self):
|
||||
pickle.dump( self.pending, open(settings["CLUSTER_MASTER_CACHEFILE"], "w") )
|
||||
log.msg("Pending saved in %s" % settings["CLUSTER_MASTER_CACHEFILE"])
|
||||
with open(settings["CLUSTER_MASTER_STATEFILE"], "w") as f:
|
||||
pickle.dump(self.pending, f)
|
||||
log.msg("Cluster master state saved in %s" % settings["CLUSTER_MASTER_STATEFILE"])
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ class ClusterMasterWeb(ClusterMaster):
|
|||
domains = args["schedule"]
|
||||
priority = int(args.get("priority", [DEFAULT_PRIORITY])[0])
|
||||
|
||||
#spider settings
|
||||
# spider settings
|
||||
slist = args.get("settings", [""])[0].split(sep)
|
||||
spider_settings = {}
|
||||
for s in slist:
|
||||
|
|
@ -167,7 +167,7 @@ class ClusterMasterWeb(ClusterMaster):
|
|||
s += "<input type='text' name='priority'>%s</input>" % DEFAULT_PRIORITY
|
||||
s += "<br />\n"
|
||||
|
||||
#spider settings
|
||||
# spider settings
|
||||
s += "Overrided spider settings:<br />\n"
|
||||
s += "<textarea name='settings' rows='4'>\n"
|
||||
s += "UNAVAILABLES_NOTIFY=2\n"
|
||||
|
|
@ -236,4 +236,4 @@ class ClusterMasterWeb(ClusterMaster):
|
|||
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
|
||||
return content
|
||||
|
|
|
|||
|
|
@ -1,4 +1,8 @@
|
|||
import sys, os, time, datetime, pickle
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import datetime
|
||||
import cPickle as pickle
|
||||
|
||||
from twisted.internet import protocol, reactor
|
||||
from twisted.spread import pb
|
||||
|
|
@ -9,6 +13,7 @@ from scrapy.conf import settings
|
|||
from scrapy.core.engine import scrapyengine
|
||||
|
||||
class ScrapyProcessProtocol(protocol.ProcessProtocol):
|
||||
|
||||
def __init__(self, procman, domain, logfile=None, spider_settings=None):
|
||||
self.procman = procman
|
||||
self.domain = domain
|
||||
|
|
@ -17,18 +22,46 @@ class ScrapyProcessProtocol(protocol.ProcessProtocol):
|
|||
self.status = "starting"
|
||||
self.pid = -1
|
||||
self.env = {}
|
||||
#We conserve original setting format for info purposes (avoid lots of unnecesary "SCRAPY_")
|
||||
# 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})
|
||||
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
|
||||
self.env["PYTHONPATH"] = ":".join(sys.path)#this is need so this crawl process knows where to locate local_scrapy_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 "<ScrapyProcess domain=%s, pid=%s, status=%s>" % (self.domain, self.pid, self.status)
|
||||
|
||||
def as_dict(self):
|
||||
return {"domain": self.domain, "pid": self.pid, "status": self.status, "settings": self.scrapy_settings, "logfile": self.logfile, "starttime": self.start_time}
|
||||
def status(self):
|
||||
"""Return this scrapy process status 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
|
||||
|
|
@ -52,22 +85,43 @@ class ClusterWorker(pb.Root):
|
|||
|
||||
self.maxproc = settings.getint('CLUSTER_WORKER_MAXPROC')
|
||||
self.logdir = settings['CLUSTER_LOGDIR']
|
||||
self.running = {}#a dict domain->ScrapyProcessControl
|
||||
self.crawlers = {}#a dict pid->scrapy process remote pb connection
|
||||
self.running = {} # dict of domain->ScrapyProcessControl
|
||||
self.crawlers = {} # dict of pid->scrapy process remote pb connection
|
||||
self.starttime = datetime.datetime.utcnow()
|
||||
port = settings.getint('CLUSTER_WORKER_PORT')
|
||||
scrapyengine.listenTCP(port, pb.PBServerFactory(self))
|
||||
log.msg("PYTHONPATH: %s" % repr(sys.path))
|
||||
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].as_dict() for k in self.running.keys() ]
|
||||
status["running"] = [self.running[k].status() 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, "Status Response.")
|
||||
status["callresponse"] = (rcode, rstring) if rstring else (0, "No request")
|
||||
return status
|
||||
|
||||
def update_master(self, domain, domain_status):
|
||||
|
|
@ -75,16 +129,17 @@ class ClusterWorker(pb.Root):
|
|||
deferred = self.__master.callRemote("update", self.status(), domain, domain_status)
|
||||
except pb.DeadReferenceError:
|
||||
self.__master = None
|
||||
log.msg("Lost connection to node %s." % (self.name), log.ERROR)
|
||||
log.msg("Lost connection to master", log.ERROR)
|
||||
else:
|
||||
deferred.addCallbacks(callback=lambda x: x, errback=lambda reason: log.msg(reason, log.ERROR))
|
||||
|
||||
def remote_set_master(self, master):
|
||||
"""Set the master for this worker"""
|
||||
self.__master = master
|
||||
return self.status()
|
||||
|
||||
def remote_stop(self, domain):
|
||||
"""Stop running 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))
|
||||
|
|
@ -97,10 +152,12 @@ class ClusterWorker(pb.Root):
|
|||
return self.status(1, "%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):
|
||||
"""Spawn process to run the given domain."""
|
||||
"""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"))
|
||||
|
|
@ -118,10 +175,13 @@ class ClusterWorker(pb.Root):
|
|||
log.msg("Unable to svn update: %s" % e, level=log.WARNING)
|
||||
except ImportError:
|
||||
log.msg("pysvn module not available.", level=log.WARNING)
|
||||
proc = reactor.spawnProcess(scrapy_proc, sys.executable, args=args, env=scrapy_proc.env)
|
||||
reactor.spawnProcess(scrapy_proc, sys.executable, args=args, env=scrapy_proc.env)
|
||||
return self.status(0, "Started process %s." % scrapy_proc)
|
||||
return self.status(2, "Domain %s already running." % domain )
|
||||
return self.status(1, "No free slot to run another process.")
|
||||
else:
|
||||
return self.status(2, "Domain %s already running." % domain )
|
||||
else:
|
||||
return self.status(1, "No free slot to run another process.")
|
||||
|
||||
def remote_register_crawler(self, pid, crawler):
|
||||
"""Register the crawler to the list of crawlers managed by this worker"""
|
||||
self.crawlers[pid] = crawler
|
||||
|
|
|
|||
Loading…
Reference in New Issue