Lab03

app_python/.dockerignore

@@ -0,0 +1,16 @@
+.git
+.gitignore
+
+*.pyc
+*.pyo
+venv/
+.venv/
+
+.env
+*.pem
+secrets/
+
+*.md
+docs/
+
+tests/

app_python/.gitignore

@@ -0,0 +1,12 @@
+# Python
+__pycache__/
+*.py[cod]
+venv/
+*.log
+
+# IDE
+.vscode/
+.idea/
+
+# OS
+.DS_Store

app_python/Dockerfile

@@ -0,0 +1,24 @@
+FROM python:3.14.3-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get upgrade -y && \
+    apt-get clean && rm -rf /var/lib/apt/lists/* && \
+    pip install --no-cache-dir --upgrade pip
+
+RUN useradd --create-home appuser && \
+    chown -R appuser:appuser /app
+
+COPY --chown=appuser:appuser requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY --chown=appuser:appuser app.py .
+
+USER appuser
+
+HEALTHCHECK --interval=30s --timeout=10s --start-period=30s \
+  CMD python -c "import urllib.request, json; urllib.request.urlopen('http://localhost:5000/health', timeout=5)" || exit 1
+
+EXPOSE 5000
+
+CMD ["python", "app.py"]

app_python/README.md

@@ -0,0 +1,65 @@
+## Overview
+
+This service provides realtime application, system, and networking data relevant for training in DevOps practices. The data includes app specifics, OS specifics, runtime stats, request data, and available endpoints listing.
+
+## Prerequisites
+
+Python `3.12.x+`.
+
+## Installation
+
+```py
+python -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+```
+
+## Running the Application
+
+Run with default parameters (`HOST=0.0.0.0`, `PORT=5000`):
+```bash
+python app.py
+```
+
+Or with custom config:
+
+```bash
+HOST=<your-value> PORT=<your-value> DEBUG=True python app.py
+```
+
+## API Endpoints
+
+- GET / - Service and system information
+- GET /health - Health check
+
+**Configuration:**
+
+| Environment Variable | Effect                                                                                  | Default Value |
+| -------------------- | --------------------------------------------------------------------------------------- | ------------- |
+| HOST                 | Specifies the host for uvicorn                                                          | `0.0.0.0`     |
+| PORT                 | Specified the launch port for uvicorn                                                   | `5000`        |
+| DEBUG                | Specifies whether to include debug information into responses (currently has no effect) | `False`       |
+
+
+## Docker
+
+This section provides command patterns for using the application in a containerized manner.
+
+
+To build the image locally, execute this command with substitued values:
+
+```bash
+docker build -t <name>:<x>.<y>.<z> 
+```
+
+To run a container, execute this command with substitued values:
+
+```bash
+docker run -d -p <external-port>:5000 <name>:<x>.<y>.<z> 
+```
+
+To pull the specific version of the image from DockerHub, execute this command with substitued values:
+
+```bash
+docker pull controlw/devops-info-service:<x>.<y>.<z> 
+```

app_python/app.py

@@ -0,0 +1,108 @@
+import logging
+import os
+
+import sys
+
+from app_stats import AppStats
+from datetime import datetime, timezone
+from pythonjsonlogger import jsonlogger
+from typing import Any
+
+import uvicorn
+from fastapi import FastAPI, Request, Response
+from fastapi.responses import JSONResponse
+
+app = FastAPI()
+app_logger = logging.getLogger("app")
+
+if not app_logger.handlers:
+    handler = logging.StreamHandler(sys.stdout)
+    formatter = jsonlogger.JsonFormatter(reserved_attrs=[], timestamp=True)
+    handler.setFormatter(formatter)
+    app_logger.addHandler(handler)
+
+request_logger = logging.getLogger("app.request")
+error_logger = logging.getLogger("app.error")
+
+request_logger.setLevel(level=logging.INFO)
+error_logger.setLevel(level=logging.ERROR)
+
+
+app_stats = AppStats(name="devops-info-service",
+                     description="DevOps course info service",
+                     major_version=1,
+                     minor_version=0,
+                     patch_version=0)
+
+@app.get("/", description="Service information")
+async def root(request: Request):
+    request_info = {
+        "client_ip": request.client.host,
+        "user_agent": request.headers.get('user-agent'),
+        "method": request.method,
+        "path": request.url.path
+    }
+
+    endpoints_info = list()
+    for _, key in enumerate(endpoint_paths):
+        path = key
+        method = next(iter(endpoint_paths[key]))
+        description = endpoint_paths[key][method]['description']
+        endpoints_info.append({"path": path, "method": method.upper(), "description": description})
+
+    return {
+        "service": app_stats.provide_service_info(),
+        "system": app_stats.provide_system_info(),
+        "runtime": app_stats.provide_runtime_info(),
+        "request": request_info,
+        "endpoints": endpoints_info
+    }
+
+@app.get("/health", description="Health check")
+async def check_health():
+    return {
+        'status': 'healthy',
+        'timestamp': datetime.now(timezone.utc).isoformat(),
+        'uptime_seconds': int(app_stats.get_uptime())
+    }
+
+@app.exception_handler(Exception)
+async def handle_general_exception(request: Request, exception: Exception):
+    error_logger.error(
+        "Unhandled exception occured in a request",
+        extra={
+            "path": request.url.path,
+            "method": request.method
+        },
+        exc_info=True)
+
+    return JSONResponse(
+        status_code=500,
+        content={"error": "Internal Server Error"}
+    )
+
+@app.middleware("http")
+async def log_request(request: Request, call_next):
+    call_time = datetime.now(timezone.utc)
+    response: Response = await call_next(request)
+    execution_time = datetime.now(timezone.utc) - call_time
+    request_logger.info(
+        "New request was processed",
+        extra={
+            "path": request.url.path,
+            "method": request.method,
+            "execution_time": execution_time,
+            "status_code": response.status_code
+        })
+    return response
+
+
+endpoint_paths: dict[str, Any] = app.openapi()['paths']
+
+if __name__ == "__main__":
+    HOST = os.getenv('HOST', '0.0.0.0')
+    PORT = int(os.getenv('PORT', 5000))
+    DEBUG = os.getenv('DEBUG', 'False').lower() == 'true'
+    logging.getLogger("uvicorn").disabled = True
+    logging.getLogger("uvicorn.access").disabled = True
+    uvicorn.run(app, host=HOST, port=PORT, log_config=None)

app_python/app_stats.py

@@ -0,0 +1,67 @@
+import platform
+import socket
+
+from datetime import datetime, timezone
+from typing import Dict
+
+
+class AppStats:
+    def __init__(self, name: str, description: str, major_version: int, minor_version: int, patch_version: int):
+        if len(name) == 0 or name is None:
+            raise ValueError("The service name must not be empty!")
+
+        if len(description) == 0 or description is None:
+            raise ValueError("The service description must not be empty!")
+
+        if major_version < 0 or minor_version < 0 or patch_version < 0:
+            raise ValueError("The service version must be non negative!")
+
+        self.start_time = datetime.now(timezone.utc)
+        self.name = name
+        self.description = description
+        self.version = f"{major_version}.{minor_version}.{patch_version}"
+        self.framework = "FastAPI"
+
+    def provide_service_info(self) -> Dict[str, Any]: 
+        return {
+            "name": self.name,
+            "version": self.version,
+            "description": self.description,
+            "framework": self.framework
+        }
+
+    def provide_system_info(self) -> Dict[str, Any]:
+        return {
+            "hostname": socket.gethostname(),
+            "platform": platform.system(),
+            "platform_version": platform.version(),
+            "architecture": platform.machine(),
+            "cpu_count": os.cpu_count(),
+            "python_version": platform.python_version()
+        }
+
+    def get_uptime(self) -> float:
+        delta = datetime.now(timezone.utc) - self.start_time
+        return delta.total_seconds()
+
+    def provide_runtime_info(self) -> Dict[str, str]:
+        uptime_seconds = int(self.get_uptime())
+        uptime_minutes = (uptime_seconds // 60) % 60
+        uptime_hours = uptime_seconds // 3600
+
+        if uptime_hours > 1 or uptime_hours == 0:
+            hours_adapted_wording = "hours"
+        else:
+            hours_adapted_wording = "hour"
+
+        if uptime_minutes > 1 or uptime_minutes == 0:
+            minutes_adapted_wording = "minutes"
+        else:
+            minutes_adapted_wording = "minute"
+
+        return {
+            "uptime_seconds": uptime_seconds,
+            "uptime_human": f"{uptime_hours} {hours_adapted_wording}, {uptime_minutes} {minutes_adapted_wording}",
+            "current_time": datetime.now(timezone.utc).isoformat(),
+            "timezone": "UTC"
+        }