| 21 | |
| 22 | |
| 23 | class WalUploader(object): |
| 24 | def __init__(self, layout, creds, gpg_key_id): |
| 25 | self.layout = layout |
| 26 | self.creds = creds |
| 27 | self.gpg_key_id = gpg_key_id |
| 28 | self.blobstore = get_blobstore(layout) |
| 29 | |
| 30 | def __call__(self, segment): |
| 31 | # TODO :: Move arbitray path construction to StorageLayout Object |
| 32 | url = '{0}/wal_{1}/{2}.lzo'.format(self.layout.prefix.rstrip('/'), |
| 33 | storage.CURRENT_VERSION, |
| 34 | segment.name) |
| 35 | |
| 36 | logger.info(msg='begin archiving a file', |
| 37 | detail=('Uploading "{wal_path}" to "{url}".' |
| 38 | .format(wal_path=segment.path, url=url)), |
| 39 | structured={'action': 'push-wal', |
| 40 | 'key': url, |
| 41 | 'seg': segment.name, |
| 42 | 'prefix': self.layout.path_prefix, |
| 43 | 'state': 'begin'}) |
| 44 | |
| 45 | structured_template = {'action': 'push-wal', |
| 46 | 'key': url, |
| 47 | 'seg': segment.name, |
| 48 | 'prefix': self.layout.path_prefix} |
| 49 | |
| 50 | try: |
| 51 | # Upload and record the rate at which it happened. |
| 52 | kib_per_second = do_lzop_put(self.creds, url, segment.path, |
| 53 | self.gpg_key_id) |
| 54 | except EnvironmentError as e: |
| 55 | if not segment.explicit and e.errno == errno.ENOENT: |
| 56 | structured = dict(state='skip', **structured_template) |
| 57 | logger.info(msg='skip parallel archiving of a file', |
| 58 | detail=('The segment {0} did not exist.' |
| 59 | .format(segment.path)), |
| 60 | structured=structured) |
| 61 | else: |
| 62 | raise |
| 63 | else: |
| 64 | structured = dict(rate=str(kib_per_second), state='complete', |
| 65 | **structured_template) |
| 66 | logger.info(msg='completed archiving to a file', |
| 67 | detail=('Archiving to "{url}" complete at ' |
| 68 | '{kib_per_second}KiB/s.' |
| 69 | .format(url=url, |
| 70 | kib_per_second=kib_per_second)), |
| 71 | structured=structured) |
| 72 | |
| 73 | return segment |
| 74 | |
| 75 | |
| 76 | class PartitionUploader(object): |