Fleets on DigitalOcean

Streamtime creates and runs a Kubernetes cluster (DigitalOcean Kubernetes) in your DigitalOcean account, then installs what the fleet’s workloads need. You register the account once, then create fleets on it.

Key Features

Managed Kubernetes: Streamtime creates, monitors and deletes the cluster for you.
Your network or a new one: use an existing VPC, or let Streamtime create one.
Optional high-availability control plane for production fleets.
Tenancy: Shared, Isolated or Dedicated. See Tenancy.
Sized from your workloads: node sizes and counts follow the capacity you choose for each workload.
Monitoring: metrics, logs and alerts for the fleet and everything on it. Logs are kept in a Spaces bucket.

Prerequisites

A DigitalOcean personal access token with no expiration and write access to Kubernetes, VPCs and Spaces. Streamtime and the Agent on the fleet both use it, so a token that expires stops the fleet from being managed.
A Base Domain for the fleet’s endpoints, added under Settings.

Fleet settings

Placement Configuration

Account: the cloud account you registered.
Region: nyc3, sfo3, ams3, sgp1, fra1 or syd1 (the regions with both Kubernetes and Spaces).

Advanced Configuration

VPC ID: an existing VPC, or leave it empty to create a new one.
VPC CIDR: the address range of a new VPC (10.0.0.0/16, 10.3.0.0/16, the default, or 172.16.0.0/16).
High-Availability Control Plane: off by default.
Public Cluster: whether the Kubernetes API is reachable from the internet (on by default).
Logs Storage Bucket Name: optional; Streamtime generates a unique name when it’s empty.
Spaces Region: optional; set it only when the cluster’s region has no Spaces endpoint.

Before you start

A Base Domain for the fleet’s endpoints, added under Settings.
For the API: an access token in STREAMTIME_TOKEN and your organization identifier in place of <org>. Replace api.example.com with your Streamtime API address.
Only administrators and infrastructure administrators can manage fleets and cloud accounts.

The API examples below share this setup:

HTTPPythonTypeScriptGo

export STREAMTIME_TOKEN=<your-access-token>
export API=https://api.example.com/organizations/<org>

import os
from streamtime_sdk.client import AuthenticatedClient

client = AuthenticatedClient(
    base_url="https://api.example.com",
    token=os.environ["STREAMTIME_TOKEN"],
)
ORG = "<org>"

Uses the generated client in streamtime-sdk/typescript/client.

import { client } from "./client/client.gen";

client.setConfig({
  baseUrl: "https://api.example.com",
  headers: { Authorization: `Bearer ${process.env.STREAMTIME_TOKEN}` },
});
const org = "<org>";

package main

import (
    "bytes"
    "crypto/rand"
    "encoding/hex"
    "encoding/json"
    "net/http"
    "os"
)

const api = "https://api.example.com/organizations/<org>"

// newKey returns a random idempotency key.
func newKey() string {
    b := make([]byte, 16)
    rand.Read(b)
    return hex.EncodeToString(b)
}

// call sends one request with the access token and a JSON body (or none).
// An optional idempotency key is sent as the Idempotency-Key header.
func call(method, path string, body any, idempotencyKey ...string) (*http.Response, error) {
    var buf bytes.Buffer
    if body != nil {
        if err := json.NewEncoder(&buf).Encode(body); err != nil {
            return nil, err
        }
    }
    req, err := http.NewRequest(method, api+path, &buf)
    if err != nil {
        return nil, err
    }
    req.Header.Set("Authorization", "Bearer "+os.Getenv("STREAMTIME_TOKEN"))
    req.Header.Set("Content-Type", "application/json")
    if len(idempotencyKey) > 0 {
        req.Header.Set("Idempotency-Key", idempotencyKey[0])
    }
    return http.DefaultClient.Do(req)
}

Register the cloud account

A cloud account holds the credentials Streamtime uses to create resources in your DigitalOcean account. Register it once; every fleet on that account uses it.

UIAPICLIMCP

Go to Settings, then Bootstrap Providers, and select + Add Bootstrap Provider.
Basic Information: enter an Identifier for the account (lowercase letters, digits and dashes, up to 24 characters).
Bootstrap Provider Details: choose DigitalOcean and fill in its credentials (see Prerequisites), then save.

The account appears in the list. Streamtime checks the credentials when you create a fleet on it.

HTTPPythonTypeScriptGo

curl -X POST "$API/bootstrap-providers/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "<cloud-account>",
    "provider": "digitalocean",
    "configuration": {"digitalocean_token": "<api-token>"}
  }'

Returns 201 with the account. The secret values aren’t returned.

Not available in the Python client yet. Register the account with the HTTP or Go example.

Not available in the TypeScript client yet. Register the account with the HTTP or Go example.

resp, err := call("POST", "/bootstrap-providers/", json.RawMessage(`{
    "identifier": "<cloud-account>",
    "provider": "digitalocean",
    "configuration": {"digitalocean_token": "<api-token>"}
}`))
// resp.StatusCode is 201 on success.

streamtime provider create \
  --identifier <cloud-account> \
  --type digitalocean \
  --config '{"digitalocean_token": "<api-token>"}'

List your accounts with streamtime provider list.

Not available in MCP yet. Register the account from the UI, API or CLI.

To change the credentials later, select Edit Credentials on the account in the UI, or send PATCH $API/bootstrap-providers/<cloud-account>/ with a new configuration. Changing them from the CLI isn’t available yet.

Create a fleet

UIAPICLIMCP

In the sidebar, select Fleet, then + Add Fleet.
Bootstrap Provider: choose DigitalOcean, then select Next.
Basic Configuration: enter an Identifier, choose the Tenancy (Shared, Isolated or Dedicated), the Base Domain and Max Tenants (how many clusters the fleet can host). Under Workloads, choose what the fleet runs (Streaming, Stream Processing, Analytics) and how much capacity each gets; only the services those workloads need are installed. Optionally turn on Enable autoscale for workload node pool and Enable AI, and pick accounts under How should this fleet be monitored?
Placement Configuration: choose the cloud account and where the fleet runs (see Fleet settings).
Advanced Configuration: review the network and node settings.
Review the cost estimate on the right and select Deploy.

The fleet appears in the list as Pending, then Provisioning, then Healthy. Creating the Kubernetes cluster can take 20 minutes or more.

If you leave the wizard early, the fleet is kept as a draft (Incomplete). Open it from the list to pick up where you left off.

HTTPPythonTypeScriptGo

Creating a fleet takes two requests. The first saves the fleet; the second starts provisioning it.

# 1. Save the fleet. The response includes "generation" and "configuration_state".
curl -X POST "$API/fleet/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "events-nyc",
    "provider": "digitalocean",
    "domain": "<base-domain>",
    "tenancy_mode": "Shared",
    "placement_config": {"account": "<cloud-account>", "region": "nyc3"},
    "advanced_config": {"do_vpc_cidr": "10.3.0.0/16"}
  }'

# 2. Start provisioning that generation.
curl -X POST "$API/fleet/events-nyc/reconcile/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"generation": 1}'

identifier, provider and domain are required. Provisioning can start only once configuration_state is valid: the placement and advanced settings are complete. List the settings and their allowed values with GET $API/fleet/configuration-profiles/digitalocean/placement-configuration/ and POST $API/fleet/configuration-profiles/digitalocean/advanced-configuration/.

Add workloads to choose what the fleet runs, for example {"streaming": {"enabled": true, "providers": ["strimzi"], "units": 6}}. Without it, every workload is enabled.

The second request returns 202 with the operation it started. Retrying with the same Idempotency-Key returns that operation again. A 409 means the generation is out of date (send the generation from GET $API/fleet/events-nyc/) or the configuration is incomplete.

Poll until status is Healthy (or Error):

curl -s "$API/fleet/events-nyc/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

The typed FleetCreate model can’t be used for this request yet, so save the fleet with the SDK’s HTTP client, then start provisioning:

import uuid

from streamtime_sdk.api.fleet import (
    organizations_fleet_reconcile_create,
    organizations_fleet_retrieve,
)
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().post(
    f"/organizations/{ORG}/fleet/",
    json={
        "identifier": "events-nyc",
        "provider": "digitalocean",
        "domain": "<base-domain>",
        "tenancy_mode": "Shared",
        "placement_config": {"account": "<cloud-account>", "region": "nyc3"},
        "advanced_config": {"do_vpc_cidr": "10.3.0.0/16"},
    },
)
resp.raise_for_status()
saved = resp.json()

if saved["configuration_state"] == "valid":
    organizations_fleet_reconcile_create.sync(
        ORG,
        "events-nyc",
        client=client,
        body=ReconcileRequest(generation=saved["generation"]),
        idempotency_key=str(uuid.uuid4()),
    )

fleet = organizations_fleet_retrieve.sync(ORG, "events-nyc", client=client)
print(fleet.status)  # Pending → Provisioning → Healthy

The generated FleetCreate type marks server-set fields as required, so send the body through the client directly, then start provisioning:

import {
  organizationsFleetReconcileCreate,
  organizationsFleetRetrieve,
} from "./client";

const { data: saved } = await client.post({
  url: "/organizations/{org_identifier}/fleet/",
  path: { org_identifier: org },
  body: {
    identifier: "events-nyc",
    provider: "digitalocean",
    domain: "<base-domain>",
    tenancy_mode: "Shared",
    placement_config: {"account": "<cloud-account>", "region": "nyc3"},
    advanced_config: {"do_vpc_cidr": "10.3.0.0/16"},
  },
});

await organizationsFleetReconcileCreate({
  path: { org_identifier: org, identifier: "events-nyc" },
  headers: { "Idempotency-Key": crypto.randomUUID() },
  body: { generation: (saved as { generation: number }).generation },
});

const { data } = await organizationsFleetRetrieve({
  path: { org_identifier: org, identifier: "events-nyc" },
});
console.log(data?.status);

resp, err := call("POST", "/fleet/", json.RawMessage(`{
    "identifier": "events-nyc",
    "provider": "digitalocean",
    "domain": "<base-domain>",
    "tenancy_mode": "Shared",
    "placement_config": {"account": "<cloud-account>", "region": "nyc3"},
    "advanced_config": {"do_vpc_cidr": "10.3.0.0/16"}
}`))
if err != nil {
    return err
}
defer resp.Body.Close()

var saved struct {
    Generation         int    `json:"generation"`
    ConfigurationState string `json:"configuration_state"`
}
if err := json.NewDecoder(resp.Body).Decode(&saved); err != nil {
    return err
}

// Start provisioning that generation.
resp, err = call("POST", "/fleet/events-nyc/reconcile/",
    map[string]any{"generation": saved.Generation}, newKey())
// Then poll: call("GET", "/fleet/events-nyc/", nil) until "status" is "Healthy".

streamtime fleet create \
  --identifier events-nyc \
  --provider digitalocean \
  --domain <base-domain> \
  --tenancy-mode Shared \
  --streaming strimzi --streaming-units 6 \
  --placement-config '{"account": "<cloud-account>", "region": "nyc3"}' \
  --advanced-config '{"do_vpc_cidr": "10.3.0.0/16"}'

When the settings are complete, the command also starts provisioning. If they aren’t, it saves the fleet and says what’s missing; finish the settings, then run streamtime fleet reconcile events-nyc.

Choose workloads with --streaming, --stream-processing and --analytics (each takes a comma-separated list of providers) and their --…-units options. Without them, every workload is enabled.

Not available in MCP yet. Create the fleet from the UI, API or CLI.

Check fleet status

A fleet’s Status is one of Incomplete, Pending, Provisioning, Updating, Restarting, Deleting, Healthy or Error. The Agent’s connection and the installed components show whether Streamtime can reach the fleet and what it runs.

UIAPICLIMCP

In the sidebar, select Fleet. Each fleet’s status shows in the list.
Open the fleet. Overview shows its status, tenancy, Kafka Units and capacity. Endpoints lists its addresses, and Monitoring its metrics.
Agent Management shows whether the Agent is installed and connected, with the history of its operations.

HTTPPythonTypeScriptGo

# Status and provisioning state
curl -s "$API/fleet/events-nyc/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Collect a fresh health report from the fleet (waits for the answer)
curl -s -X POST "$API/fleet/events-nyc/status/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# The Agent's connection
curl -s "$API/fleet/events-nyc/agent/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Installed components
curl -s "$API/fleet/events-nyc/installed-components/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

In the fleet, status is the fleet’s state and reconciliation_state the state of its last operation (queued, running, succeeded or failed). GET $API/fleet/events-nyc/status/ returns the last health report without collecting a new one. The Agent’s status reads bootstrap_complete once it is installed and the fleet is set up.

from streamtime_sdk.api.agent import get_agent_connection
from streamtime_sdk.api.fleet import organizations_fleet_retrieve
from streamtime_sdk.api.fleet_metrics import organizations_fleet_status_create

fleet = organizations_fleet_retrieve.sync(ORG, "events-nyc", client=client)
print(fleet.status, fleet.reconciliation_state)

report = organizations_fleet_status_create.sync_detailed(ORG, "events-nyc", client=client)
print(report.content.decode())

agent = get_agent_connection.sync_detailed(ORG, "events-nyc", client=client)
print(agent.content.decode())

import {
  getAgentConnection,
  organizationsFleetRetrieve,
  organizationsFleetStatusCreate,
} from "./client";

const { data: fleet } = await organizationsFleetRetrieve({
  path: { org_identifier: org, identifier: "events-nyc" },
});
console.log(fleet?.status, fleet?.reconciliation_state);

const report = await organizationsFleetStatusCreate({
  path: { org_identifier: org, fleet_id: "events-nyc" },
});
const agent = await getAgentConnection({
  path: { org_identifier: org, fleet_id: "events-nyc" },
});
console.log(report.data, agent.data);

resp, err := call("GET", "/fleet/events-nyc/", nil)
// Decode "status" and "reconciliation_state" from resp.Body.

resp, err = call("POST", "/fleet/events-nyc/status/", nil)  // fresh health report
resp, err = call("GET", "/fleet/events-nyc/agent/", nil)    // Agent connection

streamtime fleet list
streamtime fleet get events-nyc
streamtime fleet status events-nyc
streamtime fleet kubeconfig events-nyc --save ./events-nyc.kubeconfig

fleet status shows the last health report. Showing the Agent’s connection or the installed components isn’t available in the CLI yet.

Ask your assistant, for example: “Is the events-nyc fleet healthy?” or “What alerts fired on events-nyc in the last hour?”

It uses get_fleet_status (pass refresh to collect a fresh report), list_fleet_resources, get_fleet_alerts, get_fleet_logs and get_fleet_metric.

Update the fleet or the Agent

You can turn the workload node pool’s autoscaling on or off, and change the accounts that receive the fleet’s monitoring data. Streamtime can’t apply changes to a fleet’s placement or advanced settings yet: they are saved, but the update operation fails. To change them, create a new fleet.

The Agent is the Streamtime component that runs inside the fleet. Update it to a new version from Agent Management.

UIAPICLIMCP

Open the fleet and select Agent Management.
Open the actions menu (⋮) and select Update Agent.
In Update Streamtime Agent, choose the Target (Worker or Operator) and the Registry, then set the Image Tag (leave it empty for the latest), or the Chart Version or Helm Values you want to change. Start the update.

The operation shows under Command History until it finishes.

HTTPPythonTypeScriptGo

# Turn autoscaling of the workload node pool off
curl -X PATCH "$API/fleet/events-nyc/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"autoscale": false}'

# Update the Agent to a new version
curl -X POST "$API/fleet/events-nyc/agent/UpdateStreamtimeAgentWorkflow/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "update_types": ["agent"],
    "registry_url": "<registry>",
    "repository": "<repository>",
    "agent": {"image_tag": "<version>"}
  }'

registry_url and repository are where your Streamtime installation publishes the Agent image; ask your Streamtime administrator. The update returns 201 with the operation it started. Follow it with GET $API/fleet/events-nyc/agent/.

from streamtime_sdk.api.fleet import organizations_fleet_partial_update
from streamtime_sdk.api.fleet_tasks import dispatch_agent_task
from streamtime_sdk.models import DispatchAgentTaskWorkflowType, PatchedFleet

organizations_fleet_partial_update.sync(
    ORG, "events-nyc", client=client, body=PatchedFleet(autoscale=False)
)

resp = dispatch_agent_task.sync_detailed(
    ORG,
    "events-nyc",
    DispatchAgentTaskWorkflowType.UPDATESTREAMTIMEAGENTWORKFLOW,
    client=client,
    body={
        "update_types": ["agent"],
        "registry_url": "<registry>",
        "repository": "<repository>",
        "agent": {"image_tag": "<version>"},
    },
)
print(resp.status_code)  # 201

import { dispatchAgentTask, organizationsFleetPartialUpdate } from "./client";

await organizationsFleetPartialUpdate({
  path: { org_identifier: org, identifier: "events-nyc" },
  body: { autoscale: false },
});

await dispatchAgentTask({
  path: {
    org_identifier: org,
    fleet_id: "events-nyc",
    workflow_type: "UpdateStreamtimeAgentWorkflow",
  },
  body: {
    update_types: ["agent"],
    registry_url: "<registry>",
    repository: "<repository>",
    agent: { image_tag: "<version>" },
  },
});

resp, err := call("PATCH", "/fleet/events-nyc/", map[string]any{"autoscale": false})

resp, err = call("POST", "/fleet/events-nyc/agent/UpdateStreamtimeAgentWorkflow/", map[string]any{
    "update_types": []string{"agent"},
    "registry_url": "<registry>",
    "repository":   "<repository>",
    "agent":        map[string]any{"image_tag": "<version>"},
})
// resp.StatusCode is 201 once the update has started.

Not available in the CLI yet. Update the fleet or the Agent from the UI or the API.

Not available in MCP yet. Update the fleet or the Agent from the UI or the API.

Delete the fleet

Delete every cluster and workload on the fleet first; Streamtime refuses to delete a fleet that still hosts one. Deleting the fleet removes the resources Streamtime created for it.

UIAPICLIMCP

In the sidebar, select Fleet.
Open the fleet’s menu in the list, select Delete, and confirm.

The fleet shows Deleting, then disappears from the list.

HTTPPythonTypeScriptGo

curl -X DELETE "$API/fleet/events-nyc/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

Returns 202 with the delete operation it started. 403 means the fleet still hosts Kafka clusters; 409 means it still has Kafka Connect clusters, replicators or virtual Kafka clusters. Delete those first. Poll GET $API/fleet/events-nyc/: it shows Deleting, then returns 404 once the fleet is gone.

from streamtime_sdk.api.fleet import organizations_fleet_destroy

resp = organizations_fleet_destroy.sync_detailed(ORG, "events-nyc", client=client)
print(resp.status_code)  # 202; 403 or 409 while workloads remain

import { organizationsFleetDestroy } from "./client";

await organizationsFleetDestroy({
  path: { org_identifier: org, identifier: "events-nyc" },
});

resp, err := call("DELETE", "/fleet/events-nyc/", nil)
// resp.StatusCode is 202 (403 or 409 while workloads remain);
// poll GET /fleet/events-nyc/ until it returns 404.

streamtime fleet delete events-nyc --yes

Not available in MCP yet. Delete the fleet from the UI, API or CLI.

Remove the cloud account

Delete every fleet on the account first, and wait until they are gone. Streamtime doesn’t stop you from removing an account that fleets still use, and without it those fleets can’t be torn down.

UIAPICLIMCP

Go to Settings, then Bootstrap Providers.
Select Delete on the account and confirm.

HTTPPythonTypeScriptGo

curl -X DELETE "$API/bootstrap-providers/<cloud-account>/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN"

Returns 204.

from streamtime_sdk.api.bootstrap_providers import organizations_bootstrap_providers_destroy

resp = organizations_bootstrap_providers_destroy.sync_detailed(
    ORG, "<cloud-account>", client=client
)
print(resp.status_code)  # 204

import { organizationsBootstrapProvidersDestroy } from "./client";

await organizationsBootstrapProvidersDestroy({
  path: { org_identifier: org, identifier: "<cloud-account>" },
});

resp, err := call("DELETE", "/bootstrap-providers/<cloud-account>/", nil)
// resp.StatusCode is 204.

streamtime provider delete <cloud-account> --yes

Not available in MCP yet. Remove the account from the UI, API or CLI.