version-watcher — just tell me when there's something new

Renovate came up when I was looking at ways to track upstream versions. It's popular, well-maintained, and does exactly what it advertises — it opens pull requests when dependencies have newer versions available. That's the problem for me.

I've had updates break things. A new Helm chart ships with a changed default value. A container image moves a config path. An app release includes a database migration that doesn't go cleanly. When those things happen, I want to have been the one who decided to update, not a tool that decided for me. Even without auto-merge, I didn't want a queue of open update PRs to manage. I just wanted to know.

I went looking for something that would watch a mix of sources — GitHub releases, Docker Hub images, GHCR containers, PyPI packages, Helm charts — and send me a Slack message when something new was out. I couldn't find anything that did just that without dragging in a full dependency management workflow. So I wrote version-watcher.

The script

The whole thing is a single watcher.py with two dependencies: httpx for HTTP and pyyaml for config parsing.

#!/usr/bin/env python3
"""
version-watcher: Polls GitHub releases, PyPI, Docker Hub, GHCR, and Helm repos
for new versions and posts Slack notifications when changes are found.

Config:   /config/watches.yaml  (mounted from ConfigMap)
State:    /state/versions.json  (mounted from PVC — persists between runs)
Env:      SLACK_WEBHOOK_URL     (required)
          GITHUB_TOKEN          (optional — raises API rate limit)
"""

import json
import os
import re
import sys
from pathlib import Path

import httpx
import yaml

CONFIG_PATH = Path(os.environ.get("WATCHES_PATH", "/config/watches.yaml"))
STATE_PATH  = Path(os.environ.get("STATE_PATH",  "/state/versions.json"))

SLACK_WEBHOOK  = os.environ["SLACK_WEBHOOK_URL"]
GITHUB_TOKEN   = os.environ.get("GITHUB_TOKEN", "")

SEMVER_RE = re.compile(r"^v?(\d+\.\d+(?:\.\d+)?)")


# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------

def load_state() -> dict:
    if STATE_PATH.exists():
        return json.loads(STATE_PATH.read_text())
    return {}


def save_state(state: dict) -> None:
    STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
    STATE_PATH.write_text(json.dumps(state, indent=2))


# ---------------------------------------------------------------------------
# Version checkers
# ---------------------------------------------------------------------------

def _github_headers() -> dict:
    headers = {
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    if GITHUB_TOKEN:
        headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"
    return headers


def get_github_latest(repo: str) -> tuple[str, str]:
    r = httpx.get(
        f"https://api.github.com/repos/{repo}/releases/latest",
        headers=_github_headers(),
        timeout=15,
    )
    r.raise_for_status()
    data = r.json()
    return data["tag_name"].lstrip("v"), data["html_url"]


def get_pypi_latest(package: str) -> tuple[str, str]:
    r = httpx.get(f"https://pypi.org/pypi/{package}/json", timeout=15)
    r.raise_for_status()
    data = r.json()
    version = data["info"]["version"]
    return version, f"https://pypi.org/project/{package}/{version}/"


def _pick_latest_semver_tag(tags: list[str]) -> str | None:
    """From a list of tag strings, return the highest semver-like tag."""
    def sort_key(tag: str):
        m = SEMVER_RE.match(tag)
        return tuple(int(x) for x in m.group(1).split(".")) if m else (0,)

    candidates = [t for t in tags if SEMVER_RE.match(t) and "latest" not in t]
    if not candidates:
        return None
    candidates.sort(key=sort_key, reverse=True)
    return candidates[0]


def get_dockerhub_latest(image: str) -> tuple[str, str]:
    if "/" not in image:
        image = f"library/{image}"
    r = httpx.get(
        f"https://hub.docker.com/v2/repositories/{image}/tags/",
        params={"ordering": "last_updated", "page_size": 50},
        timeout=15,
    )
    r.raise_for_status()
    results = r.json().get("results", [])
    tag_names = [t["name"] for t in results]
    latest = _pick_latest_semver_tag(tag_names) or (tag_names[0] if tag_names else "unknown")
    return latest.lstrip("v"), f"https://hub.docker.com/r/{image}/tags"


def get_ghcr_latest(image: str) -> tuple[str, str]:
    """image: 'org/package'  (e.g. 'someorg/someapp')"""
    org, pkg = image.split("/", 1)
    pkg_encoded = pkg.replace("/", "%2F")
    headers = _github_headers()

    for endpoint in [
        f"https://api.github.com/orgs/{org}/packages/container/{pkg_encoded}/versions",
        f"https://api.github.com/users/{org}/packages/container/{pkg_encoded}/versions",
    ]:
        r = httpx.get(endpoint, headers=headers, params={"per_page": 10}, timeout=15)
        if r.status_code == 200:
            break
    r.raise_for_status()

    for v in r.json():
        tags = v.get("metadata", {}).get("container", {}).get("tags", [])
        winner = _pick_latest_semver_tag(tags)
        if winner:
            return winner.lstrip("v"), f"https://github.com/{image}/releases"

    # Fallback: first available tag
    for v in r.json():
        tags = v.get("metadata", {}).get("container", {}).get("tags", [])
        if tags:
            return tags[0].lstrip("v"), f"https://ghcr.io/{image}"

    return "unknown", f"https://ghcr.io/{image}"


def get_helm_latest(repo_url: str, chart: str) -> tuple[str, str]:
    r = httpx.get(f"{repo_url.rstrip('/')}/index.yaml", timeout=15)
    r.raise_for_status()
    index = yaml.safe_load(r.text)
    entries = index.get("entries", {}).get(chart, [])
    if not entries:
        raise ValueError(f"Chart '{chart}' not found in Helm repo {repo_url}")
    # Helm index is already sorted newest first
    return entries[0]["version"], repo_url


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

def check_version(watch: dict) -> tuple[str, str]:
    t = watch["type"]
    if t == "github":
        return get_github_latest(watch["repo"])
    elif t == "pypi":
        return get_pypi_latest(watch["package"])
    elif t == "docker":
        return get_dockerhub_latest(watch["image"])
    elif t == "ghcr":
        return get_ghcr_latest(watch["image"])
    elif t == "helm":
        return get_helm_latest(watch["repo"], watch["chart"])
    else:
        raise ValueError(f"Unknown watch type: {t!r}")


# ---------------------------------------------------------------------------
# Slack
# ---------------------------------------------------------------------------

_TYPE_EMOJI = {
    "github": ":github:",
    "pypi":   ":python:",
    "docker": ":whale:",
    "ghcr":   ":package:",
    "helm":   ":helm:",
}


def notify_slack(
    name: str,
    old: str | None,
    new: str,
    url: str,
    watch_type: str,
    running: str | None = None,
) -> None:
    emoji   = _TYPE_EMOJI.get(watch_type, ":bell:")
    old_str = f"`{old}`" if old else "_first seen_"
    text    = f"{emoji} *New release: {name}*\n{old_str} → *`{new}`*"
    if running and running != new:
        text += f"  ·  _you're running `{running}`_"
    payload = {
        "blocks": [
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": text},
                "accessory": {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "View Release"},
                    "url": url,
                },
            }
        ]
    }
    httpx.post(SLACK_WEBHOOK, json=payload, timeout=10).raise_for_status()


def notify_slack_running_behind(
    name: str, running: str, latest: str, url: str, watch_type: str
) -> None:
    emoji   = _TYPE_EMOJI.get(watch_type, ":bell:")
    payload = {
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"{emoji} *Update available: {name}*\n"
                        f"Running *`{running}`* · Latest `{latest}`"
                    ),
                },
                "accessory": {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "View Release"},
                    "url": url,
                },
            }
        ]
    }
    httpx.post(SLACK_WEBHOOK, json=payload, timeout=10).raise_for_status()


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> int:
    config = yaml.safe_load(CONFIG_PATH.read_text())
    state  = load_state()
    errors = 0

    for watch in config.get("watches", []):
        name    = watch["name"]
        running = watch.get("running_version", "").strip() or None
        try:
            version, url = check_version(watch)
            prev = state.get(name)

            if prev != version:
                notify_slack(name, prev, version, url, watch["type"], running)
                state[name] = version
                state.pop(f"{name}.__run_notified", None)

            elif running and running != version:
                last_notified = state.get(f"{name}.__run_notified")
                if last_notified != version:
                    notify_slack_running_behind(name, running, version, url, watch["type"])
                    state[f"{name}.__run_notified"] = version

        except Exception as exc:
            print(f"  ERR  {name}: {exc}", file=sys.stderr)
            errors += 1

    save_state(state)
    return errors


if __name__ == "__main__":
    sys.exit(main())

The container is python:3.12-slim, runs as an unprivileged user (UID 1000), and the image is built and pushed to GHCR via a GitHub Actions workflow on any change to the docker/version-watcher/ path.

FROM python:3.12-slim

ARG VERSION=1.0.0

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY watcher.py .

RUN useradd -m -u 1000 watcher
USER watcher

CMD ["python", "/app/watcher.py"]

How it's configured

The watch list lives in a watches.yaml mounted from a Kubernetes ConfigMap. It's templated by Ansible so the running_version fields get populated from group vars at deploy time, but the structure is straightforward:

watches:
  # Apps
  - name: Ghost
    type: github
    repo: TryGhost/Ghost
    running_version: "5.120.1"

  - name: Mealie
    type: github
    repo: mealie-recipes/mealie
    running_version: "2.8.0"

  # Infrastructure
  - name: Talos Linux
    type: github
    repo: siderolabs/talos
    running_version: "1.9.5"

  - name: Longhorn
    type: helm
    repo: https://charts.longhorn.io
    chart: longhorn
    running_version: "1.8.1"

  - name: kube-prometheus-stack
    type: helm
    repo: https://prometheus-community.github.io/helm-charts
    chart: kube-prometheus-stack

  - name: some-pypi-tool
    type: pypi
    package: some-pypi-tool

  - name: My GHCR Image
    type: ghcr
    image: myorg/myimage

How it's deployed

The Ansible role creates the namespace, a Kubernetes Secret containing the Slack webhook URL, the ConfigMap with watches.yaml, a small PVC for the state file, and the CronJob itself. The CronJob spec is the interesting part:

apiVersion: batch/v1
kind: CronJob
metadata:
  name: version-watcher
  namespace: version-watcher
spec:
  schedule: "0 */6 * * *"
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 3
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: Never
          securityContext:
            fsGroup: 1000
          containers:
            - name: version-watcher
              image: ghcr.io/myorg/version-watcher:1.0.0
              env:
                - name: SLACK_WEBHOOK_URL
                  valueFrom:
                    secretKeyRef:
                      name: version-watcher-secrets
                      key: SLACK_WEBHOOK_URL
                - name: WATCHES_PATH
                  value: /config/watches.yaml
                - name: STATE_PATH
                  value: /state/versions.json
              volumeMounts:
                - name: config
                  mountPath: /config
                  readOnly: true
                - name: state
                  mountPath: /state
              resources:
                requests:
                  cpu: "50m"
                  memory: "128Mi"
                limits:
                  cpu: "200m"
                  memory: "256Mi"
          volumes:
            - name: config
              configMap:
                name: version-watcher-config
            - name: state
              persistentVolumeClaim:
                claimName: version-watcher-state

concurrencyPolicy: Forbid means if a run is somehow still going when the next one is due (unlikely given it's checking a handful of APIs, but possible if something hangs), it won't stack. The webhook URL comes in from the Secret via secretKeyRef — nothing sensitive in the ConfigMap or the image.

Tracking what's actually running

Each watch entry also accepts an optional running_version field. If you set it to the version currently deployed in the cluster, version-watcher adds a second check: if the upstream hasn't changed since the last run but your running version is still behind, it fires a separate "you're behind" notification. That fires once per upstream version so it doesn't repeat every six hours. I update running_version values in Ansible group vars after each deployment, so the watcher always knows what's actually live.

It's been ticking away quietly and doing exactly what I wanted. Something ships a new release, Slack tells me, I go read the changelog, I decide whether now is a good time to update. No surprises, no automated changes, no cleaning up after a broken deployment.