OLD | NEW |
(Empty) | |
| 1 #!/usr/bin/env python |
| 2 # Copyright (c) 2006-2008 Mitch Garnaat http://garnaat.org/ |
| 3 # |
| 4 # Permission is hereby granted, free of charge, to any person obtaining a |
| 5 # copy of this software and associated documentation files (the |
| 6 # "Software"), to deal in the Software without restriction, including |
| 7 # without limitation the rights to use, copy, modify, merge, publish, dis- |
| 8 # tribute, sublicense, and/or sell copies of the Software, and to permit |
| 9 # persons to whom the Software is furnished to do so, subject to the fol- |
| 10 # lowing conditions: |
| 11 # |
| 12 # The above copyright notice and this permission notice shall be included |
| 13 # in all copies or substantial portions of the Software. |
| 14 # |
| 15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
| 16 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
| 17 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
| 18 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
| 19 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
| 20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
| 21 # IN THE SOFTWARE. |
| 22 from optparse import OptionParser |
| 23 from boto.services.servicedef import ServiceDef |
| 24 from boto.services.submit import Submitter |
| 25 from boto.services.result import ResultProcessor |
| 26 import boto |
| 27 import sys, os, StringIO |
| 28 |
| 29 class BS(object): |
| 30 |
| 31 Usage = "usage: %prog [options] config_file command" |
| 32 |
| 33 Commands = {'reset' : 'Clear input queue and output bucket', |
| 34 'submit' : 'Submit local files to the service', |
| 35 'start' : 'Start the service', |
| 36 'status' : 'Report on the status of the service buckets and queu
es', |
| 37 'retrieve' : 'Retrieve output generated by a batch', |
| 38 'batches' : 'List all batches stored in current output_domain'} |
| 39 |
| 40 def __init__(self): |
| 41 self.service_name = None |
| 42 self.parser = OptionParser(usage=self.Usage) |
| 43 self.parser.add_option("--help-commands", action="store_true", dest="hel
p_commands", |
| 44 help="provides help on the available commands") |
| 45 self.parser.add_option("-a", "--access-key", action="store", type="strin
g", |
| 46 help="your AWS Access Key") |
| 47 self.parser.add_option("-s", "--secret-key", action="store", type="strin
g", |
| 48 help="your AWS Secret Access Key") |
| 49 self.parser.add_option("-p", "--path", action="store", type="string", de
st="path", |
| 50 help="the path to local directory for submit and
retrieve") |
| 51 self.parser.add_option("-k", "--keypair", action="store", type="string",
dest="keypair", |
| 52 help="the SSH keypair used with launched instance
(s)") |
| 53 self.parser.add_option("-l", "--leave", action="store_true", dest="leave
", |
| 54 help="leave the files (don't retrieve) files duri
ng retrieve command") |
| 55 self.parser.set_defaults(leave=False) |
| 56 self.parser.add_option("-n", "--num-instances", action="store", type="st
ring", dest="num_instances", |
| 57 help="the number of launched instance(s)") |
| 58 self.parser.set_defaults(num_instances=1) |
| 59 self.parser.add_option("-i", "--ignore-dirs", action="append", type="str
ing", dest="ignore", |
| 60 help="directories that should be ignored by submi
t command") |
| 61 self.parser.add_option("-b", "--batch-id", action="store", type="string"
, dest="batch", |
| 62 help="batch identifier required by the retrieve c
ommand") |
| 63 |
| 64 def print_command_help(self): |
| 65 print '\nCommands:' |
| 66 for key in self.Commands.keys(): |
| 67 print ' %s\t\t%s' % (key, self.Commands[key]) |
| 68 |
| 69 def do_reset(self): |
| 70 iq = self.sd.get_obj('input_queue') |
| 71 if iq: |
| 72 print 'clearing out input queue' |
| 73 i = 0 |
| 74 m = iq.read() |
| 75 while m: |
| 76 i += 1 |
| 77 iq.delete_message(m) |
| 78 m = iq.read() |
| 79 print 'deleted %d messages' % i |
| 80 ob = self.sd.get_obj('output_bucket') |
| 81 ib = self.sd.get_obj('input_bucket') |
| 82 if ob: |
| 83 if ib and ob.name == ib.name: |
| 84 return |
| 85 print 'delete generated files in output bucket' |
| 86 i = 0 |
| 87 for k in ob: |
| 88 i += 1 |
| 89 k.delete() |
| 90 print 'deleted %d keys' % i |
| 91 |
| 92 def do_submit(self): |
| 93 if not self.options.path: |
| 94 self.parser.error('No path provided') |
| 95 if not os.path.exists(self.options.path): |
| 96 self.parser.error('Invalid path (%s)' % self.options.path) |
| 97 s = Submitter(self.sd) |
| 98 t = s.submit_path(self.options.path, None, self.options.ignore, None, |
| 99 None, True, self.options.path) |
| 100 print 'A total of %d files were submitted' % t[1] |
| 101 print 'Batch Identifier: %s' % t[0] |
| 102 |
| 103 def do_start(self): |
| 104 ami_id = self.sd.get('ami_id') |
| 105 instance_type = self.sd.get('instance_type', 'm1.small') |
| 106 security_group = self.sd.get('security_group', 'default') |
| 107 if not ami_id: |
| 108 self.parser.error('ami_id option is required when starting the servi
ce') |
| 109 ec2 = boto.connect_ec2() |
| 110 if not self.sd.has_section('Credentials'): |
| 111 self.sd.add_section('Credentials') |
| 112 self.sd.set('Credentials', 'aws_access_key_id', ec2.aws_access_key_i
d) |
| 113 self.sd.set('Credentials', 'aws_secret_access_key', ec2.aws_secret_a
ccess_key) |
| 114 s = StringIO.StringIO() |
| 115 self.sd.write(s) |
| 116 rs = ec2.get_all_images([ami_id]) |
| 117 img = rs[0] |
| 118 r = img.run(user_data=s.getvalue(), key_name=self.options.keypair, |
| 119 max_count=self.options.num_instances, |
| 120 instance_type=instance_type, |
| 121 security_groups=[security_group]) |
| 122 print 'Starting AMI: %s' % ami_id |
| 123 print 'Reservation %s contains the following instances:' % r.id |
| 124 for i in r.instances: |
| 125 print '\t%s' % i.id |
| 126 |
| 127 def do_status(self): |
| 128 iq = self.sd.get_obj('input_queue') |
| 129 if iq: |
| 130 print 'The input_queue (%s) contains approximately %s messages' % (i
q.id, iq.count()) |
| 131 ob = self.sd.get_obj('output_bucket') |
| 132 ib = self.sd.get_obj('input_bucket') |
| 133 if ob: |
| 134 if ib and ob.name == ib.name: |
| 135 return |
| 136 total = 0 |
| 137 for k in ob: |
| 138 total += 1 |
| 139 print 'The output_bucket (%s) contains %d keys' % (ob.name, total) |
| 140 |
| 141 def do_retrieve(self): |
| 142 if not self.options.path: |
| 143 self.parser.error('No path provided') |
| 144 if not os.path.exists(self.options.path): |
| 145 self.parser.error('Invalid path (%s)' % self.options.path) |
| 146 if not self.options.batch: |
| 147 self.parser.error('batch identifier is required for retrieve command
') |
| 148 s = ResultProcessor(self.options.batch, self.sd) |
| 149 s.get_results(self.options.path, get_file=(not self.options.leave)) |
| 150 |
| 151 def do_batches(self): |
| 152 d = self.sd.get_obj('output_domain') |
| 153 if d: |
| 154 print 'Available Batches:' |
| 155 rs = d.query("['type'='Batch']") |
| 156 for item in rs: |
| 157 print ' %s' % item.name |
| 158 else: |
| 159 self.parser.error('No output_domain specified for service') |
| 160 |
| 161 def main(self): |
| 162 self.options, self.args = self.parser.parse_args() |
| 163 if self.options.help_commands: |
| 164 self.print_command_help() |
| 165 sys.exit(0) |
| 166 if len(self.args) != 2: |
| 167 self.parser.error("config_file and command are required") |
| 168 self.config_file = self.args[0] |
| 169 self.sd = ServiceDef(self.config_file) |
| 170 self.command = self.args[1] |
| 171 if hasattr(self, 'do_%s' % self.command): |
| 172 method = getattr(self, 'do_%s' % self.command) |
| 173 method() |
| 174 else: |
| 175 self.parser.error('command (%s) not recognized' % self.command) |
| 176 |
| 177 if __name__ == "__main__": |
| 178 bs = BS() |
| 179 bs.main() |
OLD | NEW |