Metrics

dbos/_conductor/conductor.py

@@ -117,6 +117,7 @@ def run(self) -> None:
                                 executor_id=GlobalParams.executor_id,
                                 application_version=GlobalParams.app_version,
                                 hostname=socket.gethostname(),
+                                language="python",
                             )
                             websocket.send(info_response.to_json())
                             self.dbos.logger.info("Connected to DBOS conductor")
@@ -397,6 +398,36 @@ def run(self) -> None:
                                 error_message=error_message,
                             )
                             websocket.send(retention_response.to_json())
+                        elif msg_type == p.MessageType.GET_METRICS:
+                            get_metrics_message = p.GetMetricsRequest.from_json(message)
+                            self.dbos.logger.debug(
+                                f"Received metrics request for time range {get_metrics_message.start_time} to {get_metrics_message.end_time}"
+                            )
+                            metrics_data = []
+                            try:
+                                sys_metrics = self.dbos._sys_db.get_metrics(
+                                    get_metrics_message.start_time,
+                                    get_metrics_message.end_time,
+                                )
+                                metrics_data = [
+                                    p.MetricData(
+                                        metric_type=m["metric_type"],
+                                        metric_name=m["metric_name"],
+                                        count=m["count"],
+                                    )
+                                    for m in sys_metrics
+                                ]
+                            except Exception as e:
+                                error_message = f"Exception encountered when getting metrics: {traceback.format_exc()}"
+                                self.dbos.logger.error(error_message)
+
+                            get_metrics_response = p.GetMetricsResponse(
+                                type=p.MessageType.GET_METRICS,
+                                request_id=base_message.request_id,
+                                metrics=metrics_data,
+                                error_message=error_message,
+                            )
+                            websocket.send(get_metrics_response.to_json())
                         else:
                             self.dbos.logger.warning(
                                 f"Unexpected message type: {msg_type}"

dbos/_conductor/protocol.py

@@ -19,6 +19,7 @@ class MessageType(str, Enum):
     LIST_STEPS = "list_steps"
     FORK_WORKFLOW = "fork_workflow"
     RETENTION = "retention"
+    GET_METRICS = "get_metrics"
 
 
 T = TypeVar("T", bound="BaseMessage")
@@ -63,6 +64,7 @@ class ExecutorInfoResponse(BaseMessage):
     executor_id: str
     application_version: str
     hostname: Optional[str]
+    language: Optional[str]
     error_message: Optional[str] = None
 
 
@@ -339,3 +341,22 @@ class RetentionRequest(BaseMessage):
 class RetentionResponse(BaseMessage):
     success: bool
     error_message: Optional[str] = None
+
+
+@dataclass
+class GetMetricsRequest(BaseMessage):
+    start_time: str  # ISO 8601
+    end_time: str  # ISO 8601
+
+
+@dataclass
+class MetricData:
+    metric_type: str
+    metric_name: str
+    count: int
+
+
+@dataclass
+class GetMetricsResponse(BaseMessage):
+    metrics: List[MetricData]
+    error_message: Optional[str] = None

dbos/_sys_db.py

@@ -158,6 +158,16 @@ class WorkflowStatusInternal(TypedDict):
     forked_from: Optional[str]
 
 
+class MetricData(TypedDict):
+    """
+    Metrics data for workflows and steps within a time range.
+    """
+
+    metric_type: str  # Type of metric: "workflow" or "step"
+    metric_name: str  # Name of the workflow or step
+    count: int  # Count of occurrences
+
+
 class EnqueueOptionsInternal(TypedDict):
     # Unique ID for deduplication on a queue
     deduplication_id: Optional[str]
@@ -2267,3 +2277,76 @@ def garbage_collect(
             return cutoff_epoch_timestamp_ms, [
                 row[0] for row in pending_enqueued_result
             ]
+
+    def get_metrics(self, start_time: str, end_time: str) -> List[MetricData]:
+        """
+        Retrieve metrics for workflows and steps within a time range.
+
+        Args:
+            start_time: ISO 8601 formatted start time
+            end_time: ISO 8601 formatted end time
+        """
+        # Convert ISO 8601 times to epoch milliseconds
+        start_epoch_ms = int(
+            datetime.datetime.fromisoformat(start_time).timestamp() * 1000
+        )
+        end_epoch_ms = int(datetime.datetime.fromisoformat(end_time).timestamp() * 1000)
+
+        metrics: List[MetricData] = []
+
+        with self.engine.begin() as c:
+            # Query workflow metrics
+            workflow_query = (
+                sa.select(
+                    SystemSchema.workflow_status.c.name,
+                    func.count(SystemSchema.workflow_status.c.workflow_uuid).label(
+                        "count"
+                    ),
+                )
+                .where(
+                    sa.and_(
+                        SystemSchema.workflow_status.c.created_at >= start_epoch_ms,
+                        SystemSchema.workflow_status.c.created_at < end_epoch_ms,
+                    )
+                )
+                .group_by(SystemSchema.workflow_status.c.name)
+            )
+
+            workflow_results = c.execute(workflow_query).fetchall()
+            for row in workflow_results:
+                metrics.append(
+                    MetricData(
+                        metric_type="workflow",
+                        metric_name=row[0],
+                        count=row[1],
+                    )
+                )
+
+            # Query step metrics
+            step_query = (
+                sa.select(
+                    SystemSchema.operation_outputs.c.function_name,
+                    func.count().label("count"),
+                )
+                .where(
+                    sa.and_(
+                        SystemSchema.operation_outputs.c.started_at_epoch_ms
+                        >= start_epoch_ms,
+                        SystemSchema.operation_outputs.c.started_at_epoch_ms
+                        < end_epoch_ms,
+                    )
+                )
+                .group_by(SystemSchema.operation_outputs.c.function_name)
+            )
+
+            step_results = c.execute(step_query).fetchall()
+            for row in step_results:
+                metrics.append(
+                    MetricData(
+                        metric_type="step",
+                        metric_name=row[0],
+                        count=row[1],
+                    )
+                )
+
+        return metrics

dbos/cli/cli.py

@@ -71,7 +71,7 @@ def _get_db_url(
                 _app_db_name = _app_name_to_db_name(config["name"])
                 # Fallback on the same defaults than the DBOS library
                 default_url = f"sqlite:///{_app_db_name}.sqlite"
-                return default_url, default_url
+                return default_url, None
         except (FileNotFoundError, OSError):
             typer.echo(
                 f"Error: Missing database URL: please set it using CLI flags or your dbos-config.yaml file.",

tests/test_metrics.py

@@ -0,0 +1,52 @@
+import datetime
+
+from dbos import DBOS
+
+
+def test_get_metrics(dbos: DBOS, skip_with_sqlite_imprecise_time: None) -> None:
+    """Test the get_metrics method returns correct workflow and step counts."""
+
+    @DBOS.workflow()
+    def test_workflow_a() -> str:
+        test_step_x()
+        test_step_x()
+        return "a"
+
+    @DBOS.workflow()
+    def test_workflow_b() -> str:
+        test_step_y()
+        return "b"
+
+    @DBOS.step()
+    def test_step_x() -> str:
+        return "x"
+
+    @DBOS.step()
+    def test_step_y() -> str:
+        return "y"
+
+    # Record start time before creating workflows
+    start_time = datetime.datetime.now(datetime.timezone.utc)
+
+    # Execute workflows to create metrics data
+    test_workflow_a()
+    test_workflow_a()
+    test_workflow_b()
+
+    # Record end time after creating workflows
+    end_time = datetime.datetime.now(datetime.timezone.utc)
+
+    # Query metrics
+    metrics = dbos._sys_db.get_metrics(start_time.isoformat(), end_time.isoformat())
+    assert len(metrics) == 4
+
+    # Convert to dict for easier assertion
+    metrics_dict = {(m["metric_type"], m["metric_name"]): m["count"] for m in metrics}
+
+    # Verify workflow counts
+    assert metrics_dict[("workflow", test_workflow_a.__qualname__)] == 2
+    assert metrics_dict[("workflow", test_workflow_b.__qualname__)] == 1
+
+    # Verify step counts
+    assert metrics_dict[("step", test_step_x.__qualname__)] == 4
+    assert metrics_dict[("step", test_step_y.__qualname__)] == 1