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.

from typing import NamedTuple

from prodvana.client import Client, make_channel
from prodvana.proto.prodvana.desired_state.manager_pb2 import (
    GetServiceDesiredStateConvergenceSummaryReq,
)
from prodvana.proto.prodvana.desired_state.model.desired_state_pb2 import Status, Type


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


with make_channel() as channel:
    client = Client(channel=channel)
    resp = client.desired_state_manager.GetServiceDesiredStateConvergenceSummary(
        GetServiceDesiredStateConvergenceSummaryReq(
            application="my-app", service="my-service"
        )
    )
    graph = {
        HashableIdentifier(type=entity.id.type, name=entity.id.name): entity
        for entity in resp.summary.entity_graph.entities
    }
    svc_entity = graph[
        HashableIdentifier(
            type=resp.summary.entity_graph.root.type,
            name=resp.summary.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_entity = graph[HashableIdentifier(type=child.type, name=child.name)]
        print(
            f"release channel: {child_entity.desired_state.service_instance.release_channel}, status: {Status.Name(child_entity.status)}"
        )

Start a Deployment

Given an app and a service, deploy the service 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)

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"
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,
            # uncomment to reject instead of approve
            # reject=True,
        )
    )

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 "deploy" 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 (
    GetDesiredStateConvergenceReq,
)
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() as channel:
    client = Client(channel=channel)

    resp = client.desired_state_manager.GetDesiredStateConvergenceSummary(
        GetDesiredStateConvergenceReq(
            desired_state_id=ds_id,
        )
    )

    svc_entity = [
        entity
        for entity in resp.summary.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)