feat: product metrics build (#751)

* feat: product metrics build

* feat: add initial posthog product metrics (#744)

* feat: add posthog product metrics

* add rest of metrics

* add tests

* feat: add staging table for metrics (#796)

---------

Co-authored-by: Mohammad Alisafaee <[email protected]>

* feat: get product metrics through k8s watcher (#802)

* feat: product metrics using k8s cache

* feat: only track member adding metrics and add user requesting session state change

* feat: add metrics for seach (#824)

* address comments

---------

Co-authored-by: Mohammad Alisafaee <[email protected]>

Author: Panaetius

DEVELOPING.md

@@ -34,6 +34,7 @@ Within components, there are the following modules:
 * *k8s*: Kubernetes client code
 * *message_queue*: Redis streams messaging code
 * *migrations*: Database migrations
+* *metrics*: Store metrics data in a staging table
 * *namespace*: Code for handling namespaces (user/groups)
 * *platform*: Renku platform configuration code
 * *project*: Code for Project entities

bases/renku_data_services/data_api/app.py

@@ -134,6 +134,7 @@ def register_all_handlers(app: Sanic, config: Config) -> Sanic:
         session_repo=config.session_repo,
         data_connector_repo=config.data_connector_repo,
         project_migration_repo=config.project_migration_repo,
+        metrics=config.metrics,
     )
     project_session_secrets = ProjectSessionSecretBP(
         name="project_session_secrets",
@@ -146,6 +147,7 @@ def register_all_handlers(app: Sanic, config: Config) -> Sanic:
         url_prefix=url_prefix,
         authenticator=config.authenticator,
         group_repo=config.group_repo,
+        metrics=config.metrics,
     )
     session_environments = EnvironmentsBP(
         name="session_environments",
@@ -158,6 +160,7 @@ def register_all_handlers(app: Sanic, config: Config) -> Sanic:
         url_prefix=url_prefix,
         session_repo=config.session_repo,
         authenticator=config.authenticator,
+        metrics=config.metrics,
     )
     builds = (
         BuildsBP(
@@ -214,6 +217,7 @@ def register_all_handlers(app: Sanic, config: Config) -> Sanic:
         data_connector_repo=config.data_connector_repo,
         data_connector_secret_repo=config.data_connector_secret_repo,
         internal_gitlab_authenticator=config.gitlab_authenticator,
+        metrics=config.metrics,
     )
     platform_config = PlatformConfigBP(
         name="platform_config",
@@ -247,13 +251,15 @@ def register_all_handlers(app: Sanic, config: Config) -> Sanic:
         ),
         solr_config=config.solr_config,
         authz=config.authz,
+        metrics=config.metrics,
     )
     data_connectors = DataConnectorsBP(
         name="data_connectors",
         url_prefix=url_prefix,
         data_connector_repo=config.data_connector_repo,
         data_connector_secret_repo=config.data_connector_secret_repo,
         authenticator=config.authenticator,
+        metrics=config.metrics,
     )
     app.blueprint(
         [

bases/renku_data_services/data_tasks/config.py

@@ -10,12 +10,33 @@
 from renku_data_services.solr.solr_client import SolrClientConfig
 
 
+@dataclass
+class PosthogConfig:
+    """Configuration for posthog."""
+
+    enabled: bool
+    api_key: str
+    host: str
+    environment: str
+
+    @classmethod
+    def from_env(cls, prefix: str = "") -> PosthogConfig:
+        """Create posthog config from environment variables."""
+        enabled = os.environ.get(f"{prefix}POSTHOG_ENABLED", "false").lower() == "true"
+        api_key = os.environ.get(f"{prefix}POSTHOG_API_KEY", "")
+        host = os.environ.get(f"{prefix}POSTHOG_HOST", "")
+        environment = os.environ.get(f"{prefix}POSTHOG_ENVIRONMENT", "development")
+
+        return cls(enabled, api_key, host, environment)
+
+
 @dataclass
 class Config:
     """Configuration for data tasks."""
 
     db_config: DBConfig
     solr_config: SolrClientConfig
+    posthog_config: PosthogConfig
     redis_config: RedisConfig
     max_retry_wait_seconds: int
     main_log_interval_seconds: int
@@ -34,6 +55,7 @@ def env(key: str, default: str) -> str:
         max_retry = int(env("MAX_RETRY_WAIT_SECONDS", "120"))
         main_tick = int(env("MAIN_LOG_INTERVAL_SECONDS", "300"))
         solr_config = SolrClientConfig.from_env(prefix)
+        posthog_config = PosthogConfig.from_env(prefix)
         tcp_host = env("TCP_HOST", "127.0.0.1")
         tcp_port = int(env("TCP_PORT", "8001"))
 
@@ -43,6 +65,7 @@ def env(key: str, default: str) -> str:
             max_retry_wait_seconds=max_retry,
             main_log_interval_seconds=main_tick,
             solr_config=solr_config,
+            posthog_config=posthog_config,
             redis_config=redis,
             tcp_host=tcp_host,
             tcp_port=tcp_port,

bases/renku_data_services/data_tasks/task_defs.py

@@ -1,15 +1,19 @@
 """The task definitions in form of coroutines."""
 
 import asyncio
+import logging
 
 import renku_data_services.search.core as search_core
 from renku_data_services.data_tasks.config import Config
 from renku_data_services.data_tasks.taskman import TaskDefininions
 from renku_data_services.message_queue.db import EventRepository
 from renku_data_services.message_queue.redis_queue import RedisQueue
+from renku_data_services.metrics.db import MetricsRepository
 from renku_data_services.search.db import SearchUpdatesRepo
 from renku_data_services.solr.solr_client import DefaultSolrClient
 
+logger = logging.getLogger(__name__)
+
 
 async def update_search(cfg: Config) -> None:
     """Update the SOLR with data from the search staging table."""
@@ -28,6 +32,49 @@ async def send_pending_redis_events(cfg: Config) -> None:
         await asyncio.sleep(1)
 
 
+async def send_metrics_to_posthog(cfg: Config) -> None:
+    """Send pending product metrics to posthog."""
+    from posthog import Posthog
+
+    posthog = Posthog(
+        api_key=cfg.posthog_config.api_key,
+        host=cfg.posthog_config.host,
+        sync_mode=True,
+        super_properties={"environment": cfg.posthog_config.environment},
+    )
+    repo = MetricsRepository(cfg.db_config.async_session_maker)
+
+    while True:
+        try:
+            metrics = repo.get_unprocessed_metrics()
+
+            processed_ids = []
+            async for metric in metrics:
+                try:
+                    posthog.capture(
+                        distinct_id=metric.anonymous_user_id,
+                        timestamp=metric.timestamp,
+                        event=metric.event,
+                        properties=metric.metadata_ or {},
+                        # This is sent to avoid duplicate events if multiple instances of data service are running.
+                        # Posthog deduplicates events with the same timestamp, distinct_id, event, and uuid fields:
+                        # https://github.com/PostHog/posthog/issues/17211#issuecomment-1723136534
+                        uuid=metric.id.to_uuid4(),
+                    )
+                except Exception as e:
+                    logger.error(f"Failed to process metrics event {metric.id}: {e}")
+                else:
+                    processed_ids.append(metric.id)
+
+            await repo.delete_processed_metrics(processed_ids)
+        except (asyncio.CancelledError, KeyboardInterrupt) as e:
+            logger.warning(f"Exiting: {e}")
+            return
+        else:
+            # NOTE: Sleep 10 seconds between processing cycles
+            await asyncio.sleep(10)
+
+
 def all_tasks(cfg: Config) -> TaskDefininions:
     """A dict of task factories to be managed in main."""
     # Impl. note: We pass the entire config to the coroutines, because
@@ -41,5 +88,6 @@ def all_tasks(cfg: Config) -> TaskDefininions:
         {
             "update_search": lambda: update_search(cfg),
             "send_pending_events": lambda: send_pending_redis_events(cfg),
+            "send_product_metrics": lambda: send_metrics_to_posthog(cfg),
         }
     )

bases/renku_data_services/k8s_cache/config.py

@@ -1,12 +1,17 @@
 """K8s cache config."""
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Self
 
 from kubernetes.client.api_client import os
 
+from renku_data_services.crc.db import ResourcePoolRepository
 from renku_data_services.db_config.config import DBConfig
+from renku_data_services.k8s.clients import DummyCoreClient, DummySchedulingClient
+from renku_data_services.k8s.quota import QuotaRepository
 from renku_data_services.k8s_watcher.db import K8sDbCache
+from renku_data_services.metrics.core import StagingMetricsService
+from renku_data_services.metrics.db import MetricsRepository
 
 
 @dataclass
@@ -20,13 +25,54 @@ def from_env(cls) -> Self:
         return cls(renku_namespace=os.environ.get("KUBERNETES_NAMESPACE", "default"))
 
 
+@dataclass
+class _MetricsConfig:
+    """Configuration for metrics."""
+
+    enabled: bool
+
+    @classmethod
+    def from_env(cls, prefix: str = "") -> "_MetricsConfig":
+        """Create metrics config from environment variables."""
+        enabled = os.environ.get(f"{prefix}POSTHOG_ENABLED", "false").lower() == "true"
+        return cls(enabled)
+
+
 @dataclass
 class Config:
     """K8s cache config."""
 
     db: DBConfig
     k8s: _K8sConfig
+    metrics_config: _MetricsConfig
+    quota_repo: QuotaRepository
     _k8s_cache: K8sDbCache | None = None
+    _metrics_repo: MetricsRepository | None = field(default=None, repr=False, init=False)
+    _metrics: StagingMetricsService | None = field(default=None, repr=False, init=False)
+    _rp_repo: ResourcePoolRepository | None = field(default=None, repr=False, init=False)
+
+    @property
+    def metrics_repo(self) -> MetricsRepository:
+        """The DB adapter for metrics."""
+        if not self._metrics_repo:
+            self._metrics_repo = MetricsRepository(session_maker=self.db.async_session_maker)
+        return self._metrics_repo
+
+    @property
+    def metrics(self) -> StagingMetricsService:
+        """The metrics service interface."""
+        if not self._metrics:
+            self._metrics = StagingMetricsService(enabled=self.metrics_config.enabled, metrics_repo=self.metrics_repo)
+        return self._metrics
+
+    @property
+    def rp_repo(self) -> ResourcePoolRepository:
+        """The resource pool repository."""
+        if not self._rp_repo:
+            self._rp_repo = ResourcePoolRepository(
+                session_maker=self.db.async_session_maker, quotas_repo=self.quota_repo
+            )
+        return self._rp_repo
 
     @property
     def k8s_cache(self) -> K8sDbCache:
@@ -42,5 +88,13 @@ def from_env(cls, prefix: str = "") -> "Config":
         """Create a config from environment variables."""
         db = DBConfig.from_env(prefix)
         k8s_config = _K8sConfig.from_env()
+        metrics_config = _MetricsConfig.from_env(prefix)
+
+        # NOTE: We only need the QuotaRepository to instantiate the ResourcePoolRepository which is used to get
+        # the resource class and pool information for metrics. We don't need quota information for metrics at all
+        # so we use the dummy client for quotas here as we don't actually access k8s, just the db.
+        quota_repo = QuotaRepository(
+            DummyCoreClient({}, {}), DummySchedulingClient({}), namespace=k8s_config.renku_namespace
+        )
 
-        return cls(db=db, k8s=k8s_config)
+        return cls(db=db, k8s=k8s_config, metrics_config=metrics_config, quota_repo=quota_repo)

bases/renku_data_services/k8s_cache/main.py

@@ -21,7 +21,7 @@ async def main() -> None:
     clusters = [Cluster(id=ClusterId("renkulab"), namespace=config.k8s.renku_namespace, api=kr8s_api)]
 
     watcher = K8sWatcher(
-        handler=k8s_object_handler(config.k8s_cache),
+        handler=k8s_object_handler(config.k8s_cache, config.metrics, rp_repo=config.rp_repo),
         clusters={c.id: c for c in clusters},
         kinds=[AMALTHEA_SESSION_KIND, JUPYTER_SESSION_KIND],
     )

components/renku_data_services/app_config/config.py

@@ -61,6 +61,8 @@
 from renku_data_services.message_queue.db import EventRepository, ReprovisioningRepository
 from renku_data_services.message_queue.interface import IMessageQueue
 from renku_data_services.message_queue.redis_queue import RedisQueue
+from renku_data_services.metrics.core import StagingMetricsService
+from renku_data_services.metrics.db import MetricsRepository
 from renku_data_services.namespace.db import GroupRepository
 from renku_data_services.notebooks.config import NotebooksConfig
 from renku_data_services.platform.db import PlatformRepository
@@ -133,6 +135,20 @@ def from_env(cls, prefix: str = "") -> "SentryConfig":
         return cls(enabled, dsn=dsn, environment=environment, sample_rate=sample_rate)
 
 
+@dataclass
+class PosthogConfig:
+    """Configuration for posthog."""
+
+    enabled: bool
+
+    @classmethod
+    def from_env(cls, prefix: str = "") -> "PosthogConfig":
+        """Create posthog config from environment variables."""
+        enabled = os.environ.get(f"{prefix}POSTHOG_ENABLED", "false").lower() == "true"
+
+        return cls(enabled)
+
+
 @dataclass
 class TrustedProxiesConfig:
     """Configuration for trusted reverse proxies."""
@@ -261,6 +277,7 @@ class Config:
     gitlab_url: str | None
     nb_config: NotebooksConfig
     builds_config: BuildsConfig
+    posthog: PosthogConfig
 
     secrets_service_public_key: rsa.RSAPublicKey
     """The public key of the secrets service, used to encrypt user secrets that only it can decrypt."""
@@ -300,6 +317,8 @@ class Config:
     _data_connector_repo: DataConnectorRepository | None = field(default=None, repr=False, init=False)
     _data_connector_secret_repo: DataConnectorSecretRepository | None = field(default=None, repr=False, init=False)
     _cluster_repo: ClusterRepository | None = field(default=None, repr=False, init=False)
+    _metrics_repo: MetricsRepository | None = field(default=None, repr=False, init=False)
+    _metrics: StagingMetricsService | None = field(default=None, repr=False, init=False)
 
     @staticmethod
     @functools.cache
@@ -352,7 +371,7 @@ def __post_init__(self) -> None:
 
     @property
     def user_repo(self) -> UserRepository:
-        """The DB adapter for users of resoure pools and classes."""
+        """The DB adapter for users of resource pools and classes."""
         if not self._user_repo:
             self._user_repo = UserRepository(
                 session_maker=self.db.async_session_maker, quotas_repo=self.quota_repo, user_repo=self.kc_user_repo
@@ -602,6 +621,20 @@ def cluster_repo(self) -> ClusterRepository:
 
         return self._cluster_repo
 
+    @property
+    def metrics_repo(self) -> MetricsRepository:
+        """The DB adapter for metrics."""
+        if not self._metrics_repo:
+            self._metrics_repo = MetricsRepository(session_maker=self.db.async_session_maker)
+        return self._metrics_repo
+
+    @property
+    def metrics(self) -> StagingMetricsService:
+        """The metrics service interface."""
+        if not self._metrics:
+            self._metrics = StagingMetricsService(enabled=self.posthog.enabled, metrics_repo=self.metrics_repo)
+        return self._metrics
+
     @classmethod
     def from_env(cls, prefix: str = "") -> "Config":
         """Create a config from environment variables."""
@@ -644,7 +677,7 @@ def from_env(cls, prefix: str = "") -> "Config":
                 UnsavedUserInfo(id="user1", first_name="user1", last_name="doe", email="[email protected]"),
                 UnsavedUserInfo(id="user2", first_name="user2", last_name="doe", email="[email protected]"),
             ]
-            kc_api = DummyKeycloakAPI(users=[i._to_keycloak_dict() for i in dummy_users])
+            kc_api = DummyKeycloakAPI(users=[i.to_keycloak_dict() for i in dummy_users])
             redis = RedisConfig.fake()
             gitlab_url = None
         else:
@@ -698,6 +731,7 @@ def from_env(cls, prefix: str = "") -> "Config":
         message_queue = RedisQueue(redis)
         nb_config = NotebooksConfig.from_env(db)
         builds_config = BuildsConfig.from_env(prefix, k8s_namespace)
+        posthog = PosthogConfig.from_env(prefix)
 
         return cls(
             version=version,
@@ -721,4 +755,5 @@ def from_env(cls, prefix: str = "") -> "Config":
             gitlab_url=gitlab_url,
             nb_config=nb_config,
             builds_config=builds_config,
+            posthog=posthog,
         )

components/renku_data_services/base_models/core.py

@@ -77,6 +77,7 @@ class ServiceAdminId(StrEnum):
 
     migrations = "migrations"
     secrets_rotation = "secrets_rotation"
+    k8s_watcher = "k8s_watcher"
 
 
 @dataclass(kw_only=True, frozen=True)