eye-statd.pl 20 KB

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