Connection & Security

Every Ray Serve deployment that uses the KubeMQ adapter is configured through a single KubeMQAdapterConfig object. It controls where the adapter connects, how it identifies itself, and how the gRPC channel is secured. This guide walks through each connection and security option — from the zero-config default to mutual TLS — using the runnable examples that ship in the repository's examples/connection/ directory.

All examples assume a broker reachable at localhost:50000. Start one with Docker:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 50000 is the gRPC endpoint the adapter's SDK clients connect to. Port 9090 exposes the shared HTTP server used by connector endpoints such as KEDA-based queue scaling.

Prerequisites

• kubemq-rayserve installed (see Getting Started with Ray Serve)
• The broker above running and reachable
• A JWT token or TLS certificates on hand if your broker enforces authentication or TLS

Default Connection

A bare KubeMQAdapterConfig() targets localhost:50000 — the address field defaults to that value, so no arguments are required for local development. The example below builds the default config, initializes the adapter, and verifies the broker is reachable with a health_check() call.

import os
import uuid

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")


def main():
    channel = f"example-connection-{uuid.uuid4().hex[:8]}"

    # Default config connects to localhost:50000
    config = KubeMQAdapterConfig(address=BROKER)
    print(f"Config address: {config.address}")
    print(f"Config client_id: {config.client_id!r} (auto-generated if empty)")

    adapter = KubeMQTaskProcessorAdapter(config)

    class _Cfg:
        queue_name = channel
        max_retries = 0
        failed_task_queue_name = ""
        unprocessable_task_queue_name = ""

    adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())

    try:
        # Verify connectivity
        healthy = adapter.health_check()
        print(f"Health check: {healthy}")
    finally:
        adapter.stop_consumer()

The small _Cfg shim stands in for the TaskProcessorConfig that Ray Serve supplies in a real deployment. For standalone scripts like this one, the adapter only needs the four queue fields — queue_name, max_retries, failed_task_queue_name, and unprocessable_task_queue_name — to initialize.

Custom Broker Address

Set address='host:port' to point the adapter at a remote or in-cluster broker. The most common in-cluster value is kubemq:50000, the Kubernetes service name resolved inside the same namespace.

import os

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# Override the default with a custom broker address
config = KubeMQAdapterConfig(address="kubemq:50000")
print(f"Connecting to custom address: {config.address}")

adapter = KubeMQTaskProcessorAdapter(config)

Client Identification

Set client_id to give the connection a stable, recognizable name. The client ID appears in the broker logs and the web UI at port 9090, which makes it easy to tell workers apart when you run several replicas.

config = KubeMQAdapterConfig(
    address="localhost:50000",
    client_id="my-worker-01",
)
print(f"Config client_id: {config.client_id}")

When client_id is left empty (the default), the adapter generates a unique one in the form rayserve-{8hex} — a rayserve- prefix followed by the first eight characters of a UUID. This guarantees a distinct identity per connection even if you do not set one explicitly, but a fixed name is preferable for monitoring.

JWT Authentication

For brokers that require authentication, set auth_token to a JWT issued by your KubeMQ license. The adapter maps this value to the SDK's ClientConfig.auth_token, normalizing an empty string to None so an unset token is never sent as a blank credential.

import os

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

auth_token = os.environ.get("KUBEMQ_AUTH_TOKEN", "<your-jwt-token>")

config = KubeMQAdapterConfig(
    address="localhost:50000",
    auth_token=auth_token,
)

# auth_token is hidden in repr for security
print(f"Config: {config}")
print(f"Auth token set: {bool(config.auth_token)}")

The auth_token field is declared with repr=False, so it is omitted when the config object is printed or logged. Printing config shows every other field but never the token itself — important when configs end up in log lines or crash reports. Use the bool(config.auth_token) check shown above when you need to confirm a token is present without revealing it.

TLS

To encrypt the gRPC channel, set tls=True and point tls_ca_file at the CA certificate that signed the broker's certificate. The adapter assembles these fields into the SDK's TLSConfig.

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# TLS with the CA certificate that signed the broker's cert
config = KubeMQAdapterConfig(
    address="localhost:50000",
    tls=True,
    tls_ca_file="ca.pem",
)

print(f"tls={config.tls}")
print(f"tls_ca_file={config.tls_ca_file!r}")

adapter = KubeMQTaskProcessorAdapter(config)

Internally, _build_client_config() constructs a TLSConfig from the four TLS-related fields and passes it to ClientConfig:

from kubemq.core.config import TLSConfig

tls_config = TLSConfig(
    enabled=self._config.tls,
    cert_file=self._config.tls_cert_file or None,
    key_file=self._config.tls_key_file or None,
    ca_file=self._config.tls_ca_file or None,
)

Mutual TLS

For mutual TLS — where the broker also verifies the client's certificate — supply all three certificate paths in addition to tls=True: tls_cert_file (the client certificate), tls_key_file (the client private key), and tls_ca_file (the shared CA).

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# Full mTLS: client cert + key + CA
config = KubeMQAdapterConfig(
    address="localhost:50000",
    tls=True,
    tls_cert_file="client.pem",
    tls_key_file="client-key.pem",
    tls_ca_file="ca.pem",
)

print(f"tls_cert_file={config.tls_cert_file!r}")
print(f"tls_key_file={config.tls_key_file!r}")
print(f"tls_ca_file={config.tls_ca_file!r}")

adapter = KubeMQTaskProcessorAdapter(config)

The three files map directly onto the TLSConfig shown above: tls_cert_file becomes cert_file, tls_key_file becomes key_file, and tls_ca_file becomes ca_file. Each is normalized to None when left empty, so the same config shape covers plain TLS (CA only) and mutual TLS (CA plus client credentials).

The table below summarizes which fields each connection mode requires:

Health Checks

The adapter exposes two health-check methods, both backed by the SDK's ping() call against the broker:

• health_check() returns a single bool — True when the broker is reachable, False otherwise. Ideal for readiness and liveness probes.
• health_check_sync() returns a list with one status dict: [{"healthy": True}] on success, or [{"healthy": False, "error": "..."}] when the ping fails or the client is not initialized.

adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())

# Simple boolean health check
healthy = adapter.health_check()
print(f"health_check() -> {healthy}")

# Detailed health check returning a list of status dicts
detailed = adapter.health_check_sync()
print(f"health_check_sync() -> {detailed}")

# Use in a readiness probe pattern
if healthy:
    print("Broker is reachable — ready to process tasks.")
else:
    print("Broker is NOT reachable — check connection settings.")

health_check() is a thin convenience wrapper: it calls health_check_sync() and returns the healthy flag from the first entry. Use the boolean form for probes and the detailed form when you want the error string for logging.

Environment-Variable Configuration

For twelve-factor deployments, build the whole config from environment variables so the same image runs unchanged across environments. Every field has a sensible default, so unset variables fall back gracefully.

import os

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# Build config entirely from environment variables with defaults
config = KubeMQAdapterConfig(
    address=os.environ.get("KUBEMQ_ADDRESS", "localhost:50000"),
    client_id=os.environ.get("KUBEMQ_CLIENT_ID", ""),
    auth_token=os.environ.get("KUBEMQ_AUTH_TOKEN", ""),
    tls=os.environ.get("KUBEMQ_TLS", "false").lower() == "true",
    tls_ca_file=os.environ.get("KUBEMQ_TLS_CA_FILE", ""),
    tls_cert_file=os.environ.get("KUBEMQ_TLS_CERT_FILE", ""),
    tls_key_file=os.environ.get("KUBEMQ_TLS_KEY_FILE", ""),
)

print(f"address={config.address}")
print(f"client_id={config.client_id!r}")
print(f"auth_token={'[set]' if config.auth_token else '[not set]'}")
print(f"tls={config.tls}")

Note the tls line: the environment carries strings, so the example compares KUBEMQ_TLS against "true" to produce the boolean the config expects. In Kubernetes, these same variables are populated from a ConfigMap (address, client ID) and a Secret (auth token, certificate paths) — see the Kubernetes scenario below for the full Secret and ConfigMap wiring.

Related