Painless Pipelines
In Painless Enrichment, I used /_update_by_query to enrich documents in place without a full reindex. That’s fine for a one-off backfill, but the derived fields go stale every time a price or stock level changes.
The fix is a repeatable enrichment pipeline for the same products index, with stored Painless scripts running on a systemd timer. I prefer timers over cron here for the reasons in Systemd Timers vs. Cron Jobs: the job can wait for Elasticsearch to start, and its output gets logged to the journal, which makes it easy to debug.
I also split enrichment by concern (pricing and inventory) instead of building one giant script. Smaller scripts are easier to test, and you can roll one back without touching the other.
Map the new fields
Start by mapping the fields these jobs write. The first three already exist from the one-off backfill, so only the timestamps are new. Each script gets its own timestamp, so a pricing run doesn’t hide stale inventory data.
PUT /products/_mapping
{
"properties": {
"discount_pct": { "type": "integer" },
"price_band": { "type": "keyword" },
"stock_state": { "type": "keyword" },
"pricing_enriched_at": { "type": "date" },
"inventory_enriched_at": { "type": "date" }
}
}
Store the scripts
With the mapping in place, store each script under a versioned ID. A breaking change becomes a new -v2 script, and rolling back means pointing the runner at -v1 again.
PUT /_scripts/products-pricing-signals-v1
{
"script": {
"lang": "painless",
"source": """
if (ctx._source.list_price != null &&
ctx._source.list_price > 0 &&
ctx._source.price != null) {
double discount =
(ctx._source.list_price - ctx._source.price) * 100.0 / ctx._source.list_price;
ctx._source.discount_pct = (int) Math.round(discount);
}
if (ctx._source.price != null) {
if (ctx._source.price < 25) {
ctx._source.price_band = "budget";
} else if (ctx._source.price < 100) {
ctx._source.price_band = "mid";
} else {
ctx._source.price_band = "premium";
}
}
ctx._source.pricing_enriched_at = params.now;
"""
}
}
PUT /_scripts/products-inventory-state-v1
{
"script": {
"lang": "painless",
"source": """
if (ctx._source.inventory_count != null) {
if (ctx._source.inventory_count <= 0) {
ctx._source.stock_state = "out_of_stock";
} else if (ctx._source.inventory_count < 10) {
ctx._source.stock_state = "low_stock";
} else {
ctx._source.stock_state = "in_stock";
}
}
ctx._source.inventory_enriched_at = params.now;
"""
}
}
Both scripts update their timestamp even when the inputs are missing. Otherwise a product without a price would match the “never enriched” query and get rewritten on every run.
Run the jobs
Each job is an async /_update_by_query that only targets documents that were never enriched or have gone stale. The runner below polls /_tasks until a job finishes before starting the next one.
Both scripts write to the same documents, so running them in parallel would cause version conflicts. conflicts=proceed counts those and quietly skips them.
#!/usr/bin/env bash
set -euo pipefail
ES_URL="http://127.0.0.1:9200"
INDEX="products"
NOW_UTC="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"
run_job() {
local script_id="$1"
local query_json="$2"
local task_id task
task_id=$(curl -fsS -X POST "$ES_URL/$INDEX/_update_by_query?conflicts=proceed&wait_for_completion=false&slices=auto" \
-H "Content-Type: application/json" \
-d "{
\"query\": $query_json,
\"script\": {
\"id\": \"$script_id\",
\"params\": { \"now\": \"$NOW_UTC\" }
}
}" | jq -r '.task')
while true; do
task=$(curl -fsS "$ES_URL/_tasks/$task_id")
if [ "$(jq -r '.completed' <<<"$task")" = "true" ]; then
break
fi
sleep 2
done
echo "job finished: $script_id $(jq -c '{updated: .response.updated, failures: (.response.failures | length)}' <<<"$task")"
jq -e '.error == null and (.response.failures | length) == 0' <<<"$task" > /dev/null
}
run_job "products-pricing-signals-v1" '{
"bool": {
"should": [
{ "bool": { "must_not": { "exists": { "field": "pricing_enriched_at" } } } },
{ "range": { "pricing_enriched_at": { "lt": "now-25m" } } }
],
"minimum_should_match": 1
}
}'
run_job "products-inventory-state-v1" '{
"bool": {
"should": [
{ "bool": { "must_not": { "exists": { "field": "inventory_enriched_at" } } } },
{ "range": { "inventory_enriched_at": { "lt": "now-10m" } } }
],
"minimum_should_match": 1
}
}'
Pricing refreshes every other run and inventory every run. Each window is a few minutes short of a multiple of the 15-minute schedule, so a run that starts late won’t skip anything. curl -f and jq -e turn an HTTP error or a task with failures into a non-zero exit, which systemd records as a failed run.
Schedule
The service runs the script, and the timer kicks it off every 15 minutes:
# /etc/systemd/system/products-enrichment.service
[Unit]
Description=Run product enrichment scripts
After=elasticsearch.service
Requires=elasticsearch.service
[Service]
Type=oneshot
User=search
Group=search
ExecStart=/opt/search/bin/run-products-enrichment.sh
# /etc/systemd/system/products-enrichment.timer
[Unit]
Description=Run product enrichment every 15 minutes
[Timer]
OnCalendar=*:0/15
Persistent=true
RandomizedDelaySec=20
[Install]
WantedBy=timers.target
Elasticsearch runs on the same host here, so Requires and After make sure it’s up before the jobs start. And because the service is Type=oneshot, a slow run doesn’t overlap with the next one. Reload systemd and enable the timer to get it going:
sudo systemctl daemon-reload
sudo systemctl enable --now products-enrichment.timer
Monitor
list-timers shows the next and last run, and the journal has a summary line for each job:
systemctl list-timers products-enrichment.timer
journalctl -u products-enrichment.service -n 100 --no-pager
To spot-check the data, pull a few documents:
GET /products/_search
{
"size": 5,
"_source": [
"name",
"price",
"list_price",
"discount_pct",
"price_band",
"pricing_enriched_at",
"inventory_count",
"stock_state",
"inventory_enriched_at"
],
"query": {
"match_all": {}
}
}
Production-ready?
Yes, for the right scope. If you’re enriching one Elasticsearch index with a small set of idempotent scripts on a predictable schedule, this systemd-based approach is simple to debug and cheap to run.
Once the workflow spans multiple systems or needs real retries and SLAs, an orchestrator like Airflow or Dagster is usually a better fit.