Line data    Source code

       1             : //
       2             : // Copyright (c) 2017 Juniper Networks, Inc. All rights reserved.
       3             : // Copyright Contributors to the OpenSDN project
       4             : //
       5             : 
       6             : #ifndef ANALYTICS_STRUCTURED_SYSLOG_SERVER_H_
       7             : #define ANALYTICS_STRUCTURED_SYSLOG_SERVER_H_
       8             : 
       9             : #include <mutex>
      10             : #include <vector>
      11             : 
      12             : #include <boost/system/error_code.hpp>
      13             : #include <boost/asio/ip/udp.hpp>
      14             : 
      15             : #include "stat_walker.h"
      16             : #include "structured_syslog_config.h"
      17             : #include "structured_syslog_kafka_forwarder.h"
      18             : 
      19             : namespace structured_syslog {
      20             : 
      21             : //
      22             : // StructuredSyslog Server
      23             : //
      24             : class StructuredSyslogServer {
      25             :  public:
      26             :     StructuredSyslogServer(EventManager *evm, uint16_t port,
      27             :         const vector<string> &structured_syslog_tcp_forward_dst,
      28             :         const std::string &structured_syslog_kafka_broker,
      29             :         const std::string &structured_syslog_kafka_topic,
      30             :         uint16_t structured_syslog_kafka_partitions,
      31             :         uint64_t structured_syslog_active_session_map_limit,
      32             :         const Options::Kafka &kafka_options,
      33             :         ConfigClientCollector *config_client,
      34             :         StatWalker::StatTableInsertFn stat_db_cb);
      35             :     virtual ~StructuredSyslogServer();
      36             :     bool Initialize();
      37             :     void Shutdown();
      38             :     boost::asio::ip::udp::endpoint GetLocalEndpoint(
      39             :         boost::system::error_code *ec);
      40             : 
      41             :  private:
      42             :     class StructuredSyslogServerImpl;
      43             :     StructuredSyslogServerImpl *impl_;
      44             : };
      45             : 
      46             : class  StructuredSyslogQueueEntry {
      47             : public:
      48             :     size_t      length;
      49             :     boost::shared_ptr<std::string> data;
      50             :     boost::shared_ptr<std::string> json_data;
      51             :     boost::shared_ptr<std::string> skey;
      52             : 
      53             :     StructuredSyslogQueueEntry(boost::shared_ptr<std::string> d, size_t len,
      54             :                                boost::shared_ptr<std::string> jd,
      55             :                                boost::shared_ptr<std::string> key);
      56             :     virtual ~StructuredSyslogQueueEntry();
      57             : };
      58             : 
      59             : class StructuredSyslogTcpForwarderSession;
      60             : 
      61             : class StructuredSyslogTcpForwarder : public TcpServer {
      62             : public:
      63             :     StructuredSyslogTcpForwarder(EventManager *evm, const std::string &ipaddress, int port);
      64             : 
      65             :     virtual TcpSession *AllocSession(Socket *socket);
      66             :     void WriteReady(const boost::system::error_code &ec);
      67             :     void Connect();
      68             :     bool Send(const u_int8_t *data, size_t size, size_t *actual);
      69             :     bool Connected();
      70             :     void Shutdown();
      71             :     StructuredSyslogTcpForwarderSession *GetSession() const { return session_; }
      72             :     void SetSocketOptions();
      73           0 :     std::string GetIpAddress() {return ipaddress_;}
      74           0 :     int GetPort(){return port_;}
      75             : 
      76             : private:
      77             :     std::string ipaddress_;
      78             :     int port_;
      79             :     StructuredSyslogTcpForwarderSession *session_;
      80             :     std::mutex send_mutex_;
      81             :     bool ready_to_send_;
      82             : 
      83             : };
      84             : 
      85             : class StructuredSyslogForwarder {
      86             : public:
      87             :     StructuredSyslogForwarder(EventManager *evm, const vector <std::string> &tcp_forward_dst,
      88             :                                const std::string &structured_syslog_kafka_broker,
      89             :                                const std::string &structured_syslog_kafka_topic,
      90             :                                const Options::Kafka& kafka_options,
      91             :                                uint16_t structured_syslog_kafka_partitions);
      92             : 
      93             :     virtual ~StructuredSyslogForwarder();
      94             :     void Forward(boost::shared_ptr<StructuredSyslogQueueEntry> sqe);
      95             :     void Shutdown();
      96             :     bool kafkaForwarder();
      97             : 
      98             : protected:
      99             :     bool PollTcpForwarder();
     100             :     void PollTcpForwarderErrorHandler(string error_name, string error_message);
     101             :     void Init(const vector <std::string> &tcp_forward_dst,
     102             :                const std::string &structured_syslog_kafka_broker,
     103             :                const std::string &structured_syslog_kafka_topic,
     104             :                const Options::Kafka& kafka_options,
     105             :                uint16_t structured_syslog_kafka_partitions);
     106             : private:
     107             :     EventManager                           *evm_;
     108             :     std::vector<StructuredSyslogTcpForwarder*> tcpForwarder_;
     109             :     KafkaForwarder*                            kafkaForwarder_;
     110             :     Timer                                  *tcpForwarder_poll_timer_;
     111             :     static const int tcpForwarderPollInterval = 60 * 1000; // in ms
     112             : };
     113             : 
     114             : }  // namespace structured_syslog
     115             : 
     116             : #endif  // ANALYTICS_STRUCTURED_SYSLOG_SERVER_H_
     117             :