foo

host_reaper.py

@@ -24,7 +24,6 @@
 from app.queue.metrics import event_serialization_time
 from lib.handlers import ShutdownHandler
 from lib.handlers import register_shutdown
-from lib.host_delete import delete_hosts
 from lib.host_repository import find_hosts_by_staleness_reaper
 from lib.host_repository import find_hosts_sys_default_staleness
 from lib.metrics import delete_host_count
@@ -101,32 +100,63 @@ def find_hosts_to_delete(logger, session):
     return query_filters
 
 
+def find_stale_hosts_using_custom_staleness(logger, session):
+    staleness_objects = session.query(Staleness).all()
+    org_ids = []
+
+    query_filters = []
+    for staleness_obj in staleness_objects:
+        # Validate which host types for a given org_id never get deleted
+        logger.debug(f"Looking for hosts from org_id {staleness_obj.org_id} that use custom staleness")
+        org_ids.append(staleness_obj.org_id)
+        identity = create_mock_identity_with_org_id(staleness_obj.org_id)
+        query_filters.append(
+            and_(
+                (Host.org_id == staleness_obj.org_id),
+                find_hosts_by_staleness_reaper(["stale"], identity, stale_last_our=True),
+            )
+        )
+    return query_filters, org_ids
+
+
+def find_stale_hosts(logger, session):
+    query_filters, org_ids = find_stale_hosts_using_custom_staleness(logger, session)
+
+    return query_filters
+
+
 @host_reaper_fail_count.count_exceptions()
 def run(config, logger, session, event_producer, notification_event_producer, shutdown_handler, application):
     with application.app.app_context():
-        filter_hosts_to_delete = find_hosts_to_delete(logger, session)
-
-        query = session.query(Host).filter(or_(False, *filter_hosts_to_delete))
-        hosts_processed = config.host_delete_chunk_size
-        deletions_remaining = query.count()
-
-        while hosts_processed == config.host_delete_chunk_size:
-            logger.info(f"Reaper starting batch; {deletions_remaining} remaining.")
-            try:
-                events = delete_hosts(
-                    query,
-                    event_producer,
-                    notification_event_producer,
-                    config.host_delete_chunk_size,
-                    shutdown_handler.shut_down,
-                    control_rule="REAPER",
-                )
-                hosts_processed = len(list(events))
-            except InterruptedError:
-                events = []
-                hosts_processed = 0
-
-            deletions_remaining -= hosts_processed
+        # filter_hosts_to_delete = find_hosts_to_delete(logger, session)
+        #
+        # query = session.query(Host).filter(or_(False, *filter_hosts_to_delete))
+        # hosts_processed = config.host_delete_chunk_size
+        # deletions_remaining = query.count()
+        #
+        # while hosts_processed == config.host_delete_chunk_size:
+        #    logger.info(f"Reaper starting batch; {deletions_remaining} remaining.")
+        #    try:
+        #        events = delete_hosts(
+        #            query,
+        #            event_producer,
+        #            notification_event_producer,
+        #            config.host_delete_chunk_size,
+        #            shutdown_handler.shut_down,
+        #            control_rule="REAPER",
+        #        )
+        #        hosts_processed = len(list(events))
+        #    except InterruptedError:
+        #        events = []
+        #        hosts_processed = 0
+        #
+        #    deletions_remaining -= hosts_processed
+
+        filter_stale_hosts = find_stale_hosts(logger, session)
+
+        query = session.query(Host).filter(or_(False, *filter_stale_hosts))
+        res = query.all()
+        print(res)
 
 
 def main(logger):
@@ -154,6 +184,7 @@ def main(logger):
     shutdown_handler = ShutdownHandler()
     shutdown_handler.register()
     run(config, logger, session, event_producer, notification_event_producer, shutdown_handler, application)
+    # find_stale_hosts(logger, session)
 
 
 if __name__ == "__main__":

lib/host_repository.py

@@ -1,3 +1,6 @@
+from datetime import datetime
+from datetime import timedelta
+from datetime import timezone
 from enum import Enum
 from uuid import UUID
 
@@ -211,11 +214,13 @@ def find_hosts_by_staleness(staleness_types, query, identity):
     return query.filter(or_(False, *staleness_conditions))
 
 
-def find_hosts_by_staleness_reaper(staleness_types, identity):
+def find_hosts_by_staleness_reaper(staleness_types, identity, stale_last_hour=False):
+    stale_timestamp_func = stale_last_hour_timestamp_filter if stale_last_hour else stale_timestamp_filter
+
     logger.debug("find_hosts_by_staleness(%s)", staleness_types)
     staleness_obj = serialize_staleness_to_dict(get_staleness_obj(identity=identity))
     staleness_conditions = [
-        or_(False, *staleness_to_conditions(staleness_obj, staleness_types, host_type, stale_timestamp_filter))
+        or_(False, *staleness_to_conditions(staleness_obj, staleness_types, host_type, stale_timestamp_func))
         for host_type in HOST_TYPES
     ]
 
@@ -271,6 +276,23 @@ def stale_timestamp_filter(gt=None, lte=None, host_type=None):
     return and_(*filter_, (Host.system_profile_facts["host_type"].as_string() == host_type))
 
 
+def stale_last_hour_timestamp_filter(gt=None, lte=None, host_type=None):
+    offset = timedelta(minutes=2)
+    now = datetime.now(timezone.utc)
+    last_hour = now - offset
+
+    filter_ = ()
+    if gt:
+        filter_ += (Host.modified_on > gt,)
+    if lte:
+        filter_ += (
+            Host.modified_on <= lte,
+            lte > last_hour,
+            lte <= now,
+        )
+    return and_(*filter_, (Host.system_profile_facts["host_type"].as_string() == host_type))
+
+
 def contains_no_incorrect_facts_filter(canonical_facts):
     # Does not contain any incorrect CF values
     # Incorrect value = AND( key exists, NOT( contains key:value ) )