Skip to content

lablink_client_service.update_inuse_status

Callback to call the API when process state changes.

Source code in packages/client/src/lablink_client_service/update_inuse_status.py 
118
119
120
def api_callback(process_name: str, url: str, client_secret: str = ""):
    """Callback to call the API when process state changes."""
    call_api(process_name, url, client_secret=client_secret)
Source code in packages/client/src/lablink_client_service/update_inuse_status.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def call_api(process_name, url, client_secret=""):
    hostname = os.getenv("VM_NAME")
    status = is_process_running(process_name=process_name)

    headers = get_auth_headers(client_secret)

    retry_count = 0

    while retry_count < MAX_API_RETRIES:
        try:
            response = requests.post(
                url,
                json={"hostname": hostname, "status": status},
                headers=headers,
                timeout=(10, 20),
            )
            response.raise_for_status()
            logger.info(f"Updated in-use status: {status}")
            break  # Success, exit retry loop
        except requests.exceptions.Timeout:
            logger.warning(
                f"Status update timed out "
                f"(attempt {retry_count + 1}/{MAX_API_RETRIES})"
            )
        except requests.RequestException as e:
            logger.warning(
                f"Status update failed: {e} "
                f"(attempt {retry_count + 1}/{MAX_API_RETRIES})"
            )

        retry_count += 1
        if retry_count < MAX_API_RETRIES:
            jitter = random.uniform(0, 5)
            time.sleep(API_RETRY_DELAY + jitter)
    else:
        logger.error(
            f"Failed to update in-use status after {MAX_API_RETRIES} attempts"
        )

Default callback to log when a process is detected.

Source code in packages/client/src/lablink_client_service/update_inuse_status.py 
35
36
37
def default_callback(process_name: str):
    """Default callback to log when a process is detected."""
    logger.info(f"Process '{process_name}' started.")

Check if a specific process is running.

Source code in packages/client/src/lablink_client_service/update_inuse_status.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
def is_process_running(process_name: str) -> bool:
    """
    Check if a specific process is running.
    """
    for proc in psutil.process_iter():
        try:
            if any(process_name in part for part in proc.cmdline()):
                if "update_inuse_status" in " ".join(proc.cmdline()):
                    continue
                return True
        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            pass
    return False

Listen for a specific process to start or stop.

Parameters:

Name Type Description Default
process_name str

The name of the process to listen for.

 required
interval int

The interval (in seconds) to check the process status.

 20
callback_func callable

A callback function to execute when the process state changes.

 None
Source code in packages/client/src/lablink_client_service/update_inuse_status.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def listen_for_process(
    process_name: str, interval: int = 20, callback_func=None
) -> None:
    """Listen for a specific process to start or stop.

    Args:
        process_name (str): The name of the process to listen for.
        interval (int, optional): The interval (in seconds) to check the process status.
        callback_func (callable, optional): A callback function to execute when the
            process state changes.
    """

    # Set up a default callback function if none is provided
    if callback_func is None:
        callback_func = lambda: default_callback(process_name)

    logger.info(f"Monitoring process '{process_name}'")

    # Continuously check if the process is running
    process_running_prev = is_process_running(process_name)
    while True:
        # Check if the process is running now
        process_running_curr = is_process_running(process_name)

        # Compare the current state with the previous state
        if process_running_prev != process_running_curr:
            if callback_func:
                logger.info(f"Process '{process_name}' state changed.")
                callback_func()

        # Update the previous state
        process_running_prev = process_running_curr

        # Wait for the specified interval before checking again
        jitter = random.uniform(0, 5)
        time.sleep(interval + jitter)
Source code in packages/client/src/lablink_client_service/update_inuse_status.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@hydra.main(version_base=None, config_path="conf", config_name="config")
def main(cfg: Config) -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    logging.getLogger("lablink_client_service").setLevel(logging.DEBUG)

    logger.info("Starting in-use status monitoring service")

    base_url, client_secret, _ = get_client_env(cfg)
    url = f"{base_url}/api/update_inuse_status"

    # Start listening for the process
    listen_for_process(
        process_name=cfg.client.software,
        interval=20,
        callback_func=lambda: api_callback(cfg.client.software, url, client_secret),
    )