Monitor topics and streams#
This example page consolidates the HPS monitor examples into one place. It
starts with topic discovery using MonitorApi.list_topics(), then shows
the main stream types exposed by the monitor WebSocket bus for running tasks,
task definitions, and backend services.
Use these examples in this order:
Discover available tags and values with
MonitorApi.list_topics().Subscribe to task-level streams such as logs, host resources, or the process tree.
Subscribe to scheduler or service-level streams when you need cluster or backend visibility.
All examples below use the scripts in examples/monitor.
List monitor topics#
This example shows how to use MonitorApi.list_topics() to discover all
tag keys and their currently active values on the HPS monitor WebSocket bus.
Use this as a first step when building a monitoring integration because it tells
you which tasks, evaluators, log files, and metric streams are currently visible
before you commit to a targeted subscription.
Background#
The HPS monitor service is a WebSocket-based pub/sub bus. Every message on the
bus carries a set of key-value tags such as task_id, file_path, and
evaluator_name that describe what the message relates to.
MonitorApi.list_topics() sends a list_tags command over the
WebSocket and returns a dictionary mapping every known tag key to all values
that currently appear on the bus.
Typical tag keys include:
Tag key |
Meaning |
|---|---|
|
IDs of tasks that have active evaluator log streams. |
|
Names of evaluators currently connected to the monitor service. |
|
Log source type, for example |
|
Names of log files currently being tailed, for example
|
|
Job IDs associated with active streams. |
|
Project IDs associated with active streams. |
|
Status tags attached to active streams. |
How the example works#
The script follows the standard three-step setup shared by all monitor examples:
Authenticate with the top-level
Client. This performs the Keycloak OAuth exchange and stores the access token.Create a
MonitorApi. Passclient=hpsso that the monitor client can reuse the same HTTP session for monitor calls and for any JMS/RMS lookups required by methods such asMonitorApi.stream_task_host_resources(), and passws_connection_optionsto disable TLS certificate verification when connecting to a local server with a self-signed certificate.Call
MonitorApi.list_topics(). The call is synchronous: it opens a short-lived WebSocket connection, sends thelist_tagscommand, waits for the response, and returns the result as a plain Python dictionary.
The exclude_noisy parameter (True by default) suppresses
high-cardinality keys such as timestamp that change with every message and
are rarely useful for topic discovery.
Command-line options#
Option |
Description |
|---|---|
|
Base URL of the HPS server (default: |
|
HPS username. |
|
HPS password. |
|
Print only the values for one specific tag key. |
|
Filter the output to tag keys that include this task ID as a value. |
|
Maximum number of values to request per tag key (default: 1000). |
|
Print raw JSON instead of the human-readable table. |
|
Include high-cardinality keys such as |
|
Disable TLS certificate verification (required for local servers with self-signed certificates). |
Expected output#
A typical run while two tasks are running produces output like this:
TAG KEY COUNT VALUES
----------------------------------------------------------------------------------------
client_type 3 ansys.rep.evaluator.file_tail
ansys.rep.evaluator.host_resources
ansys.rep.evaluator.process_tree
evaluator_name 1 evaluator-0
file_path 2 console_output.txt
fluent.trn
job_id 2 033Vub9xM... 033VubA2K...
project_id 1 033Vub7fh...
task_id 2 033VubFI0... 033VubFI3...
Here is the list_topics.py script for this example:
# Copyright (C) 2022 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Annotated example: discover available monitor topics with MonitorApi.list_topics.
``list_topics`` sends a ``list_tags`` WebSocket command to the HPS monitor
endpoint and returns a dictionary mapping every known tag key to all currently
active values. This is useful for exploring what is visible on the WebSocket
bus before building a targeted subscription.
Typical tag keys include:
- ``task_id`` – IDs of tasks that have active evaluator log streams.
- ``evaluator_name``– Names of connected evaluators.
- ``client_type`` – Log source types (e.g. ``ansys.rep.evaluator.file_tail``).
- ``file_path`` – Log file names being tailed.
- ``job_id``, ``project_id``, ``status`` – Additional correlation tags.
Usage (local dev with self-signed cert):
pip install -e .
python examples/monitor/list_topics.py \\
--base-url https://localhost:8443/hps \\
--username repadmin \\
--password repadmin \\
--insecure
Pass ``--task-id <ID>`` to further filter the output to rows where that task
appears as a value, or ``--key client_type`` to print only the values for a
single tag key.
"""
from __future__ import annotations
import argparse
import json
import ssl
from ansys.hps.client import Client
from ansys.hps.client.monitor.api.monitor_api import MonitorApi
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="List all available monitor topic tags and their current values."
)
parser.add_argument("--base-url", default="https://localhost:8443/hps")
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument(
"--limit",
type=int,
default=1000,
help="Maximum number of values to request per tag key (default: 1000).",
)
parser.add_argument(
"--key",
default=None,
metavar="TAG_KEY",
help="If set, print only the values for this single tag key.",
)
parser.add_argument(
"--task-id",
default=None,
metavar="TASK_ID",
help="If set, filter and show only tag keys that contain this task ID as a value.",
)
parser.add_argument(
"--json",
action="store_true",
dest="as_json",
help="Output raw JSON instead of a human-readable table.",
)
parser.add_argument(
"--all",
action="store_true",
dest="show_all",
help="Include high-cardinality keys like 'timestamp' that are suppressed by default.",
)
parser.add_argument(
"--insecure",
action="store_true",
help="Disable TLS certificate verification for local/self-signed endpoints.",
)
return parser
def main() -> None:
args = _build_parser().parse_args()
# 1) Authenticate once with the top-level HPS client.
hps = Client(
args.base_url,
args.username,
args.password,
verify=not args.insecure,
)
# 2) Configure WebSocket options only when running in insecure local mode.
ws_options = {"sslopt": {"cert_reqs": ssl.CERT_NONE}} if args.insecure else None
# 3) Build a MonitorApi that reuses the authenticated HPS client.
monitor = MonitorApi(
hps,
ws_connection_options=ws_options,
timeout_seconds=30.0,
)
# 4) Call list_topics to send a list_tags WebSocket command and get back
# the full tag catalogue from the server.
# The return value is: {tag_key: [value, value, ...], ...}
# Pass exclude_noisy=False to also include high-cardinality keys like
# 'timestamp' that are suppressed by default in the API.
topics: dict[str, list[str]] = monitor.list_topics(
limit=args.limit,
exclude_noisy=not args.show_all,
)
# 5) Apply optional filters.
if args.key:
# Restrict to a single key the user asked for.
topics = {args.key: topics.get(args.key, [])}
if args.task_id:
# Keep only tag keys whose value list contains the requested task ID.
topics = {k: v for k, v in topics.items() if args.task_id in v}
# 6) Output.
if args.as_json:
print(json.dumps(topics, indent=2))
return
if not topics:
print(
"No topics found"
+ (" matching the given filters." if (args.key or args.task_id) else ".")
)
return
# Human-readable table: one row per tag key, with all values listed.
print(f"{'TAG KEY':<30} COUNT VALUES")
print("-" * 88)
for key in sorted(topics):
values = topics[key]
# Print the first value on the same line; subsequent values indented.
first, *rest = values if values else ["(none)"]
print(f"{key:<30} {len(values):>5} {first}")
for val in rest:
print(f"{'':30} {val}")
if __name__ == "__main__":
main()
Stream task logs#
This example shows how to stream evaluator file-tail log output for a running
task using MonitorApi.stream_task_logs(). Log lines are delivered over
a WebSocket connection as the evaluator writes them, making this the primary way
to watch solver output in real time without polling the server.
Background#
While a task is running, the HPS evaluator tails one or more log files and
publishes each new line as a tagged message on the monitor WebSocket bus.
MonitorApi.stream_task_logs() subscribes to the task_id and
client_type=ansys.rep.evaluator.file_tail tags for the requested task and
yields one dictionary per line.
Each yielded message dictionary contains:
Key |
Content |
|---|---|
|
The log line text. |
|
ISO-8601 timestamp of when the line was written. |
|
A nested dictionary with |
The backlog parameter controls how many historical log lines are replayed
from the server’s message buffer on connection. Set it to a larger value such
as 500 to catch up on output that was written before the subscription opened, or
to 0 to receive only new lines going forward.
Filtering by file#
When a task produces multiple log files such as console_output.txt and
fluent.trn, you can supply file_path to narrow the stream to a single
file. Omit it to receive lines from all files interleaved, distinguished by the
tags["file_path"] field in each message.
Command-line options#
Option |
Description |
|---|---|
|
Base URL of the HPS server (default: |
|
HPS username. |
|
HPS password. |
|
ID of the task to stream logs for (required). |
|
Optional log filename to filter, for example |
|
Number of historical log lines to replay on connect (default: 200). |
|
Stop after receiving this many messages. Omit for continuous streaming. |
|
Disable TLS certificate verification (required for local servers with self-signed certificates). |
Expected output#
A typical run streaming a Fluent solver console produces output like this:
Streaming logs for task 033VubFI0m4PYEREddzbN1
Press Ctrl+C to stop
----------------------------------------------------------------------------------------
2026-06-26T15:17:43 [console_output.txt]
2026-06-26T15:17:43 [console_output.txt] 1 1.0000e+00 0.0000e+00 0.0000e+00 0.0000e+00 1.0000e+00 308.0000 0.0000e+00 0.0000e+00 0:00:12 199
2026-06-26T15:17:44 [console_output.txt] 2 9.8765e-01 1.2340e-03 1.2340e-03 9.8760e-04 9.8765e-01 307.9200 3.4160e-03 0.0000e+00 0:00:11 198
...
2026-06-26T15:19:33 [console_output.txt] 76 9.6758e-03 4.5210e-04 2.3410e-04 1.1230e-04 9.6758e-03 303.3800 4.2100e-05 0.0000e+00 0:00:00 0
Here is the stream_task_logs.py script for this example:
# Copyright (C) 2022 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Stream evaluator file-tail logs for a task with ``MonitorApi.stream_task_logs``.
This example shows the standard monitoring pattern for task output:
1. Authenticate once with ``Client``.
2. Reuse the same token/session in ``MonitorApi``.
3. Subscribe to task log messages by ``task_id``.
4. Optionally narrow output to a specific ``file_path``.
By default, messages are streamed continuously until you stop the program
(``Ctrl+C``). You can set ``--max-messages`` for a bounded run.
Typical usage (local/self-signed endpoint):
python examples/monitor/stream_task_logs.py \
--base-url https://localhost:8443/hps \
--username repadmin \
--password repadmin \
--task-id <TASK_ID> \
--insecure
Filter to one file only:
python examples/monitor/stream_task_logs.py \
--base-url https://localhost:8443/hps \
--username repadmin \
--password repadmin \
--task-id <TASK_ID> \
--file-path console_output.txt \
--insecure
"""
from __future__ import annotations
import argparse
import json
import ssl
from typing import Any
from ansys.hps.client import Client
from ansys.hps.client.monitor.api.monitor_api import MonitorApi
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Stream evaluator file-tail logs for a task.")
parser.add_argument("--base-url", default="https://localhost:8443/hps")
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument("--task-id", required=True)
parser.add_argument("--file-path", default=None, help="Optional single file name to filter.")
parser.add_argument("--backlog", type=int, default=200)
parser.add_argument(
"--max-messages",
type=int,
default=None,
help="Maximum messages to stream. Omit for unlimited streaming.",
)
parser.add_argument(
"--insecure",
action="store_true",
help="Disable TLS certificate verification for local/self-signed endpoints.",
)
return parser
def _to_text(message: dict[str, Any]) -> str:
"""Normalize monitor payload text for printing."""
text = message.get("message", "")
if isinstance(text, str):
return text
return json.dumps(text)
def main() -> None:
args = _build_parser().parse_args()
# 1) Authenticate once with the top-level HPS client.
hps = Client(
args.base_url,
args.username,
args.password,
verify=not args.insecure,
)
# 2) Configure websocket options only when running in insecure local mode.
ws_options: dict[str, Any] | None = None
if args.insecure:
ws_options = {"sslopt": {"cert_reqs": ssl.CERT_NONE}}
# 3) Reuse the same authenticated client in MonitorApi.
monitor = MonitorApi(
hps,
ws_connection_options=ws_options,
timeout_seconds=30.0,
)
print(f"Streaming logs for task {args.task_id}")
if args.file_path:
print(f"Filtering file_path={args.file_path}")
print("Press Ctrl+C to stop")
print("-" * 88)
try:
# 4) stream_task_logs subscribes to task_id + evaluator file-tail client_type.
for msg in monitor.stream_task_logs(
task_id=args.task_id,
file_path=args.file_path,
backlog=args.backlog,
max_messages=args.max_messages,
):
ts = (msg.get("time") or msg.get("timestamp") or "")[:19]
tags = msg.get("tags") or {}
path = tags.get("file_path", "?")
line = _to_text(msg)
print(f"{ts} [{path}] {line}")
except KeyboardInterrupt:
print("\nInterrupted by user.")
if __name__ == "__main__":
main()
Stream task host resources#
This example shows how to stream host CPU and memory utilisation metrics for a
running task using MonitorApi.stream_task_host_resources(). Unlike
stream_task_logs(), this method requires both a project ID
and task ID. It automatically resolves the evaluator assigned to the task through JMS and
RMS and subscribes to that evaluator’s host_resources metric stream.
Background#
The HPS evaluator periodically publishes a snapshot of the host machine’s CPU
and memory state as a metric message on the monitor bus.
MonitorApi.stream_task_host_resources() subscribes to the
client_type=ansys.rep.evaluator.host_resources tag for the evaluator
running the given task and yields one dictionary per snapshot.
The message field of each yielded dictionary is a JSON string with the
following structure:
{
"cpu": {
"usage": 62.5,
"per_core": [55.1, 68.2, 70.3, 56.4]
},
"virtual_memory": {
"percent": 96.3,
"total": 34359738368,
"available": 1236172800,
"used": 33123565568
},
"swap_memory": { ... },
"disk_io": { ... }
}
The example’s _extract_metrics helper parses this payload and returns
(cpu_usage, memory_percent) as floats, handling the case where the field is
absent or unparsable.
Evaluator resolution#
MonitorApi.stream_task_host_resources() looks up the task via the JMS
API, finds the evaluator name from the task’s execution context, then queries
the RMS API to resolve the evaluator’s host identifier. This is why passing
client=hps when constructing MonitorApi is required: the method
uses that pre-authenticated client for JMS/RMS API calls.
Because JMS tasks are scoped by project,
MonitorApi.stream_task_host_resources() needs both task_id and
project_id.
If you know task_id but not project_id, use
MonitorApi.resolve_project_id_for_task() before calling
MonitorApi.stream_task_host_resources().
Both helpers raise ansys.hps.client.ClientError when the required
monitor/JMS/RMS metadata cannot be resolved.
Display modes#
The example supports two display modes selected by --interval:
Per-message mode (default,
--interval 0): prints one line per metric snapshot as it arrives, showing the timestamp, CPU usage, and memory percent.Rolling-summary mode (
--interval N): accumulates snapshots for N seconds and then prints a single summary line with the last, minimum, maximum, and average values for both CPU and memory over that window.
Command-line options#
Option |
Description |
|---|---|
|
Base URL of the HPS server (default: |
|
HPS username. |
|
HPS password. |
|
ID of the task to stream host metrics for (required). |
|
Optional ID of the project that owns |
|
Number of historical metric snapshots to replay on connect (default: 20). |
|
If greater than zero, print a rolling summary every N seconds instead of printing each raw message. |
|
Stop after receiving this many snapshots. Omit for continuous streaming. |
|
Disable TLS certificate verification (required for local servers with self-signed certificates). |
Expected output#
Per-message mode:
Streaming host resources for project 03P7JeAj5L0vjX7B8u4JQ2, task 033VubFI0m4PYEREddzbN1
(resolving evaluator via JMS/RMS...)
Press Ctrl+C to stop
time cpu mem
----------------------------------------------------------------------------------------
2026-06-26T15:18:44 62.5% 96.3%
2026-06-26T15:18:49 64.4% 96.4%
2026-06-26T15:18:57 74.2% 97.0%
Rolling-summary mode (--interval 30):
time_utc n cpu_last cpu_min cpu_max cpu_avg mem_last mem_min mem_max mem_avg
----------------------------------------------------------------------------------------
2026-06-26T15:19:00 5 74.2 62.5 74.2 67.3 97.0 96.3 97.0 96.7
Here is the stream_task_host_resources.py script for this example:
# Copyright (C) 2022 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Annotated example: stream host CPU/memory metrics with MonitorApi.stream_task_host_resources.
``stream_task_host_resources`` resolves the evaluator assigned to a task (via
JMS and RMS) and subscribes to ``host_resources`` metric messages for that
evaluator. A pre-authenticated HPS client must be passed to
``MonitorApi(...)`` so these JMS/RMS lookups can run. If
``--project-id`` is omitted, this example infers it from ``stream_task_logs``.
Usage (local dev with self-signed cert):
python examples/monitor/stream_task_host_resources.py \\
--base-url https://localhost:8443/hps \\
--username repadmin \\
--password repadmin \\
--task-id <TASK_ID> \\
--insecure
Pass ``--interval 20`` to print a rolling 20-second summary instead of every
raw message. Press Ctrl+C to stop.
"""
from __future__ import annotations
import argparse
import json
import ssl
import statistics
import time
from datetime import datetime, timezone
from typing import Any
from ansys.hps.client import Client, ClientError
from ansys.hps.client.monitor.api.monitor_api import MonitorApi
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Stream host CPU/memory metrics for a running task."
)
parser.add_argument("--base-url", default="https://localhost:8443/hps")
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument(
"--project-id",
default=None,
help="Optional project ID. If omitted, inferred from task logs.",
)
parser.add_argument("--task-id", required=True)
parser.add_argument(
"--backlog",
type=int,
default=20,
help="Historical metric snapshots to request on connect (default: 20).",
)
parser.add_argument(
"--interval",
type=int,
default=0,
metavar="SECONDS",
help=(
"If >0, accumulate samples and print a rolling summary every N seconds "
"instead of printing each raw message. Default: 0 (print every message)."
),
)
parser.add_argument(
"--max-messages",
type=int,
default=None,
help="Maximum metric messages to receive. Omit for unlimited streaming.",
)
parser.add_argument(
"--insecure",
action="store_true",
help="Disable TLS certificate verification for local/self-signed endpoints.",
)
return parser
def _extract_metrics(msg: dict[str, Any]) -> tuple[float | None, float | None]:
"""Pull CPU usage and memory percent out of a host_resources message.
The server embeds the metric payload as a JSON string in the ``message``
field with the shape::
{
"cpu": {"usage": 42.4, "per_core": [...]},
"virtual_memory": {"percent": 88.9, ...},
...
}
"""
raw = msg.get("message", "")
if isinstance(raw, str):
try:
payload = json.loads(raw)
except (ValueError, TypeError):
payload = {}
elif isinstance(raw, dict):
payload = raw
else:
payload = {}
try:
cpu = float(payload["cpu"]["usage"])
except (KeyError, TypeError, ValueError):
cpu = None
try:
mem = float(payload["virtual_memory"]["percent"])
except (KeyError, TypeError, ValueError):
mem = None
return cpu, mem
def _print_raw(msg: dict[str, Any]) -> None:
"""Print a single host-resources message in a compact tabular form."""
ts = (msg.get("time") or msg.get("timestamp") or "")[:19]
cpu, mem = _extract_metrics(msg)
cpu_s = f"{cpu:6.1f}%" if cpu is not None else " n/a"
mem_s = f"{mem:6.1f}%" if mem is not None else " n/a"
print(f"{ts} cpu {cpu_s} mem {mem_s}")
def _print_summary(
window_end: str,
cpu_vals: list[float],
mem_vals: list[float],
) -> None:
"""Print a one-line rolling window summary."""
def _fmt(vals: list[float]) -> str:
if not vals:
return " n/a n/a n/a n/a"
return f"{vals[-1]:6.1f} {min(vals):6.1f} {max(vals):6.1f} {statistics.mean(vals):6.1f}"
n = max(len(cpu_vals), len(mem_vals))
print(f"{window_end} {n:>5} cpu {_fmt(cpu_vals)} mem {_fmt(mem_vals)}")
def main() -> None:
args = _build_parser().parse_args()
# 1) Authenticate once with the top-level HPS client.
hps = Client(
args.base_url,
args.username,
args.password,
verify=not args.insecure,
)
# 2) Configure WebSocket options only when running in insecure local mode.
ws_options: dict[str, Any] | None = None
if args.insecure:
ws_options = {"sslopt": {"cert_reqs": ssl.CERT_NONE}}
# 3) Build a MonitorApi that reuses the authenticated client.
# stream_task_host_resources uses JMS/RMS APIs to resolve the evaluator.
monitor = MonitorApi(
hps,
ws_connection_options=ws_options,
timeout_seconds=30.0,
)
use_summary = args.interval > 0
cpu_vals: list[float] = []
mem_vals: list[float] = []
try:
project_id = args.project_id
if not project_id:
project_id = monitor.resolve_project_id_for_task(task_id=args.task_id)
print(f"Resolved project_id={project_id} from task logs")
print(f"Streaming host resources for project {project_id}, task {args.task_id}")
print("(resolving evaluator via JMS/RMS...)")
if use_summary:
print(f"Reporting every {args.interval}s")
print(
f"{'time_utc':<19} {'n':>5} "
f"{'cpu_last':>8} {'cpu_min':>7} {'cpu_max':>7} {'cpu_avg':>7} "
f"{'mem_last':>8} {'mem_min':>7} {'mem_max':>7} {'mem_avg':>7}"
)
else:
print("Press Ctrl+C to stop")
print(f"{'time':<19} {'cpu':>9} {'mem':>9}")
print("-" * 88)
window_start = time.monotonic()
# 4) stream_task_host_resources resolves the evaluator and subscribes to
# the host_resources metric statistic for it. Each yielded dict is one
# metric snapshot pushed by the server.
for msg in monitor.stream_task_host_resources(
task_id=args.task_id,
project_id=project_id,
backlog=args.backlog,
max_messages=args.max_messages,
):
cpu, mem = _extract_metrics(msg)
if use_summary:
# Accumulate into rolling window.
if cpu is not None:
cpu_vals.append(cpu)
if mem is not None:
mem_vals.append(mem)
now = time.monotonic()
if now - window_start >= args.interval:
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
_print_summary(ts, cpu_vals, mem_vals)
cpu_vals = []
mem_vals = []
window_start = now
else:
_print_raw(msg)
except ClientError as exc:
raise SystemExit(f"Monitor request failed: {exc}") from exc
except KeyboardInterrupt:
print("\nInterrupted by user.")
if use_summary and (cpu_vals or mem_vals):
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
_print_summary(ts, cpu_vals, mem_vals)
if __name__ == "__main__":
main()
Stream task process tree#
This example shows how to stream process-tree snapshots for a running task using
MonitorApi.stream_task_process_tree(). Each snapshot contains every
process in the task’s process group, including the task wrapper, the solver,
any co-processes, and their parent-child relationships.
Background#
The HPS evaluator periodically walks the process group of the task it is running
and publishes a snapshot as a metric message on the monitor bus.
MonitorApi.stream_task_process_tree() subscribes to the
client_type=ansys.rep.evaluator.process_tree tag for the given task and
yields one dictionary per snapshot.
Message format#
The message field of each yielded dictionary is a JSON string whose payload
is the root process node with nested children:
{
"pid": 18508,
"ppid": 0,
"name": "task_033VubFI0m4PYEREddzbN1",
"cpu_percentage": "0.2",
"memory_percentage": "0.5",
"state": "running",
"level": 0,
"children": [
{
"pid": 17904,
"ppid": 18508,
"name": "cmd.exe",
"cpu_percentage": "0.0",
"memory_percentage": "0.0",
"state": "running",
"level": 1,
"children": [
{
"pid": 14636,
"ppid": 17904,
"name": "cx2610.exe",
"cpu_percentage": "14.1",
"memory_percentage": "2.3",
"state": "running",
"level": 2,
"children": [ ... ]
}
]
}
]
}
Note that cpu_percentage and memory_percentage are strings, not
floats. The example’s _extract_processes helper recursively flattens this
tree into a plain list, and callers must cast these fields with float()
before arithmetic.
Flattening the tree#
The _extract_processes function uses a recursive _flatten helper to
convert the nested JSON tree into a flat list of process dictionaries. Once
flattened, processes can be sorted by CPU usage, filtered by name, or used to
compute aggregate statistics without recursion.
Display modes#
The example supports three display modes:
Flat sorted view (default): prints each snapshot as a table sorted by CPU usage descending, one process per row.
Tree view (
--tree): indents child processes under their parent using theppidfield to reconstruct the parent-child hierarchy.Rolling-summary mode (
--interval N): accumulates snapshots for N seconds and prints a single summary line showing the snapshot count, process count range, and the process with peak CPU usage over the window.
Command-line options#
Option |
Description |
|---|---|
|
Base URL of the HPS server (default: |
|
HPS username. |
|
HPS password. |
|
ID of the task to stream process-tree snapshots for (required). |
|
Number of historical snapshots to replay on connect (default: 20). |
|
Show child processes indented under their parent instead of a flat sorted list. |
|
If greater than zero, print a rolling summary every N seconds instead of printing each snapshot in full. |
|
Stop after receiving this many snapshots. Omit for continuous streaming. |
|
Disable TLS certificate verification (required for local servers with self-signed certificates). |
Expected output#
Flat sorted view during an active Fluent solve:
Streaming process-tree snapshots for task 033VubFI0m4PYEREddzbN1
Press Ctrl+C to stop
----------------------------------------------------------------------------------------
2026-06-26T15:19:00 [9 processes]
pid name cpu mem status
----------------------------------------------------------------
14636 cx2610.exe 14.1% 2.3% running
24088 fl2610.exe 9.7% 2.1% running
23648 ansyscl.exe 0.3% 0.1% running
18508 task_033VubFI0m4... 0.0% 0.5% running
...
Tree view (--tree) of the same snapshot:
2026-06-26T15:19:00 [9 processes]
pid name cpu mem status
----------------------------------------------------------------
18508 task_033VubFI0m4... 0.0% 0.5% running
17904 cmd.exe 0.0% 0.0% running
45592 fluent.exe 0.0% 0.0% running
14636 cx2610.exe 14.1% 2.3% running
24088 fl2610.exe 9.7% 2.1% running
23648 ansyscl.exe 0.3% 0.1% running
Here is the stream_task_process_tree.py script for this example:
# Copyright (C) 2022 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Stream process-tree metric snapshots for a task with ``MonitorApi.stream_task_process_tree``.
The evaluator pushes a new process-tree snapshot on a fixed interval while the
task is running. Each snapshot lists every process in the task's process group
with its PID, parent PID, name, CPU usage, and memory usage.
Usage (local dev with self-signed cert):
python examples/monitor/stream_task_process_tree.py \\
--base-url https://localhost:8443/hps \\
--username repadmin \\
--password repadmin \\
--task-id <TASK_ID> \\
--insecure
Pass ``--tree`` to indent child processes under their parent in each snapshot.
Pass ``--interval 20`` to print a one-line rolling summary every 20 seconds
instead of printing every snapshot in full. Press Ctrl+C to stop.
"""
from __future__ import annotations
import argparse
import json
import ssl
import time
from datetime import datetime, timezone
from typing import Any
from ansys.hps.client import Client
from ansys.hps.client.monitor.api.monitor_api import MonitorApi
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Stream process-tree metric snapshots for a running task."
)
parser.add_argument("--base-url", default="https://localhost:8443/hps")
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument("--task-id", required=True)
parser.add_argument(
"--backlog",
type=int,
default=20,
help="Historical snapshots to request on connect (default: 20).",
)
parser.add_argument(
"--tree",
action="store_true",
help="Indent child processes under their parent (uses ppid field).",
)
parser.add_argument(
"--interval",
type=int,
default=0,
metavar="SECONDS",
help=(
"If >0, accumulate snapshots and print a rolling summary every N seconds "
"instead of printing each snapshot in full. Default: 0 (print every snapshot)."
),
)
parser.add_argument(
"--max-messages",
type=int,
default=None,
help="Maximum snapshots to receive. Omit for unlimited streaming.",
)
parser.add_argument(
"--insecure",
action="store_true",
help="Disable TLS certificate verification for local/self-signed endpoints.",
)
return parser
def _extract_processes(msg: dict[str, Any]) -> list[dict[str, Any]]:
"""Return a flat process list from a process-tree snapshot message.
The server embeds the metric payload as a JSON string in the ``message``
field. The payload is the *root* process node with nested ``children``::
{
"pid": 1234, "ppid": 0, "name": "task_...",
"cpu_percentage": "92.3", # string, not float
"memory_percentage": "4.1", # string, not float
"state": "running",
"level": 0,
"children": [
{"pid": 5678, "ppid": 1234, "name": "fluent.exe", ...},
...
]
}
This function flattens the nested tree into a plain list (``children``
key stripped from each node) so callers can iterate without recursion.
"""
raw = msg.get("message", "")
if isinstance(raw, str):
try:
payload = json.loads(raw)
except (ValueError, TypeError):
payload = {}
elif isinstance(raw, dict):
payload = raw
else:
payload = {}
if not payload or "pid" not in payload:
return []
result: list[dict[str, Any]] = []
def _flatten(node: dict[str, Any]) -> None:
result.append({k: v for k, v in node.items() if k != "children"})
for child in node.get("children", []):
_flatten(child)
_flatten(payload)
return result
def _build_tree(processes: list[dict[str, Any]]) -> dict[int, list[dict[str, Any]]]:
"""Build a parent → children mapping from the process list."""
tree: dict[int, list[dict[str, Any]]] = {}
for proc in processes:
ppid = proc.get("ppid", 0) or 0
tree.setdefault(ppid, []).append(proc)
return tree
def _print_snapshot(
msg: dict[str, Any],
processes: list[dict[str, Any]],
show_tree: bool,
) -> None:
"""Print one process-tree snapshot."""
ts = (msg.get("time") or msg.get("timestamp") or "")[:19]
n = len(processes)
print(f"{ts} [{n} process{'es' if n != 1 else ''}]")
if not processes:
return
if show_tree:
children = _build_tree(processes)
pids = {p.get("pid") for p in processes}
# Find roots: processes whose ppid is not in the pid set.
roots = [p for p in processes if (p.get("ppid") or 0) not in pids]
if not roots:
roots = processes
def _walk(proc: dict[str, Any], depth: int) -> None:
indent = " " * depth
pid = proc.get("pid", "?")
name = proc.get("name", "?")
cpu_raw = proc.get("cpu_percentage")
mem_raw = proc.get("memory_percentage")
cpu = float(cpu_raw) if cpu_raw is not None else None
mem = float(mem_raw) if mem_raw is not None else None
cpu_s = f"{cpu:5.1f}%" if cpu is not None else " n/a"
mem_s = f"{mem:5.1f}%" if mem is not None else " n/a"
status = proc.get("state", "")
print(f" {indent}{pid:<8} {name:<24} cpu {cpu_s} mem {mem_s} {status}")
for child in children.get(pid, []):
_walk(child, depth + 1)
print(f" {'pid':<8} {'name':<24} {'cpu':>9} {'mem':>9} status")
print(" " + "-" * 64)
for root in roots:
_walk(root, 0)
else:
print(f" {'pid':<8} {'name':<24} {'cpu':>9} {'mem':>9} status")
print(" " + "-" * 64)
for proc in sorted(processes, key=lambda p: -float(p.get("cpu_percentage") or 0)):
pid = proc.get("pid", "?")
name = proc.get("name", "?")
cpu_raw = proc.get("cpu_percentage")
mem_raw = proc.get("memory_percentage")
cpu = float(cpu_raw) if cpu_raw is not None else None
mem = float(mem_raw) if mem_raw is not None else None
cpu_s = f"{cpu:5.1f}%" if cpu is not None else " n/a"
mem_s = f"{mem:5.1f}%" if mem is not None else " n/a"
status = proc.get("state", "")
print(f" {pid:<8} {name:<24} cpu {cpu_s} mem {mem_s} {status}")
print()
def _print_summary(
window_end: str,
snapshot_count: int,
proc_counts: list[int],
top_cpu_vals: list[tuple[float, str]],
) -> None:
"""Print a one-line rolling window summary."""
if proc_counts:
min_procs = min(proc_counts)
max_procs = max(proc_counts)
last_procs = proc_counts[-1]
procs_s = f"{last_procs} (min {min_procs} max {max_procs})"
else:
procs_s = "n/a"
if top_cpu_vals:
best_cpu, best_name = max(top_cpu_vals, key=lambda x: x[0])
top_s = f"{best_name} @ {best_cpu:.1f}%"
else:
top_s = "n/a"
print(f"{window_end} snapshots {snapshot_count:>4} procs {procs_s:<24} top cpu {top_s}")
def main() -> None:
args = _build_parser().parse_args()
# 1) Authenticate once with the top-level HPS client.
hps = Client(
args.base_url,
args.username,
args.password,
verify=not args.insecure,
)
# 2) Configure WebSocket options only when running in insecure local mode.
ws_options: dict[str, Any] | None = None
if args.insecure:
ws_options = {"sslopt": {"cert_reqs": ssl.CERT_NONE}}
# 3) Build a MonitorApi that reuses the authenticated client.
monitor = MonitorApi(
hps,
ws_connection_options=ws_options,
timeout_seconds=30.0,
)
print(f"Streaming process-tree snapshots for task {args.task_id}")
print("Press Ctrl+C to stop")
use_summary = args.interval > 0
if use_summary:
print(f"Rolling summary every {args.interval}s")
elif args.tree:
print("Showing tree view (child processes indented under parent)")
print("-" * 88)
snapshot_count = 0
proc_counts: list[int] = []
top_cpu_vals: list[tuple[float, str]] = []
window_start = time.monotonic()
try:
# 4) stream_task_process_tree subscribes to process_tree metric messages
# for the given task_id. Each yielded dict is one snapshot pushed by
# the evaluator on its reporting interval.
for msg in monitor.stream_task_process_tree(
task_id=args.task_id,
backlog=args.backlog,
max_messages=args.max_messages,
):
processes = _extract_processes(msg)
snapshot_count += 1
if use_summary:
proc_counts.append(len(processes))
if processes:
top = max(processes, key=lambda p: float(p.get("cpu_percentage") or 0))
cpu_raw = top.get("cpu_percentage")
cpu = float(cpu_raw) if cpu_raw is not None else None
name = top.get("name", "?")
if cpu is not None:
top_cpu_vals.append((cpu, name))
now = time.monotonic()
if now - window_start >= args.interval:
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
_print_summary(ts, snapshot_count, proc_counts, top_cpu_vals)
snapshot_count = 0
proc_counts = []
top_cpu_vals = []
window_start = now
else:
_print_snapshot(msg, processes, show_tree=args.tree)
except KeyboardInterrupt:
print("\nInterrupted by user.")
if use_summary and (proc_counts or snapshot_count):
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
_print_summary(ts, snapshot_count, proc_counts, top_cpu_vals)
if __name__ == "__main__":
main()
Stream scheduler job status#
This example shows how to stream job scheduler status metrics for a task
definition using MonitorApi.stream_scheduler_job_status(). Unlike
task-level methods such as stream_task_logs(), this method
takes a task definition ID rather than a task ID and requires no evaluator
resolution.
Background#
When HPS submits jobs to an external scheduler such as Slurm or LSF, the
autoscaling service (ansys.rep.scaling) periodically publishes a
scaler_instances metric message on the monitor bus. Each message reports
how many scheduler jobs are currently running, pending, and queued for a given
task definition.
MonitorApi.stream_scheduler_job_status() subscribes to messages with
the following tags:
Tag |
Value |
|---|---|
|
|
|
|
|
The task definition ID you provide. |
|
|
The message field of each yielded dictionary is a JSON string with the
following structure:
{
"running": 2,
"pending": 1,
"total": 3,
"scheduler": "slurm"
}
The example’s _extract_counts helper parses this payload and returns the
individual counts, handling the case where a field is absent or unparsable.
How the example works#
The script follows three steps:
Authenticate: call
Clientwith your username and password to obtain an access token.Create a
MonitorApi: passbase_urland the token. Noclient=argument is needed because scheduler job status does not require JMS or RMS evaluator resolution.Stream: iterate over
stream_scheduler_job_status(), printing each status update as it arrives. Press Ctrl+C to stop.
Command-line options#
Option |
Description |
|---|---|
|
Base URL of the HPS server (default: |
|
HPS username. |
|
HPS password. |
|
ID of the task definition to stream scheduler status for (required). |
|
Number of historical metric messages to replay on connect (default: 20). |
|
Stop after receiving this many messages. Omit for continuous streaming. |
|
Disable TLS certificate verification (required for local servers with self-signed certificates). |
Expected output#
Streaming scheduler job status for task definition taskdef-abc123
Press Ctrl+C to stop
time scheduler running pending total
--------------------------------------------------------------------------------
2026-06-29T10:04:11 slurm running 2 pending 1 total 3
2026-06-29T10:04:26 slurm running 3 pending 0 total 3
2026-06-29T10:04:41 slurm running 3 pending 0 total 3
Here is the stream_scheduler_job_status.py script for this example:
# Copyright (C) 2022 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Annotated example: stream scheduler job status with MonitorApi.stream_scheduler_job_status.
``stream_scheduler_job_status`` subscribes to ``scaler_instances`` metric messages
emitted by the HPS autoscaling service (``ansys.rep.scaling``) for a specific task
definition. Each message is a status update from the underlying job scheduler
(e.g. Slurm or LSF) and contains instance-count information for the scaler.
Usage (local dev with self-signed cert):
python examples/monitor/stream_scheduler_job_status.py \\
--base-url https://localhost:8443/hps \\
--username repadmin \\
--password repadmin \\
--task-definition-id <TASK_DEFINITION_ID> \\
--insecure
Press Ctrl+C to stop streaming.
"""
from __future__ import annotations
import argparse
import json
import ssl
from typing import Any
from ansys.hps.client import Client
from ansys.hps.client.monitor.api.monitor_api import MonitorApi
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Stream scheduler job status metrics for a task definition."
)
parser.add_argument("--base-url", default="https://localhost:8443/hps")
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument(
"--task-definition-id",
required=True,
help="Task definition ID to monitor scheduler job status for.",
)
parser.add_argument(
"--backlog",
type=int,
default=20,
help="Historical metric messages to request on connect (default: 20).",
)
parser.add_argument(
"--max-messages",
type=int,
default=None,
help="Maximum metric messages to receive. Omit for unlimited streaming.",
)
parser.add_argument(
"--insecure",
action="store_true",
help="Disable TLS certificate verification for local/self-signed endpoints.",
)
return parser
def _parse_payload(msg: dict[str, Any]) -> dict[str, Any]:
"""Extract the inner JSON payload from the ``message`` field.
Each ``scaler_instances`` metric message embeds the scheduler instance
record as a JSON string in the ``message`` field. The shape is::
{
"state": "SUBMITTED" | "RUNNING" | "COMPLETED" | "FAILED",
"orchestrator_id": "14072143",
"instance_id": "e2da1296-...",
"scaler_type": "slurm",
"resource_name": "oc-slurm",
"created_on": "2026-06-29T14:26:10...",
"started_on": "2026-06-29T18:26:18..." | null,
"keep": true,
"exit_state_checked": false,
"messages": [...],
...
}
"""
raw = msg.get("message", "")
if isinstance(raw, str):
try:
return json.loads(raw)
except (ValueError, TypeError):
return {}
return raw if isinstance(raw, dict) else {}
def _print_status(msg: dict[str, Any]) -> None:
"""Print a single scaler_instances message in a compact tabular form."""
ts = (msg.get("time") or msg.get("timestamp") or "")[:19]
payload = _parse_payload(msg)
state = payload.get("state") or "unknown"
scheduler = payload.get("scaler_type") or payload.get("resource_name") or "unknown"
job_id = payload.get("orchestrator_id") or "n/a"
started = (payload.get("started_on") or "")[:19] or "not started"
print(f"{ts} {scheduler:<10} {state:<12} job={job_id:<12} started={started}")
def main() -> None:
args = _build_parser().parse_args()
# 1) Authenticate once with the top-level HPS client.
hps = Client(
args.base_url,
args.username,
args.password,
verify=not args.insecure,
)
# 2) Configure WebSocket options only when running in insecure local mode.
ws_options: dict[str, Any] | None = None
if args.insecure:
ws_options = {"sslopt": {"cert_reqs": ssl.CERT_NONE}}
# 3) Build a MonitorApi that reuses the authenticated client.
monitor = MonitorApi(
hps,
ws_connection_options=ws_options,
timeout_seconds=30.0,
)
print(f"Streaming scheduler job status for task definition {args.task_definition_id}")
print("Press Ctrl+C to stop")
print(f"{'time':<19} {'scheduler':<10} {'state':<12} {'job id':<16} {'started'}")
print("-" * 80)
try:
# 4) stream_scheduler_job_status subscribes to scaler_instances metric
# messages for the given task definition. Each yielded dict is one
# status update from the job scheduler (e.g. Slurm or LSF).
for msg in monitor.stream_scheduler_job_status(
task_definition_id=args.task_definition_id,
backlog=args.backlog,
max_messages=args.max_messages,
):
_print_status(msg)
except KeyboardInterrupt:
print("\nInterrupted by user.")
if __name__ == "__main__":
main()
Stream service logs#
This example shows how to stream log messages from HPS backend services such as
the Job Management Service, autoscaling, or housekeeper using
MonitorApi.stream_service_logs(). Logs are delivered over a WebSocket
connection in real time, making this useful for debugging service behavior and
monitoring backend operations.
Background#
HPS runs several backend services that emit operational logs and diagnostics. These services include:
JMS (Job Management Service): Handles job lifecycle and execution
Scaling: Autoscaling decisions and instance management
Housekeeper: Cleanup and maintenance tasks
Evaluator: Evaluator process metrics and state transitions
MonitorApi.stream_service_logs() subscribes to messages from a specific
service using the client_type tag. Use ClientType constants for
well-known services, or pass a custom service identifier.
Each yielded message dictionary contains:
Key |
Content |
|---|---|
|
The log message text, or a structured object. |
|
ISO-8601 timestamp of when the message was logged. |
|
A nested dictionary with |
The backlog parameter controls how many historical messages are replayed
from the server’s message buffer on connection. Set it to a larger value such
as 500 to catch up on messages that were written before the subscription opened,
or to 0 to receive only new messages going forward.
Available services#
Service |
Purpose |
|---|---|
|
Job Management Service: job lifecycle and task execution |
|
RMS autoscaling: resource provisioning and instance events |
|
Cleanup and maintenance: temporary file removal and resource cleanup |
|
Evaluator metrics: process state, CPU or memory usage, and events |
Command-line options#
Option |
Description |
|---|---|
|
Base URL of the HPS server (default: |
|
HPS username. |
|
HPS password. |
|
Service to stream logs from: |
|
Number of historical messages to replay on connect (default: 100). |
|
Stop after receiving this many messages. Omit for continuous streaming. |
|
Print available services and exit. |
|
Disable TLS certificate verification (required for local servers with self-signed certificates). |
Example usage#
Stream JMS logs continuously:
python examples/monitor/stream_service_logs.py \
--base-url https://localhost:8443/hps \
--username repadmin \
--password repadmin \
--service jms \
--insecure
Stream scaling service logs with a 500-message backlog and stop after 100 messages:
python examples/monitor/stream_service_logs.py \
--base-url https://localhost:8443/hps \
--username repadmin \
--password repadmin \
--service scaling \
--backlog 500 \
--max-messages 100 \
--insecure
Expected output#
A typical run streaming JMS logs produces output like this:
Streaming logs from service: jms
Press Ctrl+C to stop
----------------------------------------------------------------------------------------
2026-06-29T14:26:10 Job 033VubFI0m4PYEREddzbN1 submitted successfully
2026-06-29T14:26:12 Assigned 2 evaluator instances for job 033VubFI0m4PYEREddzbN1
2026-06-29T14:26:15 Task 033VubFI0m4PYEREddzbN1 started on evaluator node-001
2026-06-29T14:27:03 Task 033VubFI0m4PYEREddzbN1 completed with exit code 0
...
Here is the stream_service_logs.py script for this example:
# Copyright (C) 2022 - 2026 Synopsys, Inc. and ANSYS, Inc. All rights reserved.
# SPDX-License-Identifier: MIT
#
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Stream HPS service logs with ``MonitorApi.stream_service_logs``.
This example shows how to stream log messages from HPS backend services such as
the Job Management Service (JMS), autoscaling service, or housekeeper tasks using
the monitor WebSocket.
Usage (stream JMS service logs on local/self-signed endpoint):
python examples/monitor/stream_service_logs.py \\
--base-url https://localhost:8443/hps \\
--username repadmin \\
--password repadmin \\
--service jms \\
--insecure
List available services:
python examples/monitor/stream_service_logs.py \\
--base-url https://localhost:8443/hps \\
--username repadmin \\
--password repadmin \\
--list-services \\
--insecure
"""
from __future__ import annotations
import argparse
import json
import ssl
from typing import Any
from ansys.hps.client import Client
from ansys.hps.client.monitor.api.monitor_api import ClientType, MonitorApi
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Stream logs from HPS backend services (JMS, scaling, housekeeper, etc)."
)
parser.add_argument("--base-url", default="https://localhost:8443/hps")
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument(
"--service",
default="jms",
choices=["jms", "scaling", "housekeeper", "evaluator"],
help="Service to stream logs from (default: jms).",
)
parser.add_argument(
"--backlog",
type=int,
default=100,
help="Historical messages to request on connect (default: 100).",
)
parser.add_argument(
"--max-messages",
type=int,
default=None,
help="Maximum messages to stream. Omit for unlimited streaming.",
)
parser.add_argument(
"--list-services",
action="store_true",
help="List available services and exit.",
)
parser.add_argument(
"--insecure",
action="store_true",
help="Disable TLS certificate verification for local/self-signed endpoints.",
)
return parser
def _get_client_type(service: str) -> str:
"""Map service name to ClientType constant."""
service_map = {
"jms": ClientType.JMS,
"scaling": ClientType.SCALING,
"housekeeper": ClientType.HOUSEKEEPER,
"evaluator": ClientType.EVALUATOR,
}
return service_map.get(service, ClientType.JMS)
def _to_text(message: dict[str, Any]) -> str:
"""Normalize monitor payload text for printing."""
text = message.get("message", "")
if isinstance(text, str):
return text
return json.dumps(text)
def _list_services() -> None:
"""Print available services."""
print("Available services:")
print(" jms - Job Management Service logs")
print(" scaling - RMS autoscaling service logs")
print(" housekeeper - Housekeeper task logs")
print(" evaluator - Evaluator state and metrics logs")
def main() -> None:
args = _build_parser().parse_args()
if args.list_services:
_list_services()
return
# 1) Authenticate once with the top-level HPS client.
hps = Client(
args.base_url,
args.username,
args.password,
verify=not args.insecure,
)
# 2) Configure websocket options only when running in insecure local mode.
ws_options: dict[str, Any] | None = None
if args.insecure:
ws_options = {"sslopt": {"cert_reqs": ssl.CERT_NONE}}
# 3) Reuse the same authenticated client in MonitorApi.
monitor = MonitorApi(
hps,
ws_connection_options=ws_options,
timeout_seconds=30.0,
)
client_type = _get_client_type(args.service)
print(f"Streaming logs from service: {args.service}")
print("Press Ctrl+C to stop")
print("-" * 88)
try:
# 4) stream_service_logs subscribes to the specified service by client_type.
for msg in monitor.stream_service_logs(
client_type=client_type,
backlog=args.backlog,
max_messages=args.max_messages,
):
ts = (msg.get("time") or msg.get("timestamp") or "")[:19]
line = _to_text(msg)
print(f"{ts} {line}")
except KeyboardInterrupt:
print("\nInterrupted by user.")
if __name__ == "__main__":
main()