Send logs & metrics from GCP to Epok

Path 1: Cloud Logging → Pub/Sub → Cloud Function

Standard pattern. A Logs Router sink ships every matching log to a Pub/Sub topic; a Cloud Function consumes the topic and POSTs batches to Epok. Works for Cloud Run, GKE, Compute Engine, App Engine, and any service that writes to Cloud Logging.

1. Create the Pub/Sub topic

gcloud pubsub topics create epok-logs

2. Create a Logs Router sink that writes to the topic

# Send every log from project PROJECT_ID to the topic. Narrow with
# --log-filter='resource.type="cloud_run_revision"' if you only want
# a subset.

gcloud logging sinks create epok-sink \
  pubsub.googleapis.com/projects/PROJECT_ID/topics/epok-logs \
  --log-filter='severity >= DEFAULT' \
  --description='Forward Cloud Logging entries to Epok'

# Grant the sink's writer identity permission to publish.
WRITER=$(gcloud logging sinks describe epok-sink --format='value(writerIdentity)')
gcloud pubsub topics add-iam-policy-binding epok-logs \
  --member="$WRITER" --role=roles/pubsub.publisher

3. Deploy the consumer Cloud Function (Python 3.12)

# main.py — Pub/Sub-triggered consumer for epok-logs
import base64
import json
import os
import urllib.request

EPOK_ENDPOINT = "https://ingest.getepok.dev/insert/elasticsearch/_bulk"
EPOK_API_KEY = os.environ["EPOK_API_KEY"]


def forward(event, _context):
    """Triggered by a single Pub/Sub message."""
    raw = base64.b64decode(event["data"]).decode("utf-8")
    entry = json.loads(raw)

    body = (
        json.dumps({"create": {}}) + "\n" +
        json.dumps({
            "_msg": entry.get("textPayload") or json.dumps(entry.get("jsonPayload", {})),
            "_time": entry.get("timestamp"),
            "severity": entry.get("severity", "DEFAULT").lower(),
            "service": (entry.get("resource", {}).get("labels", {}).get("service_name")
                        or entry.get("logName", "").split("/")[-1]),
            "resource_type": entry.get("resource", {}).get("type"),
        }) + "\n"
    ).encode("utf-8")

    req = urllib.request.Request(
        EPOK_ENDPOINT,
        data=body,
        headers={
            "Authorization": f"Bearer {EPOK_API_KEY}",
            "Content-Type": "application/json",
        },
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        return {"status": resp.status}

gcloud functions deploy epok-forwarder \
  --gen2 \
  --runtime=python312 \
  --region=us-central1 \
  --trigger-topic=epok-logs \
  --entry-point=forward \
  --set-env-vars=EPOK_API_KEY=epk_REPLACE_ME \
  --max-instances=10

Note: GCP's severity field uses the labels DEBUG, INFO, NOTICE, WARNING, ERROR, CRITICAL, ALERT, EMERGENCY. Epok recognizes these natively — but if you want a different mapping (e.g. NOTICE → info, ALERT → critical), set it up under Settings → Log Processing → Level Mapping Rules.

Path 2: Vector on GCE / GKE node

Skip Cloud Logging entirely if you control the host. Vector tails files or systemd journal and ships directly. Cheaper than the Pub/Sub path at high volume (no Cloud Logging egress fees).

# /etc/vector/vector.yaml — GCE/GKE host → Epok

sources:
  journal:
    type: journald
    include_units: [my-app.service, nginx.service]
    current_boot_only: false

  app:
    type: file
    include: ["/var/log/myapp/*.log"]

transforms:
  enrich:
    type: remap
    inputs: [journal, app]
    source: |
      .host = get_hostname!()
      .region = "us-central1"
      .cloud = "gcp"

sinks:
  epok:
    type: elasticsearch
    inputs: [enrich]
    endpoints: ["https://ingest.getepok.dev"]
    bulk:
      index: logs
    # Required. Epok is not a real Elasticsearch: the default api_version
    # "auto" sniffs the version against our root (404) and the sink then
    # never starts, so nothing ships and nothing says why.
    api_version: v8
    healthcheck:
      enabled: false
    auth:
      strategy: basic
      user: ${EPOK_API_KEY}
      password: x

For GKE specifically (DaemonSet manifests, RBAC), use the Kubernetes install guide instead — same Vector binary, simpler delivery.

Path 3: Cloud Monitoring metrics → Epok

For infra metrics — GCE CPU, Cloud SQL utilization, load-balancer latency, Pub/Sub backlog — run the OpenTelemetry Collector's googlecloudmonitoring receiver with a read-only service account and push to Epok over OTLP. The GCP credentials stay in your environment; Epok never holds them. Once the series land, Epok's metric detectors (saturation, anomaly, slow-drift, reporting-gap) run on them automatically.

# otel-cloudmonitoring.yaml — pull Cloud Monitoring metrics → Epok
receivers:
  googlecloudmonitoring:
    project_id: PROJECT_ID
    collection_interval: 5m
    metrics_list:
      - metric_name: "compute.googleapis.com/instance/cpu/utilization"
      - metric_name: "cloudsql.googleapis.com/database/cpu/utilization"

processors:
  resource:
    attributes:
      - key: service.name        # the service these metrics belong to
        value: gcp-infra
        action: upsert

exporters:
  otlphttp/epok:
    metrics_endpoint: https://ingest.getepok.dev/v1/metrics
    encoding: json               # set this — correct against every build
    headers: { Authorization: "Bearer ${EPOK_API_KEY}" }

service:
  pipelines:
    metrics:
      receivers: [googlecloudmonitoring]
      processors: [resource]
      exporters: [otlphttp/epok]

The collector's service account needs roles/monitoring.viewer. Cloud Monitoring bills per time series read, so list the metrics you care about. Confirm in Epok → Metrics within one collection interval.

Verify

1. Open app.getepok.dev → Live Tail. Within 60–120 seconds (Pub/Sub adds a few seconds of buffering) you should see log lines.
2. Open Services — entries should appear keyed on the resource_type (e.g. cloud_run_revision, k8s_container).
3. If your severitylabels aren't mapping cleanly, check Settings → Log Processing → Level Mapping Rules and add field-value rules for any non-standard labels.

Common gotchas

• Logs Router sink scope. Sinks are project-scoped by default. For a multi-project org, create the sink at the folder or organization level (gcloud logging sinks create --organization=ORG_ID).
• Cloud Function cold starts. First trigger after an idle period can add ~1–3 s of latency. Set --min-instances=1 for sub-second forwarding if it matters.
• Pub/Sub message size.10 MB per message. Cloud Logging entries are usually well under this; if you're shipping binary payloads, configure the sink's --output-version-format=V2.
• JSON payloads. Cloud Run / GKE structured logs arrive with jsonPayload instead of textPayload. The forwarder above serializes the whole object — if you'd rather pick fields, customize the_msg derivation in the function.

When the key is rejected (401 vs 403)

The status tells you which half of the problem you have. 401 with the body {"error":"unauthorized","detail":"..."} means no credential was read at all — a missing header, an unexpanded variable, or a scheme the log-ingest endpoints do not parse. 403 with {"error":"forbidden","detail":"Invalid API key. ..."} means a credential was read and then refused: unknown or rotated key, expired key, or a key without the ingest scope. So the ordinary "wrong key" is a 403, not a 401. On GCP the Pub/Sub path is the one that hides both, and it hides them in a particularly expensive way.

• Path 1 (Cloud Function). urllib.request.urlopen raises on any non-2xx, so the function invocation fails with HTTPError: HTTP Error 403: Forbidden (or 401if the header never got set at all) — visible in the function's own Cloud Logging entries, not in the logs you were forwarding. Because the function fails, Pub/Sub does not acknowledge the message and redelivers it. A wrong key therefore looks like a Cloud Function error rate climbing and a subscription backlog growing, never like an auth problem. If your epok-logs subscription is accumulating unacked messages, check the key before you check anything else.
• Path 2 (Vector). Because healthcheck.enabled: false is required for this sink (see Path 2 above), Vector starts cleanly with a dead key — nothing fails at boot and nothing is printed at startup. The rejection shows up only as per-request sink errors: journalctl -u vector -f.
• Path 3 (collector). The collector logs the failing export with the response status, so this is the one path where the status code appears verbatim.

The log-ingest endpoints (Paths 1–2) accept exactly three credential headers: X-API-Key: <key>, Authorization: Bearer <key>, and Authorization: Basic <base64> with the key as the username (any password — the Vector config above uses x). A ?api_key= query parameter also works, but only on drain-style sources that cannot set headers. Notably Authorization: Token <key> is not accepted there — it parses as no credential at all, so a correct key sent that way still returns 401. Path 3 is the exception: /v1/metrics is served by the app plane, which does accept Token. The plane is decided by the path, not the hostname. Full per-plane matrix: authentication.

A 403 is good news during setup. It proves your header shape is right and the plane read your credential — the key itself is then the thing to fix: wrong or rotated key, expired key, or a key missing the ingest scope (Settings → API Keys). On the search call below, 403 also covers a key without the read scope and a tenant ID in the path that the key does not belong to. Plan and feature gates are a separate matter, covered in limits.

Your first query

Seeing bytes move is not the same as being able to ask a question. Run this against the logs you just wired up — the Path 1 forwarder tags every entry with resource_type carried over from Cloud Logging, so asking for lines where that field exists returns exactly what came through the Logs Router sink.

curl -X POST https://app.getepok.dev/api/v1/tenants/YOUR_TENANT_ID/search \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "resource_type:*",
    "start": "-1h",
    "end":   "now",
    "limit": 100
  }'

Your tenant ID is not printed on the API Keys screen. The quickest way to read it: while signed in, open app.getepok.dev/auth/me — the JSON it returns includes tenant_id. The body accepts five fields and no others: query, start, end, limit (max 10000), and before — an ISO timestamp cursor for paging to older logs.

A good result is a 200 whose count is non-zero and whose logs entries carry a _time inside the last hour plus a resource_type you recognise — cloud_run_revision, k8s_container, gce_instance. Narrow to one workload with resource_type:cloud_run_revision, or go straight at the errors with severity:error(the forwarder lowercases GCP's severity labels). On the Vector path substitute cloud:gcp, which the enrich transform sets.

A 200 with "count": 0 is not an auth failure. It means the request was authenticated and routed correctly and nothing matched. Re-run with "query": "*" before you assume delivery is broken — if *returns rows, your logs are landing and it is the field names that differ from what you asked for. This matters more on GCP than elsewhere: the sink's --log-filter may simply not be matching anything.

The same query in the product: open app.getepok.dev → Explore, paste resource_type:*, set the range to the last hour. Live Tail shows you the stream; Explore is where you interrogate it.