diff --git a/.devcontainer/docker-compose.yaml b/.devcontainer/docker-compose.yaml index f6b0ab6..4ff56a8 100644 --- a/.devcontainer/docker-compose.yaml +++ b/.devcontainer/docker-compose.yaml @@ -17,14 +17,14 @@ services: image: influxdb:1.11.8 container_name: influx_database_v1-rt ports: - - "8079:8086" + - "8030:8086" volumes: - - influxdb_data:/var/lib/influxdb + - influxdb_v1_data:/var/lib/influxdb restart: always environment: INFLUXDB_DB: submodel-db INFLUXDB_ADMIN_USER: admin - INFLUXDB_ADMIN_PASSWORD: fluid40 + INFLUXDB_ADMIN_PASSWORD: fluid40secure! INFLUXDB_USER: fluid40 INFLUXDB_USER_PASSWORD: wbEx!S!#4OHcN!5X healthcheck: @@ -34,6 +34,28 @@ services: retries: 5 start_period: 10s + influx_database_v2: + image: influxdb:2.7 + container_name: influx_database_v2-rt + ports: + - "8031:8086" + volumes: + - influxdb_v2_data:/var/lib/influxdb2 + restart: always + environment: + DOCKER_INFLUXDB_INIT_MODE: setup + DOCKER_INFLUXDB_INIT_USERNAME: admin + DOCKER_INFLUXDB_INIT_PASSWORD: fluid40secure! + DOCKER_INFLUXDB_INIT_ORG: fluid40-org + DOCKER_INFLUXDB_INIT_BUCKET: fluid40-bucket + DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: wbEx!S!#4OHcN!5X + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8086/health"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 10s + grafana: image: grafana/grafana:latest container_name: grafana-rt @@ -41,7 +63,7 @@ services: - "8076:3000" environment: GF_SECURITY_ADMIN_USER: admin - GF_SECURITY_ADMIN_PASSWORD: IDKFA-fluid40 + GF_SECURITY_ADMIN_PASSWORD: fluid40 depends_on: influx_database_v1: condition: service_healthy @@ -127,5 +149,6 @@ services: - RUNTIME_CONFIGURATION_FILE=./configuration/DevContainerEnv.json volumes: - influxdb_data: + influxdb_v1_data: + influxdb_v2_data: grafana_data: diff --git a/.devcontainer/grafana/provisioning/datasources/influxdb.yaml b/.devcontainer/grafana/provisioning/datasources/influxdb.yaml index 52ee473..bbb49cf 100644 --- a/.devcontainer/grafana/provisioning/datasources/influxdb.yaml +++ b/.devcontainer/grafana/provisioning/datasources/influxdb.yaml @@ -6,7 +6,7 @@ datasources: access: proxy url: http://influx_database_v1-rt:8086 database: submodel-db - user: admin + user: fluid40 jsonData: tlsAuth: false tlsAuthWithCACert: false @@ -14,7 +14,24 @@ datasources: tlsCACert: "" tlsClientCert: "" tlsClientKey: "" - password: IDDQD-fluid40 + password: wbEx!S!#4OHcN!5X isDefault: true editable: false version: 1 + + + - name: RuntimeInfluxDBv2 + uid: influxdb-v2 + type: influxdb + access: proxy + url: http://influx_database_v2-rt:8086 + + jsonData: + version: Flux + organization: fluid40-org + defaultBucket: fluid40-bucket + tlsSkipVerify: false + + + secureJsonData: + token: wbEx!S!#4OHcN!5X diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index 02526c2..d2861d7 100644 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -104,6 +104,7 @@ jobs: ### 🧩 Build Information - Docker image: `engineeringmethodsag/fluid-runtime:${{ steps.version.outputs.version }}-${{ steps.version.outputs.hash }}` - Latest tag: `engineeringmethodsag/fluid-runtime:latest` + - [Changelog](CHANGELOG.md) env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..18893df --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,7 @@ +# Changelog + +## [0.12.1] - 2026-01-23 +- ✨Feature: Ass support for InfluxDB database version 2 +- πŸš€Improvement: Improve structure of data written to database +- πŸš€Improvement: Standardization of configuration property names (see [configuration](docs/configuration.md)) +- πŸ“„Documentation: Re-Work documentation (WIP) diff --git a/README.md b/README.md index cfc945b..eabbaa9 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# Runtime +# Micro-Service
aas_http_client @@ -9,52 +9,18 @@ [![CI](https://github.com/fluid40/runtime/actions/workflows/docker-image.yml/badge.svg?branch=main&cache-bust=1)](https://github.com/fluid40/runtime/actions) [![License](https://img.shields.io/badge/license-MIT-orange.svg)](LICENSE) -- [Runtime](#runtime) +- [Micro-Service](#micro-service) - [🌐 Overview](#-overview) - [πŸ”— Topics](#-topics) - - [🧩 Components](#-components) - - [🏭 Provided Environments](#-provided-environments) ## 🌐 Overview +The microservice implements the use case of reading operating data from a tool, making this data available within the submodel structure, and optionally writing it to an InfluxDb database. -The runtime environment uses an AID submodel from a specific shell to configure an MQTT or OPC UA broker connection. The runtime then reads machine data from the broker at regular intervals. -If an Influx database has been connected, the data is stored there. -The runtime runs as a server component and provides REST endpoints to output the currently read machine data. ## πŸ”— Topics - πŸ› οΈ [Configuration](docs/configuration.md) -- 🏭 [Environments](docs/environments.md) +- 🏭 [Development](docs/development.md) - -## 🧩 Components - -The runtime relies on several components to operate effectively. Each plays a specific role in data management and system integration: - -| Component | Requirement | Description | Link | -| ------------------------------ | ----------- | ------------------------------------------------------------------------------------- | ------------------------------------------------------------------- | -| **AAS Server** | Required | Responsible for managing all Asset Administration Shells (AAS) and their submodels. | e.g. [Link](https://github.com/eclipse-basyx/basyx-java-server-sdk) | -| **Asset Administration Shell** | Required | An AAS hosted on the AAS Server, containing a fully configured AID and AIMC submodel. | | -| **Asset Connector** | Required | Interfaces with MQTT or OPC UA brokers to retrieve real-time machine data. | [Link](https://github.com/fluid40/asset-connector) | -| **InfluxDB** | Optional | Stores the data collected by the Asset Connector. | [Link](https://www.influxdata.com/) | - - -## 🏭 Provided Environments - -The runtime is provided in two different environments, depending on your use case and setup preferences: - -1. πŸ’» [DevContainer](docs/environments.md#-devcontainer) - - Provides all components pre-installed and ready to use. - - The runtime itself must still be started manually from the source code. - - For development, testing, or debugging individual components. - -The following table summarizes the key characteristics of the two provided environments: - -| Environment | Components Provided | Runtime Startup | Use Case / Purpose | -| ---------------- | ------------------- | ----------------------------------------- | ----------------------------------------------------- | -| **DevContainer** | All components | Must be started manually from source code | Development, testing, debugging individual components | - -
-Architecture -
+## Resources diff --git a/configuration/DevContainerEnv.json b/configuration/DevContainerEnv.json index 76fc0d6..a3f7999 100644 --- a/configuration/DevContainerEnv.json +++ b/configuration/DevContainerEnv.json @@ -1,10 +1,11 @@ { "AasId": "https://fluid40.de/ids/shell/5793_5449_7830_4223", + "PollingInterval": 5, "AssetConnector": { - "base_url": "http://asset_connector-rt:8000/", - "time_out": 60, - "connection_time_out": 60, - "trust_env": false + "BaseUrl": "http://asset_connector-rt:8000/", + "TimeOut": 60, + "ConnectionTimeOut": 60, + "TrustEnv": false }, "AasServer": { "BaseUrl": "http://javaserver-rt:8077/", @@ -15,15 +16,13 @@ "TrustEnv": false, "EncodedIds": false }, + "InfluxDbVersion": "1", "InfluxDb": { - "host": "influx_database_v1-rt", - "port": 8086, - "username": "fluid40", - "database": "submodel-db", - "connection_time_out": 60, - "trust_env": false - }, - "PollingInterval": 5, - "WriteInflux": true, - "PutSubmodel": false + "Host": "influx_database_v1-rt", + "Port": 8086, + "Username": "fluid40", + "Database": "submodel-db", + "ConnectionTimeOut": 60, + "TrustEnv": false + } } diff --git a/docs/backup.mdx b/docs/backup.mdx new file mode 100644 index 0000000..4c74bb5 --- /dev/null +++ b/docs/backup.mdx @@ -0,0 +1,57 @@ +# Micro-Service + +
+aas_http_client +
+ +--- + +[![CI](https://github.com/fluid40/runtime/actions/workflows/docker-image.yml/badge.svg?branch=main&cache-bust=1)](https://github.com/fluid40/runtime/actions) +[![License](https://img.shields.io/badge/license-MIT-orange.svg)](LICENSE) + +- [Runtime](#runtime) + - [🌐 Overview](#-overview) + - [πŸ”— Topics](#-topics) + - [🧩 Components](#-components) + - [🏭 Provided Environments](#-provided-environments) + +## 🌐 Overview +The micro service implements the use case + +## πŸ”— Topics + +- πŸ› οΈ [Configuration](docs/configuration.md) +- 🏭 [Environments](docs/environments.md) + + +## 🧩 Components + +The runtime relies on several components to operate effectively. Each plays a specific role in data management and system integration: + +| Component | Requirement | Description | Link | +| ------------------------------ | ----------- | ------------------------------------------------------------------------------------- | ------------------------------------------------------------------- | +| **AAS Server** | Required | Responsible for managing all Asset Administration Shells (AAS) and their submodels. | e.g. [Link](https://github.com/eclipse-basyx/basyx-java-server-sdk) | +| **Asset Administration Shell** | Required | An AAS hosted on the AAS Server, containing a fully configured AID and AIMC submodel. | | +| **Asset Connector** | Required | Interfaces with MQTT or OPC UA brokers to retrieve real-time machine data. | [Link](https://github.com/fluid40/asset-connector) | +| **InfluxDB** | Optional | Stores the data collected by the Asset Connector. | [Link](https://www.influxdata.com/) | + + +## 🏭 Provided Environments + +The runtime is provided in two different environments, depending on your use case and setup preferences: + +1. πŸ’» [DevContainer](docs/environments.md#-devcontainer) + - Provides all components pre-installed and ready to use. + - The runtime itself must still be started manually from the source code. + - For development, testing, or debugging individual components. + +The following table summarizes the key characteristics of the two provided environments: + +| Environment | Components Provided | Runtime Startup | Use Case / Purpose | +| ---------------- | ------------------- | ----------------------------------------- | ----------------------------------------------------- | +| **DevContainer** | All components | Must be started manually from source code | Development, testing, debugging individual components | + +
+Architecture +
+ diff --git a/docs/configuration.md b/docs/configuration.md index cb00b28..67dfb35 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -3,9 +3,16 @@ - [πŸ› οΈ Configuration](#️-configuration) - [🌐 Overview](#-overview) - [πŸ“‘ Configuration Structure](#-configuration-structure) - - [πŸ“‹ Configuration Parameters overview](#-configuration-parameters-overview) - - [πŸ” AAS Server Authentication Settings](#-aas-server-authentication-settings) - - [πŸ—„οΈ Influx Database](#️-influx-database) + - [βš™οΈ General Settings](#️-general-settings) + - [πŸ”— Asset Connector Settings](#-asset-connector-settings) + - [πŸ–₯️ AAS Server Settings](#️-aas-server-settings) + - [πŸ›‘οΈ AAS Server Authentication Settings](#️-aas-server-authentication-settings) + - [πŸ”’ Basic Authentication Settings](#-basic-authentication-settings) + - [πŸ—οΈ oAuth2 Settings](#️-oauth2-settings) + - [πŸͺͺ Bearer Token Settings](#-bearer-token-settings) + - [πŸ“Š InfluxDB Settings](#-influxdb-settings) + - [πŸ“Š1️⃣ InfluxDB version 1 Settings](#1️⃣-influxdb-version-1-settings) + - [πŸ“Š2️⃣ InfluxDB version 2 Settings](#2️⃣-influxdb-version-2-settings) - [πŸ”‘ Additional Passwords](#-additional-passwords) @@ -24,99 +31,129 @@ If this variable is not set, the runtime cannot start correctly. The configuration file is written in JSON format (often stored as .json or .yml) and contains all necessary parameters for connecting the runtime with external services. It is organized into distinct sections, each responsible for a specific system component: +- General Setting: Defines the core runtime parameters, such as the AAS identifier and polling interval, required for the application to operate. +- Asset Connector settings: These are the connection settings for the Asset Connector micro service's REST API, which handles the connection to an MQTT or OPC UA broker. +- AAS Server settings: These are the connection settings for the AAS Server, including authentication, proxy, TLS/SSL verification and timeouts. +- InfluxDB Settings (optional): Database connection details for storing time-series data, e.g. host, port, credentials and database name.. -AasId: Unique identifier of the Asset Administration Shell (AAS). +A minimal configuration must define at least the `AasId`, `the AssetConnector.BaseUrl` and the `AasServer.BaseUrl`. +All other parameters are optional and fall back to reasonable defaults. -PollingInterval: Interval in seconds for retrieving values from the broker (optional, default: 5). +Passwords and secrets should be provided via environment variables, not in the configuration file. See the [Additional Passwords](#-additional-passwords) section for details. -AssetConnector: Connection settings for the Asset Connector REST API (e.g., base URL, timeout values, proxy behavior). +Example Structure for AAS Server with oAuth2 authentication and InfluxDB version 1 integration: -AasServer: Connection settings for the AAS Server, including authentication, proxy, TLS/SSL verification, and timeouts. +```yml +{ # General Setting + "AasId": "https://fluid40.de/ids/shell/8547_5433_6140_7422", + "PollingInterval": 5, + "AssetConnector": { # Asset Connector Settings + "BaseUrl": "http://mqtt_connector:8000", + "TimeOut": 200, + "ConnectionTimeOut": 100, + "TrustEnv": true, + }, + "AasServer": { # AAS Server Settings + "BaseUrl": "http://javaserver:8077/", + "ConnectionTimeOut": 100, + "SslVerify": true, + "TrustEnv": true, + "AuthenticationSettings": { # AAS Server Authentication Settings + "OAuth": { + "ClientId": "my-client-id", + "TokenUrl": "https://auth-server.example.com/oauth/token", + "GrantType": "client_credentials" + } + }, + "InfluxDbVersion": "1", # InfluxDB Version Setting + "InfluxDb": { # InfluxDB Settings + "Host": "influx_database", + "Port": 8086, + "Username": "fluid40", + "Database": "submodel-db", + "TrustEnv": true, + } +} +``` -InfluxDb (optional): Database connection details for storing time-series data (e.g., host, port, credentials, database name). +### βš™οΈ General Settings +**Required** -A detailed overview of the parameters can be found [here](#configuration-parameters). +The *General Settings* section defines the core parameters required for the runtime to operate. These include the unique identifier for the Asset Administration Shell (AAS) and the polling interval for retrieving values from the broker. Proper configuration of these fields ensures the runtime can identify itself and function at the desired update frequency. -A minimal configuration must define at least the AasId, the AssetConnector.base_url, the AasServer.base_url, and the InfluxDb.host/InfluxDb.port. -All other parameters are optional and fall back to reasonable defaults. +| Field | Description | Required | Default | +| ------------------- | ------------------------------------------------------------ | -------- | ------- | +| **AasId** | Unique identifier of the Asset Administration Shell (AAS). | βœ… Yes | – | +| **PollingInterval** | Interval (in seconds) for retrieving values from the broker. | ❌ No | `5` | Example: +```json +"AasId": "https://fluid40.de/ids/shell/8547_5433_6140_7422", +"PollingInterval": 5 +``` +### πŸ”— Asset Connector Settings +**Required** -```yml -{ - "AasId": "https://fluid40.de/ids/shell/8547_5433_6140_7422", # ID of the AAS - "PollingInterval": 5, # Polling interval in seconds for retrieving values from the broker (optional, default: 5) - "PutSubmodel": true, # Enable putting changed dynamic submodels to AAS server - "AssetConnector": { # Connection settings for the Asset Connector - "base_url": "http://mqtt_connector:8000", # The base URL of the asset connector REST API (required) - "time_out": 200, # API call timeout in seconds (optional, default: 200) - "connection_time_out": 100, # Connection establishment timeout in seconds (optional, default: 100) - "trust_env": true, # Disable proxy usage from environment (optional, default: true) - }, - "AasServer": { # Connection settings for the AAS Server - "BaseUrl": "http://javaserver:8077/", # Base URL of the AAS server (required) - "HttpsProxy": null, # HTTPS proxy (optional, default: null) - "HttpProxy": null, # HTTP proxy (optional, default: null) - "TimeOut": 200, # API call timeout in seconds (optional, default: 200) - "ConnectionTimeOut": 100, # Connection establishment timeout in seconds (optional, default: 100) - "SslVerify": true, # Verify TLS/SSL certificates (optional, default: true) - "TrustEnv": true, # Disable proxy usage from environment (optional, default: true) - "AuthenticationSettings": { # Authentication settings (optional) - "BasicAuth": { # Settings for basic auth - "Username": "admin" # Username for HTTP Basic Authentication - }, - "OAuth": { # Settings for oAuth2 - "ClientId": "my-client-id", # OAuth2 client identifier - "TokenUrl": "https://auth-server.example.com/oauth/token", # OAuth2 token endpoint URL - "GrantType": "client_credentials" # OAuth2 grant type (`client_credentials` or `password`) - } +The *Asset Connector Settings* section defines how the runtime communicates with the REST API of the Asset Connector micro service. These parameters specify the API endpoint, timeouts, and proxy handling, ensuring reliable and configurable integration with external asset data sources. + +| Field | Description | Required | Default | +| ------------------------------------ | -------------------------------------------- | -------- | ------- | +| **AssetConnector.BaseUrl** | Base URL of the Asset Connector REST API. | βœ… Yes | – | +| **AssetConnector.TimeOut** | API call timeout in seconds. | ❌ No | `200` | +| **AssetConnector.ConnectionTimeOut** | Connection establishment timeout in seconds. | ❌ No | `100` | +| **AssetConnector.TrustEnv** | Disable proxy usage from environment. | ❌ No | `true` | + +Example: +```json +"AssetConnector": { + "BaseUrl": "http://mqtt_connector:8000", + "TimeOut": 200, + "ConnectionTimeOut": 100, + "TrustEnv": true, +} +``` + +### πŸ–₯️ AAS Server Settings +**Required** + +The AAS Server Settings section defines how the runtime connects to and interacts with the Asset Administration Shell (AAS) server. These parameters include the server endpoint, timeouts, proxy settings, and SSL verification, ensuring secure and reliable communication with the AAS server for all core operations. + +**References:** +For the connection to the AAS Server the [aas-http-client](https://github.com/fluid40/aas-http-client) package is used. +For detailed information about the AAS Server configuration, see the [aas-http-client documentation](https://github.com/fluid40/aas-http-client/blob/main/docs/configuration.md). + + +| Field | Description | Required | Default | +| ------------------------------- | -------------------------------------------- | -------- | ------- | +| **AasServer.BaseUrl** | Base URL of the AAS server. | βœ… Yes | – | +| **AasServer.HttpsProxy** | HTTPS proxy. | ❌ No | `null` | +| **AasServer.HttpProxy** | HTTP proxy. | ❌ No | `null` | +| **AasServer.TimeOut** | API call timeout in seconds. | ❌ No | `200` | +| **AasServer.ConnectionTimeOut** | Connection establishment timeout in seconds. | ❌ No | `100` | +| **AasServer.SslVerify** | Verify TLS/SSL certificates. | ❌ No | `true` | +| **AasServer.TrustEnv** | Disable proxy usage from environment. | ❌ No | `true` | + +Example (with Basic Authentication): +Example: +```json +"AasServer": { + "BaseUrl": "http://javaserver:8077/", + "HttpsProxy": "", + "HttpProxy": "", + "TimeOut": 200, + "ConnectionTimeOut": 100, + "SslVerify": true, + "TrustEnv": true, + "AuthenticationSettings": { + "BasicAuth": { + "Username": "admin" } - }, - "InfluxDb": { # Connection settings for the Influx database (optional) - "host": "influx_database", # The hostname or IP address of the InfluxDB server (required) - "port": 8086, # The port number of the InfluxDB server (required) - "username": "fluid40", # The username for the InfluxDB server (optional, default: "") - "database": "submodel-db", # The database name to use in InfluxDB (optional, default: "") - "connection_time_out": 100, # Connection establishment timeout in seconds (optional, default: 100) - "trust_env": true, # Disable proxy usage from environment (optional, default: true) } } ``` - -### πŸ“‹ Configuration Parameters overview - -The configuration file is divided into several sections, each defining the parameters for a specific system component. Some fields are mandatory for the runtime to start, while others are optional and fall back to sensible defaults. The following table provides an overview of all available configuration parameters, updated to match the structure and naming from the Configuration Structure section: - -| Field | Description | Required | Default | -| ------------------------------------------------------- | ------------------------------------------------------------ | -------- | ---------------------- | -| **AasId** | Unique identifier of the Asset Administration Shell (AAS). | βœ… Yes | – | -| **PollingInterval** | Interval (in seconds) for retrieving values from the broker. | ❌ No | `5` | -| **PutSubmodel** | Enable putting changed dynamic submodels to AAS server | ❌ No | `false` | -| **AssetConnector.base_url** | Base URL of the Asset Connector REST API. | βœ… Yes | – | -| **AssetConnector.time_out** | API call timeout in seconds. | ❌ No | `200` | -| **AssetConnector.connection_time_out** | Connection establishment timeout in seconds. | ❌ No | `100` | -| **AssetConnector.trust_env** | Disable proxy usage from environment. | ❌ No | `true` | -| **AasServer.BaseUrl** | Base URL of the AAS server. | βœ… Yes | – | -| **AasServer.HttpsProxy** | HTTPS proxy. | ❌ No | `null` | -| **AasServer.HttpProxy** | HTTP proxy. | ❌ No | `null` | -| **AasServer.TimeOut** | API call timeout in seconds. | ❌ No | `200` | -| **AasServer.ConnectionTimeOut** | Connection establishment timeout in seconds. | ❌ No | `100` | -| **AasServer.SslVerify** | Verify TLS/SSL certificates. | ❌ No | `true` | -| **AasServer.TrustEnv** | Disable proxy usage from environment. | ❌ No | `true` | -| **AasServer.AuthenticationSettings.BasicAuth.Username** | Username for HTTP Basic Authentication. | ❌ No | `""` | -| **AasServer.AuthenticationSettings.OAuth.ClientId** | OAuth2 client identifier. | ❌ No | `""` | -| **AasServer.AuthenticationSettings.OAuth.TokenUrl** | OAuth2 token endpoint URL. | ❌ No | `""` | -| **AasServer.AuthenticationSettings.OAuth.GrantType** | OAuth2 grant type (`client_credentials` or `password`). | ❌ No | `"client_credentials"` | -| **InfluxDb.host** | Hostname or IP address of the InfluxDB server. | βœ… Yes | – | -| **InfluxDb.port** | Port number of the InfluxDB server. | βœ… Yes | – | -| **InfluxDb.username** | Username for the InfluxDB server. | ❌ No | `""` | -| **InfluxDb.database** | Database name to use in InfluxDB. | ❌ No | `""` | -| **InfluxDb.connection_time_out** | Connection establishment timeout in seconds. | ❌ No | `100` | -| **InfluxDb.trust_env** | Disable proxy usage from environment. | ❌ No | `true` | - - -### πŸ” AAS Server Authentication Settings +#### πŸ›‘οΈ AAS Server Authentication Settings +**Optional** The runtime supports multiple authentication methods for connecting to the AAS Server, allowing secure integration in a variety of environments. Authentication credentials should be provided using environment variables whenever possible to avoid storing sensitive information in configuration files. @@ -125,71 +162,138 @@ The runtime supports multiple authentication methods for connecting to the AAS S 1. **Basic Authentication** - Set the username in the configuration under `AasServer.AuthenticationSettings.BasicAuth.Username`. - Provide the password via the `RUNTIME_BASIC_AUTH_PW` environment variable. - - Example: - ```json - "AuthenticationSettings": { - "BasicAuth": { - "Username": "admin" - } - } - ``` - -2. **Bearer Token Authentication** - - No username is required in the configuration. - - Provide the token via the `RUNTIME_BEARER_TOKEN` environment variable. -3. **OAuth2 Client Credentials** +2. **OAuth2 Client Credentials** - Set the client ID, token URL, and grant type in the configuration under `AasServer.AuthenticationSettings.OAuth`. - Provide the client secret via the `RUNTIME_OAUTH_SECRET` environment variable. - - Example: - ```json - "AuthenticationSettings": { - "OAuth": { - "ClientId": "my-client-id", - "TokenUrl": "https://auth-server.example.com/oauth/token", - "GrantType": "client_credentials" - } - } - ``` - -**Best Practices:** -- Always use environment variables for passwords, secrets, and tokens (see [Additional Passwords](#-additional-passwords)). -- Only include the authentication method(s) you intend to use in your configuration. -- If multiple authentication methods are configured, the runtime will use them in the following order of precedence: Bearer Token β†’ OAuth2 β†’ Basic Auth. -- Ensure your authentication server endpoints and credentials are correct and accessible from the runtime environment. -**References:** -For detailed information about the AAS Server authentication configuration, see the [aas-http-client documentation](https://github.com/fluid40/aas-http-client/blob/main/docs/configuration.md). +3. **Bearer Token Authentication** + - No username is required in the configuration. + - Provide the token via the `RUNTIME_BEARER_TOKEN` environment variable. + +If *Bearer Token* or no authentication is used, the `AuthenticationSettings` node can be omitted. -### πŸ—„οΈ Influx Database +Passwords and secrets should be provided via environment variables, not in the configuration file. See the [Additional Passwords](#-additional-passwords) section for details. + +##### πŸ”’ Basic Authentication Settings + +| Field | Description | Required | Default | +| ------------------------------------------------------- | --------------------------------------- | -------- | ------- | +| **AasServer.AuthenticationSettings.BasicAuth.Username** | Username for HTTP Basic Authentication. | βœ… Yes | - | + +Provide the password via the `RUNTIME_BASIC_AUTH_PW` environment variable. + +Example: +```json +"AuthenticationSettings": { + "BasicAuth": { + "Username": "admin" + } +} +``` + +##### πŸ—οΈ oAuth2 Settings + +| Field | Description | Required | Default | +| ---------------------------------------------------- | ------------------------------------------------------- | -------- | ---------------------- | +| **AasServer.AuthenticationSettings.OAuth.ClientId** | OAuth2 client identifier. | βœ… Yes | - | +| **AasServer.AuthenticationSettings.OAuth.TokenUrl** | OAuth2 token endpoint URL. | βœ… Yes | - | +| **AasServer.AuthenticationSettings.OAuth.GrantType** | OAuth2 grant type (`client_credentials` or `password`). | ❌ No | `"client_credentials"` | + +Provide the client secret via the `RUNTIME_OAUTH_SECRET` environment variable. + +Example: +```json +"AuthenticationSettings": { + "OAuth": { + "ClientId": "my-client-id", + "TokenUrl": "https://auth-server.example.com/oauth/token", + "GrantType": "client_credentials" + } +} +``` + +##### πŸͺͺ Bearer Token Settings + +Provide the token via the `RUNTIME_BEARER_TOKEN` environment variable. +The `AuthenticationSettings` node can be omitted. + +### πŸ“Š InfluxDB Settings +**Optional** The InfluxDB integration enables the runtime to store time-series data for later analysis and visualization. This database connection is **optional**. If the `InfluxDb` section is omitted from your configuration file, the runtime will operate normally but will not persist any data to a database. **Behavior:** -- If the `InfluxDb` node is **not set** in the configuration file, no data will be written to a database. +- If the `InfluxDb` node is **not set** in the configuration file, no data will be written to a database. `InfluxDbVersion` - If the `InfluxDb` node is present but the specified database cannot be found or accessed, the runtime will continue operating, but data will not be written to the database. No critical errors will be raised, ensuring robust operation even if the database is temporarily unavailable. -**Best Practices:** -- Use environment variables to provide sensitive credentials (see [Additional Passwords](#-additional-passwords)). -- Ensure the InfluxDB server is reachable from the runtime environment. -- Regularly back up your InfluxDB data if persistence is critical. +InfluxDB version 1 and version 2 is supported. Set 'InfluxDbVersion' to: +- `1` for InfluxDB version 1 +- `2` for InfluxDB version 2 + +#### πŸ“Š1️⃣ InfluxDB version 1 Settings + +| Field | Description | Required | Default | +| ------------------------------ | ---------------------------------------------- | -------- | ------- | +| **InfluxDbVersion** | Set the version of the Influx DB | βœ… Yes | - | +| **InfluxDb.Host** | Hostname or IP address of the InfluxDB server. | βœ… Yes | – | +| **InfluxDb.Port** | Port number of the InfluxDB server. | βœ… Yes | – | +| **InfluxDb.Username** | Username for the InfluxDB server. | ❌ No | `""` | +| **InfluxDb.Database** | Database name to use in InfluxDB. | ❌ No | `""` | +| **InfluxDb.ConnectionTimeOut** | Connection establishment timeout in seconds. | ❌ No | `100` | +| **InfluxDb.TrustEnv** | Disable proxy usage from environment. | ❌ No | `true` | + +The property 'InfluxDbVersion' must be set to the value 1. +Provide the database password via the `RUNTIME_INFLUX_PW` environment variable. + +Example: +```json +"InfluxDbVersion": "1", +"InfluxDb": { + "Host": "influx_database_v1-rt", + "Port": 8086, + "Username": "fluid40", + "Database": "submodel-db", + "ConnectionTimeOut": 60, + "TrustEnv": false +} +``` + +#### πŸ“Š2️⃣ InfluxDB version 2 Settings -**Troubleshooting:** -- If data is not appearing in InfluxDB, check the runtime logs for connection errors or authentication issues. -- Verify that the `host`, `port`, and `database` values are correct and that the InfluxDB server is running. +| Field | Description | Required | Default | +| ------------------------------ | -------------------------------------------- | -------- | ------- | +| **InfluxDbVersion** | Set the version of the Influx DB | βœ… Yes | - | +| **InfluxDb.Url** | URL of the InfluxDB server. | βœ… Yes | – | +| **InfluxDb.Organization** | Organization name to use in InfluxDB. | βœ… Yes | – | +| **InfluxDb.Bucket** | Bucket name to use in InfluxDB. | βœ… Yes | - | +| **InfluxDb.ConnectionTimeOut** | Connection establishment timeout in seconds. | ❌ No | `100` | -For more details on all available parameters, see the [Configuration Parameters overview](#-configuration-parameters-overview). +The property 'InfluxDbVersion' must be set to the value 2. +Provide the database token via the `RUNTIME_INFLUX_PW` environment variable. +Example: +```json +"InfluxDbVersion": "2", +"InfluxDb": { + "Url": "http://influx_database_v2-rt:8086/", + "Organization": "fluid40-org", + "Bucket": "submodel-db", + "ConnectionTimeOut": 60 +} +``` ## πŸ”‘ Additional Passwords Some components may require authentication. Instead of storing sensitive information in the configuration file, passwords must be provided via environment variables: -| Component | Environment Variable | Description | -| ---------- | ----------------------- | --------------------------------------------------------------- | -| AAS Server | `RUNTIME_BASIC_AUTH_PW` | Password used to authenticate with the AAS Server by basic auth | -| AAS Server | `RUNTIME_OAUTH_SECRET` | Secret used to authenticate with the AAS Server by oAuth | -| AAS Server | `RUNTIME_BEARER_TOKEN` | Bearer Token used to authenticate with the AAS Server by Token | -| InfluxDB | `RUNTIME_INFLUX_PW` | Password used to authenticate with the InfluxDB | +| Component | Environment Variable | Description | +| ---------- | ----------------------- | ------------------------------------------------------------------ | +| AAS Server | `RUNTIME_BASIC_AUTH_PW` | Password used to authenticate with the AAS Server by basic auth | +| AAS Server | `RUNTIME_OAUTH_SECRET` | Secret used to authenticate with the AAS Server by oAuth | +| AAS Server | `RUNTIME_BEARER_TOKEN` | Bearer Token used to authenticate with the AAS Server by Token | +| InfluxDB | `RUNTIME_INFLUX_PW` | Password (v1) or token (v2) used to authenticate with the InfluxDB | This approach keeps credentials secure and ensures they can be easily managed across different environments (development, staging, production). + + diff --git a/docs/environments.md b/docs/development.md similarity index 100% rename from docs/environments.md rename to docs/development.md diff --git a/requirements.txt b/requirements.txt index d14cf28..bf2e0f6 100644 Binary files a/requirements.txt and b/requirements.txt differ diff --git a/runtime/core/core_process.py b/runtime/core/core_process.py index ccd5231..9610d24 100644 --- a/runtime/core/core_process.py +++ b/runtime/core/core_process.py @@ -1,5 +1,4 @@ import datetime -import json import logging from aas_http_client import SdkWrapper @@ -10,27 +9,44 @@ from runtime.core.server_initialization import RuntimeStates from runtime.interfaces.influx_1_interface import InfluxClient -from runtime.model.classes import PayloadValue +from runtime.model.classes import PayloadValue, ShellInfo router = APIRouter() logger = logging.getLogger(__name__) +counter = 1 + + +def _raise_counter(): + global counter + counter += 1 + + +def _get_counter() -> int: + return counter + def runtime_process(states: RuntimeStates): """Placeholder function for runtime process.""" + counter = _get_counter() + logger.info(f"Runtime process execution count: {counter}") + asset_values = get_asset_values(states) # Write asset values to InfluxDB if states.influx_client is not None: logger.info("Writing asset values to InfluxDB.") - _write_payloads_to_influx(asset_values, states.influx_client) + _write_payloads_to_influx(asset_values, states.influx_client, states.shell_info, counter) # Push changed submodels to AAS server if states.runtime_configuration.put_submodel: logger.info("Putting changed dynamic submodels to AAS server.") _push_submodels(states.wrapper, states.dynamic_submodel_cache) + # Increase counter + _raise_counter() + def get_asset_values(states: RuntimeStates): """Get the values from mapping configurations relations from broker. @@ -70,7 +86,7 @@ def get_asset_values(states: RuntimeStates): return asset_values -def _write_payloads_to_influx(payload_values: list[PayloadValue], influx_client: InfluxClient): +def _write_payloads_to_influx(payload_values: list[PayloadValue], influx_client: InfluxClient, shell_info: ShellInfo, counter: int): """Write payload values to InfluxDB. :param payload_values: List of PayloadValue instances. @@ -81,24 +97,45 @@ def _write_payloads_to_influx(payload_values: list[PayloadValue], influx_client: logger.warning("No payload values to write to InfluxDB.") return - payloads_dict: dict[str, dict[str, str]] = {} + fields: dict[str, dict[str, str]] = {} for payload_value in payload_values: - if payload_value.parent_name not in payloads_dict: - payloads_dict[payload_value.parent_name] = {} - payloads_dict[payload_value.parent_name][payload_value.property_name] = payload_value.value - - if influx_client: - try: - for measurement, fields in payloads_dict.items(): - # add timestamp if not present in fields - if "timestamp" not in fields: - fields["timestamp"] = datetime.datetime.now(tz=datetime.timezone.utc).isoformat() - - influx_client.write_data(fields, measurement) - except Exception as e: - logger.error(f"Could not write data to InfluxDB: {e}") - else: + fields[payload_value.property_name] = payload_value.value + + if influx_client is None: logger.error("InfluxDB client is not initialized.") + return + + measurement = shell_info.id_short.replace(" ", "_") + + pattern = "" + if "Pattern" in fields: + pattern = fields["Pattern"] + fields.pop("Pattern") + + machine_state = "" + if "MachineState" in fields: + machine_state = fields["MachineState"] + fields.pop("MachineState") + + if "Counter" in fields: + counter = fields["Counter"] + + tags = { + "source": shell_info.display_name.replace(" ", "_"), + "counter": counter, + "machine_state": machine_state, + "pattern": pattern, + } + + try: + # Ensure timestamp field exists + if "timestamp" not in fields and "Timestamp" not in fields: + fields["timestamp"] = datetime.datetime.now(tz=datetime.timezone.utc).isoformat() + + influx_client.write_data(fields, measurement, tags) + + except Exception as e: + logger.error(f"Could not write data to InfluxDB: {e}") def _update_target_property_with_payload(payload_value: PayloadValue, wrapper: SdkWrapper, submodel_cache: dict[str, model.Submodel]): diff --git a/runtime/core/server_initialization.py b/runtime/core/server_initialization.py index 73f7c0d..b7d470f 100644 --- a/runtime/core/server_initialization.py +++ b/runtime/core/server_initialization.py @@ -9,8 +9,9 @@ from fastapi import HTTPException from runtime.core.utilities import aas_parser -from runtime.interfaces import asset_connector_interface, influx_1_interface -from runtime.model.classes import RuntimeStates +from runtime.interfaces import asset_connector_interface, influx_1_interface, influx_2_interface +from runtime.interfaces.influx_interface import IInfluxClient +from runtime.model.classes import RuntimeStates, ShellInfo from runtime.model.configuration import RuntimeConfiguration logger = logging.getLogger(__name__) @@ -28,6 +29,12 @@ def initialize_server(configuration: RuntimeConfiguration) -> RuntimeStates: # get the AAS aas = _get_aas(wrapper, configuration) + shell_info = ShellInfo( + id=aas.id, + id_short=aas.id_short, + display_name=_get_aas_display_name(aas), + ) + # get the AIMC submodel from the AAS logger.info("Get AIMC submodel from AAS.") aimc_submodel = aas_parser.get_aimc_submodel(wrapper, aas) @@ -39,9 +46,9 @@ def initialize_server(configuration: RuntimeConfiguration) -> RuntimeStates: asset_connector = _connect_to_asset_connector(configuration, aid_submodels) # connect to InfluxDB if enabled - influx_client = None + influx_client: IInfluxClient | None = None if configuration.influx_db_settings: - influx_client = _connect_to_db(configuration, aas) + influx_client = _connect_to_db(configuration) else: logger.warning("No InfluxDB settings provided. Database interactions disabled.") @@ -50,6 +57,7 @@ def initialize_server(configuration: RuntimeConfiguration) -> RuntimeStates: states.wrapper = wrapper states.asset_connector = asset_connector states.mapping_configurations = mapping_configurations + states.shell_info = shell_info states.influx_client = influx_client return states @@ -117,21 +125,28 @@ def _connect_to_aas_server(configuration: RuntimeConfiguration) -> SdkWrapper | return wrapper -def _connect_to_db(configuration: RuntimeConfiguration, aas: model.AssetAdministrationShell) -> influx_1_interface.InfluxClient | None: +def _connect_to_db(configuration: RuntimeConfiguration) -> IInfluxClient | None: logger.info("Connect to InfluxDB.") logger.info("Get Influx DB password from environment variable 'RUNTIME_INFLUX_PW'") password = os.getenv("RUNTIME_INFLUX_PW") - client = influx_1_interface.create_client(configuration.influx_db_settings, password) + db_version: int = configuration.influx_db_version + client: IInfluxClient | None = None + + if db_version == 1: + client = influx_1_interface.create_client(configuration.influx_db_settings, password) + + elif db_version == 2: + client = influx_2_interface.create_client(configuration.influx_db_settings, password) + + else: + logger.error(f"Unsupported InfluxDB version '{db_version}'. Only version 1 is supported.") if not client: logger.warning("No InfluxDB client available. Database interactions disabled.") return None - influx_tag = _get_aas_display_name(aas) - client.set_tag(influx_tag) - return client diff --git a/runtime/interfaces/asset_connector_interface.py b/runtime/interfaces/asset_connector_interface.py index 870ba54..02feb55 100644 --- a/runtime/interfaces/asset_connector_interface.py +++ b/runtime/interfaces/asset_connector_interface.py @@ -19,10 +19,10 @@ class AssetConnectorClient(BaseModel): :return: A AssetConnectorClient instance. """ - base_url: str = Field(default="http://localhost:8000", description="The base URL of the Asset Connector REST API.") - time_out: int = Field(default=200, description="API call timeout in seconds.") - trust_env: bool = Field(default=True, description="Disable proxy usage from environment.") - connection_time_out: int = Field(default=100, description="Connection establishment timeout in seconds.") + base_url: str = Field(default="http://localhost:8000", description="The base URL of the Asset Connector REST API.", alias="BaseUrl") + time_out: int = Field(default=200, description="API call timeout in seconds.", alias="TimeOut") + trust_env: bool = Field(default=True, description="Disable proxy usage from environment.", alias="TrustEnv") + connection_time_out: int = Field(default=100, description="Connection establishment timeout in seconds.", alias="ConnectionTimeOut") _session: Session = PrivateAttr(default=None) def __init__(self, **data): diff --git a/runtime/interfaces/influx_1_interface.py b/runtime/interfaces/influx_1_interface.py index 6169b28..d199b99 100644 --- a/runtime/interfaces/influx_1_interface.py +++ b/runtime/interfaces/influx_1_interface.py @@ -7,18 +7,21 @@ from influxdb import InfluxDBClient from pydantic import BaseModel, Field, PrivateAttr, ValidationError +from runtime.interfaces.influx_interface import IInfluxClient + logger = logging.getLogger(__name__) -class InfluxClient(BaseModel): - host: str = Field(default="influx_database", description="The hostname or IP address of the InfluxDB server.") - port: int = Field(default=8086, description="The port number of the InfluxDB server.") - username: str = Field(default="", description="The username for the InfluxDB server.") - database: str = Field(default="", description="The database name to use in InfluxDB.") - connection_time_out: int = Field(default=100, description="Connection establishment timeout in seconds.") - trust_env: bool = Field(default=False, description="Disable proxy usage from environment.") - _client: InfluxDBClient = PrivateAttr(default=bool) - _tag: str = PrivateAttr(default=str) +class InfluxClient(BaseModel, IInfluxClient): + """Client for InfluxDB interactions.""" + + host: str = Field(default="influx_database", description="The hostname or IP address of the InfluxDB 1 database.", alias="Host") + port: int = Field(default=8086, description="The port number of the InfluxDB database.", alias="Port") + username: str = Field(default="", description="The username for the InfluxDB database.", alias="Username") + database: str = Field(default="", description="The database name to use in InfluxDB.", alias="Database") + connection_time_out: int = Field(default=100, description="Connection establishment timeout in seconds.", alias="ConnectionTimeOut") + trust_env: bool = Field(default=False, description="Disable proxy usage from environment.", alias="TrustEnv") + _client: InfluxDBClient = PrivateAttr(default=None) def initialize(self, password: str) -> None: """Initialize the InfluxDB client with the given password. @@ -31,27 +34,23 @@ def initialize(self, password: str) -> None: client = InfluxDBClient(host=self.host, port=self.port, username=self.username, password=password, database=self.database, session=session) if not client: - logger.error("Could not create InfluxDB client.") + logger.error("Could not create InfluxDB v1 client.") self._client = client - def check_and_create_database(self): - logger.debug(f"Check for database '{self.database}'.") + def create_database(self): + """Check if the specified database exists, and create it if it does not.""" + + logger.debug(f"Create database '{self.database}'") dbs = [db["name"] for db in self._client.get_list_database()] + + logger.debug(f"Check if database '{self.database}' already exists") if self.database not in dbs: - logger.debug(f"Create database '{self.database}'.") + logger.debug(f"Creating database '{self.database}'") self._client.create_database(self.database) logger.debug(f"Switch to database '{self.database}'.") self._client.switch_database(self.database) - self._tag = "fluid40-runtime" - - def set_tag(self, tag: str) -> None: - """Set the source tag for data points. - - :param tag: The source tag to set. - """ - self._tag = tag.replace(" ", "_") def ping(self) -> bool: """Ping the InfluxDB server to check if it's reachable. @@ -70,34 +69,34 @@ def ping(self) -> bool: logger.error(f"Failed to ping Influx DB server: {e}") return False - def write_data(self, data: dict, measurement_name: str, tags: dict | None = None) -> bool: + def write_data(self, fields: dict, measurement: str, tags: dict) -> bool: """Write data to the InfluxDB. - :param data: The data to write, must include a 'timestamp' field in ISO format. - :param measurement_name: The name of the measurement. - :param tags: Optional tags to include with the data. + :param fields: The data to write, must include a 'timestamp' field in ISO format. + :param measurement: The name of the measurement. + :param tags: The tags to associate with the data point. :return: True if the data was written successfully, False otherwise. """ - logger.debug(f"Writing data to InfluxDB measurement '{measurement_name}' with tags: {tags}") + logger.debug(f"Writing data to InfluxDB measurement '{measurement}' with tags: {tags}") - if data is None or not isinstance(data, dict): - logger.debug(f"'{measurement_name}': Data must be a non-empty dictionary.") + if fields is None or not isinstance(fields, dict): + logger.debug(f"'{measurement}': Data must be a non-empty dictionary.") return False - if "timestamp" not in data: - logger.error(f"{measurement_name}: Data must include a 'timestamp' field.") + if "timestamp" not in fields and "Timestamp" not in fields: + logger.error(f"{measurement}: Data must include a 'timestamp' field.") return False - if tags is None: - tags = {"source": self._tag} - point = { - "measurement": measurement_name, + "measurement": measurement, "tags": tags, - "fields": data, - "time": int(datetime.fromisoformat(data["timestamp"].replace("Z", "+00:00")).timestamp() * 1e9), + "fields": fields, + "time": int(datetime.fromisoformat(fields["timestamp"].replace("Z", "+00:00")).timestamp() * 1e9), } - logger.info(f"Writing data to InfluxDB measurement '{measurement_name}'\nTags:\n{tags}'\nValues:\n{json.dumps(data, indent=4)}") + + logger.info( + f"Writing data to InfluxDB:\nMeasurement: '{measurement}'\nTags:\n{json.dumps(tags, indent=4)}'\nValues:\n{json.dumps(fields, indent=4)}" + ) success: bool = self._client.write_points([point], time_precision="n") if not success: @@ -107,10 +106,11 @@ def write_data(self, data: dict, measurement_name: str, tags: dict | None = None return True -def create_client(config_dict: dict, password: str) -> InfluxClient: +def create_client(config_dict: dict, password: str) -> IInfluxClient: """Create a HTTP client for a asset connector connection from a given configuration. :param config_dict: The configuration dictionary for the asset connector. + :param password: The password for the InfluxDB user. :raises ValidationError: If the configuration is invalid. :return: A ConnectorClient instance or None if creation failed. """ @@ -131,7 +131,7 @@ def create_client(config_dict: dict, password: str) -> InfluxClient: if not connected: return None - client.check_and_create_database() + client.create_database() return client diff --git a/runtime/interfaces/influx_2_interface.py b/runtime/interfaces/influx_2_interface.py new file mode 100644 index 0000000..4d9b34a --- /dev/null +++ b/runtime/interfaces/influx_2_interface.py @@ -0,0 +1,191 @@ +import json +import logging +import time +from datetime import datetime +from typing import Any + +import requests +from influxdb_client import InfluxDBClient, Point +from influxdb_client.client.write_api import SYNCHRONOUS +from influxdb_client.domain.bucket_retention_rules import BucketRetentionRules +from pydantic import BaseModel, Field, PrivateAttr, ValidationError + +from runtime.interfaces.influx_interface import IInfluxClient + +logger = logging.getLogger(__name__) + + +class InfluxClient(BaseModel, IInfluxClient): + """Client for InfluxDB interactions.""" + + url: str = Field(default="influx_database", description="The base URL of the InfluxDB 2 database.", alias="Url") + organization: str = Field(..., description="The Organization name to use in InfluxDB.", alias="Organization") + bucket: str = Field(..., description="The Bucket name to use in InfluxDB.", alias="Bucket") + connection_time_out: int = Field(default=100, description="Connection establishment timeout in seconds.", alias="ConnectionTimeOut") + _client: InfluxDBClient = PrivateAttr(default=None) + _tag: str = PrivateAttr(default=str) + _organization_id: str = PrivateAttr(default=str) + _bucket_id: str = PrivateAttr(default=str) + + def initialize(self, token: str) -> None: + """Initialize the InfluxDB client with the given token. + + :param token: The token for the InfluxDB user. + """ + client = InfluxDBClient(url=self.url, token=token, org=self.organization) + + if not client: + logger.error("Could not create InfluxDB v2 client.") + return + + orgs_api = client.organizations_api() + orgs: list = orgs_api.find_organizations(org=self.organization) + + if len(orgs) == 0 or orgs[0] is None: + logger.error(f"Organization '{self.organization}' not found in InfluxDB.") + else: + self._organization_id = orgs[0].id + + self._client = client + + def create_database(self): + """Check if the specified bucket exists, and create it if it does not.""" + logger.debug(f"Create bucket '{self.bucket}'") + buckets_api = self._client.buckets_api() + + logger.debug(f"Check if bucket '{self.bucket}' already exists") + bucket = buckets_api.find_bucket_by_name(self.bucket) + + if bucket is not None: + logger.debug(f"Bucket '{self.bucket}' already exists") + self._bucket_id = bucket.id + return + + logger.debug(f"Creating bucket '{self.bucket}'") + retention = BucketRetentionRules(type="expire", every_seconds=30 * 24 * 3600) + bucket = buckets_api.create_bucket( + bucket_name=self.bucket, + org_id=self._organization_id, + retention_rules=retention, + ) + + self._bucket_id = bucket.id + + def ping(self) -> bool: + """Ping the InfluxDB server to check if it's reachable. + + :return: True if the server is reachable, False otherwise. + """ + logger.debug(f"Pinging InfluxDB server '{self.url}'.") + if self._client is None: + logger.error("InfluxDB client is not initialized.") + return False + + try: + health = self._client.health() + except Exception as e: + logger.error(f"Failed to ping Influx DB server: {e}") + return False + + if health.status != "pass": + logger.error(f"InfluxDB server health check failed: {health.message}") + return False + + return True + + def write_data(self, fields: dict, measurement: str, tags: dict) -> bool: + """Write data to the InfluxDB. + + :param fields: The data to write, must include a 'timestamp' field in ISO format. + :param measurement: The name of the measurement. + :param tags: The tags to associate with the data point. + :return: True if the data was written successfully, False otherwise. + """ + logger.debug(f"Writing data to InfluxDB measurement '{measurement}' with tags: {tags}") + + if fields is None or not isinstance(fields, dict): + logger.debug(f"'{measurement}': Data must be a non-empty dictionary.") + return False + + if "timestamp" not in fields and "Timestamp" not in fields: + logger.error(f"{measurement}: Data must include a 'timestamp' field.") + return False + + point = { + "measurement": measurement, + "tags": tags, + "fields": fields, + "time": int(datetime.fromisoformat(fields["timestamp"].replace("Z", "+00:00")).timestamp() * 1e9), + } + + logger.info( + f"Writing data to InfluxDB:\nMeasurement: '{measurement}'\nTags:\n{json.dumps(tags, indent=4)}'\nValues:\n{json.dumps(fields, indent=4)}" + ) + + try: + write_api = self._client.write_api(write_options=SYNCHRONOUS) + write_api.write(record=point, bucket=self.bucket) + except Exception as e: + logger.error(f"Failed to write data point '{point}' to InfluxDB: {e}") + return False + + return True + + +def create_client(config_dict: dict, token: str) -> IInfluxClient: + """Create a HTTP client for a asset connector connection from a given configuration. + + :param config_dict: The configuration dictionary for the asset connector. + :param token: The token for InfluxDB authentication. + :raises ValidationError: If the configuration is invalid. + :return: A ConnectorClient instance or None if creation failed. + """ + logger.info("Create Influx Database client.") + + try: + config_string = json.dumps(config_dict, indent=4) + client = InfluxClient.model_validate_json(config_string) + except ValidationError as ve: + raise ValidationError(f"Invalid Influx DB configuration file: {ve}") from ve + + if not client.organization.endswith("-org"): + logger.warning(f"Adjusting Influx DB organization name from '{client.organization}' to '{client.organization}-org'.") + client.organization = f"{client.organization}-org" + + logger.info(f"Using Influx DB configuration: '{client.url}' | organization: '{client.organization}'.") + + client.initialize(token) + + connected = _establish_connection(client) + + if not connected: + return None + + try: + client.create_database() + except Exception as e: + logger.error(f"Failed to create InfluxDB database: {e}") + return None + + return client + + +def _establish_connection(client: InfluxClient) -> bool: + start_time = time.time() + logger.info(f"Try to connect to Influx DB '{client.url}' for {client.connection_time_out} seconds") + counter: int = 0 + while True: + try: + root = client.ping() + if root: + logger.info(f"Connected to Influx DB at '{client.url}' successfully.") + return True + except requests.exceptions.ConnectionError: + pass + if time.time() - start_time > client.connection_time_out: + logger.error(f"Connection to Influx DB timed out after {client.connection_time_out} seconds.") + return False + + counter += 1 + logger.warning(f"Retrying connection to Influx DB (attempt: {counter})") + time.sleep(5) diff --git a/runtime/interfaces/influx_interface.py b/runtime/interfaces/influx_interface.py new file mode 100644 index 0000000..1772da1 --- /dev/null +++ b/runtime/interfaces/influx_interface.py @@ -0,0 +1,32 @@ +"""Interface definitions for InfluxDB clients.""" + + +class IInfluxClient: + """Interface for InfluxDB clients.""" + + def initialize(self, password: str) -> None: + """Initialize the InfluxDB client with the given password. + + :param password: The password for the InfluxDB user. + """ + raise NotImplementedError + + def ping(self) -> bool: + """Ping the InfluxDB server to check if it's reachable. + + :return: True if the server is reachable, False otherwise. + """ + raise NotImplementedError + + def write_data(self, data: dict, measurement_name: str) -> bool: + """Write data to the InfluxDB. + + :param data: The data to write, must include a 'timestamp' field in ISO format. + :param measurement_name: The name of the measurement. + :return: True if the data was written successfully, False otherwise. + """ + raise NotImplementedError + + def create_database(self): + """Check if the specified database exists, and create it if it does not.""" + raise NotImplementedError diff --git a/runtime/model/classes.py b/runtime/model/classes.py index a47d2bc..84e4a1d 100644 --- a/runtime/model/classes.py +++ b/runtime/model/classes.py @@ -6,6 +6,21 @@ from runtime.model.configuration import RuntimeConfiguration +class ShellInfo: + """Class representing shell information for a model.""" + + def __init__(self, id: str, id_short: str, display_name: str): + """Initialize ShellInfo with given parameters. + + :param id: The unique identifier of the shell. + :param id_short: The short identifier of the shell. + :param display_name: The display name of the shell. + """ + self.id = id + self.id_short = id_short + self.display_name = display_name + + class RuntimeStates: """Class representing the runtime states of the server.""" @@ -15,6 +30,7 @@ class RuntimeStates: mapping_configurations: MappingConfigurations = None influx_client: influx_1_interface.InfluxClient = None dynamic_submodel_cache: dict[str, model.Submodel] = {} + shell_info: ShellInfo = None class PayloadValue: diff --git a/runtime/model/configuration.py b/runtime/model/configuration.py index 1b0f7fd..8383945 100644 --- a/runtime/model/configuration.py +++ b/runtime/model/configuration.py @@ -25,6 +25,7 @@ class RuntimeConfiguration(BaseModel): aas_server_settings: dict = Field(..., alias="AasServer") aas_id: str = Field(..., alias="AasId", description="The ID of the AAS.") asset_connector_settings: dict = Field(..., alias="AssetConnector") + influx_db_version: int = Field(default=0, alias="InfluxDbVersion") influx_db_settings: dict = Field(default={}, alias="InfluxDb") polling_interval: int = Field( default=5, alias="PollingInterval", description="Polling interval in seconds for retrieving values from the broker."