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->($zbx, \@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->($zbx, \@items) (default: log and continue)

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->($zbx, \@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_response => $NOP,
      on_error    => sub {
         AE::log 4 => "$_[0]{zhost}:$_[0]{zport}: $_[1]"; # error
      },
      on_loss     => sub {
         my $nitems = @{ $_[1] };
         AE::log 5 => "$_[0]{zhost}:$_[0]{zport}: $nitems items lost"; # warn
      },

      @_,

      on_clear    => $NOP,
   }, $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) = @_;

   while (@{ $self->{queue} } || $self->{sending}) {
      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};

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

      my $fail = sub {
         $self->{on_error}($self, $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}) {
                  delete $self->{hdl};

                  my $res = eval { $json->decode (substr $_[0]{rbuf}, 13) }
                     or return $fail->("protocol error");

                  $self->{on_response}($self, $items, $res);

                  delete $self->{sending};

                  $self->_send2 if delete $self->{send_immediate} && $self->{queue};

                  $self->{on_clear}();
               }
            }
         },
      ;

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

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

      $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}($self, [shift @{ $self->{queue} }]);
   }

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

   my $retry = $self->{retry_min} * 2 ** $self->{retry}++;
   $retry = $self->{retry_max} if $retry > $self->{retry_max};
   $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.3
Committed:	Thu Jan 18 15:38:11 2018 UTC (6 years, 10 months ago) by root
Branch:	MAIN
Changes since 1.2:	+5 -4 lines
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	root	1.2	=item on_error => $cb->($zbx, \@items, $msg) (default: log and continue)
84	root	1.1
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	root	1.3	=item on_loss => $cb->($zbx, \@items) (default: log and continue)
90	root	1.1
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	root	1.3	=item on_response => $cb->($zbx, \@items, \%response) (default: not called)
95	root	1.1
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_response => $NOP,
118			on_error => sub {
119	root	1.2	AE::log 4 => "$_[0]{zhost}:$_[0]{zport}: $_[1]"; # error
120	root	1.1	},
121	root	1.2	on_loss => sub {
122			my $nitems = @{ $_[1] };
123			AE::log 5 => "$_[0]{zhost}:$_[0]{zport}: $nitems items lost"; # warn
124			},
125
126	root	1.1	@_,
127	root	1.2
128			on_clear => $NOP,
129	root	1.1	}, $class;
130
131			($self->{zhost}, $self->{zport}) = AnyEvent::Socket::parse_hostport $self->{server}, 10051;
132
133			$self->{host} //= do {
134			require POSIX;
135			(POSIX::uname())[1]
136			};
137
138			$self->{linger_time} //= $self->{queue_time};
139
140			$self
141			}
142
143			sub DESTROY {
144			my ($self) = @_;
145
146			$self->_wait;
147
148			%$self = ();
149			}
150
151			sub _wait {
152			my ($self) = @_;
153
154	root	1.2	while (@{ $self->{queue} } \|\| $self->{sending}) {
155			my $cv = AE::cv;
156	root	1.1
157	root	1.2	my $to = AE::timer $self->{linger_time}, 0, $cv;
158			local $self->{on_clear} = $cv;
159	root	1.1
160	root	1.2	$cv->recv;
161			}
162	root	1.1	}
163
164			=item $zbx->submit ($k, $v[, $clock[, $host]])
165
166			Submits a new key-value pair to the zabbix server. If C<$clock> is missing
167			or C<undef>, then C<AE::now> is used for the event timestamp. If C<$host>
168			is missing, then the default set during object creation is used.
169
170			=item $zbx->submit_multiple ([ [$k, $v, $clock, $host]... ])
171
172			Like C<submit>, but submits many key-value pairs at once.
173
174			=cut
175
176			sub submit_multiple {
177			my ($self, $kvcs) = @_;
178
179			push @{ $self->{queue} }, [AE::now, $kvcs];
180
181			$self->_send
182			unless $self->{sending};
183			}
184
185			sub submit {
186			my ($self, $k, $v, $clock, $host) = @_;
187
188			push @{ $self->{queue} }, [AE::now, [[$k, $v, $clock, $host]]];
189
190			$self->_send;
191			}
192
193			# start sending
194			sub _send {
195			my ($self) = @_;
196
197			if ($self->{delay}) {
198			Scalar::Util::weaken $self;
199			$self->{delay_w} \|\|= AE::timer $self->{delay}, 0, sub {
200			delete $self->{delay_w};
201			$self->{send_immediate} = 1;
202			$self->_send2 unless $self->{sending}++;
203			};
204			} else {
205			$self->{send_immediate} = 1;
206			$self->_send2 unless $self->{sending}++;
207			}
208			}
209
210			# actually do send
211			sub _send2 {
212			my ($self) = @_;
213
214			Scalar::Util::weaken $self;
215			$self->{connect_w} = AnyEvent::Socket::tcp_connect $self->{zhost}, $self->{zport}, sub {
216			my ($fh) = @_;
217
218			$fh
219			or return $self->_retry;
220
221			delete $self->{retry};
222
223	root	1.3	delete $self->{send_immediate};
224	root	1.1	my $data = delete $self->{queue};
225			my $items = [map @{ $_->[1] }, @$data];
226
227			my $fail = sub {
228	root	1.2	$self->{on_error}($self, $items, $_[0]);
229	root	1.1	$self->_retry;
230			};
231
232			$self->{hdl} = new AnyEvent::Handle
233			fh => $fh,
234			on_error => sub {
235			$fail->($_[2]);
236			},
237			on_read => sub {
238			if (13 <= length $_[0]{rbuf}) {
239			my ($zbxd, $version, $length) = unpack "a4 C Q<", $_[0]{rbuf};
240
241			$zbxd eq "ZBXD"
242			or return $fail->("protocol mismatch");
243			$version == 1
244			or return $fail->("protocol version mismatch");
245
246			if (13 + $length <= length $_[0]{rbuf}) {
247	root	1.2	delete $self->{hdl};
248
249	root	1.1	my $res = eval { $json->decode (substr $_[0]{rbuf}, 13) }
250			or return $fail->("protocol error");
251
252	root	1.2	$self->{on_response}($self, $items, $res);
253
254			delete $self->{sending};
255	root	1.1
256	root	1.2	$self->_send2 if delete $self->{send_immediate} && $self->{queue};
257
258			$self->{on_clear}();
259	root	1.1	}
260			}
261			},
262			;
263
264			my $json = $json->encode ({
265			request => "sender data",
266	root	1.3	clock => int AE::now,
267	root	1.1	data => [
268			map {
269			my $slot = $_;
270
271			map {
272			key => $_->[0],
273			value => $_->[1],
274	root	1.3	clock => int ($_->[2] // $slot->[0]),
275	root	1.1	host => $_->[3] // $self->{host},
276			}, @{ $slot->[1] }
277			} @$data
278			],
279			});
280
281			$self->{hdl}->push_write (pack "a4 C Q</a", "ZBXD", 1, $json);
282			};
283			}
284
285			sub _retry {
286			my ($self) = @_;
287
288			Scalar::Util::weaken $self;
289
290			delete $self->{hdl};
291
292			my $expire = AE::now - $self->{queue_time};
293			while (@{ $self->{queue} } && $self->{queue}[0][0] < $expire) {
294	root	1.2	$self->{on_loss}($self, [shift @{ $self->{queue} }]);
295	root	1.1	}
296
297			unless (@{ $self->{queue} }) {
298	root	1.2	delete $self->{sending};
299	root	1.1	$self->{on_clear}();
300			return;
301			}
302
303			my $retry = $self->{retry_min} * 2 ** $self->{retry}++;
304			$retry = $self->{retry_max} if $retry > $self->{retry_max};
305			$self->{retry_w} = AE::timer $retry, 0, sub {
306			delete $self->{retry_w};
307			$self->_send2;
308			};
309			}
310
311			=back
312
313			=head1 SEE ALSO
314
315			L<AnyEvent>.
316
317			=head1 AUTHOR
318
319			Marc Lehmann <schmorp@schmorp.de>
320			http://home.schmorp.de/
321
322			=cut
323
324			1
325