From ea91b1db43362fefaa845066d5bd4639bf7b29e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9mence=20Lesn=C3=A9?= Date: Mon, 25 Sep 2023 18:06:48 +0200 Subject: [PATCH 1/3] Feat: Load app conf by env var - Validate config parameters against a type, to avoid miss-configurations - Search recursively the config file in top folders - Variables passed by env are encoder in JSON - Config file values are substituted with env on the fly, allowing to source secrets from Azure Key Vault, Kubernetes Secrets, etc --- .vscode/launch.json | 9 +- app/helpers/config.py | 243 ++++++++++-------- app/helpers/dicts.py | 107 -------- .../LogUsage/LogUsageToLogAnalytics.py | 6 +- app/plugins/base.py | 41 +-- app/powerproxy.py | 82 +++--- config/config.example.yaml | 4 + 7 files changed, 202 insertions(+), 290 deletions(-) delete mode 100644 app/helpers/dicts.py diff --git a/.vscode/launch.json b/.vscode/launch.json index 3f7772b..81de172 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -5,14 +5,17 @@ "version": "0.2.0", "configurations": [ { - "name": "Debug powerproxy.py", + "name": "Debug with \"local\" config", "type": "python", "request": "launch", "cwd": "${workspaceFolder}/app", "program": "powerproxy.py", - "args": ["--config-file", "../config/config.local.yaml"], + "env": { + "POWERPROXY_CONFIG_FILE": "config.local.yaml", + "POWERPROXY_CONFIG_PATH": "${workspaceFolder}/config" + }, "console": "integratedTerminal", "justMyCode": true } ] -} \ No newline at end of file +} diff --git a/app/helpers/config.py b/app/helpers/config.py index 78d61fb..cbacd23 100644 --- a/app/helpers/config.py +++ b/app/helpers/config.py @@ -1,116 +1,141 @@ """Several methods and classes around configuration.""" +from enum import Enum +from pathlib import Path +from typing import Any, Dict, List, Optional, Type, TypeVar, Union +from uuid import UUID import json import os - import yaml -from plugins.base import PowerProxyPlugin, foreach_plugin - -# pylint: disable=relative-beyond-top-level -from .dicts import QueryDict - -# pylint: enable=relative-beyond-top-level - - -class Configuration: - """Configuration class.""" - - def __init__(self, values_dict): - """Constructor.""" - self.values_dict = QueryDict(values_dict) - self.clients = [client["name"] for client in self.get("clients")] - self.key_client_map = {client["key"]: client["name"] for client in self.get("clients")} - self.plugin_names = [plugin["name"] for plugin in self.get("plugins")] - - # instantiate plugins - self.plugins = [ - PowerProxyPlugin.get_plugin_instance( - plugin_config["name"], self, QueryDict(plugin_config) - ) - for plugin_config in self.get("plugins") - ] - foreach_plugin(self.plugins, "on_plugin_instantiated") - - def __getitem__(self, key): - """Dunder method to get config value via ["..."] syntax.""" - return self.values_dict[key] - - def get(self, path, default=None): - """Return value under given path.""" - return self.values_dict.get(path, default) - - def print(self): - """Print the current configuration.""" - Configuration.print_setting("Clients identified by API Key", ", ".join(self.clients)) - Configuration.print_setting( - "Fixed client overwrite", - f"{self['fixed_client'] if self['fixed_client'] else '(not set)'}", +import re + + +T = TypeVar("T", bool, int, float, UUID, str, Enum, list, dict, None) +TEST_SUBSTITUTION = re.compile(r"({(.*)})") +CACHE: Dict[str, T] = {} + + +class ConfigNotFound(Exception): + pass + + +def get_config( + key: str, + validate: Type[T], + default: Any = None, + required: bool = False, + sections: Optional[Union[str, List[str]]] = None, +) -> T: + """ + Get config from environment variable or config file. + """ + cache_key = ".".join( + [] + + ([] if not sections else sections if isinstance(sections, list) else [sections]) + + [key] + ) + + if cache_key in CACHE: + return CACHE[cache_key] + + # Get config from file + res = None + if sections: + if isinstance(sections, list): + res = CONFIG + for section in sections: + res = res.get(section, {}) + else: + res = CONFIG.get(sections, {}) + res = res.get(key, default) + else: + res = CONFIG.get(key, default) + + # Check if required + if required and not res: + raise ConfigNotFound(f'Cannot find config "{sections}/{key}"') + + # Convert to res_type + try: + if validate is str: # str + res = str(res) + elif validate is bool: # bool + res = bool(res) + elif validate is int: # int + res = int(res) + elif validate is float: # float + res = float(res) + elif validate is UUID: # UUID + res = UUID(res) + elif validate is list: # list + res = list(res) + elif validate is dict: # dict + res = dict(res) + elif issubclass(validate, Enum): # Enum + res = validate(res) + except (ValueError, TypeError, AttributeError): + raise ConfigNotFound( + f'Cannot convert config "{sections}/{key}" ({validate}), found "{res}" ({type(res)})' + ) + + # Check res type + if not isinstance(res, validate): + raise ConfigNotFound( + f'Cannot validate config "{sections}/{key}" ({validate}), found "{res}" ({type(res)})' ) - Configuration.print_setting("Plugins enabled", ", ".join(self.plugin_names)) - Configuration.print_setting("Azure OpenAI endpoint (backend)", self["aoai/endpoint"]) - - @staticmethod - def print_setting(name, value): - """Print the given setting name and value.""" - print(f"{name.ljust(32)}: {value}") - - @staticmethod - def from_file(file_path): - """Load configuration from file.""" - with open(file_path, "r", encoding="utf-8") as file: - return Configuration(yaml.safe_load(file)) - - @staticmethod - def from_json_string(json_string): - """Load configuration from JSON string.""" + + res = _substitute(res) + CACHE[cache_key] = res + return res + + +def _substitute(element: Union[str, List, Dict]) -> Union[str, List, Dict]: + """ + Replace chars surrounded by double quotes, like "{xxxx}", by the related env. + """ + if isinstance(element, str): + for substitution in re.findall(TEST_SUBSTITUTION, element): + env = os.environ.get(substitution[1]) + if env: + element = element.replace(substitution[0], env) + elif isinstance(element, list): + for i, v in enumerate(element): + element[i] = _substitute(v) + elif isinstance(element, dict): + for k, v in element.items(): + element.update({_substitute(k): _substitute(v)}) + return element + + +CONFIG: Dict[str, Any] = {} +CONFIG_ENV = os.environ.get("POWERPROXY_CONFIG_JSON") + +if not CONFIG_ENV: + print('No JSON config defined from "POWERPROXY_CONFIG_JSON", pass') +else: + try: + CONFIG = json.loads(CONFIG_ENV) + print("JSON config is loaded from env") + except json.JSONDecodeError as e: + print("Failed to load JSON config from env") + print(e) + +if not CONFIG: + CONFIG_FILE = os.environ.get("POWERPROXY_CONFIG_FILE", "config.yaml") + CONFIG_FOLDER = Path(os.environ.get("POWERPROXY_CONFIG_PATH", ".")).absolute() + CONFIG_PATH: Union[str, None] = None + while CONFIG_FOLDER: + CONFIG_PATH = f"{CONFIG_FOLDER}/{CONFIG_FILE}" + print(f'Try to load config from "{CONFIG_PATH}"') try: - return Configuration(json.loads(json_string)) - except ValueError: - # pylint: disable=raise-missing-from - raise ValueError( - (f"The provided config string '{json_string}' is not a valid JSON document.") - ) - # pylint: enable=raise-missing-from - - @staticmethod - def from_env_var(env_var_name="POWERPROXY_CONFIG_STRING", skip_no_env_var_exception=False): - """Load configuration from environment variable.""" - if env_var_name in os.environ: - return Configuration.from_json_string(os.environ[env_var_name]) - if not skip_no_env_var_exception: - raise ValueError( - f"Cannot load configuration from environment variable '{env_var_name}' because it " - f"does not exist." - ) - - @staticmethod - def from_args(args): - """Load configuration from script arguments.""" - result = None - if args.config_file: - result = Configuration.from_file(args.config_file) - elif args.config_env_var and args.config_env_var in os.environ: - result = Configuration.from_env_var(args.config_env_var) - elif args.config_env_var and args.config_env_var not in os.environ: - raise ValueError( - ( - f"The specified environment variable '{args.config_env_var}', which shall " - "contain the configuration for PowerProxy, does not exist." - ) - ) - elif args.config_string: - result = Configuration.from_json_string(args.config_string) - elif "POWERPROXY_CONFIG_STRING" in os.environ: - result = Configuration.from_env_var( - "POWERPROXY_CONFIG_STRING", skip_no_env_var_exception=True - ) - else: - raise ValueError( - ( - "No configuration provided. Ensure that you pass in a valid configuration " - "either by using argument '--config-file', '-config-env-var', or " - "'--config-string' or provide a valid config string in env variable named " - "'POWERPROXY_CONFIG_STRING' (in single-line JSON format)." - ) - ) - return result + with open(CONFIG_PATH, "rb") as file: + CONFIG = yaml.safe_load(file) + break + except FileNotFoundError: + if CONFIG_FOLDER.parent == CONFIG_FOLDER: + raise ConfigNotFound("Cannot find config file") + CONFIG_FOLDER = CONFIG_FOLDER.parent.parent + except Exception as e: + print(f'Cannot load config file "{CONFIG_PATH}"') + raise e + print(f'Config "{CONFIG_PATH}" loaded') diff --git a/app/helpers/dicts.py b/app/helpers/dicts.py deleted file mode 100644 index fe33ba8..0000000 --- a/app/helpers/dicts.py +++ /dev/null @@ -1,107 +0,0 @@ -"""Helper functions for working with dicts.""" - -import re - -class QueryDict(dict): - """Helper class to query and update a dict more conveniently.""" - - def __getitem__(self, key): - """Dunder method to get value at by ["..."] syntax.""" - return self.get(key) - - def get(self, path, default=None, separator="/", escape_sequence="''"): - """ - Return the value at the given path. - - If the path cannot be found, the default value is returned. - - Keys in the path can be surrounded by the given escape sequence to treat separators as - regular chars instead of separators. For instance, "abc/''def/ghi''" will return the value - at abc -> def/ghi instead of abc -> def -> ghi (assuming default values for parameters - 'separator' and 'escape_sequence' are used). - - Optionally, a path can start with a separator ("/" by default). If path is only the - separator, the entire dict is returned. - """ - if not path: - return default - - if path == separator: - return dict(self) - - keys_from_path = QueryDict._get_keys_from_path(path, separator, escape_sequence) - - path_length = len(keys_from_path) - for i, key_from_path in enumerate(keys_from_path): - if i == 0: - parent_element = dict.get(self, key_from_path, default) - continue - if i > 0 and i < path_length: - try: - parent_element = parent_element[key_from_path] - continue - # pylint: disable=bare-except - except: - return default - # pylint: enable=bare-except - return parent_element - - def set(self, path, value, separator="/", escape_sequence="''"): - """ - Set the given value at the given path. - - Path items which do not exist yet, will be added. - - Keys in the path can be surrounded by the given escape sequence to treat separators as - regular chars instead of separators. For instance, "abc/''def/ghi''" will set the value at - abc -> def/ghi instead of abc -> def -> ghi (assuming default values for parameters - 'separator' and 'escape_sequence' are used). - - Optionally, a path can start with a separator ("/" by default). If path is only the - separator, the entire dict - is set. - """ - keys_from_path = QueryDict._get_keys_from_path(path, separator, escape_sequence) - parent_element = self - is_last_element_in_path = False - for i, key_from_path in enumerate(keys_from_path): - is_last_element_in_path = i == len(keys_from_path) - 1 - if not key_from_path in parent_element: - parent_element[key_from_path] = {} - if not is_last_element_in_path and not isinstance(parent_element[key_from_path], dict): - raise ValueError( - ( - "Cannot set value. All items on the way to the last element must be of " - "type dict to avoid that data is unintentionally overwritten." - ) - ) - if not is_last_element_in_path: - parent_element = parent_element[key_from_path] - else: - parent_element[key_from_path] = value - - @staticmethod - def get_last_item_from_path(path, separator="/", escape_sequence="''"): - """Return the last item from the given path.""" - return QueryDict._get_keys_from_path(path, separator, escape_sequence)[-1] - - @staticmethod - def _get_keys_from_path(path, separator, escape_sequence): - """Get the different keys from the given path.""" - if path.startswith("/"): - path = path[1:] - - escaped_escape_sequence = re.escape(escape_sequence) - keys_from_path = [ - re.sub( - rf"^{escaped_escape_sequence}|{escaped_escape_sequence}$", - "", - element.replace(chr(0), separator), - ) - for element in re.sub( - rf"{escaped_escape_sequence}.*?{escaped_escape_sequence}", - lambda match: match.group().replace(separator, chr(0)), - path, - ).split(separator) - ] - return keys_from_path diff --git a/app/plugins/LogUsage/LogUsageToLogAnalytics.py b/app/plugins/LogUsage/LogUsageToLogAnalytics.py index ef2851a..f9730f5 100644 --- a/app/plugins/LogUsage/LogUsageToLogAnalytics.py +++ b/app/plugins/LogUsage/LogUsageToLogAnalytics.py @@ -4,8 +4,8 @@ from azure.identity import ChainedTokenCredential, ClientSecretCredential, ManagedIdentityCredential from azure.monitor.ingestion import LogsIngestionClient -from helpers.dicts import QueryDict from plugins.LogUsage.LogUsageBase import LogUsageBase +from typing import Dict, Any class LogUsageToLogAnalytics(LogUsageBase): @@ -20,9 +20,9 @@ class LogUsageToLogAnalytics(LogUsageBase): log_analytics_client = None - def __init__(self, app_configuration, plugin_configuration: QueryDict): + def __init__(self, plugin_configuration: Dict[str, Any]): """Constructor.""" - super().__init__(app_configuration, plugin_configuration) + super().__init__(plugin_configuration) self.log_ingestion_endpoint = plugin_configuration.get("log_ingestion_endpoint") self.credential_tenant_id = plugin_configuration.get("credential_tenant_id") diff --git a/app/plugins/base.py b/app/plugins/base.py index bfecdb6..c4b54fe 100644 --- a/app/plugins/base.py +++ b/app/plugins/base.py @@ -1,15 +1,31 @@ """Defines the foundation for PowerProxy plugins.""" +from helpers.config import get_config +from typing import Dict, Any, List import importlib import re +def foreach_plugin(method_name, *args): + """ + Have each plugin run the method with the given name and arguments. + """ + for plugin in PLUGINS: + if hasattr(plugin, method_name): + getattr(plugin, method_name)(*args) + else: + raise ValueError( + ( + f"Plugin class '{plugin.__class__()}' does not have a method named '{method_name}'." + ) + ) + + class PowerProxyPlugin: """A plugin for PowerProxy, doing different things at different events.""" - def __init__(self, app_configuration, plugin_configuration): + def __init__(self, plugin_configuration: Dict[str, Any]): """Constructor.""" - self.app_configuration = app_configuration self.plugin_configuration = plugin_configuration def on_plugin_instantiated(self): @@ -42,21 +58,14 @@ def get_plugin_class(plugin_name): ) @staticmethod - def get_plugin_instance(plugin_name, app_configuration, plugin_configuration): + def get_plugin_instance(plugin_name, plugin_configuration): """Return an instance of the plugin with the given name.""" plugin_class = PowerProxyPlugin.get_plugin_class(plugin_name) - return plugin_class(app_configuration, plugin_configuration) + return plugin_class(plugin_configuration) -def foreach_plugin(plugins, method_name, *args): - """Have each plugin run the method with the given name and arguments.""" - for plugin in plugins: - if hasattr(plugin, method_name): - getattr(plugin, method_name)(*args) - else: - raise ValueError( - ( - f"Plugin class '{plugin.__class__()}' does not have a method named " - f"'{method_name}'." - ) - ) +PLUGINS: List[PowerProxyPlugin] = [ + PowerProxyPlugin.get_plugin_instance(plugin["name"], plugin) + for plugin in get_config("plugins", validate=list, required=True) +] +foreach_plugin("on_plugin_instantiated") diff --git a/app/powerproxy.py b/app/powerproxy.py index 8dcbb52..b379ea1 100644 --- a/app/powerproxy.py +++ b/app/powerproxy.py @@ -7,7 +7,6 @@ # pylint: disable=invalid-name, import-error -import argparse import io import json @@ -15,47 +14,16 @@ import uvicorn from fastapi import FastAPI, Request, status from fastapi.responses import Response, StreamingResponse -from helpers.config import Configuration +from helpers.config import get_config from helpers.header import print_header from plugins.base import foreach_plugin from version import VERSION -## define script arguments -parser = argparse.ArgumentParser() -# --config-file -parser.add_argument( - "--config-file", - type=str, - help="Path to config file", -) -# --config-env-var -parser.add_argument( - "--config-env-var", - type=str, - help="Name of the environment variable containing the configuration as JSON string.", -) -# --config-string -parser.add_argument( - "--config-string", - type=str, - help="String containing the configuration as JSON string.", -) -# --port -parser.add_argument( - "--port", - type=int, - default=80, - help=( - "Port where the proxy runs. Ports <= 1024 may need special permissions in Linux. " - "Default: 80." - ), -) -args, unknown = parser.parse_known_args() -## load configuration -config = Configuration.from_args(args) +# misc +PORT = get_config('port', validate=int, default=80) -## define and run proxy app +# define and run proxy app app = FastAPI() @@ -65,11 +33,12 @@ async def startup_event(): """Invoked when the app is started.""" # print header and config values print_header(f"PowerProxy for Azure OpenAI - v{VERSION}") - Configuration.print_setting("Proxy port", args.port) - config.print() + print(f"Proxy port: {PORT}") # instantiate HTTP client for AOAI endpoint - app.state.target_client: httpx.AsyncClient = httpx.AsyncClient(base_url=config["aoai/endpoint"]) + app.state.target_client: httpx.AsyncClient = httpx.AsyncClient( + base_url=get_config("endpoint", sections="aoai", validate=str, required=True) + ) # print serve notification print() @@ -108,7 +77,7 @@ async def handle_request(request: Request, path: str): "incoming_request": request, "incoming_request_body": await request.body(), } - foreach_plugin(config.plugins, "on_new_request_received", routing_slip) + foreach_plugin("on_new_request_received", routing_slip) # identify client and replace API key if needed # notes: - When API authentication is used, we get an API key in header 'api-key'. This would @@ -126,21 +95,27 @@ async def handle_request(request: Request, path: str): for key in set(request.headers.keys()) - {"Host", "host", "Content-Length", "content-length"} } - client = None - if config["FIXED_CLIENT"]: - client = config["FIXED_CLIENT"] + + fixed_client = get_config("FIXED_CLIENT", validate=str) + client = fixed_client if fixed_client else None + if "api-key" in headers: - if headers["api-key"] not in config.key_client_map: + client_map = dict( + (client.get("key"), client.get("name")) + for client in get_config("clients", validate=list, required=True) + ) + if headers["api-key"] not in client_map: raise ValueError( ( "The provided API key is not a valid PowerProxy key. Ensure that the 'api-key' " "header contains valid API key from the PowerProxy's configuration." ) ) - client = config.key_client_map[headers["api-key"]] if client is None else client - headers["api-key"] = config["aoai/key"] + client = client_map[headers["api-key"]] if client is None else client + headers["api-key"] = get_config("key", sections="aoai", validate=str, required=True) + routing_slip["client"] = client - foreach_plugin(config.plugins, "on_client_identified", routing_slip) + foreach_plugin("on_client_identified", routing_slip) # forward request to target endpoint and get response aoai_response: httpx.Response = await app.state.target_client.request( @@ -150,8 +125,9 @@ async def handle_request(request: Request, path: str): headers=headers, content=routing_slip["incoming_request_body"], ) + routing_slip["headers_from_target"] = aoai_response.headers - foreach_plugin(config.plugins, "on_headers_from_target_received", routing_slip) + foreach_plugin("on_headers_from_target_received", routing_slip) # determine if it's an event stream or not routing_slip["is_event_stream"] = ( @@ -170,7 +146,7 @@ async def handle_request(request: Request, path: str): body = await aoai_response.aread() try: routing_slip["body_dict_from_target"] = json.load(io.BytesIO(body)) - foreach_plugin(config.plugins, "on_body_dict_from_target_available", routing_slip) + foreach_plugin("on_body_dict_from_target_available", routing_slip) # pylint: disable=bare-except except: # eat any exception in case the response cannot be parsed @@ -181,6 +157,7 @@ async def handle_request(request: Request, path: str): status_code=aoai_response.status_code, headers=routing_slip["response_headers_from_target"], ) + case True: # event stream # forward and process events as they come in @@ -195,10 +172,11 @@ async def yield_data_events(): if data != "[DONE]": routing_slip["data_from_target"] = data foreach_plugin( - config.plugins, "on_data_event_from_target_received", routing_slip + "on_data_event_from_target_received", routing_slip ) + foreach_plugin( - config.plugins, "on_end_of_target_response_stream_reached", routing_slip + "on_end_of_target_response_stream_reached", routing_slip ) return StreamingResponse( @@ -215,7 +193,7 @@ async def yield_data_events(): uvicorn.run( app, host="0.0.0.0", - port=int(args.port), + port=PORT, log_level="warning", server_header=False, date_header=False, diff --git a/config/config.example.yaml b/config/config.example.yaml index a3db92a..775b622 100644 --- a/config/config.example.yaml +++ b/config/config.example.yaml @@ -1,3 +1,7 @@ +# Port where the proxy runs. Ports <= 1024 may need special permissions in Linux. +# default: 80 +port: ___ + # the teams or use cases accessing Azure OpenAI (aka. "clients") and their keys # notes: - each client must have a unique key. PowerProxy identifies clients by the given key. # - should be a single line, even for defining multiple clients From 66a923dec82557bbda296f2b6b221e5b1eb43c2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9mence=20Lesn=C3=A9?= Date: Mon, 25 Sep 2023 18:15:29 +0200 Subject: [PATCH 2/3] Feat: Customize logs levels Dev: Facilitate logging understanding by allowing the developer to add trace/debug logs, not shown into PROD envs. --- app/helpers/header.py | 7 ------- app/helpers/logger.py | 21 +++++++++++++++++++ app/plugins/LogUsage/LogUsageToConsole.py | 6 +++++- .../LogUsage/LogUsageToLogAnalytics.py | 15 +++++++------ app/powerproxy.py | 14 ++++++------- config/config.example.yaml | 8 +++++++ 6 files changed, 50 insertions(+), 21 deletions(-) delete mode 100644 app/helpers/header.py create mode 100644 app/helpers/logger.py diff --git a/app/helpers/header.py b/app/helpers/header.py deleted file mode 100644 index c8fe112..0000000 --- a/app/helpers/header.py +++ /dev/null @@ -1,7 +0,0 @@ -"""Several methods around printing headers.""" - -def print_header(header_string): - """Print the given string as header.""" - print(len(header_string) * "-") - print(header_string) - print(len(header_string) * "-") diff --git a/app/helpers/logger.py b/app/helpers/logger.py new file mode 100644 index 0000000..a9ebbe9 --- /dev/null +++ b/app/helpers/logger.py @@ -0,0 +1,21 @@ +from helpers.config import get_config +from typing import List +import logging + + +LOGGERS: List[logging.Logger] = [] + +LOGGING_SYS_LEVEL = get_config( + "sys_level", sections=["monitoring", "logging"], validate=str, default="WARN" +) +logging.basicConfig(level=LOGGING_SYS_LEVEL) +LOGGING_APP_LEVEL = get_config( + "app_level", sections=["monitoring", "logging"], validate=str, default="INFO" +) + + +def build_logger(name: str) -> logging.Logger: + logger = logging.getLogger(name) + logger.setLevel("DEBUG") + LOGGERS.append(logger) + return logger diff --git a/app/plugins/LogUsage/LogUsageToConsole.py b/app/plugins/LogUsage/LogUsageToConsole.py index 29cdda8..e13b402 100644 --- a/app/plugins/LogUsage/LogUsageToConsole.py +++ b/app/plugins/LogUsage/LogUsageToConsole.py @@ -2,9 +2,13 @@ # pylint: disable=invalid-name,too-many-arguments,import-error,no-name-in-module,too-few-public-methods +from helpers.logger import build_logger from plugins.LogUsage.LogUsageBase import LogUsageBase +_logger = build_logger(__name__) + + class LogUsageToConsole(LogUsageBase): """Logs Azure OpenAI usage info to console.""" @@ -21,7 +25,7 @@ def _append_line( openai_region, ): """Append a new line with the given infos.""" - print( + _logger.info( "---\n" f"Request start minute : {request_start_minute}\n" f"Request start minute UTC : {request_start_minute_utc}\n" diff --git a/app/plugins/LogUsage/LogUsageToLogAnalytics.py b/app/plugins/LogUsage/LogUsageToLogAnalytics.py index f9730f5..d836003 100644 --- a/app/plugins/LogUsage/LogUsageToLogAnalytics.py +++ b/app/plugins/LogUsage/LogUsageToLogAnalytics.py @@ -6,6 +6,10 @@ from azure.monitor.ingestion import LogsIngestionClient from plugins.LogUsage.LogUsageBase import LogUsageBase from typing import Dict, Any +from helpers.logger import build_logger + + +_logger = build_logger(__name__) class LogUsageToLogAnalytics(LogUsageBase): @@ -31,12 +35,11 @@ def __init__(self, plugin_configuration: Dict[str, Any]): self.data_collection_rule_id = plugin_configuration.get("data_collection_rule_id") self.stream_name = plugin_configuration.get("stream_name") - print() - print(f"Log ingestion endpoint : {self.log_ingestion_endpoint}") - print(f"Credential Tenant ID : {self.credential_tenant_id}") - print(f"Credential Client ID : {self.credential_client_id}") - print(f"Data Collection Rule ID : {self.data_collection_rule_id}") - print(f"Stream Name : {self.stream_name}") + _logger.info(f"Log ingestion endpoint : {self.log_ingestion_endpoint}") + _logger.info(f"Credential Tenant ID : {self.credential_tenant_id}") + _logger.info(f"Credential Client ID : {self.credential_client_id}") + _logger.info(f"Data Collection Rule ID : {self.data_collection_rule_id}") + _logger.info(f"Stream Name : {self.stream_name}") def on_plugin_instantiated(self): """Run directly after the new plugin instance has been instantiated.""" diff --git a/app/powerproxy.py b/app/powerproxy.py index b379ea1..29f1b4a 100644 --- a/app/powerproxy.py +++ b/app/powerproxy.py @@ -15,12 +15,12 @@ from fastapi import FastAPI, Request, status from fastapi.responses import Response, StreamingResponse from helpers.config import get_config -from helpers.header import print_header +from helpers.logger import build_logger from plugins.base import foreach_plugin from version import VERSION - # misc +_logger = build_logger(__name__) PORT = get_config('port', validate=int, default=80) # define and run proxy app @@ -32,8 +32,8 @@ async def startup_event(): """Invoked when the app is started.""" # print header and config values - print_header(f"PowerProxy for Azure OpenAI - v{VERSION}") - print(f"Proxy port: {PORT}") + _logger.info(f"PowerProxy for Azure OpenAI - v{VERSION}") + _logger.debug(f"Proxy port: {PORT}") # instantiate HTTP client for AOAI endpoint app.state.target_client: httpx.AsyncClient = httpx.AsyncClient( @@ -41,9 +41,7 @@ async def startup_event(): ) # print serve notification - print() - print("Serving incoming requests...") - print() + _logger.info("Serving incoming requests...") # app shutdown @@ -114,10 +112,12 @@ async def handle_request(request: Request, path: str): client = client_map[headers["api-key"]] if client is None else client headers["api-key"] = get_config("key", sections="aoai", validate=str, required=True) + _logger.debug(f"Identified client: {client}") routing_slip["client"] = client foreach_plugin("on_client_identified", routing_slip) # forward request to target endpoint and get response + _logger.debug(f"Forwarded headers: {headers}") aoai_response: httpx.Response = await app.state.target_client.request( request.method, path, diff --git a/config/config.example.yaml b/config/config.example.yaml index 775b622..dad60ed 100644 --- a/config/config.example.yaml +++ b/config/config.example.yaml @@ -41,6 +41,14 @@ aoai: # note: is not required when Azure AD is used to auth against Azure OpenAI key: ___ +monitoring: + # Technical logging + logging: + # Enum: "NOSET", "DEBUG", "INFO", "WARN", "ERROR", "FATAL", "CRITICAL" + app_level: INFO + # Enum: "NOSET", "DEBUG", "INFO", "WARN", "ERROR", "FATAL", "CRITICAL" + sys_level: ERROR + # region to which the proxy shall be deployed to Azure # example: westeurope region: westeurope From 6b7d42c02675bccffe9cbf1ba810330f97da4086 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9mence=20Lesn=C3=A9?= Date: Fri, 22 Sep 2023 23:53:49 +0200 Subject: [PATCH 3/3] Feat: Monitor the technical behaviour with Application Insights --- app/helpers/app_insights.py | 54 +++++++++++++++++++++++++++++++++++++ app/helpers/logger.py | 7 +++++ app/powerproxy.py | 11 ++++++++ config/config.example.yaml | 6 +++++ requirements.txt | 6 ++++- 5 files changed, 83 insertions(+), 1 deletion(-) create mode 100644 app/helpers/app_insights.py diff --git a/app/helpers/app_insights.py b/app/helpers/app_insights.py new file mode 100644 index 0000000..a27816a --- /dev/null +++ b/app/helpers/app_insights.py @@ -0,0 +1,54 @@ +from helpers.config import get_config +from helpers.logger import enable_app_insights +from helpers.logger import build_logger + + +_logger = build_logger(__name__) +ENABLED = get_config( + "enabled", sections=["monitoring", "app_insights"], validate=bool, required=True +) + + +def init() -> None: + if ENABLED: + _setup() + enable_app_insights() + _logger.info("App Insights enabled") + + +def _setup() -> None: + """Setup OpenTelemetry for App Insights.""" + from azure.identity import DefaultAzureCredential + from opentelemetry._logs import get_logger_provider, set_logger_provider + from azure.monitor.opentelemetry.exporter import ( + AzureMonitorLogExporter, + AzureMonitorTraceExporter, + ) + from opentelemetry.sdk._logs.export import BatchLogRecordProcessor + from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor + from opentelemetry.sdk._logs import LoggerProvider + + connection_str = get_config( + "connection_str", sections=["monitoring", "app_insights"], validate=str, required=True + ) + credential = DefaultAzureCredential() + + # Logs + _logger.debug("Setting up logs exporter for App Insights") + set_logger_provider(LoggerProvider()) + log_exporter = AzureMonitorLogExporter(connection_string=connection_str, credential=credential) + get_logger_provider().add_log_record_processor(BatchLogRecordProcessor(log_exporter)) + + # Traces + # TODO: Enable sampling + _logger.debug("Setting up traces exporter for App Insights") + HTTPXClientInstrumentor().instrument() + trace.set_tracer_provider(TracerProvider()) + trace_exporter = AzureMonitorTraceExporter( + connection_string=connection_str, credential=credential + ) + span_processor = BatchSpanProcessor(trace_exporter) + trace.get_tracer_provider().add_span_processor(span_processor) diff --git a/app/helpers/logger.py b/app/helpers/logger.py index a9ebbe9..8d75b5f 100644 --- a/app/helpers/logger.py +++ b/app/helpers/logger.py @@ -1,4 +1,5 @@ from helpers.config import get_config +from opentelemetry.sdk._logs import LoggingHandler from typing import List import logging @@ -19,3 +20,9 @@ def build_logger(name: str) -> logging.Logger: logger.setLevel("DEBUG") LOGGERS.append(logger) return logger + + +def enable_app_insights() -> None: + for logger in LOGGERS: + handler = LoggingHandler() + logger.addHandler(handler) diff --git a/app/powerproxy.py b/app/powerproxy.py index 29f1b4a..df64de0 100644 --- a/app/powerproxy.py +++ b/app/powerproxy.py @@ -7,6 +7,11 @@ # pylint: disable=invalid-name, import-error +from helpers import app_insights + +# Setup tracing first +app_insights.init() + import io import json @@ -26,6 +31,12 @@ # define and run proxy app app = FastAPI() +# Instrument FastAPI with OpenTelemetry +if app_insights.ENABLED: + from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor + + FastAPIInstrumentor.instrument_app(app) + # app startup event @app.on_event("startup") diff --git a/config/config.example.yaml b/config/config.example.yaml index dad60ed..d740a97 100644 --- a/config/config.example.yaml +++ b/config/config.example.yaml @@ -48,6 +48,12 @@ monitoring: app_level: INFO # Enum: "NOSET", "DEBUG", "INFO", "WARN", "ERROR", "FATAL", "CRITICAL" sys_level: ERROR + # Azure Application Insights + app_insights: + # bool + enabled: true + # Connection string + connection_str: ___ # region to which the proxy shall be deployed to Azure # example: westeurope diff --git a/requirements.txt b/requirements.txt index fc5a9d3..4be8f53 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,8 @@ uvicorn[standard] fastapi tiktoken azure-identity -azure-monitor-ingestion \ No newline at end of file +azure-monitor-ingestion +azure-monitor-opentelemetry==1.0.0 +opentelemetry-instrumentation-fastapi==0.41b0 +opentelemetry-instrumentation-httpx==0.41b0 +opentelemetry-instrumentation-requests==0.41b0