gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1163

Browse Source 
Refactor logging statements in Ingestor and app modules to use f-strings for improved readability and consistency. Updated error and info logs to enhance clarity in the Ingestor class.
This commit is contained in:
vitor-aignosi
2025-07-16 11:17:43 -03:00
parent 5132162eaa
commit b87bfc8e68
2 changed files with 16 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ingestor/app.py
View File

@@ -21,7 +21,7 @@ def main():
except Exception as e: except Exception as e:
metrics.APP_ERRORS_TOTAL.labels( metrics.APP_ERRORS_TOTAL.labels(
pod_id=POD_ID).inc() # Increment errors pod_id=POD_ID).inc() # Increment errors
ingestor.logger.error("Failed to prepare ingestor: %s", e) ingestor.logger.error(f"Failed to prepare ingestor: {e}")
exit_signal.set() exit_signal.set()
ingestor.logger.info("Ingestor prepared. Starting main loop.") ingestor.logger.info("Ingestor prepared. Starting main loop.")

38
ingestor/ingestor.py
View File

@@ -128,7 +128,7 @@ class Ingestor:
self.ingestor_manager = IngestorManager( self.ingestor_manager = IngestorManager(
kafka_servers=self.kafka_servers, kafka_servers=self.kafka_servers,
redis_config={ redis_data={
'host': self.redis_host, 'host': self.redis_host,
'port': self.redis_port, 'port': self.redis_port,
'username': self.redis_username, 'username': self.redis_username,
@@ -151,7 +151,7 @@ class Ingestor:
# Get slot lease # Get slot lease
acquired = self.ingestor_manager.get_slot_leases() acquired = self.ingestor_manager.get_slot_leases()
self.logger.info("Acquired slots: %s", acquired) self.logger.info(f"Acquired slots: {acquired}")
self.handle_acquired_tags(acquired) self.handle_acquired_tags(acquired)
@@ -198,14 +198,14 @@ class Ingestor:
if available_slots > 0 and lacking_ingestors > 0: if available_slots > 0 and lacking_ingestors > 0:
# Some ingestors are innactive, so theres "available_slots" slots available # Some ingestors are innactive, so theres "available_slots" slots available
self.logger.info("Slots available: %s", available_slots) self.logger.info(f"Slots available: {available_slots}")
# Get slot lease # Get slot lease
self.ingestor_manager.get_slot_leases(available_slots) self.ingestor_manager.get_slot_leases(available_slots)
elif lacking_ingestors <= 0 and slot_diff > 0: elif lacking_ingestors <= 0 and slot_diff > 0:
self.logger.info("Extra slots available: %s", slot_diff) self.logger.info(f"Extra slots available: {slot_diff}")
# There's enough slots for all ingestors, but this ingestor has more than one slot # There's enough slots for all ingestors, but this ingestor has more than one slot
# So we need to drop the extra leases # So we need to drop the extra leases
@@ -230,16 +230,13 @@ class Ingestor:
""" """
self.logger.debug( self.logger.debug(
"Current managed tags: %s", self.ingestor_manager.managed_tags f"Current managed tags: {self.ingestor_manager.managed_tags}")
)
self.ingestor_manager.update_opc_servers() self.ingestor_manager.update_opc_servers()
new_managed_tags = deepcopy(self.ingestor_manager.managed_tags) new_managed_tags = deepcopy(self.ingestor_manager.managed_tags)
self.logger.debug( self.logger.debug(
"Comparing new managed tags %s with old managed tags %s", f"Comparing new managed tags {new_managed_tags} with old managed tags {old_managed_tags}"
new_managed_tags,
old_managed_tags,
) )
keys = set(new_managed_tags) | set(old_managed_tags) keys = set(new_managed_tags) | set(old_managed_tags)
@@ -247,22 +244,22 @@ class Ingestor:
changes = {k: (new_managed_tags.get(k), old_managed_tags.get(k)) changes = {k: (new_managed_tags.get(k), old_managed_tags.get(k))
for k in keys if new_managed_tags.get(k) != old_managed_tags.get(k)} for k in keys if new_managed_tags.get(k) != old_managed_tags.get(k)}
self.logger.debug("Changes: %s", changes) self.logger.debug(f"Changes: {changes}")
for slot, config in new_managed_tags.items(): for slot, config in new_managed_tags.items():
if slot not in old_managed_tags: if slot not in old_managed_tags:
self.logger.debug("Subscribing to new slot %s", slot) self.logger.debug(f"Subscribing to new slot {slot}")
self.ingestor_manager.subscribe_to_tags({slot: config}) self.ingestor_manager.subscribe_to_tags({slot: config})
continue continue
if config != old_managed_tags[slot]: if config != old_managed_tags[slot]:
self.logger.debug("Resubscribing to slot %s", slot) self.logger.debug(f"Resubscribing to slot {slot}")
self.ingestor_manager.unsubscribe_slot(slot) self.ingestor_manager.unsubscribe_slot(slot)
self.ingestor_manager.subscribe_to_tags({slot: config}) self.ingestor_manager.subscribe_to_tags({slot: config})
for slot in old_managed_tags.keys(): for slot in old_managed_tags.keys():
if slot not in new_managed_tags: if slot not in new_managed_tags:
self.logger.debug("Unsubscribing from slot %s", slot) self.logger.debug(f"Unsubscribing from slot {slot}")
self.ingestor_manager.unsubscribe_slot(slot) self.ingestor_manager.unsubscribe_slot(slot)
# Ensure the gauge is updated after any potential changes here # Ensure the gauge is updated after any potential changes here
@@ -321,16 +318,11 @@ class Ingestor:
) )
self.logger.debug( self.logger.debug(
"Active ingestors: %s, " f"Active ingestors: {ingestors}, "
"Number of slots: %s, " f"Number of slots: {number_of_slots}, "
"Number of leases: %s, " f"Number of leases: {number_of_leases}, "
"Managed tags: %s, " f"Managed tags: {self.ingestor_manager.managed_tags}, "
"Managed servers: %s", f"Managed servers: {self.ingestor_manager.opc_managers}"
ingestors,
number_of_slots,
number_of_leases,
self.ingestor_manager.managed_tags,
self.ingestor_manager.opc_managers,
) )
if not self.ingestor_manager.managed_tags: if not self.ingestor_manager.managed_tags:
# No slots acquired # No slots acquired