Changeset 21309 for lang/perl/Net-Amazon-HadoopEC2

Timestamp:

10/14/08 14:28:18 (5 years ago)

Author:

lopnor

Message:

lang/perl/Net-Amazon-HadoopEC2: implemented launch_*, find_*, terminate_*

Location:

lang/perl/Net-Amazon-HadoopEC2/trunk

Files:

• Changes (added)
• MANIFEST (modified) (2 diffs)
• README (added)
• lib/Net/Amazon/HadoopEC2.pm (modified) (3 diffs)
• lib/Net/Amazon/HadoopEC2/Cluster.pm (modified) (7 diffs)
• t/01_group.t (added)
• t/01_module.t (deleted)
• t/02_launch_cluster.t (added)
• t/03_find_cluster.t (added)
• t/04_launch_slave.t (added)
• t/05_terminate_cluster.t (added)

lang/perl/Net-Amazon-HadoopEC2/trunk/MANIFEST

@@ +1 @@
+Changes
 inc/Module/Install.pm
 inc/Module/Install/Base.pm
@@ -8 +9 @@
 inc/Module/Install/WriteAll.pm
 lib/Net/Amazon/HadoopEC2.pm
+lib/Net/Amazon/HadoopEC2/Cluster.pm
 Makefile.PL
 MANIFEST                        This list of files
 META.yml
+README
 t/00_compile.t
-t/01_module.t
+t/01_group.t
+t/02_launch_cluster.t
+t/03_find_cluster.t
+t/04_launch_slave.t
+t/05_terminate_cluster.t

lang/perl/Net-Amazon-HadoopEC2/trunk/lib/Net/Amazon/HadoopEC2.pm

@@ -26 +26 @@
 sub launch_cluster {
     my ($self, $args) = @_;
-    $self->_assert_group({group => $args->{name}}) or return;
+    $self->_assert_group({name => $args->{name}}) or return;
     my $cluster = Net::Amazon::HadoopEC2::Cluster->new(
         {
             ec2      => $self->ec2,
             name     => $args->{name},
-            image_id => $args->{image_id},
-            slaves   => $args->{slaves} || 2,
-            key_name => $args->{key_name} || 'gsg-keypair',
             key_file => $args->{key_file},
         }
     );
-    $cluster->launch or return;
+    $cluster->launch_cluster( 
+        { 
+            slaves => $args->{slaves} || 2 ,
+            image_id => $args->{image_id},
+            key_name => $args->{key_name} || 'gsg-keypair',
+        } 
+    ) or return;
     return $cluster;
 }
 
+sub find_cluster {
+    my ($self, $args) = @_;
+    $self->_find_group({name => $args->{name}}) or return;
+    my $cluster = Net::Amazon::HadoopEC2::Cluster->new(
+        {
+            ec2 => $self->ec2,
+            name => $args->{name},
+        }
+    );
+    $cluster->find_cluster or return;
+    return $cluster;
+}
+
 sub _assert_group {
     my ($self, $args) = @_;
-    my @groups = ($args->{group}, "$args->{group}-master");
+    unless ($self->_find_group($args)) {
+        return $self->_create_group($args);
+    }
+    return 1;
+}
+
+sub _find_group {
+    my ($self, $args) = @_;
+    my @groups = ($args->{name}, "$args->{name}-master");
     my $g = $self->ec2->describe_security_groups(
         GroupName => [ @groups ],
     );
     if (ref $g eq 'Net::Amazon::EC2::Errors') {
-        $g->errors->[0]->code eq 'InvalidGroup.NotFound' or return;
+        $g->errors->[0]->code eq 'InvalidGroup.NotFound' or die $g->errors->[0]->message;
+        return;
     } else {
-        return scalar @{$g} == 2 ? 1 : 0;
-    }
-    return $self->_create_group($args);
+        scalar @{$g} == 2 or die "$args->{name} doesn't seem to be Hadoop cluster.";
+        return 1;
+    }
 }
 
 sub _create_group {
     my ($self, $args) = @_;
-    my @groups = ($args->{group}, "$args->{group}-master");
+    my @groups = ($args->{name}, "$args->{name}-master");
     for my $target ( @groups ) {
         my $desc = 'Group for Hadoop ' . ($target =~ m{master} ? 'Master' : 'Slaves') . '.';
@@ -114 +139 @@
 sub _remove_group {
     my ($self, $args) = @_;
-    my @groups = ($args->{group}, "$args->{group}-master");
+    my @groups = ($args->{name}, "$args->{name}-master");
     my $success = 0;
     my $result;
@@ -180 +205 @@
 =item * launch_cluster($hashref)
 
+=item * find_cluster($hashref)
+
 =back
 

lang/perl/Net-Amazon-HadoopEC2/trunk/lib/Net/Amazon/HadoopEC2/Cluster.pm

@@ -1 +1 @@
 package Net::Amazon::HadoopEC2::Cluster;
 use Moose;
-use Moose::Util::TypeConstraints;
 use Net::SSH::Perl;
 use File::Spec;
 use File::Basename;
 use MIME::Base64;
+use Carp;
 
 has name     => ( is => 'ro', isa => 'Str', required => 1 );
 has ec2      => ( is => 'ro', isa => 'Net::Amazon::EC2', required => 1 );
-has image_id => ( is => 'ro', isa => 'Str', required => 1 );
-has slaves   => ( is => 'ro', isa => 'Str', required => 1 );
-has key_name => ( is => 'ro', isa => 'Str', required => 1,
-    default => 'gsg-keypair' );
-has key_file => ( is => 'ro', isa => 'Str', required => 1,
+has key_file => ( is => 'rw', isa => 'Str', required => 1, lazy => 1,
     default => "$ENV{HOME}/.ssh/id_rsa-gsg-keypair" );
-has master_instance => ( is => 'rw', isa => 'Net::Amazon::EC2::RunningInstances' );
-has slave_instances => ( is => 'rw', isa => 'ArrayRef[Net::Amazon::EC2::RunningInstances]' );
-has retry => ( is => 'ro', isa => 'Int', required => 1, default => 1 );
-has retry_interval => ( is => 'ro', isa => 'Int', default => 10 );
-has map_tasks => (is => 'rw', isa => 'Int', default => 2 );
-has reduce_tasks => (is => 'rw', isa => 'Int', default => 2 );
-has compress => (is => 'rw', isa => 'Str', default => 'false' );
+has master_instance => ( is => 'rw', isa => 'Maybe[Net::Amazon::EC2::RunningInstances]');
+has slave_instances => ( is => 'rw', isa => 'ArrayRef[Net::Amazon::EC2::RunningInstances]', default => sub { return [] } );
+
+has retry => ( is => 'rw', isa => 'Int', required => 1, default => 1 );
+has retry_interval => ( is => 'rw', isa => 'Int', required => 1, default => 10 );
+has map_tasks => (is => 'rw', isa => 'Int', required => 1, default => 2 );
+has reduce_tasks => (is => 'rw', isa => 'Int', required => 1, default => 2 );
+has compress => (is => 'rw', isa => 'Str', required => 1, default => 'false' );
 
 has ssh => (
@@ -50 +47 @@
 
 no Moose;
-no Moose::Util::TypeConstraints;
-
-sub launch {
-    my ($self, $args) = @_;
-    $self->launch_master($args) or return;
+
+sub find_cluster {
+    my ($self, $args) = @_;
+    my $master_group = sprintf("%s-master", $self->name);
+    my @res = @{$self->ec2->describe_instances};
+    my @master = $self->_wait_for_instances({name => $master_group}) or return;
+    scalar @master == 1 or return;
+    $self->master_instance($master[0]);
+    $self->_find_slaves;
+    return $self;
+}
+
+sub _find_slaves {
+    my ($self) = @_;
+    my @slaves = $self->_wait_for_instances({name => $self->name});
+    $self->slave_instances([ @slaves ]);
+}
+
+sub launch_cluster {
+    my ($self, $args) = @_;
+    $self->_launch_master($args) or return;
     $self->launch_slave($args) or return;
-    return 1;
-}
-
-sub launch_master {
+    return $self;
+}
+
+sub _launch_master {
     my ($self, $args) = @_;
     my $master_group = sprintf("%s-master", $self->name);
@@ -72 +85 @@
     my $result;
     $result = $self->ec2->run_instances(
-        ImageId => $self->image_id,
+        ImageId => $args->{image_id},
         MinCount => 1,
         MaxCount => 1,
-        KeyName => $self->key_name,
+        KeyName => $args->{key_name},
         SecurityGroup => $master_group,
         UserData => encode_base64($user_data_str),
@@ -81 +94 @@
     ref $result eq 'Net::Amazon::EC2::ReservationInfo' or return;
     my $master_id = $result->instances_set->[0]->instance_id;
-    while (1) {
-        $result = $self->ec2->describe_instances(
-            InstanceId => $master_id,
-        );
-        ref $result eq 'Net::Amazon::EC2::Errors' and return;
-        ref $result->[0] eq 'Net::Amazon::EC2::ReservationInfo' or return;
-        my $state = $result->[0]->instances_set->[0];
-        if ($state->instance_state->code == 16) {
-            $self->master_instance($state);
-            last;
-        }
-        $self->retry or last;
-        sleep $self->retry_interval;
-    }
-    $self->master_instance or return;
+    my ($master) = $self->_wait_for_instances({name => $master_group, instance_id => $master_id}) or return;
+    $self->master_instance($master);
     $self->push_files(
         {
@@ -103 +103 @@
     ) or return;
     $self->execute( { command => 'chmod 600 /root/.ssh/id_rsa' } )->{code} and return;
-    return 1;
+    return $self;
 }
 
 sub launch_slave {
     my ($self, $args) = @_;
+    $self->master_instance or return;
     my $user_data = {
-        MASTER_HOST => $self->master_instance->dns_name,
+        MASTER_HOST => $self->master_instance->private_dns_name,
         MAX_MAP_TASKS => $self->map_tasks,
         MAX_REDUCE_TASKS => $self->reduce_tasks,
@@ -117 +118 @@
 
     my $result = $self->ec2->run_instances(
-        ImageId => $self->image_id,
+        ImageId => $self->master_instance->image_id,
         MinCount => 1,
-        MaxCount => $self->slaves,
-        KeyName => $self->key_name,
+        MaxCount => $args->{slaves},
+        KeyName => $self->master_instance->key_name,
         SecurityGroup => $self->name,
         UserData => encode_base64($user_data_str),
     );
     ref $result eq 'Net::Amazon::EC2::ReservationInfo' or return;
-    $self->slave_instances([ $result->instances_set ]);
-    return 1;
-}
-
-sub terminate {
+    my @instances = map { $_->instance_id} @{$result->instances_set};
+    $self->_wait_for_instances({name => $self->name, instance_id => [ @instances ]});
+    $self->_find_slaves;
+    return $self;
+}
+
+sub _wait_for_instances {
+    my ($self, $args) = @_;
+    my $name = $args->{name} or croak "name not specified";
+    my $instances = $args->{instance_id} || [];
+    $instances = [ $instances ] unless ref $instances;
+    while (1) {
+        my $result = $self->ec2->describe_instances(
+            InstanceId => $instances,
+        );
+        ref $result eq 'Net::Amazon::EC2::Errors' and return;
+        ref $result->[0] eq 'Net::Amazon::EC2::ReservationInfo' or return;
+        my @found = map {@{$_->instances_set}} grep { grep {$_->group_id eq $name} @{$_->group_set}} @{$result};
+        if (grep {$_->instance_state->code == 0} @found) {
+            $self->retry or last;
+            sleep $self->retry_interval;
+        }
+        @found = grep {$_->instance_state->code == 16} @found; 
+        if (my $count_expect = scalar @{$instances}) {
+            scalar @found == $count_expect or next;
+        }
+        return @found;
+    }
+}
+
+sub terminate_cluster {
     my ($self, $args) = @_;
     my @instances = map { $_->instance_id } ($self->master_instance, @{$self->slave_instances});
-    $self->ec2->terminate_instances(
+    my $res = $self->ec2->terminate_instances(
         InstanceId => [ @instances ],
     );
+    $self->master_instance(undef);
+    $self->slave_instances( [] );
+    return $res;
+}
+
+sub terminate_slaves {
+    my ($self, $args) = @_;
+    my $existing = scalar @{$self->slave_instances};
+    my $count = $args->{slaves} || $existing;
+    $count = $existing if $count > $existing;
+    my @instances = map { $_->instance_id } @{$self->slave_instances}[ 0 .. $count - 1 ];
+    my $res = $self->ec2->terminate_instances(
+        InstanceId => [ @instances ],
+    );
+    $self->_find_slaves;
+    return $res;
 }
 
@@ -209 +252 @@
 =over 4
 
-=item * launch
-
-=item * launch_master
+=item * launch_cluster
+
+=item * find_cluster
 
 =item * launch_slave
 
-=item * terminate
+=item * terminate_cluster
+
+=item * terminate_slaves
 
 =item * execute