AnyEvent-ZabbixSender/ZabbixSender.pm

=head1 NAME

AnyEvent::ZabbixSender - simple and efficient zabbix data submission

=head1 SYNOPSIS

   use AnyEvent::ZabbixSender;

=head1 DESCRIPTION

This module is an L<AnyEvent> user, you need to make sure that you use and
run a supported event loop.

I't implements the zabbix version 2.0-3.4 protocol for item data
submission.

=head2 METHODS

=over 4

=cut

package AnyEvent::ZabbixSender;

use common::sense;

use Errno ();
use Scalar::Util ();

use AnyEvent ();
use AnyEvent::Socket ();
use AnyEvent::Handle ();

our $VERSION = '0.1';

=item $zbx = new AnyEvent::ZabbixSender [key => value...]

Creates a (virtual) connection to a zabbix server. Since each submission
requires a new TCP connection, creating the connection object does not
actually contact the server.

The connection object will linger in the destructor until all data has
been submitted or thrown away.

You can specify various configuration parameters. The term C<@items>
refers to an array with C<[key, value, clock]> array-refs.

=over 4

=item server => "$hostname:$port" (default: C<localhost:10051>)

The zabbix server to connect to.

=item host => $name (default: local nodename)

The submission host, the "technical" name from tghe zabbix configuration.

=item delay => $seconds (default: C<0>)

If non-zero, then the module will gather data submissions for up to this
number of seconds before actually submitting them as a single batch.

Submissions can get batched even if C<0>, as events submitted while the
connection is being established or retried will be batched together in any
case.

=item queue_time => $seconds (default: C<3600>)

The amount of time a data item will be queued until it is thrown away when
the server cannot be reached.

=item linger_time => $seconds (default: same as C<queue_time>)

The amount of time the module will linger in its destructor until all
items have been submitted.

=item retry_min => $seconds (default: C<30>)

=item retry_max => $seconds (default: C<300>)

The minimum and maximum retry times when the server cannot be reached.

=item on_error => $cb->(\@items, $msg) (default: log and continue)

Called on any protocol errors - these generally indicate that something
other than a zabbix server is running on a port. The given key-value pairs
are the lost items.

=item on_loss => $cb->(\@items) (default: not called)

Will be called when some data items are thrown away (this happens if the
server isn't reachable for at least C<queue_time> seconds),

=item on_response => $cb->(\@items, \%response) (default: not called)

Will be called with the (generally rather useless) response form the
zabbix server.

=back

=cut

our $NOP = sub { };

my $json = eval { require JSON::XS; JSON::XS->new } || do { require JSON::PP; JSON::PP->new };

$json->utf8;

sub new {
   my $class = shift;
   my $self  = bless {
      server      => "localhost:10051",
      delay       => 0,
      retry_min   => 30,
      retry_max   => 300,
      queue_time  => 3600,
      on_loss     => $NOP,
      on_response => $NOP,
      on_error    => sub {
         AE::log 4 => $_[0]; # error
      },
      @_,
   }, $class;

   ($self->{zhost}, $self->{zport}) = AnyEvent::Socket::parse_hostport $self->{server}, 10051;

   $self->{host} //= do {
      require POSIX;
      (POSIX::uname())[1]
   };

   $self->{linger_time} //= $self->{queue_time};

   $self
}

sub DESTROY {
   my ($self) = @_;

   $self->_wait;

   %$self = ();
}

sub _wait {
   my ($self) = @_;

   return unless @{ $self->{queue} };

   my $cv = AE::cv;

   my $to = AE::timer $self->{linger_time}, 0, $cv;
   local $self->{on_clear} = $cv;

   $cv->recv;
}

=item $zbx->submit ($k, $v[, $clock[, $host]])

Submits a new key-value pair to the zabbix server. If C<$clock> is missing
or C<undef>, then C<AE::now> is used for the event timestamp. If C<$host>
is missing, then the default set during object creation is used.

=item $zbx->submit_multiple ([ [$k, $v, $clock, $host]... ])

Like C<submit>, but submits many key-value pairs at once.

=cut

sub submit_multiple {
   my ($self, $kvcs) = @_;

   push @{ $self->{queue} }, [AE::now, $kvcs];

   $self->_send
      unless $self->{sending};
}

sub submit {
   my ($self, $k, $v, $clock, $host) = @_;

   push @{ $self->{queue} }, [AE::now, [[$k, $v, $clock, $host]]];

   $self->_send;
}

# start sending
sub _send {
   my ($self) = @_;

   if ($self->{delay}) {
      Scalar::Util::weaken $self;
      $self->{delay_w} ||= AE::timer $self->{delay}, 0, sub {
         delete $self->{delay_w};
         $self->{send_immediate} = 1;
         $self->_send2 unless $self->{sending}++;
      };
   } else {
      $self->{send_immediate} = 1;
      $self->_send2 unless $self->{sending}++;
   }
}

# actually do send
sub _send2 {
   my ($self) = @_;

   Scalar::Util::weaken $self;
   $self->{connect_w} = AnyEvent::Socket::tcp_connect $self->{zhost}, $self->{zport}, sub {
      my ($fh) = @_;

      $fh
         or return $self->_retry;
         
      delete $self->{retry};

      my $data = delete $self->{queue};
      my $items = [map @{ $_->[1] }, @$data];

      my $fail = sub {
         $self->{on_error}($items, $_[0]);
         $self->_retry;
      };

      $self->{hdl} = new AnyEvent::Handle
         fh => $fh,
         on_error  => sub {
            $fail->($_[2]);
         },
         on_read   => sub {
            if (13 <= length $_[0]{rbuf}) {
               my ($zbxd, $version, $length) = unpack "a4 C Q<", $_[0]{rbuf};

               $zbxd eq "ZBXD"
                  or return $fail->("protocol mismatch");
               $version == 1
                  or return $fail->("protocol version mismatch");

               if (13 + $length <= length $_[0]{rbuf}) {
                  my $res = eval { $json->decode (substr $_[0]{rbuf}, 13) }
                     or return $fail->("protocol error");
                  use Data::Dump; ddx $res;#d#

                  $self->{on_response}($items, $res);
                  $self->_send2 if delete $self->{send_immediate};

                  $self->{on_clear}() unless @{ $self->{queue} };
               }
            }
         },
      ;

      my $json = $json->encode ({
         request => "sender data",
         clock => AE::now,
         data => [
            map {
               my $slot = $_;

               map {
                  key   => $_->[0],
                  value => $_->[1],
                  clock => $_->[2] // $slot->[0],
                  host  => $_->[3] // $self->{host},
               }, @{ $slot->[1] }
            } @$data
         ],
      });

      warn $json;

      $self->{hdl}->push_write (pack "a4 C Q</a", "ZBXD", 1, $json);
   };
}

sub _retry {
   my ($self) = @_;

   Scalar::Util::weaken $self;

   delete $self->{hdl};

   my $expire = AE::now - $self->{queue_time};
   while (@{ $self->{queue} } && $self->{queue}[0][0] < $expire) {
      $self->{on_loss}([shift @{ $self->{queue} }]);
   }

   unless (@{ $self->{queue} }) {
      $self->{on_clear}();
      return;
   }

   my $retry = $self->{retry_min} * 2 ** $self->{retry}++;
   $retry = $self->{retry_max} if $retry > $self->{retry_max};
   warn "retry $retry\n";#d#
   $self->{retry_w} = AE::timer $retry, 0, sub {
      delete $self->{retry_w};
      $self->_send2;
   };
}

=back

=head1 SEE ALSO

L<AnyEvent>.

=head1 AUTHOR

   Marc Lehmann <schmorp@schmorp.de>
   http://home.schmorp.de/

=cut

1

Revision:	1.1
Committed:	Thu Jan 18 10:17:15 2018 UTC (6 years, 4 months ago) by root
Branch:	MAIN
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::ZabbixSender - simple and efficient zabbix data submission
4
5			=head1 SYNOPSIS
6
7			use AnyEvent::ZabbixSender;
8
9			=head1 DESCRIPTION
10
11			This module is an L<AnyEvent> user, you need to make sure that you use and
12			run a supported event loop.
13
14			I't implements the zabbix version 2.0-3.4 protocol for item data
15			submission.
16
17			=head2 METHODS
18
19			=over 4
20
21			=cut
22
23			package AnyEvent::ZabbixSender;
24
25			use common::sense;
26
27			use Errno ();
28			use Scalar::Util ();
29
30			use AnyEvent ();
31			use AnyEvent::Socket ();
32			use AnyEvent::Handle ();
33
34			our $VERSION = '0.1';
35
36			=item $zbx = new AnyEvent::ZabbixSender [key => value...]
37
38			Creates a (virtual) connection to a zabbix server. Since each submission
39			requires a new TCP connection, creating the connection object does not
40			actually contact the server.
41
42			The connection object will linger in the destructor until all data has
43			been submitted or thrown away.
44
45			You can specify various configuration parameters. The term C<@items>
46			refers to an array with C<[key, value, clock]> array-refs.
47
48			=over 4
49
50			=item server => "$hostname:$port" (default: C<localhost:10051>)
51
52			The zabbix server to connect to.
53
54			=item host => $name (default: local nodename)
55
56			The submission host, the "technical" name from tghe zabbix configuration.
57
58			=item delay => $seconds (default: C<0>)
59
60			If non-zero, then the module will gather data submissions for up to this
61			number of seconds before actually submitting them as a single batch.
62
63			Submissions can get batched even if C<0>, as events submitted while the
64			connection is being established or retried will be batched together in any
65			case.
66
67			=item queue_time => $seconds (default: C<3600>)
68
69			The amount of time a data item will be queued until it is thrown away when
70			the server cannot be reached.
71
72			=item linger_time => $seconds (default: same as C<queue_time>)
73
74			The amount of time the module will linger in its destructor until all
75			items have been submitted.
76
77			=item retry_min => $seconds (default: C<30>)
78
79			=item retry_max => $seconds (default: C<300>)
80
81			The minimum and maximum retry times when the server cannot be reached.
82
83			=item on_error => $cb->(\@items, $msg) (default: log and continue)
84
85			Called on any protocol errors - these generally indicate that something
86			other than a zabbix server is running on a port. The given key-value pairs
87			are the lost items.
88
89			=item on_loss => $cb->(\@items) (default: not called)
90
91			Will be called when some data items are thrown away (this happens if the
92			server isn't reachable for at least C<queue_time> seconds),
93
94			=item on_response => $cb->(\@items, \%response) (default: not called)
95
96			Will be called with the (generally rather useless) response form the
97			zabbix server.
98
99			=back
100
101			=cut
102
103			our $NOP = sub { };
104
105			my $json = eval { require JSON::XS; JSON::XS->new } \|\| do { require JSON::PP; JSON::PP->new };
106
107			$json->utf8;
108
109			sub new {
110			my $class = shift;
111			my $self = bless {
112			server => "localhost:10051",
113			delay => 0,
114			retry_min => 30,
115			retry_max => 300,
116			queue_time => 3600,
117			on_loss => $NOP,
118			on_response => $NOP,
119			on_error => sub {
120			AE::log 4 => $_[0]; # error
121			},
122			@_,
123			}, $class;
124
125			($self->{zhost}, $self->{zport}) = AnyEvent::Socket::parse_hostport $self->{server}, 10051;
126
127			$self->{host} //= do {
128			require POSIX;
129			(POSIX::uname())[1]
130			};
131
132			$self->{linger_time} //= $self->{queue_time};
133
134			$self
135			}
136
137			sub DESTROY {
138			my ($self) = @_;
139
140			$self->_wait;
141
142			%$self = ();
143			}
144
145			sub _wait {
146			my ($self) = @_;
147
148			return unless @{ $self->{queue} };
149
150			my $cv = AE::cv;
151
152			my $to = AE::timer $self->{linger_time}, 0, $cv;
153			local $self->{on_clear} = $cv;
154
155			$cv->recv;
156			}
157
158			=item $zbx->submit ($k, $v[, $clock[, $host]])
159
160			Submits a new key-value pair to the zabbix server. If C<$clock> is missing
161			or C<undef>, then C<AE::now> is used for the event timestamp. If C<$host>
162			is missing, then the default set during object creation is used.
163
164			=item $zbx->submit_multiple ([ [$k, $v, $clock, $host]... ])
165
166			Like C<submit>, but submits many key-value pairs at once.
167
168			=cut
169
170			sub submit_multiple {
171			my ($self, $kvcs) = @_;
172
173			push @{ $self->{queue} }, [AE::now, $kvcs];
174
175			$self->_send
176			unless $self->{sending};
177			}
178
179			sub submit {
180			my ($self, $k, $v, $clock, $host) = @_;
181
182			push @{ $self->{queue} }, [AE::now, [[$k, $v, $clock, $host]]];
183
184			$self->_send;
185			}
186
187			# start sending
188			sub _send {
189			my ($self) = @_;
190
191			if ($self->{delay}) {
192			Scalar::Util::weaken $self;
193			$self->{delay_w} \|\|= AE::timer $self->{delay}, 0, sub {
194			delete $self->{delay_w};
195			$self->{send_immediate} = 1;
196			$self->_send2 unless $self->{sending}++;
197			};
198			} else {
199			$self->{send_immediate} = 1;
200			$self->_send2 unless $self->{sending}++;
201			}
202			}
203
204			# actually do send
205			sub _send2 {
206			my ($self) = @_;
207
208			Scalar::Util::weaken $self;
209			$self->{connect_w} = AnyEvent::Socket::tcp_connect $self->{zhost}, $self->{zport}, sub {
210			my ($fh) = @_;
211
212			$fh
213			or return $self->_retry;
214
215			delete $self->{retry};
216
217			my $data = delete $self->{queue};
218			my $items = [map @{ $_->[1] }, @$data];
219
220			my $fail = sub {
221			$self->{on_error}($items, $_[0]);
222			$self->_retry;
223			};
224
225			$self->{hdl} = new AnyEvent::Handle
226			fh => $fh,
227			on_error => sub {
228			$fail->($_[2]);
229			},
230			on_read => sub {
231			if (13 <= length $_[0]{rbuf}) {
232			my ($zbxd, $version, $length) = unpack "a4 C Q<", $_[0]{rbuf};
233
234			$zbxd eq "ZBXD"
235			or return $fail->("protocol mismatch");
236			$version == 1
237			or return $fail->("protocol version mismatch");
238
239			if (13 + $length <= length $_[0]{rbuf}) {
240			my $res = eval { $json->decode (substr $_[0]{rbuf}, 13) }
241			or return $fail->("protocol error");
242			use Data::Dump; ddx $res;#d#
243
244			$self->{on_response}($items, $res);
245			$self->_send2 if delete $self->{send_immediate};
246
247			$self->{on_clear}() unless @{ $self->{queue} };
248			}
249			}
250			},
251			;
252
253			my $json = $json->encode ({
254			request => "sender data",
255			clock => AE::now,
256			data => [
257			map {
258			my $slot = $_;
259
260			map {
261			key => $_->[0],
262			value => $_->[1],
263			clock => $_->[2] // $slot->[0],
264			host => $_->[3] // $self->{host},
265			}, @{ $slot->[1] }
266			} @$data
267			],
268			});
269
270			warn $json;
271
272			$self->{hdl}->push_write (pack "a4 C Q</a", "ZBXD", 1, $json);
273			};
274			}
275
276			sub _retry {
277			my ($self) = @_;
278
279			Scalar::Util::weaken $self;
280
281			delete $self->{hdl};
282
283			my $expire = AE::now - $self->{queue_time};
284			while (@{ $self->{queue} } && $self->{queue}[0][0] < $expire) {
285			$self->{on_loss}([shift @{ $self->{queue} }]);
286			}
287
288			unless (@{ $self->{queue} }) {
289			$self->{on_clear}();
290			return;
291			}
292
293			my $retry = $self->{retry_min} * 2 ** $self->{retry}++;
294			$retry = $self->{retry_max} if $retry > $self->{retry_max};
295			warn "retry $retry\n";#d#
296			$self->{retry_w} = AE::timer $retry, 0, sub {
297			delete $self->{retry_w};
298			$self->_send2;
299			};
300			}
301
302			=back
303
304			=head1 SEE ALSO
305
306			L<AnyEvent>.
307
308			=head1 AUTHOR
309
310			Marc Lehmann <schmorp@schmorp.de>
311			http://home.schmorp.de/
312
313			=cut
314
315			1
316