Observability pipeline with streams

Fabric metrics and events pushed to a monitoring system, with alert rules and facility data.

Scenario#

A network team monitors its own routers in Datadog and has no data from the Fabric side: connection bandwidth, BGP session state on the Cloud Routers, and port errors. In this design Equinix pushes that data through a stream, in place of a job that polls the metrics API.

Architecture#

Assets are attached to one stream per environment. Subscriptions deliver metrics to the monitoring system and alerts to the paging system. IBX SmartView feeds facility data through its own subscription.

Access check#

Streams need an entitlement. /fabric/v4/streams answered 403 for the account used in the SDK's live validation. Check a credential:

Java

import com.eqixiac.equinix.core.exception.EquinixAuthorizationException;

try {
    fabric.streams().list().forEach(s -> System.out.println(s.getUuid() + " " + s.getName()));
} catch (EquinixAuthorizationException e) {
    System.err.println("This credential is not entitled to Fabric streams: " + e.getMessage());
}

Step 1: stream#

Java

import com.eqixiac.equinix.fabric.enums.StreamType;
import com.eqixiac.equinix.fabric.model.Stream;

Stream stream = fabric.streams().define()
    .withType(StreamType.TELEMETRY_STREAM)
    .withName("prod-interconnect")
    .withDescription("All production connections, ports and Cloud Routers")
    .create();

A stream belongs to a project. One stream per environment keeps production alerts separate from alerts of test resources.

Step 2: attach assets#

Java

import com.eqixiac.equinix.fabric.model.CloudRouter;
import com.eqixiac.equinix.fabric.model.Connection;
import com.eqixiac.equinix.fabric.model.Port;
import com.eqixiac.equinix.fabric.model.implementation.filter.FilterPropertyList;
import com.eqixiac.equinix.fabric.model.implementation.filter.FilterType;

FilterPropertyList inProject = new FilterPropertyList(FilterType.AND).equals("/project/projectId", projectId);

for (Connection connection : fabric.connections().search(inProject).loadAll()) {
    fabric.streamAssets().attach(streamUuid, "connections", connection.getUuid(), true);
}
for (CloudRouter router : fabric.cloudRouters().search(inProject).loadAll()) {
    fabric.streamAssets().attach(streamUuid, "routers", router.getUuid(), true);
}
for (Port port : fabric.ports().search(inProject).loadAll()) {
    fabric.streamAssets().attach(streamUuid, "ports", port.getUuid(), true);
}

Each asset is attached individually, so a connection created later is not on the stream until it is attached. Either run this loop on a schedule, or attach each new connection in the code that creates it. A repeated run has to handle an asset that is already attached, by catching EquinixConflictException or by listing the attached assets first with streamAssets().search(...).

Step 3: subscriptions#

Java

import com.eqixiac.equinix.fabric.enums.StreamSubscriptionSinkCredentialType;
import com.eqixiac.equinix.fabric.enums.StreamSubscriptionSinkType;
import com.eqixiac.equinix.fabric.enums.StreamSubscriptionType;
import com.eqixiac.equinix.fabric.model.StreamSubscription;

StreamSubscription datadog = fabric.streamSubscriptions().define(streamUuid)
    .withType(StreamSubscriptionType.STREAM_SUBSCRIPTION)
    .withName("datadog-prod")
    .withEnabled(true)
    .withSinkType(StreamSubscriptionSinkType.DATADOG)
    .withSinkHost("datadoghq.com")
    .withCredentialType(StreamSubscriptionSinkCredentialType.API_KEY)
    .withApiKey(System.getenv("DATADOG_API_KEY"))
    .create();

StreamSubscription pager = fabric.streamSubscriptions().define(streamUuid)
    .withType(StreamSubscriptionType.STREAM_SUBSCRIPTION)
    .withName("pagerduty-prod")
    .withEnabled(true)
    .withSinkType(StreamSubscriptionSinkType.PAGERDUTY)
    .withSinkHost("events.pagerduty.com")
    .withCredentialType(StreamSubscriptionSinkCredentialType.INTEGRATION_KEY)
    .withIntegrationKey(System.getenv("PAGERDUTY_INTEGRATION_KEY"))
    .create();

Each sink type expects a particular combination of host or URI and credential type, defined in the Fabric specification for stream subscriptions. The two blocks show the builder methods, and the fields for each sink type should be checked against that specification.

Equinix stores the credential. Create a key in the monitoring system that can only submit data, and name it after this subscription so that it can be revoked alone.

withMetricSelector(...) and withEventSelector(...) restrict a subscription. The paging subscription should be restricted to alert events.

Step 4: alert rules#

Java

import com.eqixiac.equinix.fabric.enums.StreamAlertRuleType;

fabric.streamAlertRules().define(streamUuid)
    .type(StreamAlertRuleType.METRIC_ALERT)
    .name("connection-utilization-high")
    .description("Receive rate above 80% of a 1 Gbps connection for 15 minutes")
    .enabled(true)
    .metricSelector(Map.of("include", List.of("equinix.fabric.connection.bandwidth_rx.usage")))
    .resourceSelector(Map.of("include", List.of("*/connections/*")))
    .detectionMethod(Map.of(
        "type", "THRESHOLD",
        "operand", "ABOVE",
        "windowSize", "PT15M",
        "warningThreshold", "700000000",
        "criticalThreshold", "800000000"))
    .create();

The selector and detection maps are passed through as written. Their keys follow the alert-rule schemas of the Fabric specification, and the example illustrates that shape and was not tested against the service. A threshold in bits per second is an absolute value, so one rule fits connections of one bandwidth. Create a rule per bandwidth tier, or alert on utilization in the monitoring system, where the provisioned bandwidth can be joined in.

Step 5: verify delivery#

Java

import com.eqixiac.equinix.fabric.model.implementation.filter.CloudEventQuery;

fabric.streamSubscriptions().list(streamUuid)
    .forEach(s -> System.out.println(s.getName() + " " + s.getState()));

fabric.cloudEvents().search(CloudEventQuery.forProject(projectId)
        .after(LocalDateTime.now(ZoneOffset.UTC).minusHours(1)))
    .forEach(e -> System.out.println(e.getTime() + " " + e.getType() + " " + e.getSubject()));

Then confirm in the sink that data points arrive, and trigger a test alert by setting a threshold below the current rate for a few minutes.

Step 6: facility data#

Power and environmental data come from IBX SmartView through its own streaming subscription, see IBX SmartView. Send it to the same monitoring system, tagged by IBX. A power event in DC6 and a port flap in DC6 two minutes later can then be seen together.

Pull for history#

A capacity review needs the history of each connection, which the metrics API returns on request.

Java

import com.eqixiac.equinix.fabric.model.Metric;

LocalDateTime to = LocalDateTime.now(ZoneOffset.UTC);
for (Connection connection : fabric.connections().search(inProject).loadAll()) {
    List<Metric> month = fabric.metrics().forConnection(connection.getUuid(), to.minusDays(30), to);
    System.out.println(connection.getName() + " " + connection.getBandwidth() + " Mbps provisioned, "
        + month.size() + " series returned");
}

Estate analysis performs this read for a whole scope and reduces the metrics of each connection to the median, the 95th percentile and the peak.

Trade-offs#

Choice	Benefit	Cost
Push through a subscription	No poller; no rate-limit exposure; near real time	Entitlement needed; credential held by Equinix; per-asset attachment to maintain
Pull from the metrics API	Works with read access only; any time range the API retains	A job to run; one request per asset per interval
Alert rules on the stream	Alerts without a monitoring system	Absolute thresholds; fewer functions than a monitoring system's own alerting
Alerting in the monitoring system	Joins with your own data; ratios, anomaly detection	Depends on delivery being healthy; monitor the subscription state too