view release on metacpan or  search on metacpan

lib/Algorithm/ToNumberMunger.pm  view on Meta::CPAN

		croak "cidr munger$where: '" . ( defined $v ? $v : 'undef' ) . "' is not a parseable IP address";
	}; ## end sub
} ## end sub _build_cidr

=head2 datetime

    { munger => 'datetime', format => '%Y-%m-%dT%H:%M:%S', part => 'epoch' }
    { munger => 'datetime', format => '%Y-%m-%d %H:%M:%S', part => 'hour' }

Parse a formatted timestamp with L<Time::Piece> (C<strptime>, so C<format> is a
standard strptime pattern) and extract one numeric C<part>:

=over 4

=item * C<epoch> (default) - seconds since the epoch.

=item * C<year>, C<mon> (1-12), C<mday> (1-31), C<hour>, C<min>, C<sec>.

=item * C<wday> - day of week, C<0>=Sunday .. C<6>=Saturday.

=item * C<yday> - day of year, C<0>-based.

=item * C<frac_day> - time of day as a fraction in C<[0, 1)>, i.e.
C<(hour*3600 + min*60 + sec) / 86400>. Handy as a cyclic-ish time-of-day feature.

=item * C<frac_week> - position within the week as a fraction in C<[0, 1)>, the
week starting Sunday to match C<wday>: C<(wday*86400 + hour*3600 + min*60 + sec)
/ 604800>. Like C<frac_day> but cycling over a week, so a weekly rhythm (weekend
vs. weekday, or a Monday-morning batch) shows up as a feature.

=item * C<sin_day> / C<cos_day>, C<sin_week> / C<cos_week> - the C<frac_*> value
mapped onto a circle, C<sin(2*pi*frac)> and C<cos(2*pi*frac)>. Prefer these over
the raw C<frac_*> when feeding the forest: a plain fraction has a false seam at
the wrap (23:59 and 00:00 sit at opposite ends, 1 vs 0, though they are a minute
apart), whereas the sin/cos pair is continuous across midnight/Sunday. Store
I<both> of a pair in two columns so the position is unambiguous.

=back

Time features often carry the anomaly (a job that normally runs at 03:00
suddenly firing at noon, or a weekday task firing on a Sunday), which is why this
is a first-class munger.

B<Multi-output form.> A cyclic pair belongs together -- C<sin> alone collides
(C<sin> is symmetric about its peak, so two different times map to one value) and
the forest then treats distinct times as identical. To emit a pair atomically,
give C<parts> (plural) and route them to two columns with C<into> (see
L</compile>):

    "time_of_week": {
        "munger": "datetime", "from": "timestamp",
        "format": "%Y-%m-%dT%H:%M:%S",
        "parts":  [ "sin_week", "cos_week" ],
        "into":   [ "time_sin", "time_cos" ]
    }

The timestamp is parsed once and both columns are filled together, so they can
never drift apart or be half-configured. C<parts> and C<into> must be the same
length. (Using C<parts> without C<into>, or C<part> with C<into>, is an error.)

B<Performance.> Two transparent accelerations, both value-identical to the plain
path: a one-slot memo returns the previous result when the same stamp string
repeats (the common case in bursty event streams); and when the format is built
from only the six numeric codes C<%Y %m %d %H %M %S> (once each, e.g.
C<%Y-%m-%dT%H:%M:%S>), parsing skips C<strptime> for a compiled regex plus
integer date math, falling back to C<strptime> for any value the regex does not
match B<or whose fields are out of range> (a month C<13>, an hour C<24>, a
C<Feb 30>) -- so an invalid stamp croaks or normalizes exactly as C<strptime>
would, never silently feeding nonsense to the date math. Like C<strptime>
without a zone code, stamps are treated as UTC.

=cut

# Fraction (in [0,1)) of the way through the day / week, shared by the frac_*
# parts and their sin/cos cyclic encodings.
sub _frac_day {
	my $t = shift;
	return ( $t->hour * 3600 + $t->min * 60 + $t->sec ) / 86400;
}

sub _frac_week {
	my $t = shift;
	return ( $t->day_of_week * 86400 + $t->hour * 3600 + $t->min * 60 + $t->sec ) / 604800;
}

my $TWO_PI = 2 * atan2( 0, -1 );    # atan2(0,-1) == pi, core-only, no POSIX

# part name => how to pull it off a Time::Piece object.
my %DATETIME_PART = (
	epoch     => sub { $_[0]->epoch },
	year      => sub { $_[0]->year },
	mon       => sub { $_[0]->mon },
	mday      => sub { $_[0]->mday },
	hour      => sub { $_[0]->hour },
	min       => sub { $_[0]->min },
	sec       => sub { $_[0]->sec },
	wday      => sub { $_[0]->day_of_week },
	yday      => sub { $_[0]->yday },
	frac_day  => \&_frac_day,
	frac_week => \&_frac_week,
	sin_day   => sub { sin( $TWO_PI * _frac_day( $_[0] ) ) },
	cos_day   => sub { cos( $TWO_PI * _frac_day( $_[0] ) ) },
	sin_week  => sub { sin( $TWO_PI * _frac_week( $_[0] ) ) },
	cos_week  => sub { cos( $TWO_PI * _frac_week( $_[0] ) ) },
);

# ---- fast fixed-format engine ----------------------------------------------
#
# Time::Piece->strptime costs microseconds per call. When the format is built
# from only the six all-numeric codes below (once each, e.g. the ubiquitous
# '%Y-%m-%dT%H:%M:%S'), we can compile it to a capture regex and derive every
# part with integer math instead -- several times faster, and bit-identical:
# both paths treat the stamp as UTC (strptime with no zone does the same).
# Anything fancier (%b, %z, %j, ...) stays on strptime.

# strptime code => [ field name, capture pattern ].
my %FAST_CODE = (
	Y => [ 'year', '[0-9]{4}' ],
	m => [ 'mon',  '[0-9]{2}' ],
	d => [ 'mday', '[0-9]{2}' ],
	H => [ 'hour', '[0-9]{2}' ],

lib/Algorithm/ToNumberMunger.pm  view on Meta::CPAN

    { munger => 'eps', prefix => 'http-req:', from => 'src_ip',
      parts => [ 'rate', 'count' ], into => [ 'req_rate', 'req_count' ] }

Per-entity sliding-window event rates via the C<iqbi-damiq> daemon shipped with
L<Algorithm::EventsPerSecond> (see
L<Algorithm::EventsPerSecond::Sukkal>). The input value becomes a meter B<key>
(after C<prefix> is prepended); by default the munger B<marks> one event against
that key and returns the key's current events-per-second, using the daemon's
C<MARKRATE> command -- mark and query in a single command with a single reply.
This is the munger behind rate columns like a per-source request rate: every
event marks its source's meter and stores the rate the meter now reads.

Unlike every other munger this one consults external state -- but the state
lives in the daemon, not here, so the munger itself remains a stateless client
and rows stay reproducible I<given> the daemon. Because the daemon is shared,
multiple writer processes marking the same keys see one B<global> rate, which an
in-process meter could never give.

Spec keys:

=over 4

=item * C<socket> - unix socket path of the daemon. Defaults to
C<$Algorithm::ToNumberMunger::EPS_SOCKET>
(C</var/run/iqbi-damiq.sock>).

=item * C<prefix> - string prepended to the input to form the key, namespacing
this column's meters (two columns keyed on the same field need different
prefixes or they share meters). No whitespace/control characters. Default C<''>.

=item * C<mark> - whether to mark the key (default C<1>). Marking rides
C<MARKRATE>: with C<< read => 'rate' >> that one command is the whole exchange;
with C<count>/C<total> the read is pipelined after it (the C<MARKRATE> rate
reply is discarded), so a marking failure still comes back as an ordinary
first reply. With C<< mark => 0 >> the munger only reads, for columns whose
marking is done elsewhere -- e.g. an NXDOMAIN rate is I<marked> by the pipeline
only on NXDOMAIN responses but I<read> on every row.

=item * C<read> - what to read: C<rate> (events/sec over the daemon's window,
default), C<count> (events inside the window), or C<total> (lifetime).

=item * C<parts> + C<into> - multi-output form (see L</compile>): read several
of C<rate>/C<count>/C<total> for the one key in a single round trip, filling one
column each. When marking, the C<MARKRATE> reply itself serves the first C<rate>
part, so C<< parts => ['rate', 'count'] >> costs exactly two commands.

=item * C<on_error> - C<'die'> (default) croaks the write when the daemon is
unreachable or replies C<ERR>; a number is returned instead as a quiet fallback.
Note C<0> is indistinguishable from a genuinely idle key, so quiet fallback
biases the column -- loud is the default on purpose.

=item * C<timeout> - per-operation socket timeout in seconds (default 5,
best-effort via C<SO_RCVTIMEO>/C<SO_SNDTIMEO>), so a wedged daemon cannot hang a
writer forever.

=back

Semantics worth knowing: a marked read B<includes the event just marked>; keys
have whitespace/control bytes replaced with C<_> to satisfy the daemon's key
rules; connections are made lazily on first use and kept open (reconnecting
transparently after a fork or an error), so compiling a plan -- including the
eager validation in C<write_info> -- needs no running daemon. Each eps column
costs one unix-socket round trip per row; the multi-output form exists so
rate+count of the same key costs one round trip, not two.

=cut

# Default socket path of the iqbi-damiq daemon.
our $EPS_SOCKET = '/var/run/iqbi-damiq.sock';

# Persistent daemon connections, keyed by socket path, shared by every eps
# munger in the process. Entries record the pid that opened them so a forked
# writer transparently reopens instead of sharing a socket with its parent.
# Connections are made lazily on first use -- never at munger build time, so a
# plan can compile (eager validation) with no daemon running.
my %EPS_CONN;

sub _eps_conn {
	my ( $path, $timeout ) = @_;
	my $c = $EPS_CONN{$path};
	return $c->{fh} if $c && $c->{pid} == $$;

	require Socket;
	require IO::Socket::UNIX;
	my $fh = IO::Socket::UNIX->new(
		Type => Socket::SOCK_STREAM(),
		Peer => $path,
	) or die "cannot connect to iqbi-damiq at $path: $!\n";

	# Best-effort read/write timeouts so a wedged daemon cannot hang a writer.
	eval {
		my $tv = pack( 'l!l!', $timeout, 0 );
		setsockopt( $fh, Socket::SOL_SOCKET(), Socket::SO_RCVTIMEO(), $tv );
		setsockopt( $fh, Socket::SOL_SOCKET(), Socket::SO_SNDTIMEO(), $tv );
	};

	$EPS_CONN{$path} = { fh => $fh, pid => $$ };
	return $fh;
} ## end sub _eps_conn

# One pipelined transaction: send $cmd (possibly several lines) and read
# $nreplies "OK n" lines, one per command sent. The munger only ever sends
# commands that reply exactly once -- MARKRATE (which marks AND returns the
# rate in one go), RATE, COUNT, TOTAL; never a bare MARK, whose reply-only-on-
# error behavior would let a failure desynchronize the reply stream. Dies on
# ERR, EOF, or timeout; the caller still drops the cached connection on error
# as belt and braces.
sub _eps_txn {
	my ( $path, $timeout, $cmd, $nreplies ) = @_;
	my $fh = _eps_conn( $path, $timeout );
	print {$fh} $cmd or die "write to iqbi-damiq failed: $!\n";
	my @out;
	for ( 1 .. $nreplies ) {
		my $reply = <$fh>;
		die "iqbi-damiq closed the connection (or timed out)\n"
			unless defined $reply;
		$reply =~ /\AOK (\S+)/
			or die "iqbi-damiq replied: $reply";
		push @out, $1 + 0;
	}
	return @out;
} ## end sub _eps_txn

# Validate the spec keys shared by the scalar and multi-output eps builders.
sub _eps_spec {
	my ( $spec, $where ) = @_;

	my $socket = defined $spec->{socket} ? $spec->{socket} : $EPS_SOCKET;
	croak "eps munger$where: 'socket' must be a non-empty path"
		unless length $socket;

	my $prefix = defined $spec->{prefix} ? $spec->{prefix} : '';
	croak "eps munger$where: 'prefix' may not contain whitespace or control " . 'characters'