Plugin System and Extensions

The plugin system in dapcli enables the extension of the CLI by allowing users to build, upload, list, and delete custom plugins (wagons) that interact with the DAP orchestrator. The architecture separates the local build process, which ensures ABI compatibility via Docker, from the remote lifecycle management, which handles plugin registration and state polling via REST APIs. This section details the client interface for orchestrator communication, the Docker-driven build process for creating compatible wagon archives, and the CLI command handlers that orchestrate these operations.

Plugin Client Interface

The plugin client, located in src/dap/plugin/client.py, provides the REST surface for interacting with the orchestrator’s plugin endpoints. It exposes functions for uploading, listing, and deleting plugins, as well as polling for upload completion.

Upload: The upload_plugin function POSTs the raw wagon archive to /api/v3.1/plugins with Content-Type: application/octet-stream and a visibility query parameter. It returns the plugin record, which may initially have a pending or uploading state for large files ¹.
Polling: The wait_plugin_uploaded function polls /api/v3.1/plugins/<id> until the state reaches uploaded or fails. It raises a DapError if the state becomes failed_uploading or invalid ².
List/Delete: list_plugins retrieves plugins via GET /api/v3.1/plugins, while delete_plugin handles deletion via DELETE /api/v3.1/plugins/<id>, treating 404 responses as success for idempotency.

The client uses open_orchestrator_client to manage sessions and authentication, supporting both user sessions and admin bootstrap results ¹.

Build Process and Wagon Creation

The build_wagon function in src/dap/plugin/client.py ensures that plugins are built with an ABI compatible with the DAP mgmtworker. It achieves this by running the build process inside a manylinux Docker container.

Containerization: By default, it uses quay.io/pypa/manylinux2014_x86_64 with Python 3.11. This can be overridden via CLI flags or environment variables like DAP_PLUGIN_BUILD_IMAGE.
Build Steps: The function creates a temporary stage directory, copies the source (excluding VCS and cache directories), and runs a shell command inside the container. This command installs build dependencies (e.g., wagon, build, setuptools), optionally runs a fetch_govc.py script, and then invokes wagon create to generate the .wgn file ².
Output: The resulting .wgn file and plugin.yaml are copied to the output directory. The function requires Docker (or Podman) to be available on the host PATH.

Lifecycle Management and CLI Commands

The CLI commands in src/dap/plugin/commands.py provide the user interface for plugin management. They resolve authentication, invoke the client functions, and handle user feedback.

Authentication: The _resolve_auth helper attempts to load a user session first, falling back to persisted admin credentials. If neither is available, it raises an error prompting the user to log in ³.
Upload Command: The dap plugin upload command accepts a wagon path, visibility, and TLS options. It calls upload_plugin and, unless --no-wait is specified, calls wait_plugin_uploaded to confirm the plugin is fully processed ⁴.
Build Command: The dap plugin build command takes a source directory, resolves build parameters (image, Python version, platform) from CLI args or environment variables, and calls build_wagon. It then prints the path to the built wagon, suggesting the subsequent upload command.

src/dap/plugin/client.py L1-120 (showing 40 of 120)

"""HTTP + build helpers for ``dap plugin``.

The orchestrator's plugin upload endpoint takes the wagon as the raw
request body (Content-Type: application/octet-stream) and ``visibility``
as a query parameter; the orchestrator reads ``plugin.yaml`` from inside
the wagon. ``build_wagon`` runs the wagon toolchain inside a manylinux
Docker container so the produced wagon's ABI tag matches the orchestrator's
mgmtworker regardless of the developer's host architecture.
"""

from __future__ import annotations

import shutil
import subprocess
import tarfile
import tempfile
import time
import uuid
from pathlib import Path
from typing import Any

from ..auth.session import Session
from ..bootstrap import BootstrapResult
from ..config import ClusterConfig
from ..errors import DapError
from ..orch_client import open_orchestrator_client

PLUGIN_POLL_SECONDS = 2
PLUGIN_POLL_TIMEOUT = 300

# manylinux base image whose cpython interpreter matches the DAP
# mgmtworker. Override via ``--container`` on the CLI or by setting
# ``DAP_PLUGIN_BUILD_IMAGE`` in the environment.
DEFAULT_BUILD_IMAGE = "quay.io/pypa/manylinux2014_x86_64"
DEFAULT_PYTHON_VERSION = "311"
DEFAULT_PLATFORM = "manylinux2014_x86_64"
DEFAULT_BUILD_PINS = (
    "wagon[dist]==1.0.3",
    "build==1.5.0",
    "setuptools==80.9.0",

src/dap/plugin/client.py L121-240 (showing 40 of 120)

                    raise DapError(f"plugin upload failed: state={state} error={body.get('error') or '<unknown>'}")
                last_state = state
            time.sleep(PLUGIN_POLL_SECONDS)
    raise DapError(f"plugin {plugin_id} did not reach state=uploaded within {timeout}s (last state={last_state!r})")


def list_plugins(
    config: ClusterConfig,
    *,
    size: int = 100,
    session: Session | None = None,
    admin: BootstrapResult | None = None,
    insecure: bool = True,
) -> list[dict[str, Any]]:
    """GET ``/api/v3.1/plugins?_size=<n>``. Returns the items list."""
    with open_orchestrator_client(config, session=session, admin=admin, insecure=insecure) as cx:
        r = cx.get(f"/api/v3.1/plugins?_size={size}")
    if r.status_code != 200:
        raise DapError(f"plugin list failed: HTTP {r.status_code} {r.text[:300]}")
    return r.json().get("items", [])


def delete_plugin(
    config: ClusterConfig,
    plugin_id: str,
    *,
    force: bool = False,
    session: Session | None = None,
    admin: BootstrapResult | None = None,
    insecure: bool = True,
) -> None:
    """DELETE ``/api/v3.1/plugins/<id>``. 404 is treated as success
    (idempotent delete).
    """
    params: dict[str, str] = {"force": "true"} if force else {}
    with open_orchestrator_client(config, session=session, admin=admin, insecure=insecure) as cx:
        r = cx.delete(f"/api/v3.1/plugins/{plugin_id}", params=params)
    if r.status_code == 404:
        return
    if r.status_code not in (200, 204):

src/dap/plugin/commands.py L1-120 (showing 40 of 120)

"""``dap plugin`` subcommand surface."""

from __future__ import annotations

import os
from pathlib import Path

import click

from ..auth.session import load_session
from ..bootstrap import load_persisted
from ..config import load_config
from ..errors import DapError
from .client import (
    DEFAULT_BUILD_IMAGE,
    DEFAULT_PLATFORM,
    DEFAULT_PYTHON_VERSION,
    build_wagon,
    delete_plugin,
    list_plugins,
    upload_plugin,
    wait_plugin_uploaded,
)


def _resolve_auth(cluster_name: str):
    try:
        return load_session(cluster_name), None
    except DapError:
        pass
    try:
        return None, load_persisted(cluster_name)
    except DapError as e:
        raise DapError("no auth available. Run `dap auth login` or `dap auth import-client-credentials` first.") from e


@click.group("plugin")
def plugin() -> None:
    """Build, upload, list, and delete orchestrator plugins (.wgn)."""

src/dap/plugin/commands.py L121-156

@click.option("--runtime", default="docker", show_default=True, type=click.Choice(["docker", "podman"]))
@click.option("--timeout", default=1200, show_default=True, help="Build timeout (seconds).")
def plugin_build(
    source_dir: str,
    output_dir: str | None,
    container: str | None,
    python_version: str | None,
    platform: str | None,
    runtime: str,
    timeout: int,
) -> None:
    """Build a wagon (.wgn) for a plugin source dir inside a manylinux container.

    The container matches the DAP mgmtworker's Python ABI by default. On a
    non-x86_64 host, register QEMU binfmt first:

        docker run --privileged --rm tonistiigi/binfmt --install amd64
    """
    src = Path(source_dir)
    dst = Path(output_dir) if output_dir else src / "dist"
    image = container or os.environ.get("DAP_PLUGIN_BUILD_IMAGE", DEFAULT_BUILD_IMAGE)
    py = python_version or os.environ.get("DAP_PLUGIN_BUILD_PYTHON", DEFAULT_PYTHON_VERSION)
    plat = platform or os.environ.get("DAP_PLUGIN_BUILD_PLATFORM", DEFAULT_PLATFORM)
    click.echo(f"building wagon from {src} (image={image}, py={py}, platform={plat}) ...")
    wagon = build_wagon(
        src,
        dst,
        container=image,
        python_version=py,
        platform=plat,
        runtime=runtime,
        timeout=float(timeout),
    )
    click.echo(f"wagon built: {wagon} ({wagon.stat().st_size:,} bytes)")
    click.echo(f"upload with: dap plugin upload {wagon}")