#!/usr/bin/env python3 """ GitOps Status Server API Simple Flask API for serving and updating status.json Listens on port 5000 and handles GET/POST requests """ import os import json import logging from flask import Flask, request, jsonify from datetime import datetime from flasgger import Swagger app = Flask(__name__) swagger = Swagger(app) # Configuration from environment STATUS_FILE = os.environ.get('STATUS_FILE', '/data/status.json') API_HOST = os.environ.get('API_HOST', '0.0.0.0') API_PORT = int(os.environ.get('API_PORT', 5000)) # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def load_status(): """Load the current status from file""" try: if os.path.exists(STATUS_FILE): with open(STATUS_FILE, 'r') as f: return json.load(f) else: logger.warning(f"Status file not found: {STATUS_FILE}") return { "repo": "unknown", "server": "unknown", "sync_status": "UNKNOWN", "drift_count": 0, "files": [], "last_check": "" } except Exception as e: logger.error(f"Error loading status: {e}") return {} def save_status(status): """Save the status to file""" try: # Ensure directory exists os.makedirs(os.path.dirname(STATUS_FILE), exist_ok=True) # Write with proper formatting with open(STATUS_FILE, 'w') as f: json.dump(status, f, indent=2) logger.info(f"Status updated: {status.get('repo', 'unknown')}/{status.get('server', 'unknown')} -> {status.get('sync_status', 'UNKNOWN')}") return True except Exception as e: logger.error(f"Error saving status: {e}") return False @app.route('/api/status', methods=['GET', 'POST', 'OPTIONS']) def api_status(): """ GitOps Status API endpoint --- get: summary: Retrieve current status responses: 200: description: Current GitOps status schema: type: object properties: repo: type: string server: type: string sync_status: type: string drift_count: type: integer files: type: array last_check: type: string post: summary: Update status with new data parameters: - in: body name: body required: true schema: type: object properties: repo: type: string example: "rsyslog" server: type: string example: "rsyslog-lab" sync_status: type: string enum: ["SYNCED", "OUT_OF_SYNC", "UNKNOWN", "PROGRESSING"] drift_count: type: integer files: type: array items: type: object last_check: type: string responses: 200: description: Status updated successfully 400: description: No JSON data provided 500: description: Failed to save status """ if request.method == 'OPTIONS': return '', 204 if request.method == 'GET': status = load_status() return jsonify(status), 200 if request.method == 'POST': try: incoming_data = request.get_json() if not incoming_data: return jsonify({"error": "No JSON data provided"}), 400 # Load current status status = load_status() # Update with incoming data (merge) status.update(incoming_data) # Add/update timestamp if not present if 'last_check' not in status or not status['last_check']: status['last_check'] = datetime.utcnow().isoformat() + 'Z' # Save updated status if save_status(status): return jsonify({ "success": True, "message": "Status updated successfully", "status": status }), 200 else: return jsonify({ "error": "Failed to save status" }), 500 except json.JSONDecodeError: return jsonify({"error": "Invalid JSON"}), 400 except Exception as e: logger.error(f"Error in POST /api/status: {e}") return jsonify({"error": str(e)}), 500 @app.route('/health', methods=['GET']) def health(): """ GET /health - Kubernetes liveness probe --- responses: 200: description: API is healthy """ return jsonify({"status": "healthy"}), 200 @app.route('/ready', methods=['GET']) def ready(): """ GET /ready - Kubernetes readiness probe --- responses: 200: description: API is ready to serve requests 503: description: API is not ready """ try: # Check if data directory is writable data_dir = os.path.dirname(STATUS_FILE) if not os.path.exists(data_dir): os.makedirs(data_dir, exist_ok=True) # Try to read or create status file if not os.path.exists(STATUS_FILE): # File doesn't exist yet, try to create it default_status = { "repo": "unknown", "server": "unknown", "sync_status": "UNKNOWN", "drift_count": 0, "files": [], "last_check": "" } save_status(default_status) # Verify we can read it status = load_status() if isinstance(status, dict): return jsonify({"status": "ready"}), 200 return jsonify({"status": "not_ready", "reason": "invalid status data"}), 503 except Exception as e: logger.error(f"Readiness check failed: {e}") return jsonify({"status": "not_ready", "error": str(e)}), 503 @app.route('/', methods=['GET']) def root(): """ GET / - API information and available endpoints --- responses: 200: description: API metadata and endpoint list """ return jsonify({ "name": "GitOps Status API", "version": "1.0.0", "endpoints": { "GET /status.json": "Retrieve current status", "GET /api/status": "Retrieve current status (JSON API)", "POST /api/status": "Update status with new data", "GET /health": "Liveness probe", "GET /ready": "Readiness probe" } }), 200 if __name__ == '__main__': logger.info(f"Starting GitOps Status API on {API_HOST}:{API_PORT}") logger.info(f"Status file location: {STATUS_FILE}") # Create directory if it doesn't exist os.makedirs(os.path.dirname(STATUS_FILE) or '.', exist_ok=True) app.run(host=API_HOST, port=API_PORT, debug=False, threaded=True)