Decode Tesla Fleet Telemetry protobuf for the NATS add-on for Splunk

.github/workflows/ci.yml

@@ -0,0 +1,32 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
+        with:
+          go-version-file: go.mod
+      - name: gofmt
+        run: test -z "$(gofmt -l .)" || { gofmt -l .; exit 1; }
+      - name: go vet
+        run: go vet ./...
+      - name: go test
+        run: go test -race ./...
+      - name: Benchmarks run
+        run: go test -run '^$' -bench . -benchtime 100x ./...
+      - name: Generated code matches the pinned protos
+        run: |
+          scripts/generate.sh
+          git diff --exit-code -- internal/protos
+      - name: Build
+        run: scripts/build.sh

.github/workflows/release.yml

@@ -0,0 +1,25 @@
+name: Release
+
+on:
+  push:
+    tags: ["v*"]
+
+permissions:
+  contents: write
+
+jobs:
+  release:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
+        with:
+          go-version-file: go.mod
+      - name: go test
+        run: go test ./...
+      - name: Build
+        run: scripts/build.sh "$GITHUB_REF_NAME"
+      - name: Publish the release
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: gh release create "$GITHUB_REF_NAME" dist/* --title "$GITHUB_REF_NAME" --generate-notes --verify-tag

.gitignore

@@ -0,0 +1 @@
+/dist/

README.md

@@ -1,2 +1,170 @@
 # telemetry-decoder
-Decodes Tesla Fleet Telemetry protobuf messages to JSON (decoder for the Splunk NATS add-on)
+
+Decodes Tesla Fleet Telemetry protobuf messages to JSON: a payload decoder for the
+[NATS Add-on for Splunk](https://github.com/Bre77/splunk_nats).
+
+Tesla Fleet Telemetry publishes protobuf on NATS subjects
+`telemetry.<VIN>.data`, `.alerts`, `.errors` and `.connectivity` (and `.metrics`).
+Without a decoder the add-on indexes those payloads as base64. With this decoder
+each message becomes a one-line JSON event whose fields Splunk extracts directly:
+
+```
+index=nats source="*.data" VehicleSpeed=* | timechart avg(VehicleSpeed) by vin
+```
+
+It is a single static binary with no runtime dependencies. The messages are
+generated from the protos in
+[Teslemetry/tesla-protocol](https://github.com/Teslemetry/tesla-protocol/tree/main/proto/telemetry)
+(`proto/telemetry`), which track Tesla's
+[fleet-telemetry](https://github.com/teslamotors/fleet-telemetry) protos.
+
+## Install
+
+1. Download the binary for the Splunk host from the
+   [latest release](https://github.com/Teslemetry/telemetry-decoder/releases)
+   (`telemetry-decoder-linux-amd64`, `-linux-arm64`, `-darwin-amd64` or
+   `-darwin-arm64`) and check it against `SHA256SUMS`, or [build it](#build).
+2. Place it in the add-on's decoder directory, owned by the Splunk user and
+   executable. Its file name there is the decoder's name:
+
+   ```
+   sudo install -o splunk -g splunk -m 755 telemetry-decoder-linux-amd64 \
+     $SPLUNK_HOME/etc/apps/nats/local/decoders/telemetry-decoder
+   ```
+
+3. On a **NATS Subscribe** input for `telemetry.>`, open **Payload decoder** and
+   set **Decoder** to `telemetry-decoder`. Leave **Decoded Source Type** at
+   `nats:json`, which allows the large events alerts produce.
+4. To try it from search first:
+
+   ```
+   | natssubscribe subject="telemetry.>" account=<account> decoder=telemetry-decoder | spath
+   ```
+
+## Arguments
+
+Set these as the input's **Decoder Arguments** (or `decoder_args=` on
+`natssubscribe`):
+
+| Argument | Default | Effect |
+|---|---|---|
+| `-format flat` | yes | Vehicle data has one top-level field per signal (see below). |
+| `-format protojson` | | Vehicle data is the protobuf JSON mapping, the same JSON fleet-telemetry writes with `transmit_decoded_records`, for every record type: fields in proto order, and vehicle data as `"data":[{"key":"VehicleSpeed","value":{"floatValue":28.6}}]`. |
+| `-proto-names` | | Proto field names (`created_at`, `network_interface`, `is_resend`) instead of the JSON names (`createdAt`, `networkInterface`, `isResend`). |
+| `-version` | | Prints the version and the tesla-protocol commit, then exits. |
+
+## Output
+
+Every event is one line of JSON. Field names follow the protobuf JSON mapping,
+as fleet-telemetry's own JSON output does (lowerCamel, enums by name, timestamps
+in RFC 3339 UTC), and every field is present even when unset (`null`, `false`,
+`0`, `""`), so an alert without `endedAt` is visibly still active. Every event
+starts with `vin` and `createdAt`, so Splunk extracts them even from large alert
+events. The VIN comes from the message, or from the subject when the message has
+none.
+
+**Vehicle data** (`telemetry.<VIN>.data`), with the default `-format flat`:
+
+```json
+{"vin":"TESTVIN0000000001","createdAt":"2026-10-15T00:13:16.273960575Z","isResend":false,"MilesToArrival":0.962417613193002,"Location":{"latitude":37.4919,"longitude":-121.9447},"Gear":"ShiftStateD","DoorState":{"DriverFront":true,"DriverRear":false,"PassengerFront":false,"PassengerRear":false,"TrunkFront":false,"TrunkRear":false},"invalid":["TpmsPressureFr"]}
+```
+
+- Each signal is a field named as in Tesla's `Field` enum, holding its value as
+  plain JSON: numbers, strings and booleans as themselves, enums by name, and
+  structured values as objects with every field (`Location.latitude`,
+  `DoorState.DriverFront`, `ScheduledChargingStartTime.hour`).
+- Signals the vehicle reports as invalid (unavailable) are listed in `invalid`
+  rather than given a value.
+- A signal newer than the decoder's protos is named by its number (`"312": 4.5`),
+  and a value type newer than the protos is `null`, so nothing is silently lost.
+  Updating the protos (see [Development](#development)) names them.
+
+**Alerts** (`telemetry.<VIN>.alerts`):
+
+```json
+{"vin":"TESTVIN0000000001","createdAt":"2026-10-15T00:15:05.180915260Z","alerts":[{"name":"CP_a101_wcFoldbackActive","audiences":["Customer","Service"],"startedAt":"2026-10-15T00:14:55.164Z","endedAt":null}]}
+```
+
+**Errors** (`telemetry.<VIN>.errors`):
+
+```json
+{"vin":"TESTVIN0000000001","createdAt":"2026-10-15T00:10:12.617991348Z","errors":[{"createdAt":"2026-10-14T23:45:04.404614863Z","name":"unsupported_field","tags":{"field_name":"LifetimeEnergyGainedRegen"},"body":""}]}
+```
+
+**Connectivity** (`telemetry.<VIN>.connectivity`):
+
+```json
+{"vin":"TESTVIN0000000001","createdAt":"2026-10-15T00:13:22.426862796Z","connectionId":"00000000-0000-4000-8000-000000000001","status":"CONNECTED","networkInterface":"wifi"}
+```
+
+**Metrics** (`telemetry.<VIN>.metrics`) follow the same mapping:
+`{"vin":...,"createdAt":...,"metrics":[{"name":...,"tags":{...},"value":...}]}`.
+
+Example searches (with an input indexing into `index=nats`):
+
+```
+index=nats source="*.data" | stats latest(BatteryLevel) latest(Odometer) by vin
+index=nats source="*.data" Location.latitude=* | table _time vin Location.latitude Location.longitude
+index=nats source="*.connectivity" | stats count by status networkInterface
+index=nats source="*.alerts" | stats count by alerts{}.name
+index=nats source="*.data" invalid{}=* | stats count by invalid{}
+```
+
+### When a payload cannot be decoded
+
+The decoder answers with an empty line, so the add-on keeps the payload as
+base64 and logs the reason, which the decoder writes to stderr as one line (for
+example `telemetry.<VIN>.foo: unknown record type "foo"`). That happens for a
+subject whose last token is not a known record type, an empty payload, a payload
+that is not valid protobuf for its record type, or a request that is not valid
+JSON or base64. The decoder keeps running and answers the next message.
+
+### Protocol and throughput
+
+The decoder implements the add-on's
+[decoder protocol](https://github.com/Bre77/splunk_nats#decoder-protocol): one
+JSON request per line on stdin, one event per line on stdout (flushed after each
+line), diagnostics on stderr, and it exits when stdin closes. It answers in a
+few microseconds: on live `telemetry.>` traffic it decodes about 200,000 messages
+a second on one core, far beyond the stream's rate (about 110 a second).
+
+## Build
+
+Needs Go (the version in `go.mod`).
+
+```
+scripts/build.sh            # static binaries for linux/darwin amd64/arm64 in dist/, with SHA256SUMS
+go build -o telemetry-decoder .   # just for this machine
+```
+
+Builds are reproducible: the same commit and version give byte-identical
+binaries.
+
+## Development
+
+```
+go vet ./... && go test ./...                         # unit, golden, protocol and binary tests
+go test -run '^$' -bench . ./internal/decoder         # throughput (BenchmarkRun: ns per message)
+```
+
+- `internal/protos` is generated, and committed. `scripts/generate.sh`
+  regenerates it from tesla-protocol at the pinned commit with pinned `protoc`
+  and `protoc-gen-go` (the latter via the `tool` directive in `go.mod`). To move
+  to a newer tesla-protocol, run `scripts/generate.sh <commit>`, update the
+  pinned commit in the script, refresh the golden files and commit. CI checks that
+  the committed code matches the pinned protos.
+- `internal/decoder/testdata/golden` holds real payloads captured from live
+  telemetry and anonymized by `tools/mkfixtures` (fake VIN, connection IDs,
+  locations, names and media), one `<name>.request` per fixture, and the
+  expected event for each output option. To refresh them from a capture of
+  decoder requests (one JSON line per message), run
+  `go run ./tools/mkfixtures < capture.jsonl` and then
+  `go test ./internal/decoder -run TestGolden -update`, and review the diff.
+- `scripts/e2e-splunk.sh` runs the decoder in a disposable Splunk container with
+  the add-on against live NATS traffic and checks that every message is decoded.
+
+## Release
+
+Pushing a `v*` tag runs the release workflow: it tests, builds the binaries with
+the tag as their version, and publishes a GitHub release with the binaries and
+`SHA256SUMS`.

go.mod

@@ -0,0 +1,9 @@
+module github.com/Teslemetry/telemetry-decoder
+
+go 1.26.0
+
+toolchain go1.27.2
+
+tool google.golang.org/protobuf/cmd/protoc-gen-go
+
+require google.golang.org/protobuf v1.36.12

go.sum

@@ -0,0 +1,4 @@
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
+google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=

internal/decoder/bench_test.go

@@ -0,0 +1,105 @@
+package decoder
+
+import (
+	"encoding/json"
+	"io"
+	"sort"
+	"testing"
+)
+
+// BenchmarkDecode decodes each fixture payload.
+func BenchmarkDecode(b *testing.B) {
+	requests := goldenRequests(b)
+	names := make([]string, 0, len(requests))
+	for name := range requests {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	for _, format := range []Format{FormatFlat, FormatProtoJSON} {
+		for _, name := range names {
+			var r Request
+			if err := json.Unmarshal(requests[name], &r); err != nil {
+				b.Fatal(err)
+			}
+			payload, err := r.DecodePayload()
+			if err != nil {
+				b.Fatal(err)
+			}
+			b.Run(string(format)+"/"+name, func(b *testing.B) {
+				d := mustNew(b, Options{Format: format})
+				var event []byte
+				b.SetBytes(int64(len(payload)))
+				for b.Loop() {
+					var err error
+					if event, err = d.Decode(event[:0], r.Subject, payload); err != nil {
+						b.Fatal(err)
+					}
+				}
+			})
+		}
+	}
+}
+
+// BenchmarkRun serves the decoder protocol over a stream cycling through every
+// fixture, as the add-on sends it: one request line in, one event line out,
+// flushed. One op is one message, so messages per second is 1e9 / ns/op.
+func BenchmarkRun(b *testing.B) {
+	var lines [][]byte
+	for _, request := range goldenRequests(b) {
+		lines = append(lines, request)
+	}
+	d := mustNew(b, Options{})
+	in := &cyclingReader{lines: lines, remaining: b.N}
+	var out countingWriter
+	b.ResetTimer()
+	if err := d.Run(in, &out, io.Discard); err != nil {
+		b.Fatal(err)
+	}
+	b.StopTimer()
+	if out.lines != b.N || out.empty != 0 {
+		b.Fatalf("%d events (%d empty) for %d requests", out.lines, out.empty, b.N)
+	}
+}
+
+// cyclingReader reads remaining request lines, cycling through lines.
+type cyclingReader struct {
+	lines     [][]byte
+	remaining int
+	next      int
+	pending   []byte
+}
+
+func (r *cyclingReader) Read(p []byte) (int, error) {
+	if len(r.pending) == 0 {
+		if r.remaining == 0 {
+			return 0, io.EOF
+		}
+		r.pending = r.lines[r.next%len(r.lines)]
+		r.next++
+		r.remaining--
+	}
+	n := copy(p, r.pending)
+	r.pending = r.pending[n:]
+	return n, nil
+}
+
+// countingWriter counts the event lines written, and the empty ones.
+type countingWriter struct {
+	lines, empty int
+	inLine       bool
+}
+
+func (w *countingWriter) Write(p []byte) (int, error) {
+	for _, c := range p {
+		if c == '\n' {
+			w.lines++
+			if !w.inLine {
+				w.empty++
+			}
+			w.inLine = false
+		} else {
+			w.inLine = true
+		}
+	}
+	return len(p), nil
+}