From 7f62faea5fa851d859c4c0bd85e61a29d91aa77b Mon Sep 17 00:00:00 2001 From: olveyra Date: Tue, 22 Jul 2008 17:21:55 +0000 Subject: [PATCH] Added cluster statistics report --HG-- extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%4091 --- .../contrib/pbcluster/master/manager.py | 32 +++++++++++++++---- .../scrapy/contrib/pbcluster/master/web.py | 8 +++++ .../contrib/pbcluster/master/ws_api.txt | 7 +++- .../pbcluster/tools/scrapy-cluster-ctl.py | 3 ++ .../contrib/pbcluster/worker/manager.py | 17 ++++++++-- .../contrib/pbcluster/worker/testworker.py | 2 ++ 6 files changed, 60 insertions(+), 9 deletions(-) diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py b/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py index 0866d3ac7..dcb94f75d 100644 --- a/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py +++ b/scrapy/trunk/scrapy/contrib/pbcluster/master/manager.py @@ -1,4 +1,4 @@ -import sys +import sys, datetime import pickle from pydispatch import dispatcher @@ -34,6 +34,7 @@ class Node: self.name = name self.master = master self.available = True + self.statistics = {"domains": {}, "scraped_total": 0 } def status_as_dict(self, verbosity=0): status = {"alive": self.alive} @@ -82,7 +83,7 @@ class Node: self.run(pending) self.master.loading.append(pending['domain']) - def get_status(self): + def update_status(self): try: deferred = self.__remote.callRemote("status") except pb.DeadReferenceError: @@ -91,6 +92,19 @@ class Node: else: deferred.addCallbacks(callback=self._set_status, errback=lambda reason: log.msg(reason, log.ERROR)) + def update_statistics(self): + + def _set_statistics(statistics): + self.statistics = statistics + + try: + deferred = self.__remote.callRemote("statistics") + except pb.DeadReferenceError: + self._set_status(None) + log.msg("Lost connection to node %s." % (self.name), log.ERROR) + else: + deferred.addCallbacks(callback=_set_statistics, errback=lambda reason: log.msg(reason, log.ERROR)) + def stop(self, domain): try: deferred = self.__remote.callRemote("stop", domain) @@ -127,6 +141,7 @@ class Node: log.msg("Lost connection to node %s." % (self.name), log.ERROR) else: deferred.addCallbacks(callback=_run_callback, errback=_run_errback) + class ClusterMaster(pb.Root): @@ -147,7 +162,7 @@ class ClusterMaster(pb.Root): self.pending = [] self.loading = [] self.nodes = {} - + self.statistics = {"domains": {}, "scraped_total": 0, "start_time": datetime.datetime.utcnow(), "pending": self.print_pending() } self.global_settings = {} #load cluster global settings for sname in settings.getlist('GLOBAL_CLUSTER_SETTINGS'): @@ -188,10 +203,16 @@ class ClusterMaster(pb.Root): _make_callback(factory, name, url) def update_nodes(self): + self.statistics["scraped_total"] = 0 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() + self.nodes[name].update_status() + self.nodes[name].update_statistics() + self.statistics["scraped_total"] += self.nodes[name].statistics["scraped_total"] + for domain in self.nodes[name].statistics["domains"]: + if not domain in self.statistics["domains"] or self.nodes[name].statistics["domains"][domain]["last_start_time"] >= self.statistics["domains"][domain]["last_start_time"]: + self.statistics["domains"][domain] = self.nodes[name].statistics["domains"][domain] else: log.msg("Reloading node. name: %s, url: %s" % (name, url) ) self.load_node(name, url) @@ -199,7 +220,7 @@ class ClusterMaster(pb.Root): def add_node(self, cworker, name): """Add node given its node""" node = Node(cworker, None, name, self) - node.get_status() + node.update_status() self.nodes[name] = node log.msg("Added cluster worker %s" % name) @@ -293,7 +314,6 @@ class ClusterMaster(pb.Root): return pending else: return self.pending - def _engine_started(self): self.load_nodes() diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py b/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py index bbe8a1d38..c96330319 100644 --- a/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py +++ b/scrapy/trunk/scrapy/contrib/pbcluster/master/web.py @@ -103,6 +103,9 @@ class ClusterMasterWeb(ClusterMaster): 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) @@ -231,4 +234,9 @@ class ClusterMasterWeb(ClusterMaster): status["pending"] = self.print_pending(int(verbosity)) status["loading"] = self.loading content = serialize(status, format) + return content + + 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 \ No newline at end of file diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt b/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt index 988f48ee5..52fc95ca4 100644 --- a/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt +++ b/scrapy/trunk/scrapy/contrib/pbcluster/master/ws_api.txt @@ -5,6 +5,8 @@ The webservice API is available at http://server:port/cluster_master/ws/ +With no parameters, webservice returns the cluster status. + Query parameters ================ @@ -30,9 +32,12 @@ Query parameters - `enable_node`: revert the state setted by 'disable_node' -'verbosity': sets the verbosity level (0 is the default minimal, 1 includes domain settings) + - `verbosity`: sets the verbosity level (0 is the default minimal, 1 includes domain settings) + + - `statistics`: shows the pending/running/scraped/lost statistics 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 diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/tools/scrapy-cluster-ctl.py b/scrapy/trunk/scrapy/contrib/pbcluster/tools/scrapy-cluster-ctl.py index 9148f5338..97c4f3776 100755 --- a/scrapy/trunk/scrapy/contrib/pbcluster/tools/scrapy-cluster-ctl.py +++ b/scrapy/trunk/scrapy/contrib/pbcluster/tools/scrapy-cluster-ctl.py @@ -19,6 +19,7 @@ def main(): 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.", default=0) (opts, args) = parser.parse_args() @@ -41,6 +42,8 @@ def main(): if opts.status: pass + elif opts.statistics: + post["statistics"] = True elif opts.schedule and domains: post["schedule"] = domains if opts.now: diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/worker/manager.py b/scrapy/trunk/scrapy/contrib/pbcluster/worker/manager.py index ee3a536ae..aaf62ebfa 100644 --- a/scrapy/trunk/scrapy/contrib/pbcluster/worker/manager.py +++ b/scrapy/trunk/scrapy/contrib/pbcluster/worker/manager.py @@ -35,10 +35,13 @@ class ScrapyProcessProtocol(protocol.ProcessProtocol): log.msg("ClusterWorker: started domain=%s, pid=%d, log=%s" % (self.domain, self.pid, self.logfile)) self.transport.closeStdin() self.status = "running" + self.procman.statistics["domains"][self.domain] = {"status": "running", "last_start_time": self.start_time} def processEnded(self, status_object): log.msg("ClusterWorker: finished domain=%s, pid=%d, log=%s" % (self.domain, self.pid, self.logfile)) del self.procman.running[self.domain] + self.procman.statistics["domains"][self.domain].update({"status": "scraped", "last_end_time": datetime.datetime.utcnow()}) + self.procman.statistics["scraped_total"] += 1 class ClusterWorker(pb.Root): @@ -49,7 +52,8 @@ class ClusterWorker(pb.Root): self.maxproc = settings.getint('CLUSTER_WORKER_MAXPROC') self.logdir = settings['CLUSTER_LOGDIR'] self.running = {} - self.starttime = time.time() + self.starttime = datetime.datetime.utcnow() + self.statistics = {"domains": {}, "scraped_total": 0} port = settings.getint('CLUSTER_WORKER_PORT') scrapyengine.listenTCP(port, pb.PBServerFactory(self)) log.msg("PYTHONPATH: %s" % repr(sys.path)) @@ -68,11 +72,20 @@ class ClusterWorker(pb.Root): def remote_status(self): return self.status() + def remote_statistics(self): + #This can detect processes that were abnormally killed (for example, by the kernel + #because of a memory ran out.) + for domain in self.statistics["domains"]: + if self.statistics["domains"][domain]["status"] == "running" and not domain in self.running: + self.statistics["domains"][domain]["status"] = "lost" + + return self.statistics + def status(self, rcode=0, rstring=None): status = {} status["running"] = [ self.running[k].as_dict() for k in self.running.keys() ] status["starttime"] = self.starttime - status["timestamp"] = time.time() + status["timestamp"] = datetime.datetime.utcnow() status["maxproc"] = self.maxproc status["loadavg"] = os.getloadavg() status["logdir"] = self.logdir diff --git a/scrapy/trunk/scrapy/contrib/pbcluster/worker/testworker.py b/scrapy/trunk/scrapy/contrib/pbcluster/worker/testworker.py index c4498e374..eee998910 100755 --- a/scrapy/trunk/scrapy/contrib/pbcluster/worker/testworker.py +++ b/scrapy/trunk/scrapy/contrib/pbcluster/worker/testworker.py @@ -17,6 +17,8 @@ elif sys.argv[0] == "-s": d.addCallback(lambda object: object.callRemote("stop", sys.argv[1])) elif sys.argv[0] == "-r": d.addCallback(lambda object: object.callRemote("run", sys.argv[1])) +elif sys.argv[0] == "-t": + d.addCallback(lambda object: object.callRemote("statistics")) d.addCallbacks(callback = util.println, errback = lambda reason: 'error: '+str(reason.value)) d.addCallback(lambda _: reactor.stop())