Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include <iostream>
7 : #include <cstdlib>
8 : #include <string>
9 : #include <map>
10 :
11 : #include <boost/array.hpp>
12 : #include <boost/bind/bind.hpp>
13 : #include <boost/shared_ptr.hpp>
14 : #include <boost/asio.hpp>
15 : #include <boost/spirit/include/qi.hpp>
16 : #include <boost/spirit/include/phoenix.hpp>
17 : #include <boost/spirit/include/phoenix_core.hpp>
18 : #include <boost/spirit/include/phoenix_operator.hpp>
19 : #include <boost/spirit/include/phoenix_stl.hpp>
20 : #if BOOST_VERSION >= 105600
21 : #include <boost/phoenix/object/construct.hpp>
22 : #else
23 : #include <boost/spirit/home/phoenix/object/construct.hpp>
24 : #endif
25 : #include <boost/uuid/uuid.hpp>
26 : #if __GNUC_PREREQ(4, 6)
27 : #pragma GCC diagnostic push
28 : #pragma GCC diagnostic ignored "-Wunused-result"
29 : #endif
30 : #include <boost/uuid/uuid_generators.hpp>
31 : #if __GNUC_PREREQ(4, 6)
32 : #pragma GCC diagnostic pop
33 : #endif
34 : #include <boost/assign/list_of.hpp>
35 : #include <boost/fusion/adapted/std_pair.hpp>
36 : #include <boost/ptr_container/ptr_map.hpp>
37 :
38 : #include <base/util.h>
39 : #include <base/logging.h>
40 :
41 : #include <sandesh/sandesh_message_builder.h>
42 :
43 : #include "generator.h"
44 : #include "syslog_collector.h"
45 : #include <boost/date_time/posix_time/posix_time.hpp>
46 :
47 : //#define SYSLOG_DEBUG 0
48 : /*** test for burst (compile <string> with -lboost_date_time -lboost_thread and
49 : * no -fno-exceptions
50 : #include <boost/thread/thread.hpp>
51 : **/
52 :
53 : using boost::asio::ip::udp;
54 : using namespace boost::asio;
55 : namespace qi = boost::spirit::qi;
56 : namespace ascii = boost::spirit::standard;
57 : namespace bt = boost::posix_time;
58 : namespace phx = boost::phoenix;
59 : using namespace boost::placeholders;
60 :
61 :
62 : class SyslogQueueEntry;
63 :
64 : class SyslogTcpSession : public TcpSession
65 : {
66 : public:
67 : typedef boost::intrusive_ptr<SyslogTcpSession> SyslogTcpSessionPtr;
68 :
69 : SyslogTcpSession (SyslogTcpListener *server, Socket *socket);
70 : virtual void OnRead (const boost::asio::const_buffer buf);
71 : };
72 :
73 : class TCPSyslogQueueEntry : public SyslogQueueEntry
74 : {
75 : private:
76 : typedef boost::intrusive_ptr<SyslogTcpSession> SyslogTcpSessionPtr;
77 : public:
78 0 : TCPSyslogQueueEntry (SyslogTcpSessionPtr ses, boost::asio::const_buffer b,
79 0 : ip::tcp::endpoint e):
80 0 : SyslogQueueEntry (b, buffer_size (b), e.address ().to_string (),
81 0 : e.port ()),
82 0 : buf_ (b), ep_ (e), session_(ses) {
83 0 : }
84 : virtual void free ();
85 0 : virtual ~TCPSyslogQueueEntry() {}
86 : private:
87 : boost::asio::const_buffer buf_;
88 : ip::tcp::endpoint ep_;
89 : SyslogTcpSessionPtr session_;
90 : };
91 :
92 : class SyslogUDPListener;
93 : class UDPSyslogQueueEntry : public SyslogQueueEntry
94 : {
95 : public:
96 :
97 0 : UDPSyslogQueueEntry (SyslogUDPListener* svr, udp::endpoint ep,
98 0 : const boost::asio::const_buffer &d, size_t l):
99 0 : SyslogQueueEntry (d, l, ep.address ().to_string (), ep.port ()),
100 0 : ep_ (ep), b_(d), server_ (svr)
101 : {
102 0 : }
103 : virtual void free ();
104 0 : virtual ~UDPSyslogQueueEntry() {}
105 : private:
106 : udp::endpoint ep_;
107 : boost::asio::const_buffer b_;
108 : SyslogUDPListener *server_;
109 : };
110 :
111 :
112 1 : SyslogParser::SyslogParser (SyslogListeners *syslog):
113 1 : work_queue_(TaskScheduler::GetInstance()->GetTaskId(
114 : "vizd::syslog"), 0, boost::bind(
115 : &SyslogParser::ClientParse, this, _1)),
116 2 : syslog_(syslog)
117 : {
118 1 : Init();
119 1 : }
120 6 : SyslogParser::~SyslogParser ()
121 : {
122 5 : genarators_.erase (genarators_.begin (), genarators_.end ());
123 6 : }
124 0 : void SyslogParser::Parse (SyslogQueueEntry *sqe) {
125 0 : work_queue_.Enqueue (sqe);
126 0 : }
127 :
128 1 : void SyslogParser::Shutdown ()
129 : {
130 1 : work_queue_.ScheduleShutdown ();
131 1 : LOG(DEBUG, __func__ << " Syslog parser shutdown done");
132 1 : }
133 :
134 4 : SyslogParser::SyslogParser ():
135 4 : work_queue_(TaskScheduler::GetInstance()->GetTaskId(
136 : "vizd::syslog"), 0, boost::bind(
137 : &SyslogParser::ClientParse, this, _1)),
138 8 : syslog_(0)
139 : {
140 4 : Init();
141 4 : }
142 :
143 5 : void SyslogParser::Init()
144 : {
145 5 : facilitynames_ = boost::assign::list_of ("auth") ("authpriv")
146 5 : ("cron") ("daemon") ("ftp") ("kern") ("lpr") ("mail") ("mark")
147 5 : ("news") ("security") ("syslog") ("user") ("uucp") ("local0")
148 5 : ("local1") ("local2") ("local3") ("local4") ("local5")
149 5 : ("local6") ("local7").convert_to_container<vector<string> >();
150 5 : }
151 0 : void SyslogParser::WaitForIdle (int max_wait)
152 : {
153 : int i;
154 0 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
155 0 : for (i = 0; !scheduler->IsEmpty() && i < max_wait; i++) {
156 0 : usleep (1000);
157 0 : LOG(DEBUG, __func__ << " Syslog queue empty? " <<
158 : scheduler->IsEmpty() << ":" << i << "/" << max_wait);
159 : }
160 0 : LOG(DEBUG, __func__ << " Syslog queue empty? " <<
161 : scheduler->IsEmpty() << ":" << i << "/" << max_wait);
162 0 : }
163 :
164 : template <typename Iterator>
165 36 : bool SyslogParser::parse_syslog (Iterator start, Iterator end, syslog_m_t &v)
166 : {
167 : using qi::int_;
168 : using qi::_1;
169 : using ascii::space;
170 : using phx::insert;
171 : using qi::eps;
172 : using qi::char_;
173 : using qi::lit;
174 : using qi::lexeme;
175 :
176 144 : qi::rule<Iterator, std::string(), ascii::space_type> word = lexeme[ +(char_ - ' ') ] ;
177 180 : qi::rule<Iterator, std::string(), ascii::space_type> word2 = lexeme[ +(char_ - ':' - ' ') ] ;
178 216 : qi::rule<Iterator, std::string(), ascii::space_type> word3 = lexeme[ +(char_ - '[' - ' ' - ':') ] ;
179 252 : qi::rule<Iterator, std::string(), ascii::space_type> word4 = lexeme[ +(char_ - '[' - ' ' - ':') >> " " ] ;
180 108 : qi::rule<Iterator, std::string(), ascii::space_type> body = lexeme[ *char_ ] ;
181 36 : qi::int_parser<int, 10, 1, 3> int3_p;
182 36 : qi::int_parser<int, 10, 1, 2> int2_p;
183 36 : qi::int_parser<int, 10, 1, 1> int1_p;
184 : struct months_: qi::symbols<char, int>
185 : {
186 36 : months_()
187 36 : {
188 : add
189 36 : ("Jan", 1)
190 36 : ("Feb", 2)
191 36 : ("Mar", 3)
192 36 : ("Apr", 4)
193 36 : ("May", 5)
194 36 : ("Jun", 6)
195 36 : ("Jul", 7)
196 36 : ("Aug", 8)
197 36 : ("Sep", 9)
198 36 : ("Oct", 10)
199 36 : ("Nov", 11)
200 36 : ("Dec", 12)
201 : ;
202 36 : }
203 36 : } months;
204 :
205 72 : bool r = qi::phrase_parse(start, end,
206 : // Begin grammar
207 : //facility severity
208 : (
209 : (
210 36 : int_ [insert(phx::ref(v),
211 36 : phx::construct<std::pair<std::string, Holder> >("msglen",
212 0 : phx::construct<Holder>("msglen", _1)))]
213 108 : >> '<' >> int_ [insert(phx::ref(v),
214 36 : phx::construct<std::pair<std::string, Holder> >("facsev",
215 108 : phx::construct<Holder>("facsev", _1)))] >> '>'
216 72 : |'<' >> int_ [insert(phx::ref(v),
217 36 : phx::construct<std::pair<std::string, Holder> >("facsev",
218 108 : phx::construct<Holder>("facsev", _1)))] >> '>'
219 : )
220 0 : >> (
221 : //month
222 36 : months [insert(phx::ref(v),
223 36 : phx::construct<std::pair<std::string, Holder> >("month",
224 0 : phx::construct<Holder>("month", _1)))]
225 : //day
226 72 : >> int_ [insert(phx::ref(v),
227 36 : phx::construct<std::pair<std::string, Holder> >("day",
228 36 : phx::construct<Holder>("day", _1)))]
229 : //hour
230 72 : >> int_ [insert(phx::ref(v),
231 36 : phx::construct<std::pair<std::string, Holder> >("hour",
232 72 : phx::construct<Holder>("hour", _1)))] >> lit(":")
233 : //min
234 72 : >> int_ [insert(phx::ref(v),
235 36 : phx::construct<std::pair<std::string, Holder> >("min",
236 72 : phx::construct<Holder>("min", _1)))] >> lit(":")
237 : //sec
238 72 : >> int_ [insert(phx::ref(v),
239 36 : phx::construct<std::pair<std::string, Holder> >("sec",
240 36 : phx::construct<Holder>("sec", _1)))]
241 : //hostname
242 144 : >> -word4 [insert(phx::ref(v),
243 36 : phx::construct<std::pair<std::string, Holder> >("hostname",
244 36 : phx::construct<Holder>("hostname", _1)))]
245 : //version
246 72 : |int_ [insert(phx::ref(v),
247 36 : phx::construct<std::pair<std::string, Holder> >("version",
248 36 : phx::construct<Holder>("version", _1)))]
249 : //year
250 72 : >>int_ [insert(phx::ref(v),
251 36 : phx::construct<std::pair<std::string, Holder> >("year",
252 72 : phx::construct<Holder>("year", _1)))] >> lit("-")
253 : //month
254 72 : >>int_ [insert(phx::ref(v),
255 36 : phx::construct<std::pair<std::string, Holder> >("month",
256 72 : phx::construct<Holder>("month", _1)))] >> lit("-")
257 : //day
258 72 : >> int_ [insert(phx::ref(v),
259 36 : phx::construct<std::pair<std::string, Holder> >("day",
260 72 : phx::construct<Holder>("day", _1)))] >> lit("T")
261 : //hour
262 72 : >> int_ [insert(phx::ref(v),
263 36 : phx::construct<std::pair<std::string, Holder> >("hour",
264 72 : phx::construct<Holder>("hour", _1)))] >> lit(":")
265 : //min
266 72 : >> int_ [insert(phx::ref(v),
267 36 : phx::construct<std::pair<std::string, Holder> >("min",
268 72 : phx::construct<Holder>("min", _1)))] >> lit(":")
269 : //sec
270 72 : >> int_ [insert(phx::ref(v),
271 36 : phx::construct<std::pair<std::string, Holder> >("sec",
272 72 : phx::construct<Holder>("sec", _1)))] >> lit(".")
273 : //msec
274 72 : >> int_ [insert(phx::ref(v),
275 36 : phx::construct<std::pair<std::string, Holder> >("msec",
276 36 : phx::construct<Holder>("msec", _1)))]
277 0 : >>( lit("Z")
278 : // tz
279 108 : |word [insert(phx::ref(v),
280 36 : phx::construct<std::pair<std::string, Holder> >("tz",
281 36 : phx::construct<Holder>("tz", _1)))]
282 : )
283 : //hostname
284 108 : >> word [insert(phx::ref(v),
285 36 : phx::construct<std::pair<std::string, Holder> >("hostname",
286 36 : phx::construct<Holder>("hostname", _1)))]
287 : )
288 : //tag body
289 : // tag=prog[pid]:
290 72 : >> ( ( word3 [insert(phx::ref(v),
291 36 : phx::construct<std::pair<std::string, Holder> >("prog",
292 36 : phx::construct<Holder>("prog", _1)))] >> lit("[")
293 72 : >> int_ [insert(phx::ref(v),
294 36 : phx::construct<std::pair<std::string, Holder> >("pid",
295 108 : phx::construct<Holder>("pid", _1)))] >> lit("]:")
296 : // tag=prog:
297 72 : |word2 [insert(phx::ref(v),
298 36 : phx::construct<std::pair<std::string, Holder> >("prog",
299 0 : phx::construct<Holder>("prog", _1))),
300 36 : insert(phx::ref(v),
301 36 : phx::construct<std::pair<std::string, Holder> >("pid",
302 108 : phx::construct<Holder>("pid", -1)))] >> lit(":")
303 : // tag=prog -
304 36 : |word2 [insert(phx::ref(v),
305 36 : phx::construct<std::pair<std::string, Holder> >("prog",
306 0 : phx::construct<Holder>("prog", _1))),
307 36 : insert(phx::ref(v),
308 36 : phx::construct<std::pair<std::string, Holder> >("pid",
309 108 : phx::construct<Holder>("pid", -1)))] >> lit("-")
310 : // body
311 : )
312 108 : >>body [insert(phx::ref(v),
313 36 : phx::construct<std::pair<std::string, Holder> >("body",
314 36 : phx::construct<Holder>("body", _1)))]
315 : //body w/ no tag
316 108 : |body [insert(phx::ref(v),
317 36 : phx::construct<std::pair<std::string, Holder> >("prog",
318 36 : phx::construct<Holder>("prog", ""))),
319 36 : insert(phx::ref(v),
320 36 : phx::construct<std::pair<std::string, Holder> >("pid",
321 36 : phx::construct<Holder>("pid", -1))),
322 36 : insert(phx::ref(v),
323 36 : phx::construct<std::pair<std::string, Holder> >("body",
324 36 : phx::construct<Holder>("body", _1)))]
325 : )
326 : )
327 : //end grammar
328 : , space);
329 72 : return (start == end) && r;
330 36 : }
331 :
332 67 : std::string SyslogParser::GetMapVals (syslog_m_t v, std::string key, std::string def)
333 : {
334 67 : std::map<std::string, Holder>::iterator i = v.find (key);
335 67 : if (i == v.end())
336 26 : return def;
337 41 : return i->second.s_val;
338 : }
339 :
340 398 : int64_t SyslogParser::GetMapVal (syslog_m_t v, std::string key, int def)
341 : {
342 398 : std::map<std::string, Holder>::iterator i = v.find (key);
343 398 : if (i == v.end())
344 29 : return def;
345 369 : return i->second.i_val;
346 : }
347 :
348 33 : void SyslogParser::GetFacilitySeverity (syslog_m_t v, int& facility, int& severity)
349 : {
350 33 : int fs = GetMapVal (v, "facsev", 0);
351 33 : severity = fs & 0x7;
352 33 : facility = fs >> 3;
353 33 : }
354 :
355 33 : void SyslogParser::GetTimestamp (syslog_m_t v, time_t& timestamp)
356 : {
357 33 : bt::ptime lt(bt::microsec_clock::local_time());
358 33 : bt::ptime ut(bt::microsec_clock::universal_time());
359 33 : tm pt_tm = bt::to_tm(lt);
360 66 : bt::ptime p(boost::gregorian::date (GetMapVal (v, "year", pt_tm.tm_year+1900),
361 66 : GetMapVal (v, "month", pt_tm.tm_mon),
362 66 : GetMapVal (v, "day", pt_tm.tm_mday)), bt::time_duration(
363 : GetMapVal (v, "hour", pt_tm.tm_hour),
364 : GetMapVal (v, "min", pt_tm.tm_min),
365 : GetMapVal (v, "sec", pt_tm.tm_sec),
366 231 : GetMapVal (v, "msec", 0)*1000));
367 33 : bt::ptime epoch(boost::gregorian::date(1970,1,1));
368 33 : if (GetMapVal (v, "year", 0) == 0) {
369 5 : bt::time_duration diff = lt - ut;
370 5 : timestamp = (p - epoch).total_microseconds()
371 5 : - diff.total_microseconds();
372 : } else {
373 56 : std::string tz = GetMapVals (v, "tz", "+0:00");
374 28 : bt::time_duration diff = bt::duration_from_string(tz.substr(1));
375 28 : if (tz.substr(0,1) == "-") {
376 2 : timestamp = (p - epoch).total_microseconds()
377 2 : + diff.total_microseconds();
378 : } else {
379 26 : timestamp = (p - epoch).total_microseconds()
380 26 : - diff.total_microseconds();
381 : }
382 28 : }
383 : /*
384 : time_t _timestamp = UTCTimestampUsec ();
385 : std::cout << "ts diff " << timestamp - _timestamp << std::endl
386 : << timestamp << std::endl << _timestamp << std::endl;
387 : */
388 33 : }
389 :
390 33 : void SyslogParser::PostParsing (syslog_m_t &v) {
391 : time_t timestamp;
392 33 : GetTimestamp (v, timestamp);
393 33 : v.insert(std::pair<std::string, Holder>("timestamp",
394 66 : Holder("timestamp", timestamp)));
395 : int f, s;
396 33 : GetFacilitySeverity (v, f, s);
397 33 : v.insert(std::pair<std::string, Holder>("facility",
398 66 : Holder("facility", f)));
399 33 : v.insert(std::pair<std::string, Holder>("severity",
400 66 : Holder("severity", s)));
401 33 : }
402 :
403 0 : SyslogGenerator* SyslogParser::GetGenerator (std::string ip)
404 : {
405 : boost::ptr_map<std::string, SyslogGenerator>::iterator i =
406 0 : genarators_.find (ip);
407 0 : if (i == genarators_.end()) {
408 :
409 0 : genarators_.insert (ip, new SyslogGenerator(syslog_, ip,
410 0 : "syslog"));
411 0 : i = genarators_.find (ip);
412 : }
413 0 : return i->second;
414 : }
415 :
416 2 : std::string SyslogParser::GetSyslogFacilityName (uint64_t f)
417 : {
418 2 : if (f < facilitynames_.size ())
419 2 : return facilitynames_[f];
420 0 : return "";
421 : }
422 :
423 2 : std::string SyslogParser::EscapeXmlTags (std::string text)
424 : {
425 2 : std::ostringstream s;
426 : #ifdef SYSLOG_DEBUG
427 : std::ostringstream ff, bb, ft;
428 : int i = 0;
429 :
430 : ft << "|" << text << "|\n";
431 : #endif
432 :
433 2 : for (std::string::const_iterator it = text.begin();
434 252 : it != text.end(); ++it) {
435 250 : switch(*it) {
436 0 : case '&': s << "&"; continue;
437 : //case '"': s << """; continue;
438 2 : case '\'': s << "'"; continue;
439 0 : case '<': s << "<"; continue;
440 0 : case '>': s << ">"; continue;
441 248 : default: if (!(0x80 & *it))
442 245 : s << *it;
443 : else
444 3 : s << "&#" << (int)((uint8_t)*it) << ";";
445 : }
446 : #ifdef SYSLOG_DEBUG
447 : if (!(i % 16)) {
448 : ft << std::endl << ff.str() + " " + bb.str();
449 : ff.str("");
450 : bb.str("");
451 : ff << std::setfill('0') << std::setw(4) << std::hex << i << " ";
452 : ff << std::setfill('0') << std::setw(2) << std::hex << (int) ((uint8_t) *it) << " ";
453 : if (isprint(*it)) {
454 : bb << *it;
455 : } else {
456 : bb << '.';
457 : }
458 : } else {
459 : ff << std::setfill('0') << std::setw(2) << std::hex << (int) ((uint8_t) *it) << " ";
460 : if (!((i+1) % 8)) {
461 : ff << " ";
462 : }
463 : if (isprint(*it)) {
464 : bb << *it;
465 : } else {
466 : bb << '.';
467 : }
468 : }
469 : i++;
470 : #endif
471 : }
472 : #ifdef SYSLOG_DEBUG
473 : int j, r = i % 16;
474 : ft << std::endl << ff.str();
475 : for (j = r; j < 16; j++)
476 : ft << " ";
477 : if (r < 7)
478 : ft << " ";
479 : if (r < 15)
480 : ft << " ";
481 : ft << " " + bb.str() + "\n[" + s.str() + "]";
482 : LOG(ERROR, __func__ << ft.str());
483 : #endif
484 :
485 4 : return s.str();
486 2 : }
487 :
488 2 : std::string SyslogParser::GetMsgBody (syslog_m_t v) {
489 4 : return EscapeXmlTags (GetMapVals (v, "body", ""));
490 : }
491 :
492 2 : std::string SyslogParser::GetModule(syslog_m_t v) {
493 2 : return GetMapVals(v, "prog", "UNKNOWN");
494 : }
495 :
496 2 : std::string SyslogParser::GetFacility(syslog_m_t v) {
497 4 : return GetSyslogFacilityName(GetMapVal(v, "facility", 0));
498 : }
499 :
500 2 : int SyslogParser::GetPID(syslog_m_t v) {
501 2 : return GetMapVal (v, "pid", -1);
502 : }
503 :
504 0 : void SyslogParser::MakeSandesh (syslog_m_t v) {
505 0 : SandeshHeader hdr;
506 0 : std::string ip(GetMapVals(v, "ip", ""));
507 :
508 0 : hdr.set_Timestamp(GetMapVal(v, "timestamp", 0));
509 0 : hdr.set_Module(GetModule(v));
510 0 : hdr.set_Source(GetMapVals(v, "hostname", ip));
511 0 : hdr.set_Type(SandeshType::SYSLOG);
512 0 : hdr.set_Level(GetMapVal(v, "severity", 0));
513 0 : hdr.set_Category(GetFacility(v));
514 0 : hdr.set_IPAddress(ip);
515 :
516 0 : int pid = GetPID(v);
517 0 : if (pid >= 0)
518 0 : hdr.set_Pid(pid);
519 :
520 :
521 0 : std::string body = EscapeXmlTags(GetMapVals (v, "body", ""));
522 0 : std::string xmsg("<Syslog>" + body + "</Syslog>");
523 0 : SandeshMessage *xmessage = syslog_->GetBuilder()->Create(
524 0 : reinterpret_cast<const uint8_t *>(xmsg.c_str()), xmsg.size());
525 0 : SandeshSyslogMessage *smessage =
526 : static_cast<SandeshSyslogMessage *>(xmessage);
527 0 : smessage->SetHeader(hdr);
528 0 : VizMsg vmsg(smessage, umn_gen_());
529 : //ParseMsgBody(body.begin(), body.end(), vmsg.keywords);
530 : //LOG(DEBUG, "[" << body << "]");
531 0 : vmsg.keyword_doc_ = body;
532 :
533 0 : GetGenerator (ip)->ReceiveSandeshMsg (&vmsg, false);
534 0 : vmsg.msg = NULL;
535 0 : delete smessage;
536 0 : }
537 :
538 0 : bool SyslogParser::ClientParse (SyslogQueueEntry *sqe) {
539 0 : std::string ip = sqe->ip;
540 0 : const uint8_t *p = buffer_cast<const uint8_t *>(sqe->data);
541 : #ifdef SYSLOG_DEBUG
542 : LOG(DEBUG, "cnt parser " << sqe->length << " bytes from (" <<
543 : ip << ":" << sqe->port << ")[");
544 :
545 : {
546 : std::string str (p, p + sqe->length);
547 : LOG(DEBUG, str << "]\n");
548 : }
549 : #endif
550 :
551 0 : syslog_m_t v;
552 0 : int len = sqe->length;
553 0 : while (!*(p + len - 1))
554 0 : --len;
555 0 : bool r = SyslogParser::parse_syslog (p, p + len, v);
556 : #ifdef SYSLOG_DEBUG
557 : LOG(DEBUG, "parsed " << r << ".");
558 : #endif
559 0 : if (r) {
560 0 : v.insert(std::pair<std::string, Holder>("ip",
561 0 : Holder("ip", ip)));
562 0 : PostParsing(v);
563 0 : MakeSandesh(v);
564 : }
565 :
566 : #ifdef SYSLOG_DEBUG
567 : LOG(DEBUG, __func__ << " syslog msg from " << ip << ":" <<
568 : GetMapVals (v, "body", "**EMPTY**"));
569 :
570 : int i = 0;
571 : while (!v.empty()) {
572 : Holder d = v.begin()->second;
573 : LOG(DEBUG, i++ << ": " << d.repr());
574 : v.erase(v.begin());
575 : }
576 : #else
577 0 : v.clear ();
578 : #endif
579 0 : sqe->free ();
580 0 : delete sqe;
581 : /*** test for burst
582 : boost::this_thread::sleep(boost::posix_time::milliseconds(20000UL));
583 : **/
584 0 : return r;
585 0 : }
586 :
587 :
588 0 : void SyslogQueueEntry::free ()
589 : {
590 0 : }
591 :
592 1 : SyslogTcpListener::SyslogTcpListener (EventManager *evm,
593 1 : SyslogMsgReadFn read_cb):
594 1 : TcpServer(evm), session_(NULL), read_cb_(read_cb)
595 : {
596 1 : }
597 :
598 0 : TcpSession *SyslogTcpListener::AllocSession(Socket *socket)
599 : {
600 0 : session_ = new SyslogTcpSession (this, socket);
601 0 : return session_;
602 : }
603 1 : void SyslogTcpListener::Shutdown ()
604 : {
605 : // server shutdown
606 1 : TcpServer::Shutdown();
607 1 : }
608 1 : void SyslogTcpListener::Start (std::string ipaddress, int port)
609 : {
610 : //Initialize (port);
611 : //LOG(DEBUG, __func__ << " Initialization of TCP syslog listener @" << port);
612 1 : LOG(ERROR, __func__ << " TCP syslog listener not supported");
613 1 : }
614 :
615 0 : void SyslogTcpListener::ReadMsg(SyslogQueueEntry *sqe) {
616 0 : read_cb_(sqe);
617 0 : }
618 :
619 1 : SyslogUDPListener::SyslogUDPListener (EventManager *evm,
620 1 : SyslogMsgReadFn read_cb): UdpServer (evm), read_cb_(read_cb)
621 : {
622 1 : }
623 1 : void SyslogUDPListener::Shutdown ()
624 : {
625 1 : UdpServer::Shutdown ();
626 1 : }
627 1 : void SyslogUDPListener::Start (std::string ipaddress, int port)
628 : {
629 1 : if (ipaddress.empty())
630 1 : Initialize (port);
631 : else
632 0 : Initialize (ipaddress, port);
633 1 : StartReceive ();
634 1 : LOG(DEBUG, __func__ << " Initialization of UDP syslog listener @" << port);
635 1 : }
636 :
637 0 : void SyslogUDPListener::HandleReceive (
638 : const boost::asio::const_buffer &recv_buffer,
639 : udp::endpoint remote_endpoint,
640 : std::size_t bytes_transferred,
641 : const boost::system::error_code& error)
642 : {
643 : // TODO: handle error
644 : UDPSyslogQueueEntry *sqe = new UDPSyslogQueueEntry (this, remote_endpoint,
645 0 : recv_buffer, bytes_transferred);
646 0 : read_cb_ (sqe);
647 0 : }
648 :
649 :
650 0 : SyslogListeners::SyslogListeners (EventManager *evm, VizCallback cb,
651 : DbHandlerPtr db_handler, std::string ipaddress,
652 0 : int port):
653 0 : parser_(new SyslogParser (this)),
654 0 : udp_listener_(new SyslogUDPListener(evm,
655 0 : boost::bind(&SyslogParser::Parse, parser_.get(), _1))),
656 0 : tcp_listener_(new SyslogTcpListener(evm,
657 0 : boost::bind(&SyslogParser::Parse, parser_.get(), _1))),
658 0 : port_(port),
659 0 : ipaddress_(ipaddress), inited_(false), cb_(cb),
660 0 : db_handler_ (db_handler),
661 0 : builder_ (SandeshMessageBuilder::GetInstance(
662 0 : SandeshMessageBuilder::SYSLOG))
663 : {
664 0 : }
665 :
666 1 : SyslogListeners::SyslogListeners (EventManager *evm, VizCallback cb,
667 1 : DbHandlerPtr db_handler, int port):
668 1 : parser_(new SyslogParser (this)),
669 2 : udp_listener_(new SyslogUDPListener(evm,
670 1 : boost::bind(&SyslogParser::Parse, parser_.get(), _1))),
671 2 : tcp_listener_(new SyslogTcpListener(evm,
672 1 : boost::bind(&SyslogParser::Parse, parser_.get(), _1))),
673 1 : port_(port), ipaddress_(),
674 1 : inited_(false), cb_(cb), db_handler_ (db_handler),
675 1 : builder_ (SandeshMessageBuilder::GetInstance(
676 1 : SandeshMessageBuilder::SYSLOG))
677 : {
678 1 : }
679 :
680 1 : void SyslogListeners::Start ()
681 : {
682 1 : if (port_ >= 0) {
683 1 : tcp_listener_->Start(ipaddress_, port_);
684 1 : udp_listener_->Start(ipaddress_, port_);
685 1 : inited_ = true;
686 : } else {
687 0 : LOG(DEBUG, __func__ << " skip starting syslog listener port:" << port_);
688 : }
689 1 : }
690 0 : bool SyslogListeners::IsRunning ()
691 : {
692 0 : return inited_;
693 : }
694 1 : void SyslogListeners::Shutdown ()
695 : {
696 1 : tcp_listener_->Shutdown ();
697 1 : udp_listener_->Shutdown ();
698 1 : parser_->Shutdown ();
699 1 : TcpServerManager::DeleteServer(tcp_listener_);
700 1 : UdpServerManager::DeleteServer(udp_listener_);
701 1 : inited_ = false;
702 1 : }
703 :
704 : int
705 0 : SyslogListeners::GetTcpPort()
706 : {
707 0 : return tcp_listener_->GetPort();
708 : }
709 :
710 : int
711 1 : SyslogListeners::GetUdpPort()
712 : {
713 1 : return udp_listener_->GetLocalEndpointPort();
714 : }
715 :
716 : void
717 0 : TCPSyslogQueueEntry::free () {
718 0 : session_->server()->DeleteSession (session_.get());
719 0 : session_->ReleaseBuffer(buf_);
720 0 : }
721 : void
722 0 : UDPSyslogQueueEntry::free () {
723 : //server_->DeallocateEndPoint (ep_);
724 0 : server_->DeallocateBuffer (b_);
725 0 : }
726 0 : SyslogTcpSession::SyslogTcpSession (SyslogTcpListener *server, Socket *socket) :
727 0 : TcpSession(server, socket) {
728 : //set_observer(boost::bind(&SyslogTcpSession::OnEvent, this, _1, _2));
729 0 : }
730 : void
731 0 : SyslogTcpSession::OnRead (const boost::asio::const_buffer buf)
732 : {
733 0 : boost::system::error_code ec;
734 : // TODO: handle error
735 0 : TCPSyslogQueueEntry *sqe = new TCPSyslogQueueEntry (SyslogTcpSessionPtr (
736 0 : this), buf, socket ()->remote_endpoint(ec));
737 0 : SyslogTcpListener *sserver = dynamic_cast<SyslogTcpListener *>(server());
738 0 : sserver->ReadMsg(sqe);
739 0 : }
740 :
|
