s3: push to multiple sacks
enable s3 to push to multiple sacks and handle them accordingly Change-Id: If68b4ef0c172f824b86b120092feb936a1c2fda2
This commit is contained in:
parent
85b41d699a
commit
25cc9d7a03
|
@ -29,6 +29,9 @@ botocore = s3.botocore
|
||||||
|
|
||||||
class S3Storage(_carbonara.CarbonaraBasedStorage):
|
class S3Storage(_carbonara.CarbonaraBasedStorage):
|
||||||
|
|
||||||
|
# NOTE(gordc): override to follow s3 partitioning logic
|
||||||
|
SACK_PREFIX = '%s/'
|
||||||
|
|
||||||
def __init__(self, conf):
|
def __init__(self, conf):
|
||||||
super(S3Storage, self).__init__(conf)
|
super(S3Storage, self).__init__(conf)
|
||||||
self.s3, self._region_name, self._bucket_prefix = (
|
self.s3, self._region_name, self._bucket_prefix = (
|
||||||
|
@ -54,10 +57,9 @@ class S3Storage(_carbonara.CarbonaraBasedStorage):
|
||||||
now = datetime.datetime.utcnow().strftime("_%Y%m%d_%H:%M:%S")
|
now = datetime.datetime.utcnow().strftime("_%Y%m%d_%H:%M:%S")
|
||||||
self.s3.put_object(
|
self.s3.put_object(
|
||||||
Bucket=self._bucket_name_measures,
|
Bucket=self._bucket_name_measures,
|
||||||
Key=(six.text_type(metric.id)
|
Key=(self.get_sack_name(self.sack_for_metric(metric.id))
|
||||||
+ "/"
|
+ six.text_type(metric.id) + "/"
|
||||||
+ six.text_type(uuid.uuid4())
|
+ six.text_type(uuid.uuid4()) + now),
|
||||||
+ now),
|
|
||||||
Body=data)
|
Body=data)
|
||||||
|
|
||||||
def _build_report(self, details):
|
def _build_report(self, details):
|
||||||
|
@ -73,8 +75,9 @@ class S3Storage(_carbonara.CarbonaraBasedStorage):
|
||||||
response = self.s3.list_objects_v2(
|
response = self.s3.list_objects_v2(
|
||||||
Bucket=self._bucket_name_measures,
|
Bucket=self._bucket_name_measures,
|
||||||
**kwargs)
|
**kwargs)
|
||||||
|
# FIXME(gordc): this can be streamlined if not details
|
||||||
for c in response.get('Contents', ()):
|
for c in response.get('Contents', ()):
|
||||||
metric, metric_file = c['Key'].split("/", 1)
|
__, metric, metric_file = c['Key'].split("/", 2)
|
||||||
metric_details[metric] += 1
|
metric_details[metric] += 1
|
||||||
return (len(metric_details), sum(metric_details.values()),
|
return (len(metric_details), sum(metric_details.values()),
|
||||||
metric_details if details else None)
|
metric_details if details else None)
|
||||||
|
@ -93,14 +96,15 @@ class S3Storage(_carbonara.CarbonaraBasedStorage):
|
||||||
kwargs = {}
|
kwargs = {}
|
||||||
response = self.s3.list_objects_v2(
|
response = self.s3.list_objects_v2(
|
||||||
Bucket=self._bucket_name_measures,
|
Bucket=self._bucket_name_measures,
|
||||||
|
Prefix=self.get_sack_name(sack),
|
||||||
Delimiter="/",
|
Delimiter="/",
|
||||||
MaxKeys=limit,
|
MaxKeys=limit,
|
||||||
**kwargs)
|
**kwargs)
|
||||||
for p in response.get('CommonPrefixes', ()):
|
for p in response.get('CommonPrefixes', ()):
|
||||||
metrics.add(p['Prefix'].rstrip('/'))
|
metrics.add(p['Prefix'].split('/', 2)[1])
|
||||||
return metrics
|
return metrics
|
||||||
|
|
||||||
def _list_measure_files_for_metric_id(self, metric_id):
|
def _list_measure_files_for_metric_id(self, sack, metric_id):
|
||||||
files = set()
|
files = set()
|
||||||
response = {}
|
response = {}
|
||||||
while response.get('IsTruncated', True):
|
while response.get('IsTruncated', True):
|
||||||
|
@ -112,7 +116,8 @@ class S3Storage(_carbonara.CarbonaraBasedStorage):
|
||||||
kwargs = {}
|
kwargs = {}
|
||||||
response = self.s3.list_objects_v2(
|
response = self.s3.list_objects_v2(
|
||||||
Bucket=self._bucket_name_measures,
|
Bucket=self._bucket_name_measures,
|
||||||
Prefix=six.text_type(metric_id) + "/",
|
Prefix=(self.get_sack_name(sack)
|
||||||
|
+ six.text_type(metric_id) + "/"),
|
||||||
**kwargs)
|
**kwargs)
|
||||||
|
|
||||||
for c in response.get('Contents', ()):
|
for c in response.get('Contents', ()):
|
||||||
|
@ -121,12 +126,14 @@ class S3Storage(_carbonara.CarbonaraBasedStorage):
|
||||||
return files
|
return files
|
||||||
|
|
||||||
def delete_unprocessed_measures_for_metric_id(self, metric_id):
|
def delete_unprocessed_measures_for_metric_id(self, metric_id):
|
||||||
files = self._list_measure_files_for_metric_id(metric_id)
|
sack = self.sack_for_metric(metric_id)
|
||||||
|
files = self._list_measure_files_for_metric_id(sack, metric_id)
|
||||||
s3.bulk_delete(self.s3, self._bucket_name_measures, files)
|
s3.bulk_delete(self.s3, self._bucket_name_measures, files)
|
||||||
|
|
||||||
@contextlib.contextmanager
|
@contextlib.contextmanager
|
||||||
def process_measure_for_metric(self, metric):
|
def process_measure_for_metric(self, metric):
|
||||||
files = self._list_measure_files_for_metric_id(metric.id)
|
sack = self.sack_for_metric(metric.id)
|
||||||
|
files = self._list_measure_files_for_metric_id(sack, metric.id)
|
||||||
|
|
||||||
measures = []
|
measures = []
|
||||||
for f in files:
|
for f in files:
|
||||||
|
|
Loading…
Reference in New Issue