Merge remote-tracking branch 'origin/syslog_telemetry_shared' into remanava_events

Author: renukamanavalan

Makefile.am

@@ -1,3 +1,3 @@
-SUBDIRS = common pyext tests
+SUBDIRS = common pyext sonic-db-cli tests
 
 ACLOCAL_AMFLAGS = -I m4

common/Makefile.am

@@ -47,6 +47,7 @@ libswsscommon_la_SOURCES = \
     select.cpp                \
     selectableevent.cpp       \
     selectabletimer.cpp       \
+    signalhandlerhelper.cpp   \
     consumertable.cpp         \
     consumertablebase.cpp     \
     consumerstatetable.cpp    \

common/configdb.h

@@ -94,8 +94,12 @@ class ConfigDBConnector_Native : public SonicV2Connector_Native
             if init_data_handler:
                 init_data_handler(init_callback_data)
 
-            while True:
+            while not SignalHandlerHelper.checkSignal(SIGNAL_INT):
                 item = self.pubsub.listen_message()
+                if 'type' not in item:
+                    # When timeout or cancelled, item will not contains 'type' 
+                    continue
+
                 if item['type'] == 'pmessage':
                     key = item['channel'].split(':', 1)[1]
                     try:

common/pubsub.cpp

@@ -79,20 +79,29 @@ bool PubSub::hasCachedData()
 
 map<string, string> PubSub::get_message(double timeout)
 {
-    map<string, string> ret;
+    return get_message_internal(timeout).second;
+}
+
+MessageResultPair PubSub::get_message_internal(double timeout)
+{
+    MessageResultPair ret;
+
     if (!m_subscribe)
     {
+        ret.first = Select::ERROR;
         return ret;
     }
 
     Selectable *selected;
     int rc = m_select.select(&selected, int(timeout));
+    ret.first = rc;
     switch (rc)
     {
         case Select::ERROR:
             throw RedisError("Failed to select", m_subscribe->getContext());
 
         case Select::TIMEOUT:
+        case Select::SIGNALINT:
             return ret;
 
         case Select::OBJECT:
@@ -110,10 +119,10 @@ map<string, string> PubSub::get_message(double timeout)
     }
 
     auto message = event->getReply<RedisMessage>();
-    ret["type"] = message.type;
-    ret["pattern"] = message.pattern;
-    ret["channel"] = message.channel;
-    ret["data"] = message.data;
+    ret.second["type"] = message.type;
+    ret.second["pattern"] = message.pattern;
+    ret.second["channel"] = message.channel;
+    ret.second["data"] = message.data;
     return ret;
 }
 
@@ -122,14 +131,17 @@ map<string, string> PubSub::get_message(double timeout)
 std::map<std::string, std::string> PubSub::listen_message()
 {
     const double GET_MESSAGE_INTERVAL = 600.0; // in seconds
+    MessageResultPair ret;
     for (;;)
     {
-        auto ret = get_message(GET_MESSAGE_INTERVAL);
-        if (!ret.empty())
+        ret = get_message_internal(GET_MESSAGE_INTERVAL);
+        if (!ret.second.empty() || ret.first == Select::SIGNALINT)
         {
-            return ret;
+            break;
         }
     }
+
+    return ret.second;
 }
 
 shared_ptr<RedisReply> PubSub::popEventBuffer()

common/pubsub.h

@@ -1,13 +1,16 @@
 #pragma once
 #include <map>
 #include <deque>
+#include <utility>
 
 #include "dbconnector.h"
 #include "select.h"
 #include "redisselect.h"
 
 namespace swss {
 
+typedef std::pair<int, std::map<std::string, std::string> > MessageResultPair;
+
 // This class is to emulate python redis-py class PubSub
 // After SWIG wrapping, it should be used in the same way
 class PubSub : protected RedisSelect
@@ -29,6 +32,7 @@ class PubSub : protected RedisSelect
 private:
     /* Pop keyspace event from event buffer. Caller should free resources. */
     std::shared_ptr<RedisReply> popEventBuffer();
+    MessageResultPair get_message_internal(double timeout = 0.0);
 
     DBConnector *m_parentConnector;
     Select m_select;

common/rediscommand.cpp

@@ -49,6 +49,16 @@ void RedisCommand::formatArgv(int argc, const char **argv, const size_t *argvlen
     }
 }
 
+void RedisCommand::format(const vector<string> &commands)
+{
+    vector<const char*> args;
+    for (auto& command : commands)
+    {
+        args.push_back(command.c_str());
+    }
+    formatArgv(static_cast<int>(args.size()), args.data(), NULL);
+}
+
 /* Format HMSET key multiple field value command */
 void RedisCommand::formatHMSET(const std::string &key,
                                const std::vector<FieldValueTuple> &values)

common/rediscommand.h

@@ -25,6 +25,7 @@ class RedisCommand {
 
     void format(const char *fmt, ...);
     void formatArgv(int argc, const char **argv, const size_t *argvlen);
+    void format(const std::vector<std::string> &commands);
 
     /* Format HMSET key multiple field value command */
 #ifndef SWIG

common/redisreply.cpp

@@ -2,6 +2,7 @@
 #include <stdint.h>
 #include <vector>
 #include <iostream>
+#include <sstream>
 #include <system_error>
 #include <functional>
 
@@ -158,7 +159,7 @@ void RedisReply::checkReplyType(int expectedType)
         const char *err = (m_reply->type == REDIS_REPLY_STRING || m_reply->type == REDIS_REPLY_ERROR) ?
             m_reply->str : "NON-STRING-REPLY";
 
-        string errmsg = "Expected to get redis type " + to_string(expectedType) + " got type " + to_string(m_reply->type) + ", err: " + err;
+        string errmsg = "Expected to get redis type " + std::to_string(expectedType) + " got type " + std::to_string(m_reply->type) + ", err: " + err;
         SWSS_LOG_ERROR("%s", errmsg.c_str());
         throw system_error(make_error_code(errc::io_error), errmsg);
     }
@@ -225,4 +226,44 @@ template<> RedisMessage RedisReply::getReply<RedisMessage>()
     return ret;
 }
 
+
+string RedisReply::to_string()
+{
+    return to_string(getContext());
+}
+
+string RedisReply::to_string(redisReply *reply)
+{
+    switch(reply->type)
+    {
+    case REDIS_REPLY_INTEGER:
+        return std::to_string(reply->integer);
+
+    case REDIS_REPLY_STRING:
+    case REDIS_REPLY_ERROR:
+    case REDIS_REPLY_STATUS:
+    case REDIS_REPLY_NIL:
+        return string(reply->str, reply->len);
+
+    case REDIS_REPLY_ARRAY:
+    {
+        stringstream result;
+        for (size_t i = 0; i < reply->elements; i++)
+        {
+            result << to_string(reply->element[i]);
+
+            if (i < reply->elements - 1)
+            {
+                result << endl;
+            }
+        }
+        return result.str();
+    }
+
+    default:
+        SWSS_LOG_ERROR("invalid type %d for message", reply->type);
+        return string();
+    }
+}
+
 }

common/redisreply.h

@@ -92,6 +92,10 @@ class RedisReply
     /* Check that the status is QUEUED, throw exception otherwise */
     void checkStatusQueued();
 
+    std::string to_string();
+
+    static std::string to_string(redisReply *reply);
+
 private:
     void checkStatus(const char *status);
     void checkReply();

common/select.cpp

@@ -1,6 +1,7 @@
 #include "common/selectable.h"
 #include "common/logger.h"
 #include "common/select.h"
+#include "common/signalhandlerhelper.h"
 #include <algorithm>
 #include <stdio.h>
 #include <sys/time.h>
@@ -9,6 +10,7 @@
 #include <unistd.h>
 #include <string.h>
 
+
 using namespace std;
 
 namespace swss {
@@ -97,7 +99,13 @@ int Select::poll_descriptors(Selectable **c, unsigned int timeout)
     {
         ret = ::epoll_wait(m_epoll_fd, events.data(), sz_selectables, timeout);
     }
-    while(ret == -1 && errno == EINTR); // Retry the select if the process was interrupted by a signal
+    while(ret == -1 && errno == EINTR && !SignalHandlerHelper::checkSignal(Signals::SIGNAL_INT)); // Retry the select if the process was interrupted by a signal
+
+    if (SignalHandlerHelper::checkSignal(Signals::SIGNAL_INT))
+    {
+        // Return if the epoll_wait was interrupted by SIGTERM
+        return Select::SIGNALINT;
+    }
 
     if (ret < 0)
         return Select::ERROR;
@@ -190,6 +198,9 @@ std::string Select::resultToString(int result)
         case swss::Select::TIMEOUT:
             return "TIMEOUT";
 
+        case swss::Select::SIGNALINT:
+            return "SIGNALINT";
+
         default:
             SWSS_LOG_WARN("unknown select result: %d", result);
             return "UNKNOWN";

common/select.h

@@ -30,6 +30,7 @@ class Select
         OBJECT = 0,
         ERROR = 1,
         TIMEOUT = 2,
+        SIGNALINT = 3,// Read operation interrupted by SIGINT
     };
 
     int select(Selectable **c, int timeout = -1);

common/selectabletimer.cpp

@@ -21,6 +21,7 @@ SelectableTimer::SelectableTimer(const timespec& interval, int pri)
         SWSS_LOG_THROW("failed to create timerfd, errno: %s", strerror(errno));
     }
     setInterval(interval);
+    m_running = false;
 }
 
 SelectableTimer::~SelectableTimer()
@@ -36,22 +37,40 @@ SelectableTimer::~SelectableTimer()
 
 void SelectableTimer::start()
 {
-    // Set the timer interval and the timer is automatically started
-    int rc = timerfd_settime(m_tfd, 0, &m_interval, NULL);
-    if (rc == -1)
+    m_mutex.lock();
+    if (!m_running)
     {
-        SWSS_LOG_THROW("failed to set timerfd, errno: %s", strerror(errno));
+        // Set the timer interval and the timer is automatically started
+        int rc = timerfd_settime(m_tfd, 0, &m_interval, NULL);
+        if (rc == -1)
+        {
+            SWSS_LOG_THROW("failed to set timerfd, errno: %s", strerror(errno));
+        }
+        else
+        {
+            m_running = true;
+        }
     }
+    m_mutex.unlock();
 }
 
 void SelectableTimer::stop()
 {
-    // Set the timer interval and the timer is automatically started
-    int rc = timerfd_settime(m_tfd, 0, &m_zero, NULL);
-    if (rc == -1)
+    m_mutex.lock();
+    if (m_running)
     {
-        SWSS_LOG_THROW("failed to set timerfd to zero, errno: %s", strerror(errno));
+        // Set the timer interval and the timer is automatically started
+        int rc = timerfd_settime(m_tfd, 0, &m_zero, NULL);
+        if (rc == -1)
+        {
+            SWSS_LOG_THROW("failed to set timerfd to zero, errno: %s", strerror(errno));
+        }
+        else
+        {
+            m_running = false;
+        }
     }
+    m_mutex.unlock();
 }
 
 void SelectableTimer::reset()

common/selectabletimer.h

@@ -3,6 +3,7 @@
 #include <string>
 #include <vector>
 #include <limits>
+#include <mutex>
 #include <sys/timerfd.h>
 #include "selectable.h"
 
@@ -22,6 +23,8 @@ class SelectableTimer : public Selectable
     uint64_t readData() override;
 
 private:
+    std::mutex m_mutex;
+    bool m_running;
     int m_tfd;
     itimerspec m_interval;
     itimerspec m_zero;