All examples here are written using our Python library, but the same API can be used in any language supported by gRPC or that can make OpenAPI HTTP calls.

Listing All Applications

from prodvana.client import Client, make_client
from prodvana.proto.prodvana.application.application_manager_pb2 import (
    ListApplicationsReq,
)

with make_channel() as channel:
    client = Client(channel=channel)
    resp = client.application_manager.ListApplications(ListApplicationsReq())
    for app in resp.applications:
        print(app.meta.name)

Listing All Services in an Application

from prodvana.client import Client, make_channel
from prodvana.proto.prodvana.service.service_manager_pb2 import ListServicesReq


with make_channel() as channel:
    client = Client(channel=channel)
    resp = client.service_manager.ListServices(
        ListServicesReq(application="my-app")
    )
    for svc in resp.services:
        print(svc.meta.name)

Get Service Convergence Status

Given an application and a service, print the current convergence status as well as the status of the individual release channels. Handle any pending desired states correctly.

from typing import NamedTuple

from prodvana.client import Client, make_channel
from prodvana.proto.prodvana.desired_state.manager_pb2 import GetDesiredStateGraphReq
from prodvana.proto.prodvana.desired_state.model.desired_state_pb2 import Status, Type


class HashableIdentifier(NamedTuple):
    type: "Type.V"
    name: str


with make_channel(org=args.org, api_token=args.api_token) as channel:
    client = Client(channel=channel)
    resp = client.desired_state_manager.GetDesiredStateGraph(
        GetDesiredStateGraphReq(
            query_by_service=GetDesiredStateGraphReq.QueryByService(
                application=app,
                service=service,
            ),
            types=[Type.SERVICE_INSTANCE],
        )
    )
    if resp.HasField("pending_set_desired_state"):
        # there is a desired state that is pending to be applied, which means that the rest of the entity graph will soon be replaced.
        if resp.pending_set_desired_state.task_status == TaskStatus.FAILED:
            print(
                f"Latest desired state failed to apply.\n{resp.pending_set_desired_state.task_result.log.decode('utf-8')}"
            )
            # continue to print the rest of the graph as the pending desired state has failed
        else:  # running
            print("Pending new desired state:")
            for (
                rc_state
            ) in (
                resp.pending_set_desired_state.compiled_desired_state.service.release_channels
            ):
                print(
                    f"release channel: {rc_state.release_channel}, pending version: {rc_state.versions[0].version}"
                )
            return  # the rest of the graph is not really relevant as it is about to be taken over by new desired state.
    graph = {
        HashableIdentifier(type=entity.id.type, name=entity.id.name): entity
        for entity in resp.entity_graph.entities
    }
    svc_entity = graph[
        HashableIdentifier(
            type=resp.entity_graph.root.type,
            name=resp.entity_graph.root.name,
        )
    ]
    print(f"status: {Status.Name(svc_entity.status)}")
    for child in svc_entity.dependencies:
        if child.type != Type.SERVICE_INSTANCE:
            continue

        child_id = HashableIdentifier(type=child.type, name=child.name)
        child_entity = graph[child_id]
        print(
            f"release channel: {child_entity.desired_state.service_instance.release_channel}, desired version: {child_entity.desired_state.service_instance.versions[0].version}, status: {Status.Name(child_entity.status)}"
        )

Create a Release

Given an Application and a Service, create a new release with an updated Docker tag.

from prodvana.client import Client, make_channel
from prodvana.proto.prodvana.application.application_manager_pb2 import (
    GetApplicationReq,
)
from prodvana.proto.prodvana.common_config.parameters_pb2 import ParameterValue
from prodvana.proto.prodvana.desired_state.manager_pb2 import SetDesiredStateReq
from prodvana.proto.prodvana.desired_state.model.desired_state_pb2 import (
    ServiceInstanceState,
    ServiceState,
    State,
    Version,
)
from prodvana.proto.prodvana.service.service_manager_pb2 import (
    ApplyParametersReq,
    GetServiceConfigReq,
    ServiceConfigVersionReference,
)


app = "my-app"
service = "my-service"
param_name = "my-image-param-name"
param_value = "my-image-tag"
with make_channel() as channel:
    client = Client(channel=channel)

    # take the latest config
    config_resp = client.service_manager.GetServiceConfig(
        GetServiceConfigReq(application=app, service=service)
    )

    # validate that the requested parameter exists and is a docker image parameter (the only one supported by this example)
    param_defs = {param.name: param for param in config_resp.config.parameters}
    assert param_name in param_defs
    assert param_defs[param_name].docker_image

    # create a new service version using the service config and requested parameter
    apply_resp = client.service_manager.ApplyParameters(
        ApplyParametersReq(
            service_config_version=ServiceConfigVersionReference(
                application=app,
                service=service,
                service_config_version=config_resp.config_version,
            ),
            parameters=[
                ParameterValue(name=param_name, docker_image_tag=param_value),
            ],
        )
    )

    # get list of release channels so we can construct desired state
    app_resp = client.application_manager.GetApplication(
        GetApplicationReq(application=app)
    )

    # construct desired state and set it, which causes convergence to begin
    desired_state = State(
        service=ServiceState(
            service=service,
            application=app,
            release_channels=[
                ServiceInstanceState(
                    release_channel=rc.name,
                    versions=[
                        Version(
                            version=apply_resp.version,
                        ),
                    ],
                )
                for rc in app_resp.application.config.release_channels
            ],
        )
    )
    ds_resp = client.desired_state_manager.SetDesiredState(
        SetDesiredStateReq(desired_state=desired_state)
    )

    # desired state id is unique for each call to SetDesiredState and can be used to get convergence status, submit manual approval, etc.
    print(ds_resp.desired_state_id)

Find Pending Manual Approvals

Given an Application and Service, find all Release Channels waiting on manual approvals. Also print out how to approve these Release Channels.

from typing import Iterator, Mapping, NamedTuple, Optional

from prodvana.client import Client, make_channel
from prodvana.proto.prodvana.desired_state.manager_pb2 import GetDesiredStateGraphReq
from prodvana.proto.prodvana.desired_state.model.desired_state_pb2 import (
    SignalType,
    Status,
    Type,
)
from prodvana.proto.prodvana.desired_state.model.entity_pb2 import Entity


class HashableIdentifier(NamedTuple):
    type: "Type.V"
    name: str


class MissingApproval(NamedTuple):
    topic: str
    signal_type: "SignalType.V"
    desired_state_id: str


def find_missing_approval(
    graph: Mapping[HashableIdentifier, Entity], release_channel: HashableIdentifier
) -> Optional[MissingApproval]:
    release_channel_entity = graph[release_channel]
    for dep in release_channel_entity.dependencies:
        if dep.type == Type.MANUAL_APPROVAL:
            manual_approval_entity = graph[
                HashableIdentifier(type=dep.type, name=dep.name)
            ]
            if manual_approval_entity.status == Status.CONVERGING:
                return MissingApproval(
                    topic=manual_approval_entity.desired_state.manual_approval.topic,
                    signal_type=SignalType.SIGNAL_MANUAL_APPROVAL,
                    desired_state_id=release_channel_entity.root_desired_state_id,
                )

    # No manual approval entities.
    # Check if anything in the tree has missing approval.

    def visit(
        graph: Mapping[HashableIdentifier, Entity], node: HashableIdentifier
    ) -> Iterator[HashableIdentifier]:
        for dep in graph[node].dependencies:
            yield HashableIdentifier(type=dep.type, name=dep.name)

    for node in visit(graph, release_channel):
        missing_approval = graph[node].missing_approval
        if missing_approval is not None and len(missing_approval.topic) > 0:
            return MissingApproval(
                topic=missing_approval.topic,
                signal_type=missing_approval.signal_type,
                desired_state_id=release_channel_entity.root_desired_state_id,
            )

    return None
  

with make_channel(org=args.org, api_token=args.api_token) as channel:
    client = Client(channel=channel)
    resp = client.desired_state_manager.GetDesiredStateGraph(
        GetDesiredStateGraphReq(
            query_by_service=GetDesiredStateGraphReq.QueryByService(
                application=app,
                service=service,
            ),
            types=[Type.SERVICE_INSTANCE, Type.MANUAL_APPROVAL],
        )
    )
    if resp.HasField("pending_set_desired_state"):
        # there is a desired state that is pending to be applied, which means that the rest of the entity graph will soon be replaced.
        if resp.pending_set_desired_state.task_status == TaskStatus.FAILED:
            print(
                f"Latest desired state failed to apply.\n{resp.pending_set_desired_state.task_result.log.decode('utf-8')}"
            )
            # continue to print the rest of the graph as the pending desired state has failed
        else:  # running
            print("Pending new desired state:")
            for (
                rc_state
            ) in (
                resp.pending_set_desired_state.compiled_desired_state.service.release_channels
            ):
                print(
                    f"release channel: {rc_state.release_channel}, pending version: {rc_state.versions[0].version}"
                )
            return  # the rest of the graph is not really relevant as it is about to be taken over by new desired state.

    graph = {
        HashableIdentifier(type=entity.id.type, name=entity.id.name): entity
        for entity in resp.entity_graph.entities
    }
    svc_entity = graph[
        HashableIdentifier(
            type=resp.entity_graph.root.type,
            name=resp.entity_graph.root.name,
        )
    ]
    print(f"status: {Status.Name(svc_entity.status)}")
    for child in svc_entity.dependencies:
        if child.type != Type.SERVICE_INSTANCE:
            continue

        child_id = HashableIdentifier(type=child.type, name=child.name)
        child_entity = graph[child_id]
        if child_entity.status == Status.WAITING_MANUAL_APPROVAL:
            print(
                f"Missing approval for {child_entity.desired_state.service_instance.release_channel}:"
            )
            missing_approval = find_missing_approval(graph, child_id)
            if missing_approval:
                print(
                    f"\tpvnctl services approve --app {app} {service} {missing_approval.topic} {missing_approval.desired_state_id} --signal-type {SignalType.Name(missing_approval.signal_type)}"
                )

Submit a Manual Approval

Submitting a manual approval requires a desired state ID and the Release Channel name to approve for.

from prodvana.client import Client, make_channel
from prodvana.proto.prodvana.desired_state.manager_pb2 import SetManualApprovalReq


ds_id = "des-..."
release_channel = "production"
# Set this based on missingApproval value or leave blank if no missingApproval field.
signal_type = "SIGNAL_MANUAL_APPROVAL"
with make_channel(org=args.org, api_token=args.api_token) as channel:
    client = Client(channel=channel)

    client.desired_state_manager.SetManualApproval(
        SetManualApprovalReq(
            desired_state_id=ds_id,
            topic=release_channel,
            signal_type=signal_type,
            # uncomment to reject instead of approve
            # reject=True,
            # NOTE: reject is only supported for SIGNAL_MANUAL_APPROVAL
        )
    )

Inspecting What Changed in a Convergence

Each desired state keeps track of the starting state when the desired state was set (e.g. when a human operator clicks "Create a New Release" on the web). This data can be used to look at what's changing in a convergence. The example below shows how to take a desired state ID, extracts out service configurations for the starting and desired states, and print out the parameters.

from prodvana.client import Client, make_channel
from prodvana.proto.prodvana.common_config.parameters_pb2 import ParameterValue
from prodvana.proto.prodvana.desired_state.manager_pb2 import GetDesiredStateGraphReq
from prodvana.proto.prodvana.service.service_manager_pb2 import GetMaterializedConfigReq


def print_param_value(param: ParameterValue) -> None:
    print(f"parameter {param.name}: ", end="")
    if param.string:
        print(param.string)
    elif param.int:
        print(f"{param.int}")
    elif param.docker_image_tag:
        print(param.docker_image_tag)
    else:
        raise Exception(f"unrecognized parameter: {param}")


ds_id = "des-..."
with make_channel(org=args.org, api_token=args.api_token) as channel:
    client = Client(channel=channel)

    resp = client.desired_state_manager.GetDesiredStateGraph(
        GetDesiredStateGraphReq(
            desired_state_id=ds_id,
        )
    )

    svc_entity = [
        entity
        for entity in resp.entity_graph.entities
        if entity.desired_state_id == ds_id
    ][0]
    starting_state = svc_entity.starting_state.service
    starting_state_release_channels = {
        rc.release_channel: rc for rc in starting_state.release_channels
    }
    desired_state = svc_entity.desired_state.service
    for desired_release_channel_state in desired_state.release_channels:
        release_channel = desired_release_channel_state.release_channel
        starting_release_channel_state = starting_state_release_channels.get(
            release_channel
        )
        print(f"release channel: {release_channel}")
        if starting_release_channel_state:  # can be None on first deployment
            print("starting state:")
            for version in starting_release_channel_state.versions:
                config_resp = client.service_manager.GetMaterializedConfig(
                    GetMaterializedConfigReq(
                        application=desired_state.application,
                        service=desired_state.service,
                        version=version.version,
                    )
                )
                svc_instance_config = [
                    cfg
                    for cfg in config_resp.compiled_service_instance_configs
                    if cfg.release_channel == release_channel
                ][0]
                for param in svc_instance_config.parameter_values:
                    print_param_value(param)
        print("desired state:")
        assert (
            len(desired_release_channel_state.versions) == 1
        ), "can only have one desired version"
        desired_version = desired_release_channel_state.versions[0]
        config_resp = client.service_manager.GetMaterializedConfig(
            GetMaterializedConfigReq(
                application=desired_state.application,
                service=desired_state.service,
                version=desired_version.version,
            )
        )
        svc_instance_config = [
            cfg
            for cfg in config_resp.compiled_service_instance_configs
            if cfg.release_channel == release_channel
        ][0]
        for param in svc_instance_config.parameter_values:
            print_param_value(param)