Files
latest_release_exporter/app/script.py

78 lines
2.4 KiB
Python

import os
import sys
import requests
from flask import Flask, request
from prometheus_client import generate_latest, Gauge, CollectorRegistry
import time
app = Flask(__name__)
# Create a CollectorRegistry
registry = CollectorRegistry()
# Create a Prometheus metric for the latest release
github_latest_release_metric = Gauge(
'github_latest_release',
'Latest release version of a GitHub repository',
['repo', 'version'],
registry=registry,
)
# GitHub API token (replace with your actual token)
github_api_token = os.environ.get('GITHUB_TOKEN')
if github_api_token is None:
sys.exit("Env variable GITHUB_TOKEN not found!")
# Function to fetch the latest release version or the latest tag from GitHub
def get_latest_version(repo_url):
api_url = f"https://api.github.com/repos/{repo_url}/releases/latest"
headers = {
'Authorization': f'Bearer {github_api_token}',
'Accept': 'application/vnd.github.v3+json'
}
response = requests.get(api_url, headers=headers)
# If the repository has releases, get the latest release version
if response.status_code == 200:
data = response.json()
return data["tag_name"]
# If the repository doesn't have releases, get the latest tag
if response.status_code == 404:
api_url = f"https://api.github.com/repos/{repo_url}/tags"
response = requests.get(api_url, headers=headers)
data = response.json()
if data:
return data[0]["name"]
# If both release and tag information is not available, return None
return None
@app.route('/metrics')
def metrics():
# Get the GitHub repository URL from the query parameter
repo_url = request.args.get('repo')
if not repo_url:
return "Error: Please provide a 'repo' parameter.", 400
try:
latest_version = get_latest_version(repo_url)
if latest_version:
print(f"Latest version: {latest_version}")
# Reset the metric to its initial state
github_latest_release_metric.clear()
# Set the metric value to 1 with appropriate labels
github_latest_release_metric.labels(repo=repo_url, version=latest_version).set(1)
except Exception as e:
print(f"Error: {e}")
return generate_latest(registry)
if __name__ == '__main__':
# Start the web service on port 8000
app.run(host='0.0.0.0', port=8000)