【Message::Passing系列】客户端收集脚本

    #!/usr/bin/perl
    package NginxLogCollector;
    use Moo;
    use MooX::Options;
    use Message::Passing::DSL;
    use MooX::Types::MooseLike::Base qw/ Str /;
    use namespace::clean -except => [qw( meta _options_data _options_config )];
    with 'Message::Passing::Role::Script';
    option filename => (
        is => 'ro',
        isa => Str,
        default => sub { '/data/nginx/logs/access.log' },
    );
    option rabbitmq => (
        is => 'ro',
        isa => Str,
        default => sub { '10.3.18.199' },
    );
    sub build_chain {
        my $self = shift;
        message_chain {
            output rabbitmq => (
                class => 'AMQP',
                exchange_name => 'logcollect',
    # 目前测试结果，发现Input::AMQP无法接收到非topic的exchange
    # CPAN上有关RabbitMQ的模块都是这个哥们写的，POD简略到没有一样，表示无语下
    #            exchange_type => 'direct',
                hostname => $self->rabbitmq,
                username => 'guest',
                password => 'guest',
            );
            output debug => (
                class => 'STDOUT',
            );
            encoder("encoder",
                class => 'JSON',
                output_to => 'rabbitmq',
                output_to => 'debug',
            );
            filter grok => (
                class => 'GrokLike',
                output_to => 'encoder',
            );
            filter logstash => (
                class => 'ToLogstash',
                output_to => 'grok',
            );
            decoder("decoder",
                class => 'JSON',
                output_to => 'logstash',
            );
            input nginxlog => (
                class => 'FileTail',
                output_to => 'decoder',
                filename => $self->filename,
           );
        };
    }
    __PACKAGE__->start unless caller;
    1;

#!/usr/bin/perl
use Moo;
use MooX::Options;
use Message::Passing::DSL;
use MooX::Types::MooseLike::Base qw/ Str /;
use namespace::clean -except => [qw( meta _options_data _options_config )];

with 'Message::Passing::Role::Script';

option elasticsearch_servers => (
    is => 'ro',
    isa => Str,
    default => sub { '10.3.18.199:9200' },
);

option rabbitmq => (
    is => 'ro',
    isa => Str,
    default => sub { '10.3.18.199' },
);

sub build_chain {
    my $self = shift;
    message_chain {
        output elasticsearch => (
            class => 'ElasticSearch',
            elasticsearch_servers => [$self->elasticsearch_servers],
        );
        decoder decoder => (
            class => 'JSON',
            output_to => 'elasticsearch',
        );
        input rabbitmq => (
            class => 'AMQP',
            exchange_name => 'logcollect',
            queue_name => 'logcollect',
            hostname => $self->rabbitmq,
            username => 'guest',
            password => 'guest',
            output_to => 'decoder',
        );
    };
}
__PACKAGE__->start unless caller;
1;