1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
import time, json, os, re
import logging, threading
from socket import socket, AF_INET, SOCK_STREAM, SHUT_WR, error as SocketError
import socket
from struct import pack, unpack
from time import sleep
from tornado import gen
from tornado.ioloop import IOLoop
from tornado.web import RequestHandler, Application, url, StaticFileHandler
from rwlock import RWLock
class ActionError(Exception):
pass
def add_monitor(mesg):
if len(reclaimed_ids):
jid = reclaimed_ids.pop()
else:
global id_cnt
jid = id_cnt
id_cnt += 1
if mesg.has_key("name"):
name = mesg["name"]
else:
name = "monitor-{0}".format(jid)
metadata = None
if mesg.has_key("metadata"):
metadata = mesg["metadata"]
stat_res[jid] = {"name": name, "records": [],
"rcnt": 0, "jid": jid, "metadata": metadata}
return json.dumps(jid)
def del_monitor(mesg):
mid = check_id(mesg)
del stat_res[mid]
reclaimed_ids.append(mid)
return ""
def check_id(mesg):
if mesg.has_key("jid"):
try:
jid = int(mesg["jid"])
except ValueError:
raise ActionError("invalid value of jid field")
else:
raise ActionError("jid field not specified")
if not stat_res.has_key(jid):
raise ActionError("the jid does not exist")
logging.info("jid: {0}".format(jid))
return jid
def add_record(mesg):
mid = check_id(mesg)
if mesg.has_key("record"):
rec = stat_res[mid]["records"]
while len(rec) >= MAX_RECORDS:
rec.pop(0)
rec.append({'rid': stat_res[mid]["rcnt"], 'rec': mesg["record"]})
stat_res[mid]["rcnt"] += 1
return ""
def alter_records(mesg):
mid = check_id(mesg)
if mesg.has_key("records"):
stat_res[mid]["records"] = mesg["records"]
return ""
def clear_records(mesg):
mid = check_id(mesg)
stat_res[mid]["records"] = []
return ""
def command_server():
global c, cmd_socket, is_exiting
cmd_socket = socket.socket(AF_INET, SOCK_STREAM)
cmd_socket.bind((HOST, SOCKET_PORT))
cmd_socket.listen(5)
while not is_exiting.isSet():
logging.debug("accepting")
(conn, addr) = cmd_socket.accept()
logging.debug("accepted")
received = conn.recv(4)
if not received:
continue
length, = unpack("<i", received)
if length <= 0 or length > MAX_LENGTH:
logging.warning("invalid header: {0}".format(length))
conn.close() # invalid header
continue
try:#
mesg = json.loads(conn.recv(length))
if not isinstance(mesg, dict):
logging.warning("not a javascript object")
continue
if not mesg.has_key("action"):
raise ActionError("action not specified")
logging.info("action: {0}".format(mesg["action"]))
try:
lock.acquire_write()
conn.send(action_map[mesg["action"]](mesg))
finally:
lock.release()
except ValueError:
logging.warning("malformed json string")
except ActionError as e:
logging.warning(e)
except KeyError as e:
logging.warning("action not found: {0}".format(mesg["action"]))
except SocketError as e:
logging.warning("socket error: {0}".format(e))
finally:
conn.close()
def cmd_shutdown():
global is_exiting, cmd_socket, local_socket_address, cmd
is_exiting.set();
cmd_socket.close()
socket(AF_INET).connect(local_socket_address)
cmd.join()
class AJAXHandler(RequestHandler):
@gen.coroutine
def get(self):
def grab_lock(self, callback=None):
try:
lock.acquire_read()
self.write(stat_res)
finally:
lock.release()
callback()
yield gen.Task(grab_lock, self)
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG,
format='[%(levelname)-7s] (%(threadName)-10s) %(message)s',)
MAX_LENGTH = 65536
MAX_RECORDS = 32
PORT = 2333
SOCKET_PORT = 2334
HOST = ''
lock = RWLock()
stat_res = {}
is_exiting = threading.Event()
reclaimed_ids = []
id_cnt = 0
action_map = {"create": add_monitor,
"drop": del_monitor,
"add": add_record,
"clear": clear_records,
"alter": alter_records}
cmd = threading.Thread(target=command_server, name="local")
cmd.setDaemon(True)
cmd.start()
try:
app = Application([url(r"/ajax", AJAXHandler),
url(r'/()', StaticFileHandler, {'path': "./static/index.html"}),
url(r'/(.*)', StaticFileHandler, {'path': "./static/"})])
app.listen(PORT)
IOLoop.current().start()
except KeyboardInterrupt:
cmd_shutdown()
|