用 Prometheus 和 Grafana 监控网页抓取把抓取结果、目标站点的回答和账单分开

https://example.com/does-not-exist
  http_status       200
  cb_status         200
  original_status   404
  domain_complexity standard
  seconds           1.4

https://www.glassdoor.com/Reviews/index.htm
  http_status       520
  cb_status         705
  original_status   403
  domain_complexity standard
  seconds           20.3

def get(self, url: str, **params: Any) -> CrawlResult:
    domain = self._limit_domain(domain_label(url))
    token_type = "js" if str(params.get("javascript", "")).lower() == "true" else self.token_type
    success_codes = self.custom_success_codes | parse_success_codes(params.get("custom_success_codes"))

    self._ensure_series(domain, token_type)
    started = self._clock()
    try:
        response = self._http.get(
            f"{self.base_url}/", params={"token": self._token, "url": url, **params}
        )
    except httpx.HTTPError as exc:
        self._record_failure(domain, token_type, exc)
        raise
    elapsed = self._clock() - started

    json_body = None
    if params.get("format") == "json":
        try:
            json_body = response.json()
        except ValueError:
            json_body = None
    signals = parse_signals(response.status_code, response.headers, json_body)
    billable = is_billable(signals.cb_status, signals.original_status, success_codes)

    # Counted on completion, next to the outcome, so success <= attempts in any window.
    self.metrics.attempts.labels(domain=domain, token_type=token_type).inc()
    self.metrics.requests.labels(
        domain=domain,
        cb_status=status_label(signals.cb_status),
        original_status=status_label(signals.original_status),
        http_status=status_label(signals.http_status),
        token_type=token_type,
        billable=str(billable).lower(),
    ).inc()
    self.metrics.latency.labels(domain=domain, complexity=signals.complexity).observe(elapsed)
    self.metrics.response_bytes.labels(domain=domain).observe(len(response.content))

    return CrawlResult(url, domain, token_type, signals, billable, elapsed, response.content)

def _record_failure(self, domain: str, token_type: str, exc: httpx.HTTPError) -> None:
    reason = "timeout" if isinstance(exc, httpx.TimeoutException) else "transport"
    self.metrics.attempts.labels(domain=domain, token_type=token_type).inc()
    self.metrics.client_errors.labels(domain=domain, token_type=token_type, reason=reason).inc()
    # Count it as a request too, so success-rate denominators include it.
    self.metrics.requests.labels(
        domain=domain, cb_status="none", original_status="none",
        http_status="none", token_type=token_type, billable="false",
    ).inc()

def preregister(self, urls: Iterable[str], token_type: str | None = None) -> None:
    """Create zero-valued series before traffic starts.

    Prometheus only sees a counter once it exists. A series born at 1 between
    two scrapes shows no increase, so the first failure would be invisible.
    """
    for url in urls:
        self._ensure_series(self._limit_domain(domain_label(url)), token_type or self.token_type)

def _ensure_series(self, domain: str, token_type: str) -> None:
    if (domain, token_type) in self._known:
        return
    self._known.add((domain, token_type))
    self.metrics.attempts.labels(domain=domain, token_type=token_type)
    self.metrics.requests.labels(
        domain=domain, cb_status="200", original_status="200",
        http_status="200", token_type=token_type, billable="true",
    )

CHARGED_CB_STATUSES = frozenset({200, 207})
CHARGED_ORIGINAL_STATUSES = frozenset({200, 201, 301, 302, 303, 307, 308, 404, 410})


def is_billable(
    cb_status: int | None,
    original_status: int | None,
    custom_success_codes: Iterable[int] = (),
) -> bool:
    if cb_status not in CHARGED_CB_STATUSES or original_status is None:
        return False
    return original_status in CHARGED_ORIGINAL_STATUSES or original_status in set(custom_success_codes)

crawlbase_requests_total{billable="true",cb_status="200",domain="example.com",http_status="200",original_status="404",token_type="normal"} 9.0
crawlbase_requests_total{billable="false",cb_status="576",domain="glassdoor.com",http_status="520",original_status="403",token_type="normal"} 3.0

def poll(self, metrics: AccountMetrics | None = None) -> AccountSnapshot | None:
    """Return a fresh snapshot if allowed, otherwise the cached one."""
    if self.seconds_until_allowed() > 0:
        if metrics:
            metrics.cache_hits.inc()
        return self._cached

    self._last_call = self._clock()
    self._save_state()
    try:
        response = self._http.get(
            f"{self.base_url}/account", params={"token": self._token, "product": self.product}
        )
        response.raise_for_status()
        snapshot = parse_account(response.json())
    except (httpx.HTTPError, ValueError) as exc:
        result = "timeout" if isinstance(exc, httpx.TimeoutException) else "error"
        status = exc.response.status_code if isinstance(exc, httpx.HTTPStatusError) else None
        log.warning("Account API call failed (%s): %s status=%s", result, type(exc).__name__, status)
        if metrics:
            metrics.api_calls.labels(result=result).inc()
        return self._cached

    self._cached = snapshot
    self._save_state()
    if metrics:
        metrics.api_calls.labels(result="ok").inc()
    return snapshot

{
  "currentMonth": {
    "totalSuccess": "<str>",
    "totalFailed": "<str>",
    "totalDue": "<float>",
    "domainStats": [
      {
        "domain": "<str>",
        "metrics": {
          "totalRequests": "<int>",
          "success": "<int>",
          "failure": "<int>",
          "successRate": "<str>"
        }
      }
    ]
  }
}

metrics.total_success.set(snapshot.total_success)
metrics.total_failed.set(snapshot.total_failed)
metrics.total_due.set(snapshot.total_due)
if snapshot.remaining_credits is not None:
    metrics.remaining_credits.labels(source="account_api").set(snapshot.remaining_credits)
elif monthly_budget:
    metrics.remaining_credits.labels(source="budget").set(
        max(0.0, monthly_budget - snapshot.total_success)
    )

global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - /etc/prometheus/rules/*.yml

scrape_configs:
  - job_name: crawlbase-scraper
    static_configs:
      - targets: ["scraper:8000"]

  # The gauges only change every 5 minutes, a slower scrape is plenty.
  - job_name: crawlbase-account
    scrape_interval: 60s
    static_configs:
      - targets: ["reconciler:8001"]

# Success = the crawl worked (cb_status 200). The denominator is
# crawlbase_attempts_total, which also counts timeouts and is created at 0.
- record: crawlbase:domain_success_ratio:rate5m
  expr: |
    clamp_max(
      sum by (domain) (rate(crawlbase_requests_total{cb_status="200"}[5m]))
      /
      sum by (domain) (rate(crawlbase_attempts_total[5m])),
      1
    )

timeseries("Success rate per domain", [target("crawlbase:domain_success_ratio:rate5m", "{{domain}}")],
           0, 6, unit="percentunit", minv=0, maxv=1,
           desc="cb_status 200 share per domain, 5 minute window. Alert fires under 90% for 10m."),

- uid: cb-domain-success-low
  title: Crawlbase domain success rate below 90%
  condition: C
  for: 10m
  noDataState: OK
  execErrState: Error
  labels:
    severity: page
  annotations:
    summary: "Success rate for {{ $labels.domain }} is {{ humanizePercentage $values.B.Value }}"
    description: >-
      cb_status 200 share has been under 90% for 10 minutes. Open the cb_status
      breakdown panel: 5xx/6xx codes are the crawl layer, 704/705/706 mean the
      target site said no (check original_status).
    runbook_url: https://crawlbase.com/docs/status-codes/
  data:
    - refId: A
      relativeTimeRange: { from: 600, to: 0 }
      datasourceUid: prometheus
      model:
        refId: A
        instant: true
        # Ignore domains with almost no traffic (under ~1 request per 2 minutes).
        expr: crawlbase:domain_success_ratio:rate5m and on (domain) crawlbase:domain_requests:rate5m > 0.008
    - refId: B
      datasourceUid: __expr__
      model: { refId: B, type: reduce, expression: A, reducer: last }
    - refId: C
      datasourceUid: __expr__
      model:
        refId: C
        type: threshold
        expression: B
        conditions:
          - evaluator: { type: lt, params: [0.9] }

# Filled from the alert: Normal token gets cb_status 503/576/705 on these.
JS_DOMAINS = {"glassdoor.com"}

URLS = [
    "https://example.com/",
    "https://www.glassdoor.com/Reviews/index.htm",
]


def main() -> None:
    load_dotenv(find_dotenv(usecwd=True))
    clients = {
        "normal": InstrumentedCrawlingAPI(os.environ["CRAWLBASE_NORMAL_TOKEN"], "normal"),
        "js": InstrumentedCrawlingAPI(os.environ["CRAWLBASE_JS_TOKEN"], "js"),
    }
    for url in URLS:
        token_type = "js" if domain_label(url) in JS_DOMAINS else "normal"
        r = clients[token_type].get(url)
        print(f"{token_type:<6} http={r.status_code} cb_status={r.headers.get('cb_status')} {url}")

cd final
cp .env.example .env          # add CRAWLBASE_NORMAL_TOKEN and CRAWLBASE_JS_TOKEN
docker compose up -d --build
# Grafana: http://localhost:3000/d/crawlbase-observability (admin / admin)