Cloudflare Workers Observability

Cloudflare Workers Observability

Production-grade observability for Cloudflare Workers: logging, metrics, tracing, and alerting.

Quick Start

// Structured logging with context
export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const requestId = crypto.randomUUID();
    const logger = createLogger(requestId, env);

    try {
      logger.info('Request received', { method: request.method, url: request.url });

      const result = await handleRequest(request, env);

      logger.info('Request completed', { status: result.status });
      return result;
    } catch (error) {
      logger.error('Request failed', { error: error.message, stack: error.stack });
      throw error;
    }
  }
};

// Simple logger factory
function createLogger(requestId: string, env: Env) {
  return {
    info: (msg: string, data?: object) => console.log(JSON.stringify({ level: 'info', requestId, msg, ...data, timestamp: Date.now() })),
    error: (msg: string, data?: object) => console.error(JSON.stringify({ level: 'error', requestId, msg, ...data, timestamp: Date.now() })),
    warn: (msg: string, data?: object) => console.warn(JSON.stringify({ level: 'warn', requestId, msg, ...data, timestamp: Date.now() })),
  };
}

Critical Rules

1. Always use structured JSON logging - Plain text logs are hard to parse and aggregate 2. Include request context - Request ID, method, path in every log entry 3. Never log sensitive data - Redact tokens, passwords, PII from logs 4. Use appropriate log levels - ERROR for failures, WARN for recoverable issues, INFO for operations 5. Sample high-volume logs - Use 1-10% sampling for request logs in production

Observability Components

Top 8 Errors Prevented

Logging Configuration

wrangler.jsonc:

{
  "name": "my-worker",
  "observability": {
    "enabled": true,
    "head_sampling_rate": 1 // 0-1, 1 = 100% of requests
  },
  "tail_consumers": [
    {
      "service": "log-aggregator", // Tail Worker name
      "environment": "production"
    }
  ],
  "analytics_engine_datasets": [
    {
      "binding": "ANALYTICS",
      "dataset": "my_worker_metrics"
    }
  ]
}

Structured Logging Pattern

interface LogEntry {
  level: 'debug' | 'info' | 'warn' | 'error';
  message: string;
  requestId: string;
  timestamp: number;
  // Contextual data
  method?: string;
  path?: string;
  status?: number;
  duration?: number;
  // Error details
  error?: {
    name: string;
    message: string;
    stack?: string;
  };
  // Custom fields
  [key: string]: unknown;
}

class Logger {
  constructor(private requestId: string, private baseContext: object = {}) {}

  private log(level: LogEntry['level'], message: string, data?: object) {
    const entry: LogEntry = {
      level,
      message,
      requestId: this.requestId,
      timestamp: Date.now(),
      ...this.baseContext,
      ...data,
    };

    // Redact sensitive fields
    const sanitized = this.redact(entry);

    const output = JSON.stringify(sanitized);
    level === 'error' ? console.error(output) : console.log(output);
  }

  private redact(entry: LogEntry): LogEntry {
    const sensitiveKeys = ['password', 'token', 'secret', 'authorization', 'cookie'];
    const redacted = { ...entry };

    for (const key of Object.keys(redacted)) {
      if (sensitiveKeys.some(s => key.toLowerCase().includes(s))) {
        redacted[key] = '[REDACTED]';
      }
    }
    return redacted;
  }

  info(message: string, data?: object) { this.log('info', message, data); }
  warn(message: string, data?: object) { this.log('warn', message, data); }
  error(message: string, data?: object) { this.log('error', message, data); }
  debug(message: string, data?: object) { this.log('debug', message, data); }
}

Analytics Engine Usage

interface Env {
  ANALYTICS: AnalyticsEngineDataset;
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const start = Date.now();
    const url = new URL(request.url);

    try {
      const response = await handleRequest(request, env);

      // Write success metric
      env.ANALYTICS.writeDataPoint({
        blobs: [request.method, url.pathname, String(response.status)],
        doubles: [Date.now() - start], // Response time in ms
        indexes: [url.pathname.split('/')[1] || 'root'], // Index for fast queries
      });

      return response;
    } catch (error) {
      // Write error metric
      env.ANALYTICS.writeDataPoint({
        blobs: [request.method, url.pathname, 'error', error.message],
        doubles: [Date.now() - start],
        indexes: ['error'],
      });
      throw error;
    }
  }
};

Tail Worker Pattern

// tail-worker.ts - Receives logs from other workers
interface TailEvent {
  scriptName: string;
  event: {
    request?: { method: string; url: string };
    response?: { status: number };
  };
  logs: Array<{
    level: string;
    message: unknown[];
    timestamp: number;
  }>;
  exceptions: Array<{
    name: string;
    message: string;
    timestamp: number;
  }>;
  outcome: 'ok' | 'exception' | 'exceededCpu' | 'exceededMemory' | 'canceled';
  eventTimestamp: number;
}

export default {
  async tail(events: TailEvent[], env: Env): Promise<void> {
    for (const event of events) {
      // Filter and forward logs
      const errorLogs = event.logs.filter(l => l.level === 'error');
      const exceptions = event.exceptions;

      if (errorLogs.length > 0 || exceptions.length > 0) {
        // Send to external logging service
        await fetch(env.LOGGING_ENDPOINT, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({
            scriptName: event.scriptName,
            timestamp: event.eventTimestamp,
            errors: errorLogs,
            exceptions,
            outcome: event.outcome,
          }),
        });
      }
    }
  }
};

When to Load References

Load specific references based on the task:

• Setting up logging? → Load references/logging.md for structured logging patterns, log levels, redaction
• Building custom metrics? → Load references/analytics-engine.md for Analytics Engine SQL queries, data modeling
• Implementing log aggregation? → Load references/tail-workers.md for Tail Worker patterns, external service integration
• Creating dashboards/tracking? → Load references/custom-metrics.md for business metrics, performance tracking
• Setting up alerts? → Load references/alerting.md for error rate monitoring, PagerDuty/Slack integration

Templates

Scripts

Resources

• Workers Observability: https://developers.cloudflare.com/workers/observability/
• Analytics Engine: https://developers.cloudflare.com/analytics/analytics-engine/
• Tail Workers: https://developers.cloudflare.com/workers/observability/tail-workers/
• Logpush: https://developers.cloudflare.com/logs/get-started/

Alerting for Cloudflare Workers

Configure monitoring and alerting for production Workers.

Alerting Architecture

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Your Worker    │────▶│   Tail Worker   │────▶│  Alert Service  │
│  (logs/metrics) │     │   (aggregates)  │     │ (Slack/PD/etc)  │
└─────────────────┘     └─────────────────┘     └─────────────────┘
                               │
                               ▼
                        ┌─────────────────┐
                        │ Analytics Engine│
                        │  (dashboards)   │
                        └─────────────────┘

Cloudflare Dashboard Alerts

Built-in Notifications

1. Go to Notifications in Cloudflare Dashboard 2. Create notification for:

• Worker errors
• Worker CPU exceeded
• Worker memory exceeded
• Deployment status

Configuring

Workers → Notifications → Create
├── Trigger: Worker Script Exception
├── Workers: [Select specific or all]
├── Delivery: Email, Webhook, PagerDuty
└── Filters: Environment, Script name

Custom Alert Implementation

Alert Manager in Tail Worker

interface Env {
  SLACK_WEBHOOK: string;
  PAGERDUTY_KEY: string;
  KV: KVNamespace; // Rate limiting
  ANALYTICS: AnalyticsEngineDataset;
}

interface AlertConfig {
  name: string;
  condition: (event: TailEvent) => boolean;
  severity: 'info' | 'warning' | 'critical';
  cooldown: number; // seconds
}

const ALERTS: AlertConfig[] = [
  {
    name: 'worker_exception',
    condition: (e) => e.outcome === 'exception' || e.exceptions.length > 0,
    severity: 'critical',
    cooldown: 300, // 5 minutes
  },
  {
    name: 'cpu_exceeded',
    condition: (e) => e.outcome === 'exceededCpu',
    severity: 'critical',
    cooldown: 300,
  },
  {
    name: 'high_error_rate',
    condition: (e) => e.event.response?.status ? e.event.response.status >= 500 : false,
    severity: 'warning',
    cooldown: 60,
  },
];

class AlertManager {
  constructor(private env: Env) {}

  async checkAndAlert(events: TailEvent[]) {
    for (const alert of ALERTS) {
      const triggered = events.filter(alert.condition);

      if (triggered.length > 0) {
        await this.sendAlert(alert, triggered);
      }
    }
  }

  private async sendAlert(alert: AlertConfig, events: TailEvent[]) {
    // Check cooldown
    const key = `alert:${alert.name}:${events[0].scriptName}`;
    const lastAlert = await this.env.KV.get(key);

    if (lastAlert) {
      const elapsed = Date.now() - parseInt(lastAlert);
      if (elapsed < alert.cooldown * 1000) return;
    }

    // Set cooldown
    await this.env.KV.put(key, String(Date.now()), {
      expirationTtl: alert.cooldown,
    });

    // Send based on severity
    switch (alert.severity) {
      case 'critical':
        await this.sendPagerDuty(alert, events);
        await this.sendSlack(alert, events);
        break;
      case 'warning':
        await this.sendSlack(alert, events);
        break;
      case 'info':
        // Log only
        break;
    }

    // Track alert
    this.env.ANALYTICS.writeDataPoint({
      blobs: [alert.name, alert.severity, events[0].scriptName],
      doubles: [events.length, 1],
      indexes: ['alert'],
    });
  }

  private async sendSlack(alert: AlertConfig, events: TailEvent[]) {
    const event = events[0];
    const exception = event.exceptions[0];

    await fetch(this.env.SLACK_WEBHOOK, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        blocks: [
          {
            type: 'header',
            text: {
              type: 'plain_text',
              text: `${alert.severity === 'critical' ? '🚨' : '⚠️'} ${alert.name}`,
            },
          },
          {
            type: 'section',
            fields: [
              { type: 'mrkdwn', text: `*Worker:*\n${event.scriptName}` },
              { type: 'mrkdwn', text: `*Outcome:*\n${event.outcome}` },
              { type: 'mrkdwn', text: `*Count:*\n${events.length} events` },
              {
                type: 'mrkdwn',
                text: `*Time:*\n${new Date(event.eventTimestamp).toISOString()}`,
              },
            ],
          },
          exception && {
            type: 'section',
            text: {
              type: 'mrkdwn',
              text: `*Error:*\n\`\`\`${exception.name}: ${exception.message}\`\`\``,
            },
          },
        ].filter(Boolean),
      }),
    });
  }

  private async sendPagerDuty(alert: AlertConfig, events: TailEvent[]) {
    const event = events[0];

    await fetch('https://events.pagerduty.com/v2/enqueue', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        routing_key: this.env.PAGERDUTY_KEY,
        event_action: 'trigger',
        dedup_key: `${alert.name}-${event.scriptName}`,
        payload: {
          summary: `[${alert.name}] ${event.scriptName}: ${event.outcome}`,
          severity: alert.severity === 'critical' ? 'critical' : 'warning',
          source: 'cloudflare-workers',
          custom_details: {
            worker: event.scriptName,
            outcome: event.outcome,
            exceptions: event.exceptions,
            eventCount: events.length,
          },
        },
      }),
    });
  }
}

export default {
  async tail(events: TailEvent[], env: Env): Promise<void> {
    const alertManager = new AlertManager(env);
    await alertManager.checkAndAlert(events);
  }
};

Error Rate Alerting

Analytics Engine-Based Alerts

// Scheduled worker that checks error rates
export default {
  async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
    const errorRate = await checkErrorRate(env);

    if (errorRate > 5) { // > 5% error rate
      await sendAlert(env, {
        title: 'High Error Rate Alert',
        message: `Error rate is ${errorRate.toFixed(2)}% (threshold: 5%)`,
        severity: 'warning',
      });
    }

    if (errorRate > 10) { // > 10% error rate
      await sendAlert(env, {
        title: 'Critical Error Rate Alert',
        message: `Error rate is ${errorRate.toFixed(2)}% (threshold: 10%)`,
        severity: 'critical',
      });
    }
  }
};

async function checkErrorRate(env: Env): Promise<number> {
  const response = await fetch(
    `https://api.cloudflare.com/client/v4/accounts/${env.ACCOUNT_ID}/analytics_engine/sql`,
    {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${env.CLOUDFLARE_API_TOKEN}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        query: `
          SELECT
            SUM(double4) as errors,
            SUM(double3) as total,
            SUM(double4) / SUM(double3) * 100 as error_rate
          FROM my_worker_metrics
          WHERE timestamp > NOW() - INTERVAL '5' MINUTE
        `,
      }),
    }
  );

  const data = await response.json();
  return data.data?.[0]?.error_rate || 0;
}

Alert Integrations

Slack

async function sendSlackAlert(webhookUrl: string, alert: Alert) {
  await fetch(webhookUrl, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      text: `${alert.severity === 'critical' ? '🚨' : '⚠️'} ${alert.title}`,
      blocks: [
        {
          type: 'header',
          text: { type: 'plain_text', text: alert.title },
        },
        {
          type: 'section',
          text: { type: 'mrkdwn', text: alert.message },
        },
        {
          type: 'context',
          elements: [
            { type: 'mrkdwn', text: `Severity: *${alert.severity}*` },
            { type: 'mrkdwn', text: `Time: ${new Date().toISOString()}` },
          ],
        },
      ],
    }),
  });
}

PagerDuty

async function sendPagerDutyAlert(routingKey: string, alert: Alert) {
  await fetch('https://events.pagerduty.com/v2/enqueue', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      routing_key: routingKey,
      event_action: 'trigger',
      dedup_key: alert.dedupKey,
      payload: {
        summary: alert.title,
        severity: alert.severity,
        source: 'cloudflare-workers',
        custom_details: alert.details,
      },
    }),
  });
}

OpsGenie

async function sendOpsGenieAlert(apiKey: string, alert: Alert) {
  await fetch('https://api.opsgenie.com/v2/alerts', {
    method: 'POST',
    headers: {
      'Authorization': `GenieKey ${apiKey}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      message: alert.title,
      description: alert.message,
      priority: alert.severity === 'critical' ? 'P1' : 'P3',
      tags: ['cloudflare-workers', alert.worker],
    }),
  });
}

Alert Best Practices

1. Set Appropriate Thresholds

2. Implement Cooldowns

• Critical: 5-10 minute cooldown
• Warning: 1-5 minute cooldown
• Info: No cooldown (log only)

3. Add Context to Alerts

{
  worker: 'api-worker',
  outcome: 'exception',
  error: 'TypeError: Cannot read property...',
  request: { method: 'POST', path: '/api/users' },
  recentChanges: 'Deployed 15 minutes ago',
  runbook: 'https://wiki.example.com/runbooks/api-worker'
}

4. Group Related Alerts

• Deduplicate by error type + worker
• Aggregate counts in single alert
• Show first occurrence + count

5. Auto-Resolve

// Resolve PagerDuty when recovered
await fetch('https://events.pagerduty.com/v2/enqueue', {
  method: 'POST',
  body: JSON.stringify({
    routing_key: key,
    event_action: 'resolve',
    dedup_key: dedupKey,
  }),
});

Analytics Engine for Cloudflare Workers

Build custom analytics and metrics dashboards with Workers Analytics Engine.

Overview

Analytics Engine is a time-series database built into Workers:

• Write from any Worker with zero latency impact
• Query via SQL API or GraphQL
• Aggregate automatically (no manual rollups)
• Retain data for 90 days

Configuration

wrangler.jsonc:

{
  "name": "my-worker",
  "analytics_engine_datasets": [
    {
      "binding": "ANALYTICS",
      "dataset": "my_worker_metrics"
    }
  ]
}

Writing Data Points

interface Env {
  ANALYTICS: AnalyticsEngineDataset;
}

// Basic write
env.ANALYTICS.writeDataPoint({
  blobs: ['GET', '/api/users', '200'],  // Strings (up to 20)
  doubles: [150, 1024],                  // Numbers (up to 20)
  indexes: ['api']                       // Fast query indexes (up to 1)
});

Data Point Structure

Best Practices

// ✅ Good: Consistent schema across all writes
env.ANALYTICS.writeDataPoint({
  blobs: [
    request.method,           // blob1: HTTP method
    url.pathname,             // blob2: Path
    String(response.status),  // blob3: Status code
    request.cf?.country || 'unknown', // blob4: Country
    request.cf?.colo || 'unknown'     // blob5: Colo
  ],
  doubles: [
    Date.now() - startTime,   // double1: Response time (ms)
    responseSize,             // double2: Response size (bytes)
    1                         // double3: Request count (for SUM)
  ],
  indexes: [url.pathname.split('/')[1] || 'root']
});

// ❌ Bad: Inconsistent schema
// Different writes with different blob meanings

Complete Metrics Example

interface Env {
  ANALYTICS: AnalyticsEngineDataset;
}

interface RequestMetrics {
  method: string;
  path: string;
  status: number;
  duration: number;
  responseSize: number;
  country: string;
  error?: string;
}

function writeMetrics(env: Env, metrics: RequestMetrics) {
  env.ANALYTICS.writeDataPoint({
    blobs: [
      metrics.method,
      metrics.path,
      String(metrics.status),
      metrics.country,
      metrics.error || ''
    ],
    doubles: [
      metrics.duration,
      metrics.responseSize,
      1, // count for aggregation
      metrics.status >= 400 ? 1 : 0, // error count
      metrics.status >= 500 ? 1 : 0  // server error count
    ],
    indexes: [metrics.status >= 400 ? 'error' : 'success']
  });
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const start = Date.now();
    const url = new URL(request.url);

    try {
      const response = await handleRequest(request, env);
      const body = await response.clone().arrayBuffer();

      ctx.waitUntil(
        Promise.resolve(writeMetrics(env, {
          method: request.method,
          path: url.pathname,
          status: response.status,
          duration: Date.now() - start,
          responseSize: body.byteLength,
          country: (request.cf?.country as string) || 'unknown'
        }))
      );

      return response;
    } catch (error) {
      writeMetrics(env, {
        method: request.method,
        path: url.pathname,
        status: 500,
        duration: Date.now() - start,
        responseSize: 0,
        country: (request.cf?.country as string) || 'unknown',
        error: error.message
      });

      throw error;
    }
  }
};

Querying Data

SQL API

# Query via API
curl -X POST "https://api.cloudflare.com/client/v4/accounts/{account_id}/analytics_engine/sql" \
  -H "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "SELECT blob1 as method, COUNT(*) as requests, AVG(double1) as avg_latency FROM my_worker_metrics WHERE timestamp > NOW() - INTERVAL '\''1'\'' HOUR GROUP BY blob1"
  }'

Common Queries

Request count by status:

SELECT
  blob3 as status,
  COUNT(*) as requests
FROM my_worker_metrics
WHERE timestamp > NOW() - INTERVAL '1' HOUR
GROUP BY blob3
ORDER BY requests DESC

Average latency by path:

SELECT
  blob2 as path,
  AVG(double1) as avg_latency_ms,
  MAX(double1) as max_latency_ms,
  COUNT(*) as requests
FROM my_worker_metrics
WHERE timestamp > NOW() - INTERVAL '1' HOUR
GROUP BY blob2
ORDER BY avg_latency_ms DESC
LIMIT 10

Error rate over time:

SELECT
  toStartOfMinute(timestamp) as minute,
  SUM(double4) / SUM(double3) * 100 as error_rate_pct,
  SUM(double3) as total_requests
FROM my_worker_metrics
WHERE timestamp > NOW() - INTERVAL '1' HOUR
GROUP BY minute
ORDER BY minute

Geographic distribution:

SELECT
  blob4 as country,
  COUNT(*) as requests,
  AVG(double1) as avg_latency_ms
FROM my_worker_metrics
WHERE timestamp > NOW() - INTERVAL '24' HOUR
GROUP BY country
ORDER BY requests DESC
LIMIT 20

P99 latency:

SELECT
  blob2 as path,
  quantile(0.99)(double1) as p99_latency_ms,
  quantile(0.95)(double1) as p95_latency_ms,
  quantile(0.50)(double1) as p50_latency_ms
FROM my_worker_metrics
WHERE timestamp > NOW() - INTERVAL '1' HOUR
GROUP BY path
ORDER BY p99_latency_ms DESC

Building Dashboards

Grafana Integration

// Worker endpoint for Grafana JSON datasource
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === '/query') {
      const body = await request.json();
      const results = await queryAnalytics(env, body.targets);
      return Response.json(results);
    }

    return new Response('Analytics API', { status: 200 });
  }
};

async function queryAnalytics(env: Env, targets: any[]) {
  // Transform Grafana targets to Analytics Engine queries
  // Return data in Grafana format
}

Custom Dashboard Worker

// Simple HTML dashboard
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const metrics = await getMetrics(env);

    return new Response(`
      <!DOCTYPE html>
      <html>
        <head><title>Worker Metrics</title></head>
        <body>
          <h1>Last Hour Metrics</h1>
          <ul>
            <li>Total Requests: ${metrics.totalRequests}</li>
            <li>Error Rate: ${metrics.errorRate.toFixed(2)}%</li>
            <li>Avg Latency: ${metrics.avgLatency.toFixed(0)}ms</li>
            <li>P99 Latency: ${metrics.p99Latency.toFixed(0)}ms</li>
          </ul>
        </body>
      </html>
    `, {
      headers: { 'Content-Type': 'text/html' }
    });
  }
};

Limits and Quotas

Cost Optimization

1. Use indexes wisely - Only index the most-queried dimension 2. Aggregate at write - Write count: 1 and SUM instead of counting rows 3. Sample high-volume - Write 10% of requests for patterns, 100% for errors 4. Batch writes - Use ctx.waitUntil() to not block responses

Custom Metrics for Cloudflare Workers

Track business metrics, performance indicators, and custom analytics.

Metrics Categories

1. Request Metrics

interface RequestMetrics {
  // Timing
  totalDuration: number;
  dbQueryTime: number;
  externalApiTime: number;
  processingTime: number;

  // Request details
  method: string;
  path: string;
  status: number;

  // Response
  responseSize: number;
  cacheStatus: 'HIT' | 'MISS' | 'BYPASS';
}

2. Business Metrics

interface BusinessMetrics {
  // User actions
  signups: number;
  logins: number;
  purchases: number;

  // Revenue
  orderValue: number;
  currency: string;

  // Engagement
  pageViews: number;
  apiCalls: number;
}

3. System Metrics

interface SystemMetrics {
  // Resource usage
  cpuTime: number;
  memoryUsed: number;

  // Errors
  errorCount: number;
  errorType: string;

  // Dependencies
  dbConnections: number;
  cacheHitRate: number;
}

Implementation Patterns

Metrics Collector Class

interface Env {
  ANALYTICS: AnalyticsEngineDataset;
}

class MetricsCollector {
  private startTime: number;
  private metrics: Map<string, number> = new Map();
  private tags: Map<string, string> = new Map();

  constructor(private env: Env) {
    this.startTime = Date.now();
  }

  // Timing helpers
  startTimer(name: string): () => void {
    const start = Date.now();
    return () => {
      this.metrics.set(`${name}_ms`, Date.now() - start);
    };
  }

  // Counter helpers
  increment(name: string, value: number = 1) {
    const current = this.metrics.get(name) || 0;
    this.metrics.set(name, current + value);
  }

  // Tag helpers
  setTag(name: string, value: string) {
    this.tags.set(name, value);
  }

  // Write to Analytics Engine
  flush() {
    const blobs: string[] = [];
    const doubles: number[] = [];

    // Convert tags to blobs
    this.tags.forEach((value, key) => {
      blobs.push(`${key}:${value}`);
    });

    // Convert metrics to doubles
    this.metrics.forEach((value, key) => {
      doubles.push(value);
    });

    // Add total duration
    doubles.push(Date.now() - this.startTime);

    this.env.ANALYTICS.writeDataPoint({
      blobs: blobs.slice(0, 20),
      doubles: doubles.slice(0, 20),
      indexes: [this.tags.get('path')?.split('/')[1] || 'root'],
    });
  }
}

Usage in Worker

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const metrics = new MetricsCollector(env);
    const url = new URL(request.url);

    // Set tags
    metrics.setTag('method', request.method);
    metrics.setTag('path', url.pathname);
    metrics.setTag('country', (request.cf?.country as string) || 'unknown');

    try {
      // Time database operation
      const stopDbTimer = metrics.startTimer('db');
      const data = await queryDatabase(env);
      stopDbTimer();

      // Time external API
      const stopApiTimer = metrics.startTimer('external_api');
      const enriched = await fetchExternalData(data);
      stopApiTimer();

      // Track business metric
      if (url.pathname === '/api/orders') {
        metrics.increment('orders_created');
        metrics.increment('order_value', enriched.total);
      }

      const response = new Response(JSON.stringify(enriched));
      metrics.setTag('status', String(response.status));
      metrics.increment('response_size', JSON.stringify(enriched).length);

      // Flush metrics after response
      ctx.waitUntil(Promise.resolve(metrics.flush()));

      return response;
    } catch (error) {
      metrics.setTag('status', '500');
      metrics.setTag('error', error.name);
      metrics.increment('errors');
      ctx.waitUntil(Promise.resolve(metrics.flush()));
      throw error;
    }
  }
};

Performance Tracking

Response Time Percentiles

// Worker that tracks detailed timing
async function handleWithTiming(request: Request, env: Env, ctx: ExecutionContext) {
  const timings = {
    start: Date.now(),
    parseComplete: 0,
    authComplete: 0,
    dbComplete: 0,
    processComplete: 0,
    end: 0,
  };

  // Parse request
  const body = await request.json();
  timings.parseComplete = Date.now();

  // Auth check
  await verifyAuth(request.headers);
  timings.authComplete = Date.now();

  // Database query
  const data = await queryDb(env, body);
  timings.dbComplete = Date.now();

  // Process
  const result = processData(data);
  timings.processComplete = Date.now();

  // Respond
  const response = Response.json(result);
  timings.end = Date.now();

  // Write timing breakdown
  ctx.waitUntil(
    Promise.resolve(
      env.ANALYTICS.writeDataPoint({
        blobs: [request.method, new URL(request.url).pathname],
        doubles: [
          timings.parseComplete - timings.start,      // parse_ms
          timings.authComplete - timings.parseComplete, // auth_ms
          timings.dbComplete - timings.authComplete,    // db_ms
          timings.processComplete - timings.dbComplete, // process_ms
          timings.end - timings.processComplete,        // respond_ms
          timings.end - timings.start,                  // total_ms
        ],
        indexes: ['timing'],
      })
    )
  );

  return response;
}

Cache Effectiveness

async function fetchWithCache(
  env: Env,
  key: string,
  fetcher: () => Promise<unknown>
): Promise<{ data: unknown; cacheStatus: string }> {
  // Try cache first
  const cached = await env.KV.get(key, 'json');

  if (cached) {
    env.ANALYTICS.writeDataPoint({
      blobs: ['cache', 'HIT'],
      doubles: [1, 0], // hit count, miss count
      indexes: ['cache'],
    });
    return { data: cached, cacheStatus: 'HIT' };
  }

  // Cache miss - fetch and store
  const data = await fetcher();
  await env.KV.put(key, JSON.stringify(data), { expirationTtl: 3600 });

  env.ANALYTICS.writeDataPoint({
    blobs: ['cache', 'MISS'],
    doubles: [0, 1], // hit count, miss count
    indexes: ['cache'],
  });

  return { data, cacheStatus: 'MISS' };
}

Business Metrics Examples

E-Commerce

function trackPurchase(env: Env, order: Order) {
  env.ANALYTICS.writeDataPoint({
    blobs: [
      'purchase',
      order.currency,
      order.paymentMethod,
      order.country,
    ],
    doubles: [
      order.total,
      order.items.length,
      order.discount,
      1, // order count
    ],
    indexes: ['purchase'],
  });
}

function trackCartAbandonment(env: Env, cart: Cart) {
  env.ANALYTICS.writeDataPoint({
    blobs: ['cart_abandoned', cart.userId],
    doubles: [cart.total, cart.items.length, 1],
    indexes: ['cart'],
  });
}

SaaS

function trackApiUsage(env: Env, userId: string, endpoint: string) {
  env.ANALYTICS.writeDataPoint({
    blobs: ['api_call', userId, endpoint],
    doubles: [1], // call count
    indexes: ['api_usage'],
  });
}

function trackFeatureUsage(env: Env, feature: string, userId: string) {
  env.ANALYTICS.writeDataPoint({
    blobs: ['feature', feature, userId],
    doubles: [1],
    indexes: ['features'],
  });
}

Querying Custom Metrics

Response Time Analysis

-- P50, P95, P99 by endpoint
SELECT
  blob2 as endpoint,
  quantile(0.50)(double6) as p50_ms,
  quantile(0.95)(double6) as p95_ms,
  quantile(0.99)(double6) as p99_ms,
  COUNT(*) as requests
FROM my_metrics
WHERE index1 = 'timing'
  AND timestamp > NOW() - INTERVAL '1' HOUR
GROUP BY endpoint
ORDER BY p99_ms DESC

Business Metrics Dashboard

-- Hourly revenue
SELECT
  toStartOfHour(timestamp) as hour,
  SUM(double1) as revenue,
  SUM(double4) as orders,
  AVG(double1 / double4) as avg_order_value
FROM my_metrics
WHERE index1 = 'purchase'
  AND timestamp > NOW() - INTERVAL '24' HOUR
GROUP BY hour
ORDER BY hour

Cache Hit Rate

SELECT
  toStartOfMinute(timestamp) as minute,
  SUM(double1) as hits,
  SUM(double2) as misses,
  SUM(double1) / (SUM(double1) + SUM(double2)) * 100 as hit_rate_pct
FROM my_metrics
WHERE index1 = 'cache'
  AND timestamp > NOW() - INTERVAL '1' HOUR
GROUP BY minute
ORDER BY minute

Best Practices

1. Define schema upfront - Document which blob/double index means what 2. Use consistent naming - Same blob positions across all writes 3. Aggregate at write time - Write counts as 1, then SUM in queries 4. Use indexes for common filters - Most-queried dimension as index 5. Sample high-volume metrics - 10% sample for request-level, 100% for business

Structured Logging for Cloudflare Workers

Comprehensive guide for implementing production-grade logging in Workers.

Log Levels

Logging Best Practices

1. Always Use Structured JSON

// ❌ Bad: Plain text
console.log(`User ${userId} logged in from ${ip}`);

// ✅ Good: Structured JSON
console.log(JSON.stringify({
  level: 'info',
  event: 'user_login',
  userId,
  ip,
  timestamp: Date.now()
}));

2. Include Request Context

// Add request ID to correlate logs
export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const requestId = request.headers.get('cf-ray') || crypto.randomUUID();

    console.log(JSON.stringify({
      requestId,
      event: 'request_start',
      method: request.method,
      url: request.url,
      cf: request.cf // Cloudflare request metadata
    }));

    // Pass requestId to all functions
    return handleRequest(request, env, requestId);
  }
};

3. Redact Sensitive Data

const SENSITIVE_PATTERNS = [
  /password/i,
  /secret/i,
  /token/i,
  /authorization/i,
  /cookie/i,
  /api[_-]?key/i,
  /credit[_-]?card/i,
  /ssn/i,
];

function redactSensitive(obj: object): object {
  const redacted = { ...obj };

  for (const [key, value] of Object.entries(redacted)) {
    if (SENSITIVE_PATTERNS.some(p => p.test(key))) {
      redacted[key] = '[REDACTED]';
    } else if (typeof value === 'object' && value !== null) {
      redacted[key] = redactSensitive(value);
    }
  }

  return redacted;
}

4. Implement Log Sampling

class SampledLogger {
  constructor(
    private sampleRate: number = 0.1, // 10% sampling
    private alwaysLogLevels: string[] = ['error', 'warn']
  ) {}

  log(level: string, message: string, data?: object) {
    // Always log errors and warnings
    if (this.alwaysLogLevels.includes(level)) {
      this.write(level, message, data);
      return;
    }

    // Sample other logs
    if (Math.random() < this.sampleRate) {
      this.write(level, message, { ...data, sampled: true });
    }
  }

  private write(level: string, message: string, data?: object) {
    console.log(JSON.stringify({
      level,
      message,
      ...data,
      timestamp: Date.now()
    }));
  }
}

Production Logger Class

interface LoggerOptions {
  service: string;
  environment: string;
  sampleRate?: number;
  redactPatterns?: RegExp[];
}

export class ProductionLogger {
  private requestId: string;
  private options: Required<LoggerOptions>;
  private startTime: number;

  constructor(requestId: string, options: LoggerOptions) {
    this.requestId = requestId;
    this.startTime = Date.now();
    this.options = {
      sampleRate: 1,
      redactPatterns: [],
      ...options
    };
  }

  private shouldLog(level: string): boolean {
    if (['error', 'warn'].includes(level)) return true;
    return Math.random() < this.options.sampleRate;
  }

  private format(level: string, message: string, data?: object) {
    return JSON.stringify({
      level,
      message,
      requestId: this.requestId,
      service: this.options.service,
      environment: this.options.environment,
      timestamp: new Date().toISOString(),
      elapsed: Date.now() - this.startTime,
      ...this.redact(data || {})
    });
  }

  private redact(data: object): object {
    const result = { ...data };
    for (const [key, value] of Object.entries(result)) {
      if (this.options.redactPatterns.some(p => p.test(key))) {
        result[key] = '[REDACTED]';
      }
    }
    return result;
  }

  debug(message: string, data?: object) {
    if (this.options.environment === 'development') {
      console.log(this.format('debug', message, data));
    }
  }

  info(message: string, data?: object) {
    if (this.shouldLog('info')) {
      console.log(this.format('info', message, data));
    }
  }

  warn(message: string, data?: object) {
    console.warn(this.format('warn', message, data));
  }

  error(message: string, error?: Error, data?: object) {
    console.error(this.format('error', message, {
      ...data,
      error: error ? {
        name: error.name,
        message: error.message,
        stack: error.stack
      } : undefined
    }));
  }

  // Log request completion with timing
  complete(status: number, data?: object) {
    this.info('request_complete', {
      ...data,
      status,
      duration: Date.now() - this.startTime
    });
  }
}

Request Logging Middleware

type Handler = (request: Request, env: Env, ctx: ExecutionContext) => Promise<Response>;

export function withLogging(handler: Handler): Handler {
  return async (request, env, ctx) => {
    const logger = new ProductionLogger(
      request.headers.get('cf-ray') || crypto.randomUUID(),
      {
        service: 'my-worker',
        environment: env.ENVIRONMENT || 'production',
        sampleRate: 0.1,
        redactPatterns: [/password/i, /token/i, /secret/i]
      }
    );

    const url = new URL(request.url);

    logger.info('request_start', {
      method: request.method,
      path: url.pathname,
      query: Object.fromEntries(url.searchParams),
      userAgent: request.headers.get('user-agent'),
      country: request.cf?.country,
      colo: request.cf?.colo
    });

    try {
      const response = await handler(request, env, ctx);

      logger.complete(response.status, {
        contentType: response.headers.get('content-type')
      });

      return response;
    } catch (error) {
      logger.error('request_failed', error as Error);
      throw error;
    }
  };
}

// Usage
export default {
  fetch: withLogging(async (request, env, ctx) => {
    // Handler logic
    return new Response('OK');
  })
};

Wrangler Logging Configuration

wrangler.jsonc:

{
  "name": "my-worker",
  "observability": {
    "enabled": true,
    "head_sampling_rate": 1 // 1 = all requests, 0.1 = 10%
  }
}

Viewing Logs

Wrangler Tail (Real-time)

# All logs
wrangler tail

# Filter by status
wrangler tail --status error

# Filter by method
wrangler tail --method POST

# Filter by search term
wrangler tail --search "user_login"

# JSON format
wrangler tail --format json

Dashboard

1. Go to Workers & Pages in Cloudflare Dashboard 2. Select your worker 3. Click "Logs" tab 4. Use filters for time range, status, search

Log Size Limits

• Single log entry: 128 KB max
• Total per request: No hard limit, but affects performance
• Recommendation: Keep entries under 10 KB

// Handle large payloads
function logLargePayload(logger: ProductionLogger, payload: object) {
  const json = JSON.stringify(payload);

  if (json.length > 10000) {
    // Log summary instead
    logger.info('large_payload', {
      size: json.length,
      keys: Object.keys(payload),
      preview: json.substring(0, 500)
    });
  } else {
    logger.info('payload', payload);
  }
}

Troubleshooting

Logs Not Appearing

1. Verify observability.enabled: true in wrangler.jsonc 2. Check sampling rate isn't 0 3. Ensure using console.log/error/warn (not custom transport) 4. Wait 1-2 minutes for log propagation

Log Truncation

• Single entry limit: 128 KB
• Solution: Chunk large payloads or use external logging

Missing Correlation

• Always include requestId in every log entry
• Use cf-ray header when available
• Pass context through function calls

Tail Workers for Log Aggregation

Real-time log streaming and aggregation with Tail Workers.

Overview

Tail Workers receive logs, exceptions, and outcomes from other Workers in real-time. Use them to:

• Forward logs to external services (Datadog, Splunk, etc.)
• Filter and transform logs
• Aggregate metrics
• Trigger alerts

Configuration

Producer Worker (wrangler.jsonc):

{
  "name": "my-api-worker",
  "tail_consumers": [
    {
      "service": "log-aggregator",
      "environment": "production"
    }
  ]
}

Tail Worker (wrangler.jsonc):

{
  "name": "log-aggregator",
  "main": "src/tail-worker.ts"
}

Tail Event Structure

interface TailEvent {
  // Which worker produced this event
  scriptName: string;

  // Request/response info (if fetch handler)
  event: {
    request?: {
      url: string;
      method: string;
      headers: Record<string, string>;
      cf?: IncomingRequestCfProperties;
    };
    response?: {
      status: number;
    };
    scheduledTime?: number; // For cron triggers
    queue?: { batchSize: number }; // For queue handlers
  };

  // Console output
  logs: Array<{
    level: 'log' | 'debug' | 'info' | 'warn' | 'error';
    message: unknown[];
    timestamp: number;
  }>;

  // Uncaught exceptions
  exceptions: Array<{
    name: string;
    message: string;
    timestamp: number;
  }>;

  // Request outcome
  outcome: 'ok' | 'exception' | 'exceededCpu' | 'exceededMemory' | 'canceled' | 'unknown';

  // When the event occurred
  eventTimestamp: number;
}

Basic Tail Worker

interface Env {
  LOGGING_ENDPOINT: string;
  LOGGING_TOKEN: string;
}

export default {
  async tail(events: TailEvent[], env: Env): Promise<void> {
    const logsToSend = [];

    for (const event of events) {
      // Process each event
      const processed = {
        worker: event.scriptName,
        timestamp: event.eventTimestamp,
        outcome: event.outcome,
        request: event.event.request ? {
          method: event.event.request.method,
          url: event.event.request.url,
        } : undefined,
        response: event.event.response ? {
          status: event.event.response.status,
        } : undefined,
        logs: event.logs.map(log => ({
          level: log.level,
          message: log.message,
          timestamp: log.timestamp,
        })),
        exceptions: event.exceptions,
      };

      logsToSend.push(processed);
    }

    // Send batch to external service
    await fetch(env.LOGGING_ENDPOINT, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${env.LOGGING_TOKEN}`,
      },
      body: JSON.stringify(logsToSend),
    });
  }
};

Filtering Logs

export default {
  async tail(events: TailEvent[], env: Env): Promise<void> {
    // Only process events with errors or warnings
    const relevantEvents = events.filter(event =>
      event.outcome !== 'ok' ||
      event.exceptions.length > 0 ||
      event.logs.some(log => ['error', 'warn'].includes(log.level))
    );

    if (relevantEvents.length === 0) return;

    // Forward filtered events
    await forwardToLoggingService(relevantEvents, env);
  }
};

External Service Integrations

Datadog

async function sendToDatadog(events: TailEvent[], env: Env) {
  const logs = events.flatMap(event => [
    // Main event log
    {
      ddsource: 'cloudflare-workers',
      ddtags: `worker:${event.scriptName},outcome:${event.outcome}`,
      hostname: 'cloudflare-edge',
      message: JSON.stringify({
        type: 'request',
        request: event.event.request,
        response: event.event.response,
        outcome: event.outcome,
      }),
      status: event.outcome === 'ok' ? 'info' : 'error',
      timestamp: event.eventTimestamp,
    },
    // Individual console logs
    ...event.logs.map(log => ({
      ddsource: 'cloudflare-workers',
      ddtags: `worker:${event.scriptName},level:${log.level}`,
      hostname: 'cloudflare-edge',
      message: JSON.stringify(log.message),
      status: log.level,
      timestamp: log.timestamp,
    })),
  ]);

  await fetch('https://http-intake.logs.datadoghq.com/api/v2/logs', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'DD-API-KEY': env.DATADOG_API_KEY,
    },
    body: JSON.stringify(logs),
  });
}

Splunk

async function sendToSplunk(events: TailEvent[], env: Env) {
  const splunkEvents = events.map(event => ({
    time: event.eventTimestamp / 1000, // Splunk uses seconds
    host: 'cloudflare-workers',
    source: event.scriptName,
    sourcetype: 'cloudflare:workers',
    event: {
      outcome: event.outcome,
      request: event.event.request,
      response: event.event.response,
      logs: event.logs,
      exceptions: event.exceptions,
    },
  }));

  await fetch(`${env.SPLUNK_HEC_URL}/services/collector/event`, {
    method: 'POST',
    headers: {
      'Authorization': `Splunk ${env.SPLUNK_HEC_TOKEN}`,
      'Content-Type': 'application/json',
    },
    body: splunkEvents.map(e => JSON.stringify(e)).join('\n'),
  });
}

Elasticsearch

async function sendToElasticsearch(events: TailEvent[], env: Env) {
  const bulk = events.flatMap(event => [
    { index: { _index: 'cloudflare-workers-logs' } },
    {
      '@timestamp': new Date(event.eventTimestamp).toISOString(),
      worker: event.scriptName,
      outcome: event.outcome,
      request: event.event.request,
      response: event.event.response,
      logs: event.logs,
      exceptions: event.exceptions,
    },
  ]);

  await fetch(`${env.ELASTICSEARCH_URL}/_bulk`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/x-ndjson',
      'Authorization': `Basic ${btoa(`${env.ES_USER}:${env.ES_PASS}`)}`,
    },
    body: bulk.map(item => JSON.stringify(item)).join('\n') + '\n',
  });
}

Alerting from Tail Workers

interface Env {
  SLACK_WEBHOOK: string;
  KV: KVNamespace; // For rate limiting alerts
}

export default {
  async tail(events: TailEvent[], env: Env): Promise<void> {
    for (const event of events) {
      // Check for critical issues
      if (event.outcome === 'exception' || event.exceptions.length > 0) {
        await sendAlert(event, env);
      }
    }
  }
};

async function sendAlert(event: TailEvent, env: Env) {
  // Rate limit: 1 alert per worker per 5 minutes
  const alertKey = `alert:${event.scriptName}`;
  const lastAlert = await env.KV.get(alertKey);

  if (lastAlert) {
    const elapsed = Date.now() - parseInt(lastAlert);
    if (elapsed < 5 * 60 * 1000) return; // Skip if within 5 minutes
  }

  await env.KV.put(alertKey, String(Date.now()), { expirationTtl: 300 });

  const exception = event.exceptions[0];
  await fetch(env.SLACK_WEBHOOK, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      blocks: [
        {
          type: 'header',
          text: { type: 'plain_text', text: '🚨 Worker Exception' },
        },
        {
          type: 'section',
          fields: [
            { type: 'mrkdwn', text: `*Worker:*\n${event.scriptName}` },
            { type: 'mrkdwn', text: `*Outcome:*\n${event.outcome}` },
            { type: 'mrkdwn', text: `*Error:*\n${exception?.name}: ${exception?.message}` },
          ],
        },
      ],
    }),
  });
}

Multi-Worker Setup

// Tail multiple workers with different handling
export default {
  async tail(events: TailEvent[], env: Env): Promise<void> {
    const grouped = groupBy(events, e => e.scriptName);

    for (const [worker, workerEvents] of Object.entries(grouped)) {
      switch (worker) {
        case 'api-worker':
          await handleApiLogs(workerEvents, env);
          break;
        case 'auth-worker':
          await handleAuthLogs(workerEvents, env);
          break;
        default:
          await handleGenericLogs(workerEvents, env);
      }
    }
  }
};

function groupBy<T>(items: T[], key: (item: T) => string): Record<string, T[]> {
  return items.reduce((acc, item) => {
    const k = key(item);
    (acc[k] ||= []).push(item);
    return acc;
  }, {} as Record<string, T[]>);
}

Limits

Troubleshooting

Tail Worker Not Receiving Events

1. Verify tail_consumers config in producer worker 2. Check Tail Worker is deployed and running 3. Ensure service names match exactly 4. Check environment matches (production/staging)

Missing Logs

1. Verify producer is using console.log/error/warn 2. Check log sampling rate in producer 3. Ensure Tail Worker isn't filtering them out

Duplicate Events

• Normal during deployments
• Implement idempotency with event fingerprints