snowplow

Overview ¶

Package snowplow provides a lightweight Snowplow event tracking implementation for Go applications.

The package enables asynchronous event collection and batch sending to Snowplow collectors with automatic retry logic and FIFO queue management to prevent event stagnation.

Basic Usage ¶

Create an emitter to handle event sending:

emitter, err := snowplow.NewEmitter("https://collector.example.com/com.snowplowanalytics.snowplow/tp2")
if err != nil {
	log.Fatal(err)
}

Create a tracker to generate events:

tracker, err := snowplow.NewTracker("my-application", emitter)
if err != nil {
	log.Fatal(err)
}
defer tracker.Stop()

Track custom events with or without properties:

// Simple event without properties
tracker.TrackEvent("user_login", nil)

// Event with properties
tracker.TrackEvent("button_clicked", map[string]any{
	"button_id": "submit",
	"page":      "/checkout",
})

// Event with nested properties
tracker.TrackEvent("page_view", map[string]any{
	"title": "Product Page",
	"user": map[string]any{
		"id":   12345,
		"role": "customer",
	},
})

Custom Contexts ¶

To attach typed custom contexts to a custom event, use TrackEventWithContexts. Custom contexts populate the Snowplow `cx` field and let analysts query their fields as first-class warehouse columns instead of parsing them out of the freeform props bag.

The package ships with GitLabStandardContext, a typed payload for the gitlab_standard iglu schema with typed enums (Realm, DeploymentType) and a sealed UserID variant. Service-owned contexts can be defined by implementing the EventContext interface:

// Universal identity / environment fields shared across GitLab services.
standard := snowplow.GitLabStandardContext{
	Environment:    "production",
	OrganizationID: 12345,
	Realm:          snowplow.RealmSaaS,
	InstanceID:     "gitlab-rails-1",
}

// Service-specific context (defined by the caller). Any type satisfying
// the EventContext interface — Schema() string and Data() any — can be
// passed; for example:
//
//   type ArtifactRegistryEventContext struct {
//       Format string `json:"format"`
//       Kind   string `json:"kind"`
//   }
//
//   func (c ArtifactRegistryEventContext) Schema() string {
//       return "iglu:com.gitlab/artifact_registry_event/jsonschema/1-0-0"
//   }
//   func (c ArtifactRegistryEventContext) Data() any { return c }
registryCtx := ArtifactRegistryEventContext{
	Format: "container",
	Kind:   "local",
}

tracker.TrackEventWithContexts(
	"artifact_registry_artifact_pushed",
	nil,
	standard,
	registryCtx,
)

Contexts may optionally implement `Validate() error`. If present, the tracker calls it before queuing the event; a validation failure aborts the event with the returned error. GitLabStandardContext.Validate covers the schema constraints: required Environment, Realm/DeploymentType enums, UserID variant, string maxLengths, and integer ranges (0..2^31-1).

TrackEvent is equivalent to TrackEventWithContexts with zero contexts and remains supported for backward compatibility.

Billing Events ¶

Track billing events for GitLab's billable usage tracking. Billing events use the billable_usage schema (iglu:com.gitlab/billable_usage/jsonschema/1-0-2) and are sent as Snowplow structured events with billing context attached.

// Minimal billing event
tracker.TrackBillingEvent(
	BillingEventRequiredInput{
		Category:      "CodeSuggestionsService",
		EventType:     "code_completions",
		Realm:         "SaaS",
		UnitOfMeasure: "tokens",
		Quantity:      150.0,
	}
)

// Billing event with full context
tracker.TrackBillingEventWithOptionalInput(
	BillingEventRequiredInput{
		Category:      "CodeSuggestionsService",
		EventType:     "code_completions",
		Realm:         "SaaS",
		UnitOfMeasure: "tokens",
		Quantity:      150.0,
	},
	BillingEventOptionalInput{
		InstanceID:      "instance-abc123",
		ProjectID:       12345,
		NamespaceID:     67890,
		RootNamespaceID: 22222,
		OrganizationID:  11111,
		Subject:         "user-456",
		GlobalUserID:    "global-user-789",
		CorrelationID:   "request-correlation-xyz",
		DeploymentType:  "self-managed",
		EntityID:        "entity-abc",
		Metadata: map[string]any{
			"model": "claude-3",
		},
	},
)

Emitter Options ¶

The emitter can be configured using functional options:

emitter, err := snowplow.NewEmitter(
	"https://collector.example.com/com.snowplowanalytics.snowplow/tp2",
	snowplow.WithCallback(func(successes, failures []snowplow.SendResult) {
		for _, s := range successes {
			log.Printf("Sent %d events (status: %d)", s.EventCount, s.StatusCode)
		}
		for _, f := range failures {
			if f.Err != nil {
				log.Printf("Failed to send %d events: %v", f.EventCount, f.Err)
			} else {
				log.Printf("Failed to send %d events (status: %d)", f.EventCount, f.StatusCode)
			}
		}
	}),
)

Available options:

WithCallback: Sets a callback function that is invoked after each send loop iteration with aggregated success and failure results. The callback receives two slices of SendResult: successes (HTTP 2xx/3xx responses) and failures (HTTP 4xx/5xx responses or local errors). Each SendResult contains EventCount (number of events in the batch), StatusCode (HTTP status or -1 for local errors), and Err (non-nil for local errors like network failures or marshal errors).

Metrics and Observability ¶

The emitter automatically collects Prometheus metrics tracking event flow and performance. Metrics are accessible via Emitter.Collector(), which returns a prometheus.Collector that can be registered with any Prometheus registry.

Available metrics include:

• Event enqueue counts
• Send success/failure counts
• Batch delivery duration (histogram)
• Queue depth and capacity
• Overflow drop counts

Example:

collector := emitter.Collector()
// Register with Prometheus
prometheus.MustRegister(collector)

For detailed metric descriptions, see the metrics package documentation.

Architecture ¶

The implementation consists of three main components:

Tracker: High-level API for tracking events. Generates Snowplow protocol-compliant payloads with unique IDs and timestamps, and encodes custom event data as base64-encoded self-describing JSON.

Emitter: Handles asynchronous batch sending of events to Snowplow collectors. Events are buffered in memory and sent in batches of up to 100 events. Failed sends are automatically retried by re-queueing events. Uses context cancellation for graceful shutdown.

Storage: Thread-safe in-memory event storage using a FIFO queue. Events are retrieved in order (oldest first) to prevent event stagnation. Implements automatic capacity management with a maximum size of 10,000 events.

Event Processing ¶

Events flow through the system as follows: Tracker.TrackEvent creates a Snowplow protocol payload, Emitter.AddAsync stores the event in the FIFO queue, Emitter automatically batches and sends events (up to 100 per batch), on failure events are re-queued to the back of the queue, and FIFO ordering ensures old events eventually get processed.

Lifecycle Management ¶

The emitter provides lifecycle management methods:

Flush: Triggers immediate sending of buffered events (non-blocking).

Stop: Cancels in-flight requests and waits for completion (blocking). After Stop is called, TrackEvent will return an error (ErrEmitterStopped).

Always call Stop before application shutdown:

tracker.Stop()

Event Structure ¶

Custom events (TrackEvent) are sent as self-describing events with standard Snowplow fields: aid (application ID), dtm (device timestamp), e (event type), eid (event ID), p (platform), stm (sent timestamp), tna (namespace), tv (tracker version), ua (user agent), and ue_px (base64-encoded self-describing JSON payload).

Billing events (TrackBillingEvent, TrackBillingEventWithOptionalInput) are sent as structured events with fields: se_ca (category), se_ac (action/event type), se_la (label/event ID), se_va (value/quantity), and cx (base64-encoded billing context). The billing context follows the billable_usage schema with required fields: event_id, event_type, realm, unit_of_measure, quantity, and timestamp.

Implementation Details ¶

Events are sent asynchronously in batches (100 events per batch) to minimize overhead. Failed sends automatically retry by re-queueing events to the back of the queue. FIFO queue ordering prevents event stagnation (oldest events are sent first). Storage capacity is limited to 10,000 events; when full, oldest events are dropped. Storage is in-memory only (events are lost on process restart). HTTP client uses connection pooling with a 5 second timeout.

Thread Safety ¶

All public APIs are thread-safe and can be called concurrently from multiple goroutines. The storage implementation uses mutex synchronization to protect internal state.

package main

import (
	"log"

	"gitlab.com/gitlab-org/labkit/v2/events/snowplow"
)

func main() {
	// Initialize emitter with collector URL.
	// The emitter handles automatic batching (5 events per batch) and retry logic.
	emitter, err := snowplow.NewEmitter("http://localhost:9093/com.snowplowanalytics.snowplow/tp2")
	if err != nil {
		log.Fatalf("Failed to create emitter: %v", err)
	}

	// Create tracker with application ID.
	tracker, err := snowplow.NewTracker("my-application", emitter)
	if err != nil {
		log.Fatalf("Failed to create tracker: %v", err)
	}

	// Track a simple event
	err = tracker.TrackEvent("user_login", nil)
	if err != nil {
		log.Printf("Failed to track event: %v", err)
	}

	// Stop tracker before shutdown
	tracker.Stop()
}

Output: