manager/src/process-state.py

   1 #!/usr/bin/env python
   2 #   Copyright (c) 2019 AT&T Intellectual Property.
   3 #   Copyright (c) 2019 Nokia.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 #
  17
  18 #
  19 #   This script does have the event handler in supervisor to follow the process state.
  20 #   Main parent process follows the events from the supervised daemon and the child process
  21 #   provides the process status info via HTTP server interface.
  22 #   When any of the process state change to PROCESS_STATE_FATAL, then the http server will
  23 #   respond with status code 500 to indicate faulty operation, normal status code is 200 (working).
  24 #   Set the script configuration to supervisord.conf as follow:
  25 #
  26 #   [eventlistener:process-state]
  27 #   command=python process-state.py
  28 #   autorestart=true
  29 #   startretries=3
  30 #   events=PROCESS_STATE
  31 #
  32 #   Following is the example supervisor daemon status for each process.
  33 #
  34 #   ver:3.0 server:supervisor serial:16 pool:process-state poolserial:16 eventname:PROCESS_STATE_FATAL len:62
  35 #   processname:sleep-exit groupname:sleep-exit from_state:BACKOFF
  36 #
  37 #   Process states are: PROCESS_STATE_STARTING, PROCESS_STATE_RUNNING, PROCESS_STATE_STOPPING,
  38 #   PROCESS_STATE_STOPPEDPROCESS_STATE_BACKOFF, PROCESS_STATE_FATAL
  39 #
  40
  41 import os
  42 import sys
  43 import signal
  44 import datetime
  45 import argparse
  46 import psutil
  47 import threading
  48 import BaseHTTPServer
  49
  50 # needed for python socket library broken pipe WA
  51 from multiprocessing import Process, Manager, Value, Lock
  52
  53 global HTTP_STATUS
  54 global PROCESS_STATE
  55 global DEBUG
  56
  57 TABLE_STYLE = ('<style>'
  58     'div { font-family: Arial, Helvetica, sans-serif; font-size:8px;}'
  59     'p { font-family: Arial, Helvetica, sans-serif;}'
  60     'h1 { font-family: Arial, Helvetica, sans-serif; font-size:30px; font-weight: bold; color:#A63434;}'
  61     'table, th, td { border: 1px solid black; border-collapse: collapse; font-size:12px;}'
  62     'th, td { padding: 3px 10px 3px 10px; text-align: left;}'
  63     'th.thr, td.tdr { padding: 3px 10px 3px 10px; text-align: right;}'
  64     'th.thc, td.tdc { padding: 3px 10px 3px 10px; text-align: center;}'
  65     'table#t1 tr:nth-child(even) { background-color: #eee;}'
  66     'table#t1 tr:nth-child(odd) { background-color: #ADC0DC;}'
  67     'table#t1 th { background-color: #214D8B; color: white;}</style>')
  68
  69 def get_pid_info(pid):
  70     pdata = None
  71     if pid is not 0:
  72         try:
  73             process = psutil.Process(pid)
  74             # these are the item lists
  75             files = process.open_files()
  76             # get the open files and connections and count number of fd str
  77             sockets = process.connections()
  78             descriptors = str(files)+str(sockets)
  79             count = descriptors.count("fd=")
  80             pdata = {"pid": process.pid,
  81                     "status": process.status(),
  82                     "cpu": process.cpu_percent(interval=0.2),
  83                     "descriptors": count,
  84                     "memory": process.memory_info().rss}
  85         except (psutil.ZombieProcess, psutil.AccessDenied, psutil.NoSuchProcess):
  86             pdata = None
  87     return pdata
  88
  89 def get_process_html_info():
  90     global PROCESS_STATE
  91
  92     try:
  93         html_data = ("<table width='800px' id='t1'>"
  94                     "<thead><tr><th>Process</th><th>Date and time</th><th>From state</th><th>to state</th>"
  95                     "<th class=thc>Pid</th><th class=thc>Fds</th><th class=thc>Mem</th><th class=thc>Cpu</th></tr></thead><tbody>")
  96         for proc,data in PROCESS_STATE.items():
  97             pid = 0
  98             descriptors = ""
  99             memory_usage = ""
 100             cpu_usage = ""
 101             if data['pid'] is not None:
 102                 pdata = get_pid_info(data['pid'])
 103                 if pdata is not None:
 104                     pid = pdata['pid']
 105                     descriptors = str(pdata['descriptors'])
 106                     memory_usage = str(pdata['memory']/1024)+" Kb"
 107                     cpu_usage = str(pdata['cpu'])+" %"
 108             html_data += ('<tr>'
 109                             '<td>'+str(proc)+'</td>'
 110                             '<td>'+str(data['time'])+'</td>'
 111                             '<td>'+str(data['from_state'])+'</td>'
 112                             '<td>'+str(data['state'])+'</td>'
 113                             '<td class=tdr>'+str(pid)+'</td>'
 114                             '<td class=tdr>'+descriptors+'</td>'
 115                             '<td class=tdr>'+memory_usage+'</td>'
 116                             '<td class=tdr>'+cpu_usage+'</td>'
 117                           '</tr>')
 118     finally:
 119         html_data += ("</tbody></table>")
 120
 121     return html_data
 122
 123 # responds to http request according to the process status
 124 class myHTTPHandler(BaseHTTPServer.BaseHTTPRequestHandler):
 125     global HTTP_STATUS
 126     global REFRESH_TIME
 127     global PROCESS_STATE
 128
 129     # write HEAD and GET to client and then close
 130     def do_HEAD(s):
 131         s.send_response(HTTP_STATUS['code'])
 132         s.send_header("Server-name", "supervisor-process-stalker 1.0")
 133         s.send_header("Content-type", "text/html")
 134         s.end_headers()
 135         s.wfile.close()
 136     def do_GET(s):
 137         try:
 138             """Respond to a GET request."""
 139             s.send_response(HTTP_STATUS['code'])
 140             s.send_header("Server-name", "supervisor-process-stalker 1.0")
 141             s.send_header("Content-type", "text/html")
 142             s.end_headers()
 143             html_data = ("<html><head><title>supervisor process event handler</title>"+TABLE_STYLE+
 144                         "<meta http-equiv='refresh' content='"+str(REFRESH_TIME)+"'/></head>"
 145                         "<body><h1>Supervisor Process Event Handler</h1>"
 146                         "<div><table width='800px' id='t1'><tr><td>Status code: "+str(HTTP_STATUS['code'])+"</td></tr></table></div>"
 147                         "<p> </p>")
 148             s.wfile.write(html_data)
 149             html = get_process_html_info()
 150             s.wfile.write(html)
 151             s.wfile.write("</body></html>")
 152             s.wfile.close()
 153         except (IOError):
 154             pass
 155
 156     # make processing silent - otherwise will mess up the event handler
 157     def log_message(self, format, *args):
 158         return
 159
 160 def HTTPServerProcess(address, port, http_status, process_state):
 161     global HTTP_STATUS
 162     global PROCESS_STATE
 163
 164     # copy the process status global variable
 165     PROCESS_STATE = process_state
 166     HTTP_STATUS = http_status
 167
 168     server = BaseHTTPServer.HTTPServer
 169     try:
 170         # redirect stdout to stderr so that the HTTP server won't kill the supervised STDIN/STDOUT interface
 171         sys.stdout = sys.stderr
 172         # Create a web server and define the handler to manage the
 173         # incoming request
 174         server = server((address, port), myHTTPHandler)
 175         # Wait forever for incoming http requests
 176         server.serve_forever()
 177
 178     except KeyboardInterrupt:
 179         write_stderr('^C received, shutting down the web server')
 180         server.socket.close()
 181
 182 def dict_print(d):
 183     for proc,data in d.items():
 184         write_stderr(str(proc))
 185         for key,val in data.items():
 186             write_stderr(str(key)+' is '+str(val))
 187
 188 # this is the default logging, only for supervised communication
 189 def write_stdout(s):
 190     # only eventlistener protocol messages may be sent to stdout
 191     sys.stdout.write(s)
 192     sys.stdout.flush()
 193
 194 def write_stderr(s):
 195     global DEBUG
 196     # this can be used for debug logging - stdout not allowed
 197     sys.stderr.write(s)
 198     sys.stderr.flush()
 199
 200 def main():
 201     global REFRESH_TIME
 202     global DEBUG
 203
 204     manager = Manager()
 205     # stores the process status info
 206     PROCESS_STATE = manager.dict()
 207     #HTTP_STATUS_CODE = Value('d', True)
 208     HTTP_STATUS = manager.dict()
 209     HTTP_STATUS['code'] = 200
 210
 211     write_stderr("HTTP STATUS SET TO "+str(HTTP_STATUS['code']))
 212
 213     # default http meta key refresh time in seconds
 214     REFRESH_TIME = 3
 215
 216     # init the default values
 217     ADDRESS = "0.0.0.0"     # bind to all interfaces
 218     PORT = 3000             # web server listen port
 219     DEBUG = False           # no logging
 220
 221     parser = argparse.ArgumentParser()
 222     parser.add_argument('--port', dest='port', help='HTTP server listen port, default 3000', required=False, type=int)
 223     parser.add_argument('--debug', dest='debug', help='sets the debug mode for logging', required=False, action='store_true')
 224     parser.add_argument('--address', dest='address', help='IP listen address (e.g. 172.16.0.3), default all interfaces', required=False, type=str)
 225     parser.add_argument('--refresh', dest='refresh', help='HTTP auto refresh time in second default is 3 seconds', required=False, type=int)
 226     args = parser.parse_args()
 227
 228     if args.port is not None:
 229         PORT = args.port
 230     if args.address is not None:
 231         ADDRESS = args.address
 232     if args.debug is not False:
 233         DEBUG = True;
 234
 235     # Start the http server, bind to address
 236     httpServer = Process(target=HTTPServerProcess, args=(ADDRESS, PORT, HTTP_STATUS, PROCESS_STATE))
 237     httpServer.start()
 238
 239     # set the signal handler this phase
 240     signal.signal(signal.SIGQUIT, doExit)
 241     signal.signal(signal.SIGTERM, doExit)
 242     signal.signal(signal.SIGINT, doExit)
 243     signal.signal(signal.SIGCLD, doExit)
 244
 245     while httpServer.is_alive():
 246         # transition from ACKNOWLEDGED to READY
 247
 248         write_stdout('READY\n')
 249         # read header line and print it to stderr
 250         line = sys.stdin.readline()
 251         write_stderr(line)
 252
 253         # read event payload and print it to stderr
 254         headers = dict([ x.split(':') for x in line.split() ])
 255         process_state = headers['eventname']
 256
 257         if process_state == 'PROCESS_STATE_FATAL':
 258             write_stderr('Status changed to FATAL')
 259             HTTP_STATUS['code'] = 500
 260
 261         short_state = process_state.replace('PROCESS_STATE_', '')
 262         length = int(headers['len'])
 263         data = sys.stdin.read(length)
 264         write_stderr(data)
 265
 266         process = dict([ x.split(':') for x in data.split() ])
 267         pid = 0
 268         if 'pid' in process:
 269             pid = int(process['pid'])
 270         now = datetime.datetime.now()
 271         timestamp=str(now.strftime("%Y/%m/%d %H:%M:%S"))
 272         PROCESS_STATE[process['processname']] = {'time': timestamp, 'state': short_state, 'from_state': process['from_state'],
 273                                                  'pid':pid}
 274         # transition from READY to ACKNOWLEDGED
 275         write_stdout('RESULT 2\nOK')
 276     httpServer.join()
 277
 278 def kill_child_processes():
 279     procs = psutil.Process().children()
 280     # send SIGTERM
 281     for p in procs:
 282         try:
 283             p.terminate()
 284         except psutil.NoSuchProcess:
 285             pass
 286
 287 def doExit(signalNumber, frame):
 288     write_stderr("Got signal: "+str(signalNumber)+" need to exit ...")
 289     kill_child_processes()
 290     exit(0)
 291
 292 if __name__ == '__main__':
 293     main()
 294