Skip to content
Draft
Changes from all commits
Commits
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
57 changes: 36 additions & 21 deletions src/drunc/process_manager/process_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import sys
import threading
import time
from collections import Counter

from daqpytools.logging import LogHandlerConf, exceptions, setup_daq_ers_logger
from druncschema.authoriser_pb2 import ActionType, SystemType
Expand Down Expand Up @@ -170,35 +171,48 @@ def find_by_uuid(pi_list, target_uuid: str):
return pi
return None

tech_name = self.configuration.pm_type.name

#! Evaluate code for these with the dead check down below
n_dead_prev = 0
dead_processes_prev = set()

while not self.stop_event.is_set():
results = self._ps_impl(q)

n_running = sum(
1
for process in results.values
if process.status_code == ProcessInstance.StatusCode.RUNNING
)
dead_processes = {
process.uuid.uuid
for process in results.values
if process.status_code == ProcessInstance.StatusCode.DEAD
}
session_running = Counter()
session_dead = Counter()
dead_processes = set()

for process in results.values:
session = f"{tech_name}_{process.process_description.metadata.session}"

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
session = f"{tech_name}_{process.process_description.metadata.session}"
session = f"{tech_name}__{process.process_description.metadata.session}"

status = process.status_code

if status == ProcessInstance.StatusCode.RUNNING:
session_running[session] += 1

elif status == ProcessInstance.StatusCode.DEAD:
session_dead[session] += 1
dead_processes.add(process.uuid.uuid)

# merge all known sessions from both counters
all_sessions = session_running.keys() | session_dead.keys()

# For future developers, if this is still slow consider using async.
for sesh in all_sessions:
self.opmon_publisher.publish(
message=ProcessStatus(
n_running=session_running[sesh],
n_dead=session_dead[sesh],
n_session=1, #! Can we change the schema? will it affect the influxdb?
),
custom_origin={"drunc_session": sesh},
)

n_dead = len(dead_processes)
n_session = len(
{
process.process_description.metadata.session
for process in results.values
}
)
self.opmon_publisher.publish(
message=ProcessStatus(
n_running=n_running, n_dead=n_dead, n_session=n_session
),
)
if n_dead_prev < n_dead:
n_dead_prev = n_dead
#! Ask pawel whats going on with the dead process check
diff_set = dead_processes - dead_processes_prev
for diff in diff_set:
if diff in self.expected_dead_applications:
Expand All @@ -218,6 +232,7 @@ def find_by_uuid(pi_list, target_uuid: str):
"drunc.process_manager",
)
self.log.critical(err_msg, extra=self.handlerconf.ERS)
self.log.warning(err_msg)

time.sleep(interval_s)

Expand Down
Loading