#!/usr/bin/env python3 """ GitOps Status Server API Multi-server status API organized by server type Supports: rsyslog, Splunk, IBM ITNM, nginx, etc. """ import os import json import logging from flask import Flask, request, jsonify from datetime import datetime, UTC 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_data(): """Load the current data 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}, creating new structure") return {} except Exception as e: logger.error(f"Error loading status: {e}") return {} def save_data(data): """Save the data to file""" try: # Ensure directory exists os.makedirs(os.path.dirname(STATUS_FILE) or '.', exist_ok=True) # Write with proper formatting with open(STATUS_FILE, 'w') as f: json.dump(data, f, indent=2) logger.info(f"Data saved successfully") return True except Exception as e: logger.error(f"Error saving data: {e}") return False @app.route('/api//status', methods=['GET', 'POST', 'OPTIONS']) def api_server_status(server_type): """ Server type status endpoint GET: Retrieve status for a server type POST: Update status for a server type --- get: summary: Retrieve status for server type parameters: - in: path name: server_type type: string required: true description: Server type (rsyslog, splunk, nginx, etc.) responses: 200: description: Server status post: summary: Update status for server type parameters: - in: path name: server_type type: string required: true description: Server type (rsyslog, splunk, nginx, etc.) - in: body name: body required: true schema: type: object properties: server: type: string example: "rsyslog-lab" sync_status: type: string enum: ["SYNCED", "OUT_OF_SYNC", "UNKNOWN", "FAILED"] example: "SYNCED" drift_count: type: integer example: 0 deployed_files: type: array items: type: object properties: name: type: string example: [{"name": "rsyslog.conf"}] drifted_files: type: array items: type: object properties: name: type: string example: [] last_check: type: string example: "2026-04-26T00:00:00Z" responses: 200: description: Status updated successfully """ if request.method == 'OPTIONS': return '', 204 if request.method == 'GET': try: data = load_data() # Return status for this server type if server_type not in data: return jsonify({ "repo": server_type, "server": "unknown", "sync_status": "UNKNOWN", "drift_count": 0, "deployed_files": [], "drifted_files": [], "last_check": "" }), 200 return jsonify(data[server_type]), 200 except Exception as e: logger.error(f"Error in GET /api/{server_type}/status: {e}") return jsonify({"error": str(e)}), 500 if request.method == 'POST': try: incoming_data = request.get_json() if not incoming_data: return jsonify({"error": "No JSON data provided"}), 400 # Load current data data = load_data() # Get existing data for this server type or create new if server_type not in data: data[server_type] = {} # Update with incoming data (merge) data[server_type].update(incoming_data) # Set repo field data[server_type]['repo'] = server_type # Add/update timestamp if not present if 'last_check' not in data[server_type] or not data[server_type]['last_check']: data[server_type]['last_check'] = datetime.now(UTC).isoformat() # Save updated data if save_data(data): logger.info(f"Status updated for {server_type}: {data[server_type].get('server', 'unknown')} -> {data[server_type].get('sync_status', 'UNKNOWN')}") return jsonify({ "success": True, "message": "Status updated successfully", "status": data[server_type] }), 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/{server_type}/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 data = load_data() # Verify we can read it if isinstance(data, dict): return jsonify({"status": "ready"}), 200 return jsonify({"status": "not_ready", "reason": "invalid data structure"}), 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": "2.0.0", "description": "Multi-server status API organized by server type", "endpoints": { "GET /api/{server_type}/status": "Retrieve status for server type (rsyslog, splunk, nginx, etc.)", "POST /api/{server_type}/status": "Update status for server type", "GET /health": "Liveness probe", "GET /ready": "Readiness probe" }, "supported_server_types": [ "rsyslog", "splunk", "ibm-itnm", "nginx", "any-server-type" ], "examples": { "rsyslog": "GET/POST /api/rsyslog/status", "splunk": "GET/POST /api/splunk/status", "nginx": "GET/POST /api/nginx/status" } }), 200 if __name__ == '__main__': logger.info(f"Starting GitOps Status API v2.0 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)