enterprise: add prometheus metrics for license usage and expiry (#17606)

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

authentik/enterprise/apps.py

@@ -1,11 +1,21 @@
 """Enterprise app config"""
 
 from django.conf import settings
+from prometheus_client import Gauge
 
 from authentik.blueprints.apps import ManagedAppConfig
 from authentik.lib.utils.time import fqdn_rand
 from authentik.tasks.schedules.common import ScheduleSpec
 
+GAUGE_LICENSE_USAGE = Gauge(
+    "authentik_enterprise_license_usage",
+    "Enterprise license usage (percentage per user type).",
+    ["user_type"],
+)
+GAUGE_LICENSE_EXPIRY = Gauge(
+    "authentik_enterprise_license_expiry_seconds", "Duration until license expires, in seconds."
+)
+
 
 class EnterpriseConfig(ManagedAppConfig):
     """Base app config for all enterprise apps"""

authentik/enterprise/license.py

@@ -217,7 +217,7 @@ class LicenseKey:
     def summary(self) -> LicenseSummary:
         """Summary of license status"""
         status = self.status()
-        latest_valid = datetime.fromtimestamp(self.exp)
+        latest_valid = datetime.fromtimestamp(self.exp).replace(tzinfo=UTC)
         return LicenseSummary(
             latest_valid=latest_valid,
             internal_users=self.internal_users,

authentik/enterprise/signals.py

@@ -5,14 +5,37 @@ from datetime import datetime
 from django.core.cache import cache
 from django.db.models.signals import post_delete, post_save, pre_save
 from django.dispatch import receiver
-from django.utils.timezone import get_current_timezone
+from django.utils.timezone import get_current_timezone, now
 
-from authentik.enterprise.license import CACHE_KEY_ENTERPRISE_LICENSE
-from authentik.enterprise.models import License
+from authentik.enterprise.apps import GAUGE_LICENSE_EXPIRY, GAUGE_LICENSE_USAGE
+from authentik.enterprise.license import CACHE_KEY_ENTERPRISE_LICENSE, LicenseKey
+from authentik.enterprise.models import License, LicenseUsageStatus
 from authentik.enterprise.tasks import enterprise_update_usage
+from authentik.root.monitoring import monitoring_set
 from authentik.tasks.schedules.models import Schedule
 
 
+@receiver(monitoring_set)
+def monitoring_set_enterprise(sender, **kwargs):
+    """set enterprise gauges"""
+    summary = LicenseKey.cached_summary()
+    if summary.status == LicenseUsageStatus.UNLICENSED:
+        return
+    percentage_internal = (
+        0
+        if summary.internal_users <= 0
+        else LicenseKey.get_internal_user_count() / (summary.internal_users / 100)
+    )
+    percentage_external = (
+        0
+        if summary.external_users <= 0
+        else LicenseKey.get_external_user_count() / (summary.external_users / 100)
+    )
+    GAUGE_LICENSE_USAGE.labels(user_type="internal").set(percentage_internal)
+    GAUGE_LICENSE_USAGE.labels(user_type="external").set(percentage_external)
+    GAUGE_LICENSE_EXPIRY.set((summary.latest_valid - now()).total_seconds())
+
+
 @receiver(pre_save, sender=License)
 def pre_save_license(sender: type[License], instance: License, **_):
     """Extract data from license jwt and save it into model"""

authentik/enterprise/tests/test_metrics.py

@@ -0,0 +1,49 @@
+"""Enterprise metrics tests"""
+
+from unittest.mock import MagicMock, patch
+
+from django.test import TestCase
+from prometheus_client import REGISTRY
+
+from authentik.core.models import User
+from authentik.core.tests.utils import create_test_user
+from authentik.enterprise.license import LicenseKey
+from authentik.enterprise.models import License
+from authentik.enterprise.tests.test_license import expiry_valid
+from authentik.lib.generators import generate_id
+from authentik.root.monitoring import monitoring_set
+
+
+class TestEnterpriseMetrics(TestCase):
+    """Enterprise metrics tests"""
+
+    @patch(
+        "authentik.enterprise.license.LicenseKey.validate",
+        MagicMock(
+            return_value=LicenseKey(
+                aud="",
+                exp=expiry_valid,
+                name=generate_id(),
+                internal_users=100,
+                external_users=100,
+            )
+        ),
+    )
+    def test_usage_empty(self):
+        """Test usage (no users)"""
+        License.objects.create(key=generate_id())
+        User.objects.all().delete()
+        create_test_user()
+        monitoring_set.send_robust(self)
+        self.assertEqual(
+            REGISTRY.get_sample_value(
+                "authentik_enterprise_license_usage", {"user_type": "internal"}
+            ),
+            1.0,
+        )
+        self.assertEqual(
+            REGISTRY.get_sample_value(
+                "authentik_enterprise_license_usage", {"user_type": "external"}
+            ),
+            0,
+        )