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 a97365e0b8..313b9e6e75 100755 --- a/debian/rules +++ b/debian/rules @@ -23,6 +23,7 @@ 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 4745144062..a3c72d1150 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