用perl对CDN节点日志进行统计

前段时间公司运维部门要求我对CDN节点的日志进行分析，具体要求如下

1、输入参数为文件名，支持统配。文件为gz压缩，可用zcat解压。统配的多个文件，
必须是一天内的。
2、文件内为一条条的访问日志。格式如下：
2011-01-02 15:55:01 122.245.127.73
"/down.eebbk.net/xzzx/h1sp/\xb8\xdf\xd6\xd0\xc9\xfa\xce\xef\xb1\xd8\xd0\xde2
\xc8\xbe\xc9\xab\xcc\xe5\xb1\xe4\xd2\xec(\xb6\xfe).avi" 206 11568567 3316812
"
&dhbig=\xbb\xc6\xb8\xd4\xbf\xce\xcc\xc3&dhsmall=\xb8\xdf\xd6\xd0\xb1\xd8\xd0
\xde2&dhtiny=\xc9\xfa\xce\xef&title=\xca\xd3\xc6\xb5\xd1\xa7\xcf\xb0xbb\xfa
H1&mode=6846" "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET
CLR 1.1.4322; .NET CLR 2.0.50727)"
3、该条日志中，2011-01-02 15:55:01 为日志时间点。其基准时间点的计算方法为，
将该时间点转换为标准的秒，再模300，得到基准时间点。如15:55。再比如，
18:23:08，其基准时间点位18:20。
4、该条日志中，表示应用层在3316812 毫秒，输出了11568567 字节。该信息折算到日
志统计上，是按如下处理：
--3316812 毫秒为 3316812/300000+1 = 11+1 = 12个日志计费点。（每个日志计费点
代表一个5分钟的区间）
--11568567字节平均分到12个日志计费点上，为该记录在该计费点上的带宽占用。如
11568567/12/300=3213bps
5、则该日志的流量将给15:00、15:05、15:10、15:15、15:20、15:25、15:30、
15:35、15:40、15:45、15:50、15:55这12个基准时间点，每个增加3123bps的值。
6、要求统计所有指定日志文件综合的各基准时间点的带宽
7、将各基准时间点的带宽值以如下格式输出显示：
time     bps
00:00   1903342
00:05   1833133
....
23:55   2312342
Total Traffic
 
8、由于该程序将是手动运行在Apache服务器上的。所以程序要轻量级，同时不用占用
太多的磁盘。或许比较理想的是直接处理zcat的输出。（服务器上没有php）

我花了5个小时，先写出了一个单线程的程序，后来又改为多线程，感觉用perl写起来比较快，之间学会了用parsewords进行解析日志，parsewords太好用了。

执行命令：perl analy_log.pl CT-ZHZ-1-N004-A-bbg04_2011010414*

测试结果，分析几十M的压缩日志，才2秒多

 

代码如下：

 

use Text::ParseWords;
use threads;
use threads::shared;
use Time::Local;
#print @ARGV;

#分析结果集

my %result:shared;
#最大线程数

my $max_thread = 2;
#线程池

my @thread_array;
my $current_thread = 0;

#检查参数

$argv_len = @ARGV;
if($argv_len == 0)
{
    print "it need filename\n";
    exit(1);
}

#处理参数开始

$cmd="ls '".join("' '",@ARGV)."' |";
#print $cmd."\n";


#处理参数结束


open(PIPE, $cmd);
@filenames = <PIPE>;
close(PIPE);
#print "[";

#print @filenames;

#print "]\n";


#多线程分析多个日志文件

foreach(@filenames){
    if( $current_thread >= $max_thread )
  {
    foreach my $thread( @thread_array )
    {
      $thread -> join( );
    }
    $current_thread = 0;
    @thread_array=();
  }
  $thread_array[$current_thread] = threads -> new(\&analy,$_);
  $current_thread ++;
}

#等待线程结束

foreach my $thread( @thread_array )
{
   $thread -> join();
}

#按时间排序输出

my @key=sort(keys(%result));
print "time\t\t\tbps\n";
foreach (@key){
        $k=$_;
        print $k."\t\t".$result{$k}."\n";
}

#去除前后空白字符

sub trim
{
        my $string = shift;
        $string =~ s/^\s+//;
        $string =~ s/\s+$//;
        return $string;
}

#分析一个日志文件

sub analy
{
    my $filename= trim(shift);
    $cmd="zcat $filename |";
    open(PIPEFILE,$cmd);
    @lines=<PIPEFILE>;
    close(PIPEFILE);
    foreach(@lines)
    {
        #分解每一行

        my @line=quotewords(" ", 1, $_);
        #print join(",",@line)."\n";

        #print $line[6]."\n";


        #根据时间合并

        my($h,$m)=split(/:/g,$line[1]);
        my($y,$mon,$d)=split(/-/g,$line[0]);
        $m=int($m/5)*5;
        my $point=int($line[6]/300000+1);
        my $bps=int($line[5]/$point/300);
     #print $line[0].$line[1]." m=$m,point=$point,bps=$bps\n";

        $endtime=timelocal(0,$m,$h,$d,$mon,$y-1900);#秒，分，时，日，月，年（year-1900）

        my @time_list=get_time_list($endtime,$point);
        #累计字节数

        my $i=0;
        for($i=0;$i<$point;$i++)
        {
            $result{$time_list[$i]}+=$bps;
        }
    }
}
#获得时间列表

sub get_time_list
{
    my $endtime=trim(shift);
    my $point=trim(shift);
    my @time_list;
    my $i=0;
    for($i=0;$i<$point;$i++)
    {
        push(@time_list,get_time_str($endtime));
        $endtime-=300;
    }
    @time_list;
}
#获得时间字符串

sub get_time_str
{
    my $endtime=trim(shift);
    #@t=gmtime($endtime);

    ($sec, $min, $hour, $day, $mon, $year, $wday, $yday, $isdst) = localtime($endtime);
    if($min<10)
    {
        $min="0$min";
    }
    if($hour<10)
    {
        $hour="0$hour";
    }
    if($day<10)
    {
        $day="0$day";
    }
    if($mon<10)
    {
        $mon="0$mon";
    }
    $year+=1900;
    my $str="$year-$mon-$day_$hour:$min";
}