◈ API / SDK/2026-03-30Advanced

Gemini API Observability in Production — Logging, Monitoring, and Cost Tracking Patterns

Learn how to build a robust observability stack for production Gemini API deployments. Covers structured logging, token usage tracking, latency monitoring, and cost optimization dashboards with full implementation code.

gemini-api²⁷⁷ observability¹² monitoring⁵ logging² production¹⁴⁰ cost-optimization³⁰ python¹⁰⁴

✦ Premium Article

When you scale a Gemini API project from prototype to production, a familiar set of questions starts surfacing. "Is the API slow because of the model, or the network?" "Why did our API bill spike this month?" "A user reported an error — which request failed and why?" The system that lets you answer these questions instantly is called observability.

This guide walks you through building a production-grade observability stack for the Gemini API, from structured logging design through cost tracking dashboards, with complete implementation code at every step. For related background, see our guides on [Gemini API Error Handling and Retry Patterns]((/articles/gemini-api/gemini-api-error-handling-retry-patterns) and [Gemini API Cost Optimization]((/articles/gemini-api/gemini-api-cost-optimization).

The Three Pillars of Observability for Gemini API

Observability is built on three pillars: Logs, Metrics, and Traces. For production Gemini API services, each pillar plays a distinct role.

Logs capture the details of individual API requests and responses — the prompt content, model output, error messages, and timing information you need for incident investigation. They give you the full story of what happened during any specific interaction.

Metrics show system health as numbers over time. Latency percentiles (P50, P95, P99), token consumption rates, error rates, and requests per second (RPS) let you monitor trends and detect degradation at a glance.

Traces visualize how a single user request flows through your system. In RAG pipelines or multi-agent systems where one user query triggers multiple Gemini API calls, traces help you pinpoint exactly which step is the bottleneck.

Designing and Implementing Structured Logging

Effective Gemini API logging starts with structure. Plain-text logs are nearly useless in production — you can't filter, aggregate, or alert on free-form text when you're handling thousands of requests per minute.

Here's a structured logging design using Python's structlog:

import structlog
import time
import uuid
from google import genai
from google.genai import types
 
# Configure structlog for JSON output
structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer()
    ]
)
logger = structlog.get_logger()
 
class GeminiObservableClient:
    """Gemini API client with built-in observability."""
 
    def __init__(self, api_key: str, default_model: str = "gemini-2.5-pro"):
        self.client = genai.Client(api_key=api_key)
        self.default_model = default_model
        self.metrics = MetricsCollector()
 
    def generate(self, prompt: str, model: str = None,
                 config: dict = None, trace_id: str = None):
        """Generate content with logging and metrics."""
        request_id = str(uuid.uuid4())[:8]
        trace_id = trace_id or str(uuid.uuid4())
        model_name = model or self.default_model
        start_time = time.monotonic()
 
        # Log request start
        logger.info("gemini_request_start",
                     request_id=request_id,
                     trace_id=trace_id,
                     model=model_name,
                     prompt_length=len(prompt),
                     config=config)
 
        try:
            response = self.client.models.generate_content(
                model=model_name,
                contents=prompt,
                config=types.GenerateContentConfig(**(config or {}))
            )
 
            elapsed = time.monotonic() - start_time
            input_tokens = response.usage_metadata.prompt_token_count
            output_tokens = response.usage_metadata.candidates_token_count
 
            # Log success
            logger.info("gemini_request_success",
                        request_id=request_id,
                        trace_id=trace_id,
                        model=model_name,
                        latency_ms=round(elapsed * 1000, 2),
                        input_tokens=input_tokens,
                        output_tokens=output_tokens,
                        total_tokens=input_tokens + output_tokens,
                        finish_reason=str(response.candidates[0].finish_reason))
 
            # Record metrics
            self.metrics.record_request(
                model=model_name,
                latency=elapsed,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                success=True
            )
 
            return response
 
        except Exception as e:
            elapsed = time.monotonic() - start_time
            # Log error
            logger.error("gemini_request_error",
                         request_id=request_id,
                         trace_id=trace_id,
                         model=model_name,
                         latency_ms=round(elapsed * 1000, 2),
                         error_type=type(e).__name__,
                         error_message=str(e))
 
            self.metrics.record_request(
                model=model_name,
                latency=elapsed,
                input_tokens=0,
                output_tokens=0,
                success=False,
                error_type=type(e).__name__
            )
            raise

This client automatically assigns a request_id and trace_id to every API call. The request_id identifies individual calls, while the trace_id ties together all calls within a single user request.

Here's what the log output looks like in practice:

# Example output:
# {"event": "gemini_request_success", "request_id": "a1b2c3d4",
#  "trace_id": "550e8400-...", "model": "gemini-2.5-pro",
#  "latency_ms": 1523.45, "input_tokens": 256,
#  "output_tokens": 1024, "total_tokens": 1280,
#  "finish_reason": "STOP", "timestamp": "2026-03-30T10:15:00Z",
#  "level": "info"}

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Master structured logging design and distributed tracing for Gemini API calls

✦Build a real-time dashboard that tracks token usage and costs across models

✦Set up latency anomaly detection and alerting to prevent outages before they happen

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Metrics Collection and Prometheus Integration

Structured logs tell you what happened to individual requests. Metrics tell you what's happening to your system as a whole. By collecting time-series metrics, you can spot trends in latency, token consumption, and error rates at a glance.

from collections import defaultdict
from dataclasses import dataclass, field
import threading
import time
 
@dataclass
class RequestMetric:
    timestamp: float
    model: str
    latency: float
    input_tokens: int
    output_tokens: int
    success: bool
    error_type: str = ""
 
class MetricsCollector:
    """Collects and aggregates Gemini API metrics."""
 
    def __init__(self):
        self._metrics: list[RequestMetric] = []
        self._lock = threading.Lock()
 
    def record_request(self, model: str, latency: float,
                       input_tokens: int, output_tokens: int,
                       success: bool, error_type: str = ""):
        metric = RequestMetric(
            timestamp=time.time(),
            model=model,
            latency=latency,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            success=success,
            error_type=error_type
        )
        with self._lock:
            self._metrics.append(metric)
 
    def get_summary(self, window_minutes: int = 60) -> dict:
        """Return a summary of metrics over the specified window."""
        cutoff = time.time() - (window_minutes * 60)
        with self._lock:
            recent = [m for m in self._metrics if m.timestamp > cutoff]
 
        if not recent:
            return {"period_minutes": window_minutes, "total_requests": 0}
 
        latencies = sorted([m.latency for m in recent if m.success])
        total_input = sum(m.input_tokens for m in recent)
        total_output = sum(m.output_tokens for m in recent)
        errors = [m for m in recent if not m.success]
 
        # Per-model breakdown
        by_model = defaultdict(lambda: {"count": 0, "tokens": 0})
        for m in recent:
            by_model[m.model]["count"] += 1
            by_model[m.model]["tokens"] += m.input_tokens + m.output_tokens
 
        return {
            "period_minutes": window_minutes,
            "total_requests": len(recent),
            "success_rate": round((len(recent) - len(errors)) / len(recent) * 100, 2),
            "latency": {
                "p50": round(latencies[len(latencies) // 2] * 1000, 2) if latencies else 0,
                "p95": round(latencies[int(len(latencies) * 0.95)] * 1000, 2) if latencies else 0,
                "p99": round(latencies[int(len(latencies) * 0.99)] * 1000, 2) if latencies else 0,
            },
            "tokens": {
                "total_input": total_input,
                "total_output": total_output,
                "total": total_input + total_output,
            },
            "errors": {
                "count": len(errors),
                "by_type": dict(defaultdict(int, {
                    e.error_type: sum(1 for x in errors if x.error_type == e.error_type)
                    for e in errors
                }))
            },
            "by_model": dict(by_model),
        }

To export these metrics in Prometheus format, integrate with the prometheus_client library:

from prometheus_client import Counter, Histogram, Gauge, start_http_server
 
# Prometheus metric definitions
gemini_requests_total = Counter(
    "gemini_api_requests_total",
    "Total Gemini API requests",
    ["model", "status"]
)
gemini_latency = Histogram(
    "gemini_api_latency_seconds",
    "Gemini API latency in seconds",
    ["model"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]
)
gemini_tokens = Counter(
    "gemini_api_tokens_total",
    "Total tokens consumed by Gemini API",
    ["model", "direction"]  # direction: input / output
)
gemini_estimated_cost = Counter(
    "gemini_api_estimated_cost_usd",
    "Estimated Gemini API cost in USD",
    ["model"]
)
 
# Start metrics server on port 9090
# start_http_server(9090)

Building a Cost Tracking System

Cost is one of the most closely watched metrics in any production AI deployment. Understanding per-model pricing and tracking costs in real time is essential for budget control.

from dataclasses import dataclass
from typing import Optional
 
@dataclass
class ModelPricing:
    """Per-model pricing in USD per 1M tokens."""
    input_per_million: float
    output_per_million: float
    context_cache_per_million: Optional[float] = None
 
# Gemini API pricing as of March 2026 (check the official pricing page regularly)
PRICING = {
    "gemini-2.5-pro": ModelPricing(
        input_per_million=1.25,
        output_per_million=10.00,
        context_cache_per_million=0.3125
    ),
    "gemini-2.5-flash": ModelPricing(
        input_per_million=0.15,
        output_per_million=0.60,
        context_cache_per_million=0.0375
    ),
    "gemini-2.0-flash": ModelPricing(
        input_per_million=0.10,
        output_per_million=0.40
    ),
    "gemini-2.0-flash-lite": ModelPricing(
        input_per_million=0.075,
        output_per_million=0.30
    ),
}
 
class CostTracker:
    """Real-time cost tracking for Gemini API usage."""
 
    def __init__(self):
        self._costs: list[dict] = []
        self._lock = threading.Lock()
 
    def calculate_cost(self, model: str, input_tokens: int,
                       output_tokens: int,
                       cached_tokens: int = 0) -> float:
        """Calculate the cost of a single request."""
        pricing = PRICING.get(model)
        if not pricing:
            logger.warning("unknown_model_pricing", model=model)
            return 0.0
 
        input_cost = (input_tokens / 1_000_000) * pricing.input_per_million
        output_cost = (output_tokens / 1_000_000) * pricing.output_per_million
 
        # Context cache discount
        cache_discount = 0.0
        if cached_tokens > 0 and pricing.context_cache_per_million:
            full_price = (cached_tokens / 1_000_000) * pricing.input_per_million
            cache_price = (cached_tokens / 1_000_000) * pricing.context_cache_per_million
            cache_discount = full_price - cache_price
 
        total = input_cost + output_cost - cache_discount
 
        cost_entry = {
            "timestamp": time.time(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cached_tokens": cached_tokens,
            "cost_usd": round(total, 6),
            "cache_savings_usd": round(cache_discount, 6)
        }
 
        with self._lock:
            self._costs.append(cost_entry)
 
        return total
 
    def get_daily_summary(self) -> dict:
        """Return today's cost summary."""
        import datetime
        today_start = datetime.datetime.now().replace(
            hour=0, minute=0, second=0
        ).timestamp()
 
        with self._lock:
            today = [c for c in self._costs if c["timestamp"] > today_start]
 
        total_cost = sum(c["cost_usd"] for c in today)
        by_model = defaultdict(float)
        for c in today:
            by_model[c["model"]] += c["cost_usd"]
 
        return {
            "date": datetime.date.today().isoformat(),
            "total_cost_usd": round(total_cost, 4),
            "total_requests": len(today),
            "by_model": {k: round(v, 4) for k, v in by_model.items()},
            "total_savings_from_cache": round(
                sum(c["cache_savings_usd"] for c in today), 4
            )
        }

Rate limiting integration matters here too — you can use your metrics data to visualize API quota consumption in real time. For details on quota management, see [Gemini API Rate Limiting and Quota Management]((/articles/gemini-api/gemini-api-rate-limiting-quota-management).

One detail that's easy to overlook is context cache savings. Gemini 2.5 Pro's context cache pricing is roughly 25% of the standard input token rate, which means workloads that repeatedly use the same context can see dramatic cost reductions. The CostTracker above factors cache discounts into its calculations so you get an accurate picture of actual spending.

Latency Monitoring and Alert Design

Detecting latency anomalies early requires thoughtful threshold design. Gemini API latency varies significantly depending on the model, prompt length, and output complexity, so static thresholds alone aren't enough. Combining them with dynamic anomaly detection based on moving averages gives you much better signal quality.

from collections import deque
import statistics
 
class LatencyMonitor:
    """Latency anomaly detection monitor."""
 
    def __init__(self, window_size: int = 100,
                 static_threshold_ms: float = 10000,
                 sigma_multiplier: float = 3.0):
        self._window = deque(maxlen=window_size)
        self.static_threshold_ms = static_threshold_ms
        self.sigma_multiplier = sigma_multiplier
 
    def record_and_check(self, latency_ms: float, model: str) -> dict:
        """Record latency and check for anomalies."""
        self._window.append(latency_ms)
 
        alert = {"is_anomaly": False, "reason": ""}
 
        # Static threshold check
        if latency_ms > self.static_threshold_ms:
            alert = {
                "is_anomaly": True,
                "reason": f"Latency {latency_ms:.0f}ms exceeded "
                          f"threshold {self.static_threshold_ms}ms"
            }
 
        # Dynamic anomaly detection (sigma rule)
        if len(self._window) >= 20:
            mean = statistics.mean(self._window)
            stdev = statistics.stdev(self._window)
            dynamic_threshold = mean + (self.sigma_multiplier * stdev)
 
            if latency_ms > dynamic_threshold and not alert["is_anomaly"]:
                alert = {
                    "is_anomaly": True,
                    "reason": f"Latency {latency_ms:.0f}ms exceeded "
                              f"dynamic threshold {dynamic_threshold:.0f}ms "
                              f"(mean={mean:.0f}, sigma={stdev:.0f})"
                }
 
        if alert["is_anomaly"]:
            logger.warning("latency_anomaly_detected",
                           model=model,
                           latency_ms=latency_ms,
                           reason=alert["reason"])
 
        return alert

In production, wire these alerts to your notification system. Here's a Slack integration example:

import json
from urllib.request import Request, urlopen
 
def send_slack_alert(webhook_url: str, alert_data: dict):
    """Send an alert to Slack."""
    message = {
        "text": ":rotating_light: Gemini API Latency Anomaly Detected",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Model*: `{alert_data.get('model', 'unknown')}`\n"
                            f"*Latency*: {alert_data.get('latency_ms', 0):.0f}ms\n"
                            f"*Reason*: {alert_data.get('reason', '')}"
                }
            }
        ]
    }
    req = Request(webhook_url,
                  data=json.dumps(message).encode(),
                  headers={"Content-Type": "application/json"})
    urlopen(req)

Distributed Tracing with OpenTelemetry

In RAG pipelines or multi-agent systems, a single user request often triggers multiple Gemini API calls. OpenTelemetry (OTel) distributed tracing lets you visualize the full request flow and pinpoint bottlenecks.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
 
# Initialize OTel
resource = Resource.create({"service.name": "gemini-api-service"})
provider = TracerProvider(resource=resource)
exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
 
tracer = trace.get_tracer("gemini.api")
 
class TracedGeminiClient(GeminiObservableClient):
    """Gemini client with OpenTelemetry tracing."""
 
    def generate_with_trace(self, prompt: str,
                            model: str = None, config: dict = None):
        model_name = model or self.default_model
 
        with tracer.start_as_current_span("gemini.generate") as span:
            span.set_attribute("gemini.model", model_name)
            span.set_attribute("gemini.prompt_length", len(prompt))
 
            try:
                response = self.generate(
                    prompt=prompt,
                    model=model_name,
                    config=config,
                    trace_id=format(span.get_span_context().trace_id, "032x")
                )
 
                span.set_attribute("gemini.input_tokens",
                                   response.usage_metadata.prompt_token_count)
                span.set_attribute("gemini.output_tokens",
                                   response.usage_metadata.candidates_token_count)
                span.set_status(trace.StatusCode.OK)
                return response
 
            except Exception as e:
                span.set_status(trace.StatusCode.ERROR, str(e))
                span.record_exception(e)
                raise
 
    def rag_pipeline(self, query: str, context_docs: list[str]):
        """Execute a RAG pipeline with full tracing."""
        with tracer.start_as_current_span("rag.pipeline") as parent:
            parent.set_attribute("rag.query", query)
            parent.set_attribute("rag.context_count", len(context_docs))
 
            # Step 1: Query expansion
            with tracer.start_as_current_span("rag.query_expansion"):
                expanded = self.generate_with_trace(
                    f"Expand this search query: {query}",
                    model="gemini-2.5-flash"
                )
 
            # Step 2: Answer generation with context
            context = "\n---\n".join(context_docs)
            with tracer.start_as_current_span("rag.answer_generation"):
                answer = self.generate_with_trace(
                    f"Context:\n{context}\n\nQuestion: {query}",
                    model="gemini-2.5-pro"
                )
 
            return answer

Traces can be visualized in backends like Jaeger, Zipkin, or Grafana Tempo. By attaching Gemini-specific attributes (model name, token counts, latency) to each span, you make bottleneck identification straightforward.

Building a Grafana Dashboard

Here's the recommended panel layout for a Grafana dashboard that monitors your Gemini API deployment:

Dashboard: "Gemini API Overview"

Request Rate (time series): rate(gemini_api_requests_total[5m]) — per-model RPS as line charts
Error Rate: rate(gemini_api_requests_total{status="error"}[5m]) / rate(gemini_api_requests_total[5m]) — target below 1%
Latency Distribution: histogram_quantile(0.95, rate(gemini_api_latency_seconds_bucket[5m])) — three lines for P50, P95, and P99
Token Consumption: increase(gemini_api_tokens_total[1h]) — stacked area chart, input vs. output
Estimated Cost: increase(gemini_api_estimated_cost_usd[24h]) — daily cumulative cost
Cost by Model: sum by (model) (increase(gemini_api_estimated_cost_usd[24h])) — pie chart showing cost distribution

At minimum, configure these three alert rules:

Error rate exceeds 5%: Slack notification when the 5-minute error rate crosses 5%
P95 latency anomaly: Slack notification when P95 exceeds 3x the normal baseline
Daily cost threshold: Slack and email notification when estimated daily cost exceeds your budget cap

Production Best Practices

Here's practical guidance for running these observability components in production.

Log lifecycle management: Storing all API logs forever will blow up your storage costs. Design a lifecycle policy — hot logs (last 7 days) in a searchable store like Elasticsearch or Cloud Logging, cold logs (7–90 days) in object storage like Cloud Storage or S3, and delete everything older.

Prompt sanitization: If you're logging full prompts, they may contain user PII. Run a sanitization pass before writing logs to strip sensitive data like email addresses, phone numbers, and credit card numbers.

import re
 
def sanitize_prompt(text: str) -> str:
    """Mask PII in prompts before logging."""
    # Email addresses
    text = re.sub(r'[\w.+-]+@[\w-]+\.[\w.]+', '[EMAIL]', text)
    # Phone numbers (US format)
    text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE]', text)
    # Credit card numbers
    text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD]', text)
    return text

Metrics-driven fallback strategy: Use your observability data to implement automatic model fallback. For example, if Gemini 2.5 Pro's P95 latency exceeds a threshold, automatically switch to Gemini 2.5 Flash for non-critical requests.

class AdaptiveModelSelector:
    """Metrics-based automatic model selection."""
 
    def __init__(self, latency_monitor: LatencyMonitor,
                 primary: str = "gemini-2.5-pro",
                 fallback: str = "gemini-2.5-flash",
                 latency_threshold_ms: float = 8000):
        self.latency_monitor = latency_monitor
        self.primary = primary
        self.fallback = fallback
        self.threshold = latency_threshold_ms
        self._use_fallback = False
 
    def select_model(self) -> str:
        """Select a model based on current latency conditions."""
        if self._use_fallback:
            # Periodically try the primary model to check recovery
            return self.primary if time.time() % 60 < 5 else self.fallback
        return self.primary
 
    def on_response(self, model: str, latency_ms: float):
        """Evaluate fallback on each response."""
        result = self.latency_monitor.record_and_check(latency_ms, model)
        if result["is_anomaly"] and model == self.primary:
            self._use_fallback = True
            logger.warning("model_fallback_activated",
                           primary=self.primary,
                           fallback=self.fallback)

Looking back

In production Gemini API deployments, observability isn't a nice-to-have — it's table stakes. Structured logs let you investigate individual requests. Metrics let you monitor system health. Cost tracking keeps your budget under control. With these three pillars in place, you can operate production AI services with confidence.

All the Python code in this article is modular and designed for incremental adoption. Start by swapping GeminiObservableClient into your existing codebase and build from there.

Thank You for Reading

Gemini Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.