feature: Add basic ProxLB API

* Add ProxLB API
  * Add DPM
  * Add Wake-on-Lan functions (by @remcohaszing)

Fixes: #141
Fixes: #217

.changelogs/1.2.0/141_add_dpm_integration.yml

@@ -0,0 +1,3 @@
+added:
+  - Add DPM support to turn off nodes in a cluster when not required (by @gyptazy) [#141]
+  - Add Wake-on-Lan functions based on pywakeonlan project (by @remcohaszing) [#141]

.changelogs/1.2.0/217_add_proxlb_api.yml

@@ -0,0 +1,2 @@
+added:
+  - Add basic ProxLB API interface based on FastAPI and Uvicorn (by @gyptazy) [#217]

.changelogs/1.2.0/release_meta.yml

@@ -0,0 +1 @@
+date: TBD

debian/control

@@ -7,6 +7,6 @@ Build-Depends: debhelper-compat (= 13), dh-python, python3-all, python3-setuptoo
 
 Package: proxlb
 Architecture: all
-Depends: ${python3:Depends}, ${misc:Depends}, python3-requests, python3-urllib3, python3-proxmoxer, python3-yaml 
+Depends: ${python3:Depends}, ${misc:Depends}, python3-requests, python3-urllib3, python3-proxmoxer, python3-yaml, python3-uvicorn, python3-fastapi
 Description: A DRS alike Load Balancer for Proxmox Clusters
  An advanced DRS alike loadbalancer for Proxmox clusters that also supports maintenance modes and affinity/anti-affinity rules.

proxlb/main.py

@@ -17,6 +17,7 @@ from utils.logger import SystemdLogger
 from utils.cli_parser import CliParser
 from utils.config_parser import ConfigParser
 from utils.proxmox_api import ProxmoxApi
+from models.proxlb_api import ProxlbApi
 from models.nodes import Nodes
 from models.guests import Guests
 from models.groups import Groups
@@ -50,6 +51,33 @@ def main():
     # Overwrite password after creating the API object
     proxlb_config["proxmox_api"]["pass"] = "********"
 
+    # Execute ProxLB API Server
+    proxlb_api = ProxlbApi(proxlb_config)
+    proxlb_api.run(proxlb_config)
+
+    # TEST
+    import time
+    time.sleep(3)
+    foo = Helper.http_client_get('http://127.0.0.1:8000/nodes')
+    print(foo)
+
+    #send
+    time.sleep(3)
+    data = {
+                    "name": "virt01",
+                    "wol_mac": "virt01",
+                    "locked": False,
+                    "ignore": False
+                }
+    Helper.http_client_post('http://127.0.0.1:8000/nodes/virt99', data)
+
+    time.sleep(3)
+    foo = Helper.http_client_get('http://127.0.0.1:8000/nodes')
+    print(foo)
+    ######
+    ######
+
+    # Execute guest balancing
     while True:
         # Get all required objects from the Proxmox cluster
         meta = {"meta": proxlb_config}

proxlb/models/proxlb_api.py

@@ -0,0 +1,181 @@
+"""
+ProxLB API
+"""
+
+
+__author__ = "Florian Paul Azim Hoberg <gyptazy>"
+__copyright__ = "Copyright (C) 2025 Florian Paul Azim Hoberg (@gyptazy)"
+__license__ = "GPL-3.0"
+
+
+import threading
+import uvicorn
+import utils.version
+from fastapi import FastAPI, HTTPException
+from utils.logger import SystemdLogger
+from fastapi.encoders import jsonable_encoder
+from pydantic import BaseModel
+from typing import Dict, Any
+
+logger = SystemdLogger()
+
+
+class ProxlbApi:
+    """
+    ProxLB API
+    """
+    def __init__(self, proxlb_config):
+        """
+        Initializes the API class with the ProxLB API object.
+        """
+        logger.debug("Starting: ProxLB API.")
+
+        # ProxLB API Object
+        self.proxlb_api = FastAPI()
+
+        # ProxLB API Model
+        class Node(BaseModel):
+            name: str = ""
+            wol_mac: str = ""
+            locked: bool = False
+            ignore: bool = False
+        # ProxLB API Data
+        nodes = {}
+
+        # ProxLB API Context Paths
+        @self.proxlb_api.get("/")
+        async def root():
+            return {"Application": "ProxLB", "version": f"{utils.version.__version__}", "status": "healthy"}
+
+        @self.proxlb_api.get("/status")
+        async def status():
+            self.api_exec_status()
+            return {"message": "status"}
+
+        @self.proxlb_api.get("/reboot")
+        async def reboot():
+            self.api_exec_reboot()
+            return {"message": "reboot"}
+
+        @self.proxlb_api.get("/shutdown")
+        async def shutdown():
+            self.api_exec_shutdown()
+            return {"message": "shutdown"}
+
+        @self.proxlb_api.get("/wol")
+        async def wol():
+            self.api_exec_wol()
+            return {"message": "wol"}
+
+        @self.proxlb_api.get("/update")
+        async def update():
+            self.api_exec_update()
+            return {"message": "update"}
+
+        @self.proxlb_api.get("/nodes")
+        async def get_node_items():
+            keys = []
+            for k, v in nodes.items():
+                keys.append(k)
+            return keys
+
+        @self.proxlb_api.get("/nodes/{item_id}", response_model=Node)
+        async def get_node_items(item_id: str):
+            return nodes[item_id]
+
+        @self.proxlb_api.patch("/nodes/{item_id}", response_model=Node)
+        async def update_node_items(item_id: str, item: Node):
+            stored_item_data = nodes[item_id]
+            stored_item_model = Node(**stored_item_data)
+            update_data = item.dict(exclude_unset=True)
+            updated_item = stored_item_model.copy(update=update_data)
+            nodes[item_id] = jsonable_encoder(updated_item)
+            return updated_item
+
+        @self.proxlb_api.post("/nodes/{item_id}", response_model=Node)
+        async def set_node_items(item_id: str, item: Node):
+            if item_id in nodes:
+                raise HTTPException(status_code=400, detail=f"Node: {item_id} already exists.")
+            print(item)
+            nodes[item_id] = jsonable_encoder(item)
+            return item
+
+        # CLI example
+        # curl -X POST "http://127.0.0.1:8000/nodes/virt01" -H "Content-Type: application/json" -d '{
+        #             "name": "virt01",
+        #             "wol_mac": "virt01",
+        #             "locked": false,
+        #             "ignore": false
+        #         }'
+
+        logger.debug("Finalized: ProxLB API.")
+
+    # ProxLB API Context Path Actions
+    def api_exec_reboot(self):
+        logger.debug("Rebooting system.")
+        print("Rebooting system.")
+
+    def api_exec_shutdown(self):
+        logger.debug("Shutting down system.")
+        print("Shutting down system.")
+
+    def api_exec_wol(self):
+        logger.debug("Sending WOL signal.")
+        print("Sending WOL signal.")
+
+    def api_exec_update(self):
+        logger.debug("Updating system.")
+        print("Updating system.")
+
+    # ProxLB API Uvicorn Server
+    def run(self, proxlb_config):
+        """
+        """
+        def exec_api_server():
+            """
+            """
+            # Define a custom formatter for match ProxLB logging syntax
+            log_config = {
+                "version": 1,
+                "disable_existing_loggers": False,
+                "formatters": {
+                    "default": {
+                        "()": "uvicorn.logging.DefaultFormatter",
+                        "fmt": "%(asctime)s - ProxLB API - INFO - %(message)s",
+                        "datefmt": "%Y-%m-%d %H:%M:%S,100",
+                    },
+                    "access": {
+                        "()": "uvicorn.logging.AccessFormatter",
+                        "fmt": "%(asctime)s - ProxLB API - ACCESS -  %(client_addr)s - \"%(request_line)s\" %(status_code)s",
+                        "datefmt": "%Y-%m-%d %H:%M:%S,100",
+                    },
+                },
+                "handlers": {
+                    "default": {
+                        "formatter": "default",
+                        "class": "logging.StreamHandler",
+                        "stream": "ext://sys.stdout",
+                    },
+                    "access": {
+                        "formatter": "access",
+                        "class": "logging.StreamHandler",
+                        "stream": "ext://sys.stdout",
+                    },
+                },
+                "loggers": {
+                    "uvicorn": {"handlers": ["default"], "level": "INFO"},
+                    "uvicorn.error": {"handlers": ["default"], "level": "INFO", "propagate": True},
+                    "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
+                },
+            }
+
+            # Run ProxLB API via Uvicorn with custom log formatter
+            uvicorn.run(self.proxlb_api, host="0.0.0.0", port=8000, log_level="info", log_config=log_config)
+
+        # Execute the Uvicorn in a threaded action to avoid blocking
+        if proxlb_config.get("api_server", {}).get("enable", False):
+            logger.debug("ProxLB API Server is enabled. Starting API server...")
+            server_thread = threading.Thread(target=exec_api_server, daemon=True)
+            server_thread.start()
+        else:
+            logger.debug("ProxLB API Server is not enabled.")

proxlb/utils/helper.py

@@ -8,10 +8,15 @@ __copyright__ = "Copyright (C) 2025 Florian Paul Azim Hoberg (@gyptazy)"
 __license__ = "GPL-3.0"
 
 
+import ipaddress
 import json
+import urllib.error
+import urllib.request
 import uuid
+import socket
 import sys
 import time
+import typing
 import utils.version
 from utils.logger import SystemdLogger
 from typing import Dict, Any
@@ -42,7 +47,7 @@ class Helper:
     """
     def __init__(self):
         """
-        Initializes the general Helper clas.
+        Initializes the general Helper class.
         """
 
     @staticmethod
@@ -162,3 +167,135 @@ class Helper:
             print(json.dumps(filtered_data, indent=4))
 
         logger.debug("Finished: print_json.")
+
+    @staticmethod
+    def create_wol_magic_packet(mac_address: str) -> bytes:
+        """
+        Create a magic packet from a given MAC adddress for wake-on-lan.
+
+        A magic packet is a packet that can be used with the for wake on lan
+        protocol to wake up a computer. The packet is constructed from the
+        mac address given as a parameter.
+
+        Parameters:
+            mac_address: The mac address that should be parsed into a magic packet.
+
+        Orig-Author: Remco Haszing @remcohaszing (https://github.com/remcohaszing/pywakeonlan)
+
+        """
+        logger.debug("Starting: create_wol_magic_packet.")
+        if len(mac_address) == 17:
+            sep = mac_address[2]
+            mac_address = mac_address.replace(sep, '')
+        elif len(mac_address) == 14:
+            sep = mac_address[4]
+            mac_address = mac_address.replace(sep, '')
+        if len(mac_address) != 12:
+            raise ValueError('Incorrect MAC address format')
+        logger.debug("Finished: create_wol_magic_packet.")
+        return bytes.fromhex('F' * 12 + mac_address * 16)
+
+    @staticmethod
+    def send_wol_packet(self, mac_address: str, interface: str, address_family: typing.Optional[socket.AddressFamily] = None) -> None:
+        """
+        Sends a magic packet to a given MAC address on a given interface for wake-on-lan.
+
+        Parameters:
+            mac_address: The mac address that should be used for wake-on-lan.
+            interface: The network interface that should be used for wake-on-lan.
+
+        Returns:
+            None
+
+        Orig-Author: Remco Haszing @remcohaszing (https://github.com/remcohaszing/pywakeonlan)
+        """
+        logger.debug("Starting: send_wol_packet.")
+        packets = [self.create_wol_magic_packet(mac) for mac in macs]
+
+        with socket.socket(address_family, socket.SOCK_DGRAM) as sock:
+            if interface is not None:
+                sock.bind((interface, 0))
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+            sock.connect(("255.255.255.255", 9))
+            for packet in packets:
+                sock.send(packet)
+
+        logger.debug("Finished: send_wol_packet.")
+
+    @staticmethod
+    def get_local_hostname() -> str:
+        """
+        Retruns the local hostname of the executing system.
+
+        Parameters:
+            None
+
+        Returns:
+            str: The local hostname of the executing system.
+        """
+        logger.debug("Starting: get_local_hostname.")
+        hostname = socket.gethostname()
+        logger.debug("Systems local hostname is: {hostname}")
+        logger.debug("Finished: get_local_hostname.")
+        return hostname
+
+    @staticmethod
+    def http_client_get(uri: str) -> str:
+        """
+        Receives the content of a GET request from a given URI.
+
+        Parameters:
+            uri (str): The URI to get the content from.
+
+        Returns:
+            str: The response content.
+        """
+        logger.debug("Starting: http_client_get.")
+        http_charset = "utf-8"
+        http_headers = {
+            "User-Agent": "ProxLB client/1.0"
+        }
+        http_request = urllib.request.Request(uri, headers=http_headers, method="GET")
+
+        try:
+            logger.debug("Get http client information from {uri}/{path}.")
+            with urllib.request.urlopen(http_request) as response:
+                http_client_content = response.read().decode(http_charset)
+                return http_client_content
+        except urllib.error.HTTPError as e:
+            print(f"HTTP error: {e.code} - {e.reason}")
+        except urllib.error.URLError as e:
+            print(f"URL error: {e.reason}")
+        logger.debug("Finished: http_client_get.")
+
+    @staticmethod
+    def http_client_post(uri: str, data: Dict[str, Any]) -> str:
+        """
+        Sends a POST request with JSON data to the given URI.
+
+        Parameters:
+            uri (str): The URI to send the POST request to.
+            data (dict): The data to send in the request body.
+
+        Returns:
+            str: The response content.
+        """
+        logger.debug("Starting: http_client_post.")
+        http_charset = "utf-8"
+        http_json_data = json.dumps(data).encode(http_charset)
+        http_headers = {
+            "User-Agent": "ProxLB client/1.0",
+            "Content-Type": "application/json"
+        }
+        http_request = urllib.request.Request(uri, data=http_json_data, headers=http_headers, method="POST")
+
+        try:
+            logger.debug(f"Sending HTTP client information to {uri}.")
+            with urllib.request.urlopen(http_request) as response:
+                http_client_content = response.read().decode(http_charset)
+                return http_client_content
+        except urllib.error.HTTPError as e:
+            print(f"HTTP error: {e.code} - {e.reason}")
+        except urllib.error.URLError as e:
+            print(f"URL error: {e.reason}")
+        logger.debug("Finished: http_client_post.")

setup.py

@@ -16,6 +16,8 @@ setup(
         "urllib3",
         "proxmoxer",
         "pyyaml",
+        "uvicorn",
+        "fastapi",
     ],
         data_files=[('/etc/systemd/system', ['service/proxlb.service']), ('/etc/proxlb/', ['config/proxlb_example.yaml'])],
 )