Upgrade Server Fleets Selected by Asset Metadata

Fleet upgrades fail when target selection relies on stale spreadsheets. This blueprint selects its targets from live data: it queries the Kestra asset catalog for io.kestra.plugin.ee.assets.VM assets in a chosen environment whose python_version metadata differs from the desired version, upgrades each machine, and writes the new version back into the asset record in the same task. A JSON export ships the refreshed catalog to external systems, and an error branch feeds failures to a Gemini-powered AI agent for root cause analysis.

How it works

1. Inputs set the target_version (default 3.11) and the env to upgrade (staging or prod).
2. query_assets (io.kestra.plugin.core.debug.Return) calls assets(metadata={"env": inputs.env, "os": "Debian"}) and filters with jq to keep only servers whose metadata.python_version differs from the target. Selection is pure metadata, no hardcoded host lists.
3. run_upgrades (io.kestra.plugin.core.flow.ForEach) iterates over the matches. get_server_details (io.kestra.plugin.core.execution.SetVariables) unpacks id, IP, and current version.
4. ssh_python_update (io.kestra.plugin.scripts.shell.Commands) simulates the upgrade over SSH. Crucially, its assets block declares the server as input and re-registers it as output with python_version set to the target, so the catalog reflects reality the moment the upgrade lands.
5. export_assets_to_file (io.kestra.plugin.ee.assets.AssetShipper with io.kestra.plugin.ee.assets.FileAssetExporter) exports the catalog as JSON for CMDBs or reporting.
6. On any failure, the errors branch runs analyze_error (io.kestra.plugin.ai.agent.AIAgent with io.kestra.plugin.ai.provider.GoogleGemini), feeding {{ errorLogs() }} to Gemini 3.5 Flash Lite and logging the diagnosis.

What you get

• Metadata-driven targeting: only outdated machines in the chosen env are touched.
• Self-updating inventory, since the upgrade task rewrites the asset record it acted on.
• A JSON catalog export after each run for downstream systems.
• AI-generated root cause analysis in the logs whenever a run fails.
• Webhook and (disabled by default) weekly schedule triggers.

Who it's for

• Platform and SRE teams running rolling runtime upgrades across fleets.
• Operations teams that need inventory to update itself during maintenance.
• Teams exploring AI-assisted incident analysis inside their workflows.

Why orchestrate this with Kestra

Configuration management tools can push a new Python, but they cannot select targets from a governed asset catalog, update that catalog transactionally with the change, or explain their own failures. Kestra combines the assets() query function, per-server ForEach task runs with isolated logs, first-class error branches, and an AI agent task in one declarative file, with schedules and webhooks to run it hands-free.

Prerequisites

• A Kestra Enterprise Edition instance (Assets are an EE feature).
• VM assets with env, os, python_version, and ip_address metadata, for example registered by the asset-provision-servers blueprint.
• A Google AI Studio API key for the Gemini error analysis.

Secrets

• UPGRADE_WEBHOOK_KEY: shared secret guarding the Webhook trigger.
• GEMINI_API_KEY: Google Gemini API key used by the AIAgent error task.

Quick start

1. Add both secrets to your Kestra instance.
2. Run asset-provision-servers first so VM assets with metadata exist.
3. Import this blueprint and execute it with your target env and version.
4. Check the per-server task runs, the refreshed asset metadata, and the exported JSON.

How to extend

• Replace the echo stub with io.kestra.plugin.fs.ssh.Command or an Ansible playbook for real upgrades.
• Enable the every_sunday_morning schedule for a weekly patch window.
• Swap the FileAssetExporter for the OpenLineage exporter to feed a lineage backend.
• Add a Pause approval before touching prod, gated on the env input with a Switch task.

Links

• Kestra Assets documentation: https://kestra.io/docs/enterprise/governance/assets
• AI plugin docs: https://kestra.io/plugins/plugin-ai