Cirrus Streaming Updater

The Cirrus Streaming Updater (SUP) updates the elasticsearch indexes for each and every mediawiki edit. The updater consists of two applications, a producer and a consumer. One producer runs per datacenter reading the events from various topics and generating a unified stream of updates that need to be applied to the search clusters. One consumer runs per cluster group (eqiad, codfw, cloudelastic) to write to. The producer only reads events from it's local datacenter. The consumer reads events from all producers in all datacenters.

The chain of events between a user clicking the 'save page' button and elasticsearch being updated is roughly as follows:

MW core approves of the edit and generates an event in the mediawiki.page-change stream.
The event is read by the streaming updater producer in the same datacenter, aggregated with related events, and results in a cirrussearch.update_pipeline.update event.
The consumer receives the event, fetches the content of the update from the mediawiki api, batches together many updates, and performs a bulk write request to the appropriate elasticsearch cluster (each consumer writes to a cluster group of three elasticsearch clusters).

In addition to page change events, a significant source of updates are mediawiki.cirrussearch.page_rerender events. These events represent changes that did not change the revision_id, but likely changed the rendered result of the page (ex: propagated template updates). While these are the higher volume inputs, the updater reads one more set of topics related to the CirrusSearch Weighted Tags functionality. These come in over a variety of streams and are generally metadata generated by async processes such as ML models or batch data processing.

There is a DPE Deep Dive from 2024-09-03 "Search Update Pipeline - Feeding search with Flink stream processing" ( Recording )

Backfilling/Reindexing

The streaming updater can also backfill existing indices for periods of time where updates, for whatever reason, were not written to the elasticsearch cluster. The backfill uses a custom helm release, which runs a second copy of the standard consumer with specific constraints on kafka offsets and wikis to process. See the "Backfill Batch" section of the cirrus-streaming-updater README .

In-place reindexing is similar, but uses different arguments. There is the top level entrypoint (python -m cirrus_reindexer ) for backfilling, and then another one ( python -m cirrus_reindexer.reindex_all ) . See the Cirrus Reindex Orchestrator repo for more details.

Deployment

The flink jobs run in 3 different environments:

wikikube staging
- producer: consuming all DCs topics and filtering on a couple wikis: testwiki, frwiki, itwiki and officewiki
- consumer-search: writing to relforge
wikikube eqiad
- producer: consumer all eqiad prefixed topics
- consumer-search: writing to the main search cluster in eqiad
- consumer-cloudelastic: writing to cloudelastic
wikikube codfw
- producer: consumer all codfw prefixed topics
- consumer-search: writing to the main search cluster in codfw

Prior to deploying a new version of the job you must ensure that it properly works in staging:

It is important to deploy a new version only if the job is properly working. If it is down you must investigate the reasons and bring the jobs back up before any attempts to upgrade it.

The deployment procedure is quite standard and can be done using helm:

cd /srv/deployment-charts/helmfile.d/services/cirrus-streaming-updater
helmfile apply -e staging -i

Please verify the diff and abort if you see an unexpected change. It may happen that someone else has tuned some settings, if in doubt please ask. Diffs you may see that you can safely ignore:

restartNonce: XXX
savepointNonce: XXX

Let the job run for a while and monitor its output with:

 kubectl get flinkdeployments.flink.apache.org

A healthy deployment should show

NAME                        JOB STATUS   LIFECYCLE STATE
flink-app-consumer-search   RUNNING      STABLE
flink-app-producer          RUNNING      STABLE

kube_env cirrus-streaming-updater staging
kubectl logs -l component=jobmanager -c flink-main-container -f | jq -r .message

If everything looks fine you can proceed to the next environments using the same procedure. But always check that the jobs are running properly before updating.

Individual jobs can be updated using a selector:

helmfile apply -e staging -i --selector name=producer

You can override some values as well:

helmfile apply -e staging -i --selector name=producer --set app.config_files.app\\.config\\.yaml.kafka-source-start-time=2025-09-18T16:00:00Z

But if you do so: please make sure to re-deploy the job after to make sure that the next deployer does not see some unexpected differences.

It is never normal to have to drop the flink state, if you run into this please take a moment to understand why the job is failing to load its state, it is likely that the new version you are pushing is introducing a bug.

Note that dropping the state may cause some data loss:

producer: the 10mins deduplication is in the state, and can contain a huge number of events in case the job was backlogged.
consumer: contains all in-flight requests to MW and the saneitizer state

San(e)itizing

San(e)itizing is a process to keep the CirrusSearch indices sane. Its primary purpose is to make sure pages that are out of date or missing in the search index will be (re-)indexed.

This process has a secondary purpose of ensuring all indexed pages have been rendered from wikitext within the last few months. It accomplishes this by indexing every n-th page it visits in such a way that after n loops over the dataset all pages will have been re-indexed.

This loop algorithm has been ported to the SUP consumer application as an additional, optional source of cirrussearch.update_pipeline.update events. It lives besides the regular kafka source, but only produces events locally at a constant, low rate.

Troubleshooting the Streaming Updater

This script outputs the kubernetes logs based on datacenter, helmfile release, and kubernetes environment.

Object Storage (Flink Savepoints/Checkpoints)

Flink automatically takes checkpoints during its course of operation, so it can recover from a crash. These checkpoints are written to the thanos-swift cluster via its S3 interface. If you are a member of the analytics-search-users group, you access the Cirrus Streaming Updater's thanos-swift account, you can login to the stat hosts and use the s3cmd cli: s3cmd -c /etc/s3cmd/cfg.d/search_platform.cfg ls