How to monitor Solace as an external consumer

I hope this helps explain what I'd like to achieve.

Keen to get some feedback etc, thanks in advance.

I've setup a local docker subpub+ using the starter guide and running the code below to test out some queries.

Note the following code is very rough and just for initial learning:

## Check the message VPN and Auth in the process

def check_message_vpn():
    apiUrl = f"http://{primaryBroker}:8080/SEMP/v2/__private_monitor__/msgVpns/{messageVpnName}?select=state"
    response = requests.get(apiUrl, auth=(apiUsername, apiPassword), verify=False)
    print(response.json())  
    try:
        response = requests.get(apiUrl, auth=(apiUsername, apiPassword), verify=False)
        if response.status_code == 200:
            vpn_state = response.json().get("data", {}).get("state", "unknown")

            if vpn_state == "up":
                print(f"Message VPN '{messageVpnName}' is up and responding correctly.")
            else:
                print(
                    f"Message VPN '{messageVpnName}' is not up. Current state: {vpn_state}"
                )
        elif response.status_code == 401:
            print("Authentication failed. Please check your username and password.")
        else:
            print(
                f"Failed to fetch the state of the message VPN. Status code: {response.status_code}"
            )
            print("Response content:", response.text)
    except requests.exceptions.RequestException as e:
        print(f"An error occurred: {e}")


## Check the queue

def check_queue_health():
    apiUrl = f"http://{primaryBroker}:8080/SEMP/v2/config/msgVpns/{messageVpnName}/queues/{queueName}"
    try:
        response = requests.get(apiUrl, auth=(apiUsername, apiPassword), verify=False)
        if response.status_code == 200:
            queue_info = response.json().get("data", {})
            ingress_enabled = queue_info.get("ingressEnabled", "unknown")
            egress_enabled = queue_info.get("egressEnabled", "unknown")
            if ingress_enabled and egress_enabled:
                print(f"Queue '{queueName}' is up and processing messages.")
            else:
                print(
                    f"Queue '{queueName}' is not processing messages correctly. Ingress: {ingress_enabled}, Egress: {egress_enabled}"
                )
        elif response.status_code == 401:
            print("Authentication failed. Please check your username and password.")
        else:
            print(
                f"Failed to fetch the state of the queue. Status code: {response.status_code}"
            )
            print("Response content:", response.text)
    except requests.exceptions.RequestException as e:
        print(f"An error occurred: {e}")

Thanks so much for this detailed response. It's appreciated!

I'll digest this and respond shortly. Thanks again!

I've tested out a way to capture service event metrics.

Added the service_interupted boolean to the service event handler class:

class SubServiceEventHandler(
    ReconnectionListener, ReconnectionAttemptListener, ServiceInterruptionListener
):
    def __init__(self):
        self.service_interrupted = False

    def on_reconnected(self, e: ServiceEvent):
        …
        …
        self.service_interrupted = False

    def on_reconnecting(self, e: ServiceEvent):
        …
        …

        self.service_interrupted = True

    def on_service_interrupted(self, e: ServiceEvent):
        …

        …
        self.service_interrupted = True

Add a function to check the value of the service_interupted variable
A thread to check the status every minute.

receiver.start()

# Set the message handler
message_handler = MyMessageHandler()
receiver.receive_async(message_handler)

# Function to check the service status every minute
def check_service_status():
    while True:
        if service_event_handler.service_interrupted:
            print("Service is interrupted")
        else:
            print("Service is running normally")
        time.sleep(60)


# Start a thread to check the service status every minute
status_thread = threading.Thread(target=check_service_status)
status_thread.daemon = True
status_thread.start()

To test this the Solace docker container has come in handy!

When working:

Publisher

Sent 'Hello, Solace! aANdP' to queue 'foo'
Message published
Sent 'Hello, Solace! DgNsT' to queue 'foo'
Message published

Subscriber
Received message: Hello, Solace! aANdP on queue foo
Service is running normally
Received message: Hello, Solace! DgNsT on queue foo
Service is running normally

When the docker container is stopped/started:

Subscriber


Received message: Hello, Solace! eHWRN on queue foo
2025-01-22 15:50:29,057 [WARNING] solace.messaging.core: [_solace_transport.py:89]  [[SERVICE: 0x7f9c8b49ba60] - [APP ID: hq/656553/00000001/geywCErEZj]] {'caller_description': 'From service event callback', 'return_code': 'Ok', 'sub_code': 'SOLCLIENT_SUBCODE_COMMUNICATION_ERROR', 'error_info_sub_code': 14, 'error_info_contents': 'Peer closed socket, fd 8, cannot read'}

on_reconnecting
Attempting to reconnect. Error cause: {'caller_description': 'From service event callback', 'return_code': 'Ok', 'sub_code': 'SOLCLIENT_SUBCODE_COMMUNICATION_ERROR', 'error_info_sub_code': 14, 'error_info_contents': 'Peer closed socket, fd 8, cannot read'}
Message: Peer closed socket, fd 8, cannot read
Service is interrupted
Service is interrupted
Service is interrupted
2025-01-22 15:53:09,865 [WARNING] solace.messaging.core: [_solace_transport.py:89]  [[SERVICE: 0x7f9c8b49ba60] - [APP ID: hq/656553/00000001/geywCErEZj]] {'caller_description': 'From service event callback', 'return_code': 'Ok', 'sub_code': 'SOLCLIENT_SUBCODE_COMMUNICATION_ERROR', 'error_info_sub_code': 14, 'error_info_contents': 'Peer closed socket, fd 8, cannot read'}

on_reconnected
Error cause: {'caller_description': 'From service event callback', 'return_code': 'Ok', 'sub_code': 'SOLCLIENT_SUBCODE_COMMUNICATION_ERROR', 'error_info_sub_code': 14, 'error_info_contents': 'Peer closed socket, fd 8, cannot read'}
Message: host 'tcp://localhost:55555', hostname 'localhost:55555' IP [::1]:55555 (host 1 of 1) (host connection attempt 1 of 1) (total reconnection attempt 1 of -1)
Service is running normally
Service is running normally
Received message: Hello, Solace! HIAQw on queue foo

While this has been fun to play around with it perhaps doesn't add anything that monitoring the bind count at the broker level?

Still can't find the docs on the session/flow/active 🔍️

Cheers