source: trunk/Monitoring/Watch/watch @ 937

#!/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
import datetime

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

import smon.repo_io as repo_io
import smon.types

CSTE_BROKER_HOST='cstest-broker.ipsl.jussieu.fr' # cstest
#CSTE_BROKER_HOST='localhost' # vesg4

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

                try:

                        # the simu exists when we are here (see TAG0001 tag)
                        s=repo_io.retrieve_simulation(message.simuid)

                        if s is None:
                                raise Exception("WATCH-ERR102","simulation not found")

                        repo_io.create_message(message,s)

                except:
                        traceback.print_exc()
                        raise

        @classmethod
        def cleanup(cls,message):
                repo_io.cleanup() # truncate/delete everything

        @classmethod
        def set_sim_status_to_error(cls,message):

                try:

                        s=repo_io.retrieve_simulation(message.simuid)

                        s.status="error"

                        repo_io.update_simulation_status(s)

                except:
                        traceback.print_exc()
                        raise

        @classmethod
        def set_sim_status_to_complete(cls,message):

                s=repo_io.retrieve_simulation(message.simuid)

                s.status="complete"

                repo_io.update_simulation_status(s)

        @classmethod
        def crea_sim(cls,message):

                s=repo_io.retrieve_simulation(message.simuid)

                if s is not None:
                        #repo_io.delete_simulation(name)

                        s.status="running"
                        repo_io.update_simulation_status(s)

                else:
                        simulation=smon.types.Simulation(name=message.simuid,status="running")


                        repo_io.create_simulation(simulation)

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

        @classmethod
        def print_stdout(cls,message):
                # used for debug

                """
                if message.file is not None:
                        print "%s %s %s %s %s\n"%(message.code,message.jobid,message.command,message.timestamp,message.file)
                else:
                        print "%s %s %s %s\n"%(message.code,message.jobid,message.command,message.timestamp)
                """

                print "%s %s %s %s\n"%(message.code,message.jobid,message.command,message.timestamp)
                #pass

        @classmethod
        def log(cls,message):
                with open("/opt/supervisor/log/supervisor.log", "a") as log_file:
                        log_file.write("%s %s %s %s %s\n"%(datetime.datetime.now().strftime('%Y%m%d_%H%M%S'), message.code,message.jobid,message.timestamp,message.command))

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

        # debug
        #
        # TAG0001: note that crea_sim must be BEFORE store_msg in the list (because when we insert the msg, we need the simu_id)
        #
        mapping = { "0000":["crea_sim", "log", "store_msg", "print_stdout"],
                                "0100":["log", "store_msg", "print_stdout", "set_sim_status_to_complete"],
                                "1000":["log", "store_msg", "print_stdout"],
                                "1100":["log", "store_msg", "print_stdout"],
                                "2000":["log", "store_msg", "print_stdout"],
                                "3000":["log", "store_msg", "print_stdout"],
                                "8888":["cleanup"],
                                "9000":["log", "store_msg", "print_stdout"],
                                "9999":["log", "store_msg", "print_stdout", "set_sim_status_to_error"] }

        # prod
        #
        # TAG0001: note that crea_sim must be BEFORE store_msg in the list (because when we insert the msg, we need the simu_id)
        #
        """
        mapping = { "0000":["crea_sim", "log", "store_msg"],
                                "0100":["log", "store_msg", "set_sim_status_to_complete"],
                                "1000":["log", "store_msg"],
                                "1100":["log", "store_msg"],
                                "2000":["log", "store_msg"],
                                "3000":["log", "store_msg"],
                                "8888":["cleanup"],
                                "9000":["log", "store_msg", "mail"],
                                "9999":["log", "store_msg", "set_sim_status_to_error", "mail"] }
        """

class Watcher():

        @classmethod
        def start(cls):
                repo_io.init() # open DB connection

        @classmethod
        def stop(cls):
                repo_io.free() # close DB connection

        @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=CSTE_BROKER_HOST))
                self.channel = connection.channel()


                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 deserialization 
                        message=None
                        try:
                                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

                                # non working
                                #print message.type

                                # working
                                #print message.code



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

                                traceback.print_exc()
                                raise



                        # manage config-card file which is attached to the "0000" code message (this file is base64 encoded and need to be unencoded)
                        #
                        if "file" in l__tmp_dic:

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

                                # add as msg attribute
                                message.file=base64_decoded_file



                        # execute actions
                        try:
                                # message code based action
                                Actions.execActions(message)

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

                                traceback.print_exc()

                                raise


                        # slow down consumer
                        #time.sleep(0.5)

                self.channel.queue_declare(queue='myqueue')

                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()
                Watcher.stop()
                sys.exit(0)

if __name__ == '__main__':

        signal.signal(signal.SIGINT, signal_handler)

        try:

                Watcher.start()

                Watcher.main()

                Watcher.stop()

                sys.exit(0)

        except Exception, e:

                traceback.print_exc()

                sys.exit(1)