source: trunk/Monitoring/Watch/watch @ 866

#!/usr/bin/env python
# -*- coding: ISO-8859-1 -*-

##################################
#  @program        smon
#  @description    simulation monitor
#  @copyright      Copyright “(c)2009 Centre National de la Recherche Scientifique CNRS. 
#                             All Rights Reserved”
#  @svn_file       $Id: watcher 2545 2013-02-01 09:58:10Z jripsl $
#  @version        $Rev: 2545 $
#  @lastrevision   $Date: 2013-02-01 10:58:10 +0100 (Fri, 01 Feb 2013) $
#  @license        CeCILL (http://dods.ipsl.jussieu.fr/jripsl/smon/LICENSE)
##################################

import pika
import base64
import json
import sys
import signal
import traceback
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

# line below is to include "smon" package in the search path
sys.path.append("/home/jripsl/snapshot/Monitoring")

import smon.repo_io as repo_io
import smon.types

"""
Code list reminder

0000 (la simulation démarre)
1000 (le job d'une simulation démarre)
2000 (PushStack)
3000 (PopStack OK)
9000 (PopStack NOK)
9999 (FATAL)
"""

class Mail():

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

class Actions():

        @classmethod
        def store_msg(cls,message):
                repo_io.create_message(message)

        @classmethod
        def set_sim_status_to_error(cls,message):
                repo_io.update_simulation_status()

        @classmethod
        def crea_sim(cls,message):

                #repo_io.retrieve_simulation(name)
                #repo_io.delete_simulation(name)

                simulation=smon.types.Simulation(name=message.simuid)

                repo_io.create_simulation(simulation)

        @classmethod
        def mail(cls):
                cls.mail_example()

        @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 %s\n"%(datetime.now().strftime('%Y%m%d_%H%M%S'), message.code,message.jobid))

        @classmethod
        def execActions(cls,message):

                message_code=message.code

                for action in MessageActionsMapping.mapping[message_code]:
                        proc_name=action

                        try:
                                getattr(Actions, proc_name)(message)
                        except Exception,e:
                                traceback.print_exc()

                                raise Exception("WATCH-ERR002","procedure error (%s,%s)"%(proc_name,str(e)))

class MessageActionsMapping():

        mapping = { "0000":["log", "store_msg", "crea_sim"],
                                "1000":["log", "store_msg"],
                                "2000":["log", "store_msg"],
                                "3000":["log", "store_msg"],
                                "9000":["log", "store_msg", "mail"],
                                "9999":["log", "store_msg", "set_sim_status_to_error", "mail"] }

class Watcher():

        @classmethod
        def start(cls):
                pass

        @classmethod
        def stop(cls):
                pass

        @classmethod
        def main(self):

                """
                # 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):

                        # msg fmt: body:base64,file:base64 (no JSON here !!!)




                        # first deserialization (no JSON here !!!)

                        fields=raw_msg.split(",")

                        l__tmp_dic={}

                        for field in fields:

                                # debug
                                print " [x] Received %s" % field

                                splitted_field=field.split(":")

                                key=splitted_field[0]
                                val=splitted_field[1]

                                l__tmp_dic[key]=val


                        # debug
                        print " [x] Received %s (encoded)" % l__tmp_dic["body"]

                        
                        # base64 decode body
                        base64_decoded_msg=base64.b64decode(l__tmp_dic["body"])


                        # debug
                        #print " [x] Received %s" % raw_msg
                        print " [x] Received %s (uudecoded)" % base64_decoded_msg 
                        #print " [x] Received %s (uudecoded)" % base64_decoded_msg 


                        message=None
                        try:
                                # body deserialization 
                                JSON_msg=json.loads(base64_decoded_msg)
                                message=smon.types.Message(JSON_msg)      # all JSON object members will be available in smon.types.Message object

                                if "file" in l__tmp_dic:

                                        # base64 decode file
                                        base64_decoded_file=base64.b64decode(l__tmp_dic["file"])

                                        # add into msg
                                        message.file=base64_decoded_file

                                # message code based action
                                Actions.execActions(message)

                        except Exception,e:
                                print "Exception occurs (exception=%s,msg=%s)"%(str(e),base64_decoded_msg)

                                #traceback.print_exc()

                                raise

                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)

        try:
                Watcher.main()

                sys.exit(0)

        except Exception, e:

                #traceback.print_exc()

                sys.exit(1)