Session logging

This fix adds the ability to provide any/all of session destinations
for SLO and sampled sessions. New flags SESSION.slo_destination
and SESSION.sample_destination will override the existing
DEFAULT.usesyslog and DEFAULT.log_flow flags.By default, the destination
list is collector. We have config variables to control SLO and sampling
rate if we are inundated.
Depends-On: I486656dc9d0866c395aabf9b9ed36cc129a12797
Partial-Bug: 1729812

Change-Id: I79f7c8879d3c1672399e86e6b7a7cc251579bd13

src/sandesh/common/flow_sandesh_impl.cc

@@ -131,15 +131,17 @@ std::string SessionAggInfoLog(
     return Xbuf.str();
 }
 
-
 void SessionEndpointObject::LogUnrolled(std::string category,
     SandeshLevel::type level,
     const std::vector<SessionEndpoint> & session_data) {
     if (!IsLevelCategoryLoggingAllowed(SandeshType::SESSION, level, category)) {
         return;
     }
     log4cplus::LogLevel Xlog4level(SandeshLevelTolog4Level(level));
-    log4cplus::Logger Xlogger = Sandesh::logger();
+    log4cplus::Logger Xlogger = Sandesh::sampled_logger();
+    log4cplus::Logger SLO_logger = Sandesh::slo_logger();
+    bool is_send_sample_to_logger_enabled = Sandesh::is_send_sampled_to_logger_enabled();
+    bool is_send_slo_to_logger_enabled = Sandesh::is_send_slo_to_logger_enabled();
     if (!Xlogger.isEnabledFor(Xlog4level)) {
         return;
     }
@@ -174,9 +176,104 @@ void SessionEndpointObject::LogUnrolled(std::string category,
                 // print SessionInfo values [aggregate data for individual session]
                 Xbuf << sessions_iter->second.log() << " ";
                 Xbuf << " ]";
-                Xlogger.forcedLog(Xlog4level, Xbuf.str());
+                // If SLO session log to SLO_logger
+                const SessionInfo & sess_info = sessions_iter->second;
+                if (sess_info.forward_flow_info.__isset.logged_bytes ||
+                    sess_info.reverse_flow_info.__isset.logged_bytes) {
+                    if (is_send_slo_to_logger_enabled) {
+                        SLO_logger.forcedLog(Xlog4level, Xbuf.str());
+                    }
+                }
+                // If sampled session log to Xlogger
+                if (sess_info.forward_flow_info.__isset.sampled_bytes ||
+                    sess_info.reverse_flow_info.__isset.sampled_bytes) {
+                    if (is_send_sample_to_logger_enabled) {
+                        Xlogger.forcedLog(Xlog4level, Xbuf.str());
+                    }
+                }
             }
         }
     }
 }
 
+/*
+ * Remove the sessions from the SessionEndPoint struct
+ * if the session need not go to the collector
+ */
+void SessionEndpointObject::adjust_session_end_point_objects(
+                        std::vector<SessionEndpoint> &session_data) {
+    bool is_send_sampled_to_collector = Sandesh::is_send_sampled_to_collector_enabled();
+    bool is_send_slo_to_collector = Sandesh::is_send_slo_to_collector_enabled();
+    std::vector<SessionEndpoint> ::iterator sep_iter;
+    for (sep_iter = session_data.begin();
+         sep_iter != session_data.end(); ++sep_iter) {
+        std::map<SessionIpPortProtocol, SessionAggInfo>::iterator
+            local_ep_iter;
+        for (local_ep_iter = sep_iter->sess_agg_info.begin();
+             local_ep_iter != sep_iter->sess_agg_info.end();
+             local_ep_iter++) {
+            // Adjust the aggregate logged and sampled info
+            if (!is_send_slo_to_collector) {
+                 local_ep_iter->second.__isset.logged_forward_bytes = false;
+                 local_ep_iter->second.__isset.logged_forward_pkts = false;
+                 local_ep_iter->second.__isset.logged_reverse_bytes = false;
+                 local_ep_iter->second.__isset.logged_reverse_pkts = false;
+             }
+             if (!is_send_sampled_to_collector) {
+                 local_ep_iter->second.__isset.sampled_forward_bytes = false;
+                 local_ep_iter->second.__isset.sampled_forward_pkts = false;
+                 local_ep_iter->second.__isset.sampled_reverse_bytes = false;
+                 local_ep_iter->second.__isset.sampled_reverse_pkts = false;
+             }
+            // Iterate the individual sessions
+            std::map<SessionIpPort, SessionInfo>::iterator sessions_iter;
+            sessions_iter = local_ep_iter->second.sessionMap.begin();
+            while (sessions_iter != local_ep_iter->second.sessionMap.end()) {
+                bool erase_session = false;
+                SessionInfo & sess_info = sessions_iter->second;
+                // Handle if its a logged session
+                if (sess_info.forward_flow_info.__isset.logged_bytes ||
+                    sess_info.reverse_flow_info.__isset.logged_bytes) {
+                    if (!is_send_slo_to_collector) {
+                        // dont erase if session is both sampled and logged
+                        if (!sess_info.forward_flow_info.__isset.
+                             sampled_bytes &&
+                            !sess_info.reverse_flow_info.__isset.
+                             sampled_bytes) {
+                             erase_session = true;
+                        } else {
+                             sess_info.forward_flow_info.__isset.logged_bytes = false;
+                             sess_info.forward_flow_info.__isset.logged_pkts = false;
+                             sess_info.reverse_flow_info.__isset.logged_bytes = false;
+                             sess_info.reverse_flow_info.__isset.logged_pkts = false;
+                        }
+                    }
+                }
+                // Handle if its a sampled session
+                if (sess_info.forward_flow_info.__isset.sampled_bytes ||
+                    sess_info.reverse_flow_info.__isset.sampled_bytes) {
+                    if (!is_send_sampled_to_collector) {
+                        // dont erase if session is both sampled and logged
+                        if (!sess_info.forward_flow_info.__isset.
+                             logged_bytes &&
+                            !sess_info.reverse_flow_info.__isset.
+                             logged_bytes) {
+                             erase_session = true;
+                        } else {
+                             sess_info.forward_flow_info.__isset.sampled_bytes = false;
+                             sess_info.forward_flow_info.__isset.sampled_pkts = false;
+                             sess_info.reverse_flow_info.__isset.sampled_bytes = false;
+                             sess_info.reverse_flow_info.__isset.sampled_pkts = false;
+                        }
+                    }
+                }
+                if (erase_session) {
+                    local_ep_iter->second.sessionMap.
+                                 erase(sessions_iter);
+                }
+                sessions_iter++;
+
+            }
+       }
+    }
+}

src/vnsw/agent/contrail-vrouter-agent.conf

@@ -44,6 +44,7 @@ log_level=SYS_NOTICE
 log_local=1
 
 # Enable/Disable local flow message logging. Possible values are 0 (disable) and 1 (enable)
+# Flag is deprecated as of 5.0.1
 # log_flow=0
 
 # Encapsulation type for tunnel. Possible values are MPLSoGRE, MPLSoUDP, VXLAN
@@ -374,3 +375,11 @@ docker_command=/usr/bin/opencontrail-vrouter-docker
 # Fallback time in seconds to find EOC in case config inactivity is not seen.
 # config_fallback_time=
 # config_inactivity_time=
+
+#[SESSION]
+# This section lets one configure destinations for sampled and SLO sessions
+# The default destination is collector for both sampled and logged sessions
+# sample_destination = collector # values can be any/all of collector, file,
+# syslog. eg., sample_destination = collector file syslog
+# slo_destination = collector # values can be any/all of collector, file,
+# syslog. eg., slo_destination = collector file syslog

src/vnsw/agent/init/agent_init.cc

@@ -106,21 +106,39 @@ int AgentInit::Start() {
     agent_->set_instance_id(InstanceId());
     agent_->set_module_type(ModuleType());
     agent_->set_module_name(module_name);
-
-
+    std::vector<std::string> v_slo_destinations = agent_param_->
+                                                   get_slo_destination();
     std::string log_property_file = agent_param_->log_property_file();
     if (log_property_file.size()) {
         LoggingInit(log_property_file);
     }
     else {
         LoggingInit(agent_param_->log_file(), agent_param_->log_file_size(),
-                    agent_param_->log_files_count(), agent_param_->use_syslog(),
+                    agent_param_->log_files_count(), false,
                     agent_param_->syslog_facility(), module_name,
                     SandeshLevelTolog4Level(
                         Sandesh::StringToLevel(agent_param_->log_level())));
     }
     agent_param_->LogConfig();
 
+    // Set the sample logger params
+    Sandesh::set_logger_appender(agent_param_->log_file(),
+                                 agent_param_->log_file_size(),
+                                 agent_param_->log_files_count(),
+                                 agent_param_->syslog_facility(),
+                                 agent_param_->get_sample_destination(),
+                                 module_name, true);
+    // Set the SLO logger params
+    Sandesh::set_logger_appender(agent_param_->log_file(),
+                                 agent_param_->log_file_size(),
+                                 agent_param_->log_files_count(),
+                                 agent_param_->syslog_facility(),
+                                 agent_param_->get_slo_destination(),
+                                 module_name, false);
+
+    Sandesh::set_send_to_collector_flags(agent_param_->get_sample_destination(),
+                                         agent_param_->get_slo_destination());
+
     int ret = agent_param_->Validate();
     if (ret != 0) {
         return ret;

src/vnsw/agent/init/agent_param.cc

@@ -345,6 +345,30 @@ void AgentParam::ParseQueue() {
     }
 }
 
+void AgentParam::ParseSessionDestinationArguments
+    (const boost::program_options::variables_map &var_map) {
+    slo_destination_.clear();
+    sample_destination_.clear();
+    GetOptValue< vector<string> >(var_map, slo_destination_,
+                                  "SESSION.slo_destination");
+    GetOptValue< vector<string> >(var_map, sample_destination_,
+                                  "SESSION.sample_destination");
+    //validate the string
+    std::set<string> valid_dest_values;
+    valid_dest_values.insert("collector");
+    valid_dest_values.insert("file");
+    valid_dest_values.insert("syslog");
+    valid_dest_values.insert("");
+    for (uint32_t i=0; i<slo_destination_.size(); i++) {
+        assert(valid_dest_values.find(slo_destination_[i]) !=
+            valid_dest_values.end());
+    }
+    for (uint32_t i=0; i<sample_destination_.size(); i++) {
+        assert(valid_dest_values.find(sample_destination_[i]) !=
+            valid_dest_values.end());
+    }
+}
+
 void AgentParam::ParseCollectorArguments
     (const boost::program_options::variables_map &var_map) {
     collector_server_list_.clear();
@@ -789,6 +813,7 @@ void AgentParam::ProcessArguments() {
     ParseRestartArguments(var_map_);
     ParseMacLearning(var_map_);
     ParseTsnServersArguments(var_map_);
+    ParseSessionDestinationArguments(var_map_);
     return;
 }
 
@@ -807,6 +832,7 @@ void AgentParam::ReInitFromConfig() {
         ParseDnsServersArguments(var_map);
         ParseCollectorArguments(var_map);
         ParseTsnServersArguments(var_map);
+        ParseSessionDestinationArguments(var_map);
 
         LogFilteredConfig();
     }
@@ -1027,7 +1053,6 @@ void AgentParam::Init(const string &config_file, const string &program_name) {
 
     config_file_ = config_file;
     program_name_ = program_name;
-
     InitFromSystem();
     InitFromConfig();
     ProcessArguments();
@@ -1502,8 +1527,10 @@ AgentParam::AgentParam(bool enable_flow_options,
          "List of 2M Huge pages to be used by vrouter");
     options_.add(restart);
     config_file_options_.add(restart);
-
     opt::options_description log("Logging options");
+    std::vector<std::string> default_session_destination;
+    std::vector<std::string> implicit_session_destination;
+    default_session_destination.push_back("collector");
     log.add_options()
         ("DEFAULT.log_category", opt::value<string>()->default_value(""),
          "Category filter for local logging of sandesh messages")
@@ -1526,7 +1553,16 @@ AgentParam::AgentParam(bool enable_flow_options,
          "Enable local logging of flow sandesh messages")
         ("DEFAULT.log_property_file", opt::value<string>()->default_value(""),
          "Log Property File")
-        ;
+        ("SESSION.slo_destination",
+         opt::value<vector<string> >()
+             ->default_value(default_session_destination, std::string("collector"))
+             ->implicit_value(implicit_session_destination, std::string("")),
+        "List of destinations. valid values are collector, file, syslog. Space delimited")
+        ("SESSION.sample_destination",
+         opt::value<std::vector<std::string> >()
+             ->default_value(default_session_destination, std::string("collector"))
+             ->implicit_value(implicit_session_destination, std::string("")),
+        "List of destinations. valid values are collector, file, syslog. Space delimited");
     options_.add(log);
     config_file_options_.add(log);
 

src/vnsw/agent/init/agent_param.h

@@ -278,6 +278,12 @@ class AgentParam  {
     const long log_file_size() const { return log_file_size_; }
     bool log_local() const { return log_local_; }
     bool log_flow() const { return log_flow_; }
+    const std::vector<std::string> &get_sample_destination() {
+        return sample_destination_;
+    }
+    const std::vector<std::string> &get_slo_destination() {
+        return slo_destination_;
+    }
     const std::string &log_level() const { return log_level_; }
     const std::string &log_category() const { return log_category_; }
     const std::string &log_property_file() const { return log_property_file_; }
@@ -615,6 +621,8 @@ class AgentParam  {
         (const boost::program_options::variables_map &v);
     void ParseMacLearning
         (const boost::program_options::variables_map &v);
+    void ParseSessionDestinationArguments
+        (const boost::program_options::variables_map &v);
 
     boost::program_options::variables_map var_map_;
     boost::program_options::options_description options_;
@@ -675,6 +683,8 @@ class AgentParam  {
     std::string log_property_file_;
     bool log_local_;
     bool log_flow_;
+    std::vector<std::string> slo_destination_;
+    std::vector<std::string> sample_destination_;
     std::string log_level_;
     std::string log_category_;
     bool use_syslog_;

src/vnsw/agent/init/test/test_agent_init.cc

@@ -547,6 +547,49 @@ TEST_F(AgentParamTest, Agent_Mac_Learning_Option_1) {
     EXPECT_EQ(param.mac_learning_update_tokens(), 510);
     EXPECT_EQ(param.mac_learning_delete_tokens(), 520);
 }
+
+TEST_F(AgentParamTest, Agent_Session_Destination_Option_Arguments) {
+    int argc = 3;
+    char *argv[] = {
+        (char *) "",
+        (char *) "--SESSION.slo_destination", (char *) "collector file",
+        (char *) "--SESSION.sample_destination", (char *) "file",
+    };
+
+    AgentParam param;
+    param.ParseArguments(argc, argv);
+    param.Init("controller/src/vnsw/agent/init/test/flows.ini", "test-param");
+    vector<string> slo_destination_list;
+    slo_destination_list.push_back(string("collector"));
+    slo_destination_list.push_back(string("file"));
+    TASK_UTIL_EXPECT_VECTOR_EQ(param.get_slo_destination(),
+                     slo_destination_list);
+    // when options are unspecified, the sessions are sent only to collector
+    argc = 1;
+    char *argv1[] = {
+        (char *) "",
+    };
+    AgentParam p;
+    p.ParseArguments(argc, argv1);
+    p.Init("controller/src/vnsw/agent/init/test/flows.ini", "test-param");
+    slo_destination_list.pop_back();
+    TASK_UTIL_EXPECT_VECTOR_EQ(p.get_slo_destination(),
+                     slo_destination_list);
+    // when empty list is passed for destination, slo_destn_list is empty
+    argc = 2;
+    char *argv2[] = {
+        (char *) "",
+        (char *) "--SESSION.slo_destination", (char *) "",
+    };
+    AgentParam p1;
+    p1.ParseArguments(argc, argv2);
+    p1.Init("controller/src/vnsw/agent/init/test/flows.ini", "test-param");
+    slo_destination_list.pop_back();
+    TASK_UTIL_EXPECT_VECTOR_EQ(p1.get_slo_destination(),
+                     slo_destination_list);
+
+}
+
 int main(int argc, char **argv) {
     ::testing::InitGoogleTest(&argc, argv);
     int ret = RUN_ALL_TESTS();