feat: add PubSub → BigQuery log pipeline addons

addons/gcp/fleet-pubsub/main.tf

@@ -0,0 +1,35 @@
+resource "google_pubsub_topic" "result" {
+  project = var.project_id
+  name    = var.result_topic_name
+}
+
+resource "google_pubsub_topic" "status" {
+  project = var.project_id
+  name    = var.status_topic_name
+}
+
+resource "google_pubsub_topic" "audit" {
+  project = var.project_id
+  name    = var.audit_topic_name
+}
+
+resource "google_pubsub_topic_iam_member" "fleet_result_publisher" {
+  project = var.project_id
+  topic   = google_pubsub_topic.result.name
+  role    = "roles/pubsub.publisher"
+  member  = "serviceAccount:${var.fleet_sa_email}"
+}
+
+resource "google_pubsub_topic_iam_member" "fleet_status_publisher" {
+  project = var.project_id
+  topic   = google_pubsub_topic.status.name
+  role    = "roles/pubsub.publisher"
+  member  = "serviceAccount:${var.fleet_sa_email}"
+}
+
+resource "google_pubsub_topic_iam_member" "fleet_audit_publisher" {
+  project = var.project_id
+  topic   = google_pubsub_topic.audit.name
+  role    = "roles/pubsub.publisher"
+  member  = "serviceAccount:${var.fleet_sa_email}"
+}

addons/gcp/fleet-pubsub/outputs.tf

@@ -0,0 +1,28 @@
+output "result_topic_name" {
+  description = "Name of the PubSub topic for osquery result logs"
+  value       = google_pubsub_topic.result.name
+}
+
+output "status_topic_name" {
+  description = "Name of the PubSub topic for osquery status logs"
+  value       = google_pubsub_topic.status.name
+}
+
+output "audit_topic_name" {
+  description = "Name of the PubSub topic for Fleet audit logs"
+  value       = google_pubsub_topic.audit.name
+}
+
+output "fleet_env_vars" {
+  description = "Map of Fleet env vars to enable PubSub logging. Merge into fleet_config.extra_env_vars."
+  value = {
+    FLEET_OSQUERY_RESULT_LOG_PLUGIN = "pubsub"
+    FLEET_OSQUERY_STATUS_LOG_PLUGIN = "pubsub"
+    FLEET_ACTIVITY_ENABLE_AUDIT_LOG = "true"
+    FLEET_PUBSUB_PROJECT            = var.project_id
+    FLEET_PUBSUB_RESULT_TOPIC       = google_pubsub_topic.result.name
+    FLEET_PUBSUB_STATUS_TOPIC       = google_pubsub_topic.status.name
+    FLEET_PUBSUB_AUDIT_TOPIC        = google_pubsub_topic.audit.name
+    FLEET_PUBSUB_ADD_ATTRIBUTES     = "true"
+  }
+}

addons/gcp/fleet-pubsub/variables.tf

@@ -0,0 +1,27 @@
+variable "project_id" {
+  description = "GCP project ID where PubSub topics are created"
+  type        = string
+}
+
+variable "fleet_sa_email" {
+  description = "Email of the Fleet Cloud Run service account (fleet-run-sa). Granted pubsub.publisher on all topics."
+  type        = string
+}
+
+variable "result_topic_name" {
+  description = "Name of the PubSub topic for osquery result logs"
+  type        = string
+  default     = "fleet-result-logs"
+}
+
+variable "status_topic_name" {
+  description = "Name of the PubSub topic for osquery status logs"
+  type        = string
+  default     = "fleet-status-logs"
+}
+
+variable "audit_topic_name" {
+  description = "Name of the PubSub topic for Fleet audit logs"
+  type        = string
+  default     = "fleet-audit-logs"
+}

addons/gcp/fleet-pubsub/versions.tf

@@ -0,0 +1,9 @@
+terraform {
+  required_version = "~> 1.11"
+  required_providers {
+    google = {
+      source  = "hashicorp/google"
+      version = ">= 6.35.0"
+    }
+  }
+}

addons/gcp/pubsub-to-bigquery/iam.tf

@@ -0,0 +1,50 @@
+data "google_project" "project" {
+  project_id = var.project_id
+}
+
+# Service account used as the Cloud Run service identity.
+# Needs BQ dataEditor + jobUser to write rows.
+resource "google_service_account" "ingest_sa" {
+  project      = var.project_id
+  account_id   = "fleet-pubsub-bq-sa"
+  display_name = "Fleet PubSub→BQ Ingest Service"
+  description  = "Identity for the fleet-pubsub-bq Cloud Run service"
+}
+
+# Service account that PubSub uses to generate OIDC tokens for push auth.
+resource "google_service_account" "pubsub_invoker_sa" {
+  project      = var.project_id
+  account_id   = "fleet-pubsub-invoker-sa"
+  display_name = "Fleet PubSub Push Invoker"
+  description  = "Used by PubSub push subscriptions to authenticate against the ingest Cloud Run service"
+}
+
+# Allow PubSub service agent to create OIDC tokens for the invoker SA.
+# Required for projects created before April 8, 2021; harmless for newer projects.
+resource "google_service_account_iam_member" "pubsub_token_creator" {
+  service_account_id = google_service_account.pubsub_invoker_sa.name
+  role               = "roles/iam.serviceAccountTokenCreator"
+  member             = "serviceAccount:service-${data.google_project.project.number}@gcp-sa-pubsub.iam.gserviceaccount.com"
+}
+
+# BQ dataEditor on the dataset lets the ingest SA insert rows.
+resource "google_bigquery_dataset_iam_member" "ingest_sa_editor" {
+  project    = local.bq_project_id
+  dataset_id = google_bigquery_dataset.fleet_logs.dataset_id
+  role       = "roles/bigquery.dataEditor"
+  member     = "serviceAccount:${google_service_account.ingest_sa.email}"
+}
+
+# BQ jobUser at project level lets the ingest SA run jobs (needed for streaming inserts).
+resource "google_project_iam_member" "ingest_sa_bq_job_user" {
+  project = local.bq_project_id
+  role    = "roles/bigquery.jobUser"
+  member  = "serviceAccount:${google_service_account.ingest_sa.email}"
+}
+
+# Standard Cloud Run logging
+resource "google_project_iam_member" "ingest_sa_log_writer" {
+  project = var.project_id
+  role    = "roles/logging.logWriter"
+  member  = "serviceAccount:${google_service_account.ingest_sa.email}"
+}

addons/gcp/pubsub-to-bigquery/main.tf

@@ -0,0 +1,223 @@
+# -------------------------------------
+# BigQuery
+# -------------------------------------
+
+resource "google_bigquery_dataset" "fleet_logs" {
+  project    = local.bq_project_id
+  dataset_id = var.bq_dataset_id
+  location   = "US"
+
+  labels = {
+    managed-by = "terraform"
+    app        = "fleet"
+  }
+}
+
+resource "google_bigquery_table" "result_logs" {
+  project             = local.bq_project_id
+  dataset_id          = google_bigquery_dataset.fleet_logs.dataset_id
+  table_id            = "result_logs"
+  deletion_protection = false
+
+  time_partitioning {
+    type  = "DAY"
+    field = "unix_time"
+  }
+
+  clustering = ["query_name", "host_identifier"]
+
+  schema = jsonencode([
+    { name = "inserted_at",     type = "TIMESTAMP", mode = "REQUIRED", description = "Time the Cloud Run service received the message" },
+    { name = "query_name",      type = "STRING",    mode = "REQUIRED", description = "Osquery query name (name field)" },
+    { name = "query_id",        type = "INTEGER",   mode = "NULLABLE", description = "Fleet query ID injected by Fleet when query is known" },
+    { name = "host_identifier", type = "STRING",    mode = "REQUIRED", description = "Osquery hostIdentifier" },
+    { name = "calendar_time",   type = "STRING",    mode = "NULLABLE", description = "Human-readable calendarTime from osquery" },
+    { name = "unix_time",       type = "TIMESTAMP", mode = "NULLABLE", description = "unixTime epoch converted to TIMESTAMP" },
+    { name = "action",          type = "STRING",    mode = "NULLABLE", description = "snapshot, added, or removed" },
+    { name = "epoch",           type = "INTEGER",   mode = "NULLABLE", description = "Schedule epoch marker" },
+    { name = "counter",         type = "INTEGER",   mode = "NULLABLE", description = "Execution counter" },
+    { name = "host_uuid",       type = "STRING",    mode = "NULLABLE", description = "decorations.host_uuid extracted for easy filtering" },
+    { name = "decorations",     type = "STRING",    mode = "NULLABLE", description = "Full decorations map as JSON string" },
+    { name = "row",             type = "STRING",    mode = "REQUIRED", description = "One result row as JSON string (one element from snapshot[], or columns, or one diffResults element)" }
+  ])
+}
+
+resource "google_bigquery_table" "status_logs" {
+  project             = local.bq_project_id
+  dataset_id          = google_bigquery_dataset.fleet_logs.dataset_id
+  table_id            = "status_logs"
+  deletion_protection = false
+
+  time_partitioning {
+    type  = "DAY"
+    field = "inserted_at"
+  }
+
+  clustering = ["severity"]
+
+  schema = jsonencode([
+    { name = "inserted_at", type = "TIMESTAMP", mode = "REQUIRED", description = "Time the Cloud Run service received the message" },
+    { name = "severity",    type = "INTEGER",   mode = "REQUIRED", description = "0=INFO, 1=WARNING, 2=ERROR" },
+    { name = "filename",    type = "STRING",    mode = "NULLABLE", description = "Source file from osquery agent" },
+    { name = "line",        type = "INTEGER",   mode = "NULLABLE", description = "Line number in source file" },
+    { name = "message",     type = "STRING",    mode = "NULLABLE", description = "Log message" },
+    { name = "version",     type = "STRING",    mode = "NULLABLE", description = "Osquery agent version" },
+    { name = "host_uuid",   type = "STRING",    mode = "NULLABLE", description = "decorations.host_uuid" },
+    { name = "decorations", type = "STRING",    mode = "NULLABLE", description = "Full decorations map as JSON string" }
+  ])
+}
+
+resource "google_bigquery_table" "audit_logs" {
+  project             = local.bq_project_id
+  dataset_id          = google_bigquery_dataset.fleet_logs.dataset_id
+  table_id            = "audit_logs"
+  deletion_protection = false
+
+  time_partitioning {
+    type  = "DAY"
+    field = "created_at"
+  }
+
+  clustering = ["type", "actor_email"]
+
+  schema = jsonencode([
+    { name = "inserted_at",     type = "TIMESTAMP", mode = "REQUIRED", description = "Time the Cloud Run service received the message" },
+    { name = "id",              type = "INTEGER",   mode = "NULLABLE", description = "Fleet activity ID" },
+    { name = "uuid",            type = "STRING",    mode = "NULLABLE", description = "Fleet activity UUID" },
+    { name = "created_at",      type = "TIMESTAMP", mode = "NULLABLE", description = "Fleet activity created_at timestamp" },
+    { name = "type",            type = "STRING",    mode = "REQUIRED", description = "Activity type (e.g. created_user, installed_software)" },
+    { name = "actor_id",        type = "INTEGER",   mode = "NULLABLE", description = "Fleet user ID (null for automation)" },
+    { name = "actor_full_name", type = "STRING",    mode = "NULLABLE", description = "Actor full name" },
+    { name = "actor_email",     type = "STRING",    mode = "NULLABLE", description = "Actor email" },
+    { name = "actor_api_only",  type = "BOOLEAN",   mode = "NULLABLE", description = "True if actor is an API-only user" },
+    { name = "fleet_initiated", type = "BOOLEAN",   mode = "NULLABLE", description = "True if triggered by Fleet automation" },
+    { name = "details",         type = "STRING",    mode = "NULLABLE", description = "Full details blob as JSON string — varies by type" }
+  ])
+}
+
+# -------------------------------------
+# Cloud Run — ingest service
+# -------------------------------------
+
+resource "google_cloud_run_v2_service" "ingest" {
+  project             = var.project_id
+  name                = "fleet-pubsub-bq"
+  location            = var.region
+  deletion_protection = false
+  ingress             = "INGRESS_TRAFFIC_ALL"
+
+  template {
+    service_account = google_service_account.ingest_sa.email
+
+    containers {
+      image = var.image
+
+      ports {
+        container_port = 8080
+      }
+
+      env {
+        name  = "BQ_PROJECT_ID"
+        value = local.bq_project_id
+      }
+      env {
+        name  = "BQ_DATASET_ID"
+        value = var.bq_dataset_id
+      }
+      env {
+        name  = "RESULT_SUBSCRIPTION"
+        value = local.result_subscription_name
+      }
+      env {
+        name  = "STATUS_SUBSCRIPTION"
+        value = local.status_subscription_name
+      }
+      env {
+        name  = "AUDIT_SUBSCRIPTION"
+        value = local.audit_subscription_name
+      }
+    }
+  }
+}
+
+# Grant the PubSub invoker SA permission to invoke the Cloud Run service.
+resource "google_cloud_run_v2_service_iam_member" "pubsub_invoker" {
+  project  = var.project_id
+  location = var.region
+  name     = google_cloud_run_v2_service.ingest.name
+  role     = "roles/run.invoker"
+  member   = "serviceAccount:${google_service_account.pubsub_invoker_sa.email}"
+}
+
+# -------------------------------------
+# PubSub push subscriptions
+# -------------------------------------
+
+resource "google_pubsub_subscription" "result" {
+  project = var.project_id
+  name    = local.result_subscription_name
+  topic   = var.result_topic_name
+
+  ack_deadline_seconds = 600
+
+  push_config {
+    push_endpoint = "${google_cloud_run_v2_service.ingest.uri}/ingest"
+
+    oidc_token {
+      service_account_email = google_service_account.pubsub_invoker_sa.email
+    }
+  }
+
+  retry_policy {
+    minimum_backoff = "10s"
+    maximum_backoff = "600s"
+  }
+
+  depends_on = [google_cloud_run_v2_service_iam_member.pubsub_invoker]
+}
+
+resource "google_pubsub_subscription" "status" {
+  project = var.project_id
+  name    = local.status_subscription_name
+  topic   = var.status_topic_name
+
+  ack_deadline_seconds = 600
+
+  push_config {
+    push_endpoint = "${google_cloud_run_v2_service.ingest.uri}/ingest"
+
+    oidc_token {
+      service_account_email = google_service_account.pubsub_invoker_sa.email
+    }
+  }
+
+  retry_policy {
+    minimum_backoff = "10s"
+    maximum_backoff = "600s"
+  }
+
+  depends_on = [google_cloud_run_v2_service_iam_member.pubsub_invoker]
+}
+
+resource "google_pubsub_subscription" "audit" {
+  project = var.project_id
+  name    = local.audit_subscription_name
+  topic   = var.audit_topic_name
+
+  ack_deadline_seconds = 600
+
+  push_config {
+    push_endpoint = "${google_cloud_run_v2_service.ingest.uri}/ingest"
+
+    oidc_token {
+      service_account_email = google_service_account.pubsub_invoker_sa.email
+    }
+  }
+
+  retry_policy {
+    minimum_backoff = "10s"
+    maximum_backoff = "600s"
+  }
+
+  depends_on = [google_cloud_run_v2_service_iam_member.pubsub_invoker]
+}

addons/gcp/pubsub-to-bigquery/outputs.tf

@@ -0,0 +1,9 @@
+output "bq_dataset_id" {
+  description = "BigQuery dataset ID"
+  value       = google_bigquery_dataset.fleet_logs.dataset_id
+}
+
+output "service_url" {
+  description = "URL of the fleet-pubsub-bq Cloud Run service"
+  value       = google_cloud_run_v2_service.ingest.uri
+}