331 lines
10 KiB
JavaScript
331 lines
10 KiB
JavaScript
/**
|
|
* PearMonitor agent RPC handlers — metrics, anomalies, alerts, jobs, ACL.
|
|
*/
|
|
import os from 'os'
|
|
import {
|
|
APP_NAME,
|
|
APP_VERSION,
|
|
PROTOCOL,
|
|
PROTOCOL_VERSION,
|
|
Roles,
|
|
} from '../../shared/protocol.js'
|
|
import { SCHEMA_VERSION } from '../../shared/schema.js'
|
|
import {
|
|
getAllChartDefs,
|
|
getContextIds,
|
|
CHART_BY_ID,
|
|
chartSummary,
|
|
} from '../../shared/metrics.js'
|
|
import { peers } from '../core/peer-registry.js'
|
|
import {
|
|
listPolicyPeers,
|
|
revokePeer as policyRevoke,
|
|
touchPeer,
|
|
} from '../core/peer-policy.js'
|
|
import { signCapability, encodeInvite } from '../../shared/crypto-auth.js'
|
|
import { getMacKey, getServerPublicKeyHex } from '../core/auth-keys.js'
|
|
import { getCollector } from '../services/collector.js'
|
|
import { getStore } from '../services/store.js'
|
|
import { getAnomalyEngine } from '../services/anomaly.js'
|
|
import { computeWeights } from '../services/weights.js'
|
|
import { queryLogs } from '../services/logs.js'
|
|
import {
|
|
listAlerts,
|
|
getAlert,
|
|
setAlertConfig,
|
|
ackAlert,
|
|
silenceAlert,
|
|
} from '../services/alerts.js'
|
|
import {
|
|
subscribeMetrics,
|
|
unsubscribeMetrics,
|
|
subscribeAnomalies,
|
|
unsubscribeAnomalies,
|
|
} from '../services/subscriptions.js'
|
|
import { getJobs, knownJobNames } from '../services/jobs.js'
|
|
import { formatAllMetrics } from '../rest/formatters.js'
|
|
import { getDb } from '../db/index.js'
|
|
import { attachLinkedPeer, isSwarmEnabled } from '../db/replicate.js'
|
|
import { listRemoteDbs } from '../db/remote.js'
|
|
import { getParentCollector, isParentEnabled } from '../services/collectors/parent.js'
|
|
|
|
/**
|
|
* @param {import('../rpc/session.js').PeerSession} session
|
|
*/
|
|
export function registerMonitorHandlers(session) {
|
|
const collector = getCollector()
|
|
const store = getStore()
|
|
const anomalies = getAnomalyEngine(os.cpus().length)
|
|
|
|
session.respond('ping', async () => ({
|
|
ok: true,
|
|
pong: Date.now(),
|
|
peerId: session.id,
|
|
}), { hot: true })
|
|
|
|
session.respond('getServerInfo', async () => {
|
|
let tunnel = { enabled: false }
|
|
try {
|
|
const { getRestTunnelInfo } = await import('../services/rest-tunnel.js')
|
|
tunnel = getRestTunnelInfo()
|
|
} catch {
|
|
// ignore
|
|
}
|
|
return {
|
|
app: APP_NAME,
|
|
version: APP_VERSION,
|
|
protocol: PROTOCOL,
|
|
protocolVersion: PROTOCOL_VERSION,
|
|
schemaVersion: SCHEMA_VERSION,
|
|
publicKeyHex: getServerPublicKeyHex(),
|
|
hostname: os.hostname(),
|
|
platform: `${os.platform()}/${os.arch()}`,
|
|
uptimeSec: Math.floor(
|
|
typeof process?.uptime === 'function' ? process.uptime() : 0
|
|
),
|
|
connectedPeers: peers.size(),
|
|
node: typeof process?.version === 'string' ? process.version : 'bare',
|
|
role: 'agent',
|
|
charts: getAllChartDefs().length,
|
|
anomalyMode: process.env.PEARDATA_ANOMALY_MODE || 'threshold',
|
|
restTunnel: tunnel,
|
|
}
|
|
})
|
|
|
|
session.respond('getAuthStatus', async () => ({
|
|
peerId: session.id,
|
|
role: session.role,
|
|
authMode: session.authMode,
|
|
displayName: session.displayName,
|
|
}))
|
|
|
|
session.respond('setDisplayName', async (args, s) => {
|
|
s.displayName = args.name
|
|
touchPeer(s.id, args.name)
|
|
return { success: true, displayName: args.name }
|
|
})
|
|
|
|
session.respond('getNodeInfo', async () =>
|
|
collector.getNodeInfo(getServerPublicKeyHex(), APP_VERSION)
|
|
)
|
|
|
|
session.respond('getHealth', async () => anomalies.getHealth())
|
|
|
|
session.respond('listContexts', async () => ({
|
|
contexts: getContextIds().map((id) => {
|
|
const charts = getAllChartDefs().filter((c) => c.context === id)
|
|
return {
|
|
id,
|
|
family: charts[0]?.family || id.split('.')[0],
|
|
title: charts[0]?.title || id,
|
|
charts: charts.map((c) => c.id),
|
|
}
|
|
}),
|
|
}))
|
|
|
|
session.respond('getContext', async (args) => {
|
|
const charts = getAllChartDefs().filter((c) => c.context === args.id || c.id === args.id)
|
|
if (!charts.length) return { error: 'unknown context', id: args.id }
|
|
return {
|
|
id: args.id,
|
|
charts: charts.map((c) => store.getMeta(c.id) || chartSummary(c)),
|
|
}
|
|
})
|
|
|
|
session.respond('listCharts', async () => ({
|
|
charts: store.listChartSummaries(),
|
|
hostname: os.hostname(),
|
|
version: APP_VERSION,
|
|
}))
|
|
|
|
session.respond('getChart', async (args) => {
|
|
const meta = store.getMeta(args.id)
|
|
if (!meta) {
|
|
const def = CHART_BY_ID.get(args.id)
|
|
if (!def) return { error: 'unknown chart', id: args.id }
|
|
return chartSummary(def)
|
|
}
|
|
return meta
|
|
})
|
|
|
|
session.respond('queryData', async (args) => store.query(args), { hot: true })
|
|
session.respond('getWeights', async (args) => computeWeights(args || {}), { hot: true })
|
|
|
|
session.respond('queryLogs', async (args) =>
|
|
queryLogs({ ...(args || {}), role: session.role })
|
|
)
|
|
|
|
session.respond('getDbInfo', async () => {
|
|
const db = getDb()
|
|
if (!db) return { enabled: false }
|
|
return {
|
|
enabled: true,
|
|
publicKeyHex: db.publicKeyHex,
|
|
discoveryKeyHex: db.discoveryKeyHex,
|
|
swarm: isSwarmEnabled(),
|
|
remotes: listRemoteDbs(),
|
|
}
|
|
})
|
|
|
|
session.respond('listPeerLinks', async () => {
|
|
const db = getDb()
|
|
if (!db) return { links: [], enabled: false }
|
|
const links = await db.listPeerLinks(getServerPublicKeyHex().slice(0, 16))
|
|
return { links, enabled: true }
|
|
})
|
|
|
|
session.respond('linkPeer', async (args) => {
|
|
const db = getDb()
|
|
if (!db) return { success: false, error: 'HyperDB disabled' }
|
|
const localNodeId = getServerPublicKeyHex().slice(0, 16)
|
|
await db.putPeerLink({
|
|
localNodeId,
|
|
remotePublicKey: args.remotePublicKey,
|
|
role: args.role || 'viewer',
|
|
alias: args.alias || null,
|
|
dbKeyHex: args.dbKeyHex || null,
|
|
discoveryKeyHex: args.discoveryKeyHex || null,
|
|
syncMode: args.syncMode || 'both',
|
|
})
|
|
const attach = await attachLinkedPeer({
|
|
discoveryKeyHex: args.discoveryKeyHex || null,
|
|
dbKeyHex: args.dbKeyHex || null,
|
|
syncMode: args.syncMode || 'both',
|
|
})
|
|
return {
|
|
success: true,
|
|
linked: true,
|
|
swarmJoined: attach.joined,
|
|
remoteOpened: attach.opened,
|
|
}
|
|
})
|
|
|
|
session.respond('getFleetHealth', async () => {
|
|
if (!isParentEnabled()) {
|
|
return {
|
|
enabled: false,
|
|
local: anomalies.getHealth(),
|
|
children: [],
|
|
}
|
|
}
|
|
const parent = getParentCollector()
|
|
return {
|
|
enabled: true,
|
|
local: anomalies.getHealth(),
|
|
children: parent.listChildren(),
|
|
summary: parent.fleetSummary(),
|
|
}
|
|
})
|
|
|
|
session.respond('listChildPeers', async () => {
|
|
if (!isParentEnabled()) return { enabled: false, children: [] }
|
|
return { enabled: true, children: getParentCollector().listChildren() }
|
|
})
|
|
|
|
session.respond('unlinkPeer', async (args) => {
|
|
const db = getDb()
|
|
if (!db) return { success: false, error: 'HyperDB disabled' }
|
|
await db.deletePeerLink(getServerPublicKeyHex().slice(0, 16), args.remotePublicKey)
|
|
return { success: true }
|
|
})
|
|
|
|
session.respond('getAllMetrics', async (args) => formatAllMetrics(args.format || 'json'))
|
|
|
|
session.respond('subscribeMetrics', async (args, s) => subscribeMetrics(s, args), {
|
|
hot: true,
|
|
})
|
|
session.respond('unsubscribeMetrics', async (_a, s) => unsubscribeMetrics(s), { hot: true })
|
|
session.respond('subscribeAnomalies', async (_a, s) => subscribeAnomalies(s))
|
|
session.respond('unsubscribeAnomalies', async (_a, s) => unsubscribeAnomalies(s))
|
|
|
|
session.respond('listAnomalies', async (args) => ({
|
|
anomalies: anomalies.listRecent(args?.limit || 50),
|
|
}))
|
|
|
|
session.respond('listAlerts', async () => ({ alerts: listAlerts() }))
|
|
session.respond('getAlert', async (args) => {
|
|
const a = getAlert(args.id)
|
|
return a || { error: 'unknown alert', id: args.id }
|
|
})
|
|
session.respond('setAlertConfig', async (args) => ({
|
|
success: true,
|
|
config: setAlertConfig(args),
|
|
}))
|
|
session.respond('ackAlert', async (args) => ackAlert(args.id))
|
|
session.respond('silenceAlert', async (args) => silenceAlert(args.id, args))
|
|
|
|
session.respond('listJobs', async () => ({
|
|
jobs: getJobs().list(),
|
|
known: knownJobNames(),
|
|
}))
|
|
session.respond('runJob', async (args) => getJobs().run(args.name, args.args || {}))
|
|
session.respond('cancelJob', async (args) => getJobs().cancel(args.id))
|
|
|
|
session.respond('mintInvite', async (args) => {
|
|
const role = args.role || Roles.operator
|
|
const ttlMs = args.ttlMs === undefined ? null : args.ttlMs
|
|
const { token, payload } = signCapability(getMacKey(), {
|
|
role,
|
|
ttlMs,
|
|
forever: ttlMs == null || ttlMs === 0,
|
|
peerId: args.peerId || null,
|
|
})
|
|
const invite = encodeInvite({
|
|
publicKeyHex: getServerPublicKeyHex(),
|
|
capability: token,
|
|
role: payload.role,
|
|
jti: payload.jti,
|
|
alias: args.alias || null,
|
|
expiresAt: payload.exp,
|
|
})
|
|
return {
|
|
success: true,
|
|
invite,
|
|
capability: token,
|
|
role: payload.role,
|
|
jti: payload.jti,
|
|
exp: payload.exp,
|
|
}
|
|
})
|
|
|
|
session.respond('listPeers', async () => ({
|
|
connected: peers.list().map((s) => ({
|
|
peerId: s.id,
|
|
role: s.role,
|
|
displayName: s.displayName,
|
|
authMode: s.authMode,
|
|
})),
|
|
known: listPolicyPeers(),
|
|
}))
|
|
|
|
session.respond('revokePeer', async (args) => {
|
|
policyRevoke(args.peerId)
|
|
const live = peers.get(args.peerId)
|
|
if (live) live.destroy()
|
|
return { success: true, peerId: args.peerId }
|
|
})
|
|
|
|
session.respond('exportSnapshot', async (args) => {
|
|
const { buildSnapshot, writeSnapshotFile, toPrometheusText, pushPrometheusText } =
|
|
await import('../services/export.js')
|
|
const snapshot = buildSnapshot()
|
|
const written = args?.write ? writeSnapshotFile(snapshot) : { path: null, bytes: 0 }
|
|
let push = null
|
|
if (args?.push || args?.pushUrl || process.env.PEARDATA_PUSHGATEWAY_URL) {
|
|
try {
|
|
const url = args?.pushUrl || process.env.PEARDATA_PUSHGATEWAY_URL
|
|
if (url) {
|
|
push = await pushPrometheusText(url, toPrometheusText(snapshot.latest))
|
|
}
|
|
} catch (err) {
|
|
push = { error: err.message }
|
|
}
|
|
}
|
|
return {
|
|
...snapshot,
|
|
file: written.path,
|
|
push,
|
|
}
|
|
})
|
|
}
|