Line data    Source code

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