Changeset 854 for trunk/Monitoring

Timestamp:

04/26/13 16:57:16 (11 years ago)

Author:

jripsl

Message:

add JSON deserialization, mail notification, logging.

Location:

trunk/Monitoring

Files:

• Analyze (added)
• Broker/README (modified) (1 diff)
• CNClient/README (modified) (3 diffs)
• CNClient/send_AMQP_msg.c (modified) (9 diffs)
• Watch/watch (modified) (5 diffs)
• doc/README (added)
• sample/config.card.base64 (added)
• script/libIGCM_mock.sh (modified) (1 diff)
• test/README (modified) (1 diff)
• test/librabbitmq-python (added)
• test/librabbitmq-python/test1 (added)
• test/librabbitmq-python/test1/client.sh (added)
• test/librabbitmq-python/test1/server.py (added)
• test/librabbitmq-python/test2 (added)
• test/librabbitmq-python/test2/consumer.py (added)
• test/librabbitmq-python/test2/producer.py (added)

trunk/Monitoring/Broker/README

@@ -42 +42 @@
                 database dir: /var/lib/rabbitmq/mnesia/rabbit
 - run
-        - to start the daemon, use command below
+        - to start the daemon, use command below as root
+                - cd /opt/rabbitmq-server-3.0.2/sbin
                 - ./rabbitmq-server -detached
         - to stop the daemon, use this

trunk/Monitoring/CNClient/README

@@ -1 +1 @@
 - This program uses rabbitmq-c library (v0.3.0)
         - https://github.com/alanxz/rabbitmq-c
-        - https://github.com/alanxz/rabbitmq-c/archive/rabbitmq-c-v0.3.0.zip
 - Compilation (static)
         - Library installation
@@ -11 +10 @@
                           so better use installation from source
                 - from source
-                        tar xzvf rabbitmq-c-v0.3.0.zip
-                        cd rabbitmq-c-v0.3.0
-                        autoreconf -i
-                        ./configure --enable-static
-                        make
-                        make install
+                        - retrieve source
+                                - wget https://github.com/alanxz/rabbitmq-c/archive/rabbitmq-c-v0.3.0.zip -O rabbitmq-c-v0.3.0.zip
+                                - unzip rabbitmq-c-v0.3.0.zip
+                                - cd rabbitmq-c-v0.3.0
+                        - compilation using autoconf (automake v1.9+, and libtool v2.2+)
+                                - autoreconf -i
+                                        - if error occurs at this step, check requirement below
+                                                - we require autotools v2.63 or better to build. I think RHEL5 ships with v2.59 which lacks the AC_PROC_CC_C99 macro.
+                                                        - You can do one of two things;
+                                                                - Install a newer version of autotools
+                                                                - Build using cmake (v2.6 or better).
+                                - ./configure --enable-static
+                                - make
+                                - make install
+                        - compilation using cmake (CMake v2.6+)
+                                - mkdir build && cd build
+                                - cmake -DBUILD_STATIC_LIBS=True -DBUILD_SHARED_LIBS=True ..
+                                - cmake --build .
+                                - make
+                                        - you got error AAA below, it's normal
+                                                make[2]: *** Pas de rÚgle pour fabriquer la cible « librabbitmq/librabbitmq.so.1.0.1 », nécessaire pour « examples/amqp_bind ». Arrêt.
+                                                make[1]: *** [examples/CMakeFiles/amqp_bind.dir/all] Erreur 2
+                                                make: *** [all] Erreur 2
+                                - make
+                                  (because of the above error (AAA), you need to run make twice)
+                                - make install
         - gcc -static -I/usr/local/include -L/usr/local/lib -Wall -o sendAMQPMsg send_AMQP_msg.c -lrabbitmq
                 - we get warning below during compilation
@@ -27 +46 @@
                         - it means that you may need to be sure all computing node have the same glibc version !!!!
                                 - also means that a different binary must be use in each computing center
-- Run
-        - ./sendAMQPMsg localhost 5672 1 10
+- usage
+        - to send a message in the queue, do
+                - ./sendAMQPMsg localhost 5672 test
+- note
+        - to show how many messages are in the queue, do as root
+                - ./rabbitmqctl list_queues

trunk/Monitoring/CNClient/send_AMQP_msg.c

@@ -16 +16 @@
 
 #define SUMMARY_EVERY_US 1000000
-
 
 
@@ -117 +116 @@
     }
 
-        // note that "amq.direct" is a special exchange
-        // (The empty exchange name is an alias for amq.direct)
+          // note that "amq.direct" is a special exchange
+          // (The empty exchange name is an alias for amq.direct)
     //die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("amq.direct"), amqp_cstring_bytes(queue_name), 0, 0, NULL, message_bytes), "Publishing");
 
@@ -158 +157 @@
   amqp_connection_state_t conn;
 
-  amqp_bytes_t reply_to_queue;
-
-  if (argc < 5) {
-    fprintf(stderr, "Usage: amqp_producer host port rate_limit message_count\n");
-    return 1;
-  }
+  //amqp_bytes_t reply_to_queue;
+
 
   hostname = argv[1];
@@ -204 +199 @@
 }
 
-
-//int main_complex(int argc, char const * const *argv) {
-int main(int argc, char const * const *argv) {
+int read_file(char *f,char **source) {
+        FILE *fp = fopen(f, "r");
+        if (fp != NULL) {
+                /* Go to the end of the file. */
+                if (fseek(fp, 0L, SEEK_END) == 0) {
+                        /* Get the size of the file. */
+                        long bufsize = ftell(fp);
+                        if (bufsize == -1) { /* Error */ }
+
+                        /* Allocate our buffer to that size. */
+                        *source = malloc(sizeof(char) * (bufsize + 1));
+
+                        /* Go back to the start of the file. */
+                        if (fseek(fp, 0L, SEEK_SET) == 0) { /* Error */ }
+
+                        /* Read the entire file into memory. */
+                        size_t newLen = fread(*source, sizeof(char), bufsize, fp);
+                        if (newLen == 0) {
+                                fputs("Error reading file\n", stderr);
+        return 1;
+                        } else {
+                                //source[++newLen] = '\0'; /* Just to be safe. */
+                        }
+                }
+                fclose(fp);
+        } else {
+                fputs("File not found\n", stderr);
+                return 1;
+        }
+
+  return 0;
+}
+
+int main(int argc, char * const *argv) {
   char const *hostname;
   int port;
-  char const *exchange;
-  char const *routingkey;
-  char const *messagebody;
+  //char const *exchange;
+  //char const *routingkey;
+  char *body_tmp;
+  char const *body;
+  char *body_final = NULL;
+  char *buf = NULL;
+  int c;
+  char *filepath = NULL;
+  int file_flag=0;
 
 
@@ -218 +250 @@
   amqp_connection_state_t conn;
 
-  if (argc < 4) {
-    fprintf(stderr, "Usage: amqp_sendstring host port messagebody\n");
-    return 1;
-  }
-
-  hostname = argv[1];
-  port = atoi(argv[2]);
-  //exchange = argv[3];
-  //routingkey = argv[4];
-  messagebody = argv[5];
+  //parsing args
+  opterr = 0;
+  while ((c = getopt (argc, argv, "b:ef:h:p:")) != -1)
+    switch (c)
+      {
+      case 'b':
+        body_tmp = optarg;
+        break;
+      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 2: ./sendAMQPMsg -h localhost -p 5672 -f /home/foobar/config.card -b '{\"jobid\":\"toto\"}'\n");
+        exit(0);
+      case 'f':
+        filepath = optarg;
+        file_flag = 1;
+        break;
+      case 'h':
+        hostname = optarg;
+        break;
+      case 'p':
+        port = atoi(optarg);
+        break;
+      case '?':
+        fprintf (stderr, "ERR001: incorrect argument '-%c'.\n", optopt);
+        exit(EXIT_FAILURE);
+      default:
+        fprintf (stderr, "ERR002: incorrect argument\n");
+        exit(EXIT_FAILURE);
+      }
+
+  //retrieve non-option argument
+  /*
+  int index;
+  for (index = optind; index < argc; index++)
+    printf ("Non-option argument %s\n", argv[index]);
+  */
+
+  // add checks here
+  // (for example, body_tmp is mandatory)
+
+  if(file_flag==1) {
+    // retrieve file contents
+
+    if( access( filepath, F_OK ) != -1 ) {
+        // file exists
+        
+        ;
+    } else {
+      // file doesn't exist
+
+      fprintf(stderr, "File not found (%s)\n",filepath);
+
+      exit(EXIT_FAILURE);
+    }
+
+    int res = read_file(filepath,&buf);
+    if (res != 0) {
+      exit(EXIT_FAILURE);
+    }
+
+    body_final=malloc(strlen(body_tmp) + 6 + strlen(buf) + 1);
+    //strcpy(body_final,"\0");
+    strcat(body_final,body_tmp);
+    strcat(body_final,",file=");
+    strcat(body_final,buf);
+
+    body=body_final;
+
+    //debug
+    //fprintf(stderr, "hostname=%s, port=%d, body=%s, filepath=%s\n", hostname, port, body, filepath);
+
+  } else {
+    // retrieve msg body from argument
+
+    body=body_tmp;
+
+  }
+
+  //exit(0);
 
   conn = amqp_new_connection();
@@ -237 +339 @@
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
 
-
-
-
-  //JRA
   amqp_exchange_declare(conn, 1, amqp_cstring_bytes("myexchange"), amqp_cstring_bytes("fanout"), 0, 0, amqp_empty_table);
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring exchange");
 
-  //JRA
-  amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1, amqp_cstring_bytes("myqueue"), 0, 1, 0, 0, amqp_empty_table); // durable && no auto-delete
+  amqp_queue_declare(conn, 1, amqp_cstring_bytes("myqueue"), 0, 1, 0, 0, amqp_empty_table); // durable && no auto-delete
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
 
   amqp_queue_bind(conn, 1, amqp_cstring_bytes("myqueue"), amqp_cstring_bytes("myexchange"), amqp_cstring_bytes(""), amqp_empty_table); //no need for binding key as we use the fanout exchange type
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Binding");
-
-
 
   { 
@@ -258 +353 @@
     props.content_type = amqp_cstring_bytes("text/plain");
     props.delivery_mode = 2; /* persistent delivery mode */
-    die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("myexchange"), amqp_cstring_bytes(""), 0, 0, &props, amqp_cstring_bytes(messagebody)), "Publishing");
+    die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("myexchange"), amqp_cstring_bytes(""), 0, 0, &props, amqp_cstring_bytes(body)), "Publishing");
   }
 
@@ -264 +359 @@
   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
   die_on_error(amqp_destroy_connection(conn), "Ending connection");
+
+  if(file_flag==1)
+    free(buf);
+
   return 0;
 }
@@ -347 +446 @@
 }
 
+/* vi: set et ts=2 sw=2: */

trunk/Monitoring/Watch/watch

@@ -1 @@
-#!/usr/bin/python -u
+#!/usr/bin/env python
 # -*- coding: ISO-8859-1 -*-
 
@@ -13 +13 @@
 ##################################
 
-from smon import dao
+#from smon import dao
+import pika
+import base64
+import json
+import sys
+import traceback
+import smtplib
+from email.mime.text import MIMEText
 
 class Watcher():
+        message_code_action_mapping = {"0000":["log","mail"],"1000":["log"],"2000":["log"],"3000":["log"],"9000":["log"],"9999":["log","mail"]}
 
         @classmethod
@@ -23 +31 @@
         @classmethod
         def start(cls):
-                dao.insert_progress_messages(cls.get_fake_progress_messages())
+                #dao.insert_progress_messages(cls.get_fake_progress_messages())
+                pass
 
         @classmethod
         def stop(cls):
                 pass
- 
+
+        @classmethod
+        def add(cls,message):
+                pass
+
+        @classmethod
+        def mail_example(cls):
+                me="jripsl@ipsl.jussieu.fr"
+                you="jripsl@ipsl.jussieu.fr"
+                body="Alarm"
+                object="Supervisor"
+
+                cls.mail(me,you,object,body)
+
+        @classmethod
+        def mail(cls):
+                cls.mail_example()
+
+        @classmethod
+        def send_mail(cls,me,you,object,body):
+                msg = MIMEText(body)
+                msg['Subject'] = object
+                msg['From'] = me
+                msg['To'] = you
+
+                # Send the message via our own SMTP server, but don't include the # envelope header.
+                s = smtplib.SMTP('localhost')
+                s.sendmail(me,[you], msg.as_string())
+                s.quit()
+
+        @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"]))
+
+        @classmethod
+        def execActions(cls,message):
+
+                message_code=message["code"]
+
+                for action in cls.message_code_action_mapping[message_code]:
+                        proc_name=action
+
+                        try:
+                                getattr(cls, proc_name)(message)
+                        except Exception,e:
+                                traceback.print_exc()
+
+                                raise Exception("WATCH-ERR002","procedure error (%s,%s)"%(proc_name,str(e)))
+
 def main():
 
+        """
         # parse args
         parser = argparse.ArgumentParser(prog='watcher')
@@ -41 +101 @@
 
         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__':
@@ -54 +145 @@
         except Exception, e:
 
-                #traceback.print_exc()
+                traceback.print_exc()
 
                 sys.exit(1)

trunk/Monitoring/script/libIGCM_mock.sh

@@ -12 +12 @@
 ##################################
 
-stack_file="../sample/stack_light"
+stack_file=$1
+send_msg="/home/jripsl/snapshot/Monitoring/CNClient/sendAMQPMsg"
+
+if [ $# -lt 1 ]; then
+        echo "Usage $0 ../sample/stack_light"
+        exit 1
+fi
 
 IFS=$'\n'
 for line in $(cat $stack_file); do
-        echo $line | awk -F" " '{print $4}'
+        #echo $line | awk -F" " '{print $4}'
+        callname=$(echo $line | awk -F" " '{print $4}' )
+        $send_msg localhost 5672 string "$callname"
 done

trunk/Monitoring/test/README

@@ -1 @@
-test_twisted.py   consumer/producer test showing how to use RabbitMQ, Twisted and txamqp together
-recv_AMQP_msg.c   basic consumer test
+test_twisted.py    consumer/producer test showing how to use RabbitMQ, Twisted and txamqp together
+recv_AMQP_msg.c    basic consumer test
+librabbitmq-python consumer/producer using librabbitmq-Python library