Compare commits

5 Commits

Author SHA1 Message Date
Pascal Scheiben
0a14e9abf2 Arranging imports 2025-09-18 13:49:54 +02:00
Pascal Scheiben
d5c2ee172f Fixing typos 2025-09-18 12:10:29 +02:00
Pascal Scheiben
2fb893df90 Enhancing comments, adding stub for business logic 2025-09-18 12:09:33 +02:00
Pascal Scheiben
3d15b89b95 Rewriting comments 2025-09-18 12:05:03 +02:00
Pascal Scheiben
d00c35ccb5 Adding config endpoint 2025-09-18 11:49:34 +02:00
25 changed files with 61 additions and 304 deletions

11
.env
View File

@@ -1,3 +1,8 @@
cluster_inventory_path = config/inventory.yml # Environment variables for NetApp ONTAP clusters
redis_host = '172.16.0.208' CLUSTER1_HOSTNAME=172.16.57.2
redis_port = '6379' CLUSTER1_USERNAME=admin
CLUSTER1_PASSWORD=Netapp12
CLUSTER2_HOSTNAME=172.16.56.2
CLUSTER2_USERNAME=admin
CLUSTER2_PASSWORD=Netapp12

View File

@@ -1,9 +0,0 @@
FROM python:latest
WORKDIR /usr/local/bin
COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt
COPY src/start.py .
CMD ["src/start.py"]

View File

@@ -1,19 +1,3 @@
# generic_api_endpoint # generic_api_endpoint
Hackathon API endpoint Hackathon API endpoint
## management summary // usecase
This API acts as a middelware for service portals and frontends (like SNOW), that can retrieve data via REST API. It manages metadata.
## ideas for future
- store the data in redis on initialization or on first request
- also first query redis, and not directly ONTAP
- documentation -> make it understandable, so that users will use it!
- add capability to apply filters/conditions on the return
- Alexeys
-
- performance based filtering
- add capability for finding best clusters, volumes
- get credentials from credential-mgmt-system
-

Binary file not shown.

Before

Width:  |  Height:  |  Size: 71 KiB

View File

@@ -1,6 +0,0 @@
- hostname: "172.16.57.2"
username: "admin"
password: "Netapp12"
- hostname: "172.16.56.2"
username: "admin"
password: "Netapp12"

View File

View File

@@ -2,6 +2,3 @@ fastapi[standard]>=0.116.2
httpx>=0.28.1 httpx>=0.28.1
redis>=6.4.0 redis>=6.4.0
python-dotenv>=1.1.1 python-dotenv>=1.1.1
pydantic
redis[hiredis]
dotenv

View File

@@ -1,3 +0,0 @@
cluster_inventory_path = ./config/inventory.yml
redis_host = '172.16.0.208'
redis_port = '6379'

View File

@@ -1,5 +1,5 @@
from src.example.router import router as example_router from src.example.router import router as example_router
from src.aggregate.aggregate_router import router as aggregate_router from .aggregate_router import router as aggregate_router
__all__ = ["example_router", "aggregate_router"] __all__ = ["example_router", "aggregate_router"]

View File

@@ -1,23 +1,23 @@
# contains the router for the aggregates endpoint # contains the router for the aggregates endpoint
from fastapi import APIRouter, Query, Request from enum import Enum
from typing import List, Dict from typing import List
from .aggregate_schema import AggregateSchema, MetricEnum
from fastapi import APIRouter, Query
from .aggregate_schema import AggregateSchema
from .aggregate_service import get_aggregates from .aggregate_service import get_aggregates
class MetricEnum(str, Enum):
relative = "relative"
absolute = "absolute"
router = APIRouter(tags=["aggregates"]) router = APIRouter(tags=["aggregates"])
@router.get("/aggregates", response_model=List[AggregateSchema]) @router.get("/aggregates", response_model=List[AggregateSchema])
async def aggregates_endpoint( async def aggregates_endpoint(
request: Request,
metric: MetricEnum = Query(MetricEnum.relative, description="Metric type"), metric: MetricEnum = Query(MetricEnum.relative, description="Metric type"),
): ):
# Extract tag parameters from query string return await get_aggregates(metric)
tags: Dict[str, str] = {}
for param_name, param_value in request.query_params.items():
if param_name.startswith("tag."):
tag_key = param_name[4:]
tags[tag_key] = param_value
return await get_aggregates(request, metric, tags)

View File

@@ -1,23 +1,8 @@
# contains the schema definitions for aggregates # contains the schema definitions for aggregates
from pydantic import BaseModel from pydantic import BaseModel
from enum import Enum
class AggregateSchema(BaseModel): class AggregateSchema(BaseModel):
aggregate: str aggregate: str
node: str node: str
available: int available: str
available_str: str
class MetricEnum(str, Enum):
relative = "relative"
absolute = "absolute"
TAG2REST = {
'worm_compliance': { 'snaplock_type': 'compliance' },
'worm_enterprise': { 'snaplock_type': 'enterprise' },
'flash': { 'block_storage.storage_type': 'ssd' },
'hdd': { 'block_storage.storage_type': 'hdd' },
'mcc': { 'block_storage.mirror.enabled': 'true' }
}

View File

@@ -1,58 +1,25 @@
# contains the business logic for aggregates # contains the business logic for aggregates
from typing import List
from typing import List, Dict from .aggregate_schema import AggregateSchema
from pprint import pprint
from fastapi import Request
from src.aggregate.aggregate_schema import AggregateSchema, MetricEnum
from logging import getLogger
from src.utils import round_bytes, get_data_from_ontap
logger = getLogger("uvicorn")
logger.setLevel("DEBUG")
# TAG2REST = { async def get_aggregates(metric: str = "relative") -> List[AggregateSchema]:
# 'worm_compliance': { 'snaplock_type': 'compliance' },
# 'worm_enterprise': { 'snaplock_type': 'enterprise' },
# 'flash': { 'block_storage.storage_type': 'ssd' },
# 'hdd': { 'block_storage.storage_type': 'hdd' },
# 'mcc': { 'block_storage.mirror.enabled': 'true' }
# }
# {
# "flash": "production",
# "performance": "gold",
# "worm": "compliance"
# }
async def get_aggregates(request: Request, metric: str = "relative", tags: Dict[str, str] = None) -> List[AggregateSchema]:
# Dummy data for demonstration # Dummy data for demonstration
# You can use the metric parameter to filter or modify results as needed # You can use the metric parameter to filter or modify results as needed
# For now, just return the same data and show metric usage # For now, just return the same data and show metric usage
logger.debug(f"Metric used: {metric}") print(f"Metric used: {metric}")
logger.debug(f"Tags used: {tags}")
# convert tags to ONTAP filter
# filter_str = ""
# if tags:
# str_filter_parts = [f"tag.{key} eq '{value}'" for key, value in tags.items()]
# param_str = "&".join([f"{TAG2REST[key]}" for key, value in tags.items()])
__aggregates = await get_data_from_ontap(request, logger, "storage/aggregates", "fields=*")
pprint(__aggregates)
if metric == MetricEnum.relative:
__aggregates = sorted(__aggregates, key=lambda r: r["space"]["block_storage"].get("used_percent"), reverse=True)
elif metric == MetricEnum.absolute:
__aggregates = sorted(__aggregates, key=lambda r: r["space"]["block_storage"].get("available"), reverse=False)
aggregates: list = [ aggregates: list = [
AggregateSchema( AggregateSchema(
aggregate=a["name"], aggregate="Aggregate A", node="cluster01-01", available="100.0TB"
node=a["node"]["name"], ),
available=a["space"]["block_storage"]["available"], AggregateSchema(
available_str=round_bytes(a["space"]["block_storage"]["available"]), aggregate="Aggregate B", node="cluster01-01", available="200.5GB"
) ),
for a in __aggregates AggregateSchema(
aggregate="Aggregate C", node="cluster01-02", available="300.75MB"
),
] ]
return aggregates return aggregates

3
src/config/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from src.config.router import router as config_router
__all__ = ["config_router"]

View File

@@ -6,7 +6,7 @@ from .schema import ConfigReturnSchema, ConfigSchema
logger = logging.getLogger("uvicorn") logger = logging.getLogger("uvicorn")
router = APIRouter(tags=["config_upload"]) router = APIRouter(tags=["config"])
@router.post( @router.post(
@@ -20,4 +20,3 @@ async def create_config(config: ConfigSchema) -> ConfigSchema:
""" """
logger.info("Received configuration data") logger.info("Received configuration data")
return config return config

View File

@@ -1,4 +1,4 @@
# contains the schema definitions for the config_upload service # contains the schema definitions for the config service
from pydantic import BaseModel from pydantic import BaseModel

2
src/config/service.py Normal file
View File

@@ -0,0 +1,2 @@
# contains the business logic for the config service
async def save_config() -> None: ...

View File

@@ -1,3 +0,0 @@
from src.config_upload.router import router as config_router
__all__ = ["config_router"]

View File

@@ -1,2 +0,0 @@
# contains the business logic for the config_upload service
async def save_config() -> None: ...

View File

@@ -1,40 +0,0 @@
import json
import logging
from redis import Redis, ConnectionError
from typing import List
from pydantic import TypeAdapter
from src.schema import ConfigSchema
def setup_db_conn(redishost, redisport: str):
''' Setup Redis connection and return it open'''
log = logging.getLogger('uvicorn')
try:
redisclient = Redis(host=redishost, port=redisport, decode_responses=True)
if redisclient.ping():
log.info(f"Connected to Redis DB {redishost} on port {redisport}")
else:
log.error(f"Cannot connect to Redis DB {redishost} on port {redisport}")
exit(1)
return redisclient
except ConnectionError as e:
print(f"FATAL: Redis DB {redishost} is unreachable on port {redisport}. Err: {e}")
return None
except Exception as e:
print(f"FATAL: {e}")
return None
def get_inventory_from_redis(redisclient: Redis):
''' Read inventory from Redis '''
cluster_inv = redisclient.hgetall('cluster_inventory')
if 'inventory' in cluster_inv:
return json.loads(cluster_inv['inventory'])
return {}
def get_config_from_db(redisclient: Redis) -> ConfigSchema:
''' Load inventory to global vars'''
GLOBAL_INVENTORY = get_inventory_from_redis(redisclient)
GLOBAL_INVENTORY_VALID = TypeAdapter(List[ConfigSchema]).validate_python(GLOBAL_INVENTORY)
return GLOBAL_INVENTORY_VALID

View File

@@ -1,7 +1,7 @@
# contains the router for the aggregate endpoint # contains the router for the aggregate endpoint
from fastapi import APIRouter from fastapi import APIRouter
from src.example.schema import ExampleSchema from .schema import ExampleSchema
router = APIRouter(tags=["aggregate"]) router = APIRouter(tags=["aggregate"])

View File

@@ -1,18 +1,7 @@
# contains the schema definitions for the aggregate service # contains the schema definitions for the aggregate service
from pydantic import BaseModel from pydantic import BaseModel
from pathlib import Path
class ExampleSchema(BaseModel): class ExampleSchema(BaseModel):
example_field: str example_field: str
another_field: int another_field: int
class ClusterCreds(BaseModel):
"""A structure to hold basic auth cluster credentials for a cluster"""
username: str
password: str
hostname: str = None
cert_filepath: Path = None
key_filepath: Path = None

View File

@@ -1,45 +0,0 @@
import os
import json
import logging
import yaml
from pathlib import Path
from dotenv import load_dotenv
from src.database import setup_db_conn
from src.schema import ConfigSchema
from typing import List
from pydantic import TypeAdapter
def initialize_config():
load_dotenv()
log = logging.getLogger("uvicorn")
ENV_INVENTORYPATH = os.getenv("cluster_inventory_path")
ENV_REDISHOST = os.getenv("redis_host")
ENV_REDISPORT = os.getenv("redis_port")
log.info(f"Found Cluster Inventory file at: {ENV_INVENTORYPATH}")
if not ENV_INVENTORYPATH or not Path(ENV_INVENTORYPATH).is_file():
print(f"FATAL: Inventory file {ENV_INVENTORYPATH} is missing or not a file.")
return False
try:
with open(ENV_INVENTORYPATH, "r") as f:
inv = yaml.safe_load(f)
inventory = json.dumps(inv)
except Exception as e:
print(f"FATAL: Cannot read inventory file {ENV_INVENTORYPATH}. Err: {e}")
return False
log.info(f"Importing configuration to DB...")
try:
GLOBAL_INVENTORY_VALID = TypeAdapter(List[ConfigSchema]).validate_python(inv)
redis_conn = setup_db_conn(ENV_REDISHOST, ENV_REDISPORT)
redis_conn.hset("cluster_inventory", mapping={"inventory": inventory})
redis_conn.close()
log.info("Configuration has been loaded.")
return True
except Exception as e:
print(f"FATAL: Redis DB error: {e}")
return False

View File

@@ -1,49 +1,27 @@
import os
import logging import logging
import httpx
from fastapi import FastAPI from fastapi import FastAPI
from src.aggregate import aggregate_router from src.aggregate import aggregate_router
from src.config_upload import config_router from src.config import config_router
from src.service import load_config
from contextlib import asynccontextmanager
from .database import setup_db_conn, get_config_from_db
from src.initialize import initialize_config
from .utils import setup_logging
logger = logging.getLogger("uvicorn") logger = logging.getLogger("uvicorn")
logger.setLevel("DEBUG")
logger.info("Starting application") logger.info("Starting application")
config = load_config()
app = FastAPI()
@asynccontextmanager
async def lifespan(app: FastAPI):
"""make loading it async"""
global shared_redis_conn, requests_client
log = logging.getLogger("uvicorn")
cfg_init_result = initialize_config()
shared_redis_conn = setup_db_conn(os.getenv("redis_host"), os.getenv("redis_port"))
if not shared_redis_conn:
log.error("Cannot connect to Redis DB. Exiting...")
exit(1)
inv_check = get_config_from_db(shared_redis_conn)
log.info(f"[DEBUG] Data validity healthcheck (DEVELOPER MODE): {inv_check}")
if not cfg_init_result:
log.error("Configuration initialization failed. Exiting...")
# exit(1)
requests_client = httpx.AsyncClient(verify=False)
yield {"redis_conn": shared_redis_conn, "requests_client": requests_client}
await requests_client.aclose()
log.info("Shutting down FastAPI app...")
setup_logging()
log = logging.getLogger("uvicorn")
log.info("Starting FastAPI app...")
app = FastAPI(lifespan=lifespan)
app.include_router(aggregate_router) app.include_router(aggregate_router)
app.include_router(config_router) app.include_router(config_router)
@app.get("/")
async def main():
return {"Hello": "World"}
@app.get("/config")
async def get_config():
"""Endpoint to get the current configuration."""
return config.model_dump()

View File

@@ -1,44 +0,0 @@
import logging
from fastapi import Request
import httpx
from src.database import get_config_from_db
def round_bytes(size_in_bytes: int) -> str:
# Helper function to convert bytes to a human-readable format
for unit in ["B", "KiB", "MiB", "GiB", "TiB", "PiB"]:
if size_in_bytes < 1024:
return f"{size_in_bytes:.2f}{unit}"
size_in_bytes /= 1024
return f"{size_in_bytes:.2f}EB"
async def get_data_from_ontap(request: Request, logger, endpoint: str, query_string: str = ""):
# get clusters from redis
redis_conn = request.state.redis_conn
config = get_config_from_db(redis_conn)
logger.debug("Got the config from REDIS: %s", config)
results = []
client = request.state.requests_client
for cluster in config:
print(f"\n\n looping, {cluster}")
url = f"https://{cluster.hostname}/api/{endpoint}"
if query_string:
url += f"?{query_string}"
try:
logger.debug(f"Fetching data from ONTAP: {url}")
response = await client.get(url, auth=(cluster.username, cluster.password))
response.raise_for_status()
results.extend(response.json()["records"])
except httpx.HTTPError as e:
logger.error(f"HTTP error occurred: {e}")
return None
return results
def setup_logging() -> None:
"""Configure logging for the application"""
logging.basicConfig(level=logging.DEBUG, format="[%(asctime)s] [%(levelname)5s] %(message)s")
print("Logger is initialized.")