view release on metacpan or  search on metacpan

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

		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;
	$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 );