orchestrata_ultimate.semantic_graph.ha_controller_cli

Reference › orchestrata_ultimate › semantic_graph › ha_controller_cli

The command-line entry point for the Memgraph HA controller in ha_controller, run as a Python module. Nodes are given as repeated --nodes arguments, each naming a node, its bolt URI and its replication host, or through the MEMGRAPH_HA_NODES environment variable as a comma-separated list. _parse_node turns one specification into a NodeConfig, _nodes_from_env parses the variable, _build_parser builds the argparse parser with flags defaulted from the environment mapping, and the async _main parses arguments and runs the controller together with its small HTTP API on the chosen port. The controller manages the replication of the semantic graph (memgraph database), not the operational orchkg graph. Nothing imports the module; it is launched directly.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/semantic_graph/ha_controller_cli.py

"""CLI entry point for the Memgraph HA controller.

Usage:
    python -m orchestrata_ultimate.semantic_graph.ha_controller_cli \
        --nodes 'memgraph-1=bolt://10.53.1.10:7687;10.53.1.10' \
        --nodes 'memgraph-2=bolt://10.53.1.11:7687;10.53.1.11' \
        --nodes 'memgraph-3=bolt://10.53.1.12:7687;10.53.1.12' \
        --port 8080

Environment variables (alternative to --nodes):
    MEMGRAPH_HA_NODES  comma-separated list of NAME|BOLT_URI|REPLICATION_HOST[:PORT]
"""

from __future__ import annotations

import argparse
import logging
import os
import sys
from collections.abc import Mapping

import anyio
import uvicorn

from orchestrata_ultimate.semantic_graph.ha_controller import (
    MemgraphHAController,
    NodeConfig,
)


def _parse_node(value: str) -> NodeConfig:
    """Parse a node specification.

    Formats:
        name=bolt_uri;replication_host
        name=bolt_uri;replication_host:port
        name|bolt_uri|replication_host
        name|bolt_uri|replication_host:port
    """
    if "=" in value:
        name, rest = value.split("=", 1)
        parts = rest.split(";")
    else:
        parts = value.split("|")
        name = parts.pop(0)

    if len(parts) < 2:
        raise ValueError(f"Invalid node spec: {value}")

    bolt_uri = parts[0]
    replication_spec = parts[1]
    if ":" in replication_spec:
        replication_host, port_str = replication_spec.rsplit(":", 1)
        replication_port = int(port_str)
    else:
        replication_host = replication_spec
        replication_port = 10000

    return NodeConfig(
        name=name,
        bolt_uri=bolt_uri,
        replication_host=replication_host,
        replication_port=replication_port,
    )


def _nodes_from_env(environ: Mapping[str, str] | None = None) -> list[NodeConfig]:
    """Parse ``MEMGRAPH_HA_NODES`` into node configs.

    Args:
        environ: Optional explicit environment mapping (for testing); when
            omitted, reads the real process environment.
    """
    env = os.environ if environ is None else environ
    raw = env.get("MEMGRAPH_HA_NODES", "")
    if not raw:
        return []
    return [_parse_node(item.strip()) for item in raw.split(",") if item.strip()]


def _build_parser(environ: Mapping[str, str] | None = None) -> argparse.ArgumentParser:
    """Build the CLI argument parser, defaulting flags from *environ*.

    Args:
        environ: Optional explicit environment mapping (for testing); when
            omitted, reads the real process environment.
    """
    env = os.environ if environ is None else environ
    parser = argparse.ArgumentParser(description="Memgraph HA controller")
    parser.add_argument(
        "--nodes",
        action="append",
        default=[],
        help="Node spec: name=bolt_uri;replication_host[:port] (repeatable)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=int(env.get("MEMGRAPH_HA_PORT", "8080")),
        help="HTTP API port (default: 8080)",
    )
    parser.add_argument(
        "--host",
        default=env.get("MEMGRAPH_HA_HOST", "0.0.0.0"),
        help="HTTP API host (default: 0.0.0.0)",
    )
    parser.add_argument(
        "--db-path",
        default=env.get("MEMGRAPH_HA_DB_PATH", "/var/lib/memgraph-ha/state.db"),
        help="SQLite state file path",
    )
    parser.add_argument(
        "--poll-interval",
        type=float,
        default=float(env.get("MEMGRAPH_HA_POLL_INTERVAL", "2.0")),
        help="Polling interval in seconds",
    )
    parser.add_argument(
        "--replication-mode",
        default=env.get("MEMGRAPH_HA_REPLICATION_MODE", "ASYNC"),
        choices=["ASYNC", "SYNC", "STRICT_SYNC"],
        help="Replication mode for replicas",
    )
    parser.add_argument(
        "--log-level",
        default=env.get("MEMGRAPH_HA_LOG_LEVEL", "INFO"),
        help="Logging level",
    )
    return parser


async def _main(
    argv: list[str] | None = None, environ: Mapping[str, str] | None = None
) -> None:
    """Parse args and run the HA controller + its HTTP API.

    Args:
        argv: CLI argv (defaults to ``sys.argv[1:]`` via argparse).
        environ: Optional explicit environment mapping (for testing); when
            omitted, reads the real process environment.
    """
    parser = _build_parser(environ)
    args = parser.parse_args(argv)

    logging.basicConfig(
        level=getattr(logging, args.log_level.upper()),
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

    nodes = (
        [_parse_node(n) for n in args.nodes] if args.nodes else _nodes_from_env(environ)
    )
    if len(nodes) < 3:
        print(
            "At least 3 Memgraph nodes are required for HA. "
            "Use --nodes or set MEMGRAPH_HA_NODES.",
            file=sys.stderr,
        )
        sys.exit(1)

    controller = MemgraphHAController(
        nodes=nodes,
        db_path=args.db_path,
        poll_interval=args.poll_interval,
        replication_mode=args.replication_mode,
    )

    # http="h11": uvicorn's "auto" protocol imports httptools, which
    # re-enables the GIL process-wide (SPEC-023). Same pin as ast_daemon.
    config = uvicorn.Config(
        controller.app,
        host=args.host,
        port=args.port,
        log_level=args.log_level.lower(),
        http="h11",
    )
    server = uvicorn.Server(config)

    async with anyio.create_task_group() as tg:
        tg.start_soon(controller.run)
        tg.start_soon(server.serve)


if __name__ == "__main__":
    anyio.run(_main)