Broker plugins

Broker plugins#

SkyPortal integrates with external alert brokers (e.g. BOOM, Fink, Lasair) through a pluggable provider interface, modeled on the follow-up facility APIs (skyportal/facility_apis/). A broker provider is a registered class; a configured connection to a broker is a Broker database record that supplies the per-instance endpoints/credentials the provider operates on.

This lets a new broker be added as a provider class (and shared helpers) instead of forking SkyPortal or re-deriving the same integration in every deployment.

Concepts#

  • BrokerAPI (skyportal/broker_apis/interface.py), the base class. Every operation is a stub that raises NotImplementedError; a provider overrides only what it supports.

  • Registry, providers are listed in the BROKERS tuple in skyportal/broker_apis/__init__.py. The broker_classnames Postgres enum (skyportal/enum_types.py) is derived from it, so provider names are validated at the database level (mirrors api_classnames). Append new providers to the end of the tuple to keep the enum stable.

  • Capabilities, BrokerAPI.implements() reports which operations a provider overrode. Handlers gate on it, and the frontend can show/hide features accordingly.

  • Broker model (skyportal/models/broker.py), one configured broker: name, broker_classname (which provider), active, ingest, and encrypted altdata (endpoints/credentials, mirroring Allocation.altdata). Only system admins may create/update/delete brokers, and altdata is redacted from non-admins. active makes the connection usable on demand; ingest is what subscribes the instance to its stream, and the two are set separately because a broker worth searching by hand is not necessarily one to consume in full.

  • Site defaults, default_alert_search, default_crossmatch and default_photometry name the broker the source page’s “Search alerts” button opens, the one its cross-matches (cone searches) run against, and the one serving the photometry the source page displays (its saved points merged with photometry pulled from the broker on demand, cached on disk for misc.minutes_to_keep_broker_photometry_cache and never written to the database). At most one broker holds each; pick them on the brokers page. The seeded data makes ALeRCE the alert search and cross-match one; no broker serves photometry until one is picked. The fetched points are scoped to the requester exactly like the alert search: a (survey, programid) group is only displayed to users whose streams cover that programid, so ZTF partnership and Caltech points stay hidden from a public-only user. PhotStat is still recomputed over everything fetched, matching how it is already computed over every saved row regardless of who is looking. Making an active broker a default re-runs its test_connection, so an unreachable one is refused; a broker that goes down later never breaks a source page, the lightcurve falls back to the saved photometry, the failure is logged, and a broker that times out is skipped for a minute so an outage does not make every source page wait for one. An object the broker simply does not know (a 404) is cached as empty and leaves the passthrough on for every other object.

Operations#

Interactive (SkyPortal → broker): query_alerts, get_alert, get_cutouts, cone_search, save_as_source, get_photometry, and filter management (get_filters, create_filter, update_filter, delete_filter, test_filter, filter_modules).

save_as_source and get_photometry come free with get_alert, but the photometry passthrough runs on every source page view under a 10s bound, so a provider whose object fetch cannot meet that sets photometry_passthrough = False and stops advertising the capability. ANTARES (walks its whole paginated alert history) and Pitt-Google (bills the deployment for a BigQuery job per view) opt out. Lasair advertises it, and reads the object from BOOM rather than from Lasair.

Ingestion (broker → SkyPortal): run_ingestion, a long-lived consumer/poller (see “Ingestion and filters” below).

Ingestion and filters (end to end)#

Once a broker is configured (a Broker record with valid altdata), ingestion, running its filters and pulling in matching alerts, is driven by a background service and gated by config.

1. Enable the ingestion service#

The broker_ingest service (services/broker_ingest/) runs one task per broker that is active, has ingest set, and whose provider implements run_ingestion. It is off by default; enable it in your config and restart the app:

brokers:
  ingest_enabled: true

With it disabled the service idles (it does not exit), so nothing is ever polled. Confirm it via make monitor (the broker_ingest entry should be RUNNING) and log/broker_ingest.log, which prints either broker ingestion disabled ... or starting ingestion for broker <id> (<name>).

brokers.ingest_enabled is a config value, not in the database, not in a broker’s altdata, and not exposed via the API. A broker’s run_ingestion: true capability only means the provider supports ingestion, not that it is running.

2. Attach filters#

A filter_kind: "query" broker (e.g. Lasair) runs one query per SkyPortal Filter linked to it. Create a Filter (on a Stream + your Group), then attach the query, for Lasair the SELECT / FROM / WHERE parts, from the broker page (or POST /api/brokers/<id>/filters/<filter_id>). Filters are re-read every poll cycle, so adding/editing one does not require a restart.

3. When and how often#

run_ingestion polls immediately on start, then sleeps altdata.poll_interval seconds (provider default; e.g. Lasair 86400 = daily, ALeRCE/ANTARES 3600), so the first batch appears within seconds of the service starting, not after a full interval. Changing poll_interval (or other broker altdata) is read once when a broker’s task starts, so restart the ingestion service to apply it.

4. Where results land#

Matching objects are registered as Candidates under the passing filter and appear on the Candidates / Scanning page filtered by it, not as Sources. From there you review and save the ones worth keeping.

5. Auto-save#

To save passing objects as Sources automatically (into the filter’s group) instead of only registering candidates, set the filter’s autosave flag, the “Auto-save passing objects as sources” checkbox in the Lasair filter builder, or POST /api/brokers/<id>/filters/<filter_id> with {"autosave": true}. Objects already saved to that group are skipped. Use it only when the filter is selective enough to trust everything it passes; otherwise keep it off and tighten the filter’s conditions so scanning stays manageable.

6. Catching up on what the stream missed#

A stream reaches back only as far as the broker retains it (Lasair keeps roughly seven days), and a consumer resumes from where it last stopped. Two controls on the filter page cover what that leaves out, and neither needs brokers.ingest_enabled:

  • Backfill from broker runs the filter’s own saved query against the broker’s database and ingests each match, so it reaches past the stream’s retention. For Lasair that is one query on the requesting user’s own token, with each match’s photometry and cutouts read from BOOM, capped at brokers.backfill.limit objects. The broker_backfill service runs one at a time. Offered by providers that set supports_backfill.

  • Replay stream bumps the credential’s stream_epoch, which suffixes the Kafka consumer group. The broker holds no offset for a group it has not seen, so the next run starts at the earliest message it still retains. Use it after fixing a topic that was delivering nothing usable; a filter whose topic was mis-routed or on the wrong stream has alerts still sitting in the broker’s cache.

Neither moves an existing consumer offset, so both are safe to run while ingestion is consuming.

Endpoints#

  • GET/POST/PATCH/DELETE /api/brokers[/{id}], manage Broker records.

  • GET /api/brokers/{id}/alerts, query alerts by objectId, candid, or ra/dec/radius. Records come back as the provider returns them, and jd_start/jd_end apply only where the provider supports them.

  • GET /api/brokers/{id}/objects/{object_id}, one object as the provider returns it.

  • GET /api/brokers/{id}/objects/{object_id}/photometry and POST /api/brokers/{id}/objects/{object_id}/save.

  • GET /api/brokers/{id}/alerts/{alert_id}/cutouts, the science, template and difference cutouts, keyed by the candid (BOOM, Babamul) or the objectId (ALeRCE, ANTARES, Fink, Lasair).

  • The /api/brokers/{id}/alerts/{object_id} forms of the three object routes still work, for callers written against them.

  • POST /api/brokers/{id}/filters/{filter_id}/backfill, queue a one-off ingest of what the broker still holds for a filter. Returns once queued.

  • POST /api/brokers/{id}/credentials/replay, read the stream again from its earliest retained message. Resets the caller’s own account, or the broker’s shared one when they have no credentials and administer it.

  • GET /api/internal/broker_apis, capabilities + config schema of every registered provider (for the frontend).

Writing a provider#

  1. Add skyportal/broker_apis/mybroker.py with a MYBROKER(BrokerAPI) class overriding the operations you support. Read per-instance config from broker.altdata. Provide form_json_schema_config (and optionally ui_json_schema, surveys, validate_config).

  2. Append it to BROKERS in skyportal/broker_apis/__init__.py.

  3. A database migration for the extended broker_apis enum is generated automatically.

See skyportal/broker_apis/generic.py (GENERICBROKER) for a working reference that talks to any REST broker via a configured base_url/token.

Contents