meerschaum.jobs.systemd

  1#! /usr/bin/env python3
  2# vim:fenc=utf-8
  3
  4"""
  5Manage `meerschaum.jobs.Job` via `systemd`.
  6
  7`SystemdExecutor` is a façade over the native executor (crate `meerschaum-jobs`, bound as
  8`meerschaum._core.Executor`): the unit files, the `systemctl --user` calls and the job state are
  9Rust.
 10"""
 11
 12import asyncio
 13import pathlib
 14import shlex
 15from datetime import datetime
 16from functools import partial
 17
 18import meerschaum as mrsm
 19from meerschaum.jobs import Job, Executor, make_executor, _native
 20from meerschaum.utils.typing import Dict, Any, List, SuccessTuple, Union, Optional
 21from meerschaum._internal.static import STATIC_CONFIG
 22from meerschaum.utils.warnings import warn, dprint
 23
 24JOB_METADATA_CACHE_SECONDS: int = STATIC_CONFIG['api']['jobs']['metadata_cache_seconds']
 25
 26
 27@make_executor
 28class SystemdExecutor(Executor):
 29    """
 30    Execute Meerschaum jobs via `systemd`.
 31    """
 32
 33    @property
 34    def _native_executor(self):
 35        """The native executor (built once per object)."""
 36        executor = self.__dict__.get('_native_executor_object', None)
 37        if executor is None:
 38            executor = _native.core().Executor('systemd')
 39            self.__dict__['_native_executor_object'] = executor
 40        return executor
 41
 42    def _warn(self) -> None:
 43        _native.emit_warnings(self._native_executor.take_warnings())
 44
 45    def get_job_names(self, debug: bool = False) -> List[str]:
 46        """
 47        Return a list of existing jobs, including hidden ones.
 48        """
 49        return self._native_executor.get_job_names()
 50
 51    def get_job_exists(self, name: str, debug: bool = False) -> bool:
 52        """
 53        Return whether a job exists.
 54        """
 55        return self._native_executor.get_job_exists(name)
 56
 57    def get_jobs(self, debug: bool = False) -> Dict[str, Job]:
 58        """
 59        Return a dictionary of `systemd` Jobs (including hidden jobs).
 60        """
 61        return {
 62            name: Job(name, executor_keys=str(self))
 63            for name in self.get_job_names(debug=debug)
 64        }
 65
 66    def get_service_name(self, name: str, debug: bool = False) -> str:
 67        """
 68        Return a job's service name.
 69        """
 70        return self._native_executor.get_service_name(name)
 71
 72    def get_service_job_path(self, name: str, debug: bool = False) -> pathlib.Path:
 73        """
 74        Return the path to the job's service directory.
 75        """
 76        return pathlib.Path(self._native_executor.get_service_job_path(name))
 77
 78    def get_service_symlink_file_path(self, name: str, debug: bool = False) -> pathlib.Path:
 79        """
 80        Return the path to where to create the service symlink.
 81        """
 82        return pathlib.Path(self._native_executor.get_service_symlink_file_path(name))
 83
 84    def get_service_file_path(self, name: str, debug: bool = False) -> pathlib.Path:
 85        """
 86        Return the path to a job's service file.
 87        """
 88        return pathlib.Path(self._native_executor.get_service_file_path(name))
 89
 90    def get_service_logs_path(self, name: str, debug: bool = False) -> pathlib.Path:
 91        """
 92        Return the path to direct service logs to.
 93        """
 94        return pathlib.Path(self._native_executor.get_service_logs_path(name))
 95
 96    def get_socket_path(self, name: str, debug: bool = False) -> pathlib.Path:
 97        """
 98        Return the path to the FIFO file.
 99        """
100        return pathlib.Path(self._native_executor.get_socket_path(name))
101
102    def get_result_path(self, name: str, debug: bool = False) -> pathlib.Path:
103        """
104        Return the path to the result file.
105        """
106        return pathlib.Path(self._native_executor.get_result_path(name))
107
108    def get_service_file_text(self, name: str, sysargs: List[str], debug: bool = False) -> str:
109        """
110        Return the contents of the unit file.
111        """
112        return self._native_executor.get_service_file_text(name, sysargs)
113
114    def get_socket_file_text(self, name: str, debug: bool = False) -> str:
115        """
116        Return the contents of the socket file.
117        """
118        return self._native_executor.get_socket_file_text(name)
119
120    def get_hidden_job(
121        self,
122        name: str,
123        sysargs: Optional[List[str]] = None,
124        properties: Optional[Dict[str, Any]] = None,
125        debug: bool = False,
126    ):
127        """
128        Return the hidden "sister" job to store a job's parameters.
129        """
130        job = Job(
131            name,
132            sysargs,
133            executor_keys='local',
134            _properties=properties,
135            _status_hook=partial(self.get_job_status, name),
136            _result_hook=partial(self.get_job_result, name),
137            _externally_managed=True,
138        )
139        return job
140
141    def get_job_metadata(self, name: str, debug: bool = False) -> Dict[str, Any]:
142        """
143        Return metadata about a job.
144        """
145        return self._native_executor.get_job_metadata(name)
146
147    def get_job_restart(self, name: str, debug: bool = False) -> bool:
148        """
149        Return whether a job restarts.
150        """
151        return self._native_executor.get_job_restart(name)
152
153    def get_job_properties(self, name: str, debug: bool = False) -> Dict[str, Any]:
154        """
155        Return the properties for a job.
156        """
157        return self._native_executor.get_job_properties(name)
158
159    def get_job_status(self, name: str, debug: bool = False) -> str:
160        """
161        Return the job's service status.
162        """
163        return self._native_executor.get_job_status(name)
164
165    def get_job_pid(self, name: str, debug: bool = False) -> Union[int, None]:
166        """
167        Return the job's service PID.
168        """
169        return self._native_executor.get_job_pid(name)
170
171    def get_job_began(self, name: str, debug: bool = False) -> Union[str, None]:
172        """
173        Return when a job began running.
174        """
175        return self._native_executor.get_job_began(name)
176
177    def get_job_ended(self, name: str, debug: bool = False) -> Union[str, None]:
178        """
179        Return when a job stopped running.
180        """
181        return self._native_executor.get_job_ended(name)
182
183    def get_job_paused(self, name: str, debug: bool = False) -> Union[str, None]:
184        """
185        Return when a job was paused.
186        """
187        return self._native_executor.get_job_paused(name)
188
189    def get_job_result(self, name: str, debug: bool = False) -> SuccessTuple:
190        """
191        Return the job's result SuccessTuple.
192        """
193        result = self._native_executor.get_job_result(name)
194        if result is None:
195            return False, "No result available."
196        return result
197
198    def get_job_sysargs(self, name: str, debug: bool = False) -> Union[List[str], None]:
199        """
200        Return the sysargs from the service file.
201        """
202        return self._native_executor.get_job_sysargs(name)
203
204    def run_command(
205        self,
206        command_args: List[str],
207        as_output: bool = False,
208        debug: bool = False,
209    ) -> Union[SuccessTuple, str]:
210        """
211        Run a `systemd` command and return success.
212
213        Parameters
214        ----------
215        command_args: List[str]
216            The command to pass to `systemctl --user`.
217
218        as_output: bool, default False
219            If `True`, return the process stdout output.
220            Defaults to a `SuccessTuple`.
221
222        Returns
223        -------
224        A `SuccessTuple` indicating success or a str for the process output.
225        """
226        if command_args[:2] != ['systemctl', '--user']:
227            command_args = ['systemctl', '--user'] + command_args
228
229        if debug:
230            dprint(shlex.join(command_args))
231
232        return self._native_executor.run_command(list(command_args[2:]), as_output)
233
234    def get_job_stdin_file(self, name: str, debug: bool = False):
235        """
236        Return a `StdinFile` for the job.
237        """
238        from meerschaum.utils.daemon import StdinFile
239        if '_stdin_files' not in self.__dict__:
240            self._stdin_files: Dict[str, StdinFile] = {}
241
242        if name not in self._stdin_files:
243            socket_path = self.get_socket_path(name, debug=debug)
244            socket_path.parent.mkdir(parents=True, exist_ok=True)
245            self._stdin_files[name] = StdinFile(socket_path)
246
247        return self._stdin_files[name]
248
249    def create_job(
250        self,
251        name: str,
252        sysargs: List[str],
253        properties: Optional[Dict[str, Any]] = None,
254        debug: bool = False,
255    ) -> SuccessTuple:
256        """
257        Create a job as a service to be run by `systemd`.
258        """
259        result = self._native_executor.create_job(name, sysargs, properties)
260        self._warn()
261        return result
262
263    def start_job(self, name: str, debug: bool = False) -> SuccessTuple:
264        """
265        Start a job.
266        """
267        result = self._native_executor.start_job(name)
268        self._warn()
269        return result
270
271    def stop_job(self, name: str, debug: bool = False) -> SuccessTuple:
272        """
273        Stop a job.
274        """
275        result = self._native_executor.stop_job(name)
276        self._warn()
277        return result
278
279    def pause_job(self, name: str, debug: bool = False) -> SuccessTuple:
280        """
281        Pause a job.
282        """
283        result = self._native_executor.pause_job(name)
284        self._warn()
285        return result
286
287    def delete_job(self, name: str, debug: bool = False) -> SuccessTuple:
288        """
289        Delete a job's service.
290        """
291        result = self._native_executor.delete_job(name)
292        self._warn()
293        return result
294
295    def get_logs(self, name: str, debug: bool = False) -> str:
296        """
297        Return a job's journal logs.
298        """
299        return self._native_executor.get_logs(name) or ''
300
301    def get_job_stop_time(self, name: str, debug: bool = False) -> Union[datetime, None]:
302        """
303        Return a job's stop time.
304        """
305        stop_time = self._native_executor.get_job_stop_time(name)
306        return datetime.fromisoformat(stop_time) if stop_time else None
307
308    def get_job_is_blocking_on_stdin(self, name: str, debug: bool = False) -> bool:
309        """
310        Return whether a job is blocking on stdin.
311        """
312        return self._native_executor.get_job_is_blocking_on_stdin(name)
313
314    def get_job_prompt_kwargs(self, name: str, debug: bool = False) -> Dict[str, Any]:
315        """
316        Return the kwargs to the blocking prompt.
317        """
318        return self._native_executor.get_job_prompt_kwargs(name)
319
320    def get_job_rotating_file(self, name: str, debug: bool = False):
321        """
322        Return a `RotatingFile` for the job's log output.
323        """
324        from meerschaum.utils.daemon import RotatingFile
325        service_logs_path = self.get_service_logs_path(name, debug=debug)
326        return RotatingFile(service_logs_path)
327
328    async def monitor_logs_async(
329        self,
330        name: str,
331        *args,
332        debug: bool = False,
333        **kwargs
334    ):
335        """
336        Monitor a job's output.
337        """
338        job = Job(name, executor_keys=str(self))
339        kwargs.update({'debug': debug})
340        await job.monitor_logs_async(*args, **kwargs)
341
342    def monitor_logs(self, *args, **kwargs):
343        """
344        Monitor a job's output.
345        """
346        asyncio.run(self.monitor_logs_async(*args, **kwargs))