Metering — Technical Specification¶
Scope: Per-feature usage tracking and cost calculation for 5 marketplace features: Data Labeling, ETL, Data Cleaning, JupyterHub Notebooks, and Video Streaming.
1. Architecture¶
Metering is marketplace-side (MongoDB). The marketplace records usage events at feature lifecycle points (job completion, session start/stop) and computes cost from resource consumption × duration × token counts.
The PNU engine is modified only to capture LLM token counts it currently discards — the vLLM response's usage dict is thrown away in llm_service._invoke() at line 135. No other PNU changes are needed for metering beyond surfacing those counts on the job status API.
┌─ Marketplace Backend (MongoDB) ───────────────────────────────────────┐
│ │
│ data_pipeline_usage_collection (NEW) │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ user_id, feature, resource_id, operation, event, status, │ │
│ │ prompt_tokens, completion_tokens, total_tokens, │ │
│ │ duration_seconds, gpu_count, gpu_profile, cpu_cores, │ │
│ │ memory_gb, storage_gb, cost_usd, │ │
│ │ started_at, ended_at, recorded_at │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ services/usage_tracker.py │
│ ├── track_inference_usage() (existing — LLM API tokens) │
│ ├── track_vdb_operation() (existing — VDB read/write) │
│ ├── record_data_pipeline_usage() (NEW — ETL / cleaning / labeling) │
│ ├── record_notebook_usage() (NEW — JupyterHub) │
│ └── record_streaming_usage() (NEW — video streaming) │
│ │
│ routes/dashboard.py │
│ └── Aggregates all collections → GET /api/dashboard/general │
│ │
│ utils/load_helper.py (NEW rates) │
│ └── 8 new hourly rates + reuses INFERENCE_TOKEN_RATE │
└───────────────────────────────────────────────────────────────────────┘
▲
│ PNU API responses include usage data (tokens, timestamps, resources)
┌─ PNU Engine (PostgreSQL + K8s + vLLM) ────────────────────────────────┐
│ │
│ services/llm_service.py │
│ └── _invoke() returns LLMResult(content, usage) ← MODIFY │
│ ↑ vLLM response already contains usage dict │
│ │
│ models/data_pipeline/data_processing.py │
│ └── DataProcessingJob ← ADD: prompt_tokens, completion_tokens, │
│ total_tokens columns │
│ │
│ services/data_pipeline/gpu_job_service.py │
│ └── _finalize_job() ← READ usage from result JSON, store on row │
│ │
│ workers/job_runner.py │
│ └── Result JSON uploaded to MinIO now includes usage dict │
│ │
│ api/v1/data_pipeline/jobs.py │
│ └── GET /jobs/{id} response includes token fields (automatic) │
└───────────────────────────────────────────────────────────────────────┘
2. Pricing Model¶
Each feature is priced by actual resource consumption × time × tokens.
| Feature | Tokens | GPU | CPU | Memory | Storage | Duration Source |
|---|---|---|---|---|---|---|
| Data Labeling (AI assist) | INFERENCE_TOKEN_RATE per token |
DATA_PIPELINE_GPU_HOURLY_RATE |
DATA_PIPELINE_CPU_HOURLY_RATE |
— | — | completed_at - started_at |
| ETL | INFERENCE_TOKEN_RATE per token |
DATA_PIPELINE_GPU_HOURLY_RATE |
DATA_PIPELINE_CPU_HOURLY_RATE |
— | — | completed_at - started_at |
| Data Cleaning | INFERENCE_TOKEN_RATE per token |
DATA_PIPELINE_GPU_HOURLY_RATE |
DATA_PIPELINE_CPU_HOURLY_RATE |
— | — | completed_at - started_at |
| JupyterHub | — | NOTEBOOK_GPU_HOURLY_RATE |
NOTEBOOK_CPU_HOURLY_RATE |
NOTEBOOK_MEMORY_HOURLY_RATE |
NOTEBOOK_STORAGE_HOURLY_RATE |
stopped_at - created_at |
| Video Streaming | — | STREAMING_GPU_HOURLY_RATE |
STREAMING_CPU_HOURLY_RATE |
— | — | completed_at - started_at |
Design decisions:
- Token cost reuses
INFERENCE_TOKEN_RATE($0.000002/token) — same LLM inference cost regardless of which feature triggered the call. One env var to tune. - GPU vs CPU: When
gpu_count > 0, cost uses the GPU hourly rate. Otherwise, falls back to CPU hourly rate. This avoids charging both GPU and CPU for the same time period. - Notebooks are the most granular — they charge for GPU + CPU + memory + storage simultaneously because a notebook pod reserves all four resources for its entire lifetime.
- Streaming charges GPU + CPU — the video streamer deployment reserves GPU (for NVENC/NVDEC) and CPU.
- Failed/cancelled jobs are recorded with
cost_usd = 0.0for accountability without charge.
3. Rate Constants¶
All rates are defined as constants in backend/utils/load_helper.py.
| Constant | Value | Unit |
|---|---|---|
INFERENCE_TOKEN_RATE |
0.000002 |
USD per token (existing) |
DATA_PIPELINE_GPU_HOURLY_RATE |
1.50 |
USD per GPU-hour |
DATA_PIPELINE_CPU_HOURLY_RATE |
0.05 |
USD per core-hour |
NOTEBOOK_GPU_HOURLY_RATE |
1.50 |
USD per GPU-hour |
NOTEBOOK_CPU_HOURLY_RATE |
0.05 |
USD per core-hour |
NOTEBOOK_MEMORY_HOURLY_RATE |
0.002 |
USD per GB-hour |
NOTEBOOK_STORAGE_HOURLY_RATE |
0.0001 |
USD per GB-hour |
STREAMING_GPU_HOURLY_RATE |
2.00 |
USD per GPU-hour |
STREAMING_CPU_HOURLY_RATE |
0.05 |
USD per core-hour |
4. Cost Formulas¶
4.1 Async Jobs (ETL, Data Cleaning, Data Labeling)¶
duration_hours = (completed_at - started_at) / 3600
token_cost = total_tokens × INFERENCE_TOKEN_RATE
resource_cost = gpu_count > 0
? gpu_count × duration_hours × DATA_PIPELINE_GPU_HOURLY_RATE
: cpu_cores × duration_hours × DATA_PIPELINE_CPU_HOURLY_RATE
total_cost = token_cost + resource_cost
Only ai_assist_bulk (labeling), extract/transform/load (ETL), and generate_schema/perform_operations (cleaning) go through async jobs and are metered here. Single-item AI assist is synchronous and goes through track_inference_usage() separately.
4.2 JupyterHub Notebooks¶
duration_hours = (stopped_at - created_at) / 3600
total_cost = (gpu_count × NOTEBOOK_GPU_HOURLY_RATE
+ cpu_cores × NOTEBOOK_CPU_HOURLY_RATE
+ memory_gb × NOTEBOOK_MEMORY_HOURLY_RATE
+ storage_gb × NOTEBOOK_STORAGE_HOURLY_RATE) × duration_hours
All four resource dimensions are charged because a notebook pod reserves GPU, CPU, memory, and storage for its entire lifetime.
Events: start (session created), stop (session stopped), delete (session hard-deleted — resources freed and row purged from PNU). The delete event uses the same cost formula but sets status: "deleted" on the MongoDB doc. A guard prevents overwriting an already-deleted doc (race between the delete route and the PNU webhook).
4.3 Video Streaming¶
duration_hours = (completed_at - started_at) / 3600
total_cost = (gpu_count × STREAMING_GPU_HOURLY_RATE
+ cpu_cores × STREAMING_CPU_HOURLY_RATE) × duration_hours
Streaming uses a higher GPU hourly rate ($2.00 vs $1.50) because video encoding is GPU-intensive with sustained NVENC/NVDEC utilization.
5. MongoDB Schema¶
Collection: data_pipeline_usage (in the main marketplace database, same DB as llm_dashboard_data)
{
"_id": ObjectId,
"user_id": "string (marketplace user ObjectId string)",
"feature": "etl | cleaning | labeling | notebook | streaming",
"resource_id": "string (job_id for async jobs, session_id for notebooks/streaming)",
"operation": "extract | transform | load | generate_schema | perform_operations | ai_assist_bulk | notebook | video_stream",
"event": "start | stop | completed | upscale | delete",
"status": "running | completed | failed | cancelled | stopped | upscaled | deleted",
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
"duration_seconds": 0,
"gpu_count": 0,
"gpu_profile": "string (e.g. '1g.35gb') or empty",
"cpu_cores": 0,
"memory_gb": 0.0,
"storage_gb": 0.0,
"cost_usd": 0.0,
"started_at": ISODate,
"ended_at": ISODate,
"recorded_at": ISODate
}
Idempotency¶
For async jobs (ETL, cleaning, labeling), the resource_id field holds the PNU job_id. Before inserting, record_data_pipeline_usage() queries find_one({"resource_id": job_id}). If a document exists, it returns early. This prevents double-counting whether completion is detected via SSE or polling.
For notebooks and streaming, the resource_id holds the session_id. The "start" event inserts a document; the "stop" event finds and updates that same document by resource_id. The "delete" event (notebooks only) does the same but sets status: "deleted". No duplicate risk because start/stop/delete are distinct user actions. The delete event is idempotent — a second call on an already-deleted doc returns early.
6. Data Flow¶
6.1 Async Jobs (ETL, Data Cleaning, Data Labeling)¶
Frontend Marketplace PNU Engine
│ │ │
│── POST /etl/extract ──────▶│ │
│ │── POST /pipelines/jobs ─▶│
│ │◀── {job_id, status} ─────│
│◀── {job_id, status} ──────│ │
│ │ │
│── GET /jobs/{id}/events (SSE) ────────────────────▶│ │
│ │── GET /pipelines/jobs/{id}/events ▶│
│ │ (httpx-sse proxy to PNU SSE) │
│ │ │
│ │ ◀── SSE: status transition events │
│ │ (via Redis pub/sub) │
│ │ │
│ │ if status == "completed": │
│ │ record_data_pipeline_usage()
│ │ (idempotent by job_id) │
│ │ │
│◀── SSE: {status: "completed", │ │
│ ...usage fields} ─────│ │
│ │
│ (fallback: GET /jobs/{id}/status every 3s │
│ if SSE fails 3×) │
Key points:
- Metering fires when a terminal status event streams through the SSE proxy (jobs_routes.stream_job_events() / gpu_training_routes.stream_job_events()), or when the REST status endpoint (get_job_status() / job_status()) detects a terminal status during polling fallback.
- Both SSE and REST paths call record_data_pipeline_usage() — the call is idempotent (keyed by resource_id == job_id), so SSE + polling fallback do not double-count.
- The PNU response includes prompt_tokens, completion_tokens, total_tokens (added in Phase 2).
- If PNU is not yet updated (Phase 1–2 not deployed), token fields default to 0 — cost is still calculated from resources × duration.
6.2 JupyterHub Notebooks¶
Frontend Marketplace PNU Engine
│ │ │
│── POST /api/tools/create ─▶│ │
│ │── POST /runtime/notebooks ▶│
│ │◀── {id, status, ...} ─────│
│ │ │
│ │ record_notebook_usage( │
│ │ event="start", │
│ │ gpu_count, cpu_cores, │
│ │ memory_gb, storage_gb) │
│ │ │
│◀── {session_id, ...} ─────│ │
│ │ │
│ ... user works in notebook ... │
│ │ │
│── POST /api/tools/ │ │
│ notebooks/{id}/stop ────▶│ │
│ │── POST /runtime/notebooks/│
│ │ {id}/stop ───────────▶│
│ │◀── {stopped_at, │
│ │ created_at, │
│ │ gpu_count, ...} ─────│
│ │ │
│ │ record_notebook_usage( │
│ │ event="stop", │
│ │ created_at, stopped_at, │
│ │ gpu_count, cpu_cores, │
│ │ memory_gb, storage_gb) │
│ │ │
│◀── {status: "stopped"} ───│ │
Key points:
- POST /api/tools/notebooks/{session_id}/stop is a new marketplace route (Phase 4, Task 4.3). The marketplace previously had no stop endpoint for notebooks.
- DELETE /api/tools/notebooks/{session_id} is the hard-delete route — proxies to PNU DELETE /runtime/notebooks/{id}?hard=true, then calls record_notebook_usage(event="delete"). Available for all session statuses. If the notebook is running, PNU stops it first, emits final billing, then purges the row.
- The start event uses resource values from the request payload (what the user requested). The stop/delete event uses values from the PNU response (what was actually allocated).
- Duration is computed from created_at (stored at start event) and stopped_at (from PNU stop/delete response).
- Upscaling (POST /api/tools/notebooks/{session_id}/upscaling): Closes the existing "start" doc by computing duration + cost at OLD specs (status → upscaled), then inserts a new "start" doc with the NEW specs. PNU also notifies via webhook POST /api/internal/notebooks/{id}/upscaled.
- Webhook (auto-stop/delete): When PNU's sweeper stops or deletes a session, PNU fires POST /api/internal/notebooks/{id}/stopped with an action field ("stopped" or "delete") so the marketplace can record the correct event type.
6.3 Video Streaming¶
Frontend Marketplace PNU Engine
│ │ │
│── POST /streaming/sessions ▶│ │
│ │── POST /pipelines/jobs ─▶│
│ │ (job_type=streaming) │
│ │◀── {id, status, ...} ────│
│ │ │
│ │ record_streaming_usage( │
│ │ event="start", │
│ │ session_id=id) │
│ │ │
│◀── {session_id, ...} ─────│ │
│ │ │
│ ... stream runs ... │
│ │ │
│── POST /streaming/ │ │
│ sessions/{id}/stop ────▶│ │
│ │── POST /pipelines/jobs/ │
│ │ {id}/cancel ────────▶│
│ │◀── {started_at, │
│ │ completed_at, │
│ │ gpu_count, ...} ─────│
│ │ │
│ │ record_streaming_usage( │
│ │ event="stop", │
│ │ started_at, │
│ │ completed_at, │
│ │ gpu_count, cpu_cores) │
│ │ │
│◀── {status: "cancelled"} ─│ │
Key points:
- The stop endpoint already exists (POST /streaming/sessions/{id}/stop). It proxies to PNU's POST /pipelines/jobs/{id}/cancel.
- Start and stop events are recorded in the same route handlers — no new routes needed for streaming.
- Duration comes from PNU's started_at and completed_at on the job object.
7. Existing Metering (Unchanged)¶
The marketplace already tracks two other resource types. These are not modified by this work but are documented here for context.
| Collection | Function | What it tracks |
|---|---|---|
llm_dashboard_data |
track_inference_usage() |
Per-user LLM token usage from direct inference API calls (/api/v1/chat/completions). Atomic $inc on tokens, cost, requests. Daily per-model breakdown. |
logs (vdb database) |
track_vdb_operation() |
Per-user vector DB read/write operations with latency, chunk count, result count. |
agentic_dashboard |
(direct insert in agentic routes) | Per-user agentic workflow executions. Count × AGENTIC_RUN_RATE. |
The new data_pipeline_usage collection follows the same pattern: per-user records, never raises on failure (metering must not break the API response), cost computed from rates in load_helper.py.
8. Key Files¶
PNU Engine (D:\PNU-GPU-MANAGEMENT)¶
| File | Role |
|---|---|
src/syntera_engine/services/llm_service.py |
LLMResult dataclass, _invoke() returns usage, _extract_usage() helper |
src/syntera_engine/services/inference_resolver.py |
run_chat_completion() returns full vLLM response (already has usage) |
src/syntera_engine/services/code_execution_service.py |
2 LLM call sites updated to use .content, accumulate usage |
src/syntera_engine/services/data_pipeline/resource_calculator.py |
1 LLM call site updated |
src/syntera_engine/services/data_pipeline/labeling/service.py |
4 LLM call sites updated, bulk worker includes usage in result |
src/syntera_engine/models/data_pipeline/data_processing.py |
DataProcessingJob model — 3 new token columns |
src/syntera_engine/services/data_pipeline/gpu_job_service.py |
_finalize_job() stores usage from result JSON onto job row |
src/syntera_engine/workers/job_runner.py |
No change — json.dumps(result) passes usage through automatically |
src/syntera_engine/api/v1/data_pipeline/jobs.py |
No change — token fields appear in API response automatically |
alembic/versions/014_add_token_tracking.py |
Migration for 3 token columns |
Marketplace (D:\syntera-marketplace)¶
| File | Role |
|---|---|
backend/database.py |
data_pipeline_usage_collection definition |
backend/utils/load_helper.py |
All pricing rate constants (env var + defaults) |
backend/services/usage_tracker.py |
3 new functions: record_data_pipeline_usage(), record_notebook_usage(), record_streaming_usage() |
backend/routes/data_pipeline/etl_routes.py |
Metering hook in get_job_status() (covers ETL + cleaning + labeling) |
backend/routes/data_pipeline/streaming_routes.py |
Start/stop metering hooks in create_stream_session() / stop_stream_session() |
backend/routes/tools/tools_routes.py |
POST /notebooks/{id}/stop, POST /notebooks/{id}/upscaling, DELETE /notebooks/{id} (hard-delete), GET /notebooks, GET /notebooks/{id} routes + start metering in create_tool() |
backend/middleware/auth_middleware.py |
Protect /api/tools/notebooks prefix |
backend/routes/internal/webhook_routes.py |
Webhook receivers: POST /api/internal/notebooks/{id}/stopped (checks action field for stop vs delete) + POST /api/internal/notebooks/{id}/upscaled |
frontend/src/components/marketplace/tools/Notebooks.jsx |
Notebooks management page (list, upscale, stop, delete with confirmation dialog) |
frontend/src/components/marketplace/tools/UpscaleNotebookModal.jsx |
Upscale modal with resource form |
frontend/src/api/dataFetching/tools.js |
API client: getNotebooks, getNotebook, stopNotebook, upscaleNotebook, deleteNotebook |
backend/routes/dashboard.py |
Aggregation queries for data_pipeline_usage_collection |
backend/models/dashboard.py |
Extended DashboardResponse with metering summary fields |
9. Implementation Phases¶
See tasks.md for the full task breakdown with checkboxes, code snippets, and dependencies.
| Phase | Repo | Focus | Tasks |
|---|---|---|---|
| Phase 1 | PNU | LLM Token Capture | 5 |
| Phase 2 | PNU | Model & Storage | 4 |
| Phase 3 | Marketplace | Infrastructure | 5 |
| Phase 4 | Marketplace | Route Hooks | 4 |
| Phase 5 | Marketplace | Dashboard Integration | 2 |
| Phase 6 | Both | Documentation Updates | 9 |
Phases 1 and 3 can proceed in parallel (different repos, no dependencies). Phase 4 depends on both Phase 2 (PNU returns token data) and Phase 3 (marketplace has metering functions). If Phase 1–2 is delayed, Phase 4 can be implemented with token fields defaulting to 0.