view release on metacpan or  search on metacpan

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

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';
		$OPT{'socket'} = File::Spec->catfile( $run, $OPT{'set'} . '.sock' );
		my $prun = defined $OPT{'pid'} ? $OPT{'pid'} : '/var/run/iforest_streamd';
		$OPT{'pid'}       = File::Spec->catfile( $prun, $OPT{'set'} . '.pid' );
		$OPT{'model_dir'} = File::Spec->catdir( $OPT{'model_dir'}, $OPT{'set'} );
	} else {
		$OPT{'socket'} = '/var/run/iforest_streamd/streamd.sock' unless defined $OPT{'socket'};
		$OPT{'pid'}    = '/var/run/iforest_streamd/streamd.pid'  unless defined $OPT{'pid'};
	}

	# Daemonizing chdirs to /, so every path used after that point must be
	# absolute -- including the socket and pid file, which are unlinked at
	# shutdown.
	for my $path_opt (qw(socket pid model_dir log)) {
		$OPT{$path_opt} = File::Spec->rel2abs( $OPT{$path_opt} )
			if defined $OPT{$path_opt};
	}

	# sun_path is 104 bytes on the BSDs and 108 on Linux (including the
	# NUL); Socket.pm just warns and TRUNCATES an over-long path, which
	# binds a socket nobody will ever find.  Refuse loudly instead.
	die(      '--socket, "'
			. $OPT{'socket'}
			. '", is '
			. length( $OPT{'socket'} )
			. ' bytes; Unix socket paths are limited to ~104 bytes -- use a shorter path'
			. "\n" )
		if length( $OPT{'socket'} ) > 100;

	# --- directories, before anything forks or binds -----------------------
	_ensure_dir( $OPT{'model_dir'},         '--model-dir' );
	_ensure_dir( dirname( $OPT{'socket'} ), '--socket' );
	_ensure_dir( dirname( $OPT{'pid'} ),    '--pid' );

	# --- refuse to double-start ---------------------------------------------
	if ( -e $OPT{'socket'} ) {
		my $probe = IO::Socket::UNIX->new( Peer => $OPT{'socket'} );
		die( 'another daemon is already listening on "' . $OPT{'socket'} . '"' . "\n" ) if $probe;
		unlink $OPT{'socket'};    # stale socket from an unclean exit
	}
	if ( -f $OPT{'pid'} ) {
		my $old = read_file( $OPT{'pid'} );
		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;

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

			return _reply( $c, { error => q{mode must be "prequential", "learn", or "score"}, @$tag } );
		}
		$c->{mode} = $mode;
		return _reply( $c, { ok => { mode => $mode }, @$tag } );
	}
	if ( $cmd eq 'stats' ) {
		return _reply(
			$c,
			{
				ok => {
					seen        => 0 + $OIF->seen,
					window      => 0 + $OIF->window_count,
					n_features  => ( defined $OIF->{n_features} ? 0 + $OIF->{n_features} : undef ),
					threshold   => 0 + _threshold(),
					connections => 0 + scalar( keys %CONN ),
					dirty       => ( $DIRTY ? 1 : 0 ),
					set         => $OPT{'set'},
				},
				@$tag
			}
		);
	} ## end if ( $cmd eq 'stats' )
	if ( $cmd eq 'save' ) {
		my $name = eval { _save_model('command') };
		if ($@) {
			( my $err = $@ ) =~ s/ at \S+ line \d+\.?\s*\z//s;
			return _reply( $c, { error => $err, @$tag } );
		}
		return _reply( $c, { ok => { saved => $name }, @$tag } );
	}
	if ( $cmd eq 'relearn-threshold' ) {
		my $ok = eval { $OIF->relearn_threshold; 1 };
		if ( !$ok ) {
			( my $err = $@ ) =~ s/ at \S+ line \d+\.?\s*\z//s;
			chomp $err;
			return _reply( $c, { error => $err, @$tag } );
		}
		return _reply( $c, { ok => { threshold => 0 + $OIF->decision_threshold }, @$tag } );
	}
	return _reply( $c, { error => 'unknown cmd "' . $cmd . '"', @$tag } );
} ## end sub _handle_cmd

sub _reply {
	my ( $c, $reply ) = @_;
	$c->{outbuf} .= $JSON->encode($reply) . "\n";
	return 1;
}

# The effective label cutoff, resolved per message so relearns and the
# contamination refresh at save time take effect immediately.
sub _threshold {
	return
		  defined $OPT{'threshold'}        ? $OPT{'threshold'}
		: defined $OIF->decision_threshold ? $OIF->decision_threshold
		:                                    0.5;
}

# One row through the model.  A JSON object is a tagged row (full munger
# plan, expanding/combining mungers included); a JSON array is positional
# (scalar mungers, like stream CSV input).  Either way the final vector
# is validated numeric before it touches the model -- JSON delivers
# typed values, so anything non-numeric left after munging is a caller
# bug worth an explicit error rather than Perl's silent string-to-0.
# Returns the score, or undef in learn mode.  Croaks on any problem.
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

return 1;