Broker ingestion

Broker ingestion#

Ingestion runs in services/broker_ingest, one asyncio task per active broker whose provider implements run_ingestion. It is off unless the config says otherwise:

brokers:
  ingest_enabled: True
  ingest_processes: 1

ingest_processes sets how many OS processes supervisor starts. Process 0 runs every broker. Higher-index processes run only providers that are safe to replicate, which means Kafka consumers, since Kafka rebalances a consumer group across them. REST pollers are not duplicated.

What a survey needs#

An alert becomes a candidate only if all of these are true:

  1. The broker subscribes to its topic. Topics come from the broker’s altdata['kafka']['topics'].

  2. The survey has a zeropoint. ZP_PER_SURVEY in skyportal/broker_apis/_save.py maps the survey name to an AB zeropoint. A survey missing from it raises on every alert, which shows up in the ingest log. Flux in nJy is scaled to Jy and takes 8.9, and magnitudes are converted against the same zeropoint.

  3. Its bands map to filters. Filters are named <survey><band> by default. A survey that names its filters otherwise needs an entry in BAND_TO_FILTER_PER_SURVEY, alongside the zeropoints.

  4. A stream carries it. The stream’s altdata names the collection and the program ids it admits.

  5. Filters are paired with the broker’s own. A Filter row carries the broker’s filter id in altdata['boom']['filter_id'], and ingestion maps that back to the SkyPortal filter.

Names do not have to match#

Four names are involved and they are set independently.

  • The survey name is what the alert carries, and it is the key into ZP_PER_SURVEY.

  • The instrument is the one named after the survey, unless INSTRUMENT_PER_SURVEY in skyportal/utils/survey.py maps it to another (DECAM is stored under DECam).

  • The stream name is a SkyPortal label for the same data.

  • The topic is the broker’s, usually <survey>_alerts_results.

Where the configuration lives#

Broker data, including the topic list and the credentials, lives in Broker.altdata, set through the API or the /brokers/{id} page.

Adding a survey to a broker#

  1. Add <SURVEY> to ZP_PER_SURVEY if it is not already there. Use the zeropoint the survey’s photometry is already on, so light curves stay continuous across a change of broker.

  2. Check how its filters are named. If they are not <survey><band>, add the band mapping.

  3. Add the topic to the broker’s altdata['kafka']['topics'].

  4. Check that the stream exists, and that each filter carries its partner’s id in altdata.

  5. Restart broker_ingest. Topics are read once, when the consumer starts.

Replaying a backlog#

A consumer group that has already read a topic will not read it again, and a broker configured with auto_offset_reset: latest starts at the end. To take in messages a broker has missed, add a second broker row that reads only that topic, with its own group_id and auto_offset_reset: earliest, then deactivate it once it has caught up.

Re-consuming an alert is safe. The save path looks for an existing candidate with the same passing_alert_id before adding one, so a replay that overlaps normal ingestion does not double up.

Retention is the limit. A backlog older than the topic’s retention is gone, whatever the offsets say.

Contents