用 Amcharts 和 ElasticSearch 做日志分析

    package AnalysisDemo;
    use Dancer ':syntax';
    use Dancer::Plugin::Ajax;
    use ElasticSearch;
    use POSIX qw(strftime);
    no  warnings;
    
    my $elsearch         = ElasticSearch->new( { %{ config->{plugins}->{ElasticSearch} } } );
    my $index_prefix     = 'demo-';
    my $type             = 'log';
    # 这里是对ip库的归类。数据是需要提前导入ES的，这可以是logstash发挥作用
    my $default_provider = {
        yidong    => [qw(CN-CRN CN-CMN)],
        jiaoyu    => [qw(CN-CER CN-CST)],
        dianxin   => [qw(CN-CHN)],
        liantong  => [qw(CN-UNI CN-CNC)],
        guangdian => [qw(CN-SCN)],
        haiwai => [qw(OS)],
    };
    
    get '/' => sub {
        # 通过 state API 获取 ES 集群现有的所有index列表
        # 因为是一个域名一个index，这样就有了前段页面上的域名下拉选择框
        my $indices = $elsearch->cluster_state->{routing_table}->{indices};
        template 'demo/chart',
          {
            providers => [ sort keys %$default_provider ],
            datasources =>
              [ grep { /^$index_prefix/ && s/$index_prefix// } keys %$indices ],
            inputfrom => strftime("%F\T%T", localtime(time()-864000)),
            inputto => strftime("%F\T%T", localtime()),
          };
    };
    
    # 这里把 api 拆成服务商和区域两个，没啥特殊原因，因为是分两回写的，汗
    # 其实可以看到最开始的请求参数类似，最后json的field名字都一样
    ajax '/api/provider' => sub {
        my $param = from_json(request->body);
        my $index = $index_prefix . $param->{'datasource'};
        my $from  = $param->{'from'} || 'now-10d';
        my $to    = $param->{'to'} || 'now';
        my $providers = $param->{'provider'};
        my ( $pct, $chartData );
        for my $provider ( sort @{$providers} ) {
            my $provider_pct;
            # 这里是比较麻烦的一点，因为一个区域在ip库里可能标记成多个，比如铁通和移动，现在都是移动
            for my $area ( @{ $default_provider->{$provider} } ) {
                my $res = pct_count( $index, $area, $from, $to );
                for my $time ( sort keys %{$res} ) {
                    $provider_pct->{$time}->{count} += $res->{$time}->{count};
                    $provider_pct->{$time}->{error} += $res->{$time}->{error};
                    $provider_pct->{$time}->{slow}  += $res->{$time}->{slow};
                }
            }
            # 这里因为可能没有错误，所以前面关闭了常用的 warnings 警告
            for my $time ( sort keys %{$provider_pct} ) {
                my $right_pct = 100;
                $right_pct =
                  100 -
                  $provider_pct->{$time}->{slow} / $provider_pct->{$time}->{count}
                  * 100;
                $pct->{$time}->{$provider} = sprintf "%.2f", $right_pct;
                $pct->{$time}->{"${provider}Err"} = sprintf "%.2f",
                  $provider_pct->{$time}->{error} / $provider_pct->{$time}->{count}
                  * 100;
                $pct->{$time}->{"${provider}Size"} = sprintf "%.0f",
                  $pct->{$time}->{"${provider}Err"};
            }
        };
    
        for my $time ( sort keys %$pct ) {
            my $data->{date} = $time;
            for my $provider ( @$providers ) {
                $data->{$provider} = $pct->{$time}->{$provider} || 100;
                $data->{"${provider}Err"} = $pct->{$time}->{"${provider}Err"} || 0;
                # 百分比太低，所以翻 5 倍来作为 bullet 的大小
                $data->{"${provider}Size"} =
                  $pct->{$time}->{"${provider}Size"} * 5 || 0;
            };
            push @$chartData, $data;
        };
    
        my $res = {
            type => "line",
            categoryField => "date",
            graphList => $providers,
            chartData => $chartData,
        };
    
        return to_json($res);
    };
    
    ajax '/api/area' => sub {
        my $param = from_json(request->body);
        my $index = $index_prefix . $param->{'datasource'};
        my $limit = $param->{'limit'} || 50;
        my $from  = $param->{'from'} || 'now-10d';
        my $to    = $param->{'to'} || 'now';
        # 这是后来写的，尽可能把 sub 拆分了，所以 ajax 这里就很简略
        # 当然因为不考虑多运营商的问题，本身也容易一些
        my $res = pct_terms( $index, $limit, $from, $to );
        return to_json($res);
    };
    
    sub pct_terms {
        my ( $index, $limit, $from, $to ) = @_;
        my $area_all_count = area_terms( $index, 0,    $limit, $from, $to );
        my $area_err_count = area_terms( $index, 2000, $limit, $from, $to );
        my ( $error, $chartData );
        for ( @{$area_err_count} ) {
            $error->{ $_->{term} } = $_->{count};
        }
        for ( @{$area_all_count} ) {
            push @$chartData, {
                area  => $_->{term},
                error => $error->{ $_->{term} } || 0,
                right => $_->{count} - $error->{ $_->{term} },
            };
        }
        my $res = {
            type => "column",
            categoryField => "area",
            graphList => [qw(right error)],
            chartData => $chartData,
        };
        return $res;
    }
    
    sub pct_count {
        my ( $index, $area, $from, $to ) = @_;
        my $level = $area eq 'OS' ? 3000 : 2000;
        my $all_count  = histo_count( $index, 0,      $area, $from, $to );
        my $slow_count = histo_count( $index, $level,   $area, $from, $to );
        my $err_count  = histo_count( $index, 'hasErr', $area, $from, $to );
        my $res;
        for ( @{$slow_count} ) {
            $res->{ $_->{time} }->{slow} = $_->{count};
        }
        for ( @{$err_count} ) {
            $res->{ $_->{time} }->{error} = $_->{count};
        }
        for ( @{$all_count} ) {
            $res->{ $_->{time} }->{count} = $_->{count};
        }
        return $res;
    }
    
    # 下面开始的两个才是真正发 ES 请求的地方

    sub area_terms {
        my ( $index, $level, $limit, $from, $to ) = @_;
        my $data = $elsearch->search(
            index  => $index,
            type   => $type,
            size   => 0,
            facets => {
                area => {
                    facet_filter => {
                        and => [
                            {
                                range => {
                                    date => {
                                        from => $from,
                                        to   => $to
                                    },
                                },
                            },
                            {
                                numeric_range =>
                                  { timeCost => { gte => $level, }, },
                            },
                        ],
                    },
                    # 使用最简单的 terms facets API，因为只用计数就好了
                    terms => {
                        field => "fromArea",
                        size  => $limit,
                    }
                }
            }
        );
        return $data->{facets}->{area}->{terms};
    }
    
    sub histo_count {
        my ( $index, $level, $area, $from, $to ) = @_;
        # 根据 level 参数判断使用 hasErr 还是 timeCost 列数据
        my $level_ref =
          $level eq 'hasErr'
          ? { term => { hasErr => 'true' } }
          : { numeric_range => { timeCost => { gt => $level } } };
        my $facets = {
            pct => {
                facet_filter => {
                    # 这里条件比较多，所以要用 bool API，不能用 and 了
                    bool => {
                        # must 可以提供多个条件作为 AND 数组
                        # 此外还有 must_not 作为 AND NOT 数组
                        # should 作为 OR 数组
                        must => [
                            {
                                range => {
                                    date => {
                                        from => $from,
                                        to   => $to
                                    },
                                },
                            },
                            { prefix => { fromArea => $area } },
                            $level_ref,
                        ],
                    },
                },
                # 这里是需要针对专门的时间列做汇总，所以用 date_histogram 了，具体说明之前有博客
                date_histogram => {
                    field    => "date",
                    interval => "1h",
                }
            }
        };
        my $data = $elsearch->search(
            index  => $index,
            type   => $type,
            facets => $facets,
            size   => 0,
        );
        return $data->{facets}->{pct}->{entries};
    }