feat: Expose metrics about used TeamForCapella license usage

A new gauge metric `t4c_licenses` was added to the `/metrics` endpoint,
which contains the number of currently used licenses per TeamForCapella
instance. The instance name is used as label for the metric.

Note: If you have multiple TeamForCapella server instances, which point
to the same license server, this is not detected. Therefore, make sure to
filter the `t4c_licenses` for one instance first.

backend/capellacollab/__main__.py

@@ -13,7 +13,8 @@
 from fastapi.middleware import cors
 
 import capellacollab.projects.toolmodels.backups.runs.interface as pipeline_runs_interface
-import capellacollab.sessions.metrics
+import capellacollab.sessions.metrics as sessions_metrics
+import capellacollab.settings.modelsources.t4c.metrics as t4c_metrics
 
 # This import statement is required and should not be removed! (Alembic will not work otherwise)
 from capellacollab.config import config
@@ -84,7 +85,8 @@ async def shutdown():
     on_startup=[
         startup,
         idletimeout.terminate_idle_sessions_in_background,
-        capellacollab.sessions.metrics.register,
+        sessions_metrics.register,
+        t4c_metrics.register,
         pipeline_runs_interface.schedule_refresh_and_trigger_pipeline_jobs,
     ],
     middleware=[

backend/capellacollab/settings/modelsources/t4c/metrics.py

@@ -0,0 +1,39 @@
+# SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import typing as t
+
+import prometheus_client
+import prometheus_client.core
+
+from capellacollab.core import database
+
+from . import crud, interface
+
+
+class UsedT4CLicensesCollector:
+    def collect(self) -> t.Iterable[prometheus_client.core.Metric]:
+        metric = prometheus_client.core.GaugeMetricFamily(
+            "t4c_licenses",
+            "Currently used T4C licenses per registered TeamForCapella instance.",
+        )
+
+        with database.SessionLocal() as db:
+            instances = crud.get_t4c_instances(db)
+
+        if not instances:
+            return
+
+        for instance in instances:
+            try:
+                t4c_status = interface.get_t4c_status(instance)
+                used_licenses = t4c_status.total - t4c_status.free
+            except:  # pylint: disable=bare-except
+                used_licenses = -1
+
+            metric.add_metric([instance.name], used_licenses)
+        yield metric
+
+
+def register() -> None:
+    prometheus_client.REGISTRY.register(UsedT4CLicensesCollector())

backend/tests/settings/conftest.py

@@ -4,8 +4,6 @@
 import pytest
 from sqlalchemy import orm
 
-from capellacollab.settings.modelsources.t4c import crud as t4c_crud
-from capellacollab.settings.modelsources.t4c import models as t4c_models
 from capellacollab.tools import crud as tools_crud
 from capellacollab.tools import models as tools_models
 from capellacollab.users import crud as users_crud
@@ -25,23 +23,3 @@ def fixture_admin_user(
     db: orm.Session, executor_name: str
 ) -> users_models.DatabaseUser:
     return users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
-
-
-@pytest.fixture(name="t4c_instance")
-def fixture_t4c_instance(
-    db: orm.Session,
-    test_tool_version: tools_models.DatabaseVersion,
-) -> t4c_models.DatabaseT4CInstance:
-    server = t4c_models.DatabaseT4CInstance(
-        name="test server",
-        license="lic",
-        host="localhost",
-        usage_api="http://localhost:8086",
-        rest_api="http://localhost:8080/api/v1.0",
-        username="user",
-        password="pass",
-        protocol=t4c_models.Protocol.tcp,
-        version=test_tool_version,
-    )
-
-    return t4c_crud.create_t4c_instance(db, server)

backend/tests/settings/teamforcapella/conftest.py

@@ -0,0 +1,42 @@
+# SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import pytest
+import responses
+from fastapi import status
+from sqlalchemy import orm
+
+from capellacollab.settings.modelsources.t4c import crud as t4c_crud
+from capellacollab.settings.modelsources.t4c import models as t4c_models
+from capellacollab.tools import models as tools_models
+
+
+@pytest.fixture(name="t4c_instance")
+def fixture_t4c_instance(
+    db: orm.Session,
+    test_tool_version: tools_models.DatabaseVersion,
+) -> t4c_models.DatabaseT4CInstance:
+    server = t4c_models.DatabaseT4CInstance(
+        name="test server",
+        license="lic",
+        host="localhost",
+        usage_api="http://localhost:8086",
+        rest_api="http://localhost:8080/api/v1.0",
+        username="user",
+        password="pass",
+        protocol=t4c_models.Protocol.tcp,
+        version=test_tool_version,
+    )
+
+    return t4c_crud.create_t4c_instance(db, server)
+
+
+@pytest.fixture(name="mock_license_server")
+def fixture_mock_license_server():
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            "http://localhost:8086/status/json",
+            status=status.HTTP_200_OK,
+            json={"status": {"used": 1, "free": 19, "total": 20}},
+        )
+        yield rsps

backend/tests/settings/test_t4c_instances.py → backend/tests/settings/teamforcapella/test_t4c_instances.py

@@ -290,20 +290,14 @@ def test_update_t4c_instance_password_empty_string(
     assert updated_t4c_instance.password == expected_password
 
 
-@responses.activate
+@pytest.mark.usefixtures("mock_license_server")
 def test_get_t4c_license_usage(
     client: testclient.TestClient,
     db: orm.Session,
     executor_name: str,
     t4c_instance: t4c_models.DatabaseT4CInstance,
 ):
     users_crud.create_user(db, executor_name, users_models.Role.ADMIN)
-    responses.get(
-        "http://localhost:8086/status/json",
-        status=status.HTTP_200_OK,
-        json={"status": {"used": 1, "free": 19, "total": 20}},
-    )
-
     response = client.get(
         f"/api/v1/settings/modelsources/t4c/{t4c_instance.id}/licenses",
     )

backend/tests/settings/teamforcapella/test_t4c_metrics.py

@@ -0,0 +1,27 @@
+# SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import pytest
+
+from capellacollab.settings.modelsources.t4c import metrics
+
+
+@pytest.mark.usefixtures("mock_license_server", "t4c_instance")
+def test_t4c_license_metrics_collector():
+    collector = metrics.UsedT4CLicensesCollector()
+
+    data = list(collector.collect())
+
+    sample = data[0].samples[0]
+    assert sample.name == "t4c_licenses"
+    assert sample.value == 1
+
+
+def test_t4c_license_metrics_collector_error():
+    collector = metrics.UsedT4CLicensesCollector()
+
+    data = list(collector.collect())
+
+    sample = data[0].samples[0]
+    assert sample.name == "t4c_licenses"
+    assert sample.value == -1