Added cluster statistics report

--HG--
extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%4091

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

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

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
 

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:

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

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