check_woodpecker_upgrade/check_woodpecker_upgrade

#!/usr/bin/python3
"""
Icinga/Nagios plugin to check Woodpecker CI instance upgrade status.

Copyright (c) 2024 Benjamin Renard <brenard@zionetrix.net>

This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License version 3
as published by the Free Software Foundation.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
"""

import argparse
import logging
import os
import re
import subprocess
import sys
import traceback

import requests

parser = argparse.ArgumentParser()

parser.add_argument("-d", "--debug", action="store_true")
parser.add_argument("-v", "--verbose", action="store_true")
parser.add_argument(
    "-p", "--path", type=str, help="Woodpecker CI bin path", default="woodpecker-server"
)
parser.add_argument(
    "-U",
    "--url",
    type=str,
    help="Woodpecker CI releases URL",
    default="https://api.github.com/repos/woodpecker-ci/woodpecker/releases",
)
parser.add_argument(
    "--pre-release",
    action="store_true",
    help="Allow pre-release (default: only stable release are considered)",
)
parser.add_argument(
    "--draft",
    action="store_true",
    help="Allow draft release (default: only stable release are considered)",
)
parser.add_argument(
    "-t", "--timeout", type=int, help="Specify timeout for HTTP requests (default: 20)", default=20
)
parser.add_argument("-u", "--upgrade", action="store_true", help="Upgrade Woodpecker CI")
parser.add_argument("-f", "--force", action="store_true", help="Force upgrade Woodpecker CI")
parser.add_argument("--arch", help="System dpkg architecture (default: auto-detect)")

options = parser.parse_args()

logging.basicConfig(
    level=logging.DEBUG if options.debug else (logging.INFO if options.verbose else logging.WARNING)
)

CURRENT = None

cmd = [options.path, "--version"]
logging.debug("Command use to retrieve current version of Woodpecker CI: %s", " ".join(cmd))
OUTPUT = None
EXCEPTION = None
try:
    OUTPUT = subprocess.check_output(cmd)
    logging.debug("Output:\n%s", OUTPUT)
    m = re.search("version ([^ ]+)$", OUTPUT.decode("utf8", errors="ignore"))
    if m:
        CURRENT = m.group(1).strip()
except Exception as err:  # pylint: disable=broad-except
    EXCEPTION = err
logging.debug("Current version: %s", CURRENT)

if not CURRENT:
    print("UNKNOWN - Fail to retrieve current Woodpecker CI")
    print(f'Command: {" ".join(cmd)}')
    print("Output:")
    print(OUTPUT if OUTPUT else "")
    print("Exception:")
    print(EXCEPTION if EXCEPTION else "")
    sys.exit(3)

CURRENT = CURRENT.replace("+", "-")
logging.debug("Cleaned current version: %s", CURRENT)

LATEST = None
try:
    logging.debug("Get releases from %s...", options.url)
    r = requests.get(options.url, timeout=options.timeout)
    data = r.json()
    logging.debug("Data retrieve:\n%s", data)
    for item in data:
        if not options.pre_release and item["prerelease"]:
            logging.debug("Ignore pre-release %s", item["name"])
            continue
        if not options.draft and item["draft"]:
            logging.debug("Ignore draft release %s", item["name"])
            continue
        LATEST = item
        break
except Exception:  # pylint: disable=broad-except
    logging.debug(
        "Exception occurred retrieving latest Woodpecker CI release from the Github API:\n%s",
        traceback.format_exc(),
    )

if LATEST is None:
    print("UNKNOWN - Fail to retrieve latest Woodpecker CI release from the Github API")
    print(f"Current version: {CURRENT}")
    sys.exit(3)

logging.debug("Latest version is %s", LATEST["name"])

if LATEST["name"] == CURRENT and not (options.upgrade and options.force):
    print(
        f"OK - The latest release of Woodpecker CI is currently used "
        f"({LATEST['name']}, published on {LATEST['published_at']})"
    )
    sys.exit(0)

if options.upgrade:
    if not options.arch:
        logging.info("Auto-detect dpkg architecture...")
        options.arch = (
            subprocess.check_output(["dpkg-architecture", "--query", "DEB_HOST_ARCH"])
            .decode("utf8")
            .strip()
        )
        logging.info("Auto-detected dpkg architecture: %s", options.arch)

    logging.info("List installed Woodpecker package...")
    INSTALLED = (
        subprocess.check_output(["dpkg-query", "-Wf", "${Package}\\n", "woodpecker-*"])
        .decode("utf8")
        .strip()
        .split("\n")
    )
    logging.info("Installed Woodpecker packages: %s", ", ".join(INSTALLED))

    for PACKAGE in INSTALLED:
        PACKAGE_ASSET = None
        for asset in LATEST["assets"]:
            if re.match(
                r"^" + PACKAGE + r"_" + LATEST["name"] + "_" + options.arch + r"\.deb$",
                asset["name"],
            ):
                PACKAGE_ASSET = asset
                break
        if not PACKAGE_ASSET:
            logging.warning("No asset found for package %s", PACKAGE)
            continue

        PATH = f"/tmp/{PACKAGE_ASSET['name']}"
        logging.info(
            "Download package %s %s to %s (%s)",
            PACKAGE,
            LATEST["name"],
            PATH,
            PACKAGE_ASSET["browser_download_url"],
        )
        r = requests.get(PACKAGE_ASSET["browser_download_url"], timeout=options.timeout)
        with open(PATH, "wb") as fp:
            fp.write(r.content)

        logging.info("Install package %s %s", PACKAGE, LATEST["name"])
        subprocess.run(["dpkg", "-i", PATH], check=True)

        logging.info("Remove temporary file")
        os.remove(PATH)

        STATUS = subprocess.run(
            ["systemctl", "is-active", PACKAGE], check=False, capture_output=True
        )
        if STATUS.stdout.decode().strip() == "active":
            logging.info("Service %s is active, restart it", PACKAGE)
            subprocess.run(["systemctl", "restart", PACKAGE], check=True)
        else:
            logging.info("No service %s is running", PACKAGE)

    sys.exit(0)

print(
    "WARNING - The version of Woodpecker CI currently used is not the latest "
    f"('{CURRENT}' vs '{LATEST['name']}', published on {LATEST['published_at']})"
)
print(f"URL: {LATEST['html_url']}")
sys.exit(1)