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