自动化运维系统雏形

java+python+robbitMQ+subprocess+threading
典型的客户端+mq+服务端架构...

import pika
import subprocess
import time
import traceback
import threading
import datetime
import signal
import os
import string
class ThreadClass(threading.Thread):
    def __init__(self, body, rpid, timeout=10):
            threading.Thread.__init__(self)
            self.body = body
            self.channel = mqConUtil().channel
            self.rpid = rpid
            self.timeout = int(timeout)
           
    def send_signal(self, sig):
            """Send a signal to the process
            """
            os.kill(self.pid, sig)
        
    def terminate(self):
        """Terminate the process with SIGTERM
        """
        self.send_signal(signal.SIGTERM)
       
    def kill(self):
        """Kill the process with SIGKILL
        """
        self.send_signal(signal.SIGKILL)
   
    def wait(self):
        """ wait child exit signal,
        """
        self.Popen.wait()

    def free_child(self):
        """
        kill process by pid
        """
        try:
            self.Popen.terminate()
            self.Popen.kill()
            self.terminate()
            self.kill()
            self.wait()
        except:
            pass
           
    def run(self):
        start = time.time()
        self.start_time = start
        now = datetime.datetime.now()
        ftime2 = time.strftime('%Y-%m-%d-%H:%M:%S', time.localtime(time.time()))
        print "%s says Hello World at time: %s" % (self.getName(), now)
        start = time.time()
        status = -11
        output = "defOutputERROR"
        pid = 0
        try:
            self.Popen = subprocess.Popen(split_cmd(self.body), stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False)
            self.pid = self.Popen.pid
            while self.Popen.poll() == None and (time.time() - self.start_time) < self.timeout :
                time.sleep(1)
            #p.wait()
            if self.Popen.poll() == None: # child is not exit yet!
                self.free_child() 
                self.returncode = -9999
                status = self.returncode
            else:
                self.returncode = self.Popen.poll()
                output = self.Popen.stdout.read()
                output += self.Popen.stderr.read()
                status = self.Popen.poll()
                pid = self.pid
        except Exception, e:
            print(e)
            print(traceback.format_exc())
        print(self.rpid, pid, str(time.time() - start), now, output, status, self.body)
        self.channel.basic_publish(exchange='', routing_key='hello2', body=str(self.rpid) + "|" + str(pid) + "|" + str(time.time() - start) + "|" + ftime2 + "|" + output.replace("\n", "") + "|" + str(status) + "|" + self.body)

def createChannel():
        connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.13.27.215', port=56729))
        return connection.channel()
def split_cmd(s):
    """
    str --> [], for subprocess.Popen()
    """
    SC = '"'
    a = s.split(' ')
    cl = []
    i = 0
    while i < len(a) :
        if a[i] == '' :
            i += 1
            continue
        if a[i][0] == SC :
            n = i
            loop = True
            while loop:
                if a[i] == '' :
                    i += 1
                    continue
                if a[i][-1] == SC :
                    loop = False
                    m = i
                i += 1
            #print a[n:m+1]
            #print ' '.join(a[n:m+1])[1:-1]
            cl.append((' '.join(a[n:m + 1]))[1:-1])
        else:
            cl.append(a[i])
            i += 1
    return cl
def callback(ch, method, properties, body):
    print " [x] Received %r" % (body)
    try:
        rpid, subbody = string.split(body, "#")
        t = ThreadClass(subbody, rpid)
        t.start()
    except Exception, e:
        print("error".join(str(subprocess.errno)))
        print(e)
        print(traceback.format_exc())
#    (status, output) = commands.getstatusoutput(body)
def setup():
        connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.13.27.215', port=56729))
        channel = connection.channel()
        channel.basic_consume(callback, queue='hello', no_ack=True)
        #channel.basic_qos(prefetch_count=1)
        #channel.queue_declare(queue='hello')
        channel.start_consuming()
def singleton(cls):
  instances = {}
  def get_instance():
    if cls not in instances:
      instances[cls] = cls()
    return instances[cls]
  return get_instance

@singleton
class mqConUtil:
       def __init__(self):
           self.channel = createChannel()
          
          
          
          
   
if __name__ == '__main__':
    while True:
            try:
                setup()
               
            except Exception, e:
                print("error".join(str(subprocess.errno)))
                print(e)
                print(traceback.format_exc())
#for i in range(1, 1):
#    connection.process_data_events();
    print '

Waiting for messages. To exit press CTRL+C-done'