Skip to main content

IT operations — service observability

The problem. A platform team runs dozens of services across a fleet of hosts. They need every service's golden signals — request latency, error rate, CPU, memory — in one place, a live alerting worker that reacts the instant an SLO is breached, and an event trail for post-incident review.

This scenario centres on live consumption: a long-running worker that tails the metrics stream and reacts in real time, rather than polling on a timer.

Set up demo data​

New workspace? Run this once (Python) to create the checkout latency series, create the subscription before we listen (with a breach), and build the dependency graph where two services share a database — so step 4's correlation finds the shared cause. Safe to re-run.

import intellistream_datahub_sdk, numpy as np, pandas as pd

client = intellistream_datahub_sdk.DataHubClient.from_env()

client.timeseries.create([intellistream_datahub_sdk.TimeSeries(external_id="service_checkout_request_latency_ms", name="Checkout p99 latency", unit="ms", value_type="float")])
client.subscriptions.create([intellistream_datahub_sdk.Subscription(
    external_id="checkout_slo", name="Checkout SLO", timeseries=["service_checkout_request_latency_ms"])])
idx = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=60, freq="1min")
lat = np.full(60, 180.0); lat[-10:] = 360.0          # p99 breaches the 300 ms SLO
client.timeseries.insert_from_lists(timestamps=idx, values=lat, ts="service_checkout_request_latency_ms")

client.resources.create(
    [intellistream_datahub_sdk.Resource(external_id=x, name=x, labels=[lbl]) for x, lbl in
     [("service_checkout", "Service"), ("service_payments", "Service"),
      ("order_db", "Database"), ("host_web_01", "Host")]],
    [intellistream_datahub_sdk.RelForm.by_external_ids("service_checkout", "order_db", "depends_on"),
     intellistream_datahub_sdk.RelForm.by_external_ids("service_payments", "order_db", "depends_on"),
     intellistream_datahub_sdk.RelForm.by_external_ids("service_checkout", "host_web_01", "runs_on")])

1. Model services and hosts​

A service runs on hosts; both are resources. Modeling the topology lets an alert on a host fan out to the services it affects.

ResourceForm service = new ResourceForm();
service.setExternalId("service_checkout");
service.setName("Checkout service");
service.setLabels(List.of("Service"));

ResourceForm host = new ResourceForm();
host.setExternalId("host_web_01");
host.setName("web-01");
host.setLabels(List.of("Host"));

RelForm runsOn = new RelForm();
runsOn.setName("runs_on");
runsOn.setFromExternalId("service_checkout");
runsOn.setToExternalId("host_web_01");

client.resources().create(List.of(service, host), List.of(runsOn));

Metrics become series: service_checkout_request_latency_ms, service_checkout_error_rate, host_web_01_cpu_utilization.

2. Alert in real time on an SLO breach​

Subscribe to the service's golden-signal series and run the SLO check inside the delivery loop — the worker reacts the moment a bad datapoint lands, and records an event the on-call dashboard is watching. See Consume live data for the listener lifecycle.

import ai.intellistream.datahub.sdk.subscriptions.SubscriptionMessage;

try (var stream = client.subscriptions().listen(List.of("checkout_slo"))
        .stream((SubscriptionMessage msg) -> {   // auto-acks after each message
            if (breachesSlo(msg.payload())) {       // e.g. p99 latency > 300 ms
                EventModel event = new EventModel();
                event.setExternalId("slo_breach_checkout_" + System.currentTimeMillis());
                event.setType("slo_breach");
                event.setStatus("firing");
                event.setMetadata(Map.of("service", "service_checkout", "slo", "latency_p99"));
                event.setEventTime(ZonedDateTime.now());
                client.events().create(List.of(event));
            }
        })) {
    awaitShutdown();   // your app lifecycle; closing the stream ends delivery
}

3. Review the incident afterward​

Every breach is a durable event. Filter by type and time to reconstruct an incident timeline, or by the service metadata to see one service's history.

EventRetreiver retriever = new EventRetreiver();
retriever.getFilter().setType(List.of("slo_breach"));
retriever.setLimit(200);
DataWrapper<EventModel> breaches = client.events().filter(retriever);

4. Which services share a failing dependency?​

When service_checkout and service_payments breach their SLOs at the same moment, the on-call question is whether they're independent incidents or both downstream of one failing component. Model dependencies as edges — a service depends_on the datastores and caches it calls — and the graph answers it: walk out from each breaching service and intersect to find the component they have in common.

ResourceNetwork checkout = client.resources().fetchRelated("service_checkout", 4);
ResourceNetwork payments = client.resources().fetchRelated("service_payments", 4);

Set<String> checkoutDeps = checkout.nodes().stream()
    .map(Resource::getExternalId).collect(Collectors.toSet());
Set<String> shared = payments.nodes().stream()
    .map(Resource::getExternalId).filter(checkoutDeps::contains).collect(Collectors.toSet());

// shared contains "order_db" → one failing dependency, not two outages

The reverse is just as useful for change management: fetchRelated from a component you're about to restart returns its blast radius — every service that depends on it. See Correlate alarms with the graph for the pattern.

See the result​

The breach trips the loop, and expanding the two affected services finds their common cause:

slo_breach event raised for service_checkout
shared dependency: order_db        ← fix the database, not each service

See also​