packages/sysSentry
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
sysSentry/add-pyxalarm-and-pySentryNotify-add-multi-users-supp.patch

737 lines
25 KiB
Diff
Raw Normal View History

add multi user support for xalarm, pyxalarm and pySentryNotify
2025-01-17 16:17:49 +08:00
From 600e20fb50f15b23252b556ca1fca806339cadfe Mon Sep 17 00:00:00 2001
From: PshySimon <caixiaomeng2@huawei.com>
Date: Thu, 26 Sep 2024 16:12:25 +0800
Subject: [PATCH] add pyxalarm and pySentryNotify, add multi users support for
xalarmd and adapt libxalarm
fix xalarm_getdesc function return not a valid str problem
xalarm add alarm msg length to 8192
fix bugs of xalarmd
fix xalarm non-uniform log formatting
---
src/libso/xalarm/register_xalarm.c | 54 +++-----
src/libso/xalarm/register_xalarm.h | 10 +-
src/python/xalarm/register_xalarm.py | 196 +++++++++++++++++++++++++++
src/python/xalarm/sentry_notify.py | 72 ++++++++++
src/python/xalarm/xalarm_api.py | 18 ++-
src/python/xalarm/xalarm_server.py | 50 ++++++-
src/python/xalarm/xalarm_transfer.py | 104 ++++++++++++--
7 files changed, 446 insertions(+), 58 deletions(-)
create mode 100644 src/python/xalarm/register_xalarm.py
create mode 100644 src/python/xalarm/sentry_notify.py
diff --git a/src/libso/xalarm/register_xalarm.c b/src/libso/xalarm/register_xalarm.c
index 9eeed74..aa0645d 100644
--- a/src/libso/xalarm/register_xalarm.c
+++ b/src/libso/xalarm/register_xalarm.c
@@ -35,7 +35,7 @@
#define ALARM_SOCKET_PERMISSION 0700
#define TIME_UNIT_MILLISECONDS 1000
-#define MAX_PARAS_LEN 511
+#define MAX_PARAS_LEN 8191
#define MIN_ALARM_ID 1001
#define MAX_ALARM_ID (MIN_ALARM_ID + MAX_NUM_OF_ALARM_ID - 1)
@@ -91,7 +91,7 @@ static int create_unix_socket(const char *path)
return -1;
}
- fd = socket(AF_UNIX, SOCK_DGRAM, 0);
+ fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (fd < 0) {
printf("socket failed:%s\n", strerror(errno));
return -1;
@@ -103,14 +103,6 @@ static int create_unix_socket(const char *path)
goto release_socket;
}
- if (access(PATH_REG_ALARM, F_OK) == 0) {
- ret = unlink(PATH_REG_ALARM);
- if (ret != 0) {
- printf("unlink register socket file failed\n");
- goto release_socket;
- }
- }
-
if (access(DIR_XALARM, F_OK) == -1) {
if (mkdir(DIR_XALARM, ALARM_DIR_PERMISSION) == -1) {
printf("mkdir %s failed\n", DIR_XALARM);
@@ -120,32 +112,22 @@ static int create_unix_socket(const char *path)
if (memset(&alarm_addr, 0, sizeof(alarm_addr)) == NULL) {
printf("create_unix_socket: memset alarm_addr failed, ret: %d\n", ret);
- goto remove_dir;
+ goto release_socket;
}
alarm_addr.sun_family = AF_UNIX;
strncpy(alarm_addr.sun_path, path, sizeof(alarm_addr.sun_path) - 1);
- if (bind(fd, (struct sockaddr *)&alarm_addr, sizeof(alarm_addr.sun_family) + strlen(alarm_addr.sun_path)) < 0) {
- printf("bind socket failed:%s\n", strerror(errno));
- goto remove_dir;
+ if (connect(fd, (struct sockaddr*)&alarm_addr, sizeof(alarm_addr)) == -1) {
+ printf("create_unix_socket: connect alarm_addr failed, ret: %d\n", ret);
+ goto release_socket;
}
if (chmod(path, ALARM_SOCKET_PERMISSION) < 0) {
printf("chmod %s failed: %s\n", path, strerror(errno));
- goto unlink_sockfile;
+ goto release_socket;
}
return fd;
-unlink_sockfile:
- ret = unlink(PATH_REG_ALARM);
- if (ret != 0) {
- printf("unlink register socket file failed\n");
- }
-remove_dir:
- ret = rmdir(DIR_XALARM);
- if (ret != 0) {
- printf("rmdir %s failed: %s\n", path, strerror(errno));
- }
release_socket:
(void)close(fd);
@@ -174,6 +156,10 @@ static void *alarm_recv(void *arg)
continue;
}
printf("recv error len:%d errno:%d\n", recvlen, errno);
+ } else if (recvlen == 0) {
+ printf("connection closed by xalarmd, maybe connections reach max num or service stopped.\n");
+ g_register_info.thread_should_stop = 1;
+ break;
}
}
return NULL;
@@ -229,6 +215,10 @@ bool xalarm_Upgrade(struct alarm_subscription_info id_filter, int client_id)
printf("%s: invalid args\n", __func__);
return false;
}
+ if (g_register_info.thread_should_stop) {
+ printf("%s: upgrade failed, alarm thread has stopped\n", __func__);
+ return false;
+ }
set_alarm_id(id_filter);
return true;
@@ -271,8 +261,6 @@ int xalarm_Register(alarm_callback_func callback, struct alarm_subscription_info
void xalarm_UnRegister(int client_id)
{
- int ret;
-
if (!g_register_info.is_registered) {
printf("%s: alarm has not registered\n", __func__);
return;
@@ -292,10 +280,6 @@ void xalarm_UnRegister(int client_id)
if (g_register_info.register_fd != -1) {
(void)close(g_register_info.register_fd);
g_register_info.register_fd = -1;
- ret = unlink(PATH_REG_ALARM);
- if (ret != 0) {
- printf("%s: unlink register socket file failed\n", __func__);
- }
}
memset(g_register_info.alarm_enable_bitmap, 0, MAX_NUM_OF_ALARM_ID * sizeof(char));
@@ -357,7 +341,7 @@ int xalarm_Report(unsigned short usAlarmId, unsigned char ucAlarmLevel,
struct sockaddr_un alarm_addr;
if ((usAlarmId < MIN_ALARM_ID || usAlarmId > MAX_ALARM_ID) ||
- (ucAlarmLevel < ALARM_LEVEL_FATAL || ucAlarmLevel > ALARM_LEVEL_DEBUG) ||
+ (ucAlarmLevel < MINOR_ALM || ucAlarmLevel > CRITICAL_ALM) ||
(ucAlarmType < ALARM_TYPE_OCCUR || ucAlarmType > ALARM_TYPE_RECOVER)) {
fprintf(stderr, "%s: alarm info invalid\n", __func__);
return -1;
@@ -368,6 +352,11 @@ int xalarm_Report(unsigned short usAlarmId, unsigned char ucAlarmLevel,
return -1;
}
+ if (pucParas == NULL || (int)strlen(pucParas) > MAX_PARAS_LEN) {
+ fprintf(stderr, "%s: alarm info invalid\n", __func__);
+ return -1;
+ }
+
if (memset(&info, 0, sizeof(struct alarm_info)) == NULL) {
fprintf(stderr, "%s: memset info failed, ret: %d\n", __func__, ret);
return -1;
@@ -671,3 +660,4 @@ int report_result(const char *task_name, enum RESULT_LEVEL result_level, const c
return RETURE_CODE_SUCCESS;
}
+
diff --git a/src/libso/xalarm/register_xalarm.h b/src/libso/xalarm/register_xalarm.h
index 1f26c6a..dcf4f03 100644
--- a/src/libso/xalarm/register_xalarm.h
+++ b/src/libso/xalarm/register_xalarm.h
@@ -11,7 +11,7 @@
#include <sys/time.h>
#include <stdbool.h>
-#define ALARM_INFO_MAX_PARAS_LEN 512
+#define ALARM_INFO_MAX_PARAS_LEN 8192
#define MAX_STRERROR_SIZE 1024
#define MAX_ALARM_TYEPS 1024
#define MIN_ALARM_ID 1001
@@ -19,11 +19,9 @@
#define MEMORY_ALARM_ID 1001
-#define ALARM_LEVEL_FATAL 1
-#define ALARM_LEVEL_ERROR 2
-#define ALARM_LEVEL_WARNING 3
-#define ALARM_LEVEL_INFO 4
-#define ALARM_LEVEL_DEBUG 5
+#define MINOR_ALM 1
+#define MAJOR_ALM 2
+#define CRITICAL_ALM 3
#define ALARM_TYPE_OCCUR 1
#define ALARM_TYPE_RECOVER 2
diff --git a/src/python/xalarm/register_xalarm.py b/src/python/xalarm/register_xalarm.py
new file mode 100644
index 0000000..6ac1eb7
--- /dev/null
+++ b/src/python/xalarm/register_xalarm.py
@@ -0,0 +1,196 @@
+import os
+import sys
+import socket
+import logging
+import threading
+import time
+import fcntl
+import inspect
+from struct import error as StructParseError
+
+from .xalarm_api import Xalarm, alarm_bin2stu
+
+
+ALARM_REPORT_LEN = 8216
+MAX_NUM_OF_ALARM_ID=128
+MIN_ALARM_ID = 1001
+MAX_ALARM_ID = (MIN_ALARM_ID + MAX_NUM_OF_ALARM_ID - 1)
+DIR_XALARM = "/var/run/xalarm"
+PATH_REG_ALARM = "/var/run/xalarm/alarm"
+PATH_REPORT_ALARM = "/var/run/xalarm/report"
+ALARM_DIR_PERMISSION = 0o0750
+ALARM_REG_SOCK_PERMISSION = 0o0700
+ALARM_SOCKET_PERMISSION = 0o0700
+TIME_UNIT_MILLISECONDS = 1000
+ALARM_REGISTER_INFO = None
+
+
+class AlarmRegister:
+ def __init__(self, id_filter: list, callback: callable):
+ self.id_filter = id_filter
+ self.callback = callback
+ self.socket = self.create_unix_socket()
+ self.is_registered = True
+ self.thread = threading.Thread(target=self.alarm_recv)
+ self.thread_should_stop = False
+
+ def check_params(self) -> bool:
+ if (len(self.id_filter) != MAX_NUM_OF_ALARM_ID):
+ sys.stderr.write("check_params: invalid param id_filter\n")
+ return False
+
+ sig = inspect.signature(self.callback)
+ if len(sig.parameters) != 1:
+ sys.stderr.write("check_params: invalid param callback\n")
+ return False
+
+ if self.socket is None:
+ sys.stderr.write("check_params: socket create failed\n")
+ return False
+ return True
+
+ def set_id_filter(self, id_filter: list) -> bool:
+ if (len(id_filter) > MAX_NUM_OF_ALARM_ID):
+ sys.stderr.write("set_id_filter: invalid param id_filter\n")
+ return False
+ self.id_filter = id_filter
+
+ def id_is_registered(self, alarm_id) -> bool:
+ if alarm_id < MIN_ALARM_ID or alarm_id > MAX_ALARM_ID:
+ return False
+ return self.id_filter[alarm_id - MIN_ALARM_ID]
+
+ def put_alarm_info(self, alarm_info: Xalarm) -> None:
+ if not self.callback or not alarm_info:
+ return
+ if not self.id_is_registered(alarm_info.alarm_id):
+ return
+ self.callback(alarm_info)
+
+ def create_unix_socket(self) -> socket.socket:
+ try:
+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ sock.setblocking(False)
+
+ if not os.access(DIR_XALARM, os.F_OK):
+ os.makedirs(DIR_XALARM)
+ os.chmod(DIR_XALARM, ALARM_DIR_PERMISSION)
+
+ sock.connect(PATH_REG_ALARM)
+ return sock
+ except (IOError, OSError, FileNotFoundError) as e:
+ sock.close()
+ sys.stderr.write(f"create_unix_socket: create socket error:{e}\n")
+ return None
+
+ def alarm_recv(self):
+ while not self.thread_should_stop:
+ try:
+ data = self.socket.recv(ALARM_REPORT_LEN)
+ if not data:
+ sys.stderr.write("connection closed by xalarmd, maybe connections reach max num or service stopped.\n")
+ self.thread_should_stop = True
+ break
+ if len(data) != ALARM_REPORT_LEN:
+ sys.stderr.write(f"server receive report msg length wrong {len(data)}\n")
+ continue
+
+ alarm_info = alarm_bin2stu(data)
+ self.put_alarm_info(alarm_info)
+ except (BlockingIOError) as e:
+ time.sleep(0.1)
+ except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError):
+ sys.stderr.write("Connection closed by the server.\n")
+ self.thread_should_stop = True
+ except (ValueError, StructParseError, InterruptedError) as e:
+ sys.stderr.write(f"{e}\n")
+ except Exception as e:
+ sys.stderr.write(f"{e}\n")
+ self.thread_should_stop = True
+
+ def start_thread(self) -> None:
+ self.thread.daemon = True
+ self.thread.start()
+
+ def stop_thread(self) -> None:
+ self.thread_should_stop = True
+ self.thread.join()
+
+
+def xalarm_register(callback: callable, id_filter: list) -> int:
+ global ALARM_REGISTER_INFO
+
+ if ALARM_REGISTER_INFO is not None:
+ sys.stderr.write("xalarm_register: alarm has registered\n")
+ return -1
+
+ ALARM_REGISTER_INFO = AlarmRegister(id_filter, callback)
+ if not ALARM_REGISTER_INFO.check_params():
+ return -1
+
+ ALARM_REGISTER_INFO.start_thread()
+
+ return 0
+
+
+def xalarm_unregister(clientId: int) -> None:
+ global ALARM_REGISTER_INFO
+ if clientId < 0:
+ sys.stderr.write("xalarm_unregister: invalid client\n")
+ return
+
+ if ALARM_REGISTER_INFO is None:
+ sys.stderr.write("xalarm_unregister: alarm has not registered\n")
+ return
+
+ ALARM_REGISTER_INFO.stop_thread()
+ ALARM_REGISTER_INFO = None
+
+
+def xalarm_upgrade(clientId: int, id_filter: list) -> bool:
+ global ALARM_REGISTER_INFO
+ if clientId < 0:
+ sys.stderr.write("xalarm_upgrade: invalid client\n")
+ return False
+ if ALARM_REGISTER_INFO is None:
+ sys.stderr.write("xalarm_upgrade: alarm has not registered\n")
+ return False
+ if ALARM_REGISTER_INFO.thread_should_stop:
+ sys.stderr.write("xalarm_upgrade: upgrade failed, alarm thread has stopped\n")
+ return False
+ ALARM_REGISTER_INFO.id_filter = id_filter
+ return True
+
+
+def xalarm_getid(alarm_info: Xalarm) -> int:
+ if not alarm_info:
+ return 0
+ return alarm_info.alarm_id
+
+
+def xalarm_getlevel(alarm_info: Xalarm) -> int:
+ if not alarm_info:
+ return 0
+ return alarm_info.alarm_level
+
+
+def xalarm_gettype(alarm_info: Xalarm) -> int:
+ if not alarm_info:
+ return 0
+ return alarm_info.alarm_type
+
+
+def xalarm_gettime(alarm_info: Xalarm) -> int:
+ if not alarm_info:
+ return 0
+ return alarm_info.timetamp.tv_sec * TIME_UNIT_MILLISECONDS + alarm_info.timetamp.tv_usec / TIME_UNIT_MILLISECONDS
+
+def xalarm_getdesc(alarm_info: Xalarm) -> str:
+ if not alarm_info:
+ return None
+ try:
+ desc_str = alarm_info.msg1.rstrip(b'\x00').decode('utf-8')
+ except UnicodeError:
+ desc_str = None
+ return desc_str
+
diff --git a/src/python/xalarm/sentry_notify.py b/src/python/xalarm/sentry_notify.py
new file mode 100644
index 0000000..1e3fa76
--- /dev/null
+++ b/src/python/xalarm/sentry_notify.py
@@ -0,0 +1,72 @@
+import os
+import sys
+import time
+import socket
+import logging
+from struct import error as StructParseError
+
+from .xalarm_api import alarm_stu2bin, Xalarm
+
+MAX_NUM_OF_ALARM_ID = 128
+MIN_ALARM_ID = 1001
+MAX_ALARM_ID = (MIN_ALARM_ID + MAX_NUM_OF_ALARM_ID - 1)
+
+MINOR_ALM = 1
+MAJOR_ALM = 2
+CRITICAL_ALM = 3
+
+ALARM_TYPE_OCCUR = 1
+ALARM_TYPE_RECOVER = 2
+
+MAX_PUC_PARAS_LEN = 8192
+
+DIR_XALARM = "/var/run/xalarm"
+PATH_REPORT_ALARM = "/var/run/xalarm/report"
+ALARM_DIR_PERMISSION = 0o750
+ALARM_SOCKET_PERMISSION = 0o700
+
+
+def check_params(alarm_id, alarm_level, alarm_type, puc_paras) -> bool:
+ if not os.path.exists(DIR_XALARM):
+ logging.error(f"check_params: {DIR_XALARM} not exist, failed")
+ return False
+
+ if not os.path.exists(PATH_REPORT_ALARM):
+ logging.error(f"check_params: {PATH_REPORT_ALARM} not exist, failed")
+ return False
+
+ if (alarm_id < MIN_ALARM_ID or alarm_id > MAX_ALARM_ID or
+ alarm_level < MINOR_ALM or alarm_level > CRITICAL_ALM or
+ alarm_type < ALARM_TYPE_OCCUR or alarm_type > ALARM_TYPE_RECOVER):
+ logging.error("check_params: alarm info invalid")
+ return False
+
+ if len(puc_paras) >= MAX_PUC_PARAS_LEN:
+ logging.error(f"check_params: alarm msg should be less than {MAX_PUC_PARAS_LEN}")
+ return False
+
+ return True
+
+def xalarm_report(alarm_id, alarm_level, alarm_type, puc_paras) -> bool:
+ if not check_params(alarm_id, alarm_level, alarm_type, puc_paras):
+ return False
+
+ try:
+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+
+ current_time = time.time()
+ current_time_seconds = int(current_time)
+ current_microseconds = int((current_time - current_time_seconds) * 1_000_000)
+ alarm_info = Xalarm(alarm_id, alarm_type, alarm_level,
+ current_time_seconds, current_microseconds, puc_paras)
+
+ sock.sendto(alarm_stu2bin(alarm_info), PATH_REPORT_ALARM)
+ except (FileNotFoundError, StructParseError, socket.error, OSError, UnicodeError) as e:
+ logging.error(f"error occurs when sending msg.")
+ return False
+ finally:
+ sock.close()
+
+ return True
+
+
diff --git a/src/python/xalarm/xalarm_api.py b/src/python/xalarm/xalarm_api.py
index 94d7638..c365019 100644
--- a/src/python/xalarm/xalarm_api.py
+++ b/src/python/xalarm/xalarm_api.py
@@ -23,6 +23,7 @@ ALARM_LEVELS = (1, 2, 3, 4, 5)
ALARM_SOCK_PATH = "/var/run/xalarm/report"
MIN_ALARM_ID = 1001
MAX_ALARM_ID = 1128
+MAX_MSG_LEN = 8192
@dataclasses.dataclass
@@ -98,7 +99,7 @@ class Xalarm:
"""msg1 setter
"""
if len(msg) > 512:
- raise ValueError("msg1 length must below 255")
+ raise ValueError("msg1 length must below 512")
self._msg1 = msg
@@ -116,3 +117,18 @@ def alarm_bin2stu(bin_data):
alarm_info.msg1 = struct_data[5]
return alarm_info
+
+
+def alarm_stu2bin(alarm_info: Xalarm):
+ alarm_msg = alarm_info.msg1
+ padding_length = MAX_MSG_LEN - len(alarm_msg)
+ if padding_length > 0:
+ alarm_msg = alarm_msg + ('\x00' * padding_length)
+ return struct.pack(
+ f'@HBBll{MAX_MSG_LEN}s',
+ alarm_info.alarm_id,
+ alarm_info.alarm_level,
+ alarm_info.alarm_type,
+ alarm_info.timetamp.tv_sec,
+ alarm_info.timetamp.tv_usec,
+ alarm_msg.encode('utf-8'))
diff --git a/src/python/xalarm/xalarm_server.py b/src/python/xalarm/xalarm_server.py
index 84db273..f90a0e2 100644
--- a/src/python/xalarm/xalarm_server.py
+++ b/src/python/xalarm/xalarm_server.py
@@ -17,16 +17,25 @@ Create: 2023-11-02
import socket
import os
import logging
+import select
+import threading
from struct import error as StructParseError
from .xalarm_api import alarm_bin2stu
-from .xalarm_transfer import check_filter, transmit_alarm
+from .xalarm_transfer import (
+ check_filter,
+ transmit_alarm,
+ wait_for_connection,
+ peroid_task_to_cleanup_connections
+)
ALARM_DIR = "/var/run/xalarm"
+USER_RECV_SOCK = "/var/run/xalarm/alarm"
SOCK_FILE = "/var/run/xalarm/report"
-ALARM_REPORT_LEN = 536
+ALARM_REPORT_LEN = 8216
ALARM_DIR_PERMISSION = 0o750
+ALARM_LISTEN_QUEUE_LEN = 5
def clear_sock_path():
@@ -37,6 +46,8 @@ def clear_sock_path():
os.chmod(ALARM_DIR, ALARM_DIR_PERMISSION)
if os.path.exists(SOCK_FILE):
os.unlink(SOCK_FILE)
+ if os.path.exists(USER_RECV_SOCK):
+ os.unlink(USER_RECV_SOCK)
def server_loop(alarm_config):
@@ -49,6 +60,25 @@ def server_loop(alarm_config):
sock.bind(SOCK_FILE)
os.chmod(SOCK_FILE, 0o600)
+ alarm_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ alarm_sock.bind(USER_RECV_SOCK)
+ os.chmod(USER_RECV_SOCK, 0o600)
+ alarm_sock.listen(ALARM_LISTEN_QUEUE_LEN)
+ alarm_sock.setblocking(False)
+
+ epoll = select.epoll()
+ epoll.register(alarm_sock.fileno(), select.EPOLLIN)
+ fd_to_socket = {alarm_sock.fileno(): alarm_sock,}
+ thread_should_stop = False
+
+ conn_thread = threading.Thread(target=wait_for_connection, args=(alarm_sock, epoll, fd_to_socket, thread_should_stop))
+ conn_thread.daemon = True
+ conn_thread.start()
+
+ cleanup_thread = threading.Thread(target=peroid_task_to_cleanup_connections, args=(alarm_sock, epoll, fd_to_socket, thread_should_stop))
+ cleanup_thread.daemon = True
+ cleanup_thread.start()
+
while True:
try:
data, _ = sock.recvfrom(ALARM_REPORT_LEN)
@@ -58,14 +88,22 @@ def server_loop(alarm_config):
logging.debug("server receive report msg length wrong %d",
len(data))
continue
-
alarm_info = alarm_bin2stu(data)
logging.debug("server bin2stu msg")
if not check_filter(alarm_info, alarm_config):
continue
+ transmit_alarm(alarm_sock, epoll, fd_to_socket, data)
+ except Exception as e:
+ logging.error(f"Error server:{e}")
+
+ thread_should_stop = True
+ conn_thread.join()
+ cleanup_thread.join()
- transmit_alarm(data)
- except (ValueError, StructParseError):
- pass
+ epoll.unregister(alarm_sock.fileno())
+ epoll.close()
+ alarm_sock.close()
+ os.unlink(USER_RECV_SOCK)
sock.close()
+
diff --git a/src/python/xalarm/xalarm_transfer.py b/src/python/xalarm/xalarm_transfer.py
index b590b43..010d2ee 100644
--- a/src/python/xalarm/xalarm_transfer.py
+++ b/src/python/xalarm/xalarm_transfer.py
@@ -16,10 +16,14 @@ Create: 2023-11-02
import socket
import logging
+import select
+from time import sleep
-USER_RECV_SOCK = "/var/run/xalarm/alarm"
MIN_ID_NUMBER = 1001
MAX_ID_NUMBER = 1128
+MAX_CONNECTION_NUM = 100
+TEST_CONNECT_BUFFER_SIZE = 32
+PEROID_SCANN_TIME = 60
def check_filter(alarm_info, alarm_filter):
@@ -35,16 +39,90 @@ def check_filter(alarm_info, alarm_filter):
return True
-def transmit_alarm(bin_data):
- """forward alarm message
+def cleanup_closed_connections(server_sock, epoll, fd_to_socket):
"""
- sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
- try:
- sock.sendto(bin_data, USER_RECV_SOCK)
- logging.debug("transfer alarm success")
- except ConnectionRefusedError:
- logging.debug("transfer sendto failed")
- except FileNotFoundError:
- logging.debug("transfer sendto failed")
- finally:
- sock.close()
+ clean invalid client socket connections saved in 'fd_to_socket'
+ :param server_sock: server socket instance of alarm
+ :param epoll: epoll instance, used to unregister invalid client connections
+ :param fd_to_socket: dict instance, used to hold client connections and server connections
+ """
+ to_remove = []
+ for fileno, connection in fd_to_socket.items():
+ if connection is server_sock:
+ continue
+ try:
+ # test whether connection still alive, use MSG_DONTWAIT to avoid blocking thread
+ # use MSG_PEEK to avoid consuming buffer data
+ data = connection.recv(TEST_CONNECT_BUFFER_SIZE, socket.MSG_DONTWAIT | socket.MSG_PEEK)
+ if not data:
+ to_remove.append(fileno)
+ except BlockingIOError:
+ pass
+ except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError):
+ to_remove.append(fileno)
+
+ for fileno in to_remove:
+ epoll.unregister(fileno)
+ fd_to_socket[fileno].close()
+ del fd_to_socket[fileno]
+ logging.info(f"cleaned up connection {fileno} for client lost connection.")
+
+
+def peroid_task_to_cleanup_connections(server_sock, epoll, fd_to_socket, thread_should_stop):
+ while not thread_should_stop:
+ sleep(PEROID_SCANN_TIME)
+ cleanup_closed_connections(server_sock, epoll, fd_to_socket)
+
+
+def wait_for_connection(server_sock, epoll, fd_to_socket, thread_should_stop):
+ """
+ thread function for catch and save client connection
+ :param server_sock: server socket instance of alarm
+ :param epoll: epoll instance, used to unregister invalid client connections
+ :param fd_to_socket: dict instance, used to hold client connections and server connections
+ :param thread_should_stop: bool instance
+ """
+ while not thread_should_stop:
+ try:
+ events = epoll.poll(1)
+
+ for fileno, event in events:
+ if fileno == server_sock.fileno():
+ connection, client_address = server_sock.accept()
+ # if reach max connection, cleanup closed connections
+ if len(fd_to_socket) - 1 >= MAX_CONNECTION_NUM:
+ cleanup_closed_connections(server_sock, epoll, fd_to_socket)
+ # if connections still reach max num, close this connection automatically
+ if len(fd_to_socket) - 1 >= MAX_CONNECTION_NUM:
+ logging.info(f"connection reach max num of {MAX_CONNECTION_NUM}, closed current connection!")
+ connection.close()
+ continue
+ epoll.register(connection.fileno(), select.EPOLLOUT)
+ fd_to_socket[connection.fileno()] = connection
+ except socket.error as e:
+ logging.debug("socket error, reason is %s", e)
+ break
+ except (KeyError, OSError, ValueError) as e:
+ logging.debug("wait for connection failed %s", e)
+
+
+def transmit_alarm(server_sock, epoll, fd_to_socket, bin_data):
+ """
+ this function is to broadcast alarm data to client, if fail to send data, remove connections held by fd_to_socket
+ :param server_sock: server socket instance of alarm
+ :param epoll: epoll instance, used to unregister invalid client connections
+ :param fd_to_socket: dict instance, used to hold client connections and server connections
+ :param bin_data: binary instance, alarm info data in C-style struct format defined in xalarm_api.py
+ """
+ to_remove = []
+ for fileno, connection in fd_to_socket.items():
+ if connection is not server_sock:
+ try:
+ connection.sendall(bin_data)
+ except (BrokenPipeError, ConnectionResetError):
+ to_remove.append(fileno)
+ for fileno in to_remove:
+ epoll.unregister(fileno)
+ fd_to_socket[fileno].close()
+ del fd_to_socket[fileno]
+ logging.info(f"cleaned up connection {fileno} for client lost connection.")
--
2.27.0
Reference in New Issue Copy Permalink