调整dashboard，用户数、主机数、账号数和连接数也都从websocket方式进行获取并订阅通知，以得到实时反馈。
Apex Liu
parent
8d4ecdf487
commit
8bc95e5c17
|
|
@ -13,10 +13,14 @@ $app.on_init = function (cb_stack) {
|
|||
bar_disk: $('#bar-disk')
|
||||
};
|
||||
|
||||
window.onresize = $app.on_screen_resize;
|
||||
$app.stat_counter = {
|
||||
user: -1,
|
||||
host: -1,
|
||||
acc: -1,
|
||||
conn: -1
|
||||
};
|
||||
|
||||
// refresh basic info every 1m.
|
||||
$app.load_basic_info();
|
||||
window.onresize = $app.on_screen_resize;
|
||||
|
||||
$app.ws = null;
|
||||
$app.init_ws();
|
||||
|
|
@ -24,27 +28,6 @@ $app.on_init = function (cb_stack) {
|
|||
cb_stack.exec();
|
||||
};
|
||||
|
||||
$app.load_basic_info = function () {
|
||||
$tp.ajax_post_json('/dashboard/do-get-basic', {},
|
||||
function (ret) {
|
||||
if (ret.code === TPE_OK) {
|
||||
$app.dom.count_user.text(ret.data.count_user);
|
||||
$app.dom.count_host.text(ret.data.count_host);
|
||||
$app.dom.count_acc.text(ret.data.count_acc);
|
||||
$app.dom.count_conn.text(ret.data.count_conn);
|
||||
} else {
|
||||
console.log('do-get-basic failed：' + tp_error_msg(ret.code, ret.message));
|
||||
}
|
||||
|
||||
},
|
||||
function () {
|
||||
console.log('can not connect to server.');
|
||||
}
|
||||
);
|
||||
|
||||
setTimeout($app.load_basic_info, 60 * 1000);
|
||||
};
|
||||
|
||||
$app.init_sys_status_info = function (data) {
|
||||
var i = 0;
|
||||
|
||||
|
|
@ -101,8 +84,6 @@ $app.init_sys_status_info = function (data) {
|
|||
}
|
||||
};
|
||||
|
||||
console.log(axis_size_cfg);
|
||||
|
||||
//=====================================
|
||||
// CPU
|
||||
//=====================================
|
||||
|
|
@ -405,6 +386,48 @@ $app.init_sys_status_info = function (data) {
|
|||
|
||||
};
|
||||
|
||||
$app.update_stat_counter = function (data) {
|
||||
if ($app.stat_counter.user === -1) {
|
||||
$app.stat_counter.user = data.user;
|
||||
$app.dom.count_user.text(data.user);
|
||||
} else if ($app.stat_counter.user !== data.user) {
|
||||
$app.stat_counter.user = data.user;
|
||||
$app.dom.count_user.fadeOut(300, function () {
|
||||
$app.dom.count_user.text(data.user).fadeIn(400);
|
||||
});
|
||||
}
|
||||
|
||||
if ($app.stat_counter.host === -1) {
|
||||
$app.stat_counter.host = data.host;
|
||||
$app.dom.count_host.text(data.host);
|
||||
} else if ($app.stat_counter.host !== data.host) {
|
||||
$app.stat_counter.host = data.host;
|
||||
$app.dom.count_host.fadeOut(300, function () {
|
||||
$app.dom.count_host.text(data.host).fadeIn(400);
|
||||
});
|
||||
}
|
||||
|
||||
if ($app.stat_counter.acc === -1) {
|
||||
$app.stat_counter.acc = data.acc;
|
||||
$app.dom.count_acc.text(data.acc);
|
||||
} else if ($app.stat_counter.acc !== data.acc) {
|
||||
$app.stat_counter.acc = data.acc;
|
||||
$app.dom.count_acc.fadeOut(300, function () {
|
||||
$app.dom.count_acc.text(data.acc).fadeIn(400);
|
||||
});
|
||||
}
|
||||
|
||||
if ($app.stat_counter.conn === -1) {
|
||||
$app.stat_counter.conn = data.conn;
|
||||
$app.dom.count_conn.text(data.conn);
|
||||
} else if ($app.stat_counter.conn !== data.conn) {
|
||||
$app.stat_counter.conn = data.conn;
|
||||
$app.dom.count_conn.fadeOut(300, function () {
|
||||
$app.dom.count_conn.text(data.conn).fadeIn(400);
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
$app.init_ws = function () {
|
||||
if ($app.ws !== null)
|
||||
delete $app.ws;
|
||||
|
|
@ -414,16 +437,13 @@ $app.init_ws = function () {
|
|||
|
||||
$app.ws.onopen = function (e) {
|
||||
$app.ws.send('{"method": "request", "param": "sys_status"}');
|
||||
// 订阅：
|
||||
// $app.ws.send('{"method": "subscribe", "params": ["sys_status"]}');
|
||||
$app.ws.send('{"method": "request", "param": "stat_counter"}');
|
||||
};
|
||||
$app.ws.onclose = function (e) {
|
||||
// console.log('[ws] ws-on-close', e);
|
||||
setTimeout($app.init_ws, 5000);
|
||||
};
|
||||
$app.ws.onmessage = function (e) {
|
||||
var t = JSON.parse(e.data);
|
||||
// console.log(t);
|
||||
|
||||
if (t.method === 'request' && t.param === 'sys_status') {
|
||||
$app.init_sys_status_info(t.data);
|
||||
|
|
@ -432,6 +452,13 @@ $app.init_ws = function () {
|
|||
return;
|
||||
}
|
||||
|
||||
if (t.method === 'request' && t.param === 'stat_counter') {
|
||||
$app.update_stat_counter(t.data);
|
||||
// 订阅数量信息
|
||||
$app.ws.send('{"method": "subscribe", "params": ["stat_counter"]}');
|
||||
return;
|
||||
}
|
||||
|
||||
if (t.method === 'subscribe' && t.param === 'sys_status') {
|
||||
$app.bar_cpu_user.shift();
|
||||
$app.bar_cpu_user.push({name: tp_format_datetime_ms(tp_utc2local_ms(t.data.t), 'HH:mm:ss'), value: [t.data.t, t.data.cpu.u]});
|
||||
|
|
@ -463,6 +490,10 @@ $app.init_ws = function () {
|
|||
{series: [{data: $app.bar_disk_read}, {data: $app.bar_disk_write}]}
|
||||
);
|
||||
}
|
||||
|
||||
if (t.method === 'subscribe' && t.param === 'stat_counter') {
|
||||
$app.update_stat_counter(t.data);
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,160 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import psutil
|
||||
from app.base.utils import tp_utc_timestamp_ms
|
||||
from app.const import *
|
||||
from app.base.wss import tp_wss
|
||||
from app.base.cron import tp_corn
|
||||
from app.model import stats
|
||||
|
||||
|
||||
class TPStats(object):
|
||||
_INTERVAL = 5 # seconds
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
import builtins
|
||||
if '__tp_stats__' in builtins.__dict__:
|
||||
raise RuntimeError('TPStats object exists, you can not create more than one instance.')
|
||||
|
||||
# 实时数据我们在内存中保留最近10分钟的数据，每5秒收集一次，共 10*60/5 = 120 条记录
|
||||
self._sys_stats = list()
|
||||
|
||||
# 网络流量和磁盘IO是递增的数据，因此要记下上一次采集的数据，以便计算间隔时间内的增量
|
||||
self._net_recv = 0
|
||||
self._net_sent = 0
|
||||
self._disk_read = 0
|
||||
self._disk_write = 0
|
||||
|
||||
self._counter_stats = {
|
||||
'user': 0,
|
||||
'host': 0,
|
||||
'acc': 0,
|
||||
'conn': 0
|
||||
}
|
||||
|
||||
def init(self):
|
||||
t = tp_utc_timestamp_ms() - 10 * 60 * 1000
|
||||
cnt = int((10 * 60 + self._INTERVAL - 1) / self._INTERVAL)
|
||||
for i in range(cnt):
|
||||
val = {
|
||||
't': t,
|
||||
'cpu': {'u': 0, 's': 0},
|
||||
'mem': {'u': 1, 't': 100},
|
||||
'disk': {'r': 0, 'w': 0},
|
||||
'net': {'r': 0, 's': 0}
|
||||
}
|
||||
self._sys_stats.append(val)
|
||||
t += self._INTERVAL * 1000
|
||||
|
||||
psutil.cpu_times_percent()
|
||||
net = psutil.net_io_counters(pernic=False)
|
||||
self._net_recv = net.bytes_recv
|
||||
self._net_sent = net.bytes_sent
|
||||
disk = psutil.disk_io_counters(perdisk=False)
|
||||
self._disk_read = disk.read_bytes
|
||||
self._disk_write = disk.write_bytes
|
||||
|
||||
err, c = stats.get_basic_stats()
|
||||
if TPE_OK == err:
|
||||
self._counter_stats = c
|
||||
|
||||
# 每 5秒 采集一次系统状态统计数据
|
||||
tp_corn().add_job('sys_status', self._check_sys_stats, first_interval_seconds=self._INTERVAL, interval_seconds=self._INTERVAL)
|
||||
# 每 一小时 重新查询一次数据库，得到用户数/主机数/账号数/连接数，避免统计数量出现偏差
|
||||
tp_corn().add_job('query_counter', self._query_counter, first_interval_seconds=60 * 60, interval_seconds=60 * 60)
|
||||
tp_wss().register_get_sys_status_callback(self.get_sys_stats)
|
||||
tp_wss().register_get_stat_counter_callback(self.get_counter_stats)
|
||||
|
||||
return True
|
||||
|
||||
def _check_sys_stats(self):
|
||||
val = {'t': tp_utc_timestamp_ms()}
|
||||
|
||||
cpu = psutil.cpu_times_percent()
|
||||
val['cpu'] = {'u': cpu.user, 's': cpu.system}
|
||||
|
||||
mem = psutil.virtual_memory()
|
||||
val['mem'] = {'u': mem.used, 't': mem.total}
|
||||
|
||||
disk = psutil.disk_io_counters(perdisk=False)
|
||||
_read = disk.read_bytes - self._disk_read
|
||||
_write = disk.write_bytes - self._disk_write
|
||||
self._disk_read = disk.read_bytes
|
||||
self._disk_write = disk.write_bytes
|
||||
|
||||
if _read < 0:
|
||||
_read = 0
|
||||
if _write < 0:
|
||||
_write = 0
|
||||
val['disk'] = {'r': int(_read / self._INTERVAL), 'w': int(_write / self._INTERVAL)}
|
||||
|
||||
net = psutil.net_io_counters(pernic=False)
|
||||
_recv = net.bytes_recv - self._net_recv
|
||||
_sent = net.bytes_sent - self._net_sent
|
||||
self._net_recv = net.bytes_recv
|
||||
self._net_sent = net.bytes_sent
|
||||
|
||||
# On some systems such as Linux, on a very busy or long-lived system, the numbers
|
||||
# returned by the kernel may overflow and wrap (restart from zero)
|
||||
if _recv < 0:
|
||||
_recv = 0
|
||||
if _sent < 0:
|
||||
_sent = 0
|
||||
val['net'] = {'r': int(_recv / self._INTERVAL), 's': int(_sent / self._INTERVAL)}
|
||||
|
||||
self._sys_stats.pop(0)
|
||||
self._sys_stats.append(val)
|
||||
|
||||
tp_wss().send_message('sys_status', val)
|
||||
|
||||
def _query_counter(self):
|
||||
# 直接从数据库中查询数据，避免长时间运行后计数不准确
|
||||
err, c = stats.get_basic_stats()
|
||||
if TPE_OK == err:
|
||||
self._counter_stats = c
|
||||
tp_wss().send_message('stat_counter', self._counter_stats)
|
||||
|
||||
def get_sys_stats(self):
|
||||
return self._sys_stats
|
||||
|
||||
def get_counter_stats(self):
|
||||
return self._counter_stats
|
||||
|
||||
def user_counter_change(self, alt_count):
|
||||
self._counter_stats['user'] += alt_count
|
||||
if self._counter_stats['user'] < 0:
|
||||
self._counter_stats['user'] = 0
|
||||
tp_wss().send_message('stat_counter', self._counter_stats)
|
||||
|
||||
def host_counter_change(self, alt_count):
|
||||
self._counter_stats['host'] += alt_count
|
||||
if self._counter_stats['host'] < 0:
|
||||
self._counter_stats['host'] = 0
|
||||
tp_wss().send_message('stat_counter', self._counter_stats)
|
||||
|
||||
def acc_counter_change(self, alt_count):
|
||||
self._counter_stats['acc'] += alt_count
|
||||
if self._counter_stats['acc'] < 0:
|
||||
self._counter_stats['acc'] = 0
|
||||
tp_wss().send_message('stat_counter', self._counter_stats)
|
||||
|
||||
def conn_counter_change(self, alt_count):
|
||||
self._counter_stats['conn'] += alt_count
|
||||
if self._counter_stats['conn'] < 0:
|
||||
self._counter_stats['conn'] = 0
|
||||
tp_wss().send_message('stat_counter', self._counter_stats)
|
||||
|
||||
|
||||
def tp_stats():
|
||||
"""
|
||||
取得 TPSysStatus 的唯一实例
|
||||
|
||||
:rtype : TPStats
|
||||
"""
|
||||
|
||||
import builtins
|
||||
if '__tp_stats__' not in builtins.__dict__:
|
||||
builtins.__dict__['__tp_stats__'] = TPStats()
|
||||
return builtins.__dict__['__tp_stats__']
|
||||
|
|
@ -1,107 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import psutil
|
||||
from app.base.utils import tp_timestamp_utc_now, tp_utc_timestamp_ms
|
||||
from app.controller.ws import tp_wss
|
||||
from app.base.cron import tp_corn
|
||||
|
||||
|
||||
class TPSysStatus(object):
|
||||
_INTERVAL = 5 # seconds
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
import builtins
|
||||
if '__tp_sys_status__' in builtins.__dict__:
|
||||
raise RuntimeError('TPSysStatus object exists, you can not create more than one instance.')
|
||||
|
||||
# 实时数据我们在内存中保留最近10分钟的数据，每5秒收集一次，共 10*60/5 = 120 条记录
|
||||
self._history = list()
|
||||
|
||||
self._disk_read = 0
|
||||
self._disk_write = 0
|
||||
self._net_recv = 0
|
||||
self._net_sent = 0
|
||||
|
||||
def init(self):
|
||||
t = tp_utc_timestamp_ms() - 10 * 60 * 1000
|
||||
cnt = int((10 * 60 + self._INTERVAL - 1) / self._INTERVAL)
|
||||
for i in range(cnt):
|
||||
val = {
|
||||
't': t,
|
||||
'cpu': {'u': 0, 's': 0},
|
||||
'mem': {'u': 1, 't': 100},
|
||||
'disk': {'r': 0, 'w': 0},
|
||||
'net': {'r': 0, 's': 0}
|
||||
}
|
||||
self._history.append(val)
|
||||
t += self._INTERVAL*1000
|
||||
|
||||
psutil.cpu_times_percent()
|
||||
net = psutil.net_io_counters(pernic=False)
|
||||
self._net_recv = net.bytes_recv
|
||||
self._net_sent = net.bytes_sent
|
||||
disk = psutil.disk_io_counters(perdisk=False)
|
||||
self._disk_read = disk.read_bytes
|
||||
self._disk_write = disk.write_bytes
|
||||
|
||||
tp_corn().add_job('sys_status', self._check_status, first_interval_seconds=self._INTERVAL, interval_seconds=self._INTERVAL)
|
||||
tp_wss().register_get_sys_status_callback(self.get_status)
|
||||
return True
|
||||
|
||||
def _check_status(self):
|
||||
val = {'t': tp_utc_timestamp_ms()}
|
||||
|
||||
cpu = psutil.cpu_times_percent()
|
||||
val['cpu'] = {'u': cpu.user, 's': cpu.system}
|
||||
|
||||
mem = psutil.virtual_memory()
|
||||
val['mem'] = {'u': mem.used, 't': mem.total}
|
||||
|
||||
disk = psutil.disk_io_counters(perdisk=False)
|
||||
_read = disk.read_bytes - self._disk_read
|
||||
_write = disk.write_bytes - self._disk_write
|
||||
self._disk_read = disk.read_bytes
|
||||
self._disk_write = disk.write_bytes
|
||||
|
||||
if _read < 0:
|
||||
_read = 0
|
||||
if _write < 0:
|
||||
_write = 0
|
||||
val['disk'] = {'r': int(_read / self._INTERVAL), 'w': int(_write / self._INTERVAL)}
|
||||
|
||||
net = psutil.net_io_counters(pernic=False)
|
||||
_recv = net.bytes_recv - self._net_recv
|
||||
_sent = net.bytes_sent - self._net_sent
|
||||
self._net_recv = net.bytes_recv
|
||||
self._net_sent = net.bytes_sent
|
||||
|
||||
# On some systems such as Linux, on a very busy or long-lived system, the numbers
|
||||
# returned by the kernel may overflow and wrap (restart from zero)
|
||||
if _recv < 0:
|
||||
_recv = 0
|
||||
if _sent < 0:
|
||||
_sent = 0
|
||||
val['net'] = {'r': int(_recv / self._INTERVAL), 's': int(_sent / self._INTERVAL)}
|
||||
|
||||
self._history.pop(0)
|
||||
self._history.append(val)
|
||||
|
||||
tp_wss().send_message('sys_status', val)
|
||||
|
||||
def get_status(self):
|
||||
return self._history
|
||||
|
||||
|
||||
def tp_sys_status():
|
||||
"""
|
||||
取得TPSysStatus管理器的唯一实例
|
||||
|
||||
:rtype : TPSysStatus
|
||||
"""
|
||||
|
||||
import builtins
|
||||
if '__tp_sys_status__' not in builtins.__dict__:
|
||||
builtins.__dict__['__tp_sys_status__'] = TPSysStatus()
|
||||
return builtins.__dict__['__tp_sys_status__']
|
||||
|
|
@ -17,7 +17,7 @@ from app.base.db import get_db
|
|||
from app.base.logger import log
|
||||
from app.base.session import tp_session
|
||||
from app.base.cron import tp_corn
|
||||
from app.base.status import tp_sys_status
|
||||
from app.base.stats import tp_stats
|
||||
|
||||
|
||||
class WebApp:
|
||||
|
|
@ -94,7 +94,7 @@ class WebApp:
|
|||
if not tp_session().init():
|
||||
log.e('can not initialize session manager.\n')
|
||||
return 0
|
||||
if not tp_sys_status().init():
|
||||
if not tp_stats().init():
|
||||
log.e('can not initialize system status collector.\n')
|
||||
return 0
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,94 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import json
|
||||
import threading
|
||||
|
||||
from app.base.logger import *
|
||||
|
||||
|
||||
class TPWebSocketServer(object):
|
||||
_clients = {}
|
||||
_lock = threading.RLock()
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
import builtins
|
||||
if '__tp_websocket_server__' in builtins.__dict__:
|
||||
raise RuntimeError('TPWebSocketServer object exists, you can not create more than one instance.')
|
||||
|
||||
self._cb_get_sys_status = None
|
||||
self._cb_get_stat_counter = None
|
||||
|
||||
def register_get_sys_status_callback(self, cb):
|
||||
self._cb_get_sys_status = cb
|
||||
|
||||
def register_get_stat_counter_callback(self, cb):
|
||||
self._cb_get_stat_counter = cb
|
||||
|
||||
def have_callbacker(self, callbacker):
|
||||
return callbacker in self._clients
|
||||
|
||||
def register(self, callbacker):
|
||||
# 记录客户端连接实例
|
||||
with self._lock:
|
||||
if not self.have_callbacker(callbacker):
|
||||
self._clients[callbacker] = {'subscribe': []}
|
||||
|
||||
def unregister(self, callbacker):
|
||||
with self._lock:
|
||||
# 删除客户端连接实例
|
||||
try:
|
||||
del self._clients[callbacker]
|
||||
except:
|
||||
# print('when unregister, not exists.')
|
||||
pass
|
||||
|
||||
def on_message(self, callbacker, message):
|
||||
print('got message', message)
|
||||
try:
|
||||
req = json.loads(message)
|
||||
except:
|
||||
log.e('need json-format request.\n')
|
||||
return
|
||||
|
||||
if req['method'] == 'subscribe':
|
||||
for p in req['params']:
|
||||
if p not in self._clients[callbacker]['subscribe']:
|
||||
self._clients[callbacker]['subscribe'].append(p)
|
||||
elif req['method'] == 'request':
|
||||
if req['param'] == 'sys_status':
|
||||
if self._cb_get_sys_status is not None:
|
||||
message = self._cb_get_sys_status()
|
||||
msg = {'method': 'request', 'param': 'sys_status', 'data': message}
|
||||
s = json.dumps(msg, separators=(',', ':'))
|
||||
callbacker.write_message(s)
|
||||
if req['param'] == 'stat_counter':
|
||||
if self._cb_get_stat_counter is not None:
|
||||
message = self._cb_get_stat_counter()
|
||||
msg = {'method': 'request', 'param': 'stat_counter', 'data': message}
|
||||
s = json.dumps(msg, separators=(',', ':'))
|
||||
callbacker.write_message(s)
|
||||
|
||||
def send_message(self, subscribe, message):
|
||||
s = None
|
||||
with self._lock:
|
||||
for c in self._clients:
|
||||
if subscribe in self._clients[c]['subscribe']:
|
||||
if s is None:
|
||||
msg = {'method': 'subscribe', 'param': subscribe, 'data': message}
|
||||
s = json.dumps(msg, separators=(',', ':'))
|
||||
c.write_message(s)
|
||||
|
||||
|
||||
def tp_wss():
|
||||
"""
|
||||
取得 TPWebSocketServer 管理器的唯一实例
|
||||
|
||||
:rtype : TPWebSocketServer
|
||||
"""
|
||||
|
||||
import builtins
|
||||
if '__tp_websocket_server__' not in builtins.__dict__:
|
||||
builtins.__dict__['__tp_websocket_server__'] = TPWebSocketServer()
|
||||
return builtins.__dict__['__tp_websocket_server__']
|
||||
|
|
@ -25,8 +25,6 @@ controllers = [
|
|||
# ====================================================
|
||||
# - 控制台页面
|
||||
(r'/dashboard', dashboard.IndexHandler),
|
||||
# - [json] 获取基本信息
|
||||
(r'/dashboard/do-get-basic', dashboard.DoGetBasicHandler),
|
||||
|
||||
# ====================================================
|
||||
# 外部调用接口
|
||||
|
|
|
|||
|
|
@ -12,8 +12,6 @@ from app.const import *
|
|||
from app.model import account
|
||||
from app.model import group
|
||||
|
||||
# cfg = tp_cfg()
|
||||
|
||||
# 临时认证ID的基数，每次使用时均递减
|
||||
tmp_auth_id_base = -1
|
||||
tmp_auth_id_lock = threading.RLock()
|
||||
|
|
@ -293,30 +291,3 @@ class DoUpdateAccountsHandler(TPBaseJsonHandler):
|
|||
return self.write_json(err)
|
||||
else:
|
||||
return self.write_json(TPE_PARAM)
|
||||
|
||||
# class DoRemoveAccountHandler(TPBaseJsonHandler):
|
||||
# def post(self):
|
||||
# ret = self.check_privilege(TP_PRIVILEGE_ACCOUNT | TP_PRIVILEGE_ACCOUNT_GROUP)
|
||||
# if ret != TPE_OK:
|
||||
# return
|
||||
#
|
||||
# args = self.get_argument('args', None)
|
||||
# if args is None:
|
||||
# return self.write_json(TPE_PARAM)
|
||||
# try:
|
||||
# args = json.loads(args)
|
||||
# except:
|
||||
# return self.write_json(TPE_JSON_FORMAT)
|
||||
#
|
||||
# try:
|
||||
# host_id = int(args['host_id'])
|
||||
# acc_id = int(args['acc_id'])
|
||||
# except:
|
||||
# log.e('\n')
|
||||
# return self.write_json(TPE_PARAM)
|
||||
#
|
||||
# if host_id <= 0 or acc_id <= 0:
|
||||
# return self.write_json(TPE_PARAM)
|
||||
#
|
||||
# err = account.delete_account(self, host_id, acc_id)
|
||||
# self.write_json(err)
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
|
||||
from app.const import *
|
||||
from app.base.controller import TPBaseHandler, TPBaseJsonHandler
|
||||
from app.model import stat
|
||||
from app.model import stats
|
||||
|
||||
|
||||
class IndexHandler(TPBaseHandler):
|
||||
|
|
@ -12,22 +12,3 @@ class IndexHandler(TPBaseHandler):
|
|||
return
|
||||
|
||||
self.render('dashboard/index.mako')
|
||||
|
||||
|
||||
class DoGetBasicHandler(TPBaseJsonHandler):
|
||||
def post(self):
|
||||
ret = self.check_privilege(TP_PRIVILEGE_LOGIN_WEB)
|
||||
if ret != TPE_OK:
|
||||
return
|
||||
|
||||
err, info = stat.get_basic()
|
||||
if err != TPE_OK:
|
||||
return self.write_json(err)
|
||||
|
||||
# ret = dict()
|
||||
# ret['count_user'] = 5
|
||||
# ret['count_host'] = 5
|
||||
# ret['count_acc'] = 5
|
||||
# ret['count_conn'] = 5
|
||||
self.write_json(TPE_OK, data=info)
|
||||
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from app.base.configs import tp_cfg
|
|||
from app.base.session import tp_session
|
||||
from app.base.core_server import core_service_async_post_http
|
||||
from app.model import record
|
||||
from app.base.stats import tp_stats
|
||||
from app.base.logger import *
|
||||
from app.base.controller import TPBaseJsonHandler
|
||||
|
||||
|
|
@ -63,7 +64,10 @@ class RpcHandler(TPBaseJsonHandler):
|
|||
|
||||
conn_id = param['conn_id']
|
||||
x = tp_session().taken('tmp-conn-info-{}'.format(conn_id), None)
|
||||
return self.write_json(0, data=x)
|
||||
if x is None:
|
||||
return self.write_json(TPE_NOT_EXISTS)
|
||||
else:
|
||||
return self.write_json(TPE_OK, data=x)
|
||||
|
||||
def _session_begin(self, param):
|
||||
try:
|
||||
|
|
@ -87,6 +91,7 @@ class RpcHandler(TPBaseJsonHandler):
|
|||
if err != TPE_OK:
|
||||
return self.write_json(err, message='can not write database.')
|
||||
else:
|
||||
tp_stats().conn_counter_change(1)
|
||||
return self.write_json(TPE_OK, data={'rid': record_id})
|
||||
|
||||
def _session_update(self, param):
|
||||
|
|
@ -106,19 +111,20 @@ class RpcHandler(TPBaseJsonHandler):
|
|||
|
||||
def _session_end(self, param):
|
||||
if 'rid' not in param or 'code' not in param:
|
||||
return self.write_json(-1, message='invalid request.')
|
||||
return self.write_json(TPE_PARAM, message='invalid request.')
|
||||
|
||||
if not record.session_end(param['rid'], param['code']):
|
||||
return self.write_json(-1, 'can not write database.')
|
||||
return self.write_json(TPE_DATABASE, 'can not write database.')
|
||||
else:
|
||||
return self.write_json(0)
|
||||
tp_stats().conn_counter_change(-1)
|
||||
return self.write_json(TPE_OK)
|
||||
|
||||
def _register_core(self, param):
|
||||
# 因为core服务启动了（之前可能非正常终止了），做一下数据库中会话状态的修复操作
|
||||
record.session_fix()
|
||||
|
||||
if 'rpc' not in param:
|
||||
return self.write_json(-1, 'invalid param.')
|
||||
return self.write_json(TPE_PARAM, 'invalid param.')
|
||||
|
||||
tp_cfg().common.core_server_rpc = param['rpc']
|
||||
|
||||
|
|
@ -132,8 +138,8 @@ class RpcHandler(TPBaseJsonHandler):
|
|||
log.d('update base server config info.\n')
|
||||
tp_cfg().update_core(ret_data)
|
||||
|
||||
return self.write_json(0)
|
||||
return self.write_json(TPE_OK)
|
||||
|
||||
def _exit(self):
|
||||
# set exit flag.
|
||||
return self.write_json(0)
|
||||
return self.write_json(TPE_OK)
|
||||
|
|
|
|||
|
|
@ -1,101 +1,15 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import json
|
||||
# import urllib.parse
|
||||
import threading
|
||||
|
||||
# import tornado.gen
|
||||
from app.const import *
|
||||
# from app.base.configs import tp_cfg
|
||||
from app.base.session import tp_session
|
||||
# from app.base.core_server import core_service_async_post_http
|
||||
# from app.model import record
|
||||
from app.base.logger import *
|
||||
# from app.base.controller import TPBaseJsonHandler
|
||||
import tornado.websocket
|
||||
|
||||
|
||||
class TPWebSocketServer(object):
|
||||
_clients = {}
|
||||
_lock = threading.RLock()
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
import builtins
|
||||
if '__tp_websocket_server__' in builtins.__dict__:
|
||||
raise RuntimeError('TPWebSocketServer object exists, you can not create more than one instance.')
|
||||
|
||||
self._cb_get_sys_status = None
|
||||
|
||||
def register_get_sys_status_callback(self, cb):
|
||||
self._cb_get_sys_status = cb
|
||||
|
||||
def have_callbacker(self, callbacker):
|
||||
return callbacker in self._clients
|
||||
|
||||
def register(self, callbacker):
|
||||
# 记录客户端连接实例
|
||||
with self._lock:
|
||||
if not self.have_callbacker(callbacker):
|
||||
self._clients[callbacker] = {'subscribe': []}
|
||||
|
||||
def unregister(self, callbacker):
|
||||
with self._lock:
|
||||
# 删除客户端连接实例
|
||||
try:
|
||||
del self._clients[callbacker]
|
||||
except:
|
||||
# print('when unregister, not exists.')
|
||||
pass
|
||||
|
||||
def on_message(self, callbacker, message):
|
||||
print('got message', message)
|
||||
try:
|
||||
req = json.loads(message)
|
||||
except:
|
||||
log.e('need json-format request.\n')
|
||||
return
|
||||
|
||||
if req['method'] == 'subscribe':
|
||||
for p in req['params']:
|
||||
if p not in self._clients[callbacker]['subscribe']:
|
||||
self._clients[callbacker]['subscribe'].append(p)
|
||||
elif req['method'] == 'request':
|
||||
if req['param'] == 'sys_status':
|
||||
if self._cb_get_sys_status is not None:
|
||||
message = self._cb_get_sys_status()
|
||||
msg = {'method': 'request', 'param': 'sys_status', 'data': message}
|
||||
s = json.dumps(msg, separators=(',', ':'))
|
||||
callbacker.write_message(s)
|
||||
|
||||
def send_message(self, subscribe, message):
|
||||
s = None
|
||||
with self._lock:
|
||||
for c in self._clients:
|
||||
if subscribe in self._clients[c]['subscribe']:
|
||||
if s is None:
|
||||
msg = {'method': 'subscribe', 'param': subscribe, 'data': message}
|
||||
s = json.dumps(msg, separators=(',', ':'))
|
||||
c.write_message(s)
|
||||
|
||||
|
||||
def tp_wss():
|
||||
"""
|
||||
取得 TPWebSocketServer 管理器的唯一实例
|
||||
|
||||
:rtype : TPWebSocketServer
|
||||
"""
|
||||
|
||||
import builtins
|
||||
if '__tp_websocket_server__' not in builtins.__dict__:
|
||||
builtins.__dict__['__tp_websocket_server__'] = TPWebSocketServer()
|
||||
return builtins.__dict__['__tp_websocket_server__']
|
||||
from app.base.session import tp_session
|
||||
from app.base.wss import tp_wss
|
||||
from app.const import *
|
||||
|
||||
|
||||
class WebSocketHandler(tornado.websocket.WebSocketHandler):
|
||||
def check_origin(self, origin): # 针对websocket处理类重写同源检查的方法
|
||||
# print('check_origin: ', origin)
|
||||
return True
|
||||
|
||||
# 接受websocket链接，保存链接实例
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ from app.base.logger import log
|
|||
from app.base.db import get_db, SQL
|
||||
from . import syslog
|
||||
from app.base.utils import tp_timestamp_utc_now
|
||||
from app.base.stats import tp_stats
|
||||
|
||||
|
||||
def get_account_info(acc_id):
|
||||
|
|
@ -215,6 +216,8 @@ def add_account(handler, host_id, args):
|
|||
# if not db_ret:
|
||||
# return TPE_DATABASE, 0
|
||||
|
||||
tp_stats().acc_counter_change(1)
|
||||
|
||||
return TPE_OK, _id
|
||||
|
||||
|
||||
|
|
@ -349,6 +352,8 @@ def remove_accounts(handler, host_id, acc_ids):
|
|||
|
||||
syslog.sys_log(handler.get_current_user(), handler.request.remote_ip, TPE_OK, "删除账号：{}".format('，'.join(acc_names)))
|
||||
|
||||
tp_stats().acc_counter_change(-1)
|
||||
|
||||
return TPE_OK
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ from app.const import *
|
|||
from app.base.logger import log
|
||||
from app.base.db import get_db, SQL
|
||||
from . import syslog
|
||||
from app.base.stats import tp_stats
|
||||
from app.base.utils import tp_timestamp_utc_now
|
||||
|
||||
|
||||
|
|
@ -113,6 +114,7 @@ def add_host(handler, args):
|
|||
if len(args['router_ip']) > 0:
|
||||
h_name += '（由{}:{}路由）'.format(args['router_ip'], args['router_port'])
|
||||
syslog.sys_log(handler.get_current_user(), handler.request.remote_ip, TPE_OK, "创建主机：{}".format(h_name))
|
||||
tp_stats().host_counter_change(1)
|
||||
|
||||
return TPE_OK, _id
|
||||
|
||||
|
|
@ -203,8 +205,10 @@ def remove_hosts(handler, hosts):
|
|||
|
||||
if len(acc_names) > 0:
|
||||
syslog.sys_log(handler.get_current_user(), handler.request.remote_ip, TPE_OK, "删除账号：{}".format('，'.join(acc_names)))
|
||||
tp_stats().acc_counter_change(0 - len(acc_names))
|
||||
if len(host_names) > 0:
|
||||
syslog.sys_log(handler.get_current_user(), handler.request.remote_ip, TPE_OK, "删除主机：{}".format('，'.join(host_names)))
|
||||
tp_stats().host_counter_change(0 - len(host_names))
|
||||
|
||||
return TPE_OK
|
||||
|
||||
|
|
@ -266,6 +270,7 @@ def update_hosts_state(handler, host_ids, state):
|
|||
else:
|
||||
return TPE_DATABASE
|
||||
|
||||
|
||||
#
|
||||
# def unlock_hosts(handler, host_ids):
|
||||
# db = get_db()
|
||||
|
|
|
|||
|
|
@ -1,88 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
from app.const import *
|
||||
from app.base.db import get_db, SQL
|
||||
from app.base.logger import log
|
||||
from app.base.utils import tp_timestamp_utc_now
|
||||
|
||||
|
||||
def get_basic():
|
||||
db = get_db()
|
||||
|
||||
ret = {'count_user': 0,
|
||||
'count_host': 0,
|
||||
'count_acc': 0,
|
||||
'count_conn': 0
|
||||
}
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}user`;'.format(tpdb=db.table_prefix)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['count_user'] = db_ret[0][0]
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}host`;'.format(tpdb=db.table_prefix)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['count_host'] = db_ret[0][0]
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}acc`;'.format(tpdb=db.table_prefix)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['count_acc'] = db_ret[0][0]
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}record` WHERE `state` IN ({sess_running},{sess_started});'.format(tpdb=db.table_prefix, sess_running=TP_SESS_STAT_RUNNING, sess_started=TP_SESS_STAT_STARTED)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['count_conn'] = db_ret[0][0]
|
||||
|
||||
return TPE_OK, ret
|
||||
|
||||
|
||||
def get_logs(sql_filter, sql_order, sql_limit):
|
||||
s = SQL(get_db())
|
||||
s.select_from('syslog', ['id', 'user_name', 'user_surname', 'client_ip', 'code', 'log_time', 'message'], alt_name='l')
|
||||
|
||||
str_where = ''
|
||||
_where = list()
|
||||
|
||||
if len(sql_filter) > 0:
|
||||
for k in sql_filter:
|
||||
if k == 'log_user_name':
|
||||
_where.append('l.user_name="{}"'.format(sql_filter[k]))
|
||||
# elif k == 'search_record':
|
||||
# _where.append('(h.name LIKE "%{}%" OR h.ip LIKE "%{}%" OR h.router_addr LIKE "%{}%" OR h.desc LIKE "%{}%" OR h.cid LIKE "%{}%")'.format(sql_filter[k], sql_filter[k], sql_filter[k], sql_filter[k], sql_filter[k]))
|
||||
|
||||
if len(_where) > 0:
|
||||
str_where = '( {} )'.format(' AND '.join(_where))
|
||||
|
||||
s.where(str_where)
|
||||
|
||||
if sql_order is not None:
|
||||
_sort = False if not sql_order['asc'] else True
|
||||
if 'log_time' == sql_order['name']:
|
||||
s.order_by('l.log_time', _sort)
|
||||
# elif 'name' == sql_order['name']:
|
||||
# s.order_by('h.name', _sort)
|
||||
# elif 'os_type' == sql_order['name']:
|
||||
# s.order_by('h.os_type', _sort)
|
||||
# elif 'cid' == sql_order['name']:
|
||||
# s.order_by('h.cid', _sort)
|
||||
# elif 'state' == sql_order['name']:
|
||||
# s.order_by('h.state', _sort)
|
||||
else:
|
||||
log.e('unknown order field: {}\n'.format(sql_order['name']))
|
||||
return TPE_PARAM, s.total_count, s.recorder
|
||||
|
||||
if len(sql_limit) > 0:
|
||||
s.limit(sql_limit['page_index'], sql_limit['per_page'])
|
||||
|
||||
err = s.query()
|
||||
return err, s.total_count, s.recorder
|
||||
|
|
@ -0,0 +1,86 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
from app.const import *
|
||||
from app.base.db import get_db
|
||||
|
||||
|
||||
def get_basic_stats():
|
||||
db = get_db()
|
||||
|
||||
ret = {'user': 0,
|
||||
'host': 0,
|
||||
'acc': 0,
|
||||
'conn': 0
|
||||
}
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}user`;'.format(tpdb=db.table_prefix)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['user'] = db_ret[0][0]
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}host`;'.format(tpdb=db.table_prefix)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['host'] = db_ret[0][0]
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}acc`;'.format(tpdb=db.table_prefix)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['acc'] = db_ret[0][0]
|
||||
|
||||
sql = 'SELECT COUNT(*) FROM `{tpdb}record` WHERE `state` IN ({sess_running},{sess_started});'.format(tpdb=db.table_prefix, sess_running=TP_SESS_STAT_RUNNING, sess_started=TP_SESS_STAT_STARTED)
|
||||
db_ret = db.query(sql)
|
||||
if not db_ret or len(db_ret) == 0:
|
||||
pass
|
||||
else:
|
||||
ret['conn'] = db_ret[0][0]
|
||||
|
||||
return TPE_OK, ret
|
||||
|
||||
|
||||
# def get_logs(sql_filter, sql_order, sql_limit):
|
||||
# s = SQL(get_db())
|
||||
# s.select_from('syslog', ['id', 'user_name', 'user_surname', 'client_ip', 'code', 'log_time', 'message'], alt_name='l')
|
||||
#
|
||||
# str_where = ''
|
||||
# _where = list()
|
||||
#
|
||||
# if len(sql_filter) > 0:
|
||||
# for k in sql_filter:
|
||||
# if k == 'log_user_name':
|
||||
# _where.append('l.user_name="{}"'.format(sql_filter[k]))
|
||||
# # elif k == 'search_record':
|
||||
# # _where.append('(h.name LIKE "%{}%" OR h.ip LIKE "%{}%" OR h.router_addr LIKE "%{}%" OR h.desc LIKE "%{}%" OR h.cid LIKE "%{}%")'.format(sql_filter[k], sql_filter[k], sql_filter[k], sql_filter[k], sql_filter[k]))
|
||||
#
|
||||
# if len(_where) > 0:
|
||||
# str_where = '( {} )'.format(' AND '.join(_where))
|
||||
#
|
||||
# s.where(str_where)
|
||||
#
|
||||
# if sql_order is not None:
|
||||
# _sort = False if not sql_order['asc'] else True
|
||||
# if 'log_time' == sql_order['name']:
|
||||
# s.order_by('l.log_time', _sort)
|
||||
# # elif 'name' == sql_order['name']:
|
||||
# # s.order_by('h.name', _sort)
|
||||
# # elif 'os_type' == sql_order['name']:
|
||||
# # s.order_by('h.os_type', _sort)
|
||||
# # elif 'cid' == sql_order['name']:
|
||||
# # s.order_by('h.cid', _sort)
|
||||
# # elif 'state' == sql_order['name']:
|
||||
# # s.order_by('h.state', _sort)
|
||||
# else:
|
||||
# log.e('unknown order field: {}\n'.format(sql_order['name']))
|
||||
# return TPE_PARAM, s.total_count, s.recorder
|
||||
#
|
||||
# if len(sql_limit) > 0:
|
||||
# s.limit(sql_limit['page_index'], sql_limit['per_page'])
|
||||
#
|
||||
# err = s.query()
|
||||
# return err, s.total_count, s.recorder
|
||||
|
|
@ -8,6 +8,7 @@ from app.base.logger import log
|
|||
from app.base.utils import tp_timestamp_utc_now, tp_generate_random
|
||||
from app.const import *
|
||||
from app.model import syslog
|
||||
from app.base.stats import tp_stats
|
||||
from app.logic.auth.password import tp_password_verify
|
||||
from app.logic.auth.oath import tp_oath_verify_code
|
||||
|
||||
|
|
@ -204,6 +205,7 @@ def create_users(handler, user_list, success, failed):
|
|||
|
||||
if len(name_list) > 0:
|
||||
syslog.sys_log(operator, handler.request.remote_ip, TPE_OK, "创建用户：{}".format('，'.join(name_list)))
|
||||
tp_stats().user_counter_change(len(name_list))
|
||||
|
||||
|
||||
def create_user(handler, args):
|
||||
|
|
@ -240,6 +242,7 @@ def create_user(handler, args):
|
|||
_id = db.last_insert_id()
|
||||
|
||||
syslog.sys_log(operator, handler.request.remote_ip, TPE_OK, "创建用户：{}".format(args['username']))
|
||||
tp_stats().user_counter_change(1)
|
||||
|
||||
return TPE_OK, _id
|
||||
|
||||
|
|
@ -512,6 +515,8 @@ def remove_users(handler, users):
|
|||
|
||||
syslog.sys_log(handler.get_current_user(), handler.request.remote_ip, TPE_OK, "删除用户：{}".format('，'.join(name_list)))
|
||||
|
||||
tp_stats().user_counter_change(0 - len(name_list))
|
||||
|
||||
return TPE_OK
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue