Skip to content

Address concurrency issue at resource_ended_at_normalization #1501

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tobias-urdin merged 1 commit into from
May 8, 2026
+62 −50
Merged

Address concurrency issue at resource_ended_at_normalization #1501

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 62 additions & 50 deletions gnocchi/chef.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,56 +120,68 @@ def resource_ended_at_normalization(self, metric_inactive_after):
moment = moment_now - datetime.timedelta(
seconds=metric_inactive_after)

inactive_metrics = self.index.list_metrics(
attribute_filter={"<": {
"last_measure_timestamp": moment}},
resource_policy_filter={"==": {"ended_at": None}}
)

LOG.debug("Inactive metrics found for processing: [%s].",
inactive_metrics)

inactive_metrics_by_resource_id = defaultdict(list)
for metric in inactive_metrics:
resource_id = metric.resource_id
inactive_metrics_by_resource_id[resource_id].append(metric)

for resource_id in inactive_metrics_by_resource_id.keys():
if resource_id is None:
LOG.debug("We do not need to process inactive metrics that do "
"not have resource. Therefore, these metrics [%s] "
"will be considered inactive, but there is nothing "
"else we can do in this process.",
inactive_metrics_by_resource_id[resource_id])
continue
resource = self.index.get_resource(
"generic", resource_id, with_metrics=True)
resource_metrics = resource.metrics
resource_inactive_metrics = inactive_metrics_by_resource_id.get(resource_id)

all_metrics_are_inactive = True
for m in resource_metrics:
if m not in resource_inactive_metrics:
all_metrics_are_inactive = False
LOG.debug("Not all metrics of resource [%s] are inactive. "
"Metric [%s] is not inactive. The inactive "
"metrics are [%s].",
resource, m, resource_inactive_metrics)
break

if all_metrics_are_inactive:
LOG.info("All metrics [%s] of resource [%s] are inactive."
"Therefore, we will mark it as finished with an"
"ended at timestmap.", resource_metrics, resource)
if resource.ended_at is not None:
LOG.debug(
"Resource [%s] already has an ended at value.", resource)
else:
LOG.info("Marking ended at timestamp for resource "
"[%s] because all of its metrics are inactive.",
resource)
self.index.update_resource(
resource.type, resource_id, ended_at=moment_now)
processing_lock = None
try:
processing_lock = self.get_sack_lock("resource_ended_at_normalization_lock")
if not processing_lock.acquire():
LOG.debug("Cannot obtain lock for the resource ended at normalization process. This means that "
"something else is already executing this step. Therefore, we can skip the processing for "
"this cycle.")
return
inactive_metrics = self.index.list_metrics(
attribute_filter={"<": {
"last_measure_timestamp": moment}},
resource_policy_filter={"==": {"ended_at": None}}
)

LOG.debug("Inactive metrics found for processing: [%s].",
inactive_metrics)

inactive_metrics_by_resource_id = defaultdict(list)
for metric in inactive_metrics:
resource_id = metric.resource_id
inactive_metrics_by_resource_id[resource_id].append(metric)

for resource_id in inactive_metrics_by_resource_id.keys():
if resource_id is None:
LOG.debug("We do not need to process inactive metrics that do "
"not have resource. Therefore, these metrics [%s] "
"will be considered inactive, but there is nothing "
"else we can do in this process.",
inactive_metrics_by_resource_id[resource_id])
continue
resource = self.index.get_resource(
"generic", resource_id, with_metrics=True)
resource_metrics = resource.metrics
resource_inactive_metrics = inactive_metrics_by_resource_id.get(resource_id)

all_metrics_are_inactive = True
for m in resource_metrics:
if m not in resource_inactive_metrics:
all_metrics_are_inactive = False
LOG.debug("Not all metrics of resource [%s] are inactive. "
"Metric [%s] is not inactive. The inactive "
"metrics are [%s].",
resource, m, resource_inactive_metrics)
break

if all_metrics_are_inactive:
LOG.info("All metrics [%s] of resource [%s] are inactive."
"Therefore, we will mark it as finished with an"
"ended at timestmap.", resource_metrics, resource)
if resource.ended_at is not None:
LOG.debug(
"Resource [%s] already has an ended at value.", resource)
else:
LOG.info("Marking ended at timestamp for resource "
"[%s] because all of its metrics are inactive.",
resource)
self.index.update_resource(
resource.type, resource_id, ended_at=moment_now)
finally:
if processing_lock:
LOG.debug("Releasing lock for the automatic resource ended at field normalization.")
processing_lock.release()

def clean_raw_data_inactive_metrics(self):
"""Cleans metrics raw data if they are inactive.
Expand Down
Loading