Build replacement appenders and thresholds before installing them and advancing the local version. Release partial replacements and old caches outside the configuration lock. In order to make the `ensureFreshness` version check as cheap as posible and only incur overhead when we actually need to update the local cache, isolate the refresh logic in a separate scope to prevent initialization and cleanup of temporary variables when the version has not changed Add regression coverage for repeated clone failures, cache preservation, cleanup, and retry without another configuration change. Update documentation to reflect the remaining freshness issue.
225 lines
7.9 KiB
Nim
225 lines
7.9 KiB
Nim
import std/[os, osproc, strutils, tempfiles, unittest]
|
|
import ../src/namespaced_logging
|
|
import ./ownership
|
|
|
|
type
|
|
CloneControl = object
|
|
failNext: bool
|
|
FallibleAppender = ref object of LogAppender
|
|
control: ptr CloneControl
|
|
|
|
method clone(app: FallibleAppender): LogAppender {.gcsafe.} =
|
|
if app.control.failNext:
|
|
app.control.failNext = false
|
|
raise newException(ValueError, "injected clone failure")
|
|
FallibleAppender(control: app.control)
|
|
|
|
type
|
|
RefreshProbeObj = object of LogAppender
|
|
destroyed: ptr int
|
|
RefreshProbe = ref RefreshProbeObj
|
|
|
|
proc `=destroy`(app: RefreshProbeObj) =
|
|
if app.destroyed != nil: inc app.destroyed[]
|
|
`=destroy`(app.namespace)
|
|
|
|
method clone(app: RefreshProbe): LogAppender {.gcsafe.} =
|
|
RefreshProbe(namespace: app.namespace, destroyed: app.destroyed)
|
|
|
|
type
|
|
LifetimeProbeObj = object
|
|
destroyed: ptr int
|
|
LifetimeProbe = SharedLogState[LifetimeProbeObj]
|
|
|
|
proc `=destroy`(probe: LifetimeProbeObj) =
|
|
if probe.destroyed != nil:
|
|
inc probe.destroyed[]
|
|
|
|
proc ignoreMessage(state: var LifetimeProbe, msg: LogMessage) {.gcsafe, noconv.} =
|
|
discard
|
|
|
|
proc watchedService(destroyed: ptr int, namespace = ""): LogService =
|
|
result = initLogService()
|
|
result.addAppender(initCustomLogAppender(
|
|
initSharedLogState(LifetimeProbeObj(destroyed: destroyed)), ignoreMessage,
|
|
namespace = namespace))
|
|
|
|
proc createAndReleaseService(destroyed: ptr int) =
|
|
var service = watchedService(destroyed)
|
|
service.setThreshold("worker", lvlInfo)
|
|
# All owning handles leave scope here. No thread or cycle is involved.
|
|
|
|
suite "ORC service handle lifetime":
|
|
test "copy, self-assignment, replacement, and move retain exactly one lifetime":
|
|
var firstDestroyed, replacedDestroyed = 0
|
|
block:
|
|
var first = watchedService(addr firstDestroyed)
|
|
var other = watchedService(addr replacedDestroyed)
|
|
other = first
|
|
check replacedDestroyed == 1
|
|
first = first
|
|
first = LogService()
|
|
check firstDestroyed == 0
|
|
|
|
var moved = move(other)
|
|
expect ValueError:
|
|
discard other.threadLocalRef()
|
|
moved.setThreshold("worker", lvlInfo)
|
|
let local = moved.threadLocalRef()
|
|
moved = LogService()
|
|
check firstDestroyed == 0
|
|
check local.getLogger("worker").isEnabled(lvlInfo)
|
|
check firstDestroyed == 1
|
|
check replacedDestroyed == 1
|
|
|
|
test "handles embedded in copied containers retain the service":
|
|
var destroyed = 0
|
|
block:
|
|
var source = watchedService(addr destroyed)
|
|
var handles = @[source, source]
|
|
var copied = handles
|
|
source = LogService()
|
|
handles.setLen(0)
|
|
check destroyed == 0
|
|
copied[0].setThreshold("worker", lvlWarn)
|
|
let local = copied[1].threadLocalRef()
|
|
check not local.getLogger("worker").isEnabled(lvlInfo)
|
|
copied.setLen(0)
|
|
check destroyed == 0
|
|
check destroyed == 1
|
|
|
|
test "an exported handle outlives the local service it came from":
|
|
var destroyed = 0
|
|
block:
|
|
var exported: LogService
|
|
block:
|
|
let local = watchedService(addr destroyed).threadLocalRef()
|
|
exported = local.service
|
|
check destroyed == 0
|
|
check exported.threadLocalRef().getLogger("worker").isEnabled(lvlInfo)
|
|
check destroyed == 1
|
|
|
|
test "failed local construction releases its shared owner and partial cache":
|
|
var destroyed = 0
|
|
var control: CloneControl
|
|
block:
|
|
var service = watchedService(addr destroyed)
|
|
service.addAppender(FallibleAppender(control: addr control))
|
|
control.failNext = true
|
|
expect ValueError:
|
|
discard service.threadLocalRef()
|
|
check destroyed == 0
|
|
check destroyed == 1
|
|
|
|
suite "ORC ownership and configuration regressions":
|
|
test "isEnabled observes a namespace change before any log call":
|
|
var service = initLogService()
|
|
service.setThreshold("worker", lvlError)
|
|
let local = service.threadLocalRef()
|
|
let logger = local.getLogger("worker")
|
|
check not logger.isEnabled(lvlInfo)
|
|
|
|
# Simulate a change from another service handle without introducing a
|
|
# reference-count race into this deterministic configuration test.
|
|
service.setThreshold("worker", lvlInfo)
|
|
check logger.isEnabled(lvlInfo)
|
|
|
|
test "a failed refresh is retried without another configuration change":
|
|
var control: CloneControl
|
|
var service = initLogService()
|
|
service.addAppender(FallibleAppender(control: addr control))
|
|
let local = service.threadLocalRef()
|
|
discard local.getLogger("worker")
|
|
service.setRootThreshold(lvlInfo)
|
|
|
|
control.failNext = true
|
|
expect ValueError:
|
|
local.ensureFreshness()
|
|
|
|
# clone now succeeds; the failed refresh must not have marked the
|
|
# incomplete local configuration as current.
|
|
local.ensureFreshness()
|
|
check local.appenders().len == 1
|
|
|
|
test "failed refresh preserves the old cache and releases partial replacements":
|
|
var oldStateDestroyed, discardedClones: int
|
|
var control: CloneControl
|
|
var service = watchedService(addr oldStateDestroyed, "old")
|
|
service.setThreshold("worker", lvlError)
|
|
let local = service.threadLocalRef()
|
|
let logger = local.getLogger("worker")
|
|
|
|
# Only the local cache retains the old appender's state after this change.
|
|
service.clearAppenders()
|
|
service.addAppender(RefreshProbe(namespace: "new", destroyed: addr discardedClones))
|
|
service.addAppender(FallibleAppender(control: addr control))
|
|
service.setThreshold("worker", lvlInfo)
|
|
let initiallyDestroyed = discardedClones
|
|
|
|
for attempt in 1 .. 2:
|
|
control.failNext = true
|
|
expect ValueError:
|
|
local.ensureFreshness()
|
|
check oldStateDestroyed == 0
|
|
check discardedClones == initiallyDestroyed + attempt
|
|
block:
|
|
let cached = local.appenders()
|
|
require cached.len == 1
|
|
check cached[0].namespace == "old"
|
|
check not logger.isEnabled(lvlInfo)
|
|
|
|
# No further configuration update is needed to retry and install both
|
|
# the new appenders and the new namespace thresholds.
|
|
local.ensureFreshness()
|
|
check oldStateDestroyed == 1
|
|
let cached = local.appenders()
|
|
require cached.len == 2
|
|
check cached[0].namespace == "new"
|
|
check logger.isEnabled(lvlInfo)
|
|
|
|
test "releasing the service releases state owned by its appenders":
|
|
var destroyed = 0
|
|
createAndReleaseService(addr destroyed)
|
|
check destroyed == 1
|
|
|
|
suite "Background writer shutdown regressions":
|
|
for destination in ["file", "console", "stderr", "file-empty", "console-empty",
|
|
"files", "file-parallel", "console-parallel"]:
|
|
test destination & " writer drains accepted messages at normal process exit":
|
|
let directory = createTempDir("namespaced_logging_shutdown_", "")
|
|
let outputPath = directory / "messages.log"
|
|
defer: removeDir(directory)
|
|
|
|
let child = startProcess(parentDir(getAppFilename()) / "writer_child",
|
|
args = [destination, outputPath], options = {poParentStreams})
|
|
defer: close(child)
|
|
let exitCode = child.waitForExit(15_000)
|
|
if exitCode == -1:
|
|
child.kill()
|
|
discard child.waitForExit()
|
|
check exitCode == 0
|
|
require fileExists(outputPath)
|
|
|
|
if destination.endsWith("-parallel"):
|
|
let received = readFile(outputPath).splitLines()
|
|
require received.len == 10_001
|
|
var next = [0, 2_500, 5_000, 7_500]
|
|
for line in received[0 ..< 10_000]:
|
|
require line.startsWith("message ")
|
|
let number = parseInt(line[8 .. ^1])
|
|
require number in 0 ..< 10_000
|
|
let worker = number div 2_500
|
|
check number == next[worker]
|
|
inc next[worker]
|
|
check next == [2_500, 5_000, 7_500, 10_000]
|
|
else:
|
|
var expected: string
|
|
for i in 0 ..< 10_000:
|
|
if destination == "file-empty" and i mod 128 == 0:
|
|
expected.add("\n")
|
|
expected.add("message " & $i & "\n")
|
|
check readFile(outputPath) == expected
|
|
if destination == "files":
|
|
require fileExists(outputPath & ".second")
|
|
check readFile(outputPath & ".second") == expected
|