scarf-analytics-pipeline

A scheduled pipeline that pulls Fastly CDN traffic logs from Grafana Loki and ships them as telemetry events to Scarf, giving OpenVSX visibility into package download and general request analytics.

How it works

A Kubernetes CronJob runs scripts/sync.py on a fixed interval (default: every 5 minutes). Each run:

1. Determines the query window. The end of the last successful run is read from a checkpoint ConfigMap (scarf-sync-checkpoint by default). Each window is SYNC_INTERVAL_MINUTES long, starting from the checkpoint (capped at the current time). If no checkpoint exists yet, it queries the last SYNC_INTERVAL_MINUTES.
2. Fetches logs from Loki via query_range for that window.
3. Parses each log line into a Scarf event, typed download or request depending on whether the URL looks like a download, deduplicated by a SHA-256 hash of the raw log line ($unique_id), and filters out lines missing both an IP and a user agent.
4. Ships events to Scarf in batches (SCARF_BATCH_SIZE, default 500) via the Scarf v2 import API.
5. Advances the checkpoint only if every batch shipped successfully. If any batch fails, the checkpoint is left unchanged so the next run retries the same window (safe, since Scarf dedupes on $unique_id).
6. Repeats with the next window until it reaches the current time. If the job falls behind (e.g. after an outage), a single run works through the backlog one window at a time. A failed window stops the run, and the next run picks up from the last window that succeeded.

Repository layout

scripts/
  sync.py             # The sync job itself
  requirements.txt    # Python dependencies
tests/
  test_sync.py        # Unit tests for sync.py
  requirements.txt    # Test dependencies (runtime deps + pytest)
charts/               # Helm chart deploying the CronJob, RBAC, etc.
kubernetes/
  helm-deploy.sh       # Deploys the Helm chart to a target environment
  namespace-rbac.yaml  # Namespace-scoped RBAC needed for Jenkins to deploy
Dockerfile            # Multi-stage build producing the sync-job image
Jenkinsfile           # CI: build, push, and deploy to staging on main

Configuration

sync.py is configured entirely through environment variables, set via charts/values.yaml and a scarf-loki-credentials secret in the target namespace.

Variable	Description	Default
LOKI_URL	Base URL of the Grafana Loki instance (secret)	—
LOKI_USER	Loki basic-auth username (secret)	—
LOKI_API_KEY	Loki basic-auth API key (secret)	—
LOKI_QUERY	LogQL query selecting the log stream to sync	—
SCARF_API_TOKEN	Scarf API token (secret)	—
SCARF_ENTITY_ID	Scarf package/entity ID to import events into	—
ORGANIZATION_NAME	Scarf organization name	OpenVSX
SCARF_BATCH_SIZE	Max events per Scarf import request	500
SYNC_INTERVAL_MINUTES	Window size queried by every run, starting from the checkpoint; must match the CronJob schedule	5
SLACK_WEBHOOK_URL	Slack incoming webhook that failed runs are reported to; alerting is disabled if unset (secret)	—
ALERT_AFTER_CONSECUTIVE_FAILURES	Consecutive failed runs before an alert is sent	1
LAG_ALERT_MINUTES	Alert when the sync is more than this many minutes behind real time and not catching up	60
DEPLOY_ENVIRONMENT	Environment name shown in alerts (set from environment in values.yaml)	unknown
CHECKPOINT_CONFIGMAP_NAME	Name of the ConfigMap used to persist the sync checkpoint	scarf-sync-checkpoint

LOKI_URL, LOKI_USER, LOKI_API_KEY, SCARF_API_TOKEN, SCARF_ENTITY_ID, and SLACK_WEBHOOK_URL are expected to come from the scarf-loki-credentials Kubernetes secret (referenced via envFrom in the CronJob template) rather than values.yaml.

Alerting

Any failed run (missing config, Loki errors, a Scarf batch that still fails after retries, or an unexpected exception) posts to SLACK_WEBHOOK_URL. The count of consecutive failed runs is kept in the checkpoint ConfigMap, so an outage produces one alert (once ALERT_AFTER_CONSECUTIVE_FAILURES is reached), not one every run, and a recovery message follows the next successful run.

A second alert covers the job falling behind without failing: if the checkpoint is more than LAG_ALERT_MINUTES behind real time and windows are taking longer to sync than the time they cover, it posts once, then again when it catches up. Working off a backlog after an outage doesn't trigger it, as long as the job is gaining ground.

This only covers failures sync.py can see. If the pod never runs or is killed (image pull errors, OOMKilled), no alert is sent.

Running locally

pip install -r scripts/requirements.txt

export LOKI_URL=https://logs-prod-018.grafana.net/
export LOKI_USER=...
export LOKI_API_KEY=...
export LOKI_QUERY='{service_name="fastly_cdn", env="production"}'
export SCARF_API_TOKEN=...
export SCARF_ENTITY_ID=...

python scripts/sync.py

Outside a cluster, checkpoint reads/writes fail gracefully (no in-cluster service account is available), so each local run falls back to the default SYNC_INTERVAL_MINUTES window.

Running tests

pip install -r tests/requirements.txt
python -m pytest tests

The tests in tests/test_sync.py fake Loki, Scarf, Slack, the checkpoint ConfigMap and the clock, so they run offline in well under a second.

Deployment

The Helm chart in charts/ deploys the CronJob along with a dedicated ServiceAccount and a Role/RoleBinding scoped to just the checkpoint ConfigMap.

./kubernetes/helm-deploy.sh staging <docker_image_tag>

See kubernetes/README.md for one-time cluster setup (applying namespace-rbac.yaml for Jenkins).

CI/CD is handled by the Jenkinsfile: every build produces and pushes a Docker image to ghcr.io/eclipsefdn/scarf-analytics, and pushes to main automatically deploy to staging via helm-deploy.sh.