meerschaum.jobs

 90class Job:
 91    """
 92    Manage a job, locally, with `systemd`, or remotely via the API.
 93    """
 94
 95    def __init__(
 96        self,
 97        name: str,
 98        sysargs: Union[List[str], str, None] = None,
 99        env: Optional[Dict[str, str]] = None,
100        executor_keys: Optional[str] = None,
101        delete_after_completion: bool = False,
102        refresh_seconds: Union[int, float, None] = None,
103        _properties: Optional[Dict[str, Any]] = None,
104        _status_hook: Optional[Callable[[], str]] = None,
105        _result_hook: Optional[Callable[[], SuccessTuple]] = None,
106        _externally_managed: bool = False,
107    ):
108        """
109        Create a new job.
110
111        Parameters
112        ----------
113        name: str
114            The name of the job to be created.
115            This will also be used as the Daemon ID.
116
117        sysargs: Union[List[str], str, None], default None
118            The sysargs of the command to be executed, e.g. 'start api'.
119
120        env: Optional[Dict[str, str]], default None
121            If provided, set these environment variables in the job's process.
122
123        executor_keys: Optional[str], default None
124            If provided, execute the job remotely on an API instance, e.g. 'api:main'
125            (or with `'systemd'`). Defaults to `'local'`, or the context's executor.
126
127        delete_after_completion: bool, default False
128            If `True`, delete this job when it has finished executing.
129
130        refresh_seconds: Union[int, float, None], default None
131            The number of seconds to sleep between refreshes.
132            Defaults to the configured value `system.cli.refresh_seconds`.
133
134        _properties: Optional[Dict[str, Any]], default None
135            If provided, use this to patch the daemon's properties.
136        """
137        native = _native.core()
138        self._core = native.Job(
139            name,
140            sysargs,
141            env=env,
142            executor_keys=executor_keys,
143            delete_after_completion=delete_after_completion,
144            refresh_seconds=refresh_seconds,
145            properties=_properties,
146            externally_managed=_externally_managed,
147            api_attributes=_native.api_attributes(executor_keys),
148        )
149        _native.emit_warnings(self._core.take_warnings())
150        self.name = name
151        self.executor_keys = self._core.executor_keys
152        self.refresh_seconds = (
153            refresh_seconds
154            if refresh_seconds is not None
155            else mrsm.get_config('system', 'cli', 'refresh_seconds')
156        )
157        if _status_hook is not None:
158            self._status_hook = _status_hook
159
160        if _result_hook is not None:
161            self._result_hook = _result_hook
162
163        self._externally_managed = _externally_managed
164
165    def _warn(self) -> None:
166        """Show the warnings the last native operation produced."""
167        _native.emit_warnings(self._core.take_warnings())
168
169    @staticmethod
170    def from_pid(pid: int, executor_keys: Optional[str] = None) -> Job:
171        """
172        Build a `Job` from the PID of a running Meerschaum process.
173
174        Parameters
175        ----------
176        pid: int
177            The PID of the process.
178
179        executor_keys: Optional[str], default None
180            The executor keys to assign to the job.
181        """
182        import os
183        try:
184            os.kill(pid, 0)
185        except ProcessLookupError as e:
186            warn(f"Process with PID {pid} does not exist.", stack=False)
187            raise e
188        except PermissionError:
189            pass
190
191        native = _native.core()
192        daemon_ids = native.jobs_daemon_ids()
193        for daemon_id in daemon_ids:
194            job = Job(daemon_id, executor_keys=executor_keys)
195            if job.pid == pid:
196                return job
197
198        cmdline_path = pathlib.Path('/proc') / str(pid) / 'cmdline'
199        if not cmdline_path.exists():
200            raise EnvironmentError(f"Is process {pid} a Meerschaum job?")
201
202        command_args = [
203            arg
204            for arg in cmdline_path.read_bytes().decode('utf-8', errors='replace').split('\0')
205            if arg
206        ]
207        # During startup the PID may change from the supervisor to its worker between
208        # the caller's PID read and the registry lookup above. Scope its persistent marker
209        # to this registry: another root may have a different Job with the same name.
210        if '--jobs-dir' in command_args:
211            import meerschaum.config.paths as paths
212            index = command_args.index('--jobs-dir') + 1
213            if (
214                index >= len(command_args)
215                or pathlib.Path(command_args[index]).resolve() != paths.DAEMON_RESOURCES_PATH.resolve()
216            ):
217                raise EnvironmentError(f"Process {pid} belongs to another Meerschaum jobs directory.")
218            for arg in command_args:
219                if arg.startswith("daemon_id='") and arg.endswith("'"):
220                    daemon_id = arg[len("daemon_id='"):-1]
221                    if daemon_id in daemon_ids:
222                        return Job(daemon_id, executor_keys=executor_keys)
223            if '__mrsm_job_supervisor__' in command_args or '__mrsm_run_daemon__' in command_args:
224                if '--daemon-id' in command_args:
225                    index = command_args.index('--daemon-id') + 1
226                    if index < len(command_args) and command_args[index] in daemon_ids:
227                        return Job(command_args[index], executor_keys=executor_keys)
228
229        if len(command_args) > 1 and command_args[1] == '-c':
230            raise EnvironmentError(f"Is process {pid} a running job?")
231
232        mrsm_ix = 0
233        for i, arg in enumerate(command_args):
234            if 'mrsm' in arg or 'meerschaum' in arg.lower():
235                mrsm_ix = i
236                break
237
238        sysargs = command_args[mrsm_ix + 1:]
239        name = None
240        for flag in ('--name', '--job-name'):
241            if flag in sysargs and sysargs.index(flag) + 1 < len(sysargs):
242                name = sysargs[sysargs.index(flag) + 1]
243                break
244        if name is None:
245            name = native.jobs_new_daemon_name()
246        return Job(name, sysargs, executor_keys=executor_keys)
247
248    def start(self, debug: bool = False) -> SuccessTuple:
249        """
250        Start the job's daemon.
251        """
252        result = self._core.start()
253        self._warn()
254        return result
255
256    def stop(
257        self,
258        timeout_seconds: Union[int, float, None] = None,
259        debug: bool = False,
260    ) -> SuccessTuple:
261        """
262        Stop the job's daemon.
263        """
264        result = self._core.stop(timeout_seconds)
265        self._warn()
266        return result
267
268    def pause(
269        self,
270        timeout_seconds: Union[int, float, None] = None,
271        debug: bool = False,
272    ) -> SuccessTuple:
273        """
274        Pause the job's daemon.
275        """
276        result = self._core.pause(timeout_seconds)
277        self._warn()
278        return result
279
280    def resume(
281        self,
282        timeout_seconds: Union[int, float, None] = None,
283        debug: bool = False,
284    ) -> SuccessTuple:
285        """
286        Resume a paused job.
287        """
288        result = self._core.resume(timeout_seconds)
289        self._warn()
290        return result
291
292    def delete(self, debug: bool = False) -> SuccessTuple:
293        """
294        Delete the job and its daemon.
295        """
296        result = self._core.delete()
297        self._warn()
298        return result
299
300    def is_running(self) -> bool:
301        """
302        Determine whether the job's daemon is running.
303        """
304        return self.status == 'running'
305
306    def exists(self, debug: bool = False) -> bool:
307        """
308        Determine whether the job exists.
309        """
310        return self._core.exists()
311
312    def get_logs(self) -> Union[str, None]:
313        """
314        Return the output text of the job's daemon.
315        """
316        return self._core.get_logs()
317
318    def monitor_logs(
319        self,
320        callback_function: Callable[[str], None] = _default_stdout_callback,
321        input_callback_function: Optional[Callable[[], str]] = None,
322        stop_callback_function: Optional[Callable[[SuccessTuple], None]] = None,
323        stop_event: Optional[asyncio.Event] = None,
324        stop_on_exit: bool = False,
325        strip_timestamps: bool = False,
326        accept_input: bool = True,
327        debug: bool = False,
328        _wait_if_stopped: bool = True,
329    ):
330        """
331        Monitor the job's log files and execute a callback on new lines.
332
333        Parameters
334        ----------
335        callback_function: Callable[[str], None], default partial(print, end='')
336            The callback to execute as new data comes in.
337            Defaults to printing the output directly to `stdout`.
338
339        input_callback_function: Optional[Callable[[], str]], default None
340            If provided, execute this callback when the daemon is blocking on stdin.
341            Defaults to `sys.stdin.readline()`.
342
343        stop_callback_function: Optional[Callable[[SuccessTuple]], str], default None
344            If provided, execute this callback when the daemon stops.
345            The job's SuccessTuple will be passed to the callback.
346
347        stop_event: Optional[asyncio.Event], default None
348            If provided, stop monitoring when this event is set.
349            You may instead raise `meerschaum.jobs.StopMonitoringLogs`
350            from within `callback_function` to stop monitoring.
351
352        stop_on_exit: bool, default False
353            If `True`, stop monitoring when the job stops.
354
355        strip_timestamps: bool, default False
356            If `True`, remove leading timestamps from lines.
357
358        accept_input: bool, default True
359            If `True`, accept input when the daemon blocks on stdin.
360        """
361        return self._monitor(
362            callback_function,
363            input_callback_function,
364            stop_callback_function,
365            stop_event,
366            stop_on_exit,
367            strip_timestamps,
368            accept_input,
369            _wait_if_stopped,
370            loop=None,
371            stop_flag=None,
372        )
373
374    async def monitor_logs_async(
375        self,
376        callback_function: Callable[[str], None] = _default_stdout_callback,
377        input_callback_function: Optional[Callable[[], str]] = None,
378        stop_callback_function: Optional[Callable[[SuccessTuple], None]] = None,
379        stop_event: Optional[asyncio.Event] = None,
380        stop_on_exit: bool = False,
381        strip_timestamps: bool = False,
382        accept_input: bool = True,
383        debug: bool = False,
384        _wait_if_stopped: bool = True,
385    ):
386        """
387        Monitor the job's log files and await a callback on new lines.
388
389        Parameters
390        ----------
391        callback_function: Callable[[str], None], default _default_stdout_callback
392            The callback to execute as new data comes in.
393            Defaults to printing the output directly to `stdout`.
394
395        input_callback_function: Optional[Callable[[], str]], default None
396            If provided, execute this callback when the daemon is blocking on stdin.
397            Defaults to `sys.stdin.readline()`.
398
399        stop_callback_function: Optional[Callable[[SuccessTuple]], str], default None
400            If provided, execute this callback when the daemon stops.
401            The job's SuccessTuple will be passed to the callback.
402
403        stop_event: Optional[asyncio.Event], default None
404            If provided, stop monitoring when this event is set.
405            You may instead raise `meerschaum.jobs.StopMonitoringLogs`
406            from within `callback_function` to stop monitoring.
407
408        stop_on_exit: bool, default False
409            If `True`, stop monitoring when the job stops.
410
411        strip_timestamps: bool, default False
412            If `True`, remove leading timestamps from lines.
413
414        accept_input: bool, default True
415            If `True`, accept input when the daemon blocks on stdin.
416        """
417        loop = asyncio.get_running_loop()
418        stop_flag = _native.core().StopFlag()
419        run = functools.partial(
420            self._monitor,
421            callback_function,
422            input_callback_function,
423            stop_callback_function,
424            stop_event,
425            stop_on_exit,
426            strip_timestamps,
427            accept_input,
428            _wait_if_stopped,
429            loop=loop,
430            stop_flag=stop_flag,
431        )
432        try:
433            await loop.run_in_executor(None, run)
434        except asyncio.CancelledError:
435            stop_flag.set()
436            raise
437        finally:
438            stop_flag.set()
439
440    def _monitor(
441        self,
442        callback_function,
443        input_callback_function,
444        stop_callback_function,
445        stop_event,
446        stop_on_exit,
447        strip_timestamps,
448        accept_input,
449        wait_if_stopped,
450        loop,
451        stop_flag,
452    ):
453        """
454        Follow the logs with the native monitor, running the callbacks on this thread's helper
455        (coroutine functions are awaited on `loop`, or on a private loop when there is none).
456        """
457        def sync(function):
458            if function is None or not inspect.iscoroutinefunction(function):
459                return function
460
461            def call(*args):
462                coroutine = function(*args)
463                if loop is not None:
464                    return asyncio.run_coroutine_threadsafe(coroutine, loop).result()
465                return asyncio.run(coroutine)
466
467            return call
468
469        def default_input_callback_function():
470            from meerschaum.utils.prompt import prompt
471            prompt_kwargs = self.get_prompt_kwargs()
472            if prompt_kwargs:
473                answer = prompt(**prompt_kwargs)
474                return answer + '\n'
475            return sys.stdin.readline()
476
477        input_callback = sync(input_callback_function) or default_input_callback_function
478        self._core.monitor_logs(
479            sync(callback_function),
480            input_callback=input_callback if accept_input else None,
481            stop_callback=sync(stop_callback_function),
482            stop_event=stop_event,
483            stop_flag=stop_flag,
484            stop_on_exit=stop_on_exit,
485            strip_timestamps=strip_timestamps,
486            accept_input=accept_input,
487            wait_if_stopped=wait_if_stopped,
488            stop_exception=StopMonitoringLogs,
489            clear_callback=_clear_screen,
490            warn=lambda message: warn(message),
491        )
492
493    def is_blocking_on_stdin(self, debug: bool = False) -> bool:
494        """
495        Return whether a job's daemon is blocking on stdin.
496        """
497        return self._core.is_blocking_on_stdin()
498
499    def get_prompt_kwargs(self, debug: bool = False) -> Dict[str, Any]:
500        """
501        Return the kwargs to the blocking `prompt()`, if available.
502        """
503        return self._core.get_prompt_kwargs()
504
505    def write_stdin(self, data):
506        """
507        Write to a job's daemon's `stdin`.
508        """
509        self._core.write_stdin(data)
510
511    @property
512    def executor(self) -> Union[Executor, None]:
513        """
514        If the job is not local, return the connector to its executor
515        (the remote API instance, `systemd`, a plugin's executor).
516        """
517        if self.executor_keys == 'local':
518            return None
519        from meerschaum.jobs._native import is_python_executor_type
520        if is_python_executor_type(self.executor_keys.split(':', maxsplit=1)[0]):
521            return mrsm.get_connector(self.executor_keys)
522        ### An executor type of a native (Rust) plugin: no Python connector exists.
523        from meerschaum.jobs._NativeExecutor import NativeExecutor
524        return NativeExecutor(self.executor_keys)
525
526    @property
527    def status(self) -> str:
528        """
529        Return the running status of the job's daemon.
530        """
531        if '_status_hook' in self.__dict__:
532            return self._status_hook()
533
534        return self._core.status
535
536    @property
537    def pid(self) -> Union[int, None]:
538        """
539        Return the PID of the job's dameon.
540        """
541        return self._core.pid
542
543    @property
544    def restart(self) -> bool:
545        """
546        Return whether to restart a stopped job.
547        """
548        return self._core.restarts
549
550    @property
551    def result(self) -> SuccessTuple:
552        """
553        Return the `SuccessTuple` when the job has terminated.
554        """
555        if '_result_hook' in self.__dict__:
556            if self.is_running():
557                return True, f"{self} is running."
558            return self._result_hook()
559
560        return self._core.result
561
562    @property
563    def sysargs(self) -> List[str]:
564        """
565        Return the sysargs to use for the Daemon.
566        """
567        return self._core.sysargs
568
569    @property
570    def health(self) -> Dict[str, Any]:
571        """
572        Return how the job's laps go (a job that repeats its work: `--loop`, `-s`, `--listen`).
573
574        Returns
575        -------
576        A dictionary with the keys:
577
578        - `state`: `'healthy'` (the last lap succeeded), `'failing'` (the last lap failed),
579          `'stalled'` (no lap for three times the expected interval, and at least a minute;
580          an hour without a sync for `--listen`; at least an hour for a first lap),
581          `'stopped'` (not running, or paused) or `'unknown'` (nothing recorded yet, e.g.
582          `'first sync in progress'`).
583        - `last_success`, `last_lap`: ISO-8601 UTC timestamps (or `None`).
584        - `consecutive_failures`: laps that failed in a row.
585        - `expected_interval_seconds`: how often laps are expected (`--min-seconds`, the
586          schedule's period, an hour for `--listen`), or `None`.
587        - `next_run`: while a running job waits for its next lap (`--min-seconds`, a cooldown
588          after a failure, a schedule's next fire), when that lap is due (ISO-8601 UTC);
589          otherwise `None`.
590        - `message`: what to show next to the state, e.g. `'last good sync 2 min ago'`.
591
592        Examples
593        --------
594        >>> job = Job('weather', 'sync pipes -c plugin:noaa --loop')
595        >>> success, msg = job.start()
596        >>> job.health['state']  # Before its first lap ends.
597        'unknown'
598        """
599        return self._core.health()
600
601    @property
602    def owner(self) -> Union[str, None]:
603        """
604        Return the user who started this job through the API server or the web console
605        (`None` for a job created locally: the CLI, Compose, systemd or Python). A job started
606        by another job has that job's owner. For an `api:` job, the server's record.
607
608        Examples
609        --------
610        >>> Job('nightly', executor_keys='api:main').owner
611        'alice'
612        """
613        return self._core.access().get('owner', None)
614
615    @property
616    def team(self) -> Union[str, None]:
617        """
618        Return the team this job belongs to: the primary team of the user who started it through
619        the API server or the web console (a job started by another job: that job's team).
620        `None` means Everyone (every local job).
621
622        Members of the team (and administrators) see the job and, with the job permissions,
623        control it. Renaming the team renames it here; deleting the team moves the job to
624        Everyone.
625        """
626        return self._core.access().get('team', None)
627
628    def get_daemon_properties(self) -> Dict[str, Any]:
629        """
630        Return the `properties` dictionary for the job's daemon.
631        """
632        return self._core.get_daemon_properties()
633
634    @property
635    def daemon(self) -> 'Daemon':
636        """
637        Return the daemon which this job manages.
638        """
639        from meerschaum.utils.daemon import Daemon
640        if self.executor_keys == 'local' and self.exists():
641            return Daemon(daemon_id=self.name)
642
643        return Daemon(
644            daemon_id=self.name,
645            sysargs=self.sysargs,
646            label=shlex.join(self.sysargs),
647            properties=self.get_daemon_properties(),
648        )
649
650    @property
651    def began(self) -> Union[datetime, None]:
652        """
653        The datetime when the job began running.
654        """
655        return _parse_dt(self._core.began)
656
657    @property
658    def ended(self) -> Union[datetime, None]:
659        """
660        The datetime when the job stopped running.
661        """
662        return _parse_dt(self._core.ended)
663
664    @property
665    def paused(self) -> Union[datetime, None]:
666        """
667        The datetime when the job was suspended while running.
668        """
669        return _parse_dt(self._core.paused)
670
671    @property
672    def stop_time(self) -> Union[datetime, None]:
673        """
674        Return the timestamp when the job was manually stopped.
675        """
676        return _parse_dt(self._core.stop_time)
677
678    @property
679    def hidden(self) -> bool:
680        """
681        Return a bool indicating whether this job should be displayed.
682        """
683        return self._core.hidden
684
685    def check_restart(self) -> SuccessTuple:
686        """
687        If `restart` is `True` and the daemon is not running,
688        restart the job.
689        Do not restart if the job was manually stopped.
690        """
691        result = self._core.check_restart()
692        self._warn()
693        return result
694
695    @property
696    def label(self) -> str:
697        """
698        Return the job's Daemon label (joined sysargs).
699        """
700        return self._core.label
701
702    @property
703    def _externally_managed_file(self) -> pathlib.Path:
704        """
705        Return the path to the externally managed file.
706        """
707        return pathlib.Path(self._core.paths()['externally_managed'])
708
709    def _set_externally_managed(self):
710        """
711        Set this job as externally managed.
712        """
713        self._externally_managed = True
714        self._core.set_externally_managed()
715
716    @property
717    def _is_externally_managed(self) -> bool:
718        """
719        Return whether this job is externally managed.
720        """
721        return self._core.is_externally_managed
722
723    @property
724    def env(self) -> Dict[str, str]:
725        """
726        Return the environment variables to set for the job's process.
727        """
728        return self._core.env
729
730    @property
731    def delete_after_completion(self) -> bool:
732        """
733        Return whether this job is configured to delete itself after completion.
734        """
735        return self._core.delete_after_completion
736
737    def __str__(self) -> str:
738        return self._core.describe()
739
740    def __repr__(self) -> str:
741        return str(self)
742
743    def __hash__(self) -> int:
744        return hash(self.name)