File Coverage

blib/lib/Net/Hadoop/YARN/ResourceManager.pm

Criterion	Covered	Total	%
statement	20	92	21.7
branch	0	44	0.0
condition	0	35	0.0
subroutine	7	16	43.7
pod			n/a
total	27	187	14.4

package Net::Hadoop::YARN::ResourceManager;
$Net::Hadoop::YARN::ResourceManager::VERSION = '0.202';
use strict;
use warnings;
use 5.10.0;

use Data::Dumper;
use Moo;
use Ref::Util qw(
    is_ref
    is_arrayref
    is_hashref
);
use Scalar::Util qw(
    refaddr
);

with 'Net::Hadoop::YARN::Roles::Common';

has '+servers' => (
    default => sub { ["localhost:8088"] },
);

has '+add_host_key' => ( default => sub { 1 } );

# After CDH 5.8.2, The RM properly issues a 302 redirect, but losing any query
# string in the original request, thus making the filters and any other parameter
# a no-op and returning huge responses including "everything" instead of a subset.
#
# eg: { queue: "root.whatever" } filter will be lost after the redirect.
#
# We can't use active_rm() in here or in the Common role, as it will trigger a
# deep recursion due to the fact that the HTTP calls are shared from there.
#
# Possibly needs to be revisited again in the future.
#
has '+no_http_redirect' => ( default => sub { 1 } );

sub active_rm {
    my $self = shift;
    my $opt  = is_hashref $_[0] ? shift @_ : {};
    my $rv;

    foreach my $server ( @{ $self->servers } ) {
        my $info = $self->info({ server => $server });
        my $haState = $info->{haState} || next;
        if ( $haState eq 'ACTIVE' ) {
            $rv = $server;
            last;
        }
    }

    if ( ! $rv ) {
        die sprintf "Failed to locate the active YARN Resource Manager from these hosts: %s",
                        join( q{, }, @{ $self->servers } ),
        ;
    }

    if ( $opt->{hostname_only} ) {
        return +( split m{[:]}xms, $rv )[0];
    }

    return $rv;
}

sub info {
    my $self = shift;
    my $opt  = is_hashref $_[0] ? shift @_ : {};

    my $res = $self->_get(
                    "cluster/info",
                    undef,
                    ( $opt->{server} or () ),
                );

    return $self->_apply_host_key(
                $res,
                $res->{clusterInfo} || $res,
            );
}

sub metrics {
    my $self = shift;
    my $opt  = is_hashref $_[0] ? shift @_ : {};
    my $res = $self->_get(
                    "cluster/metrics",
                    undef,
                    ( $opt->{server} or () ),
                );

    return $self->_apply_host_key(
                $res,
                $res->{clusterMetrics} || $res,
            );
}

sub scheduler {
    my $self = shift;
    my $res  = $self->_get("cluster/scheduler");
    return $self->_apply_host_key(
                $res,
                $res->{schedulerInfo} || $res,
            );
}

sub apps {
    my $self = shift;
    my $app_id;
    my $options;
    if ( @_ == 1 ) {
        if ( !ref $_[0] ) {
            $app_id = shift;
        }
        else {
            $options = shift;
        }
    }
    elsif ( @_ > 1 ) {
        $options = {@_};
    }
    my $res = $self->_get(
        $app_id ? "cluster/apps/$app_id" : ( "cluster/apps", { params => $options } )
    );

    return $self->_apply_host_key(
                $res,
                $res->{apps}{app} || $res->{app} || $res,
            );
}

sub appattempts {
    my $self = shift;
    my $app_id = shift or die "No app ID provided";
    my $res = $self->_get( "cluster/apps/$app_id/appattempts" );
    return $res;
}

# TODO check all states and add filter (validation)

sub appstatistics {
    my $self = shift;
    my $options;
    if ( @_ == 1 && ref $_[0] ) {
        $options = shift;
    }
    elsif ( @_ > 1 ) {
        $options = {@_};
    }
    my $res = $self->_get(
                    "cluster/appstatistics",
                    ( $options ? {
                        params => $options,
                    } : () ),
                );

    if ($res) {
        return $self->_apply_host_key(
                    $res,
                    $res->{appStatInfo}{statItem} || $res->{statItem},
                );
    }

    return;
}

sub nodes {
    my $self = shift;
    my $node_id;
    my $options;
    if ( @_ == 1 ) {
        if ( !ref $_[0] ) {
            $node_id = shift;
        }
        else {
            $options = shift;
        }
    }
    elsif ( @_ > 1 ) {
        $options = {@_};
    }
    my $res = $self->_get(
                    $node_id ? "cluster/nodes/$node_id"
                             :  (
                                    "cluster/nodes",
                                    { params => $options },
                                )
                );


    return $self->_apply_host_key(
                $res,
                $res->{nodes}{node} || $res->{node} || $res,
            );

}

sub _apply_host_key {
    my $self = shift;
    my $res  = shift;
    my $rv   = shift;


    if (   is_ref( $res )
        && is_ref( $rv )
        && ( refaddr $res eq refaddr $rv )
    ) {
        return $rv;
    }

    my $host_key  = $self->host_key;
    my $this_host = $res->{ $host_key };

    if ( is_hashref $rv ) {
        $rv->{ $host_key } = $this_host;
    }
    elsif ( is_arrayref $rv ) {
        foreach my $e ( @{ $rv } ) {
            $e->{ $host_key } = $this_host;
        }
    }
    else {
        die "Got unknown data: $rv";
    }

    return $rv;
}

1;

__END__