Merge branch 'patch-avs-invalidate-resourcegroup-cache' into 'master'

invalidate_cache on ResourceGroup change.

See merge request noc/noc!6031

core/bioseg/policies/base.py

@@ -99,7 +99,7 @@ class BaseBioSegPolicy(object):
             self.logger.info("Moving %s from %s to %s", mo.name, src.name, dst.name)
             mo.segment = dst
             mo.save()
-            mo._reset_caches()
+            ManagedObject._reset_caches(mo.id)
             dp += mo.object_profile.level
         self.logger.info(
             "%s power is increased from %d to %d (+%d)", dst.name, dst_pwr, dst_pwr + dp, dp

inv/models/resourcegroup.py

@@ -8,6 +8,7 @@
 # Python modules
 import operator
 import threading
+import logging
 from typing import List, Union, Optional
 
 # Third-party modules
@@ -24,11 +25,14 @@ from noc.models import get_model, is_document
 from noc.core.mongo.fields import PlainReferenceField
 from noc.core.model.decorator import on_delete_check, on_save, tree
 from noc.core.change.decorator import change
+from noc.core.defer import defer
 from noc.core.bi.decorator import bi_sync
 from noc.main.models.remotesystem import RemoteSystem
 from noc.main.models.label import Label
 from .technology import Technology
 
+logger = logging.getLogger(__name__)
+
 id_lock = threading.Lock()
 rx_labels_lock = threading.Lock()
 
@@ -218,15 +222,47 @@ class ResourceGroup(Document):
             and "dynamic_service_labels" in self._changed_fields
             and self.technology.service_model
         ) or (not hasattr(self, "_changed_fields") and self.dynamic_service_labels):
-            self.unset_service_group(self.technology.service_model)
-            self.add_service_group(self.technology.service_model)
+            changed_ids = []
+            self.unset_service_group(self.technology.service_model, changed_ids=changed_ids)
+            self.add_service_group(self.technology.service_model, changed_ids=changed_ids)
+            if changed_ids:
+                defer(
+                    "noc.inv.models.resourcegroup.invalidate_instance_cache",
+                    model_id=self.technology.service_model,
+                    ids=list(set(changed_ids)),
+                )
         if (
             hasattr(self, "_changed_fields")
             and "dynamic_client_labels" in self._changed_fields
             and self.technology.client_model
         ) or (not hasattr(self, "_changed_fields") and self.dynamic_client_labels):
-            self.unset_client_group(self.technology.client_model)
-            self.add_client_group(self.technology.client_model)
+            changed_ids = []
+            self.unset_client_group(self.technology.client_model, changed_ids=changed_ids)
+            self.add_client_group(self.technology.client_model, changed_ids=changed_ids)
+            if changed_ids:
+                defer(
+                    "noc.inv.models.resourcegroup.invalidate_instance_cache",
+                    model_id=self.technology.client_model,
+                    ids=list(set(changed_ids)),
+                )
+
+    @classmethod
+    def get_model_instance_ids(
+        cls, model_id: str, resource_group: str, is_client: bool = False
+    ) -> List[Union[int, str]]:
+        """
+        Getting model instance ids that setting ResourceGroup
+        :return:
+        """
+        model = get_model(model_id)
+        query = f"effective_{'client' if is_client else 'service'}_groups"
+        if is_document(model):
+            return list(model.objects.filter(**{query: resource_group}).values_list("id"))
+        return list(
+            model.objects.filter(**{f"{query}__contains": str(resource_group)}).values_list(
+                "id", flat=True
+            )
+        )
 
     @staticmethod
     def _remove_group(
@@ -301,20 +337,34 @@ class ResourceGroup(Document):
             cursor = connection.cursor()
             cursor.execute(sql, [labels])
 
-    def unset_service_group(self, model_id: str):
+    def unset_service_group(self, model_id: str, changed_ids: Optional[List[str]] = None):
+        """
+        Remove ServiceGroup from model
+        :param model_id:
+        :param changed_ids:
+        :return:
+        """
+        if changed_ids is not None:
+            changed_ids += self.get_model_instance_ids(model_id, self.id)
         self._remove_group(model_id, self.id)
 
-    def unset_client_group(self, model_id: str):
+    def unset_client_group(self, model_id: str, changed_ids: Optional[List[str]] = None):
+        if changed_ids is not None:
+            changed_ids += self.get_model_instance_ids(model_id, self.id, is_client=True)
         self._remove_group(model_id, self.id, is_client=True)
 
-    def add_service_group(self, model_id: str):
+    def add_service_group(self, model_id: str, changed_ids: Optional[List[str]] = None):
         # @todo optimize for one operation
         for ml in self.dynamic_service_labels:
             self._add_group(model_id, self.id, ml.labels)
+        if changed_ids is not None:
+            changed_ids += self.get_model_instance_ids(model_id, self.id)
 
-    def add_client_group(self, model_id: str):
+    def add_client_group(self, model_id: str, changed_ids: Optional[List[str]] = None):
         for ml in self.dynamic_service_labels:
             self._add_group(model_id, self.id, ml.labels, is_client=True)
+        if changed_ids is not None:
+            changed_ids += self.get_model_instance_ids(model_id, self.id, is_client=True)
 
     @classmethod
     def get_dynamic_service_groups(cls, labels: List[str], model: str) -> List[str]:
@@ -503,3 +553,21 @@ class ResourceGroup(Document):
                 o = model.objects.get(**q)
                 objects.add(o)
         return list(objects)
+
+
+def invalidate_instance_cache(model_id: str, ids: List[int]):
+    """
+    Defer task for invalidate instance with __reset_caches
+    :param model_id:
+    :param ids:
+    :return:
+    """
+    if not model_id:
+        return
+    logger.info(f"[{model_id}] Invalidate instances cache: {len(ids)}")
+    model = get_model(model_id)
+    if not hasattr(model, "_reset_caches"):
+        return
+
+    for o_id in ids:
+        model._reset_caches(o_id)

sa/models/managedobject.py

@@ -834,7 +834,7 @@ class ManagedObject(NOCModel):
         # Apply discovery jobs
         self.ensure_discovery_jobs()
         #
-        self._reset_caches(self.id)
+        self._reset_caches(self.id, credential=True)
         cache.delete_many(deleted_cache_keys)
         # Rebuild segment access
         if self.initial_data["id"] is None:
@@ -1407,7 +1407,7 @@ class ManagedObject(NOCModel):
             )
             # self.save()
             cache.delete("cred-%s" % self.id, version=CREDENTIAL_CACHE_VERSION)
-            self._reset_caches(self.id)
+            self._reset_caches(self.id, credential=True)
         caps = {}
         for ci in new_caps:
             cn = Capability.get_by_id(ci["capability"])
@@ -1909,7 +1909,7 @@ class ManagedObject(NOCModel):
         return self.object_profile.snmp_rate_limit
 
     @classmethod
-    def _reset_caches(cls, mo_id: int):
+    def _reset_caches(cls, mo_id: int, credential: bool = False):
         try:
             del cls._id_cache[f"managedobject-id-{mo_id}"]
         except KeyError:
@@ -1919,7 +1919,8 @@ class ManagedObject(NOCModel):
         except KeyError:
             pass
         cache.delete(f"managedobject-id-{mo_id}", version=MANAGEDOBJECT_CACHE_VERSION)
-        cache.delete(f"cred-{mo_id}", version=CREDENTIAL_CACHE_VERSION)
+        if credential:
+            cache.delete(f"cred-{mo_id}", version=CREDENTIAL_CACHE_VERSION)
 
     @property
     def events_stream_and_partition(self) -> Tuple[str, int]: