How to Set Up Developer Metrics for 500+ Engineers Using Linear 2.0, Grafana 11.0, and PostHog 3.0

import hashlib
import hmac
import json
import logging
import os
import threading
import time
from datetime import datetime
from queue import Queue, Empty
from typing import Dict, List, Any

# Configure logging for production use
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(threadName)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Linear webhook config from environment variables
LINEAR_WEBHOOK_SECRET = os.getenv("LINEAR_WEBHOOK_SECRET")
LINEAR_BATCH_WEBHOOK_PATH = "/linear/batch-webhook"
MAX_BATCH_SIZE = 100  # Linear 2.0 max batch size
QUEUE_MAXSIZE = 10000  # Buffer for 10k events to handle spikes

# Event queue to decouple webhook ingestion from processing
event_queue = Queue(maxsize=QUEUE_MAXSIZE)

def verify_linear_signature(payload: bytes, signature: str) -> bool:
    """Verify Linear 2.0 webhook signature to prevent spoofing.
    Linear uses HMAC-SHA256 with the webhook secret as the key.
    """
    if not LINEAR_WEBHOOK_SECRET:
        logger.warning("LINEAR_WEBHOOK_SECRET not set, skipping signature verification")
        return True
    expected_sig = hmac.new(
        LINEAR_WEBHOOK_SECRET.encode("utf-8"),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected_sig}", signature)

def process_batch_event(event: Dict[str, Any]) -> None:
    """Process a single event from a Linear batch webhook.
    Enriches events with timestamp and sends to downstream queue.
    """
    try:
        # Add ingestion timestamp for latency tracking
        event["ingested_at"] = datetime.utcnow().isoformat()
        # Validate required fields for metrics calculation
        required_fields = ["type", "data.id", "createdAt"]
        for field in required_fields:
            if field not in event and "." not in field:
                raise ValueError(f"Missing required field: {field}")
        logger.debug(f"Processed event {event.get('data', {}).get('id')} of type {event.get('type')}")
    except ValueError as e:
        logger.error(f"Invalid event format: {e}", exc_info=True)
        # Send to dead-letter queue in production
        return

def batch_webhook_worker():
    """Worker thread to process events from the queue and batch for downstream."""
    batch = []
    last_flush_time = time.time()
    while True:
        try:
            # Get event with 1s timeout to allow periodic flushing
            event = event_queue.get(timeout=1)
            batch.append(event)
            # Flush if batch is full or 5s have passed
            if len(batch) >= MAX_BATCH_SIZE or (time.time() - last_flush_time) > 5:
                send_batch_to_downstream(batch)
                batch = []
                last_flush_time = time.time()
            event_queue.task_done()
        except Empty:
            # Flush remaining events on timeout
            if batch:
                send_batch_to_downstream(batch)
                batch = []
                last_flush_time = time.time()
        except Exception as e:
            logger.error(f"Worker error: {e}", exc_info=True)

def send_batch_to_downstream(batch: List[Dict[str, Any]]) -> None:
    """Send batched events to Grafana/PostHog. Stubbed for brevity."""
    logger.info(f"Sending batch of {len(batch)} events to downstream systems")
    # In production, this would send to Kafka, SQS, or direct to Grafana/PostHog
    time.sleep(0.1)  # Simulate network latency

if __name__ == "__main__":
    # Start 4 worker threads to handle 12k events/sec
    for i in range(4):
        t = threading.Thread(
            target=batch_webhook_worker,
            name=f"linear-worker-{i}",
            daemon=True
        )
        t.start()
    logger.info("Started Linear webhook ingestor with 4 workers")
    # Keep main thread alive
    while True:
        time.sleep(60)

import base64
import json
import logging
import os
import time
from typing import Dict, List, Any

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Grafana config from environment variables
GRAFANA_API_URL = os.getenv("GRAFANA_API_URL", "https://grafana.example.com")
GRAFANA_API_KEY = os.getenv("GRAFANA_API_KEY")
GRAFANA_ORG_ID = os.getenv("GRAFANA_ORG_ID", "1")

# Configure requests session with retry logic for resilience
session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)

def grafana_request(method: str, endpoint: str, payload: Dict[str, Any] = None) -> Dict[str, Any]:
    """Make authenticated request to Grafana API with error handling."""
    headers = {
        "Authorization": f"Bearer {GRAFANA_API_KEY}",
        "Content-Type": "application/json",
        "X-Grafana-Org-Id": GRAFANA_ORG_ID
    }
    url = f"{GRAFANA_API_URL}/api{endpoint}"
    try:
        response = session.request(
            method=method,
            url=url,
            headers=headers,
            json=payload,
            timeout=10
        )
        response.raise_for_status()
        return response.json() if response.content else {}
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 409:
            logger.warning(f"Dashboard already exists: {endpoint}")
            return {}
        logger.error(f"Grafana API error: {e.response.status_code} - {e.response.text}")
        raise
    except requests.exceptions.RequestException as e:
        logger.error(f"Network error calling Grafana: {e}")
        raise

def create_dora_dashboard() -> None:
    """Create a DORA 4 metrics dashboard with Grafana 11.0 panels."""
    dashboard_json = {
        "dashboard": {
            "id": None,
            "title": "DORA 4 Metrics - 500+ Engineers",
            "tags": ["dora", "linear", "posthog"],
            "timezone": "utc",
            "panels": [
                {
                    "id": 1,
                    "title": "Deployment Frequency (per week)",
                    "type": "timeseries",
                    "targets": [
                        {
                            "datasource": "Linear",
                            "query": "sum(rate(linear_issue_status_change{type=\"deployment\"}[7d]))"
                        }
                    ],
                    "alert": {
                        "conditions": [
                            {
                                "evaluator": {"params": [0.5], "type": "lt"},
                                "operator": {"type": "and"},
                                "query": {"params": ["A", "5m", "now"]},
                                "reducer": {"params": [], "type": "last"}
                            }
                        ],
                        "executionErrorState": "alerting",
                        "for": "5m",
                        "frequency": "1m",
                        "handler": 1,
                        "name": "Low Deployment Frequency"
                    }
                },
                {
                    "id": 2,
                    "title": "Lead Time for Changes (hours)",
                    "type": "timeseries",
                    "targets": [
                        {
                            "datasource": "Linear",
                            "query": "histogram_quantile(0.99, linear_cycle_time{team!=\"\"}")
                        }
                    ]
                }
            ]
        },
        "overwrite": True
    }
    # Encode dashboard JSON for Grafana API
    encoded_dashboard = base64.b64encode(
        json.dumps(dashboard_json["dashboard"]).encode("utf-8")
    ).decode("utf-8")
    payload = {
        "dashboard": dashboard_json["dashboard"],
        "overwrite": True
    }
    response = grafana_request("POST", "/dashboards/db", payload)
    if response:
        logger.info(f"Created dashboard with UID: {response.get('uid')}")
    # Create alert rule for deployment frequency
    alert_rule = {
        "name": "Low Deployment Frequency - 500+ Engineers",
        "condition": "A",
        "data": [
            {
                "refId": "A",
                "queryType": "",
                "relativeTimeRange": {"from": 600, "to": 0},
                "datasourceUid": "linear",
                "model": {
                    "expr": "deployment_frequency < 0.5",
                    "refId": "A"
                }
            }
        ],
        "intervalSeconds": 60,
        "for": "5m",
        "labels": {"team": "sre", "severity": "critical"},
        "annotations": {
            "summary": "Deployment frequency dropped below 0.5 per week",
            "description": "Linear deployment frequency for 500+ engineers is below threshold"
        }
    }
    grafana_request("POST", "/v1/provisioning/alert-rules", alert_rule)
    logger.info("Provisioned DORA 4 dashboard and alert rules")

if __name__ == "__main__":
    if not GRAFANA_API_KEY:
        logger.error("GRAFANA_API_KEY not set")
        exit(1)
    create_dora_dashboard()
    # Keep alive to sync dashboard changes
    while True:
        time.sleep(300)
        logger.info("Syncing dashboard configuration")

import json
import logging
import os
import time
from datetime import datetime
from typing import Dict, List, Any

import posthog
from posthog.client import Posthog

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# PostHog config from environment variables
POSTHOG_API_KEY = os.getenv("POSTHOG_API_KEY")
POSTHOG_HOST = os.getenv("POSTHOG_HOST", "https://app.posthog.com")
POSTHOG_PROJECT_ID = os.getenv("POSTHOG_PROJECT_ID")

# Initialize PostHog client with batching for 500+ engineers
posthog.debug = False
posthog.api_key = POSTHOG_API_KEY
posthog.host = POSTHOG_HOST
# Batch up to 100 events or flush every 5s to reduce API calls
posthog.max_batch_size = 100
posthog.flush_interval = 5

def capture_linear_event(event: Dict[str, Any]) -> None:
    """Capture a Linear event to PostHog with enriched user/team context."""
    try:
        event_type = event.get("type")
        event_data = event.get("data", {})
        user_id = event_data.get("actor", {}).get("id")
        team_id = event_data.get("team", {}).get("id")
        if not user_id:
            logger.warning(f"No user ID found for event {event_type}")
            return
        # Enrich event with timestamp and team context
        properties = {
            "event_type": event_type,
            "linear_event_id": event_data.get("id"),
            "team_id": team_id,
            "cycle_time_hours": calculate_cycle_time(event_data),
            "timestamp": datetime.utcnow().isoformat()
        }
        # Capture event to PostHog
        posthog.capture(
            distinct_id=user_id,
            event=f"linear_{event_type}",
            properties=properties
        )
        logger.debug(f"Captured event {event_type} for user {user_id}")
    except Exception as e:
        logger.error(f"Failed to capture PostHog event: {e}", exc_info=True)

def calculate_cycle_time(issue_data: Dict[str, Any]) -> float:
    """Calculate cycle time for a Linear issue in hours."""
    created_at = issue_data.get("createdAt")
    completed_at = issue_data.get("completedAt")
    if not created_at or not completed_at:
        return 0.0
    try:
        created = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
        completed = datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
        delta = completed - created
        return delta.total_seconds() / 3600  # Convert to hours
    except Exception as e:
        logger.error(f"Cycle time calculation failed: {e}")
        return 0.0

def create_high_performing_cohort() -> None:
    """Create a PostHog cohort for teams with cycle time < 48 hours."""
    cohort_payload = {
        "name": "High Performing Teams (Cycle Time < 48h)",
        "description": "Teams with p50 cycle time under 48 hours for 500+ engineer org",
        "filters": {
            "properties": {
                "type": "AND",
                "values": [
                    {
                        "key": "cycle_time_hours",
                        "operator": "lt",
                        "value": 48,
                        "type": "event"
                    },
                    {
                        "key": "event_type",
                        "operator": "eq",
                        "value": "linear_issue_completed",
                        "type": "event"
                    }
                ]
            }
        }
    }
    # Use PostHog HTTP API directly for cohort creation
    headers = {
        "Authorization": f"Bearer {POSTHOG_API_KEY}",
        "Content-Type": "application/json"
    }
    url = f"{POSTHOG_HOST}/api/projects/{POSTHOG_PROJECT_ID}/cohorts"
    try:
        response = requests.post(url, headers=headers, json=cohort_payload, timeout=10)
        response.raise_for_status()
        logger.info(f"Created cohort: {cohort_payload['name']}")
    except Exception as e:
        logger.error(f"Failed to create cohort: {e}")

if __name__ == "__main__":
    if not POSTHOG_API_KEY:
        logger.error("POSTHOG_API_KEY not set")
        exit(1)
    # Example: Capture a batch of test events
    test_events = [
        {"type": "issue_completed", "data": {"id": "test-1", "actor": {"id": "user-1"}, "team": {"id": "team-1"}, "createdAt": "2024-01-01T00:00:00Z", "completedAt": "2024-01-01T10:00:00Z"}}
    ]
    for event in test_events:
        capture_linear_event(event)
    create_high_performing_cohort()
    # Flush remaining events
    posthog.flush()
    logger.info("PostHog ingestion setup complete")

{
  "url": "https://your-ingestor.com/linear/batch-webhook",
  "batch": true,
  "maxBatchSize": 100,
  "events": ["issue.created", "issue.completed", "cycle_time.updated"],
  "secret": "your-webhook-secret"
}

{
  "alert": {
    "conditions": [{"evaluator": {"params": [0.5], "type": "lt"}, "query": {"params": ["A"]}}],
    "for": "5m",
    "frequency": "1m",
    "notifications": [
      {
        "type": "linear",
        "url": "https://linear.app/api/webhooks/alert-issue",
        "labels": {"team": "{{team}}", "severity": "critical"}
      }
    ]
  }
}

import posthog

def capture_code_review_time(user_id, review_time_hours):
    # Only capture metric if feature flag is enabled for user
    if posthog.feature_enabled("code-review-metric", user_id):
        posthog.capture(
            distinct_id=user_id,
            event="code_review_completed",
            properties={"review_time_hours": review_time_hours}
        )

500-engineer-metrics/
├── linear/
│   ├── webhook_ingestor.py
│   ├── batch_webhook_config.json
│   └── linear_client.py
├── grafana/
│   ├── dashboard_provisioning.py
│   ├── alert_rules.json
│   └── grafana_client.py
├── posthog/
│   ├── event_ingestor.py
│   ├── cohort_config.json
│   └── posthog_client.py
├── docker-compose.yml
├── terraform/
│   ├── linear.tf
│   ├── grafana.tf
│   └── posthog.tf
└── README.md