feat: add syslog real-time forwarding

link-mink · Jan 20, 2022 · 0b905db · 0b905db
1 parent 53ef816
commit 0b905db
Show file tree

Hide file tree

Showing 15 changed files with 514 additions and 46 deletions.
diff --git a/Makefile.am b/Makefile.am
@@ -373,4 +373,8 @@ endif
 if ENABLE_CLIPS
 include src/services/sysagent/plugins/clips/Makefile.am
 endif
+# syslog plugin
+if ENABLE_SYSLOG
+include src/services/sysagent/plugins/syslog/Makefile.am
+endif
 endif
diff --git a/configure.ac b/configure.ac
@@ -228,6 +228,15 @@ if test "x$enable_openwrt" != "xno"; then
 
 fi
 
+# syslog
+AC_ARG_ENABLE(syslog,
+              [AS_HELP_STRING([--enable-syslog], [Enable syslog plugin [default=no]])],,
+              [enable_syslog=no])
+AM_CONDITIONAL(ENABLE_SYSLOG, test "x$enable_syslog" = "xyes")
+if test "x$enable_syslog" != "xno"; then
+    AC_DEFINE([ENABLE_SYSLOG], [1], [Enable syslog plugin])
+fi 
+
 # GDT packet chunk size
 AC_ARG_WITH(gdt-csize,
             [AS_HELP_STRING([--with-gdt-csize], [GDT packet buffer size [default = 1024]])],

diff --git a/src/gdt/gdt.asn b/src/gdt/gdt.asn
@@ -458,6 +458,10 @@ ParameterType ::= INTEGER {
     pt-mink-checksum                                        (6014), -- checksum 
     pt-mink-timeout                                         (6015), -- timeout
     pt-mink-error                                           (6016), -- error code
+    pt-mink-error-msg                                       (6017), -- error message
+    pt-mink-status                                          (6018), -- status code
+    pt-mink-status-msg                                      (6019), -- status message
+    pt-mink-persistent-correlation                          (6020), -- persistent GUID
 
     -- MINK routing                                         (6100 - 6200)
     -- ------------------------------------------------------------------

diff --git a/src/include/gdt_def.h b/src/include/gdt_def.h
@@ -549,6 +549,10 @@ namespace asn1 {
         static const int _pt_mink_checksum = 6014;
         static const int _pt_mink_timeout = 6015;
         static const int _pt_mink_error = 6016;
+        static const int _pt_mink_error_msg = 6017;
+        static const int _pt_mink_status = 6018;
+        static const int _pt_mink_status_msg = 6019;
+        static const int _pt_mink_persistent_correlation = 6020;
         static const int _pt_mink_routing_destination = 6100;
         static const int _pt_mink_routing_source = 6101;
         static const int _pt_mink_routing_gateway = 6102;

diff --git a/src/include/mink_err_codes.h b/src/include/mink_err_codes.h
@@ -14,6 +14,7 @@
 namespace mink {
     namespace error {
         enum ErrorCode {
+            EC_OK                   =  0,
             EC_JSON_MALFORMED       = -1,
             EC_REQ_TIMEOUT          = -2,
             EC_GDT_PUSH_FAILED      = -3,

diff --git a/src/proto/gdt.pb.cc b/src/proto/gdt.pb.cc
diff --git a/src/proto/gdt.pb.enums_only.h b/src/proto/gdt.pb.enums_only.h
@@ -17,6 +17,10 @@ namespace gdt_grpc {
       UNKNWON_PARAMETER = 0,
       PT_MINK_DTYPE = 6000,
       PT_MINK_DID = 6001,
+      PT_MINK_ERROR = 6016,
+      PT_MINK_ERROR_MSG = 6017,
+      PT_MINK_STATUS = 6018,
+      PT_MINK_STATUS_MSG = 6019,
       PT_CPU_USER_PERCENT = 9000,
       PT_CPU_NICE_PERCENT = 9001,
       PT_CPU_SYSTEM_PERCENT = 9002,
@@ -59,11 +63,17 @@ namespace gdt_grpc {
       PT_SP_PAYLOAD = 9039,
       PT_FU_DATA = 9040,
       PT_FU_FSIZE = 9041,
+      PT_SL_LOGLINE = 9042,
+      PT_SL_PORT = 9043,
     };
     static const std::map<int, std::string> SysagentParamMap = {
       {UNKNWON_PARAMETER, "UNKNWON_PARAMETER"}, 
       {PT_MINK_DTYPE, "PT_MINK_DTYPE"}, 
       {PT_MINK_DID, "PT_MINK_DID"}, 
+      {PT_MINK_ERROR, "PT_MINK_ERROR"}, 
+      {PT_MINK_ERROR_MSG, "PT_MINK_ERROR_MSG"}, 
+      {PT_MINK_STATUS, "PT_MINK_STATUS"}, 
+      {PT_MINK_STATUS_MSG, "PT_MINK_STATUS_MSG"}, 
       {PT_CPU_USER_PERCENT, "PT_CPU_USER_PERCENT"}, 
       {PT_CPU_NICE_PERCENT, "PT_CPU_NICE_PERCENT"}, 
       {PT_CPU_SYSTEM_PERCENT, "PT_CPU_SYSTEM_PERCENT"}, 
@@ -106,6 +116,8 @@ namespace gdt_grpc {
       {PT_SP_PAYLOAD, "PT_SP_PAYLOAD"}, 
       {PT_FU_DATA, "PT_FU_DATA"}, 
       {PT_FU_FSIZE, "PT_FU_FSIZE"}, 
+      {PT_SL_LOGLINE, "PT_SL_LOGLINE"}, 
+      {PT_SL_PORT, "PT_SL_PORT"}, 
     };
     enum SysagentCommand : int {
       UNKNWON_COMMAND = 0,
@@ -123,6 +135,8 @@ namespace gdt_grpc {
       CMD_AUTH = 12,
       CMD_SOCKET_PROXY = 13,
       CMD_FIRMWARE_UPDATE = 14,
+      CMD_SYSLOG_START = 15,
+      CMD_SYSLOG_STOP = 16,
     };
     static const std::map<int, std::string> SysagentCommandMap = {
       {UNKNWON_COMMAND, "UNKNWON_COMMAND"}, 
@@ -140,6 +154,8 @@ namespace gdt_grpc {
       {CMD_AUTH, "CMD_AUTH"}, 
       {CMD_SOCKET_PROXY, "CMD_SOCKET_PROXY"}, 
       {CMD_FIRMWARE_UPDATE, "CMD_FIRMWARE_UPDATE"}, 
+      {CMD_SYSLOG_START, "CMD_SYSLOG_START"}, 
+      {CMD_SYSLOG_STOP, "CMD_SYSLOG_STOP"}, 
     };
 }
 

diff --git a/src/proto/gdt.pb.h b/src/proto/gdt.pb.h
diff --git a/src/proto/gdt.proto b/src/proto/gdt.proto
@@ -47,6 +47,8 @@ enum SysagentCommand {
     CMD_AUTH = 12;
     CMD_SOCKET_PROXY = 13;
     CMD_FIRMWARE_UPDATE = 14;
+    CMD_SYSLOG_START = 15;
+    CMD_SYSLOG_STOP = 16;
 }
 
 /******************/
@@ -57,6 +59,10 @@ enum ParameterType {
     // mink core
     PT_MINK_DTYPE = 6000;
     PT_MINK_DID = 6001;
+    PT_MINK_ERROR = 6016;
+    PT_MINK_ERROR_MSG = 6017;
+    PT_MINK_STATUS = 6018;
+    PT_MINK_STATUS_MSG = 6019;
     // cpu info
     PT_CPU_USER_PERCENT = 9000;
     PT_CPU_NICE_PERCENT = 9001;
@@ -108,6 +114,9 @@ enum ParameterType {
     // firmware update
     PT_FU_DATA = 9040;
     PT_FU_FSIZE = 9041;
+    // syslog
+    PT_SL_LOGLINE = 9042;
+    PT_SL_PORT = 9043;
 }
 
 /*****************/

diff --git a/src/services/json_rpc/events.cpp b/src/services/json_rpc/events.cpp
@@ -16,7 +16,6 @@
 #include <zlib.h>
 
 using data_vec_t = std::vector<uint8_t>;
-namespace beast = boost::beast;
 
 #ifdef ENABLE_CONFIGD
 EVHbeatMissed::EVHbeatMissed(mink::Atomic<uint8_t> *_activity_flag): activity_flag(_activity_flag) {}
@@ -69,14 +68,14 @@ static void cmap_process_timeout(mink_utils::CorrelationMap<JrpcPayload> &cmap){
     for (auto it = cmap.begin(), it_next = it; it != cmap.end(); it = it_next) {
         // next
         ++it_next;
-
         // calculate timeout
         if(now - it->second.ts <= it->second.data_timeout) continue;
+        // skip persistent
+        if(it->second.data.persistent) continue;
         // payload
         JrpcPayload &pld = it->second.data;
         // session pointer
         std::shared_ptr<WebSocketBase> ws = pld.cdata;
-        int id = pld.id; 
         // remove from list
         cmap.remove(it);
     }
@@ -195,6 +194,9 @@ void EVSrvcMsgRecv::run(gdt::GDTCallbackArgs *args){
     const mink_utils::VariantParam *vp_guid = smsg->vpget(asn1::ParameterType::_pt_mink_guid);
     if(!vp_guid) return;
 
+    // persistent guid
+    const mink_utils::VariantParam *vp_p_guid = smsg->vpget(asn1::ParameterType::_pt_mink_persistent_correlation);
+
     // error check
     const mink_utils::VariantParam *vp_err = smsg->vpget(asn1::ParameterType::_pt_mink_error);
 
@@ -209,6 +211,12 @@ void EVSrvcMsgRecv::run(gdt::GDTCallbackArgs *args){
         dd->cmap.unlock();
         return;
     }
+    // set as persistent (if requested)
+    if (vp_p_guid)
+        pld->persistent = true;
+    else
+        pld->persistent = false;
+
     // id
     int id = pld->id;
     auto ts_req = pld->ts;
@@ -219,7 +227,7 @@ void EVSrvcMsgRecv::run(gdt::GDTCallbackArgs *args){
     auto j = json_rpc::JsonRpc::gen_response(id);
     // update ts
     dd->cmap.update_ts(guid);
-    dd->cmap.remove(guid);
+    if(!pld->persistent) dd->cmap.remove(guid);
     // unlock
     dd->cmap.unlock();
 
@@ -300,7 +308,7 @@ void EVSrvcMsgRecv::run(gdt::GDTCallbackArgs *args){
                                         it->second.get_size(),
                                         z_out_buff,
                                         sizeof(z_out_buff));
-
+            
             // compressed/uncompressed
             if (val.empty()) {
                 val.assign(reinterpret_cast<char *>(od), it->second.get_size());
@@ -338,6 +346,10 @@ void EVSrvcMsgRecv::run(gdt::GDTCallbackArgs *args){
         //                             beast::bind_front_handler(&WebSocketBase::on_write,
         //                                                       ws));
     } catch (std::exception &e) {
+        mink::CURRENT_DAEMON->log(mink::LLT_ERROR, 
+                                  "JSON RPC error: [%s]",
+                                  e.what());
+
         // send error
         handle_error(nullptr, id, ws);
     }

diff --git a/src/services/json_rpc/jrpc.h b/src/services/json_rpc/jrpc.h
@@ -48,6 +48,7 @@ struct JrpcPayload {
     mink_utils::Guid guid;
     std::shared_ptr<WebSocketBase> cdata;
     int id;
+    bool persistent = false;
     std::chrono::time_point<std::chrono::system_clock> ts;
 };
 

diff --git a/src/services/json_rpc/ws_server.h b/src/services/json_rpc/ws_server.h
@@ -441,7 +441,7 @@ class WebSocketSession : public WebSocketBase {
                             // correlate guid
                             dd->cmap.lock();
                             JrpcPayload *pld = dd->cmap.get(g);
-                            if (pld) {
+                            if (pld && !pld->persistent) {
                                 dd->cmap.remove(g);
                                 dd->cmap.unlock();
                                 std::string th_rpl = Jrpc::gen_err(mink::error::EC_REQ_TIMEOUT, id).dump();