RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

removed obsolete scrapy.contrib.cluster

This commit is contained in:
Pablo Hoffman 2009-08-10 21:02:46 -03:00
parent a0e2086b1b
commit 7ac7597950
16 changed files with 0 additions and 1184 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
docs/ref/settings.rst
View File

@ -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

15
scrapy/conf/default_settings.py
View File

@ -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 = []

3
scrapy/contrib/cluster/__init__.py
View File

@ -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

0
scrapy/contrib/cluster/crawler/__init__.py
View File

41
scrapy/contrib/cluster/crawler/manager.py
View File

@ -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))

13
scrapy/contrib/cluster/hooks/__init__.py
View File

@ -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.
"""

23
scrapy/contrib/cluster/hooks/svn.py
View File

@ -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)

0
scrapy/contrib/cluster/master/__init__.py
View File

390
scrapy/contrib/cluster/master/manager.py
View File

@ -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)

239
scrapy/contrib/cluster/master/web.py
View File

@ -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 += "<h2>Home</h2>\n"
s += "<table border='1'>\n"
s += "<tr><th>&nbsp;</th><th>Name</th><th>Available</th><th>Running</th><th>Load.avg</th></tr>\n"
for node in self.nodes.itervalues():
#chkbox = "<input type='checkbox' name='shutdown' value='%s' />" % domain if node.status in ["up", "idle"] else "&nbsp;"
nodelink = "<a href='nodes/#%s'>%s</a>" % (node.name, node.name)
chkbox = "&nbsp;"
loadavg = "%.2f %.2f %.2f" % node.loadavg
s += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%d/%d</td><td>%s</td></tr>\n" % \
(chkbox, nodelink, node.available, len(node.running), node.maxproc, loadavg)
s += "</table>\n"
s += "</body>\n"
s += "</html>\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 += "<h2><a name='%s'>%s</h2>\n" % (node.name, node.name)
s += "<h3>Running domains</h3>\n"
if node.running:
s += "<form method='post' action='.'>\n"
s += "<table border='1'>\n"
s += "<tr><th>&nbsp;</th><th>PID</th><th>Domain</th><th>Status</th><th>Running time</th><th>Log file</th></tr>\n"
for proc in node.running:
chkbox = "<input type='checkbox' name='stop' value='%s' />" % proc['domain'] if proc['status'] == "running" else "&nbsp;"
start_time = proc.get('starttime', None)
elapsed = now - start_time if start_time else None
s += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>\n" % \
(chkbox, proc['pid'], proc['domain'], proc['status'], elapsed, proc['logfile'])
s += "</table>\n"
s += "<input type='hidden' name='node' value='%s'>\n" % node.name
s += "<p><input type='submit' value='Stop selected domains on %s'></p>\n" % node.name
s += "</form>\n"
else:
s += "<p>No running domains on %s</p>\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 += "<h2>Schedule domains</h2>\n"
s += "Inactive domains (not running or pending)<br />"
s += "<form method='post' action='.'>\n"
s += "<select name='schedule' multiple='multiple' size='10'>\n"
for domain in sorted(inactive_domains):
s += "<option>%s</option>\n" % domain
s += "</select>\n"
s += "<br />\n"
s += "Priority:<br />\n"
s += "<input type='text' name='priority'>%s</input>" % 20
s += "<br />\n"
# spider settings
s += "Overrided spider settings:<br />\n"
s += "<textarea name='settings' rows='4'>\n"
s += "UNAVAILABLES_NOTIFY=2\n"
s += "</textarea>\n"
s += "<br />\n"
s += "<p><input type='submit' value='Schedule selected domains'></p>\n"
s += "</form>\n"
s += "<h2>Domains</h2>\n"
s += "<table border='1'>\n"
s += "<tr><th>Domain</th><th>Status</th><th>Node</th></tr>\n"
s += self._domains_table(self.running, '<b>running</b>')
s += "</table>\n"
# pending domains
s += "<h3>Pending domains</h3>\n"
if self.pending:
s += "<form method='post' action='.'>\n"
s += "<select name='remove' multiple='multiple' size='10'>\n"
for p in self.pending:
s += "<option value='%s'>%s (P:%s)</option>\n" % (p['domain'], p['domain'],p['priority'])
s += "</select>\n"
s += "<p><input type='submit' value='Remove selected pending domains'></p>\n"
s += "</form>\n"
else:
s += "<p>No pending domains</p>\n"
return str(s)
def render_header(self):
s = banner(self)
s += "<p>Nav: "
s += "<a href='/cluster_master/'>Home</a> | "
s += "<a href='/cluster_master/domains/'>Domains</a> | "
s += "<a href='/cluster_master/nodes/'>Nodes</a> (<a href='/cluster_master/nodes/?updatenodes=1'>update</a>)"
s += "</p>"
return s
def _domains_table(self, dict_, status):
s = ""
for domain, node in dict_.iteritems():
s += "<tr><td>%s</td><td>%s</td><td>%s</td></tr>\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

51
scrapy/contrib/cluster/master/ws_api.txt
View File

@ -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 <setting_name>=<value> 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

80
scrapy/contrib/cluster/tools/scrapy-cluster-ctl.py
View File

@ -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()

24
scrapy/contrib/cluster/tools/test-worker.py
View File

@ -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()

0
scrapy/contrib/cluster/worker/__init__.py
View File

208
scrapy/contrib/cluster/worker/manager.py
View File

@ -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 "<ScrapyProcess domain=%s, pid=%s, status=%s>" % (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

5
scrapy/tests/test_engine.py
View File

@ -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