From dde92c9dfb810c77d3cc0458008a974547f295c4 Mon Sep 17 00:00:00 2001 From: Peter Silva Date: Sun, 15 Jun 2025 12:55:25 -0400 Subject: [PATCH] user mode systemd unit, with co-ordinated cgroups. Add option setting cgroupOverride, example usage: cgroupOverride /sys/fs/cgroup/user.slice/user-1000.slice/user@1000.service/app.slice/metpx-sr3.service this means that when sr3 starts up a process, either by analyst intervention or sr3 sanity, it will add the processes launched to this cgroup. systemD uses this group, so when starting up entire thing with systemD it will see what is already running. so systemD view and sr3 view of which processes are running should be consistent now. otoh... if systemd things the unit is stopped, it does not look at the cgroup, so it will not see things running if they are started manually. once it is started though, it does see the ones manually started before. ~ ~ --- debian/metpx-sr3.service | 23 ---------------- debian/rules | 2 +- sarracenia/config/__init__.py | 2 +- sarracenia/sr.py | 50 +++++++++++++++++++++++++++++++---- tools/metpx-sr3_user.service | 30 --------------------- 5 files changed, 47 insertions(+), 60 deletions(-) delete mode 100644 debian/metpx-sr3.service delete mode 100644 tools/metpx-sr3_user.service diff --git a/debian/metpx-sr3.service b/debian/metpx-sr3.service deleted file mode 100644 index fc54df8b3b..0000000000 --- a/debian/metpx-sr3.service +++ /dev/null @@ -1,23 +0,0 @@ -# This file provides a service definition for sarra -# It is equivalent to a /etc/init.d script when systemd is init system. -# Packager should place this file in: /usr/lib/systemd/system -# user and group 'sarra' need to exist and that account needs to -# contain the configuration for the pump (as per sarra standards) -# - -[Unit] -Description=Sarracenia File Copy Service -After=network-online.target local-fs.target remote-fs.target -Requires=network-online.target - -[Service] -Type=forking -ExecStart=/usr/bin/sr3 start -User=sarra -Group=sarra - -ExecReload=/usr/bin/sr3 reload -ExecStop=/usr/bin/sr3 stop - -[Install] -WantedBy=multi-user.target diff --git a/debian/rules b/debian/rules index 3876af6cf7..6ea0fed0f4 100755 --- a/debian/rules +++ b/debian/rules @@ -20,7 +20,7 @@ override_dh_auto_build: dh_auto_build override_dh_installinit: - dh_installinit --name=metpx-sr3 --no-start + dh_installsystemduser --name=metpx-sr3 --no-enable %: dh $@ --with python3 --buildsystem=pybuild diff --git a/sarracenia/config/__init__.py b/sarracenia/config/__init__.py index 292817f40b..a9ec7f0b92 100755 --- a/sarracenia/config/__init__.py +++ b/sarracenia/config/__init__.py @@ -193,7 +193,7 @@ def __repr__(self) -> str: 'pollUrl', 'post_baseUrl', 'post_baseDir', 'post_broker', 'post_exchange', 'post_exchangeSuffix', 'post_format', 'post_topic', 'queueName', 'queueShare', 'sendTo', 'rename', 'report_exchange', 'source', 'strip', 'timezone', 'nodupe_ttl', 'nodupe_driver', - 'nodupe_basis', 'tlsRigour', 'topic' + 'nodupe_basis', 'tlsRigour', 'topic', 'cgroupOverride' ] r""" diff --git a/sarracenia/sr.py b/sarracenia/sr.py index 10a2aa0646..d3ea59f61f 100755 --- a/sarracenia/sr.py +++ b/sarracenia/sr.py @@ -192,17 +192,41 @@ def _launch_instance(self, component_path, c, cfg, i): try: if self.configs[c][cfg]['options'].logStdout: - subprocess.Popen(cmd) + instance_proc=subprocess.Popen(cmd) else: with open(lfn, "a") as lf: - subprocess.Popen(cmd, + instance_proc=subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=lf, stderr=subprocess.STDOUT) + #print( f"launched: {cmd}" ) + except Exception as ex: print("failed to launch: %s >%s >2&1 (reason: %s) " % (' '.join(cmd), lfn, ex)) + return + + try: + cgroup=None + if hasattr(self.configs[c][cfg]['options'], 'cgroupOverride'): + cgroup=self.configs[c][cfg]['options'].cgroupOverride + elif 'cgroupDirectory' in self.states[c][cfg] and len(self.states[c][cfg]['cgroupDirectory']) > 0: + cgroup=list(self.states[c][cfg]['cgroupDirectory'])[0] + + if cgroup: + if not os.path.exists(cgroup): + os.makedirs(cgroup) + logger.critical( f"created group: {cgroup}") + + logger.critical( f"join cgroup: {cgroup}") + with open(cgroup+os.sep+"cgroup.procs", "a" ) as cgpf: + cgpf.write( f"{instance_proc.pid}") + else: + logger.critical(" No cgroup to join" ) + + except Exception as ex: + logger.critical( f"unable to join cgroup: {ex}" ) def save_procs(self, File="procs.json"): """ @@ -291,11 +315,23 @@ def _read_procs(self): return for proc in psutil.process_iter(): try: - self._filter_sr_proc( - proc.as_dict( - ['pid', 'cmdline', 'name', 'username', 'create_time', 'memory_full_info', 'cpu_times'])) + pd = proc.as_dict( ['pid', 'cmdline', 'name', 'username', 'create_time', 'memory_full_info', 'cpu_times']) + if 'MEMORY_PRESSURE_WATCH' in proc.environ(): + pd['MEMORY_PRESSURE_WATCH']= proc.environ()['MEMORY_PRESSURE_WATCH'] + pd['cgroupDirectory'] = pd['MEMORY_PRESSURE_WATCH'].replace('/memory.pressure','') + self._filter_sr_proc( pd ) + except: pass # the process went away while iterating. avoid spurious message. + + cgroupDirectories=set([]) + for p in self.procs: + if 'cgroupDirectory' in self.procs[p]: + cgroupDirectories |= set([ self.procs[p]['cgroupDirectory'] ]) + + if len(cgroupDirectories) > 1: + logger.debug( f"inconsistent linux cgroups for running flows: {cgroupDirectories}" ) + def _read_configs(self): # read in configurations. @@ -1040,6 +1076,7 @@ def _resolve(self): if ('instance_pids' in self.states[c][cfg]) and (len(self.states[c][cfg]['instance_pids']) >= 0): self.states[c][cfg]['missing_instances'] = [] self.states[c][cfg]['hung_instances'] = [] + self.states[c][cfg]['cgroupDirectory'] = set([]) observed_instances = 0 hung_instances=0 resource_usage={ 'uss': 0, 'rss': 0, 'vms':0, 'user_cpu': 0.0, 'system_cpu':0.0 } @@ -1062,6 +1099,9 @@ def _resolve(self): resource_usage[ 'system_cpu' ] += self.procs[pid]['cpu']['system'] self.resources[ 'system_cpu' ] += self.procs[pid]['cpu']['system'] + if 'cgroupDirectory' in self.procs[pid]: + self.states[c][cfg]['cgroupDirectory'] |= set( [self.procs[pid]['cgroupDirectory']] ) + if ('logAge' in self.states[c][cfg]) and (i in self.states[c][cfg]['logAge'] ) and \ ( self.states[c][cfg]['logAge'][i] > self.configs[c][cfg]['options'].runStateThreshold_hung ): hung_instances += 1 diff --git a/tools/metpx-sr3_user.service b/tools/metpx-sr3_user.service deleted file mode 100644 index 4a4fbdaa5f..0000000000 --- a/tools/metpx-sr3_user.service +++ /dev/null @@ -1,30 +0,0 @@ -# Per user service file. To run sarracenia for each user. -# this will run as long as a user is logged in. -# systemctl --user enable sarracenia -# will run the package using the user's configuration. -# reference: https://wiki.archlinux.org/index.php/Systemd/User#Basic_setup -# but when they logout, it will be ended. -# -# ways of ensuring persistence (regardless of login status) are a systemd -# configuration issue, see (loginctl enable-linger ) -# -# should be installed by a packager in: /usr/lib/systemd/user -# failing that: -# or an admin in /etc/systemd/user -# failing that: -# or by a user int ~/.config/systemd/user -# - -[Unit] -Description=Sarracenia V3 File Copy Service -After=network.target - -[Service] -Type=forking -ExecStart=/usr/bin/sr3 start - -ExecReload=/usr/bin/sr3 reload -ExecStop=/usr/bin/sr3 stop - -[Install] -WantedBy=default.target