Algorithm-Classifier-IsolationForest
view release on metacpan or search on metacpan
lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm view on Meta::CPAN
. 'the set name is appended as a subdirectory.',
{ 'default' => '/var/db/iforest_streamd', 'completion' => 'files' }
],
[
'save-interval=i',
'Seconds between periodic model saves (only when learning happened).',
{ 'default' => 300 }
],
[ 'keep=i', 'Prune all but the newest N timestamped model files after each save.' ],
[ 'f|foreground', 'Do not daemonize; log to stderr unless --log is given.' ],
[
'log=s',
'Log file. Defaults to <model-dir>/streamd.log when daemonized; stderr in the foreground.',
{ 'completion' => 'files' }
],
[ 'socket-mode=s', 'Octal permissions to chmod the socket file to (e.g. 0660).' ],
[ 'threshold=f', 'Alternative decision threshold to use for the label field. 0 < $val < 1' ],
# creation knobs, used only when <model-dir>/latest.json does not exist yet
[ 'n=i', 'Number of isolation trees in the ensemble (new models only).' ],
[ 'window=i', 'Sliding window size; 0 disables forgetting (new models only).' ],
[ 'eta=i', 'max_leaf_samples: points a leaf accumulates before splitting (new models only).' ],
[ 'growth=s', "Leaf split-requirement growth, 'adaptive' or 'fixed' (new models only)." ],
[ 'subsample=f', 'Per-tree stream subsampling probability, in (0, 1] (new models only).' ],
[ 's=i', 'Seed int (new models only).' ],
[
'c=f',
'Contamination. Expected fraction of anomalies, in (0, 0.5]; the decision threshold is '
. 'relearned from the window before every save (new models only).'
],
[
't=s@',
'Feature name tag. Pass once per feature; enables the tagged (JSON object) row form '
. '(new models only).'
],
[
'mungers=s',
'JSON file of Algorithm::ToNumberMunger specs, keyed by feature tag (new models only; requires -t).',
{ 'completion' => 'files' }
],
[
'prototype=s',
'JSON prototype file to create the model from (new models only). May not be combined '
. 'with -t or --mungers. See PROTOTYPES in the module POD.',
{ 'completion' => 'files' }
],
);
} ## end sub opt_spec
sub abstract { 'Run an Online Isolation Forest scoring daemon on a Unix socket, speaking JSON lines' }
sub description {
'Runs a prequential scoring daemon around an Online Isolation Forest
model (Algorithm::Classifier::IsolationForest::Online): clients connect
to the Unix domain socket and exchange one JSON document per line.
At startup the daemon resumes from <model-dir>/latest.json when it
exists; otherwise it creates a new model from the creation knobs (-n,
--window, --eta, --growth, --subsample, -s, -c, -t, --mungers,
--prototype -- the same set `iforest stream` takes). The model is saved
to a timestamped file in --model-dir every --save-interval seconds
(only when something was learned), on SIGUSR1, on the save command, and
at shutdown; the symlink latest.json is atomically repointed at every
save, so a restart resumes the stream losing at most one interval.
Requests are JSON objects carrying exactly one of "row", "rows", or
"cmd", an optional "mode", and an optional "tag" (any JSON value,
echoed back verbatim in the reply -- a correlation tag, not to be
confused with feature tags):
{"row": [0.1, 0.7]} -> {"score": 0.41, "label": 0}
{"row": {"cpu": 0.1, "mem": 0.7}} -> {"score": 0.41, "label": 0}
{"rows": [[...], {...}], "tag": "b7"} -> {"scores": [[0.41,0], ...], "tag": "b7"}
{"rows": [[...]], "mode": "learn"} -> {"ok": {"learned": 1}}
{"cmd": "mode", "mode": "score"} -> {"ok": {"mode": "score"}}
{"cmd": "ping"} -> {"ok": "pong"}
{"cmd": "stats"} -> {"ok": {"seen": ..., ...}}
{"cmd": "save"} -> {"ok": {"saved": "oiforest-....json"}}
{"cmd": "relearn-threshold"} -> {"ok": {"threshold": 0.61}}
anything invalid -> {"error": "...", "tag": ...}
The array row form is positional (scalar mungers applied, like stream
CSV input); the object form is a tagged row and runs the full munger
plan, including expanding and combining mungers -- and, being JSON, the
raw values may safely contain commas, newlines, or any unicode.
A worked tagged example. Create the daemon around raw HTTP request
data, with mungers turning the raw values into numbers (mungers.json
here; a --prototype carrying the same schema works identically):
{ "method": { "munger": "http_method_enum", "default": -1 },
"path_len": { "munger": "length", "from": "path" },
"host_entropy": { "munger": "entropy", "from": "host" } }
iforest streamd --set web -t method -t path_len -t host_entropy \
--mungers mungers.json -c 0.05
Clients then send the raw values themselves -- note the input fields
are the munger SOURCES (method, path, host), not the feature tags,
because the plan derives path_len and host_entropy from them:
-> {"row": {"method": "GET", "path": "/index.html",
"host": "www.example.com"}, "tag": "r-1"}
<- {"score": 0.31, "label": 0, "tag": "r-1"}
-> {"row": {"method": "BREW", "path": "/aa,a\"a.php",
"host": "kq3xv9z2.biz"}, "tag": "r-2"}
<- {"score": 0.74, "label": 1, "tag": "r-2"}
The same rows work from the shell via
`iforest streamc --set web --jsonl -i rows.jsonl`.
Modes are prequential (score each row against the model as it stood, then
learn it -- the default), learn (learn only), and score (score only);
"mode" on a row/rows message overrides the connection default set by
the mode command for that message. A bad row gets an {"error": ...}
reply on that message only; the connection and the daemon live on (for
a "rows" batch, rows before the failing one were already processed).
Multiple concurrent connections are supported; rows are applied to the
one shared model in the order their lines arrive, which defines the
stream order.
--set NAME runs a named instance: the set name is appended to
--model-dir (so its saves, latest.json, and default log live under
their own subdirectory) and the socket/pid become <set>.sock /
<set>.pid under the run dir -- with --set, --socket and --pid name the
base run dir instead of the files. Several sets run side by side, each
with its own model, resume state, and double-start protection:
iforest streamd --set web
iforest streamd --set dns --prototype dns-proto.json -c 0.02
Set names must match /\A[A-Za-z0-9+\-@_]+\z/; since the class has no
"." or "/", a set name can only ever create one new path segment.
Everything under --model-dir and the socket/pid directories is created
at startup when missing; when that fails (e.g. running unprivileged
with the /var defaults) the daemon dies immediately, before forking,
naming the directory and the flag to override.
';
} ## end sub description
sub validate {
my ( $self, $opt, $args ) = @_;
# Anchored with \A/\z rather than ^/$ ($ tolerates a trailing newline).
# The class has no '.' or '/', so a set name can only ever create one
# new path segment -- no traversal is expressible.
if ( defined( $opt->{'set'} ) && $opt->{'set'} !~ /\A[A-Za-z0-9+\-@_]+\z/ ) {
$self->usage_error( '--set, "'
. $opt->{'set'}
. '", must match /\A[A-Za-z0-9+\-@_]+\z/ (letters, digits, and + - @ _ only)' );
}
if ( $opt->{'save_interval'} < 1 ) {
$self->usage_error( '--save-interval, "' . $opt->{'save_interval'} . '", must be >= 1 second' );
}
if ( defined( $opt->{'keep'} ) && $opt->{'keep'} < 1 ) {
$self->usage_error( '--keep, "' . $opt->{'keep'} . '", must be >= 1' );
}
if ( defined( $opt->{'threshold'} ) && ( $opt->{'threshold'} <= 0 || $opt->{'threshold'} >= 1 ) ) {
$self->usage_error( '--threshold, "' . $opt->{'threshold'} . '", needs to be greater than 0 and less than 1' );
}
if ( defined( $opt->{'growth'} ) && $opt->{'growth'} !~ /\A(?:adaptive|fixed)\z/ ) {
$self->usage_error( '--growth, "' . $opt->{'growth'} . '", must be either adaptive or fixed' );
}
if ( defined( $opt->{'socket_mode'} ) && $opt->{'socket_mode'} !~ /\A0?[0-7]{3}\z/ ) {
$self->usage_error( '--socket-mode, "' . $opt->{'socket_mode'} . '", must be octal like 0660' );
}
if ( defined( $opt->{'mungers'} ) ) {
if ( !-f $opt->{'mungers'} ) {
$self->usage_error( '--mungers, "' . $opt->{'mungers'} . '", is not a file or does not exist' );
} elsif ( !-r $opt->{'mungers'} ) {
$self->usage_error( '--mungers, "' . $opt->{'mungers'} . '", is not readable' );
} elsif ( !defined( $opt->{'t'} ) ) {
$self->usage_error('--mungers requires feature tags (-t) to compile against');
}
}
if ( defined( $opt->{'prototype'} ) ) {
if ( !-f $opt->{'prototype'} ) {
$self->usage_error( '--prototype, "' . $opt->{'prototype'} . '", is not a file or does not exist' );
} elsif ( !-r $opt->{'prototype'} ) {
$self->usage_error( '--prototype, "' . $opt->{'prototype'} . '", is not readable' );
}
if ( defined( $opt->{'t'} ) || defined( $opt->{'mungers'} ) ) {
$self->usage_error(
'--prototype may not be combined with -t or --mungers; the schema comes only from the prototype');
}
} ## end if ( defined( $opt->{'prototype'} ) )
return 1;
} ## end sub validate
sub execute {
my ( $self, $opt, $args ) = @_;
# JSON::MaybeXS is required lazily so a box without it still has a
# working iforest CLI (App::Cmd loads every command module up front).
eval { require JSON::MaybeXS; 1 }
or die( 'iforest streamd requires JSON::MaybeXS for its wire protocol; install it: ' . $@ );
$JSON = JSON::MaybeXS->new( utf8 => 1, canonical => 1, allow_nonref => 0 );
%OPT = %$opt;
# --set turns --socket/--pid into base run dirs holding <set>.sock /
# <set>.pid and appends the set name to --model-dir, so several named
# daemons run side by side with no other flags. Without a set the
# flags are the socket/pid files themselves, defaulting as documented.
if ( defined $OPT{'set'} ) {
my $run = defined $OPT{'socket'} ? $OPT{'socket'} : '/var/run/iforest_streamd';
lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm view on Meta::CPAN
$next_save = time + $OPT{'save_interval'};
}
} ## end while ($RUN)
# --- shutdown ------------------------------------------------------------
_log('shutting down');
_save_model('shutdown') if $DIRTY;
for my $c ( values %CONN ) {
close $c->{sock};
}
%CONN = ();
close $listener;
unlink $OPT{'socket'};
unlink $OPT{'pid'};
_log('bye');
return 1;
} ## end sub execute
#-------------------------------------------------------------------------------
# startup helpers
#-------------------------------------------------------------------------------
# Make sure a directory the daemon needs exists and is writable, creating
# it when it does not. Run before daemonizing so a permissions problem is
# reported to the terminal that started the daemon rather than buried in a
# log the user has not found yet.
#
# Args:
# $dir :: the directory to create or check.
# $flag :: the option that asked for it, e.g. '--model-dir'. Only used
# to name the fix in the error message.
#
# Returns: 1 when the directory exists and is writable. Dies otherwise,
# telling the user to create it, fix its permissions, or point $flag
# somewhere else.
#
# Example:
# _ensure_dir( $OPT{'model_dir'}, '--model-dir' );
sub _ensure_dir {
my ( $dir, $flag ) = @_;
if ( !-d $dir ) {
my $err;
make_path( $dir, { mode => oct('0755'), error => \$err } );
die( 'could not create "'
. $dir
. '" (needed for '
. $flag
. '); create it, fix permissions, or point '
. $flag
. ' somewhere writable'
. "\n" )
if !-d $dir;
} ## end if ( !-d $dir )
die( '"' . $dir . '" (needed for ' . $flag . ') is not writable; fix permissions or override ' . $flag . "\n" )
unless -w $dir;
return 1;
} ## end sub _ensure_dir
# Classic double-fork daemonization. The parents leave via POSIX::_exit
# so no END blocks (Inline's, App::Cmd's) run twice. The second fork is
# what guarantees the daemon can never reacquire a controlling terminal.
#
# Args: none. Reads nothing and takes nothing -- the caller decides
# whether to daemonize at all (-f keeps the process in the foreground).
#
# Returns: 1, in the grandchild only. The two parents never return: they
# _exit(0) immediately, so the caller either continues as the daemon or
# does not continue. Dies on a failed fork, setsid, chdir or STDIN
# reopen.
#
# Example:
# _daemonize() unless $OPT{'f'};
# # from here on we are the daemon
sub _daemonize {
defined( my $pid = fork() ) or die( 'fork failed: ' . $! . "\n" );
POSIX::_exit(0) if $pid;
setsid() or die( 'setsid failed: ' . $! . "\n" );
defined( $pid = fork() ) or die( 'second fork failed: ' . $! . "\n" );
POSIX::_exit(0) if $pid;
chdir '/' or die( 'chdir / failed: ' . $! . "\n" );
open( STDIN, '<', '/dev/null' ) or die( 'reopen STDIN failed: ' . $! . "\n" );
return 1;
} ## end sub _daemonize
# Point the daemon's logging at wherever --log said, falling back to
# STDERR. When daemonized, STDOUT and STDERR are reopened onto the log
# too, so a warn from anywhere in the process still lands somewhere the
# operator can read. Everything is unbuffered: a daemon's log is useless
# if the interesting line is still sitting in a buffer when it wedges.
#
# Args: none. Reads $OPT{'log'} and $OPT{'f'}.
#
# Returns: 1. Sets the package's $LOG_FH as its whole purpose. Dies when
# the log file cannot be opened.
#
# Example:
# _open_log();
# _log('listening');
sub _open_log {
if ( defined $OPT{'log'} ) {
open( my $fh, '>>', $OPT{'log'} ) or die( 'failed to open log "' . $OPT{'log'} . '": ' . $! . "\n" );
$fh->autoflush(1);
$LOG_FH = $fh;
if ( !$OPT{'f'} ) {
open( STDOUT, '>>', $OPT{'log'} ) or die( 'reopen STDOUT failed: ' . $! . "\n" );
open( STDERR, '>>', $OPT{'log'} ) or die( 'reopen STDERR failed: ' . $! . "\n" );
STDOUT->autoflush(1);
STDERR->autoflush(1);
}
} else {
$LOG_FH = \*STDERR;
}
return 1;
} ## end sub _open_log
# Write one timestamped, pid-stamped line to the log. The pid matters
# because several named instances (--set) can share one log file.
#
# Args:
# $msg :: the message, without a trailing newline -- one is added.
#
# Returns: 1.
#
# Example:
# _log( 'saved ' . $name . ' (' . $why . ')' );
# # 2026-08-08T14:02:11 [4821] saved oiforest-20260808-140211.json (interval)
sub _log {
my ($msg) = @_;
print {$LOG_FH} strftime( '%Y-%m-%dT%H:%M:%S', localtime ) . ' [' . $$ . '] ' . $msg . "\n";
return 1;
}
#-------------------------------------------------------------------------------
# model persistence
#-------------------------------------------------------------------------------
# Timestamped save + atomic symlink flip. The symlink stores a relative
# name, so the whole model directory can be moved without breaking it.
lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm view on Meta::CPAN
# interval save skip a quiet period. Dies on a row that is neither shape,
# on a cell left non-numeric after munging, or on anything the model
# itself croaks about.
#
# Example:
# _apply_row( { method => 'GET', host => 'h' }, 'prequential' ); # 0.41
# _apply_row( [ 0.2, 0.7 ], 'learn' ); # undef
sub _apply_row {
my ( $row, $mode ) = @_;
my $vec;
if ( ref $row eq 'HASH' ) {
$vec = $OIF->tagged_row_to_array( $row, 'streamd' );
} elsif ( ref $row eq 'ARRAY' ) {
$vec = $row;
if ( ref $OIF->{mungers} eq 'HASH' && %{ $OIF->{mungers} } ) {
$vec = $OIF->munge_rows( [$row] )->[0];
}
} else {
die 'row must be a JSON array (positional) or object (tagged)' . "\n";
}
for my $col ( 0 .. $#$vec ) {
next if !defined $vec->[$col]; # undef defers to the model's missing policy
die 'column ' . ( $col + 1 ) . ' is not a number after munging' . "\n"
unless looks_like_number( $vec->[$col] );
}
if ( $mode eq 'learn' ) {
$OIF->learn( [$vec] );
$DIRTY = 1;
return undef;
}
if ( $mode eq 'score' ) {
return $OIF->score_samples( [$vec] )->[0];
}
my $score = $OIF->score_learn( [$vec] )->[0];
$DIRTY = 1;
return $score;
} ## end sub _apply_row
=head1 NAME
Algorithm::Classifier::IsolationForest::App::Command::streamd - Run an Online Isolation Forest scoring daemon on a Unix socket, speaking JSON lines
=head1 DESCRIPTION
Wraps the prequential loop of C<iforest stream> in a daemon: it listens
on a Unix domain socket, serves many concurrent connections from one
shared model, and exchanges one JSON document per line. Because values
travel as JSON rather than positionally, raw input headed for mungers may
safely contain commas, newlines or any unicode, and object rows run the
full munger plan -- expanding and combining mungers included, which
positional CSV cannot express.
An optional C<"tag"> on a request is echoed back verbatim. Errors are
always per-message: a bad row gets an C<{"error": ...}> reply and the
connection lives on.
Models are saved to C<--model-dir> as timestamped files every
C<--save-interval> seconds when learning has happened, plus on the C<save>
command, on SIGUSR1, and at shutdown. The C<latest.json> symlink is
repointed atomically at each save and resumed from at the next startup, so
a crash loses at most one interval of learning.
Several named instances can run side by side: C<--set> gives each its own
socket, pid file, model subdirectory and resume state.
C<iforest streamc> is the matching client.
Run it as C<iforest streamd>; C<iforest help streamd> lists every option.
=head1 METHODS
L<App::Cmd> calls these while dispatching the subcommand. Nothing else
should.
=head2 opt_spec
Returns this command's option specifications, as the list of arrayrefs
L<Getopt::Long::Descriptive> expects.
=head2 abstract
Returns the one-line summary C<iforest commands> prints beside the
command name.
=head2 description
Returns the long help text C<iforest help streamd> prints under the option
list.
=head2 validate
Checks the parsed options before anything is read or written, so a
mistake costs nothing.
Checks that C<--set> is a usable instance name, that C<--save-interval>,
C<--keep>, C<--threshold>, C<--growth> and C<--socket-mode> hold sane
values, and that a C<--mungers> spec is readable and accompanied by the
feature tags (C<-t>) it compiles against.
Takes the parsed options hashref and the arrayref of remaining
arguments. Calls C<usage_error>, which prints the usage and exits, on
the first problem it finds, and returns 1 when everything checks out.
=head2 execute
Prepares the runtime and model directories, daemonizes unless C<-f> was
given, and runs the accept/serve loop until it is asked to stop.
Takes the parsed options hashref and the arrayref of remaining
arguments, and returns 1.
=cut
return 1;
( run in 0.659 second using v1.01-cache-2.11-cpan-4ef0a570458 )