From df54ed004193a9e85607ff48bb198644f9b4e3df Mon Sep 17 00:00:00 2001 From: Pablo Hoffman Date: Tue, 30 Nov 2010 02:26:31 -0200 Subject: [PATCH] Some Scrapyd enhancements: * added minimal web ui * return unique id per job (spider scheduled) * store one log per spider run (job) and rotate them, keeping the last N logs (where N is configurable through settings) --- docs/topics/scrapyd.rst | 40 +++++++++++---- scrapyd/app.py | 2 +- scrapyd/default_scrapyd.conf | 1 + scrapyd/environ.py | 13 ++++- scrapyd/interfaces.py | 4 +- scrapyd/launcher.py | 28 +++++++---- scrapyd/tests/test_environ.py | 8 +-- scrapyd/tests/test_utils.py | 2 +- scrapyd/utils.py | 2 +- scrapyd/webservice.py | 45 ++++------------- scrapyd/website.py | 94 +++++++++++++++++++++++++++++++++++ 11 files changed, 175 insertions(+), 64 deletions(-) create mode 100644 scrapyd/website.py diff --git a/docs/topics/scrapyd.rst b/docs/topics/scrapyd.rst index 4b9e12fb8..331b41738 100644 --- a/docs/topics/scrapyd.rst +++ b/docs/topics/scrapyd.rst @@ -27,17 +27,15 @@ How Scrapyd works ================= Scrapyd is an application (typically run as a daemon) that continually polls -for spiders that need to run (ie. those projects that have spiders enqueued). +for spiders that need to run. When a spider needs to run, a process is started to crawl the spider:: scrapy crawl my spider Scrapyd also runs multiple processes in parallel, allocating them in a fixed -number of "slots", which defaults to the number of cpu processors available in -the system, but this can be changed with the ``max_proc`` and -``max_proc_per_cpu`` options. It also starts as many processes as possible to -handle the load. +number of slots given by the `max_proc`_ and `max_proc_per_cpu`_ options, +starting as many processes as possible to handle the load. In addition to dispatching and managing processes, Scrapyd provides a :ref:`JSON web service ` to upload new project versions @@ -46,6 +44,9 @@ you want to implement your own custom Scrapyd. The components are pluggable and can be changed, if you're familiar with the `Twisted Application Framework`_ which Scrapyd is implemented in. +Starting from 0.11, Scrapyd also provides a minimal :ref:`web interface +`. + Starting Scrapyd ================ @@ -109,11 +110,15 @@ The standard error captured from Scrapyd and any sub-process spawned from it. Remember to check this file if you're having problems, as the errors may not get logged to the ``scrapyd.log`` file. -/var/log/scrapyd/slotN.log -~~~~~~~~~~~~~~~~~~~~~~~~~~ +/var/log/scrapyd/project +~~~~~~~~~~~~~~~~~~~~~~~~ -The log files of Scrapy processes started from Scrapyd, one per slot. These are -standard :ref:`Scrapy log files `. +Besides the main service log file, Scrapyd stores one log file per crawling +process in:: + + /var/log/scrapyd/PROJECT/SPIDER/ID.log + +Where ``ID`` is a unique id for the run. /var/lib/scrapyd/ ~~~~~~~~~~~~~~~~~ @@ -175,7 +180,12 @@ spider queues). logs_dir -------- -The directory where the Scrapy processes logs (``slotN.log``) will be stored. +The directory where the Scrapy processes logs will be stored. + +logs_to_keep +------------ + +The number of logs to keep per spider. Defaults to ``5``. egg_runner ---------- @@ -346,6 +356,16 @@ To schedule a spider run:: For more resources see: :ref:`topics-scrapyd-jsonapi` for more available resources. +.. _topics-scrapyd-webui: + +Web Interface +============= + +.. versionadded:: 0.11 + +Scrapyd comes with a minimal web interface (for monitoring running processes +and accessing logs) which can be accessed at http://localhost:6800/ + .. _topics-scrapyd-jsonapi: JSON API reference diff --git a/scrapyd/app.py b/scrapyd/app.py index 466b4d44f..2ce777c1a 100644 --- a/scrapyd/app.py +++ b/scrapyd/app.py @@ -8,7 +8,7 @@ from .eggstorage import FilesystemEggStorage from .scheduler import SpiderScheduler from .poller import QueuePoller from .environ import Environment -from .webservice import Root +from .website import Root from .config import Config def application(): diff --git a/scrapyd/default_scrapyd.conf b/scrapyd/default_scrapyd.conf index 69ad18aa5..b73dcf4be 100644 --- a/scrapyd/default_scrapyd.conf +++ b/scrapyd/default_scrapyd.conf @@ -1,6 +1,7 @@ [scrapyd] eggs_dir = eggs logs_dir = logs +logs_to_keep = 5 dbs_dir = dbs max_proc = 0 max_proc_per_cpu = 4 diff --git a/scrapyd/environ.py b/scrapyd/environ.py index 211377e38..70bef2974 100644 --- a/scrapyd/environ.py +++ b/scrapyd/environ.py @@ -11,6 +11,7 @@ class Environment(object): def __init__(self, config, initenv=os.environ): self.dbs_dir = config.get('dbs_dir', 'dbs') self.logs_dir = config.get('logs_dir', 'logs') + self.logs_to_keep = config.getint('logs_to_keep', 5) if config.cp.has_section('settings'): self.settings = dict(config.cp.items('settings')) else: @@ -27,7 +28,15 @@ class Environment(object): env['SCRAPY_SETTINGS_MODULE'] = self.settings[project] dbpath = os.path.join(self.dbs_dir, '%s.db' % project) env['SCRAPY_SQLITE_DB'] = dbpath - logpath = os.path.join(self.logs_dir, 'slot%s.log' % slot) - env['SCRAPY_LOG_FILE'] = logpath + env['SCRAPY_LOG_FILE'] = self._get_log_file(message) return env + def _get_log_file(self, message): + logsdir = os.path.join(self.logs_dir, message['project'], \ + message['spider']) + if not os.path.exists(logsdir): + os.makedirs(logsdir) + to_delete = sorted(os.listdir(logsdir), reverse=True)[:-self.logs_to_keep] + for x in to_delete: + os.remove(os.path.join(logsdir, x)) + return os.path.join(logsdir, "%s.log" % message['_id']) diff --git a/scrapyd/interfaces.py b/scrapyd/interfaces.py index 30a27e637..c9c6f839e 100644 --- a/scrapyd/interfaces.py +++ b/scrapyd/interfaces.py @@ -65,7 +65,9 @@ class IEnvironment(Interface): def get_environment(message, slot, eggpath): """Return the environment variables to use for running the process. - `message` is the message received from the IPoller.next() + `message` is the message received from the IPoller.next() augmented to + contain (at least) the following keys: project, spider, _id (where + _id is a unique identifier for this run) `slot` is the Launcher slot where the process will be running. `eggpath` is the path to an eggfile that contains the project code. The `eggpath` may be `None` if no egg was found for the project, in diff --git a/scrapyd/launcher.py b/scrapyd/launcher.py index 845a35f8b..a2eff56b9 100644 --- a/scrapyd/launcher.py +++ b/scrapyd/launcher.py @@ -1,22 +1,26 @@ import sys, os from shutil import copyfileobj from tempfile import mkstemp +from datetime import datetime from twisted.internet import reactor, defer, protocol, error from twisted.application.service import Service from twisted.python import log from scrapy.utils.py26 import cpu_count -from scrapy.utils.python import unicode_to_str +from scrapy.utils.python import stringify_dict from scrapyd.utils import get_crawl_args from .interfaces import IPoller, IEggStorage, IEnvironment class Launcher(Service): + name = 'launcher' + def __init__(self, config, app): + self.processes = {} self.max_proc = config.getint('max_proc', 0) if not self.max_proc: - self.max_proc = cpu_count() * config.getint('max_proc_per_cpu') + self.max_proc = cpu_count() * config.getint('max_proc_per_cpu', 4) self.egg_runner = config.get('egg_runner', 'scrapyd.eggrunner') self.app = app @@ -43,31 +47,35 @@ class Launcher(Service): return eggpath def _spawn_process(self, message, slot): - project = unicode_to_str(message['project']) - spider = unicode_to_str(message['spider']) + msg = stringify_dict(message, keys_only=False) + project = msg['project'] eggpath = self._get_eggpath(project) args = [sys.executable, '-m', self.egg_runner, 'crawl'] - args += get_crawl_args(message) + args += get_crawl_args(msg) e = self.app.getComponent(IEnvironment) - env = e.get_environment(message, slot, eggpath) - pp = ScrapyProcessProtocol(eggpath, slot, project, spider) + env = e.get_environment(msg, slot, eggpath) + pp = ScrapyProcessProtocol(eggpath, slot, project, msg['spider'], msg['_id']) pp.deferred.addBoth(self._process_finished, eggpath, slot) reactor.spawnProcess(pp, sys.executable, args=args, env=env) + self.processes[slot] = pp def _process_finished(self, _, eggpath, slot): if eggpath: os.remove(eggpath) + self.processes.pop(slot) self._wait_for_project(slot) class ScrapyProcessProtocol(protocol.ProcessProtocol): - def __init__(self, eggfile, slot, project, spider): + def __init__(self, eggfile, slot, project, spider, job): self.eggfile = eggfile self.slot = slot self.pid = None self.project = project self.spider = spider + self.job = job + self.start_time = datetime.now() self.deferred = defer.Deferred() def outReceived(self, data): @@ -88,6 +96,6 @@ class ScrapyProcessProtocol(protocol.ProcessProtocol): self.deferred.callback(self) def log(self, msg): - msg += "project=%r spider=%r slot=%r pid=%r egg=%r" % (self.project, \ - self.spider, self.slot, self.pid, self.eggfile) + msg += "project=%r spider=%r job=%r pid=%r egg=%r" % (self.project, \ + self.spider, self.job, self.pid, self.eggfile) log.msg(msg, system="Launcher") diff --git a/scrapyd/tests/test_environ.py b/scrapyd/tests/test_environ.py index a1a2e2e75..2133a5d5b 100644 --- a/scrapyd/tests/test_environ.py +++ b/scrapyd/tests/test_environ.py @@ -22,21 +22,21 @@ class EggStorageTest(unittest.TestCase): verifyObject(IEnvironment, self.environ) def test_get_environment_with_eggfile(self): - msg = {'project': 'mybot'} + msg = {'project': 'mybot', 'spider': 'myspider', '_id': 'ID'} slot = 3 env = self.environ.get_environment(msg, slot, '/path/to/file.egg') self.assertEqual(env['SCRAPY_PROJECT'], 'mybot') self.assert_(env['SCRAPY_SQLITE_DB'].endswith('mybot.db')) - self.assert_(env['SCRAPY_LOG_FILE'].endswith('slot3.log')) + self.assert_(env['SCRAPY_LOG_FILE'].endswith('/mybot/myspider/ID.log')) self.assert_(env['SCRAPY_EGGFILE'].endswith('/path/to/file.egg')) self.failIf('SCRAPY_SETTINGS_MODULE' in env) def test_get_environment_without_eggfile(self): - msg = {'project': 'newbot'} + msg = {'project': 'newbot', 'spider': 'myspider', '_id': 'ID'} slot = 3 env = self.environ.get_environment(msg, slot, None) self.assertEqual(env['SCRAPY_PROJECT'], 'newbot') self.assert_(env['SCRAPY_SQLITE_DB'].endswith('newbot.db')) - self.assert_(env['SCRAPY_LOG_FILE'].endswith('slot3.log')) + self.assert_(env['SCRAPY_LOG_FILE'].endswith('/newbot/myspider/ID.log')) self.assertEqual(env['SCRAPY_SETTINGS_MODULE'], 'newbot.settings') self.failIf('SCRAPY_EGGFILE' in env) diff --git a/scrapyd/tests/test_utils.py b/scrapyd/tests/test_utils.py index 68aed0053..8bc174100 100644 --- a/scrapyd/tests/test_utils.py +++ b/scrapyd/tests/test_utils.py @@ -7,7 +7,7 @@ class UtilsTest(unittest.TestCase): def test_get_crawl_args(self): msg = {'project': 'lolo', 'spider': 'lala'} self.assertEqual(get_crawl_args(msg), ['lala']) - msg = {'project': 'lolo', 'spider': 'lala', 'arg1': 'val1'} + msg = {'project': 'lolo', 'spider': 'lala', 'arg1': u'val1'} cargs = get_crawl_args(msg) self.assertEqual(cargs, ['lala', '-a', 'arg1=val1']) assert all(isinstance(x, str) for x in cargs), cargs diff --git a/scrapyd/utils.py b/scrapyd/utils.py index 6adf1faba..0971f6f0d 100644 --- a/scrapyd/utils.py +++ b/scrapyd/utils.py @@ -37,7 +37,7 @@ def get_crawl_args(message): msg = message.copy() args = [unicode_to_str(msg['spider'])] del msg['project'], msg['spider'] - for k, v in stringify_dict(msg).items(): + for k, v in stringify_dict(msg, keys_only=False).items(): args += ['-a'] args += ['%s=%s' % (k, v)] return args diff --git a/scrapyd/webservice.py b/scrapyd/webservice.py index e66bb3691..6c6fa5b73 100644 --- a/scrapyd/webservice.py +++ b/scrapyd/webservice.py @@ -1,11 +1,9 @@ import cgi import traceback +import uuid from cStringIO import StringIO -from twisted.web.resource import Resource - from scrapy.utils.txweb import JsonResource -from .interfaces import IPoller, IEggStorage, ISpiderScheduler from .eggutils import get_spider_list_from_eggfile class WsResource(JsonResource): @@ -29,9 +27,11 @@ class Schedule(WsResource): args = dict((k, v[0]) for k, v in txrequest.args.items()) project = args.pop('project') spider = args.pop('spider') - sched = self.root.app.getComponent(ISpiderScheduler) - sched.schedule(project, spider, **args) - return {"status": "ok"} + jobid = uuid.uuid1().hex + args['_id'] = jobid + self.root.scheduler.schedule(project, spider, **args) + jobids = {spider: jobid} + return {"status": "ok", "jobs": jobids} class AddVersion(WsResource): @@ -43,8 +43,7 @@ class AddVersion(WsResource): version = d['version'][0] eggf = StringIO(d['egg'][0]) spiders = get_spider_list_from_eggfile(eggf, project) - eggstorage = self.root.app.getComponent(IEggStorage) - eggstorage.put(eggf, project, version) + self.root.eggstorage.put(eggf, project, version) self.root.update_projects() return {"status": "ok", "project": project, "version": version, \ "spiders": len(spiders)} @@ -52,23 +51,21 @@ class AddVersion(WsResource): class ListProjects(WsResource): def render_GET(self, txrequest): - projects = self.root.app.getComponent(ISpiderScheduler).list_projects() + projects = self.root.scheduler.list_projects() return {"status": "ok", "projects": projects} class ListVersions(WsResource): def render_GET(self, txrequest): project = txrequest.args['project'][0] - eggstorage = self.root.app.getComponent(IEggStorage) - versions = eggstorage.list(project) + versions = self.root.eggstorage.list(project) return {"status": "ok", "versions": versions} class ListSpiders(WsResource): def render_GET(self, txrequest): project = txrequest.args['project'][0] - eggstorage = self.root.app.getComponent(IEggStorage) - _, eggf = eggstorage.get(project) + _, eggf = self.root.eggstorage.get(project) spiders = get_spider_list_from_eggfile(eggf, project, \ eggrunner=self.root.egg_runner) return {"status": "ok", "spiders": spiders} @@ -81,8 +78,7 @@ class DeleteProject(WsResource): return {"status": "ok"} def _delete_version(self, project, version=None): - eggstorage = self.root.app.getComponent(IEggStorage) - eggstorage.delete(project, version) + self.root.eggstorage.delete(project, version) self.root.update_projects() class DeleteVersion(DeleteProject): @@ -93,22 +89,3 @@ class DeleteVersion(DeleteProject): self._delete_version(project, version) return {"status": "ok"} -class Root(Resource): - - def __init__(self, config, app): - Resource.__init__(self) - self.debug = config.getboolean('debug', False) - self.eggrunner = config.get('egg_runner') - self.app = app - self.putChild('schedule.json', Schedule(self)) - self.putChild('addversion.json', AddVersion(self)) - self.putChild('listprojects.json', ListProjects(self)) - self.putChild('listversions.json', ListVersions(self)) - self.putChild('listspiders.json', ListSpiders(self)) - self.putChild('delproject.json', DeleteProject(self)) - self.putChild('delversion.json', DeleteVersion(self)) - self.update_projects() - - def update_projects(self): - self.app.getComponent(IPoller).update_projects() - self.app.getComponent(ISpiderScheduler).update_projects() diff --git a/scrapyd/website.py b/scrapyd/website.py new file mode 100644 index 000000000..c53866832 --- /dev/null +++ b/scrapyd/website.py @@ -0,0 +1,94 @@ +from datetime import datetime + +from twisted.web import resource, static +from twisted.application.service import IServiceCollection +from .interfaces import IPoller, IEggStorage, ISpiderScheduler + +from . import webservice + +class Root(resource.Resource): + + def __init__(self, config, app): + resource.Resource.__init__(self) + self.debug = config.getboolean('debug', False) + self.eggrunner = config.get('egg_runner') + logsdir = config.get('logs_dir') + self.app = app + self.putChild('', Home()) + self.putChild('schedule.json', webservice.Schedule(self)) + self.putChild('addversion.json', webservice.AddVersion(self)) + self.putChild('listprojects.json', webservice.ListProjects(self)) + self.putChild('listversions.json', webservice.ListVersions(self)) + self.putChild('listspiders.json', webservice.ListSpiders(self)) + self.putChild('delproject.json', webservice.DeleteProject(self)) + self.putChild('delversion.json', webservice.DeleteVersion(self)) + self.putChild('logs', static.File(logsdir, 'text/plain')) + self.putChild('procmon', ProcessMonitor(self)) + self.update_projects() + + def update_projects(self): + self.poller.update_projects() + self.scheduler.update_projects() + + @property + def launcher(self): + app = IServiceCollection(self.app, self.app) + return app.getServiceNamed('launcher') + + @property + def scheduler(self): + return self.app.getComponent(ISpiderScheduler) + + @property + def eggstorage(self): + return self.app.getComponent(IEggStorage) + + @property + def poller(self): + return self.app.getComponent(IPoller) + + +class Home(resource.Resource): + + def render_GET(self, txrequest): + return """ + +Scrapyd + +

Scrapyd

+ + + +""" + + +class ProcessMonitor(resource.Resource): + + def __init__(self, root): + resource.Resource.__init__(self) + self.root = root + + def render(self, txrequest): + s = "Scrapyd" + s += "" + s += "

Process monitor

" + s += "

Go back

" + s += "" + s += "" + s += "" + s += "" + for p in self.root.launcher.processes.values(): + s += "" + for a in ['project', 'spider', 'job', 'pid']: + s += "" % getattr(p, a) + s += "" % (datetime.now() - p.start_time) + s += "" + s += "
ProjectSpiderJobPIDRuntime
%s%s
" + s += "" + s += "" + return s +