besu_launcher.star

shared_utils = import_module("../../shared_utils/shared_utils.star")
input_parser = import_module("../../package_io/input_parser.star")
el_context = import_module("../../el/el_context.star")
el_admin_node_info = import_module("../../el/el_admin_node_info.star")
el_shared = import_module("../el_shared.star")
node_metrics = import_module("../../node_metrics_info.star")
constants = import_module("../../package_io/constants.star")

# The dirpath of the execution data directory on the client container
EXECUTION_DATA_DIRPATH_ON_CLIENT_CONTAINER = "/data/besu/execution-data"

METRICS_PATH = "/metrics"

RPC_PORT_NUM = 8545
WS_PORT_NUM = 8546
DISCOVERY_PORT_NUM = 30303
ENGINE_HTTP_RPC_PORT_NUM = 8551
METRICS_PORT_NUM = 9001

JAVA_OPTS = {"JAVA_OPTS": "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n"}

ENTRYPOINT_ARGS = ["sh", "-c"]

VERBOSITY_LEVELS = {
    constants.GLOBAL_LOG_LEVEL.error: "ERROR",
    constants.GLOBAL_LOG_LEVEL.warn: "WARN",
    constants.GLOBAL_LOG_LEVEL.info: "INFO",
    constants.GLOBAL_LOG_LEVEL.debug: "DEBUG",
    constants.GLOBAL_LOG_LEVEL.trace: "TRACE",
    constants.GLOBAL_LOG_LEVEL.custom: "CUSTOM",
}


def launch(
    plan,
    launcher,
    service_name,
    participant,
    global_log_level,
    existing_el_clients,
    persistent,
    tolerations,
    node_selectors,
    port_publisher,
    participant_index,
    network_params,
    extra_files_artifacts,
    bootnodoor_el_enr=None,
    el_binary_artifact=None,
    otel_otlp_grpc_url=None,
):
    cl_client_name = service_name.split("-")[3]

    config = get_config(
        plan,
        launcher,
        participant,
        service_name,
        existing_el_clients,
        cl_client_name,
        global_log_level,
        persistent,
        tolerations,
        node_selectors,
        port_publisher,
        participant_index,
        network_params,
        extra_files_artifacts,
        bootnodoor_el_enr,
        el_binary_artifact,
        otel_otlp_grpc_url,
    )

    service = plan.add_service(
        service_name, config, force_update=participant.el_force_restart
    )

    return get_el_context(
        plan,
        service_name,
        service,
        launcher,
    )


def get_config(
    plan,
    launcher,
    participant,
    service_name,
    existing_el_clients,
    cl_client_name,
    global_log_level,
    persistent,
    tolerations,
    node_selectors,
    port_publisher,
    participant_index,
    network_params,
    extra_files_artifacts,
    bootnodoor_el_enr=None,
    el_binary_artifact=None,
    otel_otlp_grpc_url=None,
):
    log_level = input_parser.get_client_log_level_or_default(
        participant.el_log_level, global_log_level, VERBOSITY_LEVELS
    )

    public_ports = {}
    public_ports_for_component = None
    if port_publisher.el_enabled:
        public_ports_for_component = shared_utils.get_public_ports_for_component(
            "el", port_publisher, participant_index
        )
        public_ports = el_shared.get_general_el_public_port_specs(
            public_ports_for_component
        )
        additional_public_port_assignments = {
            constants.RPC_PORT_ID: public_ports_for_component[3],
            constants.WS_PORT_ID: public_ports_for_component[4],
        }
        public_ports.update(
            shared_utils.get_port_specs(additional_public_port_assignments)
        )

    discovery_port_tcp = (
        public_ports_for_component[0]
        if public_ports_for_component
        else DISCOVERY_PORT_NUM
    )
    discovery_port_udp = (
        public_ports_for_component[0]
        if public_ports_for_component
        else DISCOVERY_PORT_NUM
    )

    used_port_assignments = {
        constants.TCP_DISCOVERY_PORT_ID: discovery_port_tcp,
        constants.UDP_DISCOVERY_PORT_ID: discovery_port_udp,
        constants.ENGINE_RPC_PORT_ID: ENGINE_HTTP_RPC_PORT_NUM,
        constants.METRICS_PORT_ID: METRICS_PORT_NUM,
        constants.RPC_PORT_ID: RPC_PORT_NUM,
        constants.WS_PORT_ID: WS_PORT_NUM,
    }
    used_ports = shared_utils.get_port_specs(used_port_assignments)

    cmd = [
        "besu",
    ]

    if log_level != "CUSTOM":
        cmd.append("--logging=" + log_level)

    cmd += [
        "--data-path=" + EXECUTION_DATA_DIRPATH_ON_CLIENT_CONTAINER,
        "--host-allowlist=*",
        "--rpc-http-enabled=true",
        "--rpc-http-host=0.0.0.0",
        "--rpc-http-port={0}".format(RPC_PORT_NUM),
        "--rpc-http-api=ADMIN,ETH,NET,DEBUG,TXPOOL,ENGINE,TRACE,WEB3",
        "--rpc-http-cors-origins=*",
        "--rpc-http-max-active-connections=300",
        "--rpc-ws-enabled=true",
        "--rpc-ws-host=0.0.0.0",
        "--rpc-ws-port={0}".format(WS_PORT_NUM),
        "--rpc-ws-api=ADMIN,ETH,NET,DEBUG,TXPOOL,ENGINE,TRACE,WEB3",
        "--p2p-enabled=true",
        "--p2p-host=" + port_publisher.el_nat_exit_ip,
        "--p2p-port={0}".format(discovery_port_tcp),
        "--discovery-mode=V5",
        "--engine-rpc-enabled=true",
        "--engine-jwt-secret=" + constants.JWT_MOUNT_PATH_ON_CONTAINER,
        "--engine-host-allowlist=*",
        "--engine-rpc-port={0}".format(ENGINE_HTTP_RPC_PORT_NUM),
        "{0}".format(
            "--sync-mode=FULL"
            if network_params.network in constants.NETWORK_NAME.kurtosis
            or participant.el_storage_type == "archive"
            else "--sync-mode=SNAP"
        ),
        "--metrics-enabled=true",
        "--metrics-host=0.0.0.0",
        "--metrics-port={0}".format(METRICS_PORT_NUM),
        "--min-gas-price=1000000000",
    ]

    # Configure storage type - besu defaults to full (BONSAI), use FOREST for archive
    if participant.el_storage_type == "archive":
        cmd.append("--data-storage-format=FOREST")

    if network_params.gas_limit > 0:
        cmd.append("--target-gas-limit={0}".format(network_params.gas_limit))

    if network_params.network not in constants.PUBLIC_NETWORKS:
        cmd.append(
            "--genesis-file="
            + constants.GENESIS_CONFIG_MOUNT_PATH_ON_CONTAINER
            + "/besu.json"
        )
    else:
        cmd.append("--network=" + network_params.network)

    # Handle bootnode configuration with bootnodoor_el_enr override
    if bootnodoor_el_enr != None:
        cmd.append("--bootnodes=" + bootnodoor_el_enr)
    elif (
        network_params.network == constants.NETWORK_NAME.kurtosis
        or constants.NETWORK_NAME.shadowfork in network_params.network
    ):
        el_bootnode_enrs = [
            ctx.enr
            for ctx in existing_el_clients[: constants.MAX_ENODE_ENTRIES]
            if ctx.enr
        ]
        if len(el_bootnode_enrs) > 0:
            cmd.append("--bootnodes=" + ",".join(el_bootnode_enrs))
    elif (
        network_params.network not in constants.PUBLIC_NETWORKS
        and constants.NETWORK_NAME.shadowfork not in network_params.network
    ):
        cmd.append(
            "--bootnodes="
            + shared_utils.get_devnet_el_enrs(
                plan, launcher.el_cl_genesis_data.files_artifact_uuid
            )
        )

    if len(participant.el_extra_params) > 0:
        # we do this as extra_params isn't a normal [] but a proto repeated array
        cmd.extend([param for param in participant.el_extra_params])

    cmd_str = " ".join(cmd)

    env_vars = (
        shared_utils.with_otel_env_vars(
            participant.el_extra_env_vars,
            otel_otlp_grpc_url,
            service_name,
        )
        | JAVA_OPTS
    )

    files = {
        constants.GENESIS_DATA_MOUNTPOINT_ON_CLIENTS: launcher.el_cl_genesis_data.files_artifact_uuid,
        constants.JWT_MOUNTPOINT_ON_CLIENTS: launcher.jwt_file,
    }

    if persistent:
        volume_size_key = (
            "devnets" if "devnet" in network_params.network else network_params.network
        )
        files[EXECUTION_DATA_DIRPATH_ON_CLIENT_CONTAINER] = Directory(
            persistent_key="data-{0}".format(service_name),
            size=int(participant.el_volume_size)
            if int(participant.el_volume_size) > 0
            else constants.VOLUME_SIZE[volume_size_key][
                constants.EL_TYPE.besu + "_volume_size"
            ],
        )

    # Add extra mounts - automatically handle file uploads
    processed_mounts = shared_utils.process_extra_mounts(
        plan, participant.el_extra_mounts, extra_files_artifacts
    )
    for mount_path, artifact in processed_mounts.items():
        files[mount_path] = artifact

    # Binary injection - mount custom binary directory if provided
    if el_binary_artifact != None:
        files["/opt/bin"] = el_binary_artifact.artifact
        # Copy injected binary to override default, then run original command
        final_cmd_str = (
            "cp /opt/bin/{0} /opt/besu/bin/besu && ".format(el_binary_artifact.filename)
            + cmd_str
        )
    else:
        final_cmd_str = cmd_str

    config_args = {
        "image": participant.el_image,
        "ports": used_ports,
        "public_ports": public_ports,
        "publish_udp": port_publisher.el_enabled,
        "cmd": [final_cmd_str],
        "files": files,
        "entrypoint": ENTRYPOINT_ARGS,
        "private_ip_address_placeholder": constants.PRIVATE_IP_ADDRESS_PLACEHOLDER,
        "env_vars": env_vars,
        "labels": shared_utils.label_maker(
            client=constants.EL_TYPE.besu,
            client_type=constants.CLIENT_TYPES.el,
            image=participant.el_image[-constants.MAX_LABEL_LENGTH :],
            connected_client=cl_client_name,
            extra_labels=participant.el_extra_labels
            | {constants.NODE_INDEX_LABEL_KEY: str(participant_index + 1)},
            supernode=participant.supernode,
        ),
        "user": User(uid=0, gid=0),
        "tolerations": tolerations,
        "node_selectors": node_selectors,
    }

    if participant.el_min_cpu > 0:
        config_args["min_cpu"] = participant.el_min_cpu
    if participant.el_max_cpu > 0:
        config_args["max_cpu"] = participant.el_max_cpu
    if participant.el_min_mem > 0:
        config_args["min_memory"] = participant.el_min_mem
    if participant.el_max_mem > 0:
        config_args["max_memory"] = participant.el_max_mem
    if len(participant.el_devices) > 0:
        config_args["devices"] = participant.el_devices
    return ServiceConfig(**config_args)


# makes request to [service_name] for enode and returns a full el_context
def get_el_context(
    plan,
    service_name,
    service,
    launcher,
):
    enode, enr = el_admin_node_info.get_enode_enr_for_node(
        plan, service_name, constants.RPC_PORT_ID
    )

    metrics_url = "{0}:{1}".format(service.name, METRICS_PORT_NUM)
    besu_metrics_info = node_metrics.new_node_metrics_info(
        service_name, METRICS_PATH, metrics_url
    )
    http_url = "http://{0}:{1}".format(service.name, RPC_PORT_NUM)
    ws_url = "ws://{0}:{1}".format(service.name, WS_PORT_NUM)

    return el_context.new_el_context(
        client_name="besu",
        enode=enode,
        dns_name=service.name,
        rpc_port_num=RPC_PORT_NUM,
        ws_port_num=WS_PORT_NUM,
        engine_rpc_port_num=ENGINE_HTTP_RPC_PORT_NUM,
        rpc_http_url=http_url,
        ws_url=ws_url,
        service_name=service_name,
        el_metrics_info=[besu_metrics_info],
        ip_addr=service.ip_address,
        enr=enr,
    )


def new_besu_launcher(el_cl_genesis_data, jwt_file):
    return struct(
        el_cl_genesis_data=el_cl_genesis_data,
        jwt_file=jwt_file,
    )