NFCLI Client API

PICLE Shell CLient

Client that implements interactive shell to work with NorFab.

ShowNorfabWorkersModel

Bases: BaseModel

run(*args: object, **kwargs: object) -> object staticmethod

Show broker-known workers with live process metrics, grouped by service.

Source code in norfab\clients\nfcli_shell\nfcli_shell_client.py

@staticmethod
def run(*args: object, **kwargs: object) -> object:
    """Show broker-known workers with live process metrics, grouped by service."""
    nfclient = getattr(builtins, "NFCLIENT", NFCLIENT)
    reply = nfclient.mmi("mmi.service.broker", "show_workers")
    if reply["errors"]:
        return "\n".join(reply["errors"])

    workers = [worker for worker in reply["results"] if worker["name"]]
    alive = [worker["name"] for worker in workers if worker["status"] == "alive"]
    results = (
        nfclient.run_job(
            "all", "get_stats", workers=alive, timeout=kwargs.get("timeout", 10)
        )
        if alive
        else {}
    ) or {}
    rows = []
    for worker in sorted(workers, key=lambda item: (item["service"], item["name"])):
        result = results.get(worker["name"], {})
        if result.get("failed"):
            log.error(
                "worker '%s' statistics failed: %s",
                worker["name"],
                result.get("errors"),
            )
        stats = {} if result.get("failed") else (result.get("result") or {})
        process = stats.get("process", {})
        rows.append(
            {
                "Worker": worker["name"],
                "Service": worker["service"],
                "Status": worker["status"],
                "Uptime": (
                    str(timedelta(seconds=process["uptime_seconds"]))
                    if process
                    else "-"
                ),
                "RAM MB": (
                    round(process["memory_rss_mbyte"])
                    if process
                    else "-"
                ),
                "CPU": process.get("cpu_percent", "-"),
                "Jobs": stats.get("jobs", {}).get("total_jobs", "-"),
            }
        )
    return (
        rows,
        Outputters.outputter_tabulate_table,
        {"showindex": True, "tablefmt": "simple"},
    )

NorFabShell

Bases: BaseModel

cmd_preloop_override() -> None classmethod

This method called before CMD loop starts

Source code in norfab\clients\nfcli_shell\nfcli_shell_client.py

@classmethod
def cmd_preloop_override(self) -> None:
    """This method called before CMD loop starts"""
    pass

mount_shell_plugins(shell: App, inventory: object) -> None

Mounts shell plugins to the given shell application.

This function iterates over the plugins in the inventory and mounts those that have an "nfcli" configuration to the shell application.

Parameters:

Name	Type	Description	Default
shell	App	The shell application to which the plugins will be mounted.	required
inventory	object	An object containing the plugins to be mounted. It should have an attribute plugins which is a dictionary where keys are service names and values are service data dictionaries.	required

Returns:

Type	Description
None	None

Source code in norfab\clients\nfcli_shell\nfcli_shell_client.py

def mount_shell_plugins(shell: App, inventory: object) -> None:
    """
    Mounts shell plugins to the given shell application.

    This function iterates over the plugins in the inventory and mounts
    those that have an "nfcli" configuration to the shell application.

    Args:
        shell (App): The shell application to which the plugins will be mounted.
        inventory (object): An object containing the plugins to be mounted.
                            It should have an attribute `plugins` which is a dictionary
                            where keys are service names and values are service data dictionaries.

    Returns:
        None
    """
    for service_name, service_data in inventory.plugins.items():
        if service_data.get("nfcli"):
            plugin = inventory.load_plugin(service_name)
            shell.model_mount(
                path=plugin[service_name]["nfcli"]["mount_path"],
                model=plugin[service_name]["nfcli"]["shell_model"],
            )

    # mount MAN commands
    shell.model_mount(
        path=["man", "tasks"],
        model=ManTasks,
        description="SHow NorFab services tasks documentation",
    )