<commit_before>package GRNOC::Simp::Comp::Worker;

use strict;
use warnings;

### REQUIRED IMPORTS ###
use AnyEvent;
use Carp;
use Data::Dumper;
use Data::Munge qw();
use List::MoreUtils qw(any);
use Moo;
use Time::HiRes qw(gettimeofday tv_interval);
use Try::Tiny;

use GRNOC::Config;
use GRNOC::Log;
use GRNOC::RabbitMQ::Client;
use GRNOC::RabbitMQ::Dispatcher;
use GRNOC::RabbitMQ::Method;
use GRNOC::WebService::Regex;

### REQUIRED ATTRIBUTES ###

=head1 public attributes

=over 12

=item config
=item rmq_config
=item logger
=item composites
=item worker_id

=back

=cut

has config => (
    is       => 'ro',
    required => 1
);

has rmq_config => (
    is       => 'ro',
    required => 1
);

has logger => (
    is       => 'ro',
    required => 1
);

has composites => (
    is       => 'ro',
    required => 1
);

has worker_id => (
    is       => 'ro',
    required => 1
);

### INTERNAL ATTRIBUTES ###

=head2 private attributes

=over 12

=item is_running
=item dispatcher
=item rmq_client
=item do_shutdown
=item rmq_dispatcher

=back

=cut

has is_running => (
    is      => 'rwp',
    default => 0
);

has dispatcher => (
    is => 'rwp'
);

has rmq_client => (
    is => 'rwp'
);

has do_shutdown => (
    is      => 'rwp',
    default => 0
);

has rmq_dispatcher => (
    is      => 'rwp',
    default => sub { undef }
);

my %_FUNCTIONS;    # Used by _function_one_val
my %_RPN_FUNCS;    # Used by _rpn_calc

### PUBLIC METHODS ###

=head2 public_methods

=over 12

=item start

=back

=cut


=head2 start
    Starts the Comp Worker process
=cut
sub start {

    my ($self) = @_;

    $self->_set_do_shutdown(0);

    while (1) {

        # we use try catch to, react to issues such as com failure
        # when any error condition is found,
        # the reactor stops and we then reinitialize
        $self->logger->debug($self->worker_id . " restarting.");

        $self->_start();

        exit(0) if $self->do_shutdown;

        sleep 2;
    }
}


=head2 _check_rabbitmq()
    Checks the whether the RabbitMQ client and dispatcher are connected.
    If either is not connected it will return false.
=cut
sub _check_rabbitmq {

    my $self      = shift;

    my $client_conn = $self->rmq_client && $self->rmq_client->connected ? 1 : 0;
    my $dispat_conn = $self->rmq_dispatcher && $self->rmq_dispatcher->connected ? 1 : 0;

    return $client_conn && $dispat_conn ? 1 : 0;
}


=head2 _connect_rmq_client()
    Creates a RabbitMQ Client and sets it for the worker
=cut
sub _connect_rmq_client {

    my $self = shift;

    $self->logger->debug("Connecting the RabbitMQ client");

    my $rmq_client = GRNOC::RabbitMQ::Client->new(
        host           => $self->rmq_config->{'ip'},
        port           => $self->rmq_config->{'port'},
        user           => $self->rmq_config->{'user'},
        pass           => $self->rmq_config->{'password'},
        exchange       => 'Simp',
        topic          => 'Simp.Data',
        timeout        => 15,
    );
    $self->_set_rmq_client($rmq_client);
}

=head2 _connect_rmq_dispatcher()
    Creates a RabbitMQ Dispatcher and sets it for the worker
=cut
sub _connect_rmq_dispatcher {

    my $self = shift;

    $self->logger->debug("Connecting the RabbitMQ dispatcher");

    my $rmq_dispatcher = GRNOC::RabbitMQ::Dispatcher->new(
        host           => $self->rmq_config->{'ip'},
        port           => $self->rmq_config->{'port'},
        user           => $self->rmq_config->{'user'},
        pass           => $self->rmq_config->{'password'},
        exchange       => 'Simp',
        topic          => 'Simp.Comp',
        queue_name     => 'Simp.Comp',
    );
    $self->_set_rmq_dispatcher($rmq_dispatcher);
}

=head2 _setup_rabbitmq
    This will set up the RabbitMQ rmq_client and dispatcher for the worker.
    It also create all of the methods for RMQ from the composites.
=cut
sub _setup_rabbitmq {

    my ($self) = @_;

    $self->logger->debug('Setting up RabbitMQ');

    $self->_connect_rmq_client();
    $self->_connect_rmq_dispatcher();

    my %predefined_param = map { $_ => 1 } ('node', 'period', 'exclude_regexp');

    # Register every composite as a RabbitMQ method calling _get() using its parameters
    for my $composite_name (keys %{$self->composites}) {

        my $composite = $self->composites->{$composite_name};

        my $method = GRNOC::RabbitMQ::Method->new(
            name        => "$composite_name",
            async       => 1,
            callback    => sub { $self->_get($composite, @_) },
            description => "retrieve composite data of type $composite_name, we should add a descr to the config"
        );

        $method->add_input_parameter(
            name        => 'node',
            description => 'nodes to retrieve data for',
            required    => 1,
            multiple    => 1,
            pattern     => $GRNOC::WebService::Regex::TEXT
        );

        $method->add_input_parameter(
            name        => 'period',
            description => "period of time to request for the data!",
            required    => 0,
            multiple    => 0,
            pattern     => $GRNOC::WebService::Regex::ANY_NUMBER
        );

        $method->add_input_parameter(
            name        => 'exclude_regexp',
            description => 'a set of var=regexp pairs, where if scan variable var matches the regexp, we exclude it from the results',
            required    => 0,
            multiple    => 1,
            pattern     => '^([^=]+=.*)$'
        );


        # Process any input variables the composite has
        my $inputs = [];
        if (exists $composite->{'variables'}[0]{'input'}) {
            $inputs = $composite->{'variables'}[0]{'input'};
        }

        while ( my ($input, $attr) = each(%{$composite->{'inputs'}}) ) {

            unless (exists $predefined_param{$input}) {

                $method->add_input_parameter(
                    name        => $input,
                    description => "we will add description to the config file later",
                    required    => exists $attr->{'required'} ? $attr->{'required'} : 1,
                    multiple    => 1,
                    pattern     => $GRNOC::WebService::Regex::TEXT
                );
            }
        }

        $self->rmq_dispatcher->register_method($method);
    }

    $self->config->{'force_array'} = 0;

    # Create the RabbitMQ ping method
    my $ping = GRNOC::RabbitMQ::Method->new(
        name        => "ping",
        callback    => sub { $self->_ping() },
        description => "function to test latency"
    );
    $self->rmq_dispatcher->register_method($ping);

    # Begin the event loop that handles requests coming
    # in from the RabbitMQ queue
    if ($self->_check_rabbitmq()) {
        $self->logger->debug('Entering RabbitMQ event loop');
        $self->rmq_dispatcher->start_consuming();
    }
    else {
        $self->logger->error('ERROR: Simp.Comp worker could not connect to RabbitMQ, retrying...');
    }

    # If a handler calls "stop_consuming()" it removes
    # the dispatcher definition and returns
    $self->_set_rmq_dispatcher(undef);
}


=head2 _start
    This is the internal method used to start up a Worker
=cut
sub _start {

    my ($self) = @_;

    my $worker_id = $self->worker_id;

    # flag that we're running
    $self->_set_is_running(1);

    # change our process name
    $0 = "simp_comp ($worker_id) [worker]";

    # setup signal handlers
    $SIG{'TERM'} = sub {
        $self->logger->info("Received SIG TERM.");
        $self->_stop();
    };

    $SIG{'HUP'} = sub {
        $self->logger->info("Received SIG HUP.");
    };

    $self->_setup_rabbitmq();
}


### PRIVATE METHODS ###

=head2 _stop
    Stops the Comp Worker
=cut
sub _stop {

    my $self = shift;
    $self->_set_do_shutdown(1);

    my $dispatcher = $self->rmq_dispatcher;
    $dispatcher->stop_consuming() if defined($dispatcher);
}


=head2 _ping
    Used to check connectivity to RabbitMQ
=cut
sub _ping {

    my $self = shift;

    return gettimeofday();
}


=head2 _get

    ### SUMMARY ###

    This is the primary method invoked when TSDS requests data
    It will run through retrieving data from Redis, processing, and sending the result


    ### PROCESS OVERVIEW ###

    Processes within simp-comp work asynchronously.
    A series of callbacks and functions use the $cv[*] AnyEvent condition variables
    to signal the beginning and end of a process, triggering the next process when ready.
    
    --- Here is the process order:
        1. _get_scans -> _scan_cb / Processes the scan configs and then gets their data via callback
        2. _digest_scans          / Combines the scan data for all scans into a single OID tree
        3. _get_data  -> _data_cb / Processes the val configs and then gets their data via callback
        4. _digest_data           / Builds data objects from all val data and then extracts them from the tree into a flat array
        5. _do_conversions        / Applies functions specified in configs to their val data

    
    ### DATA STRUCTURE ###
    
    Data is accumulated in a global %results hash:
    
    --- Results from the _get_scans -> _scan_cb -> _digest_scans phase:
    $results{'scan_tree'}{$node}{$scan_id} = { OID element tree returned from scans with empty leaves }
    $results{'scan_vals'}{$node}{$scan_id} = { OID element tree returned from scans with hashes of their values }
    
    This is from legacy code, no performance is gained with it. Leaving if hidden meaning is revealed...
    $results{'scan_exclude'}{$node}{$oid} = 1 (Any value here marks an oid to be excluded)
    
    --- Results from the _get_data -> _data_cb -> _digest_data phase:
    $results{'data'}{$host}{$val_id} = { OID element tree of hashes containting data returned for the val }
    $results{'hostvar'}{$host}{$hostvar_name} = The host variables (_get_data and _hostvar_cb)
    
    --- Results from the _do_conversions phase:
    $results{'final'}{$host} = [ Array of all data objects for the host that is passed back to the caller ]

=cut
sub _get {

    my $start     = [gettimeofday];
    my $self      = shift;
    my $composite = shift;
    my $rpc_ref   = shift;
    my $params    = shift;

    if (!defined($params->{'period'}{'value'})) {
        $params->{'period'}{'value'} = 60;
    }

    # Initialize the results hash for _get to pass from method to method
    my %results;

    # Make sure the final results hash exists, even if we get zero results
    $results{'final'} = {};

    # Initialize the variables map and add any inputs to it
    $results{'var_map'} = {};

    # Process any input variables the composite has
    if ($composite->{'inputs'}) {
        while ( my ($input, $attr) = each(%{$composite->{'inputs'}}) ) {
            $results{'var_map'}{$attr->{'value'}} = $input;
        }
    }

    # Initialize any configured constants for the composite
    $results{'constants'} = {};
    if ($composite->{'constants'}) {
        while ( my ($const, $value) = each(%{$composite->{'constants'}}) ) {
            $results{'constants'}{$const} = $value;
        }
    }

    my $success_callback = $rpc_ref->{'success_callback'};

    # Initialize the AnyEvent condition vars for the processing pipeline
    my @cv = map { AnyEvent->condvar; } (0 .. 5);

    # Step through data processing asynchronously using AnyEvent
    $cv[0]->begin(sub { $self->_get_scans($composite, $params, \%results, $cv[1]); });
    $cv[1]->begin(sub { $self->_digest_scans($params, \%results, $cv[2]); });
    $cv[2]->begin(sub { $self->_get_data($composite, $params, \%results, $cv[3]); });
    $cv[3]->begin(sub { $self->_digest_data($composite, $params, \%results, $cv[4]); });
    $cv[4]->begin(sub { $self->_do_conversions($composite, $params, \%results, $cv[5]); });
    $cv[5]->begin(
        sub {
            my $end = [gettimeofday];
            my $resp_time = tv_interval($start, $end);
            $self->logger->info("REQTIME COMP $resp_time");
            &$success_callback($results{'final'});
            undef %results;
            undef $composite;
            undef $params;
            undef @cv;
            undef $success_callback;
        }
    );

    # Reset the async pipeline:
    $cv[0]->end;
}

=head2 debug_dump
    This is a function to optimize debug logging involving calls to Data::Dumper
    It will only evaluate the calls to Dumper() and the passed variable if in debug mode
    Default behavior for logger->debug() will process data even when debug is not active
=cut
sub debug_dump {
     my $self  = shift;
     my $value = shift;

     $self->logger->debug({
        filter => \&Data::Dumper::Dumper, 
        value  => $value
     });
}


=head2 _trim_data
    Checks all branch and leaf elems of a val_tree, removing any not listed in map_tree
=cut
sub _trim_data {

    my $self     = shift;
    my $val_tree = shift;
    my $map_tree = shift;

    return if (ref($val_tree) ne ref({}));

    # Loop over the val keys at the reference root of val
    for my $key (keys %{$val_tree})
    {
        next if ($key eq 'value' || $key eq 'time');

        # Check if the key from val exists as a key in the map
        if (exists $map_tree->{$key})
        {
            # If the existing value points to another hash
            if (   ref($val_tree->{$key}) eq ref({})
                && ref($map_tree->{$key}) eq ref({}))
            {
                $self->_trim_data($val_tree->{$key}, $map_tree->{$key});
            }
        }
        else
        {
            $self->logger->debug(
                "$key not found in map_tree, removing it from val_tree!");
            delete $val_tree->{$key};
        }
    }

    return $val_tree;
}

# Transforms OID values and data into a hash tree, preserving OID element dependencies
sub _transform_oids {

    my $self = shift;
    my $oids = shift;
    my $data = shift;
    my $map  = shift;
    my $type = shift;

    # The final translation hash
    my %trans = ('vals' => {});

    my $is_scan = 0;
    if (defined $type && $type eq 'scan') {
        $trans{'blanks'} = {};
        $is_scan++;
    }
    else {
        $type = 'none';
    }

    my @legend;    # Store vars in order of parent->child

    for my $oid (@{$oids}) {

        # Get the OID's returned value from polling
        my $value = $data->{$oid};
        next if (!defined $value);

        if (!defined $map->{'first_var'}) {
            $trans{'vals'}{'static'} = $value;
            $trans{'legend'} = ['static'];

            if ($is_scan) {
                $trans{'blanks'}{'static'} = {};
            }

            next;
        }

        my @split_oid = split(/\./, $oid);

        # Check if the last var happens before the end of the data OID elements
        if (defined $map->{'last_var'} && $map->{'last_var'} < $#split_oid) {

            # Combine any tailing elements in the OID with the last var
            my $var_tail = join('.', splice(@split_oid, $map->{'last_var'}, $#split_oid));
            push(@split_oid, $var_tail);
        }

        # Make a reference point to the base of %vals
        my $ref = $trans{'vals'};
        my $blank_ref;

        # Do the same for blanks if it's for a scan
        if ($is_scan) {
            $blank_ref = $trans{'blanks'};
        }

        # Starting from the 1st var, loop over OID elements
        for (my $i = $map->{'first_var'}; $i <= $#split_oid; $i++) {

            # Get any matching var from the map's split OID
            my $var = $map->{'split_oid'}[$i];
            if (defined $var) {

                if (!exists $trans{'legend'}) {
                    push @legend, $var;
                }
            }
            else {
                next;
            }

            # Get the var's value in the polled OID
            my $oid_elem = $split_oid[$i];

            # If it's not a key at the reference point,
            # make it one and give it a val
            if (!exists $ref->{$oid_elem}) {

                # On the last key (leaf), assign the data
                if ($i == $#split_oid) {
                    $ref->{$oid_elem} = $value;
                    $ref->{$oid_elem}{'suffix'} = $oid_elem;
                }

                # Otherwise init a new hash in that key for another var
                # and set it in val results
                else {
                    $ref->{$oid_elem} = {};
                }

                # Always assign blank hashes to elems in the blanks hash tree
                if ($is_scan) {
                    $blank_ref->{$oid_elem} = {};
                }
            }

            # Switch the reference point to the elem's new hash
            $ref = $ref->{$oid_elem};

            if ($is_scan) {
                $blank_ref = $blank_ref->{$oid_elem};
            }
        }

        # Set the legend if it hasn't been
        if (!exists $trans{'legend'}) {
            $trans{'legend'} = \@legend;
        }
    }

    return \%trans;
}


# Polls for the scan elements of a composite
sub _get_scans {

    my $self      = shift;
    my $composite = shift;    # The instance XPath from the config
    my $params    = shift;    # Parameters to request
    my $results   = shift;    # Global $results hash
    my $cv        = shift;    # AnyEvent condition var (assumes it's been begin()'ed)

    $self->logger->debug("Running _get_scans");

    # Get the array of hosts from params
    my $hosts = $params->{'node'}{'value'};

    # find the set of exclude patterns, and group them by var
    my %exclude_patterns;
    for my $pattern (@{$params->{'exclude_regexp'}{'value'}}) {
        $pattern =~ /^([^=]+)=(.*)$/;
        push @{$exclude_patterns{$1}}, $2;
    }

    #--- This function will execute multiple scans in "parallel" using the begin/end approach
    #--- We use $cv to signal when all those scans are done
    #--- these should be moved to the constructor

    # Make sure several root hashes exist
    for my $host (@$hosts) {

        $results->{'scan_tree'}{$host} = {};
        $results->{'scan_vals'}{$host} = {};
        $results->{'data'}{$host}      = {};
        $results->{'hostvar'}{$host}   = {};
    }

    my $scans = $composite->{'scans'};

    while ( my ($scan, $attr) = each(%$scans) ) {

        my $suffix = $attr->{'suffix'};
        my $oid    = $attr->{'oid'};

        # Example Scan:
        # <scan oid_suffix="ifIdx" poll_value="ifName" oid="1.3.6.1.2.1.31.1.1.1.18.*" />

        # Add the scan to the var map
        $results->{'var_map'}->{$scan} = $suffix;

        # Ensure that a map exists for the scan
        if (!defined $scans->{$scan}->{'map'}) {
            $self->logger->error("ERROR: no map exists for scan \"$scan\"!");
            next;
        }

        # Get the base oid to poll, (defaults to the original OID if no vars)
        my $base_oid = $attr->{'map'}{'base_oid'};

        # Callback to get the polling data for the base oid
        # of the scan into results
        $cv->begin;
        $self->rmq_client->get(
            node           => $hosts,
            oidmatch       => $base_oid,
            async_callback => sub {
                my $data = shift;
                $self->_scan_cb($data->{'results'}, $hosts, $results, $scan, $attr);
                $cv->end;
            }
        );
    }

    $self->logger->debug("Completed _get_scans");

    # Signals _digest_scans to start when all callbacks complete
    $cv->end;
}


# Gets data for a scan for the OIDs we want to scan
# and sets scan_tree and scan_vals
sub _scan_cb {

    my $self    = shift;
    my $data    = shift;
    my $hosts   = shift;
    my $results = shift;
    my $scan    = shift;
    my $attr    = shift;

    my $scan_map    = $attr->{'map'};
    my $scan_suffix = $attr->{'suffix'};
    my $scan_oid    = $attr->{'oid'};
    my $targets     = $attr->{'targets'};
    my $excludes    = $attr->{'excludes'};

    # True = Add no results, but possibly blacklist OID values
    my $exclude_only = $attr->{'ex_only'};

    $self->logger->debug("Running _scan_cb for $scan_suffix");

    for my $host (@$hosts) {
        
        if (!$data->{$host}) {
            $self->logger->error("No scan data could be retrieved for $host in callback for $scan_oid");
            next;
        }

        # Track the OIDs we want after exclusions and targets are factored in
        my @oids;

        # Return only entries matching specified value regexps,
        # if value regexps are specified
        my $has_targets = (defined($targets) && (scalar(@$targets) > 0));

        # Check our oids and keep only the ones we want
        for my $data_oid (keys %{$data->{$host}}) {

            my $oid_value = $data->{$host}{$data_oid}{'value'};

            # Blacklist the value if it matches an exclude
            next if (any { $oid_value =~ /$_/ } @$excludes);

            # Skip the OID if the host is exclusion only and
            # is using target matches or the value matches a target
            if ($exclude_only) {
                next unless (!$has_targets || (any { $scan_oid =~ /$_/ } @$targets));
            }
            # If we didn't exclude the oid, add it to our OIDs to translate
            else {
                push @oids, $data_oid;
            }
        }

        $self->logger->debug(scalar(@oids) . " OIDs were pushed for transformation");

        # Add the data for the scan to results if we're not excluding the host
        if (!$exclude_only && scalar(@oids)) {

            # Transform the OID data into an OID tree with empty leaves to fill
            my $scan_transform = $self->_transform_oids(\@oids, $data->{$host}, $scan_map, 'scan');

            # Set scan_tree with the blank OID tree and legend from the scan
            $results->{'scan_tree'}{$host}{$scan_suffix}{'vals'}   = $scan_transform->{'blanks'};
            $results->{'scan_tree'}{$host}{$scan_suffix}{'legend'} = $scan_transform->{'legend'};

            # Set scan_vals to the tree of values from the scan
            $results->{'scan_vals'}{$host}{$scan_suffix} = $scan_transform->{'vals'};
        }
    }

    $self->logger->debug("Finished running _scan_cb for $scan_suffix");
}


# Recursively combine the OID tree for a scan with one of another scan
sub _combine_scans {

    my $self     = shift;
    my $scan     = shift;
    my $combined = shift;

    return if (!scalar(%{$scan}));

    for my $key (keys %{$scan}) {
        if (!exists $combined->{$key}) {
            $combined->{$key} = {};
        }
        else{<commit_msg>SCTASK0055424 Consolidated results hash initialization<commit_after>package GRNOC::Simp::Comp::Worker;

use strict;
use warnings;

### REQUIRED IMPORTS ###
use AnyEvent;
use Carp;
use Data::Dumper;
use Data::Munge qw();
use List::MoreUtils qw(any);
use Moo;
use Time::HiRes qw(gettimeofday tv_interval);
use Try::Tiny;

use GRNOC::Config;
use GRNOC::Log;
use GRNOC::RabbitMQ::Client;
use GRNOC::RabbitMQ::Dispatcher;
use GRNOC::RabbitMQ::Method;
use GRNOC::WebService::Regex;

### REQUIRED ATTRIBUTES ###

=head1 public attributes

=over 12

=item config
=item rmq_config
=item logger
=item composites
=item worker_id

=back

=cut

has config => (
    is       => 'ro',
    required => 1
);

has rmq_config => (
    is       => 'ro',
    required => 1
);

has logger => (
    is       => 'ro',
    required => 1
);

has composites => (
    is       => 'ro',
    required => 1
);

has worker_id => (
    is       => 'ro',
    required => 1
);

### INTERNAL ATTRIBUTES ###

=head2 private attributes

=over 12

=item is_running
=item dispatcher
=item rmq_client
=item do_shutdown
=item rmq_dispatcher

=back

=cut

has is_running => (
    is      => 'rwp',
    default => 0
);

has dispatcher => (
    is => 'rwp'
);

has rmq_client => (
    is => 'rwp'
);

has do_shutdown => (
    is      => 'rwp',
    default => 0
);

has rmq_dispatcher => (
    is      => 'rwp',
    default => sub { undef }
);

my %_FUNCTIONS;    # Used by _function_one_val
my %_RPN_FUNCS;    # Used by _rpn_calc

### PUBLIC METHODS ###

=head2 public_methods

=over 12

=item start

=back

=cut


=head2 start
    Starts the Comp Worker process
=cut
sub start {

    my ($self) = @_;

    $self->_set_do_shutdown(0);

    while (1) {

        # we use try catch to, react to issues such as com failure
        # when any error condition is found,
        # the reactor stops and we then reinitialize
        $self->logger->debug($self->worker_id . " restarting.");

        $self->_start();

        exit(0) if $self->do_shutdown;

        sleep 2;
    }
}


=head2 _check_rabbitmq()
    Checks the whether the RabbitMQ client and dispatcher are connected.
    If either is not connected it will return false.
=cut
sub _check_rabbitmq {

    my $self      = shift;

    my $client_conn = $self->rmq_client && $self->rmq_client->connected ? 1 : 0;
    my $dispat_conn = $self->rmq_dispatcher && $self->rmq_dispatcher->connected ? 1 : 0;

    return $client_conn && $dispat_conn ? 1 : 0;
}


=head2 _connect_rmq_client()
    Creates a RabbitMQ Client and sets it for the worker
=cut
sub _connect_rmq_client {

    my $self = shift;

    $self->logger->debug("Connecting the RabbitMQ client");

    my $rmq_client = GRNOC::RabbitMQ::Client->new(
        host           => $self->rmq_config->{'ip'},
        port           => $self->rmq_config->{'port'},
        user           => $self->rmq_config->{'user'},
        pass           => $self->rmq_config->{'password'},
        exchange       => 'Simp',
        topic          => 'Simp.Data',
        timeout        => 15,
    );
    $self->_set_rmq_client($rmq_client);
}

=head2 _connect_rmq_dispatcher()
    Creates a RabbitMQ Dispatcher and sets it for the worker
=cut
sub _connect_rmq_dispatcher {

    my $self = shift;

    $self->logger->debug("Connecting the RabbitMQ dispatcher");

    my $rmq_dispatcher = GRNOC::RabbitMQ::Dispatcher->new(
        host           => $self->rmq_config->{'ip'},
        port           => $self->rmq_config->{'port'},
        user           => $self->rmq_config->{'user'},
        pass           => $self->rmq_config->{'password'},
        exchange       => 'Simp',
        topic          => 'Simp.Comp',
        queue_name     => 'Simp.Comp',
    );
    $self->_set_rmq_dispatcher($rmq_dispatcher);
}

=head2 _setup_rabbitmq
    This will set up the RabbitMQ rmq_client and dispatcher for the worker.
    It also create all of the methods for RMQ from the composites.
=cut
sub _setup_rabbitmq {

    my ($self) = @_;

    $self->logger->debug('Setting up RabbitMQ');

    $self->_connect_rmq_client();
    $self->_connect_rmq_dispatcher();

    my %predefined_param = map { $_ => 1 } ('node', 'period', 'exclude_regexp');

    # Register every composite as a RabbitMQ method calling _get() using its parameters
    for my $composite_name (keys %{$self->composites}) {

        my $composite = $self->composites->{$composite_name};

        my $method = GRNOC::RabbitMQ::Method->new(
            name        => "$composite_name",
            async       => 1,
            callback    => sub { $self->_get($composite, @_) },
            description => "retrieve composite data of type $composite_name, we should add a descr to the config"
        );

        $method->add_input_parameter(
            name        => 'node',
            description => 'nodes to retrieve data for',
            required    => 1,
            multiple    => 1,
            pattern     => $GRNOC::WebService::Regex::TEXT
        );

        $method->add_input_parameter(
            name        => 'period',
            description => "period of time to request for the data!",
            required    => 0,
            multiple    => 0,
            pattern     => $GRNOC::WebService::Regex::ANY_NUMBER
        );

        $method->add_input_parameter(
            name        => 'exclude_regexp',
            description => 'a set of var=regexp pairs, where if scan variable var matches the regexp, we exclude it from the results',
            required    => 0,
            multiple    => 1,
            pattern     => '^([^=]+=.*)$'
        );


        # Process any input variables the composite has
        my $inputs = [];
        if (exists $composite->{'variables'}[0]{'input'}) {
            $inputs = $composite->{'variables'}[0]{'input'};
        }

        while ( my ($input, $attr) = each(%{$composite->{'inputs'}}) ) {

            unless (exists $predefined_param{$input}) {

                $method->add_input_parameter(
                    name        => $input,
                    description => "we will add description to the config file later",
                    required    => exists $attr->{'required'} ? $attr->{'required'} : 1,
                    multiple    => 1,
                    pattern     => $GRNOC::WebService::Regex::TEXT
                );
            }
        }

        $self->rmq_dispatcher->register_method($method);
    }

    $self->config->{'force_array'} = 0;

    # Create the RabbitMQ ping method
    my $ping = GRNOC::RabbitMQ::Method->new(
        name        => "ping",
        callback    => sub { $self->_ping() },
        description => "function to test latency"
    );
    $self->rmq_dispatcher->register_method($ping);

    # Begin the event loop that handles requests coming
    # in from the RabbitMQ queue
    if ($self->_check_rabbitmq()) {
        $self->logger->debug('Entering RabbitMQ event loop');
        $self->rmq_dispatcher->start_consuming();
    }
    else {
        $self->logger->error('ERROR: Simp.Comp worker could not connect to RabbitMQ, retrying...');
    }

    # If a handler calls "stop_consuming()" it removes
    # the dispatcher definition and returns
    $self->_set_rmq_dispatcher(undef);
}


=head2 _start
    This is the internal method used to start up a Worker
=cut
sub _start {

    my ($self) = @_;

    my $worker_id = $self->worker_id;

    # flag that we're running
    $self->_set_is_running(1);

    # change our process name
    $0 = "simp_comp ($worker_id) [worker]";

    # setup signal handlers
    $SIG{'TERM'} = sub {
        $self->logger->info("Received SIG TERM.");
        $self->_stop();
    };

    $SIG{'HUP'} = sub {
        $self->logger->info("Received SIG HUP.");
    };

    $self->_setup_rabbitmq();
}


### PRIVATE METHODS ###

=head2 _stop
    Stops the Comp Worker
=cut
sub _stop {

    my $self = shift;
    $self->_set_do_shutdown(1);

    my $dispatcher = $self->rmq_dispatcher;
    $dispatcher->stop_consuming() if defined($dispatcher);
}


=head2 _ping
    Used to check connectivity to RabbitMQ
=cut
sub _ping {

    my $self = shift;

    return gettimeofday();
}


=head2 _get

    ### SUMMARY ###

    This is the primary method invoked when TSDS requests data
    It will run through retrieving data from Redis, processing, and sending the result


    ### PROCESS OVERVIEW ###

    Processes within simp-comp work asynchronously.
    A series of callbacks and functions use the $cv[*] AnyEvent condition variables
    to signal the beginning and end of a process, triggering the next process when ready.
    
    --- Here is the process order:
        1. _get_scans -> _scan_cb / Processes the scan configs and then gets their data via callback
        2. _digest_scans          / Combines the scan data for all scans into a single OID tree
        3. _get_data  -> _data_cb / Processes the val configs and then gets their data via callback
        4. _digest_data           / Builds data objects from all val data and then extracts them from the tree into a flat array
        5. _do_conversions        / Applies functions specified in configs to their val data

    
    ### DATA STRUCTURE ###
    
    Data is accumulated in a global %results hash:
    
    --- Results from the _get_scans -> _scan_cb -> _digest_scans phase:
    $results{'scan_tree'}{$node}{$scan_id} = { OID element tree returned from scans with empty leaves }
    $results{'scan_vals'}{$node}{$scan_id} = { OID element tree returned from scans with hashes of their values }
    
    This is from legacy code, no performance is gained with it. Leaving if hidden meaning is revealed...
    $results{'scan_exclude'}{$node}{$oid} = 1 (Any value here marks an oid to be excluded)
    
    --- Results from the _get_data -> _data_cb -> _digest_data phase:
    $results{'data'}{$host}{$val_id} = { OID element tree of hashes containting data returned for the val }
    $results{'hostvar'}{$host}{$hostvar_name} = The host variables (_get_data and _hostvar_cb)
    
    --- Results from the _do_conversions phase:
    $results{'final'}{$host} = [ Array of all data objects for the host that is passed back to the caller ]

=cut
sub _get {

    my $start     = [gettimeofday];
    my $self      = shift;
    my $composite = shift;
    my $rpc_ref   = shift;
    my $args      = shift;

    # Set the period for args as a failsafe
    if (!defined($args->{'period'}{'value'})) {
        $args->{'period'}{'value'} = 60;
    }

    # Initialize the results hash for _get to pass from method to method
    my $results = $self->_init_results($args, $composite);
    $self->debug_dump($results);

    my $success_callback = $rpc_ref->{'success_callback'};

    # Initialize the AnyEvent condition vars for the processing pipeline
    my @cv = map { AnyEvent->condvar; } (0 .. 5);

    # Step through data processing asynchronously using AnyEvent
#=head2 OLD FRAMEWORK
    $cv[0]->begin(sub { $self->_get_scans($composite, $args, $results, $cv[1]); });
    $cv[1]->begin(sub { $self->_digest_scans($args, $results, $cv[2]); });
    $cv[2]->begin(sub { $self->_get_data($composite, $args, $results, $cv[3]); });
    $cv[3]->begin(sub { $self->_digest_data($composite, $args, $results, $cv[4]); });
    $cv[4]->begin(sub { $self->_do_conversions($composite, $args, $results, $cv[5]); });
    $cv[5]->begin(
        sub {
            my $end = [gettimeofday];
            my $resp_time = tv_interval($start, $end);
            $self->logger->info("REQTIME COMP $resp_time");
            &$success_callback($results->{'final'});
            undef $results;
            undef $composite;
            undef $args;
            undef @cv;
            undef $success_callback;
        }
    );
#=cut


    # Reset the async pipeline:
    $cv[0]->end;
}


=head2
    Initializes the environment hash containing all data about a single _get request.
    The environment is cleared after _get finishes running.
=cut
sub _init_results {
    my $self      = shift;
    my $args      = shift;
    my $composite = shift;

    # Create the environment hash
    my %env = (
        hosts     => $args->{node}{value},
        final     => {},
        var_map   => {},
        constants => {},
        scan_tree => {},
        scan_vals => {},
        data      => {},
        meta      => {},
        hostvar   => {}
    );

    # Map any input variables for the composite
    if ($composite->{'inputs'}) {
        while ( my ($input, $attr) = each(%{$composite->{'inputs'}}) ) {
            $env{'var_map'}{$attr->{'value'}} = $input;
        }
    }

    # Map any constants for the composite
    if ($composite->{'constants'}) {
        while ( my ($const, $value) = each(%{$composite->{'constants'}}) ) {
            $env{'constants'}{$const} = $value;
        }
    }

    # Create host-specific hash entries
    for my $host (@{$env{hosts}}) {
        $env{'scan_tree'}{$host} = {};
        $env{'scan_vals'}{$host} = {};
        $env{'data'}{$host}      = {};
        $env{'hostvar'}{$host}   = {};
    }

    # Map scan poll_value and oid_suffix names
    while ( my ($name, $attr) = each(%{$composite->{scans}}) ) {
        $env{'var_map'}{$name} = $attr->{suffix};
    }

    # Map metadata names for TSDS
    while ( my ($name, $attr) = each(%{$composite->{data}}) ) {
        $env{'meta'}{$name} = "*$name" if ($attr->{data_type} eq 'metadata');
    }

    return \%env;
}


=head2 debug_dump
    This is a function to optimize debug logging involving calls to Data::Dumper
    It will only evaluate the calls to Dumper() and the passed variable if in debug mode
    Default behavior for logger->debug() will process data even when debug is not active
=cut
sub debug_dump {
     my $self  = shift;
     my $value = shift;

     $self->logger->debug({
        filter => \&Data::Dumper::Dumper, 
        value  => $value
     });
}


=head2 _trim_data
    Checks all branch and leaf elems of a val_tree, removing any not listed in map_tree
=cut
sub _trim_data {

    my $self     = shift;
    my $val_tree = shift;
    my $map_tree = shift;

    return if (ref($val_tree) ne ref({}));

    # Loop over the val keys at the reference root of val
    for my $key (keys %{$val_tree})
    {
        next if ($key eq 'value' || $key eq 'time');

        # Check if the key from val exists as a key in the map
        if (exists $map_tree->{$key})
        {
            # If the existing value points to another hash
            if (   ref($val_tree->{$key}) eq ref({})
                && ref($map_tree->{$key}) eq ref({}))
            {
                $self->_trim_data($val_tree->{$key}, $map_tree->{$key});
            }
        }
        else
        {
            $self->logger->debug(
                "$key not found in map_tree, removing it from val_tree!");
            delete $val_tree->{$key};
        }
    }

    return $val_tree;
}

# Transforms OID values and data into a hash tree, preserving OID element dependencies
sub _transform_oids {

    my $self = shift;
    my $oids = shift;
    my $data = shift;
    my $map  = shift;
    my $type = shift;

    # The final translation hash
    my %trans = ('vals' => {});

    my $is_scan = 0;
    if (defined $type && $type eq 'scan') {
        $trans{'blanks'} = {};
        $is_scan++;
    }
    else {
        $type = 'none';
    }

    my @legend;    # Store vars in order of parent->child

    for my $oid (@{$oids}) {

        # Get the OID's returned value from polling
        my $value = $data->{$oid};
        next if (!defined $value);

        if (!defined $map->{'first_var'}) {
            $trans{'vals'}{'static'} = $value;
            $trans{'legend'} = ['static'];

            if ($is_scan) {
                $trans{'blanks'}{'static'} = {};
            }

            next;
        }

        my @split_oid = split(/\./, $oid);

        # Check if the last var happens before the end of the data OID elements
        if (defined $map->{'last_var'} && $map->{'last_var'} < $#split_oid) {

            # Combine any tailing elements in the OID with the last var
            my $var_tail = join('.', splice(@split_oid, $map->{'last_var'}, $#split_oid));
            push(@split_oid, $var_tail);
        }

        # Make a reference point to the base of %vals
        my $ref = $trans{'vals'};
        my $blank_ref;

        # Do the same for blanks if it's for a scan
        if ($is_scan) {
            $blank_ref = $trans{'blanks'};
        }

        # Starting from the 1st var, loop over OID elements
        for (my $i = $map->{'first_var'}; $i <= $#split_oid; $i++) {

            # Get any matching var from the map's split OID
            my $var = $map->{'split_oid'}[$i];
            if (defined $var) {

                if (!exists $trans{'legend'}) {
                    push @legend, $var;
                }
            }
            else {
                next;
            }

            # Get the var's value in the polled OID
            my $oid_elem = $split_oid[$i];

            # If it's not a key at the reference point,
            # make it one and give it a val
            if (!exists $ref->{$oid_elem}) {

                # On the last key (leaf), assign the data
                if ($i == $#split_oid) {
                    $ref->{$oid_elem} = $value;
                    $ref->{$oid_elem}{'suffix'} = $oid_elem;
                }

                # Otherwise init a new hash in that key for another var
                # and set it in val results
                else {
                    $ref->{$oid_elem} = {};
                }

                # Always assign blank hashes to elems in the blanks hash tree
                if ($is_scan) {
                    $blank_ref->{$oid_elem} = {};
                }
            }

            # Switch the reference point to the elem's new hash
            $ref = $ref->{$oid_elem};

            if ($is_scan) {
                $blank_ref = $blank_ref->{$oid_elem};
            }
        }

        # Set the legend if it hasn't been
        if (!exists $trans{'legend'}) {
            $trans{'legend'} = \@legend;
        }
    }

    return \%trans;
}


sub _gather_data {
    my $self = shift;
    my $composite = shift;
    my $params = shift;
    my $results = shift;
    my $cv = shift;

    $self->logger->debug('Gathering data');

    my %exclusions;
    for my $pattern (@{$params->{exclude_regexp}{value}}) {
        $pattern =~ /^([^=]+)=(.*)$/;
        push @{$exclusions{$1}}, $2;
    }


    my $scans = $composite->{scans};
    my $data  = $composite->{data};


}


# Polls for the scan elements of a composite
sub _get_scans {

    my $self      = shift;
    my $composite = shift;    # The instance XPath from the config
    my $params    = shift;    # Parameters to request
    my $results   = shift;    # Global $results hash
    my $cv        = shift;    # AnyEvent condition var (assumes it's been begin()'ed)

    $self->logger->debug("Running _get_scans");

    my $scans = $composite->{'scans'};

    while ( my ($scan, $attr) = each(%$scans) ) {

        my $suffix = $attr->{'suffix'};
        my $oid    = $attr->{'oid'};

        # Example Scan:
        # <scan oid_suffix="ifIdx" poll_value="ifName" oid="1.3.6.1.2.1.31.1.1.1.18.*" />

        # Ensure that a map exists for the scan
        if (!defined $attr->{'map'}) {
            $self->logger->error("ERROR: no map exists for scan \"$scan\"!");
            next;
        }

        # Get the base oid to poll, (defaults to the original OID if no vars)
        my $base_oid = $attr->{'map'}{'base_oid'};

        # Callback to get the polling data for the base oid
        # of the scan into results
        $cv->begin;
        $self->rmq_client->get(
            node           => $results->{hosts},
            oidmatch       => $base_oid,
            async_callback => sub {
                my $data = shift;
                $self->_scan_cb($data->{'results'}, $results->{hosts}, $results, $scan, $attr);
                $cv->end;
            }
        );
    }

    $self->logger->debug("Completed _get_scans");

    # Signals _digest_scans to start when all callbacks complete
    $cv->end;
}


# Gets data for a scan for the OIDs we want to scan
# and sets scan_tree and scan_vals
sub _scan_cb {

    my $self    = shift;
    my $data    = shift;
    my $hosts   = shift;
    my $results = shift;
    my $scan    = shift;
    my $attr    = shift;

    my $scan_map    = $attr->{'map'};
    my $scan_suffix = $attr->{'suffix'};
    my $scan_oid    = $attr->{'oid'};

    $self->logger->debug("Running _scan_cb for $scan_suffix");

    for my $host (@$hosts) {
        
        if (!$data->{$host}) {
            $self->logger->error("$host has no scan data for $scan_oid");
            next;
        }

        # Track the OIDs we want after exclusions and targets are factored in
        my @oids;

        # Only keep OIDs that have data
        for my $data_oid (keys %{$data->{$host}}) {
            my $oid_value = $data->{$host}{$data_oid}{'value'};
            push(@oids, $data_oid) if (defined($oid_value));
        }

        $self->logger->debug(scalar(@oids) . " OIDs were pushed for transformation");
        
        # Skip the host if no OIDs with data were returned
        next if (!scalar(@oids));

        # Transform the OID data into an OID tree with empty leaves to fill
        my $scan_transform = $self->_transform_oids(\@oids, $data->{$host}, $scan_map, 'scan');

        # Set scan_tree with the blank OID tree and legend from the scan
        $results->{'scan_tree'}{$host}{$scan_suffix}{'vals'}   = $scan_transform->{'blanks'};
        $results->{'scan_tree'}{$host}{$scan_suffix}{'legend'} = $scan_transform->{'legend'};

        # Set scan_vals to the tree of values from the scan
        $results->{'scan_vals'}{$host}{$scan_suffix} = $scan_transform->{'vals'};
    }

    $self->logger->debug("Finished running _scan_cb for $scan_suffix");
}


# Recursively combine the OID tree for a scan with one of another scan
sub _combine_scans {

    my $self     = shift;
    my $scan     = shift;
    my $combined = shift;

    return if (!scalar(%{$scan}));

    for my $key (keys %{$scan}) {
        if (!exists $combined->{$key}) {
            $combined->{$key} = {};
        }
        else{
{'fraction_non_alphanumeric': 0.13866660722644494, 'fraction_numerical': 0.004413436462118003, 'mean_word_length': 2.1353693479628206, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 552, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6174962', 'n_tokens_mistral': 14340, 'n_tokens_neox': 13579, 'n_words': 4126}