[Pkg-nagios-changes] [SCM] UNNAMED PROJECT branch, debian/master, updated. 810edbdd3feedbfe37f4a65bee50b57b2f60fa2a

Naparuba naparuba at gmail.com
Tue Feb 28 22:13:30 UTC 2012


The following commit has been merged in the debian/master branch:
commit 751791e1a1c138498660b356ace835ec4a2ac722
Author: Naparuba <naparuba at gmail.com>
Date:   Fri Jan 13 11:16:40 2012 +0100

    Add : skonfUI worker launch.

diff --git a/shinken/daemons/skonfdaemon.py b/shinken/daemons/skonfdaemon.py
index 32f6f2d..b0e2f8c 100644
--- a/shinken/daemons/skonfdaemon.py
+++ b/shinken/daemons/skonfdaemon.py
@@ -20,10 +20,19 @@
 #You should have received a copy of the GNU Affero General Public License
 #along with Shinken.  If not, see <http://www.gnu.org/licenses/>.
 
+# Try to see if we are in an android device or not
+is_android = True
+try:
+   import android
+except ImportError:
+   is_android = False
+
+
 import sys
 import os
 import time
 import traceback
+import threading
 from Queue import Empty
 import socket
 
@@ -35,7 +44,7 @@ from shinken.log import logger
 from shinken.brok import Brok
 from shinken.external_command import ExternalCommand
 from shinken.util import safe_print
-
+from shinken.skonfuiworker import SkonfUIWorker
 
 # Now the bottle HTTP part :)
 from shinken.webui.bottle import Bottle, run, static_file, view, route, request, response
@@ -156,6 +165,8 @@ class Skonf(Daemon):
         self.interface = IForArbiter(self)
         self.conf = Config()
 
+        self.workers = {}   # dict of active workers
+
 
     # Use for adding things like broks
     def add(self, b):
@@ -422,6 +433,20 @@ class Skonf(Daemon):
             self.do_daemon_init_and_start()
             self.uri_arb = self.pyro_daemon.register(self.interface, "ForArbiter")
 
+            # Under Android, we do not have multiprocessing lib
+            # so use standard Queue threads things
+            # but in multiprocess, we are also using a Queue(). It's just
+            # not the same
+            if is_android:
+                self.returns_queue = Queue()
+            else:
+                self.returns_queue = self.manager.Queue()
+                
+            # For multiprocess things, we should not have
+            # sockettimeouts. will be set explicitly in Pyro calls
+            import socket
+            socket.setdefaulttimeout(None)
+
             # ok we are now fully daemon (if requested)
             # now we can start our "external" modules (if any) :
             self.modules_manager.start_external_instances()
@@ -566,11 +591,24 @@ class Skonf(Daemon):
 
         # Declare the whole app static files AFTER the plugin ones
         self.declare_common_static()
+        
+        # Start sub workers
+        for i in xrange(1, 3):
+            self.create_and_launch_worker()
+
+        # Launch the data thread"
+        self.workersmanager_thread = threading.Thread(None, self.workersmanager, 'httpthread')
+        self.workersmanager_thread.start()
+        # TODO : look for alive and killing
 
-        print "Starting Skonf application"
+        print "Starting SkonfUI app"
         srv = run(host=self.http_host, port=self.http_port, server=self.http_backend)
-        
-        
+
+
+    def workersmanager(self):
+        while True:
+            print "Workers manager thread"
+            time.sleep(1)
 
 
     # Here we will load all plugins (pages) under the webui/plugins
@@ -795,3 +833,35 @@ class Skonf(Daemon):
 
         #c = self.datamgr.get_contact(user_name)
         return user_name
+
+
+
+
+
+    # Create and launch a new worker, and put it into self.workers
+    def create_and_launch_worker(self):
+        # ceate the input queue of this worker
+        try:
+           if is_android:
+              q = Queue()
+           else:
+              q = self.manager.Queue()
+        # If we got no /dev/shm on linux, we can got problem here. 
+        # Must raise with a good message
+        except OSError, exp:
+            # We look for the "Function not implemented" under Linux
+            if exp.errno == 38 and os.name == 'posix':
+                logger.log("ERROR : get an exception (%s). If you are under Linux, please check that your /dev/shm directory exists." % (str(exp)))
+            raise
+            
+
+        w = SkonfUIWorker(1, q, self.returns_queue, 1, mortal=False, max_plugins_output_length = 1, target=None)
+        w.module_name = 'skonfuiworker'
+
+        # save this worker
+        self.workers[w.id] = w
+        
+        logger.log("[%s] Allocating new %s Worker : %s" % (self.name, w.module_name, w.id))
+        
+        # Ok, all is good. Start it!
+        w.start()
diff --git a/shinken/skonfuiworker.py b/shinken/skonfuiworker.py
new file mode 100644
index 0000000..e16b381
--- /dev/null
+++ b/shinken/skonfuiworker.py
@@ -0,0 +1,112 @@
+#!/usr/bin/env python
+# Copyright (C) 2009-2012 :
+#     Gabes Jean, naparuba at gmail.com
+#     Gerhard Lausser, Gerhard.Lausser at consol.de
+#     Gregory Starck, g.starck at gmail.com
+#     Hartmut Goebel, h.goebel at goebel-consult.de
+# 
+# This file is part of Shinken.
+# 
+# Shinken is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# 
+# Shinken is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+# 
+# You should have received a copy of the GNU Affero General Public License
+# along with Shinken.  If not, see <http://www.gnu.org/licenses/>.
+
+
+# This class is used for poller and reactionner to work.
+# The worker is a process launch by theses process and read Message in a Queue
+# (self.s) (slave)
+# They launch the Check and then send the result in the Queue self.m (master)
+# they can die if they do not do anything (param timeout)
+
+from Queue import Empty
+
+# In android, we sould use threads, not process
+is_android = True
+try:
+   import android
+except ImportError:
+   is_android = False
+
+if not is_android:
+   from multiprocessing import Process, Queue
+else:
+   from Queue import Queue
+   from threading import Thread as Process
+
+import time
+import sys
+import signal
+
+from shinken.worker import Worker
+
+
+# SkonfuiWorker class is a sub one for the Worker one of the poller/reactionners
+# it just take it's jobs from the mongodb, instead of the queues()
+class SkonfUIWorker(Worker):
+    id = 0
+    _process = None
+    _mortal = None
+    _idletime = None
+    _timeout = None
+    _c = None
+
+
+    def get_new_scan(self):
+       print "I ask for a scan"
+       time.sleep(1)
+
+    def launch_new_scan(self):
+       print "I try to launch scan", self.scan
+    
+
+    # id = id of the worker
+    # s = Global Queue Master->Slave
+    # m = Queue Slave->Master
+    # return_queue = queue managed by manager
+    # c = Control Queue for the worker
+    def work(self, s, returns_queue, c):
+        ## restore default signal handler for the workers:
+        # but on android, we are a thread, so don't do it
+        if not is_android:
+           signal.signal(signal.SIGTERM, signal.SIG_DFL)
+        timeout = 1.0
+        self.scan = None
+        self.returns_queue = returns_queue
+        self.s = s
+        self.t_each_loop = time.time()
+        while True:
+            begin = time.time()
+            msg = None
+            cmsg = None
+            
+            # If we are dying (big problem!) we do not
+            # take new jobs, we just finished the current one
+            if not self.i_am_dying:
+                self.get_new_scan()
+                self.launch_new_scan()
+
+            # Now get order from master
+            try:
+                cmsg = c.get(block=False)
+                if cmsg.get_type() == 'Die':
+                   print "[%d]Dad say we are dying..." % self.id
+                   break
+            except :
+                pass
+
+            # Manage a possible time change (our avant will be change with the diff)
+            diff = self.check_for_system_time_change()
+            begin += diff
+
+            timeout -= time.time() - begin
+            if timeout < 0:
+                timeout = 1.0
diff --git a/shinken/webui/plugins_skonf/newhosts/newhosts.py b/shinken/webui/plugins_skonf/newhosts/newhosts.py
index 3361c03..864b59f 100644
--- a/shinken/webui/plugins_skonf/newhosts/newhosts.py
+++ b/shinken/webui/plugins_skonf/newhosts/newhosts.py
@@ -23,7 +23,7 @@
 
 
 from shinken.webui.bottle import redirect
-from shinken.utils import to_bool
+from shinken.util import to_bool
 
 ### Will be populated by the UI with it's own value
 app = None

-- 
UNNAMED PROJECT



More information about the Pkg-nagios-changes mailing list