view release on metacpan or  search on metacpan

lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm  view on Meta::CPAN

use POSIX            qw(setsid strftime);
use Errno            qw();

# The daemon is a singleton per process, so its runtime state lives in
# file-scoped lexicals rather than being threaded through every helper.
my $JSON;          # JSON::MaybeXS codec (required at runtime, see execute)
my $OIF;           # the online model
my %OPT;           # resolved options
my $LOG_FH;        # log handle (STDERR in foreground without --log)
my %CONN;          # fileno => { sock, inbuf, outbuf, mode }
my $DIRTY = 0;     # learned anything since the last save?
my $RUN;           # cleared by SIGTERM/SIGINT
my $SAVE_NOW;      # set by SIGUSR1
my $REOPEN_LOG;    # set by SIGHUP

# Sanity caps on per-connection buffers: a client is allowed big batch
# messages, but one that streams an endless line (or stops reading its
# replies) gets dropped instead of eating the daemon's memory.
use constant MAX_INBUF  => 16 * 1024 * 1024;
use constant MAX_OUTBUF => 16 * 1024 * 1024;

sub opt_spec {
	return (
		[
			'set=s',
			'Named instance. Appended to --model-dir, and the socket/pid become <set>.sock / <set>.pid '
				. 'under the run dir, so several daemons run side by side with no other flags. '
				. 'Must match /\A[A-Za-z0-9+\-@_]+\z/.'
		],
		[
			'socket=s',
			'Unix domain socket to listen on; default /var/run/iforest_streamd/streamd.sock. With --set '
				. 'this is instead the base run dir (default /var/run/iforest_streamd) the <set>.sock is created in.',
			{ 'completion' => 'files' }
		],
		[
			'pid=s',
			'Where to write the daemon pid; default /var/run/iforest_streamd/streamd.pid. With --set '
				. 'this is instead the base run dir (default /var/run/iforest_streamd) the <set>.pid is created in.',
			{ 'completion' => 'files' }
		],
		[
			'model-dir=s',
			'Directory timestamped model saves land in; the symlink latest.json in it always points '
				. 'at the newest, and the daemon resumes from it at startup when it exists. With --set '
				. '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"}

lib/Algorithm/Classifier/IsolationForest/App/Command/streamd.pm  view on Meta::CPAN

		chomp $old if defined $old;
		if ( defined $old && $old =~ /\A\d+\z/ && ( kill( 0, $old ) || $!{EPERM} ) ) {
			die( 'another daemon appears to be running (pid ' . $old . ' from "' . $OPT{'pid'} . '")' . "\n" );
		}
		unlink $OPT{'pid'};       # stale pid file
	}

	# --- resume or create the model ----------------------------------------
	my $latest = File::Spec->catfile( $OPT{'model_dir'}, 'latest.json' );
	if ( -e $latest ) {
		$OIF = Algorithm::Classifier::IsolationForest->load($latest);
		die( '"' . $latest . '" is not an online model; streamd only works on those' . "\n" )
			unless ref $OIF eq 'Algorithm::Classifier::IsolationForest::Online';
	} elsif ( defined $OPT{'prototype'} ) {
		my $proto = eval {
			Algorithm::Classifier::IsolationForest->validate_prototype( scalar read_file( $OPT{'prototype'} ) );
		};
		die( '--prototype, "' . $OPT{'prototype'} . '", is not a valid prototype: ' . $@ ) if $@;
		die( '--prototype, "' . $OPT{'prototype'} . '", is for a batch model; streamd needs an online one' . "\n" )
			unless $proto->{class} eq 'online';

		my %overrides;
		$overrides{'n_trees'}          = $OPT{'n'}         if defined $OPT{'n'};
		$overrides{'window_size'}      = $OPT{'window'}    if defined $OPT{'window'};
		$overrides{'max_leaf_samples'} = $OPT{'eta'}       if defined $OPT{'eta'};
		$overrides{'growth'}           = $OPT{'growth'}    if defined $OPT{'growth'};
		$overrides{'subsample'}        = $OPT{'subsample'} if defined $OPT{'subsample'};
		$overrides{'seed'}             = $OPT{'s'}         if defined $OPT{'s'};
		$overrides{'contamination'}    = $OPT{'c'}         if defined $OPT{'c'};

		$OIF = eval { Algorithm::Classifier::IsolationForest->new_from_prototype( $proto, %overrides ) };
		die( '--prototype, "' . $OPT{'prototype'} . '", failed to create a model: ' . $@ ) if $@;
	} else {
		my $mungers;
		if ( defined $OPT{'mungers'} ) {
			$mungers = eval { $JSON->decode( scalar read_file( $OPT{'mungers'} ) ) };
			die( '--mungers, "' . $OPT{'mungers'} . '", did not parse as JSON: ' . $@ ) if $@;
			die( '--mungers, "' . $OPT{'mungers'} . '", must be a JSON object of tag => spec' )
				unless ref $mungers eq 'HASH';
		}
		$OIF = Algorithm::Classifier::IsolationForest::Online->new(
			'n_trees'          => $OPT{'n'},
			'window_size'      => $OPT{'window'},
			'max_leaf_samples' => $OPT{'eta'},
			'growth'           => $OPT{'growth'},
			'subsample'        => $OPT{'subsample'},
			'seed'             => $OPT{'s'},
			'contamination'    => $OPT{'c'},
			'feature_names'    => $OPT{'t'},
			'mungers'          => $mungers,
		);
	} ## end else [ if ( -e $latest ) ]

	# --- bind, then daemonize (the listening fd survives the forks) --------
	my $listener = IO::Socket::UNIX->new(
		Local  => $OPT{'socket'},
		Listen => 64,
	) or die( 'failed to listen on "' . $OPT{'socket'} . '": ' . $! . "\n" );
	$listener->blocking(0);
	if ( defined $OPT{'socket_mode'} ) {
		chmod( oct( $OPT{'socket_mode'} ), $OPT{'socket'} )
			or die( 'failed to chmod "' . $OPT{'socket'} . '" to ' . $OPT{'socket_mode'} . ': ' . $! . "\n" );
	}

	if ( !$OPT{'f'} ) {
		$OPT{'log'} = File::Spec->catfile( $OPT{'model_dir'}, 'streamd.log' )
			unless defined $OPT{'log'};
		_daemonize();
	}
	_open_log();

	write_file( $OPT{'pid'}, { 'atomic' => 1 }, $$ . "\n" );

	# --- signals -------------------------------------------------------------
	$RUN        = 1;
	$SAVE_NOW   = 0;
	$REOPEN_LOG = 0;
	local $SIG{TERM} = sub { $RUN        = 0 };
	local $SIG{INT}  = sub { $RUN        = 0 };
	local $SIG{USR1} = sub { $SAVE_NOW   = 1 };
	local $SIG{HUP}  = sub { $REOPEN_LOG = 1 };
	local $SIG{PIPE} = 'IGNORE';

	_log(     'listening on '
			. $OPT{'socket'}
			. ( -e $latest          ? ' (resumed '                : ' (new model, ' )
			. ( defined $OPT{'set'} ? 'set=' . $OPT{'set'} . ', ' : '' ) . 'seen='
			. $OIF->seen
			. ', save-interval='
			. $OPT{'save_interval'} . 's'
			. ', model-dir='
			. $OPT{'model_dir'}
			. ')' );

	# --- event loop ----------------------------------------------------------
	my $rsel      = IO::Select->new($listener);
	my $wsel      = IO::Select->new();
	my $next_save = time + $OPT{'save_interval'};

	while ($RUN) {
		my $timeout = $next_save - time;
		$timeout = 0 if $timeout < 0;

		my @ready = $rsel->can_read($timeout);
		for my $s (@ready) {
			if ( $s == $listener ) {
				while ( my $cl = $listener->accept ) {
					$cl->blocking(0);
					$CONN{ fileno($cl) } = { sock => $cl, inbuf => '', outbuf => '', mode => 'prequential' };
					$rsel->add($cl);
				}
				next;
			}
			_read_from( $s, $rsel, $wsel );
		} ## end for my $s (@ready)

		# Drain clients whose replies did not fit in one write.
		if ( $wsel->count ) {
			for my $s ( $wsel->can_write(0) ) {
				_flush( $s, $rsel, $wsel );
			}
		}