ShMonitor

ShMonitor use ShTracerHost as backend for polling output of a command.

class duetector.monitors.sh_monitor.ShTracerHost(backend, timeout=5)

Bases: object

Host for Shell, provide a way to poll shell command.

Use subprocess.Popen to run shell command.

attach(tracer)

detach(tracer)

get_poller(tracer) → Callable[[None], None]

Provide a callback function for Poller.

Use subprocess.Popen to run shell command, pipe stdout to callback.

poll(tracer)

Poll a tracer.

poll_all()

Poll all tracers.

set_callback(tracer, callback)

Set callback for tracer.

class duetector.monitors.sh_monitor.ShMonitor(config: dict[str, Any] | None = None, *args, **kwargs)

Bases: Monitor

A monitor for shell command.

property auto_init

Auto init tracers when init monitor.

config_scope: str | None = 'monitor.sh'

Config scope for this monitor.

default_config = {'auto_init': True, 'backend_args': {'max_workers': 10}, 'disabled': False, 'poller': {'call_when_shutdown': True, 'interval_ms': 500}, 'timeout': 5}

Default config for this monitor.

Two sub-configs are available:

• auto_init: Auto init tracers when init monitor.

• timeout: Timeout for shell command.

init()

poll(tracer: ShellTracer)

Poll a tracer. Should be implemented by subclass.

poll_all()

Poll all tracers. Depends on self.poll.

property timeout

Timeout for shell command.