1
0

eye-statd.pl 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651
  1. #!/usr/bin/perl -w
  2. use utf8;
  3. use open ":encoding(utf8)";
  4. use English;
  5. use base;
  6. use FindBin '$Bin';
  7. use lib "/opt/Eye/scripts";
  8. use strict;
  9. use DBI;
  10. use Time::Local;
  11. use Net::Patricia;
  12. use Data::Dumper;
  13. use Date::Parse;
  14. use DateTime;
  15. use eyelib::config;
  16. use eyelib::main;
  17. use eyelib::net_utils;
  18. use eyelib::database;
  19. use Socket qw(AF_INET6 inet_ntop);
  20. use IO::Socket;
  21. use Data::Dumper;
  22. use threads;
  23. my @router_ref = ();
  24. my @interfaces = ();
  25. my %router_svi;
  26. my %routers;
  27. my %wan_dev;
  28. my %lan_dev;
  29. my @traffic = ();
  30. my $saving = 0;
  31. #user statistics for cached data
  32. my %user_stats;
  33. my $MAXREAD = 9216;
  34. my $timeshift = get_option($dbh,55)*60;
  35. my $save_path = get_option($dbh,72);
  36. my $thread_count = $cpu_count;
  37. #save traffic to DB
  38. my $traf_lastflush = time();
  39. # NetFlow
  40. my $server_port = 2055;
  41. my $netflow5_header_len = 24;
  42. my $netflow5_flowrec_len = 48;
  43. my $netflow9_header_len = 20;
  44. my $netflow9_templates = {};
  45. # reap dead children
  46. $SIG{CHLD} = \&REAPER;
  47. $SIG{TERM} = \&TERM;
  48. $SIG{INT} = \&TERM;
  49. $SIG{HUP} = \&INIT;
  50. sub REAPER {
  51. wait;
  52. $saving = 0;
  53. $SIG{CHLD} = \&REAPER;
  54. }
  55. sub TERM {
  56. print "SIGTERM received\n";
  57. flush_traffic(1);
  58. while (wait() != -1) {}
  59. exit 0;
  60. }
  61. sub INIT {
  62. # Create new database handle. If we can't connect, die()
  63. my $hdb = init_db();
  64. InitSubnets();
  65. init_option($hdb);
  66. $save_path = get_option($dbh,72);
  67. $timeshift = get_option($hdb,55)*60;
  68. @router_ref = get_records_sql($hdb,"SELECT * FROM devices WHERE deleted=0 AND device_type=2 AND snmp_version>0 ORDER by ip" );
  69. @interfaces = get_records_sql($hdb,"SELECT * FROM `device_l3_interfaces` ORDER by device_id" );
  70. #router device_id by known device ip
  71. foreach my $row (@router_ref) {
  72. $routers{$row->{id}}=$row;
  73. my @auth_list = get_records_sql($hdb,"SELECT ip FROM User_auth WHERE deleted=0 AND user_id=".$row->{user_id});
  74. foreach my $auth (@auth_list) {
  75. $router_svi{$auth->{ip}}{id}=$row->{id};
  76. $router_svi{$auth->{ip}}{save}=$row->{netflow_save};
  77. }
  78. }
  79. #snmp index for WAN/LAN interface by device id
  80. foreach my $row (@interfaces) {
  81. if ($row->{interface_type}) { $wan_dev{$row->{device_id}}{$row->{snmpin}}=1; } else { $lan_dev{$row->{device_id}}{$row->{snmpin}}=1; }
  82. }
  83. #get userid list
  84. my @auth_list_ref = get_records_sql($hdb,"SELECT id,ip,save_traf FROM User_auth where deleted=0 ORDER by id");
  85. foreach my $row (@auth_list_ref) {
  86. $user_stats{$row->{ip}}{auth_id}=$row->{id};
  87. $user_stats{$row->{ip}}{save_traf}=$row->{save_traf};
  88. }
  89. $hdb->disconnect();
  90. }
  91. ############### MAIN ##########################
  92. #close default database
  93. $dbh->disconnect();
  94. INIT();
  95. my $lsn_nflow;
  96. my $sel = IO::Select->new();
  97. # prepare to listen for NetFlow UDP packets
  98. if ($server_port > 0) {
  99. $lsn_nflow = IO::Socket::INET->new(LocalPort => $server_port, Proto => "udp")
  100. or die "Couldn't be a NetFlow UDP server on port $server_port : $@\n";
  101. $sel->add($lsn_nflow);
  102. }
  103. my ($him,$datagram,$flags);
  104. # main datagram receive loop
  105. while (1) {
  106. while (my @ready = $sel->can_read) {
  107. foreach my $server (@ready) {
  108. $him = $server->recv($datagram, $MAXREAD);
  109. next if (!$him);
  110. my ($port, $ipaddr) = sockaddr_in($server->peername);
  111. if (defined($lsn_nflow) && $server == $lsn_nflow) {
  112. my ($version) = unpack("n", $datagram);
  113. if ($version == 5) {
  114. parse_netflow_v5($datagram, $ipaddr);
  115. } elsif ($version == 9) {
  116. parse_netflow_v9($datagram, $ipaddr);
  117. } else {
  118. print "unknown NetFlow version: $version\n";
  119. }
  120. }
  121. }
  122. }
  123. }
  124. sub parse_netflow_v5 {
  125. my $datagram = shift;
  126. my $ipaddr = shift;
  127. my ($version, $count, $sysuptime, $unix_secs, $unix_nsecs,
  128. $flow_sequence, $engine_type, $engine_id, $aggregation,
  129. $agg_version) = unpack("nnNNNNCCCC", $datagram);
  130. my $flowrecs = substr($datagram, $netflow5_header_len);
  131. #0 - N 0-3 srcaddr Source IP address
  132. #1 - N 4-7 dstaddr Destination IP address
  133. #2 - N 8-11 nexthop IP address of next hop router
  134. #3 - n 12-13 input SNMP index of input interface
  135. #4 - n 14-15 output SNMP index of output interface
  136. #5 - N 16-19 dPkts Packets in the flow
  137. #6 - N 20-23 dOctets Total number of Layer 3 bytes in the packets of the flow
  138. #7 - N 24-27 First SysUptime at start of flow
  139. #8 - N 28-31 Last SysUptime at the time the last packet of the flow was received
  140. #9 - n 32-33 src_port TCP/UDP source port number or equivalent
  141. #10- n 34-35 dst_port TCP/UDP destination port number or equivalent
  142. #11- C 36 pad1 Unused (zero) byte
  143. #12- C 37 tcp_flags Cumulative OR of TCP flags
  144. #13- C 38 prot IP protocol type (for example, TCP = 6; UDP = 17)
  145. #14- C 39 tos IP type of service (ToS)
  146. #15- n 40-41 src_as Autonomous system number of the source, either origin or peer
  147. #16- n 42-43 dst_as Autonomous system number of the destination, either origin or peer
  148. #17- C 44 src_mask Source address prefix mask bits
  149. #18- C 45 dst_mask Destination address prefix mask bits
  150. #19- n 46-47 pad2 Unused (zero) bytes
  151. for (my $i = 0; $i < $count; $i++) {
  152. my $flowrec = substr($datagram, $netflow5_header_len + ($i*$netflow5_flowrec_len), $netflow5_flowrec_len);
  153. my @flowdata = unpack("NNNnnNNNNnnCCCCnnCCn", $flowrec);
  154. my %flow;
  155. $flow{src_ip} = join '.', unpack 'C4', pack 'N', $flowdata[0];
  156. $flow{dst_ip} = join '.', unpack 'C4', pack 'N', $flowdata[1];
  157. $flow{snmp_in} = $flowdata[3] || 0;
  158. $flow{snmp_out} = $flowdata[4] || 0;
  159. $flow{pkts} = $flowdata[5] || 0;
  160. $flow{octets} = $flowdata[6] || 0;
  161. $flow{src_port} = $flowdata[9] || 0;
  162. $flow{dst_port} = $flowdata[10] || 0;
  163. $flow{proto} = $flowdata[13] || 0;
  164. $flow{xsrc_ip} = $flow{src_ip};
  165. $flow{xdst_ip} = $flow{dst_ip};
  166. $flow{starttime} = time();
  167. $flow{netflow_v} = '5';
  168. $flow{ipv} = '4';
  169. save_flow($ipaddr, \%flow);
  170. }
  171. }
  172. sub parse_netflow_v9 {
  173. my $datagram = shift;
  174. my $ipaddr = shift;
  175. # Parse packet
  176. my ($version, $count, $sysuptime, $unix_secs, $seqno, $source_id, @flowsets) = unpack("nnNNNN(nnX4/a)*", $datagram);
  177. # Loop through FlowSets and take appropriate action
  178. for (my $i = 0; $i < scalar @flowsets; $i += 2) {
  179. my $flowsetid = $flowsets[$i];
  180. my $flowsetdata = substr($flowsets[$i+1], 4); # chop off id/length
  181. if ($flowsetid == 0) {
  182. # 0 = Template FlowSet
  183. parse_netflow_v9_template_flowset($flowsetdata, $ipaddr, $source_id);
  184. } elsif ($flowsetid == 1) {
  185. # 1 - Options Template FlowSet
  186. } elsif ($flowsetid > 255) {
  187. # > 255: Data FlowSet
  188. parse_netflow_v9_data_flowset($flowsetid, $flowsetdata, $ipaddr, $source_id);
  189. } else {
  190. # reserved FlowSet
  191. print "Unknown FlowSet ID $flowsetid found\n";
  192. }
  193. }
  194. }
  195. sub parse_netflow_v9_template_flowset {
  196. my $templatedata = shift;
  197. my $ipaddr = shift;
  198. my $source_id = shift;
  199. # Note: there may be multiple templates in a Template FlowSet
  200. my @template_ints = unpack("n*", $templatedata);
  201. my $i = 0;
  202. while ($i < scalar @template_ints) {
  203. my $template_id = $template_ints[$i];
  204. my $fldcount = $template_ints[$i+1];
  205. last if (!defined($template_id) || !defined($fldcount));
  206. # print "Updated template ID $template_id (source ID $source_id, from " . inet_ntoa($ipaddr) . ")\n" if ($debug);
  207. my $template = [@template_ints[($i+2) .. ($i+2+$fldcount*2-1)]];
  208. $netflow9_templates->{$ipaddr}->{$source_id}->{$template_id}->{'template'} = $template;
  209. # Calculate total length of template data
  210. my $totallen = 0;
  211. for (my $j = 1; $j < scalar @$template; $j += 2) {
  212. $totallen += $template->[$j];
  213. }
  214. $netflow9_templates->{$ipaddr}->{$source_id}->{$template_id}->{'len'} = $totallen;
  215. $i += (2 + $fldcount*2);
  216. }
  217. }
  218. sub parse_netflow_v9_data_flowset {
  219. my $flowsetid = shift;
  220. my $flowsetdata = shift;
  221. my $ipaddr = shift;
  222. my $source_id = shift;
  223. my $template = $netflow9_templates->{$ipaddr}->{$source_id}->{$flowsetid}->{'template'};
  224. if (!defined($template)) {
  225. # print "Template ID $flowsetid from $source_id/" . inet_ntoa($ipaddr) . " does not (yet) exist\n" if ($debug);
  226. return;
  227. }
  228. # Flowset record types
  229. #define NF9_IN_BYTES 1
  230. #define NF9_IN_PACKETS 2
  231. #define NF9_IN_PROTOCOL 4
  232. #define NF9_L4_SRC_PORT 7
  233. #define NF9_IPV4_SRC_ADDR 8
  234. #define NF9_INPUT_SNMP 10
  235. #define NF9_L4_DST_PORT 11
  236. #define NF9_IPV4_DST_ADDR 12
  237. #define NF9_OUTPUT_SNMP 14
  238. #define NF9_OUT_BYTES 23
  239. #define NF9_OUT_PKTS 24
  240. #define NF9_DIRECTION 61
  241. #define NF_F_XLATE_SRC_ADDR_IPV4 225
  242. #define NF_F_XLATE_DST_ADDR_IPV4 226
  243. #define NF_F_XLATE_SRC_PORT 227
  244. #define NF_F_XLATE_DST_PORT 228
  245. #define NF9_IPV6_SRC_ADDR 27
  246. #define NF9_IPV6_DST_ADDR 28
  247. #define NF_F_XLATE_SRC_ADDR_IPV6 281
  248. #define NF_F_XLATE_DST_ADDR_IPV6 282
  249. my $len = $netflow9_templates->{$ipaddr}->{$source_id}->{$flowsetid}->{'len'};
  250. my $offset = 0;
  251. my $datalen = length($flowsetdata);
  252. while (($offset + $len) <= $datalen) {
  253. my %flow;
  254. $flow{netflow_v} = '9';
  255. $flow{ipv} = '4';
  256. $flow{starttime} = time();
  257. for (my $i = 0; $i < scalar @$template; $i += 2) {
  258. my $field_type = $template->[$i];
  259. my $field_length = $template->[$i+1];
  260. my $value = substr($flowsetdata, $offset, $field_length);
  261. $offset += $field_length;
  262. # IN_BYTES
  263. if ($field_type == 1) {
  264. if ($field_length == 4) {
  265. $flow{octets} = unpack("N", $value);
  266. } elsif ($field_length == 8) {
  267. $flow{octets} = unpack("Q>", $value);
  268. }
  269. }
  270. # IN_PACKETS
  271. elsif ($field_type == 2) {
  272. if ($field_length == 4) {
  273. $flow{pkts} = unpack("N", $value);
  274. } elsif ($field_length == 8) {
  275. $flow{pkts} = unpack("Q>", $value);
  276. }
  277. }
  278. # IN_PROTOCOL
  279. elsif ($field_type == 4) { $flow{proto} = unpack("C", $value); }
  280. # L4_SRC_PORT
  281. elsif ($field_type == 7) { $flow{src_port} = unpack("n", $value); }
  282. # IPV4_SRC_ADDR
  283. elsif ($field_type == 8) { $flow{src_ip} = inet_ntop(AF_INET, $value); }
  284. # INPUT_SNMP
  285. elsif ($field_type == 10) {
  286. if ($field_length == 2) {
  287. $flow{snmp_in} = unpack("n", $value);
  288. } elsif ($field_length == 4) {
  289. $flow{snmp_in} = unpack("N", $value);
  290. }
  291. }
  292. # L4_DST_PORT
  293. elsif ($field_type == 11) { $flow{dst_port} = unpack("n", $value); }
  294. # IPV4_DST_ADDR
  295. elsif ($field_type == 12) { $flow{dst_ip} = inet_ntop(AF_INET, $value); }
  296. # OUTPUT_SNMP
  297. elsif ($field_type == 14) {
  298. if ($field_length == 2) {
  299. $flow{snmp_out} = unpack("n", $value);
  300. } elsif ($field_length == 4) {
  301. $flow{snmp_out} = unpack("N", $value);
  302. }
  303. }
  304. # IP_PROTOCOL_VERSION
  305. elsif ($field_type == 60) { my $ipversion = unpack("C", $value);
  306. #skip ipv6
  307. if ($ipversion == 6) { %flow=(); last; }
  308. }
  309. # XLATE_SRC_ADDR_IPV4
  310. elsif ($field_type == 225) { $flow{xsrc_ip} = inet_ntop(AF_INET, $value); }
  311. # XLATE_DST_ADDR_IPV4
  312. elsif ($field_type == 226) { $flow{xdst_ip} = inet_ntop(AF_INET, $value); }
  313. }
  314. $flow{snmp_in} = 0 if (!$flow{snmp_in});
  315. $flow{snmp_out} = 0 if (!$flow{snmp_out});
  316. $flow{octets} = 0 if (!$flow{octets});
  317. $flow{pkts} = 0 if (!$flow{pkts});
  318. if (%flow and $flow{snmp_in} and $flow{snmp_out}) { save_flow($ipaddr, \%flow); }
  319. }
  320. }
  321. sub save_flow {
  322. my $router_ip = shift;
  323. my $flow = shift;
  324. $router_ip = inet_ntoa($router_ip);
  325. #direction for user, 0 - in, 1 - out
  326. $flow->{direction} = '0';
  327. my $router_id;
  328. #skip unknown router
  329. if (exists $router_svi{$router_ip}) {
  330. $router_id = $router_svi{$router_ip}{id};
  331. $flow->{router_ip} = $router_ip;
  332. $flow->{device_id} = $router_id;
  333. $flow->{save} = $router_svi{$router_ip}{save};
  334. } else { return; }
  335. #skip input traffic for router
  336. if (exists $wan_dev{$router_id}->{$flow->{snmp_out}} and exists $wan_dev{$router_id}->{$flow->{snmp_in}}) { return; }
  337. #skip local traffic for router
  338. if (!exists $wan_dev{$router_id}->{$flow->{snmp_out}} and ! exists $wan_dev{$router_id}->{$flow->{snmp_in}}) { return; }
  339. if (exists $wan_dev{$router_id}->{$flow->{snmp_out}}) { $flow->{direction} = 1; }
  340. # print Dumper($flow) if ($debug);
  341. push(@traffic,$flow);
  342. flush_traffic(0);
  343. }
  344. sub flush_traffic {
  345. my $force = shift || 0;
  346. if (!$force && ($saving || ((time - $traf_lastflush) < $timeshift))) { return; }
  347. $saving++;
  348. my $pid = fork();
  349. if (!defined $pid) {
  350. $saving = 0;
  351. print "cannot fork! Save traffic and exit...\n";
  352. } elsif ($pid != 0) {
  353. # in parent
  354. $traf_lastflush = time();
  355. #clean main cache
  356. @traffic = ();
  357. return;
  358. }
  359. #create oper-cache
  360. my @flush_table = ();
  361. push(@flush_table,@traffic);
  362. #clean main cache
  363. INIT();
  364. my $hdb=init_db();
  365. #saved packet by users
  366. my @detail_traffic = ();
  367. my %saved_netflow = ();
  368. my %routers_found;
  369. #last packet timestamp
  370. my $last_time = time();
  371. foreach my $traf_record (@flush_table) {
  372. my ($auth_id,$l_src_ip,$l_dst_ip,$user_ip,$router_id);
  373. $router_id = $traf_record->{device_id};
  374. if ($traf_record->{save}) {
  375. push(@{$saved_netflow{$traf_record->{device_id}}},join(';',$traf_record->{starttime},$traf_record->{proto},$traf_record->{snmp_in},$traf_record->{snmp_out},$traf_record->{src_ip},$traf_record->{dst_ip},$traf_record->{xsrc_ip},$traf_record->{xdst_ip},$traf_record->{src_port},$traf_record->{dst_port},$traf_record->{octets},$traf_record->{pkts}));
  376. }
  377. $routers_found{$router_id} = 1;
  378. #outbound traffic
  379. if ($traf_record->{direction}) {
  380. if (exists $user_stats{$traf_record->{src_ip}}) {
  381. $user_ip = $traf_record->{src_ip};
  382. $l_src_ip = $traf_record->{src_ip};
  383. $l_dst_ip = $traf_record->{dst_ip};
  384. if (exists $user_stats{$user_ip}{$router_id}{out}) {
  385. $user_stats{$user_ip}{$router_id}{out}+=$traf_record->{octets};
  386. } else {
  387. $user_stats{$user_ip}{$router_id}{out}=$traf_record->{octets};
  388. }
  389. if (exists $user_stats{$user_ip}{$router_id}{pkt_out}) {
  390. $user_stats{$user_ip}{$router_id}{pkt_out}+=$traf_record->{pkts};
  391. } else {
  392. $user_stats{$user_ip}{$router_id}{pkt_out}=$traf_record->{pkts};
  393. }
  394. }
  395. if (!$user_ip and $config_ref{add_unknown_user}) {
  396. $user_ip = $traf_record->{src_ip};
  397. $auth_id = new_auth($hdb,$user_ip);
  398. $user_stats{$user_ip}{auth_id}=$auth_id;
  399. $user_stats{$user_ip}{$router_id}{in}=0;
  400. $user_stats{$user_ip}{$router_id}{out}=$traf_record->{octets};
  401. $user_stats{$user_ip}{$router_id}{pkt_in}=0;
  402. $user_stats{$user_ip}{$router_id}{pkt_out}=$traf_record->{pkts};
  403. $user_stats{$user_ip}{save_traf}=$config_ref{save_detail};
  404. }
  405. #inbound traffic
  406. } else {
  407. if (exists $user_stats{$traf_record->{xdst_ip}}) {
  408. $user_ip = $traf_record->{xdst_ip};
  409. $l_src_ip = $traf_record->{src_ip};
  410. $l_dst_ip = $traf_record->{xdst_ip};
  411. if (exists $user_stats{$user_ip}{$router_id}{in}) {
  412. $user_stats{$user_ip}{$router_id}{in}+=$traf_record->{octets};
  413. } else {
  414. $user_stats{$user_ip}{$router_id}{in}=$traf_record->{octets};
  415. }
  416. if (exists $user_stats{$user_ip}{$router_id}{pkt_in}) {
  417. $user_stats{$user_ip}{$router_id}{pkt_in}+=$traf_record->{pkts};
  418. } else {
  419. $user_stats{$user_ip}{$router_id}{pkt_in}=$traf_record->{pkts};
  420. }
  421. }
  422. }
  423. next if (!$user_ip);
  424. $last_time = $traf_record->{starttime};
  425. $user_stats{$user_ip}{last_found} = $last_time;
  426. next if (!$config_ref{save_detail} and !$user_stats{$user_ip}{save_traf});
  427. my $l_src_ip_aton=StrToIp($l_src_ip);
  428. my $l_dst_ip_aton=StrToIp($l_dst_ip);
  429. my ($sec,$min,$hour,$day,$month,$year,$zone) = (localtime($last_time))[0,1,2,3,4,5];
  430. $month++;
  431. $year += 1900;
  432. my $full_time = sprintf "%04d-%02d-%02d %02d:%02d:%02d",$year,$month,$day,$hour,$min,$sec;
  433. my @detail_array = ($user_stats{$user_ip}->{auth_id},$router_id,$full_time,$traf_record->{proto},$l_src_ip_aton,$l_dst_ip_aton,$traf_record->{src_port},$traf_record->{dst_port},$traf_record->{octets},$traf_record->{pkts});
  434. push(@detail_traffic,\@detail_array);
  435. }
  436. @flush_table=();
  437. #start hour
  438. my ($sec,$min,$hour,$day,$month,$year) = (localtime($last_time))[0,1,2,3,4,5];
  439. #save netflow
  440. $save_path=~s/\/$//;
  441. foreach my $dev_id (keys %saved_netflow) {
  442. my $netflow_file_path = $save_path.'/'.$dev_id.'/'.sprintf "%04d/%02d/%02d/%02d/",$year+1900,$month+1,$day,$hour;
  443. my $nmin = int($min/10)*10;
  444. my $netflow_file_name = $netflow_file_path.sprintf "%04d%02d%02d-%02d%02d.csv",$year+1900,$month+1,$day,$hour,$nmin;
  445. if ($saved_netflow{$dev_id} and scalar @{$saved_netflow{$dev_id}}) {
  446. use File::Path;
  447. File::Path::make_path($netflow_file_path);
  448. if ( -e $netflow_file_name) {
  449. open (ND,">>$netflow_file_name") || die("Error open file $netflow_file_name!!! die...");
  450. binmode(ND,':utf8');
  451. } else {
  452. open (ND,">$netflow_file_name") || die("Error open file $netflow_file_name!!! die...");
  453. binmode(ND,':utf8');
  454. print ND join(';',"time","proto","snmp_in","snmp_out","src_ip","dst_ip","xsrc_ip","xdst_ip","src_port","dst_port","octets","pkts")."\n";
  455. }
  456. foreach my $row (@{$saved_netflow{$dev_id}}) {
  457. next if (!$row);
  458. print ND $row."\n";
  459. }
  460. close ND;
  461. @{$saved_netflow{$dev_id}}=();
  462. }
  463. }
  464. undef %saved_netflow;
  465. #save statistics
  466. #start stat time
  467. my $hour_date1 = $hdb->quote(sprintf "%04d-%02d-%02d %02d:00:00",$year+1900,$month+1,$day,$hour);
  468. #end hour
  469. ($hour,$day,$month,$year) = (localtime($last_time+3600))[2,3,4,5];
  470. my $hour_date2 = $hdb->quote(sprintf "%04d-%02d-%02d %02d:00:00",$year+1900,$month+1,$day,$hour);
  471. my @batch_sql_traf=();
  472. #print Dumper(\%user_stats) if ($debug);
  473. # update database
  474. foreach my $user_ip (keys %user_stats) {
  475. next if (!exists $user_stats{$user_ip}{last_found});
  476. my $user_ip_aton=StrToIp($user_ip);
  477. my $auth_id = $user_stats{$user_ip}{auth_id};
  478. #last flow for user
  479. my ($sec,$min,$hour,$day,$month,$year) = (localtime($user_stats{$user_ip}{last_found}))[0,1,2,3,4,5];
  480. #flow time string
  481. my $flow_date = $hdb->quote(sprintf "%04d-%02d-%02d %02d:%02d:%02d",$year+1900,$month+1,$day,$hour,$min,$sec);
  482. #last found timestamp
  483. my $tSQL="UPDATE User_auth SET `last_found`=$flow_date WHERE id='$auth_id'";
  484. push (@batch_sql_traf,$tSQL);
  485. #per router stats
  486. foreach my $router_id (keys %routers_found) {
  487. next if (!exists $user_stats{$user_ip}{$router_id});
  488. if (!exists $user_stats{$user_ip}{$router_id}{in}) { $user_stats{$user_ip}{$router_id}{in} = 0; }
  489. if (!exists $user_stats{$user_ip}{$router_id}{out}) { $user_stats{$user_ip}{$router_id}{out} = 0; }
  490. #skip empty stats
  491. if ($user_stats{$user_ip}{$router_id}{in} + $user_stats{$user_ip}{$router_id}{out} ==0) { next; }
  492. #packet count per router
  493. if (!exists $user_stats{$user_ip}{$router_id}{pkt_in}) { $user_stats{$user_ip}{$router_id}{pkt_in} = 0; }
  494. if (!exists $user_stats{$user_ip}{$router_id}{pkt_out}) { $user_stats{$user_ip}{$router_id}{pkt_out} = 0; }
  495. #current stats
  496. my $tSQL="INSERT INTO User_stats_full (timestamp,auth_id,router_id,byte_in,byte_out,pkt_in,pkt_out,step) VALUES($flow_date,'$auth_id','$router_id','$user_stats{$user_ip}{$router_id}{in}','$user_stats{$user_ip}{$router_id}{out}','$user_stats{$user_ip}{$router_id}{pkt_in}','$user_stats{$user_ip}{$router_id}{pkt_out}','$timeshift')";
  497. push (@batch_sql_traf,$tSQL);
  498. #hour stats
  499. # get current stats
  500. my $sql = "SELECT id, byte_in, byte_out FROM User_stats WHERE `timestamp`>=$hour_date1 AND `timestamp`<$hour_date2 AND router_id=$router_id AND auth_id=$auth_id";
  501. my $hour_stat = get_record_sql($hdb,$sql);
  502. if (!$hour_stat) {
  503. my $dSQL="INSERT INTO User_stats (timestamp,auth_id,router_id,byte_in,byte_out) VALUES($flow_date,'$auth_id','$router_id','$user_stats{$user_ip}{$router_id}{in}','$user_stats{$user_ip}{$router_id}{out}')";
  504. push (@batch_sql_traf,$dSQL);
  505. next;
  506. }
  507. if (!$hour_stat->{byte_in}) { $hour_stat->{byte_in}=0; }
  508. if (!$hour_stat->{byte_out}) { $hour_stat->{byte_out}=0; }
  509. $hour_stat->{byte_in} += $user_stats{$user_ip}{$router_id}{in};
  510. $hour_stat->{byte_out} += $user_stats{$user_ip}{$router_id}{out};
  511. $tSQL="UPDATE User_stats SET byte_in='".$hour_stat->{byte_in}."', byte_out='".$hour_stat->{byte_out}."' WHERE id='".$auth_id."' AND router_id='".$router_id."'";
  512. push (@batch_sql_traf,$tSQL);
  513. }
  514. }
  515. #update statistics in DB
  516. batch_db_sql($hdb,\@batch_sql_traf);
  517. db_log_debug($hdb,"Recalc quotes started");
  518. foreach my $router_id (keys %routers_found) { recalc_quotes($hdb,$router_id); }
  519. db_log_debug($hdb,"Recalc quotes stopped");
  520. if (scalar(@detail_traffic)) {
  521. db_log_debug($hdb,"Start write traffic detail to DB. ".scalar @detail_traffic." lines count") if ($debug);
  522. if ($config_ref{DBTYPE} eq 'mysql') {
  523. batch_db_sql_csv("Traffic_detail", \@detail_traffic);
  524. } else {
  525. my $index = 0;
  526. my @tmp=();
  527. my $item_per_thread = int(scalar @detail_traffic / $thread_count);
  528. my @threads=();
  529. foreach my $row (@detail_traffic) {
  530. push(@tmp,$row);
  531. $index++;
  532. if ($index<=$item_per_thread) { next; }
  533. my @tmp1=();
  534. push(@tmp1,@tmp);
  535. @tmp=();
  536. push(@threads, threads->create(\&batch_db_sql_csv, "Traffic_detail", \@tmp1));
  537. }
  538. if (scalar(@tmp)) {
  539. push(@threads, threads->create(\&batch_db_sql_csv, "Traffic_detail", \@tmp));
  540. }
  541. foreach my $t (@threads) { $t->join(); }
  542. @tmp=();
  543. }
  544. @detail_traffic = ();
  545. db_log_debug($hdb,"Write traffic detail to DB stopped") if ($debug);
  546. }
  547. $hdb->disconnect();
  548. $saving = 0;
  549. exit;
  550. }