Enable resource "un-delete" process

Author: rafaelweingartner

gnocchi/chef.py

@@ -21,6 +21,7 @@
 
 from gnocchi import carbonara
 from gnocchi import indexer
+from gnocchi import utils
 
 LOG = daiquiri.getLogger(__name__)
 
@@ -51,6 +52,7 @@ def __init__(self, coord, incoming, index, storage):
         self.index = index
         self.storage = storage
 
+
     def resource_ended_at_normalization(self, metric_inactive_after):
         """Marks resources as ended at if needed.
 
@@ -64,12 +66,9 @@ def resource_ended_at_normalization(self, metric_inactive_after):
         inactive, according to `metric_inactive_after` parameter. Therefore,
         we do not need to lock these metrics while processing, as they are
         inactive, and chances are that they will not receive measures anymore.
-        We will not "undelete" the resource, if it starts receiving measures
-        again. If there will ever be a case like this, we need to implement
-        some workflow to handle it.
         """
 
-        momment_now = datetime.datetime.utcnow()
+        momment_now = utils.utcnow()
         momment = momment_now - datetime.timedelta(
             seconds=metric_inactive_after)
 
@@ -91,6 +90,13 @@ def resource_ended_at_normalization(self, metric_inactive_after):
             metrics_by_resource_id[resource_id].append(metric)
 
         for resource_id in 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.",
+                          metrics_by_resource_id[resource_id])
+                continue
             resource = self.index.get_resource(
                 "generic", resource_id, with_metrics=True)
             resource_metrics = resource.metrics
@@ -101,13 +107,15 @@ def resource_ended_at_normalization(self, metric_inactive_after):
                 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.", resource, m)
+                              "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.")
+                         "ended at timestmap.", resource_metrics, resource)
                 if resource.ended_at is not None:
                     LOG.debug(
                         "Resource [%s] already has an ended at value.", resource)

gnocchi/cli/metricd.py

@@ -278,7 +278,8 @@ def _run_job(self):
         LOG.debug("Finished the cleaning of raw data points for metrics that "
                   "are no longer receiving measures.")
 
-        if self.conf.metricd.metric_inactive_after:
+        if (self.conf.metricd.metric_inactive_after and
+                self.conf.metricd.metric_inactive_after > 0):
             LOG.debug("Starting resource ended at field normalization.")
             self.chef.resource_ended_at_normalization(
                 self.conf.metricd.metric_inactive_after)

gnocchi/indexer/alembic/versions/f89ed2e3c2ec_create_last_measure_timestamp_column.py

@@ -14,18 +14,17 @@
 #
 
 """Create last measure push timestamp column
-
 Revision ID: f89ed2e3c2ec
 Revises: 18fff4509e3e
 Create Date: 2024-04-24 09:16:00
-
 """
 
 from alembic import op
 
-import datetime
 import sqlalchemy
 
+from sqlalchemy.sql import func
+
 # revision identifiers, used by Alembic.
 revision = 'f89ed2e3c2ec'
 down_revision = '18fff4509e3e'
@@ -37,4 +36,4 @@ def upgrade():
     op.add_column(
         "metric", sqlalchemy.Column(
             "last_measure_timestamp", sqlalchemy.DateTime,
-            nullable=True, default=datetime.datetime.utcnow()))
+            nullable=False, server_default=func.current_timestamp()))

gnocchi/indexer/sqlalchemy_base.py

@@ -20,12 +20,12 @@
 import sqlalchemy
 from sqlalchemy.ext import declarative
 from sqlalchemy.orm import declarative_base
+from sqlalchemy.sql import func
 
 import sqlalchemy_utils
 
 from gnocchi import archive_policy
 from gnocchi import indexer
-from gnocchi.indexer import sqlalchemy_types as types
 from gnocchi import resource_type
 from gnocchi import utils
 
@@ -120,7 +120,7 @@ class Metric(Base, GnocchiBase, indexer.Metric):
     # chances are that the resource ceased existing in the backend.
     last_measure_timestamp = sqlalchemy.Column(
         "last_measure_timestamp", sqlalchemy.DateTime,
-        nullable=True, default=datetime.datetime.utcnow())
+        nullable=False, server_default=func.current_timestamp())
 
     def jsonify(self):
         d = {
@@ -199,7 +199,7 @@ class ResourceType(Base, GnocchiBase, resource_type.ResourceType):
                                               name="resource_type_state_enum"),
                               nullable=False,
                               server_default="creating")
-    updated_at = sqlalchemy.Column(types.TimestampUTC, nullable=False,
+    updated_at = sqlalchemy.Column(indexer.types.TimestampUTC, nullable=False,
                                    # NOTE(jd): We would like to use
                                    # sqlalchemy.func.now, but we can't
                                    # because the type of PreciseTimestamp in
@@ -263,11 +263,12 @@ def type(cls):
             nullable=False)
 
     creator = sqlalchemy.Column(sqlalchemy.String(255))
-    started_at = sqlalchemy.Column(types.TimestampUTC, nullable=False,
+    started_at = sqlalchemy.Column(indexer.types.TimestampUTC, nullable=False,
                                    default=lambda: utils.utcnow())
-    revision_start = sqlalchemy.Column(types.TimestampUTC, nullable=False,
+    revision_start = sqlalchemy.Column(indexer.types.TimestampUTC,
+                                       nullable=False,
                                        default=lambda: utils.utcnow())
-    ended_at = sqlalchemy.Column(types.TimestampUTC)
+    ended_at = sqlalchemy.Column(indexer.types.TimestampUTC)
     user_id = sqlalchemy.Column(sqlalchemy.String(255))
     project_id = sqlalchemy.Column(sqlalchemy.String(255))
     original_resource_id = sqlalchemy.Column(sqlalchemy.String(255),
@@ -307,7 +308,8 @@ class ResourceHistory(ResourceMixin, Base, GnocchiBase):
                                ondelete="CASCADE",
                                name="fk_rh_id_resource_id"),
                            nullable=False)
-    revision_end = sqlalchemy.Column(types.TimestampUTC, nullable=False,
+    revision_end = sqlalchemy.Column(indexer.types.TimestampUTC,
+                                     nullable=False,
                                      default=lambda: utils.utcnow())
     metrics = sqlalchemy.orm.relationship(
         Metric, primaryjoin="Metric.resource_id == ResourceHistory.id",

gnocchi/storage/__init__.py

@@ -691,6 +691,23 @@ def _map_compute_splits_operations(bound_timeserie):
             # Mark when the metric receives its latest measures
             indexer_driver.update_last_measure_timestmap(metric.id)
 
+            resource_id = metric.resource_id
+            if resource_id:
+                resource = indexer_driver.get_resource('generic', resource_id)
+                LOG.debug("Checking if resource [%s] of metric [%s] with "
+                          "resource ID [%s] needs to be 'undeleted.'",
+                          resource, metric.id, resource_id)
+                if resource.ended_at is not None:
+                    LOG.info("Resource [%s] was marked with a timestamp for the "
+                             "'ended_at' field. However, it received a "
+                             "measurement for metric [%s]. Therefore, we undelete "
+                             "it.", resource, metric)
+                    indexer_driver.update_resource(
+                        "generic", resource_id, ended_at=None)
+            else:
+                LOG.debug("Metric [%s] does not have a resource "
+                          "assigned to it.", metric)
+
         with self.statistics.time("splits delete"):
             self._delete_metric_splits(splits_to_delete)
         self.statistics["splits delete"] += len(splits_to_delete)