Install
Godot
Add AnyAnalytics to a Godot 4 game with a GDScript autoload.
One GDScript file, added as an autoload, tracks any Godot 4 game. It keeps events on disk while the player is offline and sends them in batches in the background.
Install
Add the script
Create
any_analytics.gdwith the code below. Your project's setup page shows it with your write key filled in.Register the autoload
Add it as an Autoload named AnyAnalytics: Project → Project Settings → Globals → Autoload.
Run the game
A
$session_startevent is sent on startup. It shows up in Live events within about 10 seconds.
# any_analytics.gd: queues events on disk and posts them to AnyAnalytics (Godot 4).
# Every event gets a uuid when tracked, so a retried batch is never counted twice.
extends Node
const HOST := "https://api.anyanalytics.org"
const WRITE_KEY := "vxw_YOUR_WRITE_KEY"
const FLUSH_INTERVAL := 10.0
const MAX_QUEUE := 1000 # oldest events are dropped beyond this
const MAX_BATCH := 100
const ID_PATH := "user://anyanalytics_id"
const QUEUE_PATH := "user://anyanalytics_queue.json"
var distinct_id := ""
var session_id := ""
var _queue: Array = []
var _in_flight: Array = [] # the one batch being sent
var _batch_size := MAX_BATCH
var _backoff := 0.0
var _retry_at := 0.0 # seconds since startup
var _warned_full := false
var _crypto := Crypto.new()
var _http := HTTPRequest.new()
func _ready() -> void:
add_child(_http)
_http.request_completed.connect(_on_request_completed)
distinct_id = _load_id()
session_id = _uuid()
_queue = _load_queue()
var timer := Timer.new()
timer.wait_time = FLUSH_INTERVAL
timer.autostart = true
timer.timeout.connect(_on_timer)
add_child(timer)
track("$session_start")
func _on_timer() -> void:
_save_queue()
flush()
func track(event: String, properties: Dictionary = {}) -> void:
_queue.append({
"uuid": _uuid(), # kept across retries for deduplication
"event": event,
"distinct_id": distinct_id,
"session_id": session_id,
"timestamp": _now_iso(),
"properties": properties,
"context": {
"platform": OS.get_name(),
"app_version": str(ProjectSettings.get_setting("application/config/version", "1.0.0")),
"os": OS.get_name(),
"os_version": OS.get_version(),
"device": OS.get_model_name(),
"locale": OS.get_locale(),
},
})
if _queue.size() > MAX_QUEUE:
_queue = _queue.slice(_queue.size() - MAX_QUEUE)
if not _warned_full:
_warned_full = true
push_warning("[AnyAnalytics] Queue full, dropping the oldest events.")
if _queue.size() >= 20:
flush()
func identify(user_id: String, traits: Dictionary = {}) -> void:
var anonymous_id := distinct_id
distinct_id = user_id
_save_id(user_id)
track("$identify", {"$anon_distinct_id": anonymous_id, "$set": traits})
# One request at a time; events leave the queue only once the server accepts them.
func flush() -> void:
if not _in_flight.is_empty() or _queue.is_empty() or _now() < _retry_at:
return
_in_flight = _queue.slice(0, _batch_size)
_queue = _queue.slice(_batch_size)
var headers := PackedStringArray(["Content-Type: application/json", "Authorization: Bearer " + WRITE_KEY])
var body := JSON.stringify({"sent_at": _now_iso(), "batch": _in_flight})
if _http.request(HOST + "/v1/batch", headers, HTTPClient.METHOD_POST, body) != OK:
_queue = _in_flight + _queue
_in_flight = []
_retry_later("")
func _on_request_completed(result: int, code: int, headers: PackedStringArray, body: PackedByteArray) -> void:
var batch := _in_flight
_in_flight = []
var ok := result == HTTPRequest.RESULT_SUCCESS
if ok and code >= 200 and code < 300:
_backoff = 0.0
_batch_size = MAX_BATCH
elif ok and code == 413 and batch.size() > 1:
# Body too large: retry the same events in smaller batches
_batch_size = maxi(1, int(batch.size() / 2.0))
_queue = batch + _queue
elif not ok or code == 408 or code == 429 or code >= 500:
# Offline, timeout, rate limited or server busy: keep the events, back off
_queue = batch + _queue
_retry_later(_header(headers, "retry-after"))
else:
# Other 4xx (bad key, invalid payload) or one oversized event: never succeeds
push_warning("[AnyAnalytics] Dropped %d event(s): HTTP %d %s" % [batch.size(), code, body.get_string_from_utf8()])
_save_queue()
if ok and (code < 300 or code == 413):
flush.call_deferred() # keep draining
# Exponential backoff with jitter (1s doubling to 60s); Retry-After wins when sent.
func _retry_later(retry_after: String) -> void:
_backoff = 1.0 if _backoff == 0.0 else minf(_backoff * 2.0, 60.0)
var delay: float = retry_after.to_float() if retry_after.is_valid_float() else _backoff * randf_range(0.5, 1.0)
_retry_at = _now() + delay
func _header(headers: PackedStringArray, header_name: String) -> String:
for line in headers:
if line.to_lower().begins_with(header_name + ":"):
return line.substr(header_name.length() + 1).strip_edges()
return ""
func _notification(what: int) -> void:
if what == NOTIFICATION_APPLICATION_PAUSED or what == NOTIFICATION_WM_CLOSE_REQUEST:
_save_queue() # anything unsent goes out on the next launch, same uuids
flush()
func _save_queue() -> void:
var file := FileAccess.open(QUEUE_PATH, FileAccess.WRITE)
if file:
file.store_string(JSON.stringify(_in_flight + _queue))
func _load_queue() -> Array:
if not FileAccess.file_exists(QUEUE_PATH):
return []
var saved = JSON.parse_string(FileAccess.get_file_as_string(QUEUE_PATH))
return saved if saved is Array else []
func _load_id() -> String:
if FileAccess.file_exists(ID_PATH):
return FileAccess.get_file_as_string(ID_PATH)
var id := _uuid()
_save_id(id)
return id
func _save_id(id: String) -> void:
var file := FileAccess.open(ID_PATH, FileAccess.WRITE)
if file:
file.store_string(id)
func _now() -> float:
return Time.get_ticks_msec() / 1000.0
func _now_iso() -> String:
return Time.get_datetime_string_from_system(true) + "Z"
# Random (v4) UUID
func _uuid() -> String:
var b := _crypto.generate_random_bytes(16)
b[6] = (b[6] & 0x0f) | 0x40 # version 4
b[8] = (b[8] & 0x3f) | 0x80 # RFC 4122 variant
var h := b.hex_encode()
return "%s-%s-%s-%s-%s" % [h.substr(0, 8), h.substr(8, 4), h.substr(12, 4), h.substr(16, 4), h.substr(20, 12)]- Undelivered events are saved under user:// and retried with backoff (offline, 429, 5xx). Each event keeps its uuid, so retries are never counted twice.
- Exporting to the web? Godot's HTTPRequest works there too, the ingestion API allows CORS.
Note
The script is indented with tabs, as Godot expects. If your editor converts them to spaces when you paste, switch the file back to tabs.
Track events
The autoload is available everywhere as AnyAnalytics:
AnyAnalytics.track("level_completed", {"level": 5, "score": 1200, "duration": 82})Properties are a plain Dictionary of strings, numbers, booleans, arrays and nested dictionaries. Events are sent every 10 seconds, as soon as 20 are queued, and when the game is paused or closed. Anything that couldn't be sent goes out on the next launch.
Identify players
Each install gets a random player id, saved under user://. When the player signs in, switch to your own id:
# After the player signs in
AnyAnalytics.identify(player_id, {"name": player_name, "platform": "steam"})Their earlier anonymous events are linked to the same player, and the traits are saved on their profile. The new id is remembered across launches.
Levels and the Game report
The Game report builds level funnels from events named level_start, level_complete and level_fail, each with a level property:
# When the player starts a level
AnyAnalytics.track("level_start", {"level": 3})
# When they finish it, or fail it
AnyAnalytics.track("level_complete", {"level": 3, "score": 1200})
AnyAnalytics.track("level_fail", {"level": 3})See Game events for every event and property the report reads, including errors and crashes.
Sessions and context
Each launch of the game is one session. Every event carries the platform, your project's version (Project Settings → Application → Config → Version), the operating system and its version, the device model and the locale.