zabbix sender и пачка значений.

[Melanxolik]

Интересует вопрос, не могу найти ответа на данный момент, как с помощью zabbix_sender передать пачку значений в том же json.
Основная задумка исключить промежуточный файл, выдавать напрямую сразу после формирования, но делать 100-200 вызовов zabbix_sender не вижу смысла, хочу одним вызовом все значения без промежуточного файла.

[Jimson]

На правах примера

package Jimson::Zabbix::Sender;

# $VCS: perl/trunk/Jimson/Zabbix/Sender.pm, rev 10, 2014-11-14 14:30:04Z, jimson $

our $VERSION = '0.3';

use strict;
use warnings;
use IO::Socket;
use IO::Select;
use JSON;
use Unix::Syslog qw( :subs :macros );
use Jimson::Utils qw( logging writetofile );

our $SendMaximumOfElements = 50;


sub new {
    my ($invocant, %args) = @_;
    my $class = ref $invocant || $invocant;

    my ($trapper) = grep defined, delete @args{qw(Trapper trapper TRAPPER Server server SERVER)}, '127.0.0.1:10051';
    my ($dumpfile) = grep defined, delete @args{qw(DumpFile Dumpfile dumpfile DUMPFILE)}, '/tmp/zabbix_sender_errors';
    my ($timeout) = grep defined, delete @args{qw(Timeout timeout TIMEOUT)}, 10;

    my $self = {
        __trapper   => $trapper,
        __dumpfile  => $dumpfile,
        __timeout   => $timeout,
        __queue     => {},          # очередь данных на отправку, ключ - hostname прокси или 'NULL', значение - ссылка на массив хэшей с данными
        __socket    => undef,       # сокет zabbix trapper, открывается при отправке данных, закрывается после получения ответа
        __select    => undef,
        __json      => undef,       # отправляеммый JSON, скаляр очищается после получаения подтверждения от zabbix trapper
        __response  => '',          # буфер полученных данных от zabbix trapper (анализ подтверждения после отправки данных)
        __waitwhen  => 0,           # таймер-таймаут ожидания ответа от zabbix trapper, так же используется если не удалось открыть TCP соединение
    };
    bless $self, $class;
    return $self;
}


sub queue {
    my $self = shift;
    return unless %{$self->{__queue}} or $self->{__json};
    1;
}

sub working {
    my $self = shift;
    return unless $self->{__socket} or ($self->queue and $self->{__waitwhen} < time);
    1;
}

sub queuelength {
    my $self = shift;
    my $count = 0;
    for ( values $self->{__queue} ) {
        $count += @{$_};
    }
    return $count;
}

sub enqueue {
    my $self = shift;
    my $proxy = ref $_[0] ? 'NULL' : shift;
    for my $data ( @_ ) {
        push @{$self->{__queue}->{$proxy}}, $data if ref $data eq 'HASH';
    }
}

sub send {
    my $self = shift;
    if ( %{$self->{__queue}} and not $self->{__json} ) {
        my $proxy = (keys $self->{__queue})[0];
        if ( $proxy eq 'NULL' ) {
            $self->{__json} = encode_json {
                request => 'sender data',
                data    => [ splice($self->{__queue}->{$proxy}, 0, $SendMaximumOfElements) ]
            };
        } else {
            $self->{__json} = encode_json {
                request => 'history data',
                host    => $proxy,
                data    => [ splice($self->{__queue}->{$proxy}, 0, $SendMaximumOfElements) ]
            };
        }
        delete $self->{__queue}->{$proxy} unless @{$self->{__queue}->{$proxy}};
    }
    if ( $self->{__json} and not $self->{__socket} and $self->{__waitwhen} < time ) {
        eval {
            local $SIG{ALRM} = sub { die 'ZABBIX TIMEOUT' };
            alarm($self->{__timeout});
            $self->{__socket} = IO::Socket::INET->new( PeerAddr => $self->{__trapper}, Proto => 'tcp' );
            alarm(0);
        };
        alarm(0);
        if ( $self->{__socket} ) {
            $self->{__select} = IO::Select->new( $self->{__socket} );
            $self->{__socket}->send( pack('a4 C L L a*', "ZBXD", 1, length($self->{__json}), 0, $self->{__json}) );
            $self->{__waitwhen} = time + $self->{__timeout};
        } else {
            $self->{__waitwhen} = time + $self->{__timeout} * 6;
            logging LOG_CRIT, "Can't connect to Zabbix trapper: %s", $@ =~ /ZABBIX TIMEOUT/ ? "Timeout" : $!;
        }
    }
}

sub listen {
    my ($self, $timeout) = @_;
    $timeout = 0 unless $timeout;
    while ( $self->{__socket} and $self->{__select}->can_read($timeout) ) {
        my $rc = $self->{__socket}->sysread( $self->{__response}, 1024, length($self->{__response}) );
        if ( $rc ) {
            if ( length($self->{__response}) > 13 ) {
                my ($ident, $guard, $size, $high, $msg) = unpack('a4 C L L a*', $self->{__response});
                if ( $ident eq 'ZBXD' and $guard == 1 and $high == 0 and $size > 0 ) {
                    if ( length($self->{__response}) >= $size + 13 ) {
                        my $hash = eval { JSON->new()->decode($msg) };
                        if ( $@ or not $hash or ref $hash ne 'HASH' or not $hash->{response} ) {
                            $self->__dumpResponse('INCORRECT ANSWER');
                            logging LOG_CRIT, "Incorrect response from Zabbix trapper.";
                            $self->__reset(1);
                        } else {
                            if ( $hash->{response} eq 'success' ) {
                                if ( $hash->{info} and $hash->{info} =~ /processed:?\s+(\d+);?\s+failed:?\s+(\d+);?\s+total:?\s+(\d+)/i ) {
                                    $self->__dumpResponse('FAILED') if $2 or $1 != $3;
                                } else {
                                    $self->__dumpResponse('NO INFO IN RESPONSE');
                                }
                            } else {
                                $self->__dumpResponse('REJECT');
                                logging LOG_CRIT, "Zabbix trapper: data rejected.";
                            }
                            $self->{__json} = undef;
                            $self->__reset();
                        }
                    }
                } else {
                    $self->__dumpResponse('INCORRECT ANSWER');
                    logging LOG_CRIT, "Incorrect response from Zabbix trapper.";
                    $self->__reset(1);
                }
            }
        } else {
            logging LOG_ERR, "Zabbix trapper socket error: $!" unless defined $rc;
            $self->__reset(1);
        }
    }
    if ( $self->{__socket} ) {
        if ( $self->{__waitwhen} < time ) {
            $self->__dumpResponse('TIMEOUT');
            logging LOG_CRIT, "Timeout response from Zabbix trapper.";
            $self->__reset();
        }
    }
}

sub __reset {
    my ($self, $pause) = @_;
    $self->{__socket}->close();
    $self->{__socket} = $self->{__select} = undef;
    $self->{__response} = '';
    $self->{__waitwhen} = $pause ? time + $self->{__timeout} : 0;
}

sub __dumpResponse {
    my ($self, $cause) = @_;
    writetofile {
        print "\n", $cause, "\nJSON: ", $self->{__json} || '<empty>', "\nRESPONSE: ", $self->{__response} || '<empty>', "\n";
    } $self->{__dumpfile};
}

1;
__END__

Что бы заработало как есть нужно реализовать logging() и writetofile() и убрать "use Jimson::Utils", ну или просто убрать строчки где используются эти функции. Пользоваться, например, так:

use Jimson::Zabbix:Sender;

Jimson::Zabbix::Sender::SendMaximumOfElements = 100500;

my $sender = Jimson::Zabbix::Sender->new( Trapper => '192.168.1.1:10051', Timeout => 2, DumpFile => '/var/tmp/mysender.errors' );

myfunction() {
    $sender->enqueue($proxy, { host => $host, key => $key, clock => $timestamp, value => $value });
    $sender->enqueue($proxy, { host => $host, key => $key, clock => $timestamp, value => $value });
    $sender->enqueue($proxy, { host => $host, key => $key, clock => $timestamp, value => $value });
}

while ($running) {
    myfunction();
    $sender->send();
    $sender->listen(0.2);
    sleep(1) unless $sender->working();
}

Вообщем оно само кэширует, само отсылает, само следит за сокетом и умеет централизованно (непосредственно на сервер) трапить данные хостов мониторящихся через прокси. Если хост мониторится сервером, то первый аргумент enqueue() должен быть 'NULL' вроде (я хэш host -> proxy вытаскиваю из базы напрямую).

Я это писал для демонов которые самостоятельно трекают данные и отсылают их на сервер.

[Melanxolik]

Спасибо, но вероятно это прийдется отложить, столкнулся с другой проблемой.
А разве графики нельзя построить по элементами полученными через zabbix trapper?
У меня в lld прилетают данные через траппер от iostat, но как оказалось графики не рисуются для трапов.

может у кого-то будут идеи?

[Jimson]

Тип данных какой? Графику совершенно пофигу должно быть на то какой тип проверки, ему важно только что бы данные были числовыми. У меня все строится замечательно, но у меня 2.0.6, возможно в новые версии насовали какие то ограничения? но зачем? смысла нет.

С графиками была другая проблемка (мелкая), дело в том что для данных траппер нельзя задать интервал опроса (в базе это items.delay) как следствие если данные перестают приходить то на графике линия не прерывается а рисуется в виде прямой. Что бы это исправить достаточно просто установить нужное значение для этих элементов в базе:
update items set delay=60 where key_ like '....';
При этом это не нужно делать постоянно, достаточно один раз выставить для вновь созданного элемента данных шаблона или прототипа, затем при создании элемента из шаблона или прототипа, а так же при клонированиях заданный "delay" будет копироваться.

[Melanxolik]

Приношу извинения, проблема оказалась в скрипте, данные просто не поступали и брались последнее значение.
Думаю потихоньку буду разбиратся дальше, сложная система получается.