diff --git a/authentik/admin/metrics.py b/authentik/admin/metrics.py
new file mode 100644
index 000000000..4866efe95
--- /dev/null
+++ b/authentik/admin/metrics.py
@@ -0,0 +1,87 @@
+"""Metrics"""
+from contextlib import contextmanager
+from enum import Enum
+from timeit import default_timer
+
+from django.core.cache import cache
+from django_redis.client import DefaultClient
+from redis import Redis
+from redis.exceptions import ResponseError
+from structlog.stdlib import BoundLogger, get_logger
+
+
+class Timeseries(Enum):
+    """An enum of all timeseries"""
+
+    policies_execution_count = "authentik_policies_execution_count"
+    policies_execution_timing = "authentik_policies_execution_timing"
+    flows_execution_count = "authentik_flows_execution_count"
+    flows_stages_execution_count = "authentik_flows_stages_execution_count"
+    flows_stages_execution_timing = "authentik_flows_stages_execution_timing"
+    users_login_count = "authentik_users_login_count"
+
+
+class MetricsManager:
+    """RedisTSDB metrics"""
+
+    supported = False
+    logger: BoundLogger
+    retention: int
+
+    def __init__(self) -> None:
+        self.supported = self.redis_tsdb_supported()
+        self.logger = get_logger()
+        # 1 week in ms
+        self.retention = 7 * 24 * 60 * 60 * 1000
+
+    def redis_tsdb_supported(self):
+        """Check if redis has the timeseries module loaded"""
+        modules = self.get_client().module_list()
+        supported = any(module[b"name"] == b"timeseries" for module in modules)
+        return supported
+
+    def get_client(self) -> Redis:
+        cache_client: DefaultClient = cache.client
+        return cache_client.get_client()
+
+    def make_key(self, ts: Timeseries, *key_parts) -> str:
+        """Construct a unique key"""
+        return "_".join([ts.value] + list(key_parts))
+
+    @contextmanager
+    def inc(self, ts: Timeseries, *key_parts, **labels):
+        """Increase counter with labels"""
+        if not self.supported:
+            yield
+            return
+        client = self.get_client()
+        yield
+        labels["base_ts"] = ts.value
+        client.ts().incrby(
+            self.make_key(ts, *key_parts),
+            1,
+            retention_msecs=self.retention,
+            labels=labels,
+        )
+
+    @contextmanager
+    def observe(self, ts: Timeseries, *key_parts, **labels):
+        """Observe time and save as a sample"""
+        if not self.supported:
+            yield
+            return
+        client = self.get_client()
+        start = default_timer()
+        yield
+        duration = default_timer() - start
+        labels["base_ts"] = ts.value
+        client.ts().add(
+            self.make_key(ts, *key_parts),
+            "*",
+            retention_msecs=self.retention,
+            value=duration,
+            labels=labels,
+        )
+
+
+metrics = MetricsManager()
diff --git a/authentik/api/v3/config.py b/authentik/api/v3/config.py
index 640e03453..57a8110f1 100644
--- a/authentik/api/v3/config.py
+++ b/authentik/api/v3/config.py
@@ -16,7 +16,7 @@ from rest_framework.permissions import AllowAny
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
-
+from authentik.admin.metrics import metrics
 from authentik.core.api.utils import PassiveSerializer
 from authentik.events.geo import GEOIP_READER
 from authentik.lib.config import CONFIG
@@ -29,6 +29,7 @@ class Capabilities(models.TextChoices):
     CAN_GEO_IP = "can_geo_ip"
     CAN_IMPERSONATE = "can_impersonate"
     CAN_DEBUG = "can_debug"
+    CAN_TSDB = "can_tsdb"
     IS_ENTERPRISE = "is_enterprise"
 
 
@@ -71,6 +72,8 @@ class ConfigView(APIView):
             caps.append(Capabilities.CAN_IMPERSONATE)
         if settings.DEBUG:  # pragma: no cover
             caps.append(Capabilities.CAN_DEBUG)
+        if metrics.supported:
+            caps.append(Capabilities.CAN_TSDB)
         if "authentik.enterprise" in settings.INSTALLED_APPS:
             caps.append(Capabilities.IS_ENTERPRISE)
         return caps
diff --git a/authentik/events/signals.py b/authentik/events/signals.py
index d6e65c136..a6dd2e6d4 100644
--- a/authentik/events/signals.py
+++ b/authentik/events/signals.py
@@ -6,6 +6,7 @@ from django.db.models.signals import post_save, pre_delete
 from django.dispatch import receiver
 from django.http import HttpRequest
 
+from authentik.admin.metrics import Timeseries, metrics
 from authentik.core.models import User
 from authentik.core.signals import login_failed, password_changed
 from authentik.events.models import Event, EventAction
@@ -34,7 +35,8 @@ def on_user_logged_in(sender, request: HttpRequest, user: User, **_):
             # Save the login method used
             kwargs[PLAN_CONTEXT_METHOD] = flow_plan.context[PLAN_CONTEXT_METHOD]
             kwargs[PLAN_CONTEXT_METHOD_ARGS] = flow_plan.context.get(PLAN_CONTEXT_METHOD_ARGS, {})
-    event = Event.new(EventAction.LOGIN, **kwargs).from_http(request, user=user)
+    with metrics.inc(Timeseries.users_login_count, str(user.pk)):
+        event = Event.new(EventAction.LOGIN, **kwargs).from_http(request, user=user)
     request.session[SESSION_LOGIN_EVENT] = event
 
 
diff --git a/authentik/flows/views/executor.py b/authentik/flows/views/executor.py
index 1279940b6..63970c238 100644
--- a/authentik/flows/views/executor.py
+++ b/authentik/flows/views/executor.py
@@ -22,6 +22,7 @@ from sentry_sdk.api import set_tag
 from sentry_sdk.hub import Hub
 from structlog.stdlib import BoundLogger, get_logger
 
+from authentik.admin.metrics import Timeseries, metrics
 from authentik.core.models import Application
 from authentik.events.models import Event, EventAction, cleanse_dict
 from authentik.flows.challenge import (
@@ -291,7 +292,13 @@ class FlowExecutorView(APIView):
             with Hub.current.start_span(
                 op="authentik.flow.executor.stage",
                 description=class_to_path(self.current_stage_view.__class__),
-            ) as span:
+            ) as span, metrics.inc(
+                Timeseries.flows_stages_execution_count,
+                str(self.current_stage.pk),
+            ), metrics.observe(
+                Timeseries.flows_stages_execution_timing,
+                str(self.current_stage.pk),
+            ):
                 span.set_data("Method", "GET")
                 span.set_data("authentik Stage", self.current_stage_view)
                 span.set_data("authentik Flow", self.flow.slug)
@@ -335,7 +342,13 @@ class FlowExecutorView(APIView):
             with Hub.current.start_span(
                 op="authentik.flow.executor.stage",
                 description=class_to_path(self.current_stage_view.__class__),
-            ) as span:
+            ) as span, metrics.inc(
+                Timeseries.flows_stages_execution_count,
+                str(self.current_stage.pk),
+            ), metrics.observe(
+                Timeseries.flows_stages_execution_timing,
+                str(self.current_stage.pk),
+            ):
                 span.set_data("Method", "POST")
                 span.set_data("authentik Stage", self.current_stage_view)
                 span.set_data("authentik Flow", self.flow.slug)
@@ -441,6 +454,13 @@ class FlowExecutorView(APIView):
             # It's only deleted on a fresh executions
             # SESSION_KEY_HISTORY,
         ]
+        # Increase the flow execution as this function gets called on successful and
+        # failed flow executions
+        with metrics.inc(
+            Timeseries.flows_execution_count,
+            flow_pk=str(self.flow.pk),
+        ):
+            pass
         self._logger.debug("f(exec): cleaning up")
         for key in keys_to_delete:
             if key in self.request.session:
diff --git a/authentik/policies/models.py b/authentik/policies/models.py
index 0eb353206..7dd8b4d74 100644
--- a/authentik/policies/models.py
+++ b/authentik/policies/models.py
@@ -6,6 +6,7 @@ from django.utils.translation import gettext_lazy as _
 from model_utils.managers import InheritanceManager
 from rest_framework.serializers import BaseSerializer
 
+from authentik.admin.metrics import Timeseries, metrics
 from authentik.lib.models import (
     CreatedUpdatedModel,
     InheritanceAutoManager,
@@ -93,14 +94,18 @@ class PolicyBinding(SerializerModel):
 
     def passes(self, request: PolicyRequest) -> PolicyResult:
         """Check if request passes this PolicyBinding, check policy, group or user"""
-        if self.policy:
-            self.policy: Policy
-            return self.policy.passes(request)
-        if self.group:
-            return PolicyResult(self.group.is_member(request.user))
-        if self.user:
-            return PolicyResult(request.user == self.user)
-        return PolicyResult(False)
+        with metrics.inc(Timeseries.policies_execution_count, str(self.policy_binding_uuid)):
+            if self.policy:
+                self.policy: Policy
+                with metrics.observe(
+                    Timeseries.policies_execution_timing, str(self.policy.pk),
+                ):
+                    return self.policy.passes(request)
+            if self.group:
+                return PolicyResult(self.group.is_member(request.user))
+            if self.user:
+                return PolicyResult(request.user == self.user)
+            return PolicyResult(False)
 
     @property
     def serializer(self) -> type[BaseSerializer]:
diff --git a/schema.yml b/schema.yml
index 857e4d2a1..d12cd16c3 100644
--- a/schema.yml
+++ b/schema.yml
@@ -27786,6 +27786,7 @@ components:
       - can_geo_ip
       - can_impersonate
       - can_debug
+      - can_tsdb
       - is_enterprise
       type: string
       description: |-
@@ -27793,6 +27794,7 @@ components:
         * `can_geo_ip` - Can Geo Ip
         * `can_impersonate` - Can Impersonate
         * `can_debug` - Can Debug
+        * `can_tsdb` - Can Tsdb
         * `is_enterprise` - Is Enterprise
     CaptchaChallenge:
       type: object