Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions docs/cli.md
Original file line number Diff line number Diff line change
Expand Up @@ -273,7 +273,7 @@ are stored under `.lousho/` in the current directory. See
```bash
npx lousho observe start # collector on http://127.0.0.1:7777
npx lousho observe status # health + fleet counts
npx lousho observe stop # pidfile + SIGTERM
npx lousho observe stop # <db>.pid + SIGTERM (checks the daemon serves --db)
npx lousho observe rebuild-rollups # recompute insight aggregates
npx lousho observe token create --project myapp # bearer key for non-loopback auth
```
Expand All @@ -283,7 +283,13 @@ set: traces, fleet monitoring, eval scores, the prompt registry and the
context hub in one SQLite file (default `~/.lousho/observe.db`). The Studio
Observe tabs read it. `start` takes `--port` (or `LOUSHO_OBSERVE_PORT`),
`--db`, `--host`, `--require-token` and a repeatable `--webhook` for alert
hooks. See [The observe daemon](./observability.md#the-observe-daemon).
hooks. Each daemon writes its own `<db name>.pid` beside the db
(`observe.db` gives `observe.pid`), so two dbs in one folder never share a
pidfile. `start` exits 1 when the port already serves a different db, and
`stop` refuses to signal when the daemon at the pidfile's port reports another
db, and falls back to signalling when it cannot verify (unreachable, or an old
one-line pidfile). See
[The observe daemon](./observability.md#the-observe-daemon).

## `lousho prompt`

Expand Down
10 changes: 8 additions & 2 deletions llms-full.txt
Original file line number Diff line number Diff line change
Expand Up @@ -5150,7 +5150,7 @@ are stored under `.lousho/` in the current directory. See
```bash
npx lousho observe start # collector on http://127.0.0.1:7777
npx lousho observe status # health + fleet counts
npx lousho observe stop # pidfile + SIGTERM
npx lousho observe stop # <db>.pid + SIGTERM (checks the daemon serves --db)
npx lousho observe rebuild-rollups # recompute insight aggregates
npx lousho observe token create --project myapp # bearer key for non-loopback auth
```
Expand All @@ -5160,7 +5160,13 @@ set: traces, fleet monitoring, eval scores, the prompt registry and the
context hub in one SQLite file (default `~/.lousho/observe.db`). The Studio
Observe tabs read it. `start` takes `--port` (or `LOUSHO_OBSERVE_PORT`),
`--db`, `--host`, `--require-token` and a repeatable `--webhook` for alert
hooks. See [The observe daemon](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md#the-observe-daemon).
hooks. Each daemon writes its own `<db name>.pid` beside the db
(`observe.db` gives `observe.pid`), so two dbs in one folder never share a
pidfile. `start` exits 1 when the port already serves a different db, and
`stop` refuses to signal when the daemon at the pidfile's port reports another
db, and falls back to signalling when it cannot verify (unreachable, or an old
one-line pidfile). See
[The observe daemon](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md#the-observe-daemon).

## `lousho prompt`

Expand Down
35 changes: 34 additions & 1 deletion src/cli/observe.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,8 @@ describe('lousho observe', () => {
});

it('start on the daemon\'s own port reports "already running" and exits 0', async () => {
const daemon = await start();
const daemon = await startObserveDaemon({ dbPath: dbPath(), port: 0 });
daemons.push(daemon);
expect(await runObserve(['start', '--port', String(daemon.port), '--db', dbPath()])).toBe(0);
expect(logs.at(-1)).toContain('already running');
});
Expand Down Expand Up @@ -93,6 +94,38 @@ describe('lousho observe', () => {
expect(fs.existsSync(pidPath)).toBe(false);
});

it('start --db b.db against a daemon serving a.db exits 1 naming the other db', async () => {
const a = path.join(cwd, 'a.db');
const daemon = await startObserveDaemon({ dbPath: a, port: 0 });
daemons.push(daemon);
expect(await runObserve(['start', '--port', String(daemon.port), '--db', path.join(cwd, 'b.db')])).toBe(1);
expect(errors.at(-1)).toContain(`port ${daemon.port} is serving ${path.resolve(a)}`);
// the same db is still "already running"
expect(await runObserve(['start', '--port', String(daemon.port), '--db', a])).toBe(0);
expect(logs.at(-1)).toContain('already running');
});

it('stop --db a.db does not signal a pid whose daemon serves b.db', async () => {
const b = await startObserveDaemon({ dbPath: path.join(cwd, 'b.db'), port: 0 });
daemons.push(b);
const a = path.join(cwd, 'a.db');
fs.writeFileSync(defaultPidPath(a), `${process.pid}\n${b.port}\n`);
const kill = vi.spyOn(process, 'kill').mockImplementation(() => true);
expect(await runObserve(['stop', '--db', a])).toBe(1);
expect(kill).not.toHaveBeenCalled();
expect(errors.at(-1)).toContain('serves');
expect(fs.existsSync(defaultPidPath(a))).toBe(true);
});

it('stop signals the pid when its daemon serves this db', async () => {
const a = path.join(cwd, 'a.db');
const d = await startObserveDaemon({ dbPath: a, port: 0, pidPath: defaultPidPath(a) });
daemons.push(d);
const kill = vi.spyOn(process, 'kill').mockImplementation(() => true);
expect(await runObserve(['stop', '--db', a])).toBe(0);
expect(kill).toHaveBeenCalledWith(process.pid, 'SIGTERM');
});

it('rebuild-rollups rebuilds the buckets of a local db', async () => {
expect(await runObserve(['rebuild-rollups', '--db', dbPath()])).toBe(0);
expect(logs.at(-1)).toContain('rebuilt');
Expand Down
78 changes: 60 additions & 18 deletions src/cli/observe.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import {
defaultDbPath,
defaultPidPath,
isObserveDaemon,
sameDbPath,
probeHealth,
startObserveDaemon,
type HealthProbe,
Expand Down Expand Up @@ -114,6 +115,25 @@ function envPort(): number {

/* ------------------------------------------------------------------- start */

/** `start` hit EADDRINUSE: say who owns the port. 0 = this db's daemon already runs; 1 = anything else. */
async function reportPortInUse(port: number, host: string, dbPath: string): Promise<number> {
const health = await probeHealth(port, host);
if (!isObserveDaemon(health)) {
console.error(
`lousho observe: port ${port} is taken by another service (its /health is not ${OBSERVE_SERVICE}). Pass --port <free>, or stop that service.`
);
return 1;
}
if (health.db !== undefined && !sameDbPath(health.db, dbPath)) {
console.error(
`lousho observe: port ${port} is serving ${health.db}, not ${dbPath}. Pass --port <free> for this db, or stop that daemon first.`
);
return 1;
}
console.log(`lousho observe: already running on http://${host}:${port} (service ${OBSERVE_SERVICE}, version ${health.version ?? 'unknown'}).`);
return 0;
}

async function runStart(args: string[]): Promise<number> {
const spec = SUB_SPECS.start;
const { values, help } = parseCommand(spec, args);
Expand All @@ -132,17 +152,7 @@ async function runStart(args: string[]): Promise<number> {
try {
daemon = await startObserveDaemon({ dbPath, port, host, requireToken, pidPath, webhooks });
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'EADDRINUSE') {
const health = await probeHealth(port, host);
if (isObserveDaemon(health)) {
console.log(`lousho observe: already running on http://${host}:${port} (service ${OBSERVE_SERVICE}, version ${health.version ?? 'unknown'}).`);
return 0;
}
console.error(
`lousho observe: port ${port} is taken by another service (its /health is not ${OBSERVE_SERVICE}). Pass --port <free>, or stop that service.`
);
return 1;
}
if ((error as NodeJS.ErrnoException).code === 'EADDRINUSE') return reportPortInUse(port, host, dbPath);
throw error;
}

Expand Down Expand Up @@ -215,6 +225,36 @@ async function runStatus(args: string[]): Promise<number> {

/* -------------------------------------------------------------------- stop */

/** The pidfile's pid (line 1) and, for newer daemons, the port it listens on (line 2). */
function readPidfile(pidPath: string): { pid?: number; port?: number } {
let text: string;
try {
text = fs.readFileSync(pidPath, 'utf8');
} catch {
return {};
}
const [pidLine = '', portLine] = text.split(/\r?\n/);
const pid = Number(pidLine.trim());
const port = Number(portLine?.trim());
return {
...(Number.isInteger(pid) && pid > 0 ? { pid } : {}),
...(Number.isInteger(port) && port > 0 ? { port } : {}),
};
}

/**
* What the daemon at the pidfile's port serves when that is not `dbPath`
* (a different db, or another service); `undefined` when it matches or cannot
* be verified (no port recorded, nothing answers).
*/
async function stopMismatch(port: number | undefined, dbPath: string): Promise<string | undefined> {
if (port === undefined) return undefined;
const health = await probeHealth(port);
if (health === undefined) return undefined;
if (!isObserveDaemon(health)) return 'another service';
return health.db !== undefined && !sameDbPath(health.db, dbPath) ? health.db : undefined;
}

async function runStop(args: string[]): Promise<number> {
const spec = SUB_SPECS.stop;
const { values, help } = parseCommand(spec, args);
Expand All @@ -224,18 +264,20 @@ async function runStop(args: string[]): Promise<number> {
}
const dbPath = dbPathOf(values.db);
const pidPath = defaultPidPath(dbPath);
let pid: number | undefined;
try {
pid = Number(fs.readFileSync(pidPath, 'utf8').trim());
} catch {
pid = undefined;
}
if (pid === undefined || !Number.isInteger(pid) || pid <= 0) {
const { pid, port: pidPort } = readPidfile(pidPath);
if (pid === undefined) {
console.error(
`lousho observe: no pidfile at ${pidPath}. If a daemon is running, stop it with Ctrl+C (or 'kill <pid>' / Task Manager).`
);
return 1;
}
const mismatch = await stopMismatch(pidPort, dbPath);
if (mismatch !== undefined) {
console.error(
`lousho observe: pid ${pid} in ${pidPath} serves ${mismatch}, not ${dbPath}; not signalling it. Remove the stale pidfile if it is wrong.`
);
return 1;
}
try {
process.kill(pid, 'SIGTERM');
} catch (error) {
Expand Down
23 changes: 21 additions & 2 deletions src/observe/daemon.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,10 @@
* port-collision detection.
*/
import { afterEach, describe, expect, it } from 'vitest';
import { isObserveDaemon, probeHealth, startObserveDaemon, type ObserveDaemon } from './daemon';
import * as nodePath from 'node:path';
import { defaultPidPath, isObserveDaemon, probeHealth, sameDbPath, startObserveDaemon, type ObserveDaemon } from './daemon';
import { OBSERVE_SERVICE } from './envelope';
import { createToken } from './store';
import { createToken, openDb } from './store';

const T0 = 1_700_000_000_000;

Expand Down Expand Up @@ -57,6 +58,24 @@ describe('GET /health', () => {
});
});

describe('per-db pidfile and db identity', () => {
it('names the pidfile after the db so two dbs in one folder never share it', () => {
expect(defaultPidPath('/d/a.db')).toBe(nodePath.resolve('/d/a.pid'));
expect(defaultPidPath('/d/b.db')).toBe(nodePath.resolve('/d/b.pid'));
expect(defaultPidPath('/d/observe.db')).toBe(nodePath.resolve('/d/observe.pid'));
expect(defaultPidPath('/d/data.sqlite')).toBe(nodePath.resolve('/d/data.sqlite.pid'));
});

it('/health reports the absolute db path; sameDbPath compares resolved paths', async () => {
const daemon = await start({ dbPath: 'mem-test.db', db: openDb(':memory:') });
const health = await probeHealth(daemon.port);
expect(health?.db).toBe(nodePath.resolve('mem-test.db'));
expect(sameDbPath(health?.db, 'mem-test.db')).toBe(true);
expect(sameDbPath(health?.db, 'other.db')).toBe(false);
expect(sameDbPath(undefined, 'other.db')).toBe(false);
});
});

describe('POST /api/v1/ingest', () => {
it('accepts a batch and serves it back through the read endpoints', async () => {
const daemon = await start();
Expand Down
24 changes: 21 additions & 3 deletions src/observe/daemon.ts
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,23 @@ export function defaultDbPath(homeDir: string = os.homedir()): string {
return path.join(homeDir, '.lousho', 'observe.db');
}

/** Where the pidfile goes for a given db path (same directory, fixed name). */
/**
* Where the pidfile goes for a given db path: `<dir>/<db name without .db>.pid`,
* so two daemons whose dbs share a folder never share a pidfile
* (`observe.db` keeps `observe.pid`).
*/
export function defaultPidPath(dbPath: string): string {
return path.join(path.dirname(resolveForFs(dbPath)), 'observe.pid');
const resolved = resolveForFs(dbPath);
const base = path.basename(resolved);
const stem = base.toLowerCase().endsWith('.db') && base.length > 3 ? base.slice(0, -3) : base;
return path.join(path.dirname(resolved), `${stem}.pid`);
}

/** Whether two db paths name the same file (resolved; case-insensitive on win32). */
export function sameDbPath(a: string | undefined, b: string): boolean {
if (a === undefined) return false;
const norm = (p: string): string => (process.platform === 'win32' ? resolveForFs(p).toLowerCase() : resolveForFs(p));
return norm(a) === norm(b);
}

function resolveForFs(p: string): string {
Expand Down Expand Up @@ -154,6 +168,7 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr
let bodyRead: Promise<string> | undefined;
const ctx: RouteContext = {
db,
dbPath,
req,
res,
params: match.params,
Expand Down Expand Up @@ -205,7 +220,8 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr
const pidPath = options.pidPath;
if (pidPath !== undefined) {
fs.mkdirSync(path.dirname(pidPath), { recursive: true });
fs.writeFileSync(pidPath, String(process.pid), 'utf8');
// Line 2 is the port, so `stop` can ask this daemon which db it serves.
fs.writeFileSync(pidPath, `${process.pid}\n${boundPort}\n`, 'utf8');
}

// The monitoring sweep: rewrites agents.status from heartbeat freshness,
Expand Down Expand Up @@ -250,6 +266,8 @@ export interface HealthProbe {
service: string;
ok: boolean;
version?: string;
/** Absolute path of the db the daemon serves. */
db?: string;
}

/**
Expand Down
2 changes: 2 additions & 0 deletions src/observe/routes/httpKit.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@ export { sendJson };
/** What a handler gets: the request, the matched `:params`, the query, the db, and a bounded body reader. */
export interface RouteContext {
db: SqlDatabase;
/** The db file path the daemon was started with (`:memory:` for tests). */
dbPath: string;
req: http.IncomingMessage;
res: http.ServerResponse;
params: Record<string, string>;
Expand Down
1 change: 1 addition & 0 deletions src/observe/routes/scores.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ function get(database: SqlDatabase, query: string): Captured {
if (!matched) throw new Error('scores route not registered');
matched.route.handle({
db: database,
dbPath: ':memory:',
req: {} as http.IncomingMessage,
res,
params: matched.params,
Expand Down
6 changes: 4 additions & 2 deletions src/observe/routes/status.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
* row-count surface `lousho observe status` prints.
*/

import * as path from 'node:path';
import { VERSION } from '../../version';
import { observeCounts } from '../store';
import { sendJson, type Route } from './httpKit';
Expand All @@ -14,8 +15,9 @@ export const statusRoutes: readonly Route[] = [
{
method: 'GET',
path: '/health',
handle: ({ res }) => {
sendJson(res, 200, { service: 'lousho-observe', ok: true, version: VERSION });
handle: ({ res, dbPath }) => {
// `db` lets `start`/`stop` tell which database a port or pid serves.
sendJson(res, 200, { service: 'lousho-observe', ok: true, version: VERSION, db: dbPath === ':memory:' ? dbPath : path.resolve(dbPath) });
},
},
{
Expand Down
Loading