Changeset 857 for trunk/Monitoring

Timestamp:

05/15/13 15:25:03 (11 years ago)

Author:

jripsl

Message:

• start database communication implementation.
• add usleep( 200 000 ) in CNClient to prevent overflow on the server side.

Location:

trunk/Monitoring

Files:

• CNClient (modified) (1 prop)
• CNClient/send_AMQP_msg.c (modified) (2 diffs)
• Watch/watch (modified) (3 diffs)
• doc/README (modified) (1 diff)

trunk/Monitoring/CNClient/send_AMQP_msg.c

@@ -260 +260 @@
       case 'e':
         fprintf(stderr, "Usage: amqp_producer host port rate_limit message_count\n");
-        fprintf(stderr, "Example 1: ./sendAMQPMsg -h localhost -p 5672 -b '{\"jobid\":\"toto\"}'\n");
+
+        fprintf(stderr, "Example 1: ./sendAMQPMsg -h localhost -p 5672 -b 'eyJqb2JpZCI6InRvdG8iLCJjb2RlIjoiMDAwMCJ9Cg=='\n");
+
+        // obsolete (from now all sent data must be B64 encoded)
         fprintf(stderr, "Example 2: ./sendAMQPMsg -h localhost -p 5672 -f /home/foobar/config.card -b '{\"jobid\":\"toto\"}'\n");
+
+        fprintf(stderr, "Example 3: ./sendAMQPMsg -h localhost -p 5672 -b 'am9iaWQ6dG90byBjb2RlOjAwMDAK' -f ../sample/config.card.base64\n");
         exit(0);
       case 'f':
@@ -363 +368 @@
     free(buf);
 
+
+  //this is to prevent overwhelming rabbitMQ server (and to prevent triggering rabbitMQ defensive behaviours (e.g. connection blocking))
+  usleep( 200 000 ); // O.2 second
+
+
   return 0;
 }

trunk/Monitoring/Watch/watch

@@ -18 +18 @@
 import json
 import sys
+import signal
 import traceback
 import smtplib
 from email.mime.text import MIMEText
+from datetime import datetime
+
+
+
+
+sys.path.append("/home/jripsl/snapshot/src") 
+import elixir
+import prodiguer_repository
+
+#import prodiguer_repository.operations.mq_hooks_stub as mq_hooks
+import prodiguer_repository.operations.mq_hooks as mq_hooks
+
+
+
+"""
+mq_hooks.retrieve_simulation(name)
+mq_hooks.delete_simulation(name)
+mq_hooks.create_simulation(activity,
+mq_hooks.update_simulation_status(name,
+mq_hooks.retrieve_messages(name)
+mq_hooks.delete_messages(name)
+"""
+
+"""
+from prodiguer_repository.models import (
+        ExecutionState,
+        EXECUTION_STATE_RUNNING,
+        EXECUTION_STATE_SET,
+        Message,
+        Simulation,
+        SIMULATION_SPACE_TEST
+)
+"""
+
+
+
+"""
+scenarios from ticket 136
+
+libIGCM simulent des erreurs (killed, segfault, 1% d'erreur)
+le watcher temps réel change le status des erreurs détéctés
+le watcher temps différé détécte et change le statut des killed 
+"""
 
 class Watcher():
-        message_code_action_mapping = {"0000":["log","mail"],"1000":["log"],"2000":["log"],"3000":["log"],"9000":["log"],"9999":["log","mail"]}
+        # prod
+        #message_code_action_mapping = {"0000":["log","mail"],"1000":["log"],"2000":["log"],"3000":["log"],"9000":["log"],"9999":["log","mail"]}
+
+        # dev
+        message_code_action_mapping = {"0000":["log","print_stdout"],"1000":[],"2000":[],"3000":[],"9000":["log"],"9999":["log"]}
+        #message_code_action_mapping = {"0000":[],"1000":[],"2000":[],"3000":[],"9000":["log"],"9999":["log"]}
+
+        # debug
+        #message_code_action_mapping = {"0000":["print_stdout"],"1000":["log"],"2000":[],"3000":[],"9000":["log"],"9999":["log"]}
 
         @classmethod
@@ -68 +120 @@
 
         @classmethod
+        def print_stdout(cls,message):
+                # used for debug
+
+                if "file" in message:
+                        print "%s %s %s\n"%(message["code"],message["jobid"],message["file"])
+                else:
+                        print "%s %s\n"%(message["code"],message["jobid"])
+
+        @classmethod
         def log(cls,message):
 
                 with open("/home/jripsl/supervisor/log/supervisor.log", "a") as log_file:
-                        log_file.write("%s %s\n"%(message["code"],message["jobid"]))
+                        log_file.write("%s %s %s\n"%(datetime.now().strftime('%Y%m%d_%H%M%S'), message["code"],message["jobid"]))
 
         @classmethod
@@ -88 +149 @@
                                 raise Exception("WATCH-ERR002","procedure error (%s,%s)"%(proc_name,str(e)))
 
-def main():
-
-        """
-        # parse args
-        parser = argparse.ArgumentParser(prog='watcher')
-        parser.add_argument('-v', dest='verbose',required=False,action='store_true')
-        args = parser.parse_args()
-
-        # check
-        if not os.path.exists(SMON.smon_home):
+        @classmethod
+        def main(self):
+
+                """
+                _CONNECTION = "postgresql://postgres:Silence107!@localhost:5432/prodiguer" # Repo connection string.
+                prodiguer_repository.connect(_CONNECTION) # Connect to repo.
+                mq_hooks.create_message("test", 2, "bla")
+                elixir.session.commit()
+                #elixir.session.rollback()
+                """
+                mq_hooks.create_message()
+
+                """
+                # parse args
+                parser = argparse.ArgumentParser(prog='watcher')
+                parser.add_argument('-v', dest='verbose',required=False,action='store_true')
+                args = parser.parse_args()
+
+                # check
+                if not os.path.exists(SMON.smon_home):
+                        sys.exit(1)
+
+                SMON.init_singleton()
+                """
+
+                connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost'))
+                self.channel = connection.channel()
+
+                #self.channel.queue_declare(queue='myqueue')
+
+                print ' [*] Waiting for messages. To exit press CTRL+C'
+
+                def callback(ch, method, properties, raw_msg):
+                        #
+                        #return
+
+                        base64_decoded_msg=base64.b64decode(raw_msg)
+
+                        # debug
+                        #print " [x] Received %s" % raw_msg
+                        #print " [x] Received %s (uudecoded)" % base64_decoded_msg 
+
+                        message=None
+                        try:
+                                message=json.loads(base64_decoded_msg)
+
+                                # message code based action
+                                Watcher.execActions(message)
+                        except Exception,e:
+                                print "Exception occurs (exception=%s,msg=%s)"%(str(e),base64_decoded_msg)
+
+                self.channel.basic_consume(callback, queue='myqueue', no_ack=True)
+
+                self.channel.start_consuming()
+
+
+                """
+                SMON.free_singleton()
+                """
+
+def signal_handler(signal, frame):
+                print 'You pressed Ctrl+C!'
+                Watcher.channel.stop_consuming()
+                sys.exit(0)
+
+if __name__ == '__main__':
+
+        signal.signal(signal.SIGINT, signal_handler)
+        #print 'Press Ctrl+C'
+
+        try:
+                Watcher.main()
+
+                sys.exit(0)
+
+        except Exception, e:
+
+                traceback.print_exc()
+
                 sys.exit(1)
-
-        SMON.init_singleton()
-        """
-
-        connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost'))
-        channel = connection.channel()
-
-        #channel.queue_declare(queue='myqueue')
-
-        print ' [*] Waiting for messages. To exit press CTRL+C'
-
-        def callback(ch, method, properties, raw_msg):
-                #
-                #return
-
-                base64_decoded_msg=base64.b64decode(raw_msg)
-
-                # debug
-                #print " [x] Received %s" % raw_msg
-                #print " [x] Received %s (uudecoded)" % base64_decoded_msg 
-
-                try:
-                        message=json.loads(base64_decoded_msg)
-                except Exception,e:
-                        print base64_decoded_msg
-                
-                # message code based action
-                Watcher.execActions(message)
-
-        channel.basic_consume(callback, queue='myqueue', no_ack=True)
-
-        channel.start_consuming()
-
-
-        """
-        SMON.free_singleton()
-        """
-
-if __name__ == '__main__':
-        try:
-                main()
-
-                sys.exit(0)
-
-        except Exception, e:
-
-                traceback.print_exc()
-
-                sys.exit(1)

trunk/Monitoring/doc/README

@@ -9 +9 @@
                 - Virtualenv installation
                         - TODO: add virtual env install instructions here
-                - libraries installation
-                        - cd /opt
-                        - cd python2.6_ve/
-                        - cd bin
-                        - source activate
-                        - pip install pika         # RabbitMQ client library
-                        - pip install Sqlalchemy
-                        - pip install elixir
+                - Python PYPI libraries installation
+                        - procedure (as root)
+                                - cd /opt
+                                - cd python2.6_ve/
+                                - cd bin
+                                - source activate
+                                - pip install <package name>
+                        - packages list
+                                - packages below are needed by SMON
+                                        - pip install pika # RabbitMQ client library
+                                - packages below are needed by Prodiguer library
+                                        -
+                                                - pip install Sqlalchemy
+                                                - pip install elixir
+                                                - pip install simplejson
+                                        - note: some Elixir/SQLAlchemy versions are not compatible.
+                                          to be sure, you can use version below which are compatible
+                                                - Elixir-0.7.1-py2.6.egg-info
+                                                - SQLAlchemy-0.7.8-py2.6.egg-info
+                - Prodiguer library installation
+                        - svn co svn+ssh://<login here>@forge.ipsl.jussieu.fr/ipsl/forge/projets/prodiguer/svn/sw/src/python/trunk/prodiguer_repository/src
+                        - edit "watch" script and edit line below accordingly
+                                - sys.path.append("/home/jripsl/snapshot/src")