Painless Pipelines
I used /_update_by_query in Painless Enrichment to update my documents without reindexing. That’s fine if you only want to do this once. Of course, those fields will get stale as prices and stock levels change.
What you want is some way to run these enrichments against your products index on a continuous basis. You can save Painless scripts in Elasticsearch and run them with a systemd timer. I chose a timer instead of a cron job for the reasons in Systemd Timers vs. Cron Jobs. It’ll wait until Elasticsearch is running, and you can see the output in your journal.
I decided to have two enrichment scripts, one for price and one for stock, rather than one massive script. Now they’re easier to test and can be rolled back independently.
Map the new fields
You’ll need to add a mapping for your new fields. You already have the first three from your one-off enrichment, so you just need to add timestamps. I added one for each script so you know when price and stock were each last updated.
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 your mapping in place, you can save your Painless scripts with a version number. If you ever need to make a change that could break things, you can save a new script with -v2 on the end. Rolling back would be as simple as pointing your runner to the -v1 script.
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;
"""
}
}
As you can see, they’ll both update their timestamp even if the inputs don’t exist. If we didn’t do this, any product that didn’t have a price would be selected by our “never enriched” query and be rewritten each time.
Run the jobs
Your jobs will select any documents that haven’t been enriched or are out of date and run /_update_by_query on them. These will run asynchronously, and the runner below will poll /_tasks until one completes, then move on to the next.
As both of these scripts are updating the same documents, we want to avoid running them at the same time. By setting conflicts=proceed, we’ll keep track of the number of conflicts and skip 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
}
}'
We want to update prices every other run and stock every run, so each window will be a couple of minutes less than a multiple of 15 minutes. If your job is started a little late, it won’t skip anything. Using curl -f and jq -e will return a non-zero exit code if there’s an HTTP error or a task with failures. Systemd will then show that run as failed.
Schedule
Our service will just call our script, and our timer will call the service 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
I have it using Requires and After since Elasticsearch is running on the same machine. Type=oneshot will ensure that a long-running job doesn’t overlap with the next job. Run this to reload systemd and enable your timer:
sudo systemctl daemon-reload
sudo systemctl enable --now products-enrichment.timer
Monitor
list-timers will display when your last job ran and when it will run next. You should also see a line in your journal that summarizes each job:
systemctl list-timers products-enrichment.timer
journalctl -u products-enrichment.service -n 100 --no-pager
Grab some docs to make sure it’s working:
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?
For this, yes. If you’re trying to enrich data in one Elasticsearch index with a couple of idempotent scripts that run every so often, this will be super easy to debug and keep costs low.
If you need to run jobs on many systems or need more robust retries and SLAs, check out Airflow or Dagster.