path: root/client/scireclient.pl

#!/usr/bin/perl

# $Id$

use strict;
use warnings;

use Scire::Job;
use Scire::Communicator;
use Getopt::Long;
use Data::Dumper;
use File::Path;
use Sys::Hostname;
use POSIX qw/WEXITSTATUS setuid/;

my $ETC_DIR = "/etc/scire";
my $SCIRE_CONFIG_FILE = "${ETC_DIR}/scire.conf";
my %conf;
my $comm;

run_main();

sub run_main {
	parse_command_line();
	my $conf_file = (defined($conf{config})) ? $conf{config} : $SCIRE_CONFIG_FILE;
	read_config_file($conf_file);

	check_job_dir();

	my $exitcode = talk_to_server();
	if($exitcode != 0) {
		if($conf{daemon}) {
			# We'll schedule another pass here later
		} else {
			debug("We couldn't communicate with the server...exiting!");
			exit(1);
		}
	}

	my @new_jobs = glob("$conf{job_dir}/queue/*.job");
	for (@new_jobs) {
		my $job = Scire::Job->new();
		$job->load_jobfile($_);
		$job->set_stdout_file("$conf{job_dir}/queue/$job->{jobid}.out");
		$job->set_stderr_file("$conf{job_dir}/queue/$job->{jobid}.err");
		$job->set_script_file("$conf{job_dir}/queue/$job->{jobid}.script");
		my $exitcode = $job->run();
		if(!$exitcode) {
			# Successful job completion
			system("mv $conf{job_dir}/queue/$job->{jobid}.* $conf{job_dir}/done/");
		} else {
			# Job failed
			system("mv $conf{job_dir}/queue/$job->{jobid}.* $conf{job_dir}/failed/");
		}
	}

	talk_to_server();
}

sub talk_to_server {
	# This functions forks a new process just for the purpose of dropping privileges.
	my $pid = fork();
	if($pid) {
		debug("Waiting for PID ${pid} to finish");
		waitpid($pid, 0);
		my $exitcode = WEXITSTATUS($?);
		debug("PID ${pid} has finished with status ${exitcode}");
		return $exitcode;
	} else {
		# We'll need to add a call to setuid() here at some point

		#ok folks so here's how this thang goes down.
		#1. Connect.
		$comm = Scire::Communicator->new( host => $conf{host}, user => $conf{user}, port => $conf{port} );
		$comm->create_connection();

		#2. Register with the DB. (only it knows if you're allowed to be active)
		#   If we do not have a defined key file, we assume this is the first run of this client
		#   so we register them instead of trying to identify.
		if(defined($conf{key_file}) and (-f $conf{key_file})) {
			if(!identify_client()) {
				exit(1);
			}
		} else {
			register_client();
			exit(0);
		}

		#3. Scan the jobs directory.  If there are done/failed jobs, report them.  Note jobs in running or queue.
		my @existing_jobs = scan_jobs_dir();
		#4. Fetch the jobs list
		get_jobs();
		#5. ???
		#6. Profit!

		$comm->close_connection();
		exit(0);
	}
}

sub parse_command_line {
	GetOptions(
		'debug|d' => \$conf{debug},
		'daemon|D' => \$conf{daemon},
		'dry-run' => \$conf{dry_run},
		'help|h' => \$conf{help},
		'config|c=s' => \$conf{config},
		'threads|t=i' => \$conf{max_threads},
		
		#config overrides.
		'host=s' => \$conf{host},
		'port=i' => \$conf{port},
		'user|u=s' => \$conf{user},
		'server_script=s' => \$conf{server_script},
		'job_dir' => \$conf{job_dir},
	);
	if ($conf{help}) {
		print "\nusage: scireclient.pl [--debug or -d]\n\t [--dry-run]"
			."\t [--config=CONF or -c] \n\t [--threads=# or -t] \t [--help or -h] \n"
			."\t [[--host=HOST] \t [--port=PORT] \t [--user=USER or -u] \n\t"
			." [--server_script=foo.pl] \t [--job_dir=/tmp/jobs] \n";
		exit 0;
	}

}

sub check_job_dir {
	my @checkdirs = ($conf{job_dir}, "$conf{job_dir}/queue", "$conf{job_dir}/done", "$conf{job_dir}/failed", "$conf{job_dir}/run");
	for my $dir (@checkdirs) {
		if (! -d $dir) {
			print "WARNING! ${dir} does not exist...creating\n";
			mkpath( $dir, {verbose => 1, mode => 0660})
				or die("Couldn't make ${dir} w/ perms 0660: $!");
		}
	}
}

sub read_config_file {
	my $conf_file = shift;
	my %config_defaults = (
		"key_file" => "${ETC_DIR}/client_key",
		"debug" => 0,
	);
	open(FH, "< ${conf_file}") or die("Couldn't open the config file ${conf_file}: $!");
	while (<FH>) {
		chomp;
		next if /^\s*(?:#|$)/;
		if(/^\s*(.+?)\s*=\s*(.+?)\s*(?:#.*)?$/) {
			unless(defined($conf{lc($1)})) {  #Don't overwrite anything specified in cmdline
				$conf{lc($1)} = $2;
			}
		}
	}
	close(FH) or die("Couldn't close the config file ${conf_file}: $!");
	for(keys %config_defaults) {
		if(!defined $conf{$_}) {
			$conf{$_} = $config_defaults{$_};
		}
	}
}

sub register_client {
#	my $mac = "00:11:22:33:44:55";
#	my $ip = "192.168.2.3";
	my ($mac, $ip) = get_interface_info(defined $conf{interface} && $conf{interface} ? $conf{interface} : "eth0");
	my $hostname = hostname();
	my ($status, $message) = $comm->send_command("REGISTER", $mac, $ip, $hostname);
	die "Could not register client $mac w/ ip $ip and hostname $hostname. Got: $message" if (! defined $status or $status ne "OK");
	debug("Client registered. Status is pending. digest is $message");
	open(FILE, ">$conf{key_file}") or die("Couldn't open key file $conf{key_file} for writing: $!");
	print FILE "$message\n";
	close(FILE);
}

sub identify_client {
	open(FILE, $conf{key_file}) or die("Couldn't open client_key $conf{key_file}: $!");
	my $digest = <FILE>;
	chomp $digest;
	close(FILE);
	my ($status, $message) = $comm->send_command("IDENTIFY", $digest);
	unless (defined $status && $status eq "OK") {
		print "ERROR Could not identify to server: $message\n";
		return 0;
	}
	debug("Client identified");
	return 1;
}

sub get_jobs {
	my ($status, $jobs) = $comm->send_command("GET_JOBS");
	unless (defined $status && $status eq "OK") {
		print "Could not get jobs list from server: $status\n";
		return 0;
	}
	if (defined($jobs) && $jobs) {
		$jobs =~ s/\s//g;  #Remove all whitespace
		my @jobs_list = split(/,/, $jobs);
		foreach my $job (@jobs_list) {
			my ($status, $filename) = $comm->send_command("GET_JOB", $job);
			#SCP the file to $conf{job_dir}/queue/

			system("cp $filename $conf{job_dir}/queue/") and die("Can't copy file: $!");  #Temporary hack.  only works locally.
			# XXX: Modify this to fetch a file instead
			debug("Fetched job $job ");
			my ($status2,$message) = $comm->send_command("JOB_FETCHED", $job);
			unless (defined $status2 && $status2 eq "OK") {
				die("ERROR Could not signal job was fetched: $message\n");
			}

		}
		#This function doesn't actually need to do anything with the list of jobs, the executor handles that part.	
	}
}

sub scan_jobs_dir {
	#Scan the dirs for job files.
	my @existing_jobs = glob("$conf{job_dir}/queue/*.job");
	my @failed_jobs = glob("$conf{job_dir}/failed/*.job");
	my @done_jobs = glob("$conf{job_dir}/done/*.job");
	
	# XXX: this function should just scan the various job dirs, create a Scire::Job object
	# for each job found, and return a structure containing the info, so that another
	# function can act on the completed jobs

	#Report on those jobs needing reporting.
	foreach my $job_file (@failed_jobs) {
		$job_file =~ /(\d+)\.job/;
		my $jobid = $1;
		my ($status, $message) = $comm->send_command("SET_JOB_STATUS", $jobid, "Failed");
		open(FILE, $job_file) or die "Couldn't open job file $job_file: $!";
		my $job_data = join("", <FILE>);
		close(FILE);
		
	}
	#may be able to use same code as above.
	foreach my $job_file (@done_jobs) {
		$job_file =~ /(\d+)\.job/;
		my $jobid = $1;
		my ($status, $message) = $comm->send_command("SET_JOB_STATUS", $jobid, "Done");
		# XXX: Send job output
	}
	
	return @existing_jobs;
}

sub debug {
	my $msg = shift;
	if($conf{debug}) {
		print STDERR $msg."\n";
	}
}

sub get_interface_info {
	my $interface = shift;

	my $info = `/sbin/ifconfig ${interface}`;
	$info =~ /^.+HWaddr ([a-zA-Z0-9:]+).+inet addr:([0-9.]+).+$/s;
	my ($mac, $ip) = ($1, $2);
	return ($mac, $ip);
}