221 lines
21 KiB
Markdown
Executable File
221 lines
21 KiB
Markdown
Executable File
# PyInfra
|
|
|
|
1. [ About ](#about)
|
|
2. [ Configuration ](#configuration)
|
|
3. [ Queue Manager ](#queue-manager)
|
|
4. [ Module Installation ](#module-installation)
|
|
5. [ Scripts ](#scripts)
|
|
6. [ Tests ](#tests)
|
|
7. [ Opentelemetry protobuf dependency hell ](#opentelemetry-protobuf-dependency-hell)
|
|
|
|
## About
|
|
|
|
Shared library for the research team, containing code related to infrastructure and communication with other services.
|
|
Offers a simple interface for processing data and sending responses via AMQP, monitoring via Prometheus and storage
|
|
access via S3 or Azure. Also export traces via OpenTelemetry for queue messages and webserver requests.
|
|
|
|
To start, see the [complete example](pyinfra/examples.py) which shows how to use all features of the service and can be
|
|
imported and used directly for default research service pipelines (data ID in message, download data from storage,
|
|
upload result while offering Prometheus monitoring, /health and /ready endpoints and multi tenancy support).
|
|
|
|
## Configuration
|
|
|
|
Configuration is done via `Dynaconf`. This means that you can use environment variables, a `.env` file or `.toml`
|
|
file(s) to configure the service. You can also combine these methods. The precedence is
|
|
`environment variables > .env > .toml`. It is recommended to load settings with the provided
|
|
[`load_settings`](pyinfra/config/loader.py) function, which you can combine with the provided
|
|
[`parse_args`](pyinfra/config/loader.py) function. This allows you to load settings from a `.toml` file or a folder with
|
|
`.toml` files and override them with environment variables.
|
|
|
|
The following table shows all necessary settings. You can find a preconfigured settings file for this service in
|
|
bitbucket. These are the complete settings, you only need all if using all features of the service as described in
|
|
the [complete example](pyinfra/examples.py).
|
|
|
|
| Environment Variable | Internal / .toml Name | Description |
|
|
| ------------------------------------------ | --------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
|
| LOGGING\_\_LEVEL | logging.level | Log level |
|
|
| DYNAMIC_TENANT_QUEUES\_\_ENABLED | dynamic_tenant_queues.enabled | Enable queues per tenant that are dynamically created mode |
|
|
| METRICS\_\_PROMETHEUS\_\_ENABLED | metrics.prometheus.enabled | Enable Prometheus metrics collection |
|
|
| METRICS\_\_PROMETHEUS\_\_PREFIX | metrics.prometheus.prefix | Prefix for Prometheus metrics (e.g. {product}-{service}) |
|
|
| WEBSERVER\_\_HOST | webserver.host | Host of the webserver (offering e.g. /prometheus, /ready and /health endpoints) |
|
|
| WEBSERVER\_\_PORT | webserver.port | Port of the webserver |
|
|
| RABBITMQ\_\_HOST | rabbitmq.host | Host of the RabbitMQ server |
|
|
| RABBITMQ\_\_PORT | rabbitmq.port | Port of the RabbitMQ server |
|
|
| RABBITMQ\_\_USERNAME | rabbitmq.username | Username for the RabbitMQ server |
|
|
| RABBITMQ\_\_PASSWORD | rabbitmq.password | Password for the RabbitMQ server |
|
|
| RABBITMQ\_\_HEARTBEAT | rabbitmq.heartbeat | Heartbeat for the RabbitMQ server |
|
|
| RABBITMQ\_\_CONNECTION_SLEEP | rabbitmq.connection_sleep | Sleep time intervals during message processing. Has to be a divider of heartbeat, and shouldn't be too big, since only in these intervals queue interactions happen (like receiving new messages) This is also the minimum time the service needs to process a message. |
|
|
| RABBITMQ\_\_INPUT_QUEUE | rabbitmq.input_queue | Name of the input queue in single queue setting |
|
|
| RABBITMQ\_\_OUTPUT_QUEUE | rabbitmq.output_queue | Name of the output queue in single queue setting |
|
|
| RABBITMQ\_\_DEAD_LETTER_QUEUE | rabbitmq.dead_letter_queue | Name of the dead letter queue in single queue setting |
|
|
| RABBITMQ\_\_TENANT_EVENT_QUEUE_SUFFIX | rabbitmq.tenant_event_queue_suffix | Suffix for the tenant event queue in multi tenant/queue setting |
|
|
| RABBITMQ\_\_TENANT_EVENT_DLQ_SUFFIX | rabbitmq.tenant_event_dlq_suffix | Suffix for the dead letter queue in multi tenant/queue setting |
|
|
| RABBITMQ\_\_TENANT_EXCHANGE_NAME | rabbitmq.tenant_exchange_name | Name of tenant exchange in multi tenant/queue setting |
|
|
| RABBITMQ\_\_QUEUE_EXPIRATION_TIME | rabbitmq.queue_expiration_time | Time until queue expiration in multi tenant/queue setting |
|
|
| RABBITMQ\_\_SERVICE_REQUEST_QUEUE_PREFIX | rabbitmq.service_request_queue_prefix | Service request queue prefix in multi tenant/queue setting |
|
|
| RABBITMQ\_\_SERVICE_REQUEST_EXCHANGE_NAME | rabbitmq.service_request_exchange_name | Service request exchange name in multi tenant/queue setting |
|
|
| RABBITMQ\_\_SERVICE_RESPONSE_EXCHANGE_NAME | rabbitmq.service_response_exchange_name | Service response exchange name in multi tenant/queue setting |
|
|
| RABBITMQ\_\_SERVICE_DLQ_NAME | rabbitmq.service_dlq_name | Service dead letter queue name in multi tenant/queue setting |
|
|
| STORAGE\_\_BACKEND | storage.backend | Storage backend to use (currently only "s3" and "azure" are supported) |
|
|
| STORAGE\_\_S3\_\_BUCKET | storage.s3.bucket | Name of the S3 bucket |
|
|
| STORAGE\_\_S3\_\_ENDPOINT | storage.s3.endpoint | Endpoint of the S3 server |
|
|
| STORAGE\_\_S3\_\_KEY | storage.s3.key | Access key for the S3 server |
|
|
| STORAGE\_\_S3\_\_SECRET | storage.s3.secret | Secret key for the S3 server |
|
|
| STORAGE\_\_S3\_\_REGION | storage.s3.region | Region of the S3 server |
|
|
| STORAGE\_\_AZURE\_\_CONTAINER | storage.azure.container_name | Name of the Azure container |
|
|
| STORAGE\_\_AZURE\_\_CONNECTION_STRING | storage.azure.connection_string | Connection string for the Azure server |
|
|
| STORAGE\_\_TENANT_SERVER\_\_PUBLIC_KEY | storage.tenant_server.public_key | Public key of the tenant server |
|
|
| STORAGE\_\_TENANT_SERVER\_\_ENDPOINT | storage.tenant_server.endpoint | Endpoint of the tenant server |
|
|
| TRACING\_\_ENABLED | tracing.enabled | Enable tracing |
|
|
| TRACING\_\_TYPE | tracing.type | Tracing mode - possible values: "opentelemetry", "azure_monitor" (Excpects APPLICATIONINSIGHTS_CONNECTION_STRING environment variable.) |
|
|
| TRACING\_\_OPENTELEMETRY\_\_ENDPOINT | tracing.opentelemetry.endpoint | Endpoint to which OpenTelemetry traces are exported |
|
|
| TRACING\_\_OPENTELEMETRY\_\_SERVICE_NAME | tracing.opentelemetry.service_name | Name of the service as displayed in the traces collected |
|
|
| TRACING\_\_OPENTELEMETRY\_\_EXPORTER | tracing.opentelemetry.exporter | Name of exporter |
|
|
| KUBERNETES\_\_POD_NAME | kubernetes.pod_name | Service pod name |
|
|
|
|
## Setup
|
|
**IMPORTANT** you need to set the following environment variables before running the setup script:
|
|
- ``$NEXUS_USER`` your Nexus user (usually equal to firstname.lastname@knecon.com)
|
|
- ``$NEXUS_PASSWORD`` your Nexus password (usually equal to your Azure Login)
|
|
|
|
```shell
|
|
# create venv and activate it
|
|
source ./scripts/setup/devenvsetup.sh {{ cookiecutter.python_version }} $NEXUS_USER $NEXUS_PASSWORD
|
|
source .venv/bin/activate
|
|
```
|
|
|
|
### OpenTelemetry
|
|
|
|
Open telemetry (vis its Python SDK) is set up to be as unobtrusive as possible; for typical use cases it can be
|
|
configured
|
|
from environment variables, without additional work in the microservice app, although additional confiuration is
|
|
possible.
|
|
|
|
`TRACING__OPENTELEMETRY__ENDPOINT` should typically be set
|
|
to `http://otel-collector-opentelemetry-collector.otel-collector:4318/v1/traces`.
|
|
|
|
## Queue Manager
|
|
|
|
The queue manager is responsible for consuming messages from the input queue, processing them and sending the response
|
|
to the output queue. The default callback also downloads data from the storage and uploads the result to the storage.
|
|
The response message does not contain the data itself, but the identifiers from the input message (including headers
|
|
beginning with "X-").
|
|
|
|
### Standalone Usage
|
|
|
|
```python
|
|
from pyinfra.queue.manager import QueueManager
|
|
from pyinfra.queue.callback import make_download_process_upload_callback, DataProcessor
|
|
from pyinfra.config.loader import load_settings
|
|
|
|
settings = load_settings("path/to/settings")
|
|
processing_function: DataProcessor # function should expect a dict (json) or bytes (pdf) as input and should return a json serializable object.
|
|
|
|
queue_manager = QueueManager(settings)
|
|
callback = make_download_process_upload_callback(processing_function, settings)
|
|
queue_manager.start_consuming(make_download_process_upload_callback(callback, settings))
|
|
```
|
|
|
|
### Usage in a Service
|
|
|
|
This is the recommended way to use the module. This includes the webserver, Prometheus metrics and health endpoints.
|
|
Custom endpoints can be added by adding a new route to the `app` object beforehand. Settings are loaded from files
|
|
specified as CLI arguments (e.g. `--settings-path path/to/settings.toml`). The values can also be set or overriden via
|
|
environment variables (e.g. `LOGGING__LEVEL=DEBUG`).
|
|
|
|
The callback can be replaced with a custom one, for example if the data to process is contained in the message itself
|
|
and not on the storage.
|
|
|
|
```python
|
|
from pyinfra.config.loader import load_settings, parse_settings_path
|
|
from pyinfra.examples import start_standard_queue_consumer
|
|
from pyinfra.queue.callback import make_download_process_upload_callback, DataProcessor
|
|
|
|
processing_function: DataProcessor
|
|
|
|
arguments = parse_settings_path()
|
|
settings = load_settings(arguments.settings_path)
|
|
|
|
callback = make_download_process_upload_callback(processing_function, settings)
|
|
start_standard_queue_consumer(callback, settings) # optionally also pass a fastAPI app object with preconfigured routes
|
|
```
|
|
|
|
### AMQP input message:
|
|
|
|
Either use the legacy format with dossierId and fileId as strings or the new format where absolute paths are used.
|
|
All headers beginning with "X-" are forwarded to the message processor, and returned in the response message (e.g.
|
|
"X-TENANT-ID" is used to acquire storage information for the tenant).
|
|
|
|
```json
|
|
{
|
|
"targetFilePath": "",
|
|
"responseFilePath": ""
|
|
}
|
|
```
|
|
|
|
or
|
|
|
|
```json
|
|
{
|
|
"dossierId": "",
|
|
"fileId": "",
|
|
"targetFileExtension": "",
|
|
"responseFileExtension": ""
|
|
}
|
|
```
|
|
|
|
## Module Installation
|
|
|
|
Add the respective version of the pyinfra package to your pyproject.toml file. Make sure to add our gitlab registry as a
|
|
source.
|
|
For now, all internal packages used by pyinfra also have to be added to the pyproject.toml file (namely kn-utils).
|
|
Execute `poetry lock` and `poetry install` to install the packages.
|
|
|
|
You can look up the latest version of the package in
|
|
the [gitlab registry](https://gitlab.knecon.com/knecon/research/pyinfra/-/packages).
|
|
For the used versions of internal dependencies, please refer to the [pyproject.toml](pyproject.toml) file.
|
|
|
|
```toml
|
|
[tool.poetry.dependencies]
|
|
pyinfra = { version = "x.x.x", source = "gitlab-research" }
|
|
kn-utils = { version = "x.x.x", source = "gitlab-research" }
|
|
|
|
[[tool.poetry.source]]
|
|
name = "gitlab-research"
|
|
url = "https://gitlab.knecon.com/api/v4/groups/19/-/packages/pypi/simple"
|
|
priority = "explicit"
|
|
```
|
|
|
|
## Scripts
|
|
|
|
### Run pyinfra locally
|
|
|
|
**Shell 1**: Start minio and rabbitmq containers
|
|
|
|
```bash
|
|
$ cd tests && docker compose up
|
|
```
|
|
|
|
**Shell 2**: Start pyinfra with callback mock
|
|
|
|
```bash
|
|
$ python scripts/start_pyinfra.py
|
|
```
|
|
|
|
**Shell 3**: Upload dummy content on storage and publish message
|
|
|
|
```bash
|
|
$ python scripts/send_request.py
|
|
```
|
|
|
|
## Tests
|
|
|
|
Tests require a running minio and rabbitmq container, meaning you have to run `docker compose up` in the tests folder
|
|
before running the tests.
|
|
|
|
## OpenTelemetry Protobuf Dependency Hell
|
|
|
|
**Note**: Status 2025/01/09: the currently used `opentelemetry-exporter-otlp-proto-http` version `1.25.0` requires
|
|
a `protobuf` version < `5.x.x` and is not compatible with the latest protobuf version `5.27.x`. This is an [open issue](https://github.com/open-telemetry/opentelemetry-python/issues/3958) in opentelemetry, because [support for 4.25.x ends in Q2 '25](https://protobuf.dev/support/version-support/#python).
|
|
Therefore, we should keep this in mind and update the dependency once opentelemetry includes support for `protobuf 5.27.x`.
|