← Back to branch summary

~justin-fathomdb/nova/justinsb-openstack-api-volumes

« back to all changes in this revision

Viewing changes to vendor/boto/boto/services/result.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/boto/boto/services/result.py
 
1
#!/usr/bin/env python
 
2
# Copyright (c) 2006,2007 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
 
 
23
import os
 
24
from datetime import datetime, timedelta
 
25
from boto.utils import parse_ts
 
26
import boto
 
27
 
 
28
class ResultProcessor:
 
29
    
 
30
    LogFileName = 'log.csv'
 
31
 
 
32
    def __init__(self, batch_name, sd, mimetype_files=None):
 
33
        self.sd = sd
 
34
        self.batch = batch_name
 
35
        self.log_fp = None
 
36
        self.num_files = 0
 
37
        self.total_time = 0
 
38
        self.min_time = timedelta.max
 
39
        self.max_time = timedelta.min
 
40
        self.earliest_time = datetime.max
 
41
        self.latest_time = datetime.min
 
42
        self.queue = self.sd.get_obj('output_queue')
 
43
        self.domain = self.sd.get_obj('output_domain')
 
44
 
 
45
    def calculate_stats(self, msg):
 
46
        start_time = parse_ts(msg['Service-Read'])
 
47
        end_time = parse_ts(msg['Service-Write'])
 
48
        elapsed_time = end_time - start_time
 
49
        if elapsed_time > self.max_time:
 
50
            self.max_time = elapsed_time
 
51
        if elapsed_time < self.min_time:
 
52
            self.min_time = elapsed_time
 
53
        self.total_time += elapsed_time.seconds
 
54
        if start_time < self.earliest_time:
 
55
            self.earliest_time = start_time
 
56
        if end_time > self.latest_time:
 
57
            self.latest_time = end_time
 
58
 
 
59
    def log_message(self, msg, path):
 
60
        keys = msg.keys()
 
61
        keys.sort()
 
62
        if not self.log_fp:
 
63
            self.log_fp = open(os.path.join(path, self.LogFileName), 'w')
 
64
            line = ','.join(keys)
 
65
            self.log_fp.write(line+'\n')
 
66
        values = []
 
67
        for key in keys:
 
68
            value = msg[key]
 
69
            if value.find(',') > 0:
 
70
                value = '"%s"' % value
 
71
            values.append(value)
 
72
        line = ','.join(values)
 
73
        self.log_fp.write(line+'\n')
 
74
 
 
75
    def process_record(self, record, path, get_file=True):
 
76
        self.log_message(record, path)
 
77
        self.calculate_stats(record)
 
78
        outputs = record['OutputKey'].split(',')
 
79
        if record.has_key('OutputBucket'):
 
80
            bucket = boto.lookup('s3', record['OutputBucket'])
 
81
        else:
 
82
            bucket = boto.lookup('s3', record['Bucket'])
 
83
        for output in outputs:
 
84
            if get_file:
 
85
                key_name = output.split(';')[0]
 
86
                key = bucket.lookup(key_name)
 
87
                file_name = os.path.join(path, key_name)
 
88
                print 'retrieving file: %s to %s' % (key_name, file_name)
 
89
                key.get_contents_to_filename(file_name)
 
90
            self.num_files += 1
 
91
 
 
92
    def get_results_from_queue(self, path, get_file=True, delete_msg=True):
 
93
        m = self.queue.read()
 
94
        while m:
 
95
            if m.has_key('Batch') and m['Batch'] == self.batch:
 
96
                self.process_record(m, path, get_file)
 
97
                if delete_msg:
 
98
                    self.queue.delete_message(m)
 
99
            m = self.queue.read()
 
100
 
 
101
    def get_results_from_domain(self, path, get_file=True):
 
102
        rs = self.domain.query("['Batch'='%s']" % self.batch)
 
103
        for item in rs:
 
104
            self.process_record(item, path, get_file)
 
105
 
 
106
    def get_results_from_bucket(self, path):
 
107
        bucket = self.sd.get_obj('output_bucket')
 
108
        if bucket:
 
109
            print 'No output queue or domain, just retrieving files from output_bucket'
 
110
            for key in bucket:
 
111
                file_name = os.path.join(path, key)
 
112
                print 'retrieving file: %s to %s' % (key, file_name)
 
113
                key.get_contents_to_filename(file_name)
 
114
                self.num_files + 1
 
115
 
 
116
    def get_results(self, path, get_file=True, delete_msg=True):
 
117
        if not os.path.isdir(path):
 
118
            os.mkdir(path)
 
119
        if self.queue:
 
120
            self.get_results_from_queue(path, get_file)
 
121
        elif self.domain:
 
122
            self.get_results_from_domain(path, get_file)
 
123
        else:
 
124
            self.get_results_from_bucket(path)
 
125
        if self.log_fp:
 
126
            self.log_fp.close()
 
127
        print '%d results successfully retrieved.' % self.num_files
 
128
        if self.num_files > 0:
 
129
            self.avg_time = float(self.total_time)/self.num_files
 
130
            print 'Minimum Processing Time: %d' % self.min_time.seconds
 
131
            print 'Maximum Processing Time: %d' % self.max_time.seconds
 
132
            print 'Average Processing Time: %f' % self.avg_time
 
133
            self.elapsed_time = self.latest_time-self.earliest_time
 
134
            print 'Elapsed Time: %d' % self.elapsed_time.seconds
 
135
            tput = 1.0 / ((self.elapsed_time.seconds/60.0) / self.num_files)
 
136
            print 'Throughput: %f transactions / minute' % tput
 
137