fix: preserve demo records across Collector restarts and config reloads

Signed-off-by: 千乘妍 (Xiaoyaner) <xiaoyaner0201@users.noreply.github.com>
This commit is contained in:
千乘妍 (Xiaoyaner) 2026-11-03 16:47:34 +08:00
parent 5f8523160b
commit d1fa71c9a9
6 changed files with 39 additions and 6 deletions

View file

@ -10,7 +10,8 @@ processors:
spike_limit_mib: 16
exporters:
file:
path: /business-logs.json
path: /data/business-logs.json
append: true
flush_interval: 100ms
service:
pipelines:

View file

@ -11,8 +11,11 @@ services:
- "127.0.0.1:${SKILLHUB_BUSINESS_LOGS_DEMO_PORT:-4318}:4318"
volumes:
- ./collector.yaml:/etc/otelcol-contrib/config.yaml:ro
- business_logs_demo_data:/data
networks:
- business-logs
volumes:
business_logs_demo_data:
networks:
business-logs:
external: true

View file

@ -33,7 +33,7 @@ case "$COMMAND" in
up)
case "$BACKEND" in host|container) ;; *) echo 'BUSINESS_LOGS_BACKEND must be host or container' >&2; exit 1 ;; esac
docker network inspect "$SKILLHUB_BUSINESS_LOGS_NETWORK" >/dev/null 2>&1 || docker network create "$SKILLHUB_BUSINESS_LOGS_NETWORK" >/dev/null
compose up -d
compose up -d --force-recreate
if [ "$BACKEND" = host ]; then
mkdir -p "$STATE_DIR"
printf '%s\n' "$SKILLHUB_BUSINESS_LOGS_DEMO_PORT" >"$STATE_DIR/host-port"
@ -48,9 +48,9 @@ case "$COMMAND" in
status) compose ps ;;
records|verify)
command -v jq >/dev/null || { echo 'Install jq to query demo records' >&2; exit 1; }
scratch=$(mktemp -d)
scratch=$(mktemp -d "${TMPDIR:-/tmp}/skillhub-business-logs.XXXXXX")
trap 'rm -rf "$scratch"' EXIT HUP INT TERM
compose cp business-logs-collector:/business-logs.json "$scratch/records.jsonl" >/dev/null
compose cp business-logs-collector:/data/business-logs.json "$scratch/records.jsonl" >/dev/null
if [ "$COMMAND" = records ]; then
cat "$scratch/records.jsonl"
else

View file

@ -1,7 +1,7 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT=$(cd "$(dirname "$0")/../.." && pwd)
TMP=$(mktemp -d)
TMP=$(mktemp -d "${TMPDIR:-/tmp}/skillhub-business-logs-test.XXXXXX")
trap 'rm -rf "$TMP"' EXIT
mkdir -p "$TMP/bin" "$TMP/home"
cat >"$TMP/bin/docker" <<'SH'

View file

@ -90,6 +90,8 @@ class BusinessLogsDeploymentTest(unittest.TestCase):
demo = json.loads(result.stdout)["services"]["business-logs-collector"]
self.assertEqual(demo["ports"][0]["host_ip"], "127.0.0.1")
self.assertNotIn("container_name", demo)
self.assertTrue(any(volume["target"] == "/data" and volume["type"] == "volume" for volume in demo["volumes"]))
self.assertIn("append: true", (ROOT / "docs/examples/business-event-logs/collector.yaml").read_text())
@unittest.skipUnless(shutil.which("helm"), "helm executable required")
def test_helm_defaults_secret_reference_and_invalid_enabled_config(self):

View file

@ -41,6 +41,7 @@ class BusinessLogsCollectorTest {
.withCreateContainerCmdModifier(cmd -> {
cmd.withName("skillhub-1009-collector-integration");
cmd.withUser("0:0");
cmd.withVolumes(new com.github.dockerjava.api.model.Volume("/data"));
cmd.getHostConfig().withMemory(128L * 1024 * 1024).withNanoCPUs(500_000_000L);
})
.waitingFor(Wait.forLogMessage(".*Everything is ready.*", 1));
@ -105,6 +106,32 @@ class BusinessLogsCollectorTest {
if (evidence != null) { Files.writeString(Path.of(evidence, "collector-received.jsonl"), raw); }
System.out.println("BUSINESS_LOGS real Collector 0.111.0: queried 27 records / 9 event names / 27 unique uids; modes none,otel-sdk(sample=0),external-agent; 128 MiB, 0.5 CPU");
String endpoint = "http://" + COLLECTOR.getHost() + ":" + COLLECTOR.getMappedPort(4318) + "/v1/logs";
// A real stop/start must not truncate the local trial's records. Wait for
// the restarted HTTP pipeline before reading, rather than racing file open.
int previousPort = COLLECTOR.getMappedPort(4318);
COLLECTOR.getDockerClient().stopContainerCmd(COLLECTOR.getContainerId()).exec();
COLLECTOR.getDockerClient().startContainerCmd(COLLECTOR.getContainerId()).exec();
// Docker may allocate another dynamic host port on restart. Inspect the
// actual binding rather than using GenericContainer's cached initial info.
String restartedPort = COLLECTOR.getDockerClient().inspectContainerCmd(COLLECTOR.getContainerId()).exec()
.getNetworkSettings().getPorts().getBindings()
.get(com.github.dockerjava.api.model.ExposedPort.tcp(4318))[0].getHostPortSpec();
endpoint = "http://" + COLLECTOR.getHost() + ":" + restartedPort + "/v1/logs";
System.out.printf("BUSINESS_LOGS Collector restart mapped port: %d -> %s%n", previousPort, restartedPort);
var client = java.net.http.HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(1)).build();
var probe = java.net.http.HttpRequest.newBuilder(java.net.URI.create(endpoint))
.timeout(Duration.ofSeconds(1)).header("Content-Type", "application/x-protobuf")
.POST(java.net.http.HttpRequest.BodyPublishers.noBody()).build();
try {
org.awaitility.Awaitility.await().atMost(Duration.ofSeconds(10)).ignoreExceptions().untilAsserted(() ->
assertThat(client.send(probe, java.net.http.HttpResponse.BodyHandlers.discarding()).statusCode()).isEqualTo(200));
} finally {
client.close();
}
assertThat(records(awaitFile(27)).stream()
.map(record -> attributes(record).get("log.record.uid").asText())
.collect(java.util.stream.Collectors.toSet())).isEqualTo(uids);
System.out.println("BUSINESS_LOGS real Collector stop/start: all 27 original record UIDs retained");
COLLECTOR.getDockerClient().pauseContainerCmd(COLLECTOR.getContainerId()).exec();
try {
outageRunner(endpoint).run(context -> {
@ -151,7 +178,7 @@ class BusinessLogsCollectorTest {
Exception lastError = null;
do {
try {
raw = COLLECTOR.copyFileFromContainer("/business-logs.json", stream -> new String(stream.readAllBytes(), java.nio.charset.StandardCharsets.UTF_8));
raw = COLLECTOR.copyFileFromContainer("/data/business-logs.json", stream -> new String(stream.readAllBytes(), java.nio.charset.StandardCharsets.UTF_8));
if (records(raw).size() >= expected) { return raw; }
} catch (Exception ex) { lastError = ex; }
TimeUnit.MILLISECONDS.sleep(100);