Poll topology

A fleet poller for GET /topology built on the stepped as_of_utc pattern: walk the evaluation instant across a time range at a fixed step, one request per step, and each response is the latest-per-entity fleet state as of that instant. Setting the step equal to freshness_seconds gives clean, non-smearing frames.

The same loop serves two jobs:

Replay: step from a start time to an end time to reconstruct history.
Live polling: keep stepping at the wall clock (or just omit as_of_utc) to follow the present.

Each step sleeps one second between requests to stay under the per-tenant budget of 60 requests/minute (rate limits). Set CONSTELLATION_API_BASE and CONSTELLATION_API_TOKEN first (common setup).

TypeScript
Python
Go
cURL

const BASE = process.env.CONSTELLATION_API_BASE ?? 'https://api.constellation.space'
const TOKEN = process.env.CONSTELLATION_API_TOKEN
if (!TOKEN) throw new Error('CONSTELLATION_API_TOKEN is not set')

const STEP_SECONDS = 300
const START = new Date('2026-07-09T12:00:00Z')
const END = new Date('2026-07-09T14:00:00Z')

type Entity = { entity_id: string; observed_at: string; tags?: Record<string, string> }
type Topology = { as_of_utc: string; tenant_key: string; entities: Entity[] }

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms))

async function snapshot(asOf: Date): Promise<Topology> {
  const url = new URL(`${BASE}/topology`)
  url.searchParams.set('measurement', 'satellite')
  url.searchParams.set('freshness_seconds', String(STEP_SECONDS))
  url.searchParams.set('as_of_utc', asOf.toISOString())
  for (;;) {
    const res = await fetch(url, { headers: { 'x-api-key': TOKEN! } })
    if (res.status === 429 || res.status === 503) {
      await sleep(Number(res.headers.get('retry-after') ?? '5') * 1000)
      continue
    }
    if (!res.ok) throw new Error(`topology read failed: ${res.status} ${await res.text()}`)
    return (await res.json()) as Topology
  }
}

for (let t = START.getTime(); t <= END.getTime(); t += STEP_SECONDS * 1000) {
  const topo = await snapshot(new Date(t))
  const nodes = new Set(topo.entities.map((e) => e.tags?.node_id ?? e.entity_id))
  console.log(`${topo.as_of_utc}  ${nodes.size} nodes`)
  await sleep(1000) // pace requests under the tenant rate budget
}

import os
import time
from datetime import datetime, timedelta, timezone

import requests

BASE = os.environ.get("CONSTELLATION_API_BASE", "https://api.constellation.space")
TOKEN = os.environ["CONSTELLATION_API_TOKEN"]

STEP_SECONDS = 300
START = datetime(2026, 7, 9, 12, 0, tzinfo=timezone.utc)
END = datetime(2026, 7, 9, 14, 0, tzinfo=timezone.utc)


def snapshot(as_of):
    params = {
        "measurement": "satellite",
        "freshness_seconds": STEP_SECONDS,
        "as_of_utc": as_of.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    while True:
        resp = requests.get(
            f"{BASE}/topology",
            params=params,
            headers={"x-api-key": TOKEN},
            timeout=30,
        )
        if resp.status_code in (429, 503):
            time.sleep(float(resp.headers.get("Retry-After", "5")))
            continue
        resp.raise_for_status()
        return resp.json()


t = START
while t <= END:
    topo = snapshot(t)
    nodes = {e.get("tags", {}).get("node_id") or e["entity_id"] for e in topo["entities"]}
    print(f"{topo['as_of_utc']}  {len(nodes)} nodes")
    t += timedelta(seconds=STEP_SECONDS)
    time.sleep(1)  # pace requests under the tenant rate budget

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"time"
)

const stepSeconds = 300

type Entity struct {
	EntityID   string            `json:"entity_id"`
	ObservedAt string            `json:"observed_at"`
	Tags       map[string]string `json:"tags"`
}

type Topology struct {
	AsOfUTC   string   `json:"as_of_utc"`
	TenantKey string   `json:"tenant_key"`
	Entities  []Entity `json:"entities"`
}

func snapshot(base, token string, asOf time.Time) (*Topology, error) {
	q := url.Values{}
	q.Set("measurement", "satellite")
	q.Set("freshness_seconds", strconv.Itoa(stepSeconds))
	q.Set("as_of_utc", asOf.UTC().Format(time.RFC3339))
	for {
		req, err := http.NewRequest(http.MethodGet, base+"/topology?"+q.Encode(), nil)
		if err != nil {
			return nil, err
		}
		req.Header.Set("x-api-key", token)
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			return nil, err
		}
		if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == http.StatusServiceUnavailable {
			resp.Body.Close()
			wait, _ := strconv.Atoi(resp.Header.Get("Retry-After"))
			if wait <= 0 {
				wait = 5
			}
			time.Sleep(time.Duration(wait) * time.Second)
			continue
		}
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusOK {
			return nil, fmt.Errorf("topology read failed: %d", resp.StatusCode)
		}
		var topo Topology
		if err := json.NewDecoder(resp.Body).Decode(&topo); err != nil {
			return nil, err
		}
		return &topo, nil
	}
}

func main() {
	base := os.Getenv("CONSTELLATION_API_BASE")
	if base == "" {
		base = "https://api.constellation.space"
	}
	token := os.Getenv("CONSTELLATION_API_TOKEN")
	if token == "" {
		log.Fatal("CONSTELLATION_API_TOKEN is not set")
	}

	start := time.Date(2026, 7, 9, 12, 0, 0, 0, time.UTC)
	end := time.Date(2026, 7, 9, 14, 0, 0, 0, time.UTC)

	for t := start; !t.After(end); t = t.Add(stepSeconds * time.Second) {
		topo, err := snapshot(base, token, t)
		if err != nil {
			log.Fatal(err)
		}
		nodes := map[string]struct{}{}
		for _, e := range topo.Entities {
			id := e.Tags["node_id"]
			if id == "" {
				id = e.EntityID
			}
			nodes[id] = struct{}{}
		}
		fmt.Printf("%s  %d nodes\n", topo.AsOfUTC, len(nodes))
		time.Sleep(time.Second) // pace requests under the tenant rate budget
	}
}

Requires GNU date (gdate from coreutils on macOS) and jq.

#!/usr/bin/env bash
set -euo pipefail

BASE="${CONSTELLATION_API_BASE:-https://api.constellation.space}"
: "${CONSTELLATION_API_TOKEN:?set CONSTELLATION_API_TOKEN}"

START="2026-07-09T12:00:00Z"
END="2026-07-09T14:00:00Z"
STEP_SECONDS=300

epoch=$(date -u -d "$START" +%s)
end_epoch=$(date -u -d "$END" +%s)

while ((epoch <= end_epoch)); do
  as_of=$(date -u -d "@$epoch" +%Y-%m-%dT%H:%M:%SZ)
  response=$(curl -sS -w '\n%{http_code}' \
    -H "x-api-key: $CONSTELLATION_API_TOKEN" \
    "$BASE/topology?measurement=satellite&freshness_seconds=${STEP_SECONDS}&as_of_utc=${as_of}")
  status=$(tail -n1 <<<"$response")
  body=$(sed '$d' <<<"$response")

  if [[ "$status" == "429" || "$status" == "503" ]]; then
    wait=$(jq -r '.retry_after_seconds // 5' <<<"$body")
    sleep "$wait"
    continue                          # retry the same step
  fi
  if [[ "$status" != "200" ]]; then
    echo "topology read failed ($status): $body" >&2
    exit 1
  fi

  count=$(jq '[.entities[] | (.tags.node_id // .entity_id)] | unique | length' <<<"$body")
  echo "$as_of  $count nodes"

  epoch=$((epoch + STEP_SECONDS))
  sleep 1                             # pace requests under the tenant rate budget
done

Expected output:

2026-07-09T12:00:00Z  24 nodes
2026-07-09T12:05:00Z  24 nodes
2026-07-09T12:10:00Z  23 nodes

A node count dropping between frames means an entity went silent for longer than the freshness window; that reads as "stale", not "deleted" (topology semantics).

Topology API reference: parameters, the 50,000-row cap, response schema.
Integration patterns: choosing cadences and freshness windows.
Ingest telemetry: the write side of this loop.

Related​

Related