Merge branch 'syslog_telemetry_shared' of https://github.com/renukamanavalan/sonic-swss-common into syslog_telemetry_shared

Author: renukamanavalan

common/events.cpp

@@ -1,4 +1,5 @@
 #include "events_pi.h"
+#include "events_wrap.h"
 
 /*
  *  Track created publishers to avoid duplicates
@@ -73,6 +74,7 @@ EventPublisher::publish(const string tag, const event_params_t *params)
 {
     int rc;
     internal_event_t event_data;
+    string key(m_event_source + ":" + tag);
 
     if (m_event_service.is_active()) {
         string s;
@@ -107,7 +109,7 @@ EventPublisher::publish(const string tag, const event_params_t *params)
             map_to_str(*params).c_str());
 
     {
-    map_str_str_t event_str_map = { { m_event_source + ":" + tag, param_str}};
+    map_str_str_t event_str_map = { { key, param_str}};
 
     rc = serialize(event_str_map, event_data[EVENT_STR_DATA]);
     RET_ON_ERR(rc == 0, "failed to serialize event str %s", 
@@ -119,6 +121,22 @@ EventPublisher::publish(const string tag, const event_params_t *params)
 
     rc = zmq_message_send(m_socket, m_event_source, event_data);
     RET_ON_ERR(rc == 0, "failed to send for tag %s", tag.c_str());
+
+    {
+        nlohmann::json msg = nlohmann::json::object();
+        {
+            nlohmann::json params_data = nlohmann::json::object();
+
+            for (event_params_t::const_iterator itc = params->begin();
+                    itc != params->end(); ++itc) {
+                params_data[itc->first] = itc->second;
+            }
+            msg[key] = params_data;
+        }
+        string json_str(msg.dump());
+        SWSS_LOG_INFO("EVENT_PUBLISHED: %s", json_str.c_str());
+    }
+
 out:
     return rc;
 }
@@ -127,10 +145,10 @@ event_handle_t
 events_init_publisher(const string event_source)
 {
     event_handle_t ret = NULL;
-    lst_publishers_t::iterator it = s_publishers.find(event_source);
-    if (it != s_publishers.end()) {
+    lst_publishers_t::const_iterator itc = s_publishers.find(event_source);
+    if (itc != s_publishers.end()) {
         // Pre-exists
-        ret = it->second;
+        ret = itc->second;
     }
     else {
         EventPublisher *p =  new EventPublisher();
@@ -348,17 +366,15 @@ EventSubscriber::event_receive(string &key, event_params_t &params, int &missed_
         /* Find any missed events for this runtime ID */
         missed_cnt = 0;
         sequence_t seq = str_to_seq(event_data[EVENT_SEQUENCE]);
-        track_info_t::iterator it = m_track.find(event_data[EVENT_RUNTIME_ID]);
-        if (it != m_track.end()) {
+        track_info_t::const_iterator itc = m_track.find(event_data[EVENT_RUNTIME_ID]);
+        if (itc != m_track.end()) {
             /* current seq - last read - 1 == 0 if none missed */
-            missed_cnt = seq - it->second.seq - 1;
-            it->second = evt_info_t(seq);
+            missed_cnt = seq - itc->second.seq - 1;
         }
         else {
             if (m_track.size() > (MAX_PUBLISHERS_COUNT + 10)) {
                 prune_track();
             }
-            m_track[event_data[EVENT_RUNTIME_ID]] = evt_info_t(seq);
         }
         if (missed_cnt >= 0) {
             map_str_str_t ev;
@@ -376,6 +392,8 @@ EventSubscriber::event_receive(string &key, event_params_t &params, int &missed_
             rc = deserialize(ev.begin()->second, params);
             RET_ON_ERR(rc == 0, "failed to deserialize params %s",
                     ev.begin()->second.substr(0, 32).c_str());
+            
+            m_track[event_data[EVENT_RUNTIME_ID]] = evt_info_t(seq);
 
         }
         else {
@@ -395,15 +413,21 @@ event_handle_t
 events_init_subscriber(bool use_cache, int recv_timeout,
         const event_subscribe_sources_t *sources)
 {
+    EventSubscriber *sub = NULL;
+
     if (s_subscriber == NULL) {
-        EventSubscriber *sub = new EventSubscriber();
+        sub = new EventSubscriber();
 
         RET_ON_ERR(sub->init(use_cache, recv_timeout, sources) == 0,
                 "Failed to init subscriber");
 
         s_subscriber = sub;
+        sub = NULL;
     }
 out:
+    if (sub != NULL) {
+        delete sub;
+    }
     return s_subscriber;
 }
 
@@ -418,24 +442,170 @@ events_deinit_subscriber(event_handle_t handle)
 }
 
 
+event_receive_op_t
+event_receive(event_handle_t handle)
+{
+    event_receive_op_t op;
+
+    if ((handle == s_subscriber) && (s_subscriber != NULL)) {
+        op.rc = s_subscriber->event_receive(op.key, op.params, op.missed_cnt);
+    }
+    else {
+        op.rc = -1;
+    }
+    return op;
+}
+
+void *
+events_init_publisher_wrap(const char *args)
+{
+    SWSS_LOG_DEBUG("events_init_publisher_wrap: args=%s",
+            (args != NULL ? args : "<null pointer>"));
+
+    if (args == NULL) {
+        return NULL;
+    }
+    const auto &data = nlohmann::json::parse(args);
+
+    string source;
+    for (auto it = data.cbegin(); it != data.cend(); ++it) {
+        if ((it.key() == ARGS_SOURCE) && (*it).is_string()) {
+            source = it.value();
+        }
+    }
+    return events_init_publisher(source);
+}
+
+
+void
+events_deinit_publisher_wrap(void *handle) 
+{
+    events_deinit_publisher(handle);
+}
+
 
 int
-event_receive(event_handle_t handle, string &key,
-        event_params_t &params, int &missed_cnt)
+event_publish_wrap(void *handle, const char *args)
 {
-    if ((handle == s_subscriber) && (s_subscriber != NULL)) {
-        return s_subscriber->event_receive(key, params, missed_cnt);
+    string tag;
+    event_params_t params;
+
+    SWSS_LOG_DEBUG("events_init_publisher_wrap: handle=%p args=%s",
+            handle, (args != NULL ? args : "<null pointer>"));
+
+    if (args == NULL) {
+        return -1;
     }
-    return -1;
+    const auto &data = nlohmann::json::parse(args);
+
+    for (auto it = data.cbegin(); it != data.cend(); ++it) {
+        if ((it.key() == ARGS_TAG) && (*it).is_string()) {
+            tag = it.value();
+        }
+        else if ((it.key() == ARGS_PARAMS) && (*it).is_object()) {
+            const auto &params_data = *it;
+            for (auto itp = params_data.cbegin(); itp != params_data.cend(); ++itp) {
+                if ((*itp).is_string()) {
+                    params[itp.key()] = itp.value();
+                }
+            }
+        }
+    }
+    return event_publish(handle, tag, &params);
 }
 
+void *
+events_init_subscriber_wrap(const char *args)
+{
+    bool use_cache = true;
+    int recv_timeout = -1;
+    event_subscribe_sources_t sources;
 
-event_receive_op_t
-event_receive_wrap(event_handle_t handle)
+    SWSS_LOG_DEBUG("events_init_subsriber_wrap: args:%s", args);
+
+    if (args != NULL) {
+        const auto &data = nlohmann::json::parse(args);
+
+        for (auto it = data.cbegin(); it != data.cend(); ++it) {
+            if ((it.key() == ARGS_USE_CACHE) && (*it).is_boolean()) {
+                use_cache = it.value();
+            }
+            else if ((it.key() == ARGS_RECV_TIMEOUT) && (*it).is_number_integer()) {
+                recv_timeout = it.value();
+            }
+        }
+    }
+    void *handle = events_init_subscriber(use_cache, recv_timeout);
+    SWSS_LOG_DEBUG("events_init_subscriber_wrap: handle=%p", handle);
+    return handle;
+}
+
+
+void
+events_deinit_subscriber_wrap(void *handle) 
 {
-    event_receive_op_t op;
+    SWSS_LOG_DEBUG("events_deinit_subsriber_wrap: args=%p", handle);
 
-    op.rc = event_receive(handle, op.key, op.params, op.missed_cnt);
-    return op;
+    events_deinit_subscriber(handle);
 }
 
+
+int
+event_receive_wrap(void *handle, char *event_str,
+        int event_str_sz, char *missed_cnt_str, int missed_cnt_str_sz)
+{
+    event_receive_op_t evt;
+    int rc = 0;
+
+    SWSS_LOG_DEBUG("events_receive_wrap handle=%p event-sz=%d missed-sz=%d\n",
+            handle, event_str_sz, missed_cnt_str_sz);
+
+    evt = event_receive(handle);
+
+    if (evt.rc == 0) {
+        nlohmann::json res = nlohmann::json::object();
+        
+        {
+            nlohmann::json params_data = nlohmann::json::object();
+
+            for (event_params_t::const_iterator itc = evt.params.begin(); itc != evt.params.end(); ++itc) {
+                params_data[itc->first] = itc->second;
+            }
+            res[evt.key] = params_data;
+        }
+        string json_str(res.dump());
+        rc = snprintf(event_str, event_str_sz, "%s", json_str.c_str());
+        if (rc >= event_str_sz) {
+            SWSS_LOG_ERROR("truncated event buffer.need=%d given=%d",
+                    rc, event_str_sz);
+            event_str[event_str_sz-1] = 0;
+        }
+
+        int rc_missed = snprintf(missed_cnt_str, missed_cnt_str_sz, "%d", evt.missed_cnt);
+        if (rc_missed >= missed_cnt_str_sz) {
+            SWSS_LOG_ERROR("missed cnt (%d) buffer.need=%d given=%d",
+                    evt.missed_cnt, rc_missed, missed_cnt_str_sz);
+            missed_cnt_str[missed_cnt_str_sz-1] = 0;
+        }
+    }
+    else if (evt.rc > 0) {
+        // timeout
+        rc = 0;
+    }
+    else {
+        rc = evt.rc;
+    }
+
+    SWSS_LOG_DEBUG("events_receive_wrap rc=%d event_str=%s missed=%s\n",
+            rc, event_str, missed_cnt_str);
+
+    return rc;
+}
+
+
+void swssSetLogPriority(int pri)
+{
+    swss::Logger::setMinPrio((swss::Logger::Priority) pri);
+}
+
+

common/events.h

@@ -151,6 +151,13 @@ event_handle_t events_init_subscriber(bool use_cache=false,
 void events_deinit_subscriber(event_handle_t handle);
 
 
+typedef struct {
+    int rc;                 /* Return value of event_receive */
+    std::string key;        /* key */
+    event_params_t params;  /* Params received */
+    int missed_cnt;         /* missed count */
+} event_receive_op_t;
+
 /*
  * Receive an event.
  * A blocking call unless the subscriber is created with a timeout.
@@ -189,40 +196,11 @@ void events_deinit_subscriber(event_handle_t handle);
  *
  * return:
  *  0   - On success
- *  > 0 - On failure, returns zmq_errno, if failure is zmq socket related.
+ *  > 0 - Implies failure due to timeout.
  *  < 0 - For all other failures
  *
  */
-int event_receive(event_handle_t handle, std::string &key,
-        event_params_t &params, int &missed_cnt);
-
-/*
- * event_receive_wrap
- *
- * Returns o/p as structured.
- * This is handy for invocation via python.
- *
- * input:
- *  handle - As obtained from events_init_subscriber
- *
- * output:
- *  None
- *
- * Return:
- *  struct that gets return value and all o/p params of event_receive
- */
-typedef struct {
-    int rc;     /* Return value of event_receive */
-    /* o/p params from event receive */
-    std::string key;
-    event_params_t params;
-    int missed_cnt;
-} event_receive_op_t;
-
-event_receive_op_t event_receive_wrap(event_handle_t handle);
+event_receive_op_t event_receive(event_handle_t handle);
 
-/* Non ZMQ Error codes */
-#define ERR_MESSAGE_INVALID -2
-#define ERR_OTHER -1
 
 #endif /* !_EVENTS_H */ 

common/events_common.h

@@ -18,16 +18,28 @@
 #include <boost/archive/text_oarchive.hpp>
 
 #include "logger.h"
-#include "events.h"
 
 using namespace std;
 using namespace chrono;
 
+#define ERR_MESSAGE_INVALID -2
+#define ERR_OTHER -1
+
 /*
  * Max count of possible concurrent event publishers
  * We maintain a cache of last seen sequence number per publisher.
  * This provides a MAX ceiling for cache.
- * Any more publishers over this count should indicate a serious bug.
+ * A publisher exiting will be not known to receiver, so some publishers
+ * who gets periodically invoked as every N seconds, could over time
+ * cause the cache over flow. 
+ * whenever the cache hits this max, old instances are removed.
+ *
+ * RFE: Publishers who publish at least 1 event, declare their
+ *      exit via a reserved event, so receivers could drop tracking
+ *      hence avoid overflow. Else, a long running process with no event
+ *      but still active could lose its record. This is still not very
+ *      serious, as first message from such publisher is less likely
+ *      to be lost and it could create a record.
  */
 #define MAX_PUBLISHERS_COUNT  1000