Logs (direct mode)

Other processes append NDJSON lines to files. ndb only reads them. You query the files in place, with no import.

When to use

Your apps already write log or event files. Use direct mode when each app can keep its own file and ndb runs on the same machine.

Collection

Declared in examples/ndb.yml:

yaml

app_logs:
  write: direct
  path: ./data/logs/*.ndjson
  id: id
  time: ts
  indexes: [app, level]

The writer is not ndb

append.sh plays the app. It writes one JSON line per event to data/logs/<app>-<pid>.ndjson. The rules are: one writer per file, one line per write(), a newline at the end. It never talks to ndb.

bash

#!/usr/bin/env bash
# The "app" of the direct-mode example: a plain process that appends one JSON
# line per event to its own file. It never talks to ndb. ndb reads the files.
#
#   append.sh APP PID 'JSON LINE'   append one line to data/logs/APP-PID.ndjson
#   append.sh --seed                write the history lines, only if the files are missing
#
# Rules for direct mode: one writer per file, one line per write(), end with a newline.
set -euo pipefail

dir="${LOG_DIR:-$(cd "$(dirname "$0")/.." && pwd)/data/logs}"

mkdir -p "$dir"

if [[ "${1-}" == "--seed" ]]; then
	if [[ ! -e "$dir/web-1001.ndjson" ]]; then
		cat >"$dir/web-1001.ndjson" <<'LINES'
{"id":"web-1001-1","ts":"2026-11-01T10:00:01Z","app":"web","level":"info","msg":"GET /api/v1/orders","path":"/api/v1/orders","ms":12}
{"id":"web-1001-2","ts":"2026-11-01T10:00:05Z","app":"web","level":"info","msg":"GET /api/v1/users","path":"/api/v1/users","ms":8}
{"id":"web-1001-3","ts":"2026-11-01T10:01:10Z","app":"web","level":"error","msg":"payment gateway timeout","path":"/api/v2/payments","ms":3000}
{"id":"web-1001-4","ts":"2026-11-01T10:02:00Z","app":"web","level":"warn","msg":"slow query","path":"/api/v1/orders","ms":900}
{"id":"web-1001-5","ts":"2026-11-01T10:03:30Z","app":"web","level":"info","msg":"GET /health","path":"/health","ms":1}
{"id":"web-1001-6","ts":"2026-11-01T10:04:00Z","app":"web","level":"error","msg":"payment declined","path":"/api/v2/payments","ms":220}
{"id":"web-1001-7","ts":"2026-11-01T10:05:00Z","app":"web","level":"info","msg":"GET /api/v2/orders","path":"/api/v2/orders","ms":15}
{"id":"web-1001-8","ts":"2026-11-01T10:06:00Z","app":"web","level":"error","msg":"db connection reset","path":"/api/v1/orders","ms":40}
LINES
	fi

	if [[ ! -e "$dir/worker-2002.ndjson" ]]; then
		cat >"$dir/worker-2002.ndjson" <<'LINES'
{"id":"worker-2002-1","ts":"2026-11-01T10:00:30Z","app":"worker","level":"info","msg":"job started","ms":0}
{"id":"worker-2002-2","ts":"2026-11-01T10:02:30Z","app":"worker","level":"error","msg":"job failed: smtp timeout","ms":5000}
{"id":"worker-2002-3","ts":"2026-11-01T10:03:00Z","app":"worker","level":"info","msg":"job started","ms":0}
{"id":"worker-2002-4","ts":"2026-11-01T10:05:30Z","app":"worker","level":"error","msg":"job failed: smtp refused","ms":150}
LINES
	fi

	exit 0
fi

app="$1" pid="$2" line="$3"

printf '%s\n' "$line" >>"$dir/$app-$pid.ndjson"

Filters and sort

level is an indexed equality, ms >= 800 is a range, and the history filter uses the time field.

failures, err := logs.Where(history, ndb.Eq("level", "error")).Sort(ndb.Desc("ts")).All(ctx)
if err != nil {
	return err
}

fmt.Println("errors, newest first:")

for _, l := range failures {
	fmt.Printf("  %s %-6s %s\n", l.TS.Format("15:04:05"), l.App, l.Msg)
}

slow, err := logs.Where(history, ndb.Gte("ms", 800)).Count(ctx)
if err != nil {
	return err
}

fmt.Printf("slow (ms >= 800): %d\n", slow)

failures = ndb.find("app_logs", where: HISTORY.merge(level: "error"), sort: { ts: -1 }).rows

puts "errors, newest first:"

failures.each do |l|
  puts format("  %s %-6s %s", l["ts"][11, 8], l["app"], l["msg"])
end

slow = count(ndb, HISTORY.merge(ms: { "$gte" => 800 }))

puts "slow (ms >= 800): #{slow}"

const failures = await logs.where(history).where({ level: 'error' }).sort({ ts: -1 })

console.log('errors, newest first:')

for (const l of failures) {
  console.log(`  ${l.ts.toISOString().slice(11, 19)} ${l.app.padEnd(6)} ${l.msg}`)
}

const slow = await logs.where(history).where({ ms: { $gte: 800 } }).count()

console.log(`slow (ms >= 800): ${slow}`)

echo "errors, newest first:"

ndb POST /v1/c/app_logs/find "{\"where\":{$history,\"level\":\"error\"},\"sort\":{\"ts\":-1}}" |
	jq -r '"  \(.ts[11:19]) \(.app + "      " | .[0:6]) \(.msg)"'

echo "slow (ms >= 800): $(count "$history,\"ms\":{\"\$gte\":800}")"

Regular expressions

$regex uses Go RE2. A literal prefix such as ^/api can use an index when the field has one.

payments, err := logs.Where(history, ndb.Regex("path", "^/api/v[0-9]+/payments")).Count(ctx)
if err != nil {
	return err
}

fmt.Printf("payment paths ($regex): %d\n", payments)

payments = count(ndb, HISTORY.merge(path: { "$regex" => "^/api/v[0-9]+/payments" }))

puts "payment paths ($regex): #{payments}"

const payments = await logs.where(history).where({ path: { $regex: '^/api/v[0-9]+/payments' } }).count()

console.log(`payment paths ($regex): ${payments}`)

echo "payment paths (\$regex): $(count "$history,\"path\":{\"\$regex\":\"^/api/v[0-9]+/payments\"}")"

Cursor pages

Pass the cursor of one page as after for the next. It works only with the same where and sort.

var pages []string

query := logs.Where(history).Sort(ndb.Asc("ts")).Limit(5)

for {
	rows, err := query.All(ctx)
	if err != nil {
		return err
	}

	pages = append(pages, strconv.Itoa(len(rows)))

	cursor := query.NextCursor()

	if cursor == "" {
		break
	}

	query = logs.Where(history).Sort(ndb.Asc("ts")).Limit(5).After(cursor)
}

fmt.Printf("pages (limit 5): %s\n", strings.Join(pages, ", "))

pages = []
cursor = nil

loop do
  page = ndb.find("app_logs", where: HISTORY, sort: { ts: 1 }, limit: 5, after: cursor)

  pages << page.rows.size
  cursor = page.next_cursor

  break if cursor.nil?
end

puts "pages (limit 5): #{pages.join(", ")}"

const pages: number[] = []
let query = logs.where(history).sort({ ts: 1 }).limit(5)

for (;;) {
  const rows = await query

  pages.push(rows.length)

  if (query.nextCursor === '') {
    break
  }

  query = query.after(query.nextCursor)
}

console.log(`pages (limit 5): ${pages.join(', ')}`)

headers=$(mktemp)
body="{\"where\":{$history},\"sort\":{\"ts\":1},\"limit\":5}"
pages=()

while :; do
	rows=$(ndb POST /v1/c/app_logs/find "$body" -D "$headers" | wc -l | tr -d ' ')
	cursor=$(tr -d '\r' <"$headers" | awk -F': ' 'tolower($1) == "x-ndb-next" { print $2 }')

	pages+=("$rows")

	if [[ -z "$cursor" ]]; then
		break
	fi

	body="{\"where\":{$history},\"sort\":{\"ts\":1},\"limit\":5,\"after\":\"$cursor\"}"
done

rm -f "$headers"

echo "pages (limit 5): $(IFS=,; echo "${pages[*]}" | sed 's/,/, /g')"

A new line appears

The program starts append.sh, then waits until the new id is found. ndb sees file events at once, with a 5 s poll as a fallback.

id := fmt.Sprintf("go-%d", time.Now().UnixMilli())
line := fmt.Sprintf(`{"id":%q,"ts":%q,"app":"cli","level":"info","msg":"hello from go"}`, id, time.Now().UTC().Format(time.RFC3339))

if err := exec.Command("../append.sh", "cli", "3003", line).Run(); err != nil {
	return err
}

fmt.Println("append: cli wrote one line")

if err := waitFor(func() (bool, error) {
	n, err := logs.Where(ndb.Eq("id", id)).Count(ctx)

	return n == 1, err
}); err != nil {
	return err
}

fmt.Println("the new line is in the collection")

id = "ruby-#{(Time.now.to_f * 1000).to_i}"
line = { id: id, ts: Time.now.utc.strftime("%Y-%m-%dT%H:%M:%SZ"), app: "cli", level: "info", msg: "hello from ruby" }

system(File.expand_path("../append.sh", __dir__), "cli", "3003", JSON.generate(line), exception: true)

puts "append: cli wrote one line"

wait_for { count(ndb, { id: id }) == 1 }

puts "the new line is in the collection"

const id = `ts-${Date.now()}`
const line = JSON.stringify({ id, ts: new Date().toISOString().slice(0, 19) + 'Z', app: 'cli', level: 'info', msg: 'hello from ts' })

execFileSync(new URL('../append.sh', import.meta.url).pathname, ['cli', '3003', line])

console.log('append: cli wrote one line')

await waitFor(async () => (await logs.where({ id }).count()) === 1)

console.log('the new line is in the collection')

id="curl-$(date +%s)$RANDOM"
line="{\"id\":\"$id\",\"ts\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"app\":\"cli\",\"level\":\"info\",\"msg\":\"hello from curl\"}"

"$here/append.sh" cli 3003 "$line"

echo "append: cli wrote one line"

wait_for "\"id\":\"$id\"" 1

echo "the new line is in the collection"

Output

The four programs print the same text.

text

history: 12 lines from 2 files
errors, newest first:
  10:06:00 web    db connection reset
  10:05:30 worker job failed: smtp refused
  10:04:00 web    payment declined
  10:02:30 worker job failed: smtp timeout
  10:01:10 web    payment gateway timeout
slow (ms >= 800): 3
payment paths ($regex): 2
pages (limit 5): 5, 5, 2
append: cli wrote one line
the new line is in the collection

Notes

A direct collection is read only. insert on it is direct_collection, so these programs have no write step.
The Go struct and the TypeScript schema declare id and time as in ndb.yml. A different value is catalog_conflict. The Ruby client does not declare the collection at all.
Every SDK supports every step of this example.
More: Storage, Clients.

Run it

cd examples
make run-logs LANG=go   # go, ruby, ts or curl
make run-logs           # every language

Source: examples/logs. The programs read NDB_URL and NDB_TOKEN, and they can run again and again.

Logs (direct mode) ​

When to use ​

Collection ​

The writer is not ndb ​

Filters and sort ​

Regular expressions ​

Cursor pages ​

A new line appears ​

Output ​

Notes ​

Run it ​