refactor(core): replace legacy logger with Effect logging (#31310)
This commit is contained in:
Dax
GitHub
parent
0a7cb20e66
commit
c06ad7c881
152 changed files with 697 additions and 2242 deletions
|
|
@ -1,73 +0,0 @@
|
|||
import { Cause, Effect, Logger, References } from "effect"
|
||||
import * as Log from "../util/log"
|
||||
|
||||
type Fields = Record<string, unknown>
|
||||
|
||||
const normalizeKey = (key: string) => (key === "sessionID" ? "session.id" : key)
|
||||
|
||||
export interface Handle {
|
||||
readonly debug: (msg?: unknown, extra?: Fields) => Effect.Effect<void>
|
||||
readonly info: (msg?: unknown, extra?: Fields) => Effect.Effect<void>
|
||||
readonly warn: (msg?: unknown, extra?: Fields) => Effect.Effect<void>
|
||||
readonly error: (msg?: unknown, extra?: Fields) => Effect.Effect<void>
|
||||
readonly with: (extra: Fields) => Handle
|
||||
}
|
||||
|
||||
const clean = (input?: Fields): Fields =>
|
||||
Object.fromEntries(
|
||||
Object.entries(input ?? {})
|
||||
.filter((entry) => entry[1] !== undefined && entry[1] !== null)
|
||||
.map(([key, value]) => [normalizeKey(key), value]),
|
||||
)
|
||||
|
||||
const text = (input: unknown): string => {
|
||||
// oxlint-disable-next-line no-base-to-string
|
||||
if (Array.isArray(input)) return input.map((item) => String(item)).join(" ")
|
||||
// oxlint-disable-next-line no-base-to-string
|
||||
return input === undefined ? "" : String(input)
|
||||
}
|
||||
|
||||
const call = (run: (msg?: unknown) => Effect.Effect<void>, base: Fields, msg?: unknown, extra?: Fields) => {
|
||||
const ann = clean({ ...base, ...extra })
|
||||
const fx = run(msg)
|
||||
return Object.keys(ann).length ? Effect.annotateLogs(fx, ann) : fx
|
||||
}
|
||||
|
||||
export const logger = Logger.make((opts) => {
|
||||
const extra = clean(opts.fiber.getRef(References.CurrentLogAnnotations))
|
||||
const now = opts.date.getTime()
|
||||
for (const [key, start] of opts.fiber.getRef(References.CurrentLogSpans)) {
|
||||
extra[`logSpan.${key}`] = `${now - start}ms`
|
||||
}
|
||||
if (opts.cause.reasons.length > 0) {
|
||||
extra.cause = Cause.pretty(opts.cause)
|
||||
}
|
||||
|
||||
const svc = typeof extra.service === "string" ? extra.service : undefined
|
||||
if (svc) delete extra.service
|
||||
const log = svc ? Log.create({ service: svc }) : Log.Default
|
||||
const msg = text(opts.message)
|
||||
|
||||
switch (opts.logLevel) {
|
||||
case "Trace":
|
||||
case "Debug":
|
||||
return log.debug(msg, extra)
|
||||
case "Warn":
|
||||
return log.warn(msg, extra)
|
||||
case "Error":
|
||||
case "Fatal":
|
||||
return log.error(msg, extra)
|
||||
default:
|
||||
return log.info(msg, extra)
|
||||
}
|
||||
})
|
||||
|
||||
export const layer = Logger.layer([logger], { mergeWithExisting: false })
|
||||
|
||||
export const create = (base: Fields = {}): Handle => ({
|
||||
debug: (msg, extra) => call((item) => Effect.logDebug(item), base, msg, extra),
|
||||
info: (msg, extra) => call((item) => Effect.logInfo(item), base, msg, extra),
|
||||
warn: (msg, extra) => call((item) => Effect.logWarning(item), base, msg, extra),
|
||||
error: (msg, extra) => call((item) => Effect.logError(item), base, msg, extra),
|
||||
with: (extra) => create({ ...base, ...extra }),
|
||||
})
|
||||
|
|
@ -1,107 +0,0 @@
|
|||
import { Effect, Layer, Logger } from "effect"
|
||||
import { FetchHttpClient } from "effect/unstable/http"
|
||||
import { OtlpLogger, OtlpSerialization } from "effect/unstable/observability"
|
||||
import * as EffectLogger from "./logger"
|
||||
import { Flag } from "../flag/flag"
|
||||
import { InstallationChannel, InstallationVersion } from "../installation/version"
|
||||
import { ensureProcessMetadata } from "../util/opencode-process"
|
||||
|
||||
const base = Flag.OTEL_EXPORTER_OTLP_ENDPOINT
|
||||
export const enabled = !!base
|
||||
const processID = crypto.randomUUID()
|
||||
|
||||
const headers = Flag.OTEL_EXPORTER_OTLP_HEADERS
|
||||
? Flag.OTEL_EXPORTER_OTLP_HEADERS.split(",").reduce(
|
||||
(acc, x) => {
|
||||
const [key, ...value] = x.split("=")
|
||||
acc[key] = value.join("=")
|
||||
return acc
|
||||
},
|
||||
{} as Record<string, string>,
|
||||
)
|
||||
: undefined
|
||||
|
||||
export function resource(): { serviceName: string; serviceVersion: string; attributes: Record<string, string> } {
|
||||
const processMetadata = ensureProcessMetadata("main")
|
||||
const attributes: Record<string, string> = (() => {
|
||||
const value = process.env.OTEL_RESOURCE_ATTRIBUTES
|
||||
if (!value) return {}
|
||||
try {
|
||||
return Object.fromEntries(
|
||||
value.split(",").map((entry) => {
|
||||
const index = entry.indexOf("=")
|
||||
if (index < 1) throw new Error("Invalid OTEL_RESOURCE_ATTRIBUTES entry")
|
||||
return [decodeURIComponent(entry.slice(0, index)), decodeURIComponent(entry.slice(index + 1))]
|
||||
}),
|
||||
)
|
||||
} catch {
|
||||
return {}
|
||||
}
|
||||
})()
|
||||
|
||||
return {
|
||||
serviceName: "opencode",
|
||||
serviceVersion: InstallationVersion,
|
||||
attributes: {
|
||||
...attributes,
|
||||
"deployment.environment.name": InstallationChannel,
|
||||
"opencode.client": Flag.OPENCODE_CLIENT,
|
||||
"opencode.process_role": processMetadata.processRole,
|
||||
"opencode.run_id": processMetadata.runID,
|
||||
"service.instance.id": processID,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function logs() {
|
||||
return Logger.layer(
|
||||
[
|
||||
EffectLogger.logger,
|
||||
OtlpLogger.make({
|
||||
url: `${base}/v1/logs`,
|
||||
resource: resource(),
|
||||
headers,
|
||||
}),
|
||||
],
|
||||
{ mergeWithExisting: false },
|
||||
).pipe(Layer.provide(OtlpSerialization.layerJson), Layer.provide(FetchHttpClient.layer))
|
||||
}
|
||||
|
||||
const traces = async () => {
|
||||
const NodeSdk = await import("@effect/opentelemetry/NodeSdk")
|
||||
const OTLP = await import("@opentelemetry/exporter-trace-otlp-http")
|
||||
const SdkBase = await import("@opentelemetry/sdk-trace-base")
|
||||
|
||||
// @effect/opentelemetry creates a NodeTracerProvider but never calls
|
||||
// register(), so the global @opentelemetry/api context manager stays
|
||||
// as the no-op default. Non-Effect code (like the AI SDK) that calls
|
||||
// tracer.startActiveSpan() relies on context.active() to find the
|
||||
// parent span - without a real context manager every span starts a
|
||||
// new trace. Registering AsyncLocalStorageContextManager fixes this.
|
||||
const { AsyncLocalStorageContextManager } = await import("@opentelemetry/context-async-hooks")
|
||||
const { context } = await import("@opentelemetry/api")
|
||||
const mgr = new AsyncLocalStorageContextManager()
|
||||
mgr.enable()
|
||||
context.setGlobalContextManager(mgr)
|
||||
|
||||
return NodeSdk.layer(() => ({
|
||||
resource: resource(),
|
||||
spanProcessor: new SdkBase.BatchSpanProcessor(
|
||||
new OTLP.OTLPTraceExporter({
|
||||
url: `${base}/v1/traces`,
|
||||
headers,
|
||||
}),
|
||||
),
|
||||
}))
|
||||
}
|
||||
|
||||
export const layer = !base
|
||||
? EffectLogger.layer
|
||||
: Layer.unwrap(
|
||||
Effect.gen(function* () {
|
||||
const trace = yield* Effect.promise(traces)
|
||||
return Layer.mergeAll(trace, logs())
|
||||
}),
|
||||
)
|
||||
|
||||
export const Observability = { enabled, layer }
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
import { Layer, type Context, ManagedRuntime, type Effect } from "effect"
|
||||
import { memoMap } from "./memo-map"
|
||||
import { Observability } from "./observability"
|
||||
import { Observability } from "../observability"
|
||||
|
||||
export function makeRuntime<I, S, E>(service: Context.Service<I, S>, layer: Layer.Layer<I, E>) {
|
||||
let rt: ManagedRuntime.ManagedRuntime<I, E> | undefined
|
||||
|
|
|
|||
|
|
@ -410,9 +410,7 @@ export const layerWith = (options?: LayerOptions) =>
|
|||
Effect.catchCauseIf(
|
||||
(cause) => !Cause.hasInterrupts(cause),
|
||||
(cause) =>
|
||||
Effect.logError("Event observer failed").pipe(
|
||||
Effect.annotateLogs({ eventID: event.id, eventType: event.type, kind, cause }),
|
||||
),
|
||||
Effect.logError("Event observer failed", { eventID: event.id, eventType: event.type, kind, cause }),
|
||||
),
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -10,11 +10,8 @@ import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner
|
|||
import { CrossSpawnSpawner } from "../cross-spawn-spawner"
|
||||
import { Global } from "../global"
|
||||
import { NonNegativeInt } from "../schema"
|
||||
import * as Log from "../util/log"
|
||||
import { sanitizedProcessEnv } from "../util/opencode-process"
|
||||
import { which } from "../util/which"
|
||||
|
||||
const log = Log.create({ service: "ripgrep" })
|
||||
const VERSION = "15.1.0"
|
||||
const PLATFORM = {
|
||||
"arm64-darwin": { platform: "aarch64-apple-darwin", extension: "tar.gz" },
|
||||
|
|
@ -146,7 +143,9 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Ri
|
|||
export const use = serviceUse(Service)
|
||||
|
||||
function env() {
|
||||
const env = sanitizedProcessEnv()
|
||||
const env = Object.fromEntries(
|
||||
Object.entries(process.env).filter((entry): entry is [string, string] => entry[1] !== undefined),
|
||||
)
|
||||
delete env.RIPGREP_CONFIG_PATH
|
||||
return env
|
||||
}
|
||||
|
|
@ -307,7 +306,7 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | ChildProcessSpa
|
|||
const url = `https://github.com/BurntSushi/ripgrep/releases/download/${VERSION}/${filename}`
|
||||
const archive = path.join(Global.Path.bin, filename)
|
||||
|
||||
log.info("downloading ripgrep", { url })
|
||||
yield* Effect.logInfo("downloading ripgrep", { url })
|
||||
yield* fs.ensureDir(Global.Path.bin).pipe(Effect.orDie)
|
||||
|
||||
const bytes = yield* HttpClientRequest.get(url).pipe(
|
||||
|
|
@ -418,7 +417,7 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | ChildProcessSpa
|
|||
})
|
||||
|
||||
const tree: Interface["tree"] = Effect.fn("Ripgrep.tree")(function* (input: TreeInput) {
|
||||
log.info("tree", input)
|
||||
yield* Effect.logInfo("tree", input)
|
||||
const list = Array.from(yield* files({ cwd: input.cwd, signal: input.signal }).pipe(Stream.runCollect))
|
||||
|
||||
interface Node {
|
||||
|
|
|
|||
|
|
@ -4,13 +4,11 @@ import type { PlatformError } from "effect/PlatformError"
|
|||
import { FSUtil } from "../fs-util"
|
||||
import { Glob } from "../util/glob"
|
||||
import { Global } from "../global"
|
||||
import * as Log from "../util/log"
|
||||
import { serviceUse } from "../effect/service-use"
|
||||
import { makeRuntime } from "../effect/runtime"
|
||||
import { Fff } from "#fff"
|
||||
import { Ripgrep } from "./ripgrep"
|
||||
|
||||
const log = Log.create({ service: "file.search" })
|
||||
const root = path.join(Global.Path.cache, "fff")
|
||||
|
||||
export type Item = Ripgrep.Item
|
||||
|
|
@ -220,12 +218,14 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
if (!scanned.ok || !scanned.value) {
|
||||
yield* fffSync("destroy picker", () => pick.destroy()).pipe(Effect.ignore)
|
||||
state.pick.delete(dir)
|
||||
log.warn("fff scan not ready", { dir })
|
||||
yield* Effect.logWarning("fff scan not ready", { dir })
|
||||
return yield* Effect.fail(new Error(scanned.ok ? "fff scan timed out" : scanned.error))
|
||||
}
|
||||
|
||||
const git = yield* fffSync("refresh git status", () => pick.refreshGitStatus())
|
||||
if (!git.ok) log.warn("fff git refresh failed", { dir, error: git.error })
|
||||
if (!git.ok) {
|
||||
yield* Effect.logWarning("fff git refresh failed", { dir, error: git.error })
|
||||
}
|
||||
})
|
||||
|
||||
// Create (or return) the picker for a directory. Creation is synchronous
|
||||
|
|
@ -244,10 +244,7 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
if (pending) return yield* Deferred.await(pending)
|
||||
|
||||
const available = yield* fffSync("check availability", () => Fff.available()).pipe(
|
||||
Effect.catch((error) => {
|
||||
log.warn("fff availability check failed", { error })
|
||||
return Effect.succeed(false)
|
||||
}),
|
||||
Effect.catch((error) => Effect.logWarning("fff availability check failed", { error }).pipe(Effect.as(false))),
|
||||
)
|
||||
if (!available) return undefined
|
||||
|
||||
|
|
@ -272,7 +269,7 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
}),
|
||||
)
|
||||
if (!made.ok) {
|
||||
log.warn("fff init failed", { dir, error: made.error })
|
||||
yield* Effect.logWarning("fff init failed", { dir, error: made.error })
|
||||
const err = new Error(made.error)
|
||||
yield* Deferred.fail(gate, err)
|
||||
return yield* Effect.fail(err)
|
||||
|
|
@ -355,14 +352,15 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
pageSize: limit,
|
||||
}),
|
||||
).pipe(
|
||||
Effect.catch((error) => {
|
||||
log.warn(`fff ${kind} search failed`, { dir, query, error })
|
||||
return Effect.succeed<Fff.Result<string[]> | undefined>(undefined)
|
||||
}),
|
||||
Effect.catch((error) =>
|
||||
Effect.logWarning(`fff ${kind} search failed`, { dir, query, error }).pipe(
|
||||
Effect.as<Fff.Result<string[]> | undefined>(undefined),
|
||||
),
|
||||
),
|
||||
)
|
||||
if (!fffResult) return undefined
|
||||
if (!fffResult.ok) {
|
||||
log.warn(`fff ${kind} search failed`, { dir, query, error: fffResult.error })
|
||||
yield* Effect.logWarning(`fff ${kind} search failed`, { dir, query, error: fffResult.error })
|
||||
return undefined
|
||||
}
|
||||
|
||||
|
|
@ -393,14 +391,15 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
timeBudgetMs: 1_500,
|
||||
}),
|
||||
).pipe(
|
||||
Effect.catch((error) => {
|
||||
log.warn("fff grep failed", { dir, pattern: input.pattern, error })
|
||||
return Effect.succeed<Fff.Result<Fff.Grep> | undefined>(undefined)
|
||||
}),
|
||||
Effect.catch((error) =>
|
||||
Effect.logWarning("fff grep failed", { dir, pattern: input.pattern, error }).pipe(
|
||||
Effect.as<Fff.Result<Fff.Grep> | undefined>(undefined),
|
||||
),
|
||||
),
|
||||
)
|
||||
if (!fffGrep) return yield* rip(input)
|
||||
if (!fffGrep.ok) {
|
||||
log.warn("fff grep failed", { dir, pattern: input.pattern, error: fffGrep.error })
|
||||
yield* Effect.logWarning("fff grep failed", { dir, pattern: input.pattern, error: fffGrep.error })
|
||||
return yield* rip(input)
|
||||
}
|
||||
|
||||
|
|
@ -432,10 +431,11 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
pageSize: limit,
|
||||
}),
|
||||
).pipe(
|
||||
Effect.catch((error) => {
|
||||
log.warn("fff glob failed", { dir, pattern: input.pattern, error })
|
||||
return Effect.succeed<Fff.Result<Fff.Search> | undefined>(undefined)
|
||||
}),
|
||||
Effect.catch((error) =>
|
||||
Effect.logWarning("fff glob failed", { dir, pattern: input.pattern, error }).pipe(
|
||||
Effect.as<Fff.Result<Fff.Search> | undefined>(undefined),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
if (fffGlob?.ok) {
|
||||
|
|
@ -453,7 +453,7 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
truncated: fffGlob.value.totalMatched > rows.length,
|
||||
}
|
||||
} else if (fffGlob) {
|
||||
log.warn("fff glob failed", { dir, pattern: input.pattern, error: fffGlob.error })
|
||||
yield* Effect.logWarning("fff glob failed", { dir, pattern: input.pattern, error: fffGlob.error })
|
||||
// fall through to the fallback
|
||||
}
|
||||
}
|
||||
|
|
@ -500,13 +500,16 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Ripgrep.Service
|
|||
if (!entry) return
|
||||
|
||||
const out = yield* fffSync("track query", () => entry.pick.trackQuery(row.text, file)).pipe(
|
||||
Effect.catch((error) => {
|
||||
log.warn("fff track query failed", { dir: row.dir, query: row.text, file, error })
|
||||
return Effect.succeed<Fff.Result<boolean> | undefined>(undefined)
|
||||
}),
|
||||
Effect.catch((error) =>
|
||||
Effect.logWarning("fff track query failed", { dir: row.dir, query: row.text, file, error }).pipe(
|
||||
Effect.as<Fff.Result<boolean> | undefined>(undefined),
|
||||
),
|
||||
),
|
||||
)
|
||||
if (!out) return
|
||||
if (!out.ok) log.warn("fff track query failed", { dir: row.dir, query: row.text, file, error: out.error })
|
||||
if (!out.ok) {
|
||||
yield* Effect.logWarning("fff track query failed", { dir: row.dir, query: row.text, file, error: out.error })
|
||||
}
|
||||
})
|
||||
|
||||
return Service.of({ files, tree, search, file, glob, open, warm, release })
|
||||
|
|
|
|||
|
|
@ -12,13 +12,11 @@ import { FSUtil } from "../fs-util"
|
|||
import { Git } from "../git"
|
||||
import { Location } from "../location"
|
||||
import { lazy } from "../util/lazy"
|
||||
import * as Log from "../util/log"
|
||||
import { Ignore } from "./ignore"
|
||||
import { Protected } from "./protected"
|
||||
|
||||
declare const OPENCODE_LIBC: string | undefined
|
||||
|
||||
const log = Log.create({ service: "file.watcher" })
|
||||
const SUBSCRIBE_TIMEOUT_MS = 10_000
|
||||
|
||||
export const Event = {
|
||||
|
|
@ -38,8 +36,7 @@ const watcher = lazy((): typeof import("@parcel/watcher") | undefined => {
|
|||
`@parcel/watcher-${process.platform}-${process.arch}${process.platform === "linux" ? `-${libc || "glibc"}` : ""}`,
|
||||
)
|
||||
return createWrapper(binding) as typeof import("@parcel/watcher")
|
||||
} catch (error) {
|
||||
log.error("failed to load watcher binding", { error })
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
})
|
||||
|
|
@ -71,14 +68,17 @@ export const layer = Layer.effect(
|
|||
const backend = getBackend()
|
||||
const location = yield* Location.Service
|
||||
if (!backend) {
|
||||
log.error("watcher backend not supported", { directory: location.directory, platform: process.platform })
|
||||
yield* Effect.logError("watcher backend not supported", {
|
||||
directory: location.directory,
|
||||
platform: process.platform,
|
||||
})
|
||||
return Service.of({})
|
||||
}
|
||||
|
||||
const w = watcher()
|
||||
if (!w) return Service.of({})
|
||||
|
||||
log.info("watcher backend", { directory: location.directory, platform: process.platform, backend })
|
||||
yield* Effect.logInfo("watcher backend", { directory: location.directory, platform: process.platform, backend })
|
||||
const events = yield* EventV2.Service
|
||||
const fs = yield* FSUtil.Service
|
||||
const git = yield* Git.Service
|
||||
|
|
@ -103,9 +103,8 @@ export const layer = Layer.effect(
|
|||
Effect.tap((subscription) => Effect.sync(() => subscriptions.push(subscription))),
|
||||
Effect.timeout(SUBSCRIBE_TIMEOUT_MS),
|
||||
Effect.catchCause((cause) => {
|
||||
log.error("failed to subscribe", { directory, cause: Cause.pretty(cause) })
|
||||
pending.then((subscription) => subscription.unsubscribe()).catch(() => {})
|
||||
return Effect.void
|
||||
return Effect.logError("failed to subscribe", { directory, cause: Cause.pretty(cause) })
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
|
@ -133,8 +132,9 @@ export const layer = Layer.effect(
|
|||
return Service.of({})
|
||||
}).pipe(
|
||||
Effect.catchCause((cause) => {
|
||||
log.error("failed to init watcher service", { cause: Cause.pretty(cause) })
|
||||
return Effect.succeed(Service.of({}))
|
||||
return Effect.logError("failed to init watcher service", { cause: Cause.pretty(cause) }).pipe(
|
||||
Effect.as(Service.of({})),
|
||||
)
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -228,7 +228,7 @@ export const layer = Layer.effect(
|
|||
}),
|
||||
).pipe(
|
||||
Effect.tapCause((cause) =>
|
||||
Effect.logError("Failed to fetch models.dev").pipe(Effect.annotateLogs("cause", cause)),
|
||||
Effect.logError("Failed to fetch models.dev", { cause: cause }),
|
||||
),
|
||||
Effect.ignore,
|
||||
)
|
||||
|
|
|
|||
21
packages/core/src/observability.ts
Normal file
21
packages/core/src/observability.ts
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
export * as Observability from "./observability"
|
||||
|
||||
import { NodeFileSystem } from "@effect/platform-node"
|
||||
import { Effect, Layer, Logger, References } from "effect"
|
||||
import { FetchHttpClient } from "effect/unstable/http"
|
||||
import { OtlpSerialization } from "effect/unstable/observability"
|
||||
import { Logging } from "./observability/logging"
|
||||
import { Otlp } from "./observability/otlp"
|
||||
|
||||
export const layer = Layer.unwrap(
|
||||
Effect.gen(function* () {
|
||||
const logs = Logger.layer([...Logging.loggers(), ...Otlp.loggers()], { mergeWithExisting: false }).pipe(
|
||||
Layer.provide(NodeFileSystem.layer),
|
||||
Layer.provide(OtlpSerialization.layerJson),
|
||||
Layer.provide(FetchHttpClient.layer),
|
||||
Layer.orDie,
|
||||
Layer.merge(Layer.succeed(References.MinimumLogLevel, Logging.minimumLogLevel())),
|
||||
)
|
||||
return Layer.merge(logs, yield* Effect.promise(Otlp.tracingLayer))
|
||||
}),
|
||||
)
|
||||
66
packages/core/src/observability/logging.ts
Normal file
66
packages/core/src/observability/logging.ts
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
import { Formatter, Logger, type LogLevel } from "effect"
|
||||
import path from "path"
|
||||
import { Global } from "../global"
|
||||
import { runID } from "./shared"
|
||||
|
||||
function formatter(id: string = runID) {
|
||||
return Logger.map(Logger.formatStructured, (output) => {
|
||||
const messages = Array.isArray(output.message) ? output.message : [output.message]
|
||||
return [
|
||||
["timestamp", output.timestamp],
|
||||
["level", output.level],
|
||||
["run", id],
|
||||
...messages.flatMap((value) => (plain(value) ? flatten(value) : [["message", value] as const])),
|
||||
...(output.cause === undefined ? [] : [["cause", output.cause] as const]),
|
||||
...flatten(output.spans),
|
||||
...flatten(output.annotations),
|
||||
]
|
||||
.map(([key, value]) => `${key}=${format(value)}`)
|
||||
.join(" ")
|
||||
})
|
||||
}
|
||||
|
||||
function flatten(input: Record<string, unknown>, prefix = "", seen = new WeakSet<object>()): Array<readonly [string, unknown]> {
|
||||
if (seen.has(input)) return [[prefix, "[Circular]"]]
|
||||
seen.add(input)
|
||||
const entries = Object.entries(input)
|
||||
if (entries.length === 0 && prefix) return [[prefix, input]]
|
||||
return entries.flatMap(([key, value]) => {
|
||||
const path = prefix ? `${prefix}.${key}` : key
|
||||
return plain(value) ? flatten(value, path, seen) : [[path, value] as const]
|
||||
})
|
||||
}
|
||||
|
||||
function plain(input: unknown): input is Record<string, unknown> {
|
||||
if (input === null || typeof input !== "object" || Array.isArray(input)) return false
|
||||
const prototype = Object.getPrototypeOf(input)
|
||||
return prototype === Object.prototype || prototype === null
|
||||
}
|
||||
|
||||
function format(input: unknown) {
|
||||
const value = typeof input === "string" ? input : Formatter.format(input)
|
||||
return /^[^\s="\\]+$/.test(value) ? value : JSON.stringify(value)
|
||||
}
|
||||
|
||||
export function fileLogger(file = path.join(Global.Path.log, "opencode.log"), id: string = runID) {
|
||||
return Logger.toFile(formatter(id), file, { flag: "a", batchWindow: 0 })
|
||||
}
|
||||
|
||||
const stderrLogger = Logger.make((options) => process.stderr.write(formatter().log(options) + "\n"))
|
||||
|
||||
export function minimumLogLevel() {
|
||||
const value = process.env.OPENCODE_LOG_LEVEL?.toUpperCase()
|
||||
const levels = {
|
||||
DEBUG: "Debug",
|
||||
INFO: "Info",
|
||||
WARN: "Warn",
|
||||
ERROR: "Error",
|
||||
} as const satisfies Record<string, LogLevel.LogLevel>
|
||||
return value && value in levels ? levels[value as keyof typeof levels] : levels.INFO
|
||||
}
|
||||
|
||||
export function loggers() {
|
||||
return process.env.OPENCODE_PRINT_LOGS === "1" ? [fileLogger(), stderrLogger] : [fileLogger()]
|
||||
}
|
||||
|
||||
export * as Logging from "./logging"
|
||||
79
packages/core/src/observability/otlp.ts
Normal file
79
packages/core/src/observability/otlp.ts
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
import { Layer } from "effect"
|
||||
import { OtlpLogger } from "effect/unstable/observability"
|
||||
import { Flag } from "../flag/flag"
|
||||
import { InstallationChannel, InstallationVersion } from "../installation/version"
|
||||
import { runID } from "./shared"
|
||||
|
||||
const endpoint = Flag.OTEL_EXPORTER_OTLP_ENDPOINT
|
||||
|
||||
const headers = Flag.OTEL_EXPORTER_OTLP_HEADERS
|
||||
? Flag.OTEL_EXPORTER_OTLP_HEADERS.split(",").reduce(
|
||||
(acc, entry) => {
|
||||
const [key, ...value] = entry.split("=")
|
||||
acc[key] = value.join("=")
|
||||
return acc
|
||||
},
|
||||
{} as Record<string, string>,
|
||||
)
|
||||
: undefined
|
||||
|
||||
function resourceAttributes() {
|
||||
const value = process.env.OTEL_RESOURCE_ATTRIBUTES
|
||||
if (!value) return {}
|
||||
try {
|
||||
return Object.fromEntries(
|
||||
value.split(",").map((entry) => {
|
||||
const index = entry.indexOf("=")
|
||||
if (index < 1) throw new Error("Invalid OTEL_RESOURCE_ATTRIBUTES entry")
|
||||
return [decodeURIComponent(entry.slice(0, index)), decodeURIComponent(entry.slice(index + 1))]
|
||||
}),
|
||||
)
|
||||
} catch {
|
||||
return {}
|
||||
}
|
||||
}
|
||||
|
||||
export function resource(): { serviceName: string; serviceVersion: string; attributes: Record<string, string> } {
|
||||
return {
|
||||
serviceName: "opencode",
|
||||
serviceVersion: InstallationVersion,
|
||||
attributes: {
|
||||
...resourceAttributes(),
|
||||
"deployment.environment.name": InstallationChannel,
|
||||
"opencode.client": Flag.OPENCODE_CLIENT,
|
||||
"opencode.run": runID,
|
||||
"service.instance.id": runID,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
export function loggers() {
|
||||
if (!endpoint) return []
|
||||
return [OtlpLogger.make({ url: `${endpoint}/v1/logs`, resource: resource(), headers })]
|
||||
}
|
||||
|
||||
export async function tracingLayer() {
|
||||
if (!endpoint) return Layer.empty
|
||||
const NodeSdk = await import("@effect/opentelemetry/NodeSdk")
|
||||
const OTLP = await import("@opentelemetry/exporter-trace-otlp-http")
|
||||
const SdkBase = await import("@opentelemetry/sdk-trace-base")
|
||||
const { AsyncLocalStorageContextManager } = await import("@opentelemetry/context-async-hooks")
|
||||
const { context } = await import("@opentelemetry/api")
|
||||
|
||||
// The Effect Node SDK does not register a global context manager, but the AI SDK uses it to parent spans.
|
||||
const manager = new AsyncLocalStorageContextManager()
|
||||
manager.enable()
|
||||
context.setGlobalContextManager(manager)
|
||||
|
||||
return NodeSdk.layer(() => ({
|
||||
resource: resource(),
|
||||
spanProcessor: new SdkBase.BatchSpanProcessor(
|
||||
new OTLP.OTLPTraceExporter({
|
||||
url: `${endpoint}/v1/traces`,
|
||||
headers,
|
||||
}),
|
||||
),
|
||||
}))
|
||||
}
|
||||
|
||||
export * as Otlp from "./otlp"
|
||||
1
packages/core/src/observability/shared.ts
Normal file
1
packages/core/src/observability/shared.ts
Normal file
|
|
@ -0,0 +1 @@
|
|||
export const runID = crypto.randomUUID().slice(0, 8)
|
||||
|
|
@ -103,9 +103,7 @@ export const layer = Layer.effect(
|
|||
(materializer) =>
|
||||
materializer.run.pipe(
|
||||
Effect.catchCause((cause) =>
|
||||
Effect.logWarning("failed to materialize project reference").pipe(
|
||||
Effect.annotateLogs({ name: materializer.name, repository: materializer.repository, cause }),
|
||||
),
|
||||
Effect.logWarning("failed to materialize project reference", { name: materializer.name, repository: materializer.repository, cause }),
|
||||
),
|
||||
),
|
||||
{ concurrency: 4, discard: true },
|
||||
|
|
|
|||
|
|
@ -7,9 +7,7 @@ import { Location } from "./location"
|
|||
import { NonNegativeInt, PositiveInt } from "./schema"
|
||||
import { PtyID } from "./pty/schema"
|
||||
import { lazy } from "./util/lazy"
|
||||
import * as Log from "./util/log"
|
||||
|
||||
const log = Log.create({ service: "pty" })
|
||||
const BUFFER_LIMIT = 1024 * 1024 * 2
|
||||
const BUFFER_CHUNK = 64 * 1024
|
||||
const encoder = new TextEncoder()
|
||||
|
|
@ -158,7 +156,7 @@ export const layer = Layer.effect(
|
|||
const session = sessions.get(id)
|
||||
if (!session) return false
|
||||
sessions.delete(id)
|
||||
log.info("removing session", { id })
|
||||
yield* Effect.logInfo("removing session", { id })
|
||||
teardown(session)
|
||||
yield* events.publish(Event.Deleted, { id: session.info.id })
|
||||
return true
|
||||
|
|
@ -179,7 +177,7 @@ export const layer = Layer.effect(
|
|||
|
||||
const create = Effect.fn("Pty.create")(function* (input: PreparedCreate) {
|
||||
const id = PtyID.ascending()
|
||||
log.info("creating session", { id, cmd: input.command, args: input.args, cwd: input.cwd })
|
||||
yield* Effect.logInfo("creating session", { id, cmd: input.command, args: input.args, cwd: input.cwd })
|
||||
const { spawn } = yield* Effect.promise(() => pty())
|
||||
const proc = yield* Effect.sync(() =>
|
||||
spawn(input.command, input.args, {
|
||||
|
|
@ -231,7 +229,7 @@ export const layer = Layer.effect(
|
|||
if (session.info.status === "exited") return
|
||||
runFork(
|
||||
Effect.gen(function* () {
|
||||
log.info("session exited", { id, exitCode })
|
||||
yield* Effect.logInfo("session exited", { id, exitCode })
|
||||
session.info.status = "exited"
|
||||
yield* events.publish(Event.Exited, { id, exitCode })
|
||||
yield* removeSession(id)
|
||||
|
|
@ -263,7 +261,7 @@ export const layer = Layer.effect(
|
|||
|
||||
const connect = Effect.fn("Pty.connect")(function* (id: PtyID, ws: Socket, cursor?: number) {
|
||||
const session = yield* requireSession(id).pipe(Effect.tapError(() => Effect.sync(() => ws.close())))
|
||||
log.info("client connected to session", { id, directory: location.directory })
|
||||
yield* Effect.logInfo("client connected to session", { id, directory: location.directory })
|
||||
const sub = sock(ws)
|
||||
session.subscribers.delete(sub)
|
||||
session.subscribers.set(sub, ws)
|
||||
|
|
@ -299,7 +297,6 @@ export const layer = Layer.effect(
|
|||
session.process.write(typeof message === "string" ? message : new TextDecoder().decode(message))
|
||||
},
|
||||
onClose: () => {
|
||||
log.info("client disconnected from session", { id })
|
||||
cleanup()
|
||||
},
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,4 +5,4 @@ export const logFailure = (
|
|||
message: "Failed to drain Session" | "Failed to wake Session",
|
||||
sessionID: SessionSchema.ID,
|
||||
cause: Cause.Cause<unknown>,
|
||||
) => Effect.logError(message, cause).pipe(Effect.annotateLogs("sessionID", sessionID))
|
||||
) => Effect.logError(message, cause).pipe(Effect.annotateLogs({ sessionID }))
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } fr
|
|||
import { FSUtil } from "../fs-util"
|
||||
import { Global } from "../global"
|
||||
import { AbsolutePath } from "../schema"
|
||||
import * as Log from "../util/log"
|
||||
|
||||
const skillConcurrency = 4
|
||||
const fileConcurrency = 8
|
||||
|
|
@ -71,7 +70,6 @@ export const layer = Layer.effect(
|
|||
Effect.gen(function* () {
|
||||
const fs = yield* FSUtil.Service
|
||||
const global = yield* Global.Service
|
||||
const log = Log.create({ service: "skill-discovery" })
|
||||
const http = (yield* HttpClient.HttpClient).pipe(
|
||||
HttpClient.retryTransient({
|
||||
retryOn: "errors-and-responses",
|
||||
|
|
@ -87,7 +85,7 @@ export const layer = Layer.effect(
|
|||
http.execute,
|
||||
Effect.flatMap((response) => response.arrayBuffer),
|
||||
Effect.flatMap((body) => fs.writeWithDirs(destination, new Uint8Array(body))),
|
||||
Effect.catch((error) => Effect.sync(() => log.error("failed to download skill file", { url, error }))),
|
||||
Effect.catch((error) => Effect.logError("failed to download skill file", { url, error })),
|
||||
)
|
||||
})
|
||||
|
||||
|
|
@ -100,10 +98,9 @@ export const layer = Layer.effect(
|
|||
HttpClientRequest.acceptJson,
|
||||
http.execute,
|
||||
Effect.flatMap(HttpClientResponse.schemaBodyJson(Index)),
|
||||
Effect.catch((error) => {
|
||||
log.error("failed to fetch skill index", { url: index, error })
|
||||
return Effect.succeed(undefined)
|
||||
}),
|
||||
Effect.catch((error) =>
|
||||
Effect.logError("failed to fetch skill index", { url: index, error }).pipe(Effect.as(undefined)),
|
||||
),
|
||||
)
|
||||
if (!data) return []
|
||||
|
||||
|
|
@ -111,17 +108,14 @@ export const layer = Layer.effect(
|
|||
return yield* Effect.forEach(
|
||||
data.skills.flatMap((skill) => {
|
||||
if (!isSafeSegment(skill.name)) {
|
||||
log.warn("skill entry has unsafe name", { url: index, skill: skill.name })
|
||||
return []
|
||||
}
|
||||
if (!skill.files.includes("SKILL.md") && !skill.files.includes(`${skill.name}.md`)) {
|
||||
log.warn("skill entry missing Markdown definition", { url: index, skill: skill.name })
|
||||
return []
|
||||
}
|
||||
|
||||
const root = path.resolve(sourceRoot, skill.name)
|
||||
if (!FSUtil.contains(sourceRoot, root) || root === sourceRoot) {
|
||||
log.warn("skill entry escapes cache root", { url: index, skill: skill.name })
|
||||
return []
|
||||
}
|
||||
|
||||
|
|
@ -144,7 +138,6 @@ export const layer = Layer.effect(
|
|||
}
|
||||
})
|
||||
if (files.some((file) => file === undefined)) {
|
||||
log.warn("skill entry has unsafe file", { url: index, skill: skill.name })
|
||||
return []
|
||||
}
|
||||
return [{ skill, root, files: files as { url: string; destination: string }[] }]
|
||||
|
|
|
|||
|
|
@ -1,197 +0,0 @@
|
|||
export * as Log from "./log"
|
||||
|
||||
import path from "path"
|
||||
import fs from "fs/promises"
|
||||
import { createWriteStream } from "fs"
|
||||
import * as Global from "../global"
|
||||
import { Schema } from "effect"
|
||||
import { Glob } from "./glob"
|
||||
|
||||
export const Level = Schema.Literals(["DEBUG", "INFO", "WARN", "ERROR"]).annotate({
|
||||
identifier: "LogLevel",
|
||||
description: "Log level",
|
||||
})
|
||||
export type Level = Schema.Schema.Type<typeof Level>
|
||||
|
||||
const levelPriority: Record<Level, number> = {
|
||||
DEBUG: 0,
|
||||
INFO: 1,
|
||||
WARN: 2,
|
||||
ERROR: 3,
|
||||
}
|
||||
const keep = 10
|
||||
const initializedRunID = "OPENCODE_LOG_INITIALIZED_RUN_ID"
|
||||
|
||||
let level: Level = "INFO"
|
||||
|
||||
function shouldLog(input: Level): boolean {
|
||||
return levelPriority[input] >= levelPriority[level]
|
||||
}
|
||||
|
||||
export type Logger = {
|
||||
debug(message?: any, extra?: Record<string, any>): void
|
||||
info(message?: any, extra?: Record<string, any>): void
|
||||
error(message?: any, extra?: Record<string, any>): void
|
||||
warn(message?: any, extra?: Record<string, any>): void
|
||||
tag(key: string, value: string): Logger
|
||||
clone(): Logger
|
||||
time(
|
||||
message: string,
|
||||
extra?: Record<string, any>,
|
||||
): {
|
||||
stop(): void
|
||||
[Symbol.dispose](): void
|
||||
}
|
||||
}
|
||||
|
||||
const loggers = new Map<string, Logger>()
|
||||
|
||||
export const Default = create({ service: "default" })
|
||||
|
||||
export interface Options {
|
||||
print: boolean
|
||||
dev?: boolean
|
||||
level?: Level
|
||||
}
|
||||
|
||||
let logpath = ""
|
||||
export function file() {
|
||||
return logpath
|
||||
}
|
||||
export function getLevel(): Level {
|
||||
return level
|
||||
}
|
||||
let write = (msg: any) => {
|
||||
process.stderr.write(msg)
|
||||
return msg.length
|
||||
}
|
||||
|
||||
export async function init(options: Options) {
|
||||
if (options.level) level = options.level
|
||||
void cleanup(Global.Path.log)
|
||||
if (options.print) return
|
||||
logpath = path.join(
|
||||
Global.Path.log,
|
||||
options.dev ? "dev.log" : new Date().toISOString().split(".")[0].replace(/:/g, "") + ".log",
|
||||
)
|
||||
const runID = process.env.OPENCODE_RUN_ID
|
||||
const shouldTruncate = !options.dev || !runID || process.env[initializedRunID] !== runID
|
||||
if (shouldTruncate) await fs.truncate(logpath).catch(() => {})
|
||||
if (options.dev && runID) process.env[initializedRunID] = runID
|
||||
const stream = createWriteStream(logpath, { flags: "a" })
|
||||
write = async (msg: any) => {
|
||||
return new Promise((resolve, reject) => {
|
||||
stream.write(msg, (err) => {
|
||||
if (err) reject(err)
|
||||
else resolve(msg.length)
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async function cleanup(dir: string) {
|
||||
const files = (
|
||||
await Glob.scan("????-??-??T??????.log", {
|
||||
cwd: dir,
|
||||
absolute: false,
|
||||
include: "file",
|
||||
}).catch(() => [])
|
||||
)
|
||||
.filter((file) => path.basename(file) === file)
|
||||
.sort()
|
||||
if (files.length <= keep) return
|
||||
|
||||
const doomed = files.slice(0, -keep)
|
||||
await Promise.all(doomed.map((file) => fs.unlink(path.join(dir, file)).catch(() => {})))
|
||||
}
|
||||
|
||||
function formatError(error: Error, depth = 0): string {
|
||||
const result = error.message
|
||||
return error.cause instanceof Error && depth < 10
|
||||
? result + " Caused by: " + formatError(error.cause, depth + 1)
|
||||
: result
|
||||
}
|
||||
|
||||
let last = Date.now()
|
||||
export function create(tags?: Record<string, any>) {
|
||||
tags = tags || {}
|
||||
|
||||
const service = tags["service"]
|
||||
if (service && typeof service === "string") {
|
||||
const cached = loggers.get(service)
|
||||
if (cached) {
|
||||
return cached
|
||||
}
|
||||
}
|
||||
|
||||
function build(message: any, extra?: Record<string, any>) {
|
||||
const prefix = Object.entries({
|
||||
...tags,
|
||||
...extra,
|
||||
})
|
||||
.filter(([_, value]) => value !== undefined && value !== null)
|
||||
.map(([key, value]) => {
|
||||
const prefix = `${key}=`
|
||||
if (value instanceof Error) return prefix + formatError(value)
|
||||
if (typeof value === "object") return prefix + JSON.stringify(value)
|
||||
return prefix + value
|
||||
})
|
||||
.join(" ")
|
||||
const next = new Date()
|
||||
const diff = next.getTime() - last
|
||||
last = next.getTime()
|
||||
return [next.toISOString().split(".")[0], "+" + diff + "ms", prefix, message].filter(Boolean).join(" ") + "\n"
|
||||
}
|
||||
const result: Logger = {
|
||||
debug(message?: any, extra?: Record<string, any>) {
|
||||
if (shouldLog("DEBUG")) {
|
||||
write("DEBUG " + build(message, extra))
|
||||
}
|
||||
},
|
||||
info(message?: any, extra?: Record<string, any>) {
|
||||
if (shouldLog("INFO")) {
|
||||
write("INFO " + build(message, extra))
|
||||
}
|
||||
},
|
||||
error(message?: any, extra?: Record<string, any>) {
|
||||
if (shouldLog("ERROR")) {
|
||||
write("ERROR " + build(message, extra))
|
||||
}
|
||||
},
|
||||
warn(message?: any, extra?: Record<string, any>) {
|
||||
if (shouldLog("WARN")) {
|
||||
write("WARN " + build(message, extra))
|
||||
}
|
||||
},
|
||||
tag(key: string, value: string) {
|
||||
if (tags) tags[key] = value
|
||||
return result
|
||||
},
|
||||
clone() {
|
||||
return create({ ...tags })
|
||||
},
|
||||
time(message: string, extra?: Record<string, any>) {
|
||||
const now = Date.now()
|
||||
result.info(message, { status: "started", ...extra })
|
||||
function stop() {
|
||||
result.info(message, {
|
||||
status: "completed",
|
||||
duration: Date.now() - now,
|
||||
...extra,
|
||||
})
|
||||
}
|
||||
return {
|
||||
stop,
|
||||
[Symbol.dispose]() {
|
||||
stop()
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
if (service && typeof service === "string") {
|
||||
loggers.set(service, result)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
|
@ -1,24 +0,0 @@
|
|||
export const OPENCODE_RUN_ID = "OPENCODE_RUN_ID"
|
||||
export const OPENCODE_PROCESS_ROLE = "OPENCODE_PROCESS_ROLE"
|
||||
|
||||
export function ensureRunID() {
|
||||
return (process.env[OPENCODE_RUN_ID] ??= crypto.randomUUID())
|
||||
}
|
||||
|
||||
export function ensureProcessRole(fallback: "main" | "worker") {
|
||||
return (process.env[OPENCODE_PROCESS_ROLE] ??= fallback)
|
||||
}
|
||||
|
||||
export function ensureProcessMetadata(fallback: "main" | "worker") {
|
||||
return {
|
||||
runID: ensureRunID(),
|
||||
processRole: ensureProcessRole(fallback),
|
||||
}
|
||||
}
|
||||
|
||||
export function sanitizedProcessEnv(overrides?: Record<string, string>) {
|
||||
const env = Object.fromEntries(
|
||||
Object.entries(process.env).filter((entry): entry is [string, string] => entry[1] !== undefined),
|
||||
)
|
||||
return overrides ? Object.assign(env, overrides) : env
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue