diff --git a/docs/examples/business-event-logs/collector.yaml b/docs/examples/business-event-logs/collector.yaml
new file mode 100644
index 000000000..1ce8d3783
--- /dev/null
+++ b/docs/examples/business-event-logs/collector.yaml
@@ -0,0 +1,20 @@
+receivers:
+ otlp:
+ protocols:
+ http:
+ endpoint: 0.0.0.0:4318
+processors:
+ memory_limiter:
+ check_interval: 1s
+ limit_mib: 64
+ spike_limit_mib: 16
+exporters:
+ file:
+ path: /business-logs.json
+ flush_interval: 100ms
+service:
+ pipelines:
+ logs:
+ receivers: [otlp]
+ processors: [memory_limiter]
+ exporters: [file]
diff --git a/docs/examples/business-event-logs/compose.yaml b/docs/examples/business-event-logs/compose.yaml
new file mode 100644
index 000000000..e7428c8a5
--- /dev/null
+++ b/docs/examples/business-event-logs/compose.yaml
@@ -0,0 +1,14 @@
+name: skillhub-1009-business-logs
+services:
+ collector:
+ image: otel/opentelemetry-collector-contrib:0.111.0
+ container_name: skillhub-1009-collector
+ command: ["--config=/etc/otelcol-contrib/config.yaml"]
+ user: "0:0"
+ mem_limit: 128m
+ cpus: 0.5
+ ports:
+ - "127.0.0.1:4318:4318"
+ volumes:
+ - ./collector.yaml:/etc/otelcol-contrib/config.yaml:ro
+
diff --git a/server/skillhub-app/pom.xml b/server/skillhub-app/pom.xml
index 454bb8aa8..8db07a16c 100644
--- a/server/skillhub-app/pom.xml
+++ b/server/skillhub-app/pom.xml
@@ -59,6 +59,10 @@
io.opentelemetry
opentelemetry-exporter-otlp
+
+ io.opentelemetry
+ opentelemetry-sdk-logs
+
org.apache.skywalking
apm-toolkit-logback-1.x
diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessEventListener.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessEventListener.java
new file mode 100644
index 000000000..b2a1d1226
--- /dev/null
+++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessEventListener.java
@@ -0,0 +1,43 @@
+package com.iflytek.skillhub.observability.business;
+
+import com.iflytek.skillhub.domain.event.SkillBusinessEvent;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.context.Context;
+import org.springframework.context.event.EventListener;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+/** Synchronous O(1) snapshot capture; only SDK admission runs after commit, never network IO. */
+public final class BusinessEventListener {
+ private final BusinessEventMapper mapper;
+ private final BusinessLogsPipeline pipeline;
+ private final BusinessLogsMetrics metrics;
+
+ BusinessEventListener(BusinessEventMapper mapper, BusinessLogsPipeline pipeline, BusinessLogsMetrics metrics) {
+ this.mapper = mapper;
+ this.pipeline = pipeline;
+ this.metrics = metrics;
+ }
+
+ @EventListener
+ public void onEvent(SkillBusinessEvent event) {
+ try {
+ mapper.map(event).ifPresent(record -> {
+ SpanContext span = Span.current().getSpanContext();
+ // Retain only valid standard span identity, never baggage or arbitrary request/MDC data.
+ Context context = span.isValid() ? Context.root().with(Span.wrap(span)) : Context.root();
+ if (TransactionSynchronizationManager.isActualTransactionActive()
+ && TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+ @Override public void afterCommit() { pipeline.emit(record, context); }
+ });
+ } else {
+ pipeline.emit(record, context);
+ }
+ });
+ } catch (RuntimeException ex) {
+ metrics.error();
+ }
+ }
+}
diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessEventMapper.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessEventMapper.java
new file mode 100644
index 000000000..50fb43ef9
--- /dev/null
+++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessEventMapper.java
@@ -0,0 +1,84 @@
+package com.iflytek.skillhub.observability.business;
+
+import com.iflytek.skillhub.domain.event.*;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+/** Sole data boundary. Never reflect/serialize an event or look up an entity here. */
+public final class BusinessEventMapper {
+ public static final Set EVENT_NAMES = Set.of("skill.publication.effective", "skill.download.prepared",
+ "skill.review.approved", "skill.review.rejected", "skill.promotion.approved", "skill.promotion.rejected",
+ "skill.status.changed", "skill.visibility.changed", "skill.version.yanked");
+ private static final Set SOURCES = Set.of("auto_publish", "private_confirm", "review", "promotion",
+ "download", "governance", "visibility");
+ private final Set events;
+ private final Set namespaces;
+
+ BusinessEventMapper(BusinessLogsProperties properties) {
+ events = Set.copyOf(properties.getEventAllowlist());
+ namespaces = Set.copyOf(properties.getNamespaceAllowlist());
+ }
+
+ public Optional map(SkillBusinessEvent event) {
+ BusinessEventSnapshot snapshot = event.snapshot();
+ if (snapshot == null || snapshot.namespaceId() == null || snapshot.visibility() == null
+ || snapshot.occurredAt() == null || event.skillId() == null
+ || snapshot.namespaceId() <= 0 || event.skillId() <= 0
+ || snapshot.source() == null || snapshot.actorType() == null
+ || !SOURCES.contains(snapshot.source())
+ || !Set.of("user", "system").contains(snapshot.actorType())
+ || (!namespaces.isEmpty() && !namespaces.contains(snapshot.namespaceId()))) {
+ return Optional.empty();
+ }
+ AttributesBuilder attributes = Attributes.builder();
+ String name;
+ switch (event) {
+ case SkillPublishedEvent e -> { name = "skill.publication.effective"; id(attributes, "version", e.versionId()); }
+ case SkillDownloadedEvent e -> { name = "skill.download.prepared"; id(attributes, "version", e.versionId()); }
+ case ReviewApprovedEvent e -> { name = "skill.review.approved"; id(attributes, "version", e.versionId()); id(attributes, "review", e.reviewId()); }
+ case ReviewRejectedEvent e -> { name = "skill.review.rejected"; id(attributes, "version", e.versionId()); id(attributes, "review", e.reviewId()); }
+ case PromotionApprovedEvent e -> { name = "skill.promotion.approved"; id(attributes, "promotion", e.promotionId()); }
+ case PromotionRejectedEvent e -> { name = "skill.promotion.rejected"; id(attributes, "promotion", e.promotionId()); }
+ case SkillStatusChangedEvent e -> {
+ if (e.oldStatus() == null || e.newStatus() == null || e.oldStatus() == e.newStatus()) { return Optional.empty(); }
+ name = "skill.status.changed";
+ attributes.put("skillhub.status.old", e.oldStatus().name()).put("skillhub.status.new", e.newStatus().name());
+ }
+ case SkillVisibilityChangedEvent e -> {
+ if (e.previousVisibility() == null || e.visibility() == null || e.previousVisibility() == e.visibility()) { return Optional.empty(); }
+ name = "skill.visibility.changed";
+ attributes.put("skillhub.visibility.old", e.previousVisibility().name()).put("skillhub.visibility.new", e.visibility().name());
+ }
+ case SkillVersionYankedEvent e -> {
+ if (!e.wasPublished()) { return Optional.empty(); }
+ name = "skill.version.yanked";
+ id(attributes, "version", e.versionId());
+ }
+ default -> { return Optional.empty(); }
+ }
+ if (!events.contains(name)) { return Optional.empty(); }
+ String uid = UUID.randomUUID().toString();
+ attributes.put("skillhub.event.name", name).put("skillhub.skill.id", event.skillId())
+ .put("skillhub.namespace.id", snapshot.namespaceId()).put("skillhub.visibility", snapshot.visibility().name())
+ .put("skillhub.actor.type", snapshot.actorType()).put("skillhub.source", snapshot.source())
+ .put("log.record.uid", uid);
+ if ("user".equals(snapshot.actorType()) && snapshot.actorId() != null && !snapshot.actorId().isBlank()) {
+ attributes.put("user.id", bounded(snapshot.actorId()));
+ }
+ return Optional.of(new MappedRecord(snapshot.occurredAt(), attributes.build(), uid));
+ }
+
+ static String bounded(String value) {
+ return value.length() <= 256 ? value : value.substring(0, 256);
+ }
+
+ private static void id(AttributesBuilder attributes, String kind, Long value) {
+ if (value != null) { attributes.put("skillhub." + kind + ".id", value); }
+ }
+
+ public record MappedRecord(Instant timestamp, Attributes attributes, String uid) {}
+}
diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsConfiguration.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsConfiguration.java
new file mode 100644
index 000000000..2e39c7f2c
--- /dev/null
+++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsConfiguration.java
@@ -0,0 +1,49 @@
+package com.iflytek.skillhub.observability.business;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporter;
+import io.opentelemetry.sdk.common.export.RetryPolicy;
+import io.opentelemetry.sdk.resources.Resource;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+
+/** Separate signal lifecycle; no global OTel registration or tracing beans. */
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties(BusinessLogsProperties.class)
+@ConditionalOnProperty(prefix = "skillhub.observability.business-logs", name = "enabled", havingValue = "true")
+public class BusinessLogsConfiguration {
+ @Bean
+ BusinessLogsMetrics businessLogsMetrics(MeterRegistry registry) { return new BusinessLogsMetrics(registry); }
+
+ @Bean(destroyMethod = "close")
+ BusinessLogsPipeline businessLogsPipeline(BusinessLogsProperties properties, BusinessLogsMetrics metrics,
+ Environment environment) {
+ properties.validate();
+ var builder = OtlpHttpLogRecordExporter.builder().setEndpoint(properties.getEndpoint())
+ .setTimeout(properties.getTimeout()).setCompression(properties.getCompression())
+ .setMeterProvider(MeterProvider.noop());
+ properties.parsedHeaders().forEach(builder::addHeader);
+ if (properties.getRetryMaxAttempts() > 1) {
+ builder.setRetryPolicy(RetryPolicy.builder().setMaxAttempts(properties.getRetryMaxAttempts())
+ .setInitialBackoff(properties.getRetryInitialBackoff()).setMaxBackoff(properties.getRetryMaxBackoff())
+ .setBackoffMultiplier(2).build());
+ }
+ Resource resource = Resource.create(Attributes.builder()
+ .put("service.name", BusinessEventMapper.bounded(environment.getProperty("spring.application.name", "skillhub")))
+ .put("service.version", BusinessEventMapper.bounded(environment.getProperty("skillhub.observability.service-version", "unknown")))
+ .put("deployment.environment", BusinessEventMapper.bounded(environment.getProperty("skillhub.observability.service-environment", "local")))
+ .build());
+ return new BusinessLogsPipeline(properties, resource, builder.build(), metrics);
+ }
+
+ @Bean
+ BusinessEventListener businessEventListener(BusinessLogsProperties properties, BusinessLogsPipeline pipeline,
+ BusinessLogsMetrics metrics) {
+ return new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics);
+ }
+}
diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsMetrics.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsMetrics.java
new file mode 100644
index 000000000..bb9bb3090
--- /dev/null
+++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsMetrics.java
@@ -0,0 +1,42 @@
+package com.iflytek.skillhub.observability.business;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import java.util.concurrent.atomic.AtomicLong;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Fixed-cardinality counters; diagnostics contain no events, exception text, URLs or credentials. */
+final class BusinessLogsMetrics {
+ private static final Logger log = LoggerFactory.getLogger(BusinessLogsMetrics.class);
+ private final AtomicLong lastWarning = new AtomicLong();
+ final Counter enqueued;
+ final Counter queueFull;
+ final Counter closed;
+ final Counter errors;
+ final Counter exportFailures;
+ final Counter exported;
+
+ BusinessLogsMetrics(MeterRegistry registry) {
+ enqueued = registry.counter("skillhub.business.logs.enqueued");
+ queueFull = registry.counter("skillhub.business.logs.dropped", "reason", "queue_full");
+ closed = registry.counter("skillhub.business.logs.dropped", "reason", "shutdown");
+ errors = registry.counter("skillhub.business.logs.errors");
+ exportFailures = registry.counter("skillhub.business.logs.export.failures");
+ exported = registry.counter("skillhub.business.logs.exported");
+ }
+
+ void error() {
+ errors.increment();
+ warn();
+ }
+
+ void warn() {
+ long now = System.nanoTime();
+ long last = lastWarning.get();
+ if ((last == 0 || now - last >= java.util.concurrent.TimeUnit.MINUTES.toNanos(1))
+ && lastWarning.compareAndSet(last, now)) {
+ log.warn("Business Logs telemetry failed; inspect skillhub.business.logs metrics and receiver health");
+ }
+ }
+}
diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsPipeline.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsPipeline.java
new file mode 100644
index 000000000..5851cb128
--- /dev/null
+++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsPipeline.java
@@ -0,0 +1,116 @@
+package com.iflytek.skillhub.observability.business;
+
+import io.opentelemetry.api.logs.Logger;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.logs.LogLimits;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord;
+import io.opentelemetry.sdk.logs.SdkLoggerProvider;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
+import io.opentelemetry.sdk.logs.export.LogRecordExporter;
+import io.opentelemetry.sdk.resources.Resource;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Dedicated SDK pipeline. Admission permits bound queued + in-flight records together, conservatively
+ * below the SDK queue capacity, providing exact admission metrics without accessing SDK internals.
+ * All network batching/retry remains in the standard SDK/exporter.
+ */
+public final class BusinessLogsPipeline implements AutoCloseable {
+ private final SdkLoggerProvider provider;
+ private final Logger logger;
+ private final LogRecordExporter exporter;
+ private final BusinessLogsMetrics metrics;
+ private final Duration shutdownTimeout;
+ private final AtomicBoolean accepting = new AtomicBoolean(true);
+
+ BusinessLogsPipeline(BusinessLogsProperties properties, Resource resource,
+ LogRecordExporter exporter, BusinessLogsMetrics metrics) {
+ this.exporter = exporter;
+ this.metrics = metrics;
+ this.shutdownTimeout = properties.getShutdownTimeout();
+ Semaphore capacity = new Semaphore(properties.getMaxQueueSize());
+ LogRecordExporter measured = new LogRecordExporter() {
+ @Override
+ public CompletableResultCode export(Collection records) {
+ int count = records.size();
+ try {
+ CompletableResultCode result = exporter.export(records);
+ result.whenComplete(() -> {
+ capacity.release(count);
+ if (result.isSuccess()) { metrics.exported.increment(count); }
+ else { metrics.exportFailures.increment(); metrics.warn(); }
+ });
+ return result;
+ } catch (RuntimeException ex) {
+ capacity.release(count);
+ metrics.exportFailures.increment();
+ metrics.warn();
+ return CompletableResultCode.ofFailure();
+ }
+ }
+ @Override public CompletableResultCode flush() { return exporter.flush(); }
+ @Override public CompletableResultCode shutdown() { return exporter.shutdown(); }
+ };
+ BatchLogRecordProcessor batch = BatchLogRecordProcessor.builder(measured)
+ .setMeterProvider(MeterProvider.noop())
+ .setMaxQueueSize(properties.getMaxQueueSize())
+ .setMaxExportBatchSize(properties.getMaxExportBatchSize())
+ .setScheduleDelay(properties.getScheduleDelay())
+ .setExporterTimeout(properties.exportBudget()).build();
+ LogRecordProcessor admission = new LogRecordProcessor() {
+ @Override
+ public void onEmit(Context context, ReadWriteLogRecord record) {
+ if (!accepting.get()) { metrics.closed.increment(); return; }
+ if (!capacity.tryAcquire()) { metrics.queueFull.increment(); return; }
+ try {
+ batch.onEmit(context, record);
+ metrics.enqueued.increment();
+ } catch (RuntimeException ex) {
+ capacity.release();
+ metrics.error();
+ }
+ }
+ @Override public CompletableResultCode forceFlush() { return batch.forceFlush(); }
+ @Override public CompletableResultCode shutdown() { return batch.shutdown(); }
+ };
+ provider = SdkLoggerProvider.builder().setResource(resource)
+ .setLogLimits(() -> LogLimits.builder().setMaxNumberOfAttributes(16)
+ .setMaxAttributeValueLength(256).build())
+ .addLogRecordProcessor(admission).build();
+ logger = provider.loggerBuilder("com.iflytek.skillhub.business").setInstrumentationVersion("1").build();
+ }
+
+ void emit(BusinessEventMapper.MappedRecord record, Context context) {
+ try {
+ if (!accepting.get()) { metrics.closed.increment(); return; }
+ logger.logRecordBuilder().setTimestamp(record.timestamp()).setContext(context)
+ .setAllAttributes(record.attributes()).emit();
+ } catch (RuntimeException ex) {
+ metrics.error();
+ }
+ }
+
+ CompletableResultCode forceFlush() { return provider.forceFlush(); }
+
+ @Override
+ public void close() {
+ if (!accepting.getAndSet(false)) { return; }
+ try {
+ CompletableResultCode shutdown = provider.shutdown();
+ shutdown.join(shutdownTimeout.toMillis(), TimeUnit.MILLISECONDS);
+ if (!shutdown.isDone()) { exporter.shutdown(); }
+ } catch (RuntimeException ex) {
+ metrics.error();
+ // Also cancel network work if shutdown itself unexpectedly fails.
+ try { exporter.shutdown(); } catch (RuntimeException shutdownException) { metrics.error(); }
+ }
+ }
+}
diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsProperties.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsProperties.java
new file mode 100644
index 000000000..6ee7368a5
--- /dev/null
+++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsProperties.java
@@ -0,0 +1,126 @@
+package com.iflytek.skillhub.observability.business;
+
+import java.net.URI;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/** Startup-only operator settings. Deliberately no toString: headers contain secrets. */
+@ConfigurationProperties("skillhub.observability.business-logs")
+public class BusinessLogsProperties {
+ private boolean enabled;
+ private String endpoint = "";
+ private String headers = "";
+ private Duration timeout = Duration.ofSeconds(3);
+ private String compression = "gzip";
+ private int maxQueueSize = 2048;
+ private int maxExportBatchSize = 256;
+ private Duration scheduleDelay = Duration.ofSeconds(1);
+ private Duration shutdownTimeout = Duration.ofSeconds(5);
+ private int retryMaxAttempts = 3;
+ private Duration retryInitialBackoff = Duration.ofMillis(200);
+ private Duration retryMaxBackoff = Duration.ofSeconds(1);
+ private Set eventAllowlist = BusinessEventMapper.EVENT_NAMES;
+ private Set namespaceAllowlist = Set.of();
+
+ public void validate() {
+ if (!enabled) { return; }
+ try {
+ URI uri = URI.create(endpoint);
+ String host = uri.getHost();
+ if (!("http".equals(uri.getScheme()) || "https".equals(uri.getScheme()))
+ || host == null || uri.getUserInfo() != null || uri.getQuery() != null || uri.getFragment() != null
+ || uri.getPath() == null || uri.getPath().isBlank() || "/".equals(uri.getPath())
+ || host.startsWith("169.254.") || host.equals("100.100.100.200")
+ || host.equalsIgnoreCase("metadata.google.internal") || host.equalsIgnoreCase("metadata")
+ || host.equalsIgnoreCase("[fd00:ec2::254]") || host.equals("0.0.0.0")) {
+ throw new IllegalArgumentException();
+ }
+ } catch (RuntimeException ex) {
+ throw new IllegalStateException("skillhub.observability.business-logs.endpoint requires an explicit HTTP(S) Logs URL without credentials, query, fragment or metadata target");
+ }
+ require(maxQueueSize >= 1 && maxQueueSize <= 65536, "max-queue-size must be 1..65536");
+ require(maxExportBatchSize >= 1 && maxExportBatchSize <= Math.min(4096, maxQueueSize),
+ "max-export-batch-size must be 1..min(4096,max-queue-size)");
+ duration(timeout, 100, 30000, "timeout");
+ duration(scheduleDelay, 10, 60000, "schedule-delay");
+ duration(shutdownTimeout, 100, 30000, "shutdown-timeout");
+ duration(retryInitialBackoff, 1, 5000, "retry-initial-backoff");
+ duration(retryMaxBackoff, 1, 10000, "retry-max-backoff");
+ require(retryMaxAttempts >= 1 && retryMaxAttempts <= 5, "retry-max-attempts must be 1..5");
+ require(retryInitialBackoff.compareTo(retryMaxBackoff) <= 0, "retry backoff bounds are inconsistent");
+ require(Set.of("none", "gzip").contains(compression), "compression must be none or gzip");
+ require(eventAllowlist != null && BusinessEventMapper.EVENT_NAMES.containsAll(eventAllowlist), "event-allowlist contains unsupported event names");
+ require(namespaceAllowlist != null && namespaceAllowlist.size() <= 1000
+ && namespaceAllowlist.stream().allMatch(id -> id != null && id > 0), "namespace-allowlist requires at most 1000 positive IDs");
+ parsedHeaders();
+ }
+
+ Map parsedHeaders() {
+ try {
+ if (headers == null || headers.isBlank()) { return Map.of(); }
+ if (headers.length() > 8192) { throw new IllegalArgumentException(); }
+ Map result = new LinkedHashMap<>();
+ String[] pairs = headers.split(",", -1);
+ if (pairs.length > 16) { throw new IllegalArgumentException(); }
+ for (String header : pairs) {
+ String[] pair = header.split("=", 2);
+ if (pair.length != 2 || result.size() >= 16) { throw new IllegalArgumentException(); }
+ String name = URLDecoder.decode(pair[0].trim(), StandardCharsets.UTF_8);
+ String value = URLDecoder.decode(pair[1].trim(), StandardCharsets.UTF_8);
+ if (!name.matches("[A-Za-z0-9!#$%&'*+.^_`|~-]{1,128}")
+ || value.chars().anyMatch(c -> c < 32 || c > 126)) { throw new IllegalArgumentException(); }
+ result.put(name, value);
+ }
+ return Map.copyOf(result);
+ } catch (RuntimeException ex) {
+ throw new IllegalStateException("business-logs.headers must contain at most 16 URL-encoded name=value pairs (8192 characters), without control characters");
+ }
+ }
+
+ private static void duration(Duration value, long min, long max, String key) {
+ require(value != null && value.compareTo(Duration.ofMillis(min)) >= 0
+ && value.compareTo(Duration.ofMillis(max)) <= 0, key + " is outside documented bounds");
+ }
+
+ private static void require(boolean condition, String message) {
+ if (!condition) { throw new IllegalStateException("skillhub.observability.business-logs: " + message); }
+ }
+
+ Duration exportBudget() {
+ return timeout.plus(retryMaxBackoff).multipliedBy(retryMaxAttempts).plusSeconds(1);
+ }
+
+ public boolean isEnabled() { return enabled; }
+ public void setEnabled(boolean enabled) { this.enabled = enabled; }
+ public String getEndpoint() { return endpoint; }
+ public void setEndpoint(String endpoint) { this.endpoint = endpoint; }
+ public String getHeaders() { return headers; }
+ public void setHeaders(String headers) { this.headers = headers; }
+ public Duration getTimeout() { return timeout; }
+ public void setTimeout(Duration timeout) { this.timeout = timeout; }
+ public String getCompression() { return compression; }
+ public void setCompression(String compression) { this.compression = compression; }
+ public int getMaxQueueSize() { return maxQueueSize; }
+ public void setMaxQueueSize(int maxQueueSize) { this.maxQueueSize = maxQueueSize; }
+ public int getMaxExportBatchSize() { return maxExportBatchSize; }
+ public void setMaxExportBatchSize(int maxExportBatchSize) { this.maxExportBatchSize = maxExportBatchSize; }
+ public Duration getScheduleDelay() { return scheduleDelay; }
+ public void setScheduleDelay(Duration scheduleDelay) { this.scheduleDelay = scheduleDelay; }
+ public Duration getShutdownTimeout() { return shutdownTimeout; }
+ public void setShutdownTimeout(Duration shutdownTimeout) { this.shutdownTimeout = shutdownTimeout; }
+ public int getRetryMaxAttempts() { return retryMaxAttempts; }
+ public void setRetryMaxAttempts(int retryMaxAttempts) { this.retryMaxAttempts = retryMaxAttempts; }
+ public Duration getRetryInitialBackoff() { return retryInitialBackoff; }
+ public void setRetryInitialBackoff(Duration retryInitialBackoff) { this.retryInitialBackoff = retryInitialBackoff; }
+ public Duration getRetryMaxBackoff() { return retryMaxBackoff; }
+ public void setRetryMaxBackoff(Duration retryMaxBackoff) { this.retryMaxBackoff = retryMaxBackoff; }
+ public Set getEventAllowlist() { return eventAllowlist; }
+ public void setEventAllowlist(Set eventAllowlist) { this.eventAllowlist = eventAllowlist; }
+ public Set getNamespaceAllowlist() { return namespaceAllowlist; }
+ public void setNamespaceAllowlist(Set namespaceAllowlist) { this.namespaceAllowlist = namespaceAllowlist; }
+}
diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/package-info.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/package-info.java
new file mode 100644
index 000000000..a81685cb1
--- /dev/null
+++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/package-info.java
@@ -0,0 +1,2 @@
+/** Opt-in, independent, lossy business analytics adapter; never an audit or application-log bridge. */
+package com.iflytek.skillhub.observability.business;
diff --git a/server/skillhub-app/src/main/resources/application.yml b/server/skillhub-app/src/main/resources/application.yml
index 3b2f58f82..c54db5379 100644
--- a/server/skillhub-app/src/main/resources/application.yml
+++ b/server/skillhub-app/src/main/resources/application.yml
@@ -170,6 +170,22 @@ skillhub:
confirmation-enabled: ${SKILLHUB_SUITE_BUNDLE_CONFIRMATION_ENABLED:false}
observability:
tracing-mode: ${SKILLHUB_TRACING_MODE:none}
+ business-logs:
+ enabled: ${SKILLHUB_BUSINESS_LOGS_ENABLED:false}
+ # Logs-specific URL is used as-is; include /v1/logs (not /v1/traces).
+ endpoint: ${SKILLHUB_BUSINESS_LOGS_ENDPOINT:}
+ headers: ${SKILLHUB_BUSINESS_LOGS_HEADERS:}
+ timeout: ${SKILLHUB_BUSINESS_LOGS_TIMEOUT:3s}
+ compression: ${SKILLHUB_BUSINESS_LOGS_COMPRESSION:gzip}
+ max-queue-size: ${SKILLHUB_BUSINESS_LOGS_MAX_QUEUE_SIZE:2048}
+ max-export-batch-size: ${SKILLHUB_BUSINESS_LOGS_MAX_EXPORT_BATCH_SIZE:256}
+ schedule-delay: ${SKILLHUB_BUSINESS_LOGS_SCHEDULE_DELAY:1s}
+ shutdown-timeout: ${SKILLHUB_BUSINESS_LOGS_SHUTDOWN_TIMEOUT:5s}
+ retry-max-attempts: ${SKILLHUB_BUSINESS_LOGS_RETRY_MAX_ATTEMPTS:3}
+ retry-initial-backoff: ${SKILLHUB_BUSINESS_LOGS_RETRY_INITIAL_BACKOFF:200ms}
+ retry-max-backoff: ${SKILLHUB_BUSINESS_LOGS_RETRY_MAX_BACKOFF:1s}
+ event-allowlist: ${SKILLHUB_BUSINESS_LOGS_EVENT_ALLOWLIST:skill.publication.effective,skill.download.prepared,skill.review.approved,skill.review.rejected,skill.promotion.approved,skill.promotion.rejected,skill.status.changed,skill.visibility.changed,skill.version.yanked}
+ namespace-allowlist: ${SKILLHUB_BUSINESS_LOGS_NAMESPACE_ALLOWLIST:}
log-format: ${SKILLHUB_LOG_FORMAT:text}
log-async-queue-size: ${SKILLHUB_LOG_ASYNC_QUEUE_SIZE:1024}
service-version: ${SKILLHUB_SERVICE_VERSION:unknown}
diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessEventMapperTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessEventMapperTest.java
new file mode 100644
index 000000000..05e0a31d4
--- /dev/null
+++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessEventMapperTest.java
@@ -0,0 +1,80 @@
+package com.iflytek.skillhub.observability.business;
+
+import com.iflytek.skillhub.domain.event.*;
+import com.iflytek.skillhub.domain.skill.SkillVisibility;
+import io.opentelemetry.api.common.AttributeKey;
+import java.time.Instant;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class BusinessEventMapperTest {
+ private final BusinessEventMapper mapper = new BusinessEventMapper(new BusinessLogsProperties());
+ private final BusinessEventSnapshot snapshot = new BusinessEventSnapshot(
+ 7L, SkillVisibility.PRIVATE, "reviewer-opaque", "user", "review", Instant.parse("2026-10-01T00:00:00Z"));
+
+ @Test
+ void rejectionExportsOnlyExplicitAllowlistNeverReasonOrSubmitter() {
+ var record = mapper.map(new ReviewRejectedEvent(3L, 1L, 2L, "reviewer-opaque", "secret-email",
+ "token=secret signed-url IP UA content", snapshot)).orElseThrow();
+ assertThat(record.timestamp()).isEqualTo(snapshot.occurredAt());
+ assertThat(record.attributes().asMap().keySet()).extracting(AttributeKey::getKey).containsExactlyInAnyOrder(
+ "skillhub.event.name", "skillhub.skill.id", "skillhub.version.id", "skillhub.namespace.id",
+ "skillhub.visibility", "skillhub.source", "skillhub.actor.type", "user.id", "log.record.uid",
+ "skillhub.review.id");
+ assertThat(record.attributes().toString()).doesNotContain("secret", "signed-url", "content", "submitter");
+ assertThat(record.attributes().get(AttributeKey.stringKey("user.id"))).isEqualTo("reviewer-opaque");
+ }
+
+ @Test
+ void anonymousHasNoUserIdAndMappingHasStableUid() {
+ var anonymous = new BusinessEventSnapshot(7L, SkillVisibility.PUBLIC, null, "user", "download", snapshot.occurredAt());
+ var record = mapper.map(new SkillDownloadedEvent(1L, 2L, anonymous)).orElseThrow();
+ assertThat(record.attributes().get(AttributeKey.stringKey("user.id"))).isNull();
+ assertThat(record.attributes().get(AttributeKey.stringKey("skillhub.actor.type"))).isEqualTo("user");
+ assertThat(record.attributes().get(AttributeKey.stringKey("log.record.uid")))
+ .isEqualTo(record.uid()).matches("[0-9a-f-]{36}");
+ }
+
+ @Test
+ void namespaceAndEventAllowlistFailClosedAndMissingSnapshotIsNotEnrichedFromDatabase() {
+ var properties = new BusinessLogsProperties();
+ properties.setNamespaceAllowlist(Set.of(8L));
+ assertThat(new BusinessEventMapper(properties).map(new SkillPublishedEvent(1L, 2L, "actor", snapshot))).isEmpty();
+ properties.setNamespaceAllowlist(Set.of());
+ properties.setEventAllowlist(Set.of("skill.download.prepared"));
+ assertThat(new BusinessEventMapper(properties).map(new SkillPublishedEvent(1L, 2L, "actor", snapshot))).isEmpty();
+ assertThat(mapper.map(new SkillPublishedEvent(1L, 2L, "actor"))).isEmpty();
+ }
+
+ @Test
+ void userValuesAreLengthBoundedAndSystemDoesNotExportServiceAccountAsHuman() {
+ var system = new BusinessEventSnapshot(7L, SkillVisibility.PRIVATE, "x".repeat(1000), "system", "auto_publish", snapshot.occurredAt());
+ var record = mapper.map(new SkillPublishedEvent(1L, 2L, "ignored", system)).orElseThrow();
+ assertThat(record.attributes().get(AttributeKey.stringKey("user.id"))).isNull();
+ var human = new BusinessEventSnapshot(7L, SkillVisibility.PRIVATE, "x".repeat(1000), "user", "review", snapshot.occurredAt());
+ assertThat(mapper.map(new SkillPublishedEvent(1L, 2L, "ignored", human)).orElseThrow()
+ .attributes().get(AttributeKey.stringKey("user.id"))).hasSize(256);
+ }
+
+ @Test
+ void unknownEventAndNullContextFieldsAreNeverSerializedOrEnriched() {
+ record SensitiveEvent(Long skillId, BusinessEventSnapshot snapshot, String password) implements SkillBusinessEvent {}
+ assertThat(mapper.map(new SensitiveEvent(1L, snapshot, "must-never-appear"))).isEmpty();
+ var missingSource = new BusinessEventSnapshot(7L, SkillVisibility.PRIVATE, "actor", "user", null, snapshot.occurredAt());
+ assertThat(mapper.map(new SkillPublishedEvent(1L, 2L, "actor", missingSource))).isEmpty();
+ assertThat(mapper.map(new SkillVersionYankedEvent(1L, 2L, "actor", false, snapshot))).isEmpty();
+ assertThat(mapper.map(new SkillStatusChangedEvent(1L, com.iflytek.skillhub.domain.skill.SkillStatus.ACTIVE,
+ com.iflytek.skillhub.domain.skill.SkillStatus.ACTIVE, snapshot))).isEmpty();
+ }
+
+ @Test
+ void reservedBuiltInPublisherIsSystemEvenThroughOrdinaryPrivilegedPublishPath() {
+ var skill = new com.iflytek.skillhub.domain.skill.Skill(7L, "not-exported", "builtin-skill-publisher", SkillVisibility.PUBLIC);
+ var captured = BusinessEventSnapshot.capture(skill, "builtin-skill-publisher", "user", "auto_publish");
+ assertThat(captured.actorType()).isEqualTo("system");
+ var record = mapper.map(new SkillPublishedEvent(1L, 2L, "builtin-skill-publisher", captured)).orElseThrow();
+ assertThat(record.attributes().get(AttributeKey.stringKey("user.id"))).isNull();
+ }
+}
diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsCollectorTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsCollectorTest.java
new file mode 100644
index 000000000..d17bdfeec
--- /dev/null
+++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsCollectorTest.java
@@ -0,0 +1,186 @@
+package com.iflytek.skillhub.observability.business;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iflytek.skillhub.domain.event.*;
+import com.iflytek.skillhub.domain.skill.SkillStatus;
+import com.iflytek.skillhub.domain.skill.SkillVisibility;
+import com.iflytek.skillhub.observability.tracing.SkillHubTracingConfiguration;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.opentelemetry.api.common.AttributeKey;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.MountableFile;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Actual production configuration sends protobuf over HTTP to a pinned real Collector/file sink. */
+@Testcontainers
+class BusinessLogsCollectorTest {
+ @Container
+ static final GenericContainer COLLECTOR = new GenericContainer<>("otel/opentelemetry-collector-contrib:0.111.0")
+ .withExposedPorts(4318)
+ .withCopyFileToContainer(MountableFile.forHostPath(Path.of("../../docs/examples/business-event-logs/collector.yaml").toAbsolutePath()),
+ "/etc/otelcol-contrib/config.yaml")
+ .withCreateContainerCmdModifier(cmd -> {
+ cmd.withName("skillhub-1009-collector-integration");
+ cmd.withUser("0:0");
+ cmd.getHostConfig().withMemory(128L * 1024 * 1024).withNanoCPUs(500_000_000L);
+ })
+ .waitingFor(Wait.forLogMessage(".*Everything is ready.*", 1));
+
+ @Configuration(proxyBeanMethods = false)
+ @EnableAutoConfiguration
+ @Import({SkillHubTracingConfiguration.class, BusinessLogsConfiguration.class})
+ static class Application {}
+
+ @Test
+ void realCollectorReceivesNineFactsInAllThreeTraceModesAndResourceLimitsAreExplicit() throws Exception {
+ var hostConfig = COLLECTOR.getCurrentContainerInfo().getHostConfig();
+ assertThat(hostConfig.getMemory()).isEqualTo(128L * 1024 * 1024);
+ assertThat(hostConfig.getNanoCPUs()).isEqualTo(500_000_000L);
+ var snapshot = new BusinessEventSnapshot(7L, SkillVisibility.PRIVATE, "reviewer", "user", "review", Instant.now());
+ var events = List.of(
+ new SkillPublishedEvent(1L, 2L, "reviewer", snapshot),
+ new SkillDownloadedEvent(1L, 2L, new BusinessEventSnapshot(7L, SkillVisibility.PUBLIC, null, "user", "download", snapshot.occurredAt())),
+ new ReviewApprovedEvent(3L, 1L, 2L, "reviewer", "private-owner", snapshot),
+ new ReviewRejectedEvent(3L, 1L, 2L, "reviewer", "private-owner", "SECRET-CONTENT-URL-TOKEN", snapshot),
+ new PromotionApprovedEvent(4L, 1L, "reviewer", "private-owner", snapshot),
+ new PromotionRejectedEvent(4L, 1L, "reviewer", "private-owner", "SECRET-CONTENT-URL-TOKEN", snapshot),
+ new SkillStatusChangedEvent(1L, SkillStatus.ACTIVE, SkillStatus.ARCHIVED, snapshot),
+ new SkillVisibilityChangedEvent(1L, SkillVisibility.PUBLIC, SkillVisibility.PRIVATE, snapshot),
+ new SkillVersionYankedEvent(1L, 2L, "reviewer", true, snapshot));
+ for (String mode : List.of("none", "otel-sdk", "external-agent")) {
+ new ApplicationContextRunner().withUserConfiguration(Application.class)
+ .withBean(SimpleMeterRegistry.class, SimpleMeterRegistry::new)
+ .withPropertyValues("spring.flyway.enabled=false", "spring.jpa.hibernate.ddl-auto=none",
+ "skillhub.observability.tracing-mode=" + mode,
+ "management.tracing.sampling.probability=0",
+ "skillhub.observability.service-version=collector-test",
+ "skillhub.observability.service-environment=test",
+ "skillhub.observability.business-logs.enabled=true",
+ "skillhub.observability.business-logs.endpoint=http://" + COLLECTOR.getHost() + ":" + COLLECTOR.getMappedPort(4318) + "/v1/logs")
+ .run(context -> {
+ assertThat(context).hasNotFailed();
+ events.forEach(context::publishEvent);
+ var result = context.getBean(BusinessLogsPipeline.class).forceFlush();
+ result.join(10, TimeUnit.SECONDS);
+ assertThat(result.isSuccess()).isTrue();
+ var registry = context.getBean(SimpleMeterRegistry.class);
+ assertThat(registry.get("skillhub.business.logs.exported").counter().count()).isEqualTo(9);
+ });
+ }
+ String raw = awaitFile(27);
+ assertThat(raw).doesNotContain("SECRET-CONTENT-URL-TOKEN", "private-owner", "request.id", "authorization");
+ List records = records(raw);
+ assertThat(records).hasSize(27);
+ var names = new HashSet();
+ var uids = new HashSet();
+ for (JsonNode record : records) {
+ var attrs = attributes(record);
+ names.add(attrs.get("skillhub.event.name").asText());
+ assertThat(uids.add(attrs.get("log.record.uid").asText())).isTrue();
+ assertThat(attrs.size()).isLessThanOrEqualTo(16);
+ assertThat(attrs.get("skillhub.namespace.id").asText()).isEqualTo("7");
+ if (attrs.get("skillhub.event.name").asText().equals("skill.download.prepared")) { assertThat(attrs).doesNotContainKey("user.id"); }
+ }
+ assertThat(names).containsExactlyInAnyOrderElementsOf(BusinessEventMapper.EVENT_NAMES);
+ String evidence = System.getProperty("skillhub.test.evidence-dir");
+ 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";
+ COLLECTOR.getDockerClient().pauseContainerCmd(COLLECTOR.getContainerId()).exec();
+ try {
+ outageRunner(endpoint).run(context -> {
+ assertThat(context).hasNotFailed();
+ long start = System.nanoTime();
+ for (int i = 0; i < 1000; i++) { context.publishEvent(events.getFirst()); }
+ long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ assertThat(elapsed).isLessThan(1000);
+ context.getBean(BusinessLogsPipeline.class).forceFlush().join(5, TimeUnit.SECONDS);
+ var registry = context.getBean(SimpleMeterRegistry.class);
+ double dropped = registry.get("skillhub.business.logs.dropped").tag("reason", "queue_full").counter().count();
+ assertThat(dropped).isGreaterThanOrEqualTo(998);
+ assertThat(registry.get("skillhub.business.logs.export.failures").counter().count()).isPositive();
+ System.out.printf("BUSINESS_LOGS paused real Collector: 1000 caller events in %d ms, drops=%.0f; timeout failures counted%n", elapsed, dropped);
+ });
+ } finally { COLLECTOR.getDockerClient().unpauseContainerCmd(COLLECTOR.getContainerId()).exec(); }
+ COLLECTOR.stop();
+ outageRunner(endpoint).run(context -> {
+ assertThat(context).hasNotFailed();
+ long start = System.nanoTime();
+ context.publishEvent(events.getFirst());
+ long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ assertThat(elapsed).isLessThan(1000);
+ context.getBean(BusinessLogsPipeline.class).forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(context.getBean(SimpleMeterRegistry.class).get("skillhub.business.logs.export.failures").counter().count()).isEqualTo(1);
+ System.out.printf("BUSINESS_LOGS stopped real Collector: caller=%d ms; failure counted, no business exception%n", elapsed);
+ });
+ }
+
+ private static ApplicationContextRunner outageRunner(String endpoint) {
+ return new ApplicationContextRunner().withUserConfiguration(BusinessLogsConfiguration.class)
+ .withBean(SimpleMeterRegistry.class, SimpleMeterRegistry::new)
+ .withPropertyValues("skillhub.observability.business-logs.enabled=true",
+ "skillhub.observability.business-logs.endpoint=" + endpoint,
+ "skillhub.observability.business-logs.max-queue-size=2",
+ "skillhub.observability.business-logs.max-export-batch-size=1",
+ "skillhub.observability.business-logs.timeout=500ms",
+ "skillhub.observability.business-logs.retry-max-attempts=1");
+ }
+
+ private static String awaitFile(int expected) throws Exception {
+ long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos();
+ String raw = "";
+ Exception lastError = null;
+ do {
+ try {
+ raw = COLLECTOR.copyFileFromContainer("/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);
+ } while (System.nanoTime() < deadline);
+ String evidence = System.getProperty("skillhub.test.evidence-dir");
+ if (evidence != null) {
+ Files.writeString(Path.of(evidence, "collector-debug.jsonl"), raw);
+ Files.writeString(Path.of(evidence, "collector-debug.log"), COLLECTOR.getLogs());
+ }
+ throw new AssertionError("Collector file did not contain " + expected + " records; raw characters=" + raw.length(), lastError);
+ }
+
+ static List records(String raw) throws Exception {
+ List result = new ArrayList<>();
+ var mapper = new ObjectMapper();
+ for (String line : raw.lines().filter(s -> !s.isBlank()).toList()) {
+ JsonNode root = mapper.readTree(line);
+ for (JsonNode resource : root.path("resourceLogs")) {
+ for (JsonNode scope : resource.path("scopeLogs")) {
+ for (JsonNode record : scope.path("logRecords")) { result.add(record); }
+ }
+ }
+ }
+ return result;
+ }
+
+ static Map attributes(JsonNode record) {
+ var result = new java.util.HashMap();
+ for (JsonNode attr : record.path("attributes")) { result.put(attr.path("key").asText(), attr.path("value").elements().next()); }
+ return result;
+ }
+}
diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsConfigurationTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsConfigurationTest.java
new file mode 100644
index 000000000..e1330f0dd
--- /dev/null
+++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsConfigurationTest.java
@@ -0,0 +1,81 @@
+package com.iflytek.skillhub.observability.business;
+
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class BusinessLogsConfigurationTest {
+ private final ApplicationContextRunner runner = new ApplicationContextRunner()
+ .withUserConfiguration(BusinessLogsConfiguration.class)
+ .withBean(SimpleMeterRegistry.class, SimpleMeterRegistry::new);
+
+ @Test
+ void defaultOffCreatesNoProviderPipelineOrListener() {
+ runner.run(context -> {
+ assertThat(context).hasNotFailed();
+ assertThat(context).doesNotHaveBean(BusinessLogsPipeline.class);
+ assertThat(context).doesNotHaveBean(BusinessEventListener.class);
+ assertThat(context).doesNotHaveBean(io.opentelemetry.sdk.logs.SdkLoggerProvider.class);
+ });
+ }
+
+ @Test
+ void enabledRequiresExplicitEndpoint() {
+ runner.withPropertyValues("skillhub.observability.business-logs.enabled=true")
+ .run(context -> assertThat(context).hasFailed());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"http://169.254.169.254/v1/logs", "http://metadata.google.internal/v1/logs",
+ "http://user:password@localhost/v1/logs", "http://localhost/v1/logs?secret=x", "file:///v1/logs"})
+ void invalidDestinationsFailWithoutEchoingInput(String endpoint) {
+ var properties = new BusinessLogsProperties();
+ properties.setEnabled(true);
+ properties.setEndpoint(endpoint);
+ assertThatThrownBy(properties::validate).isInstanceOf(IllegalStateException.class)
+ .hasMessageNotContaining(endpoint).hasMessageNotContaining("password").hasMessageNotContaining("secret=x");
+ }
+
+ @Test
+ void budgetsAndEventNamesAreValidated() {
+ var properties = new BusinessLogsProperties();
+ properties.setEnabled(true);
+ properties.setEndpoint("http://localhost:4318/v1/logs");
+ properties.setMaxQueueSize(1);
+ properties.setMaxExportBatchSize(2);
+ assertThatThrownBy(properties::validate).isInstanceOf(IllegalStateException.class);
+ properties.setMaxExportBatchSize(1);
+ properties.setEventAllowlist(java.util.Set.of("not-a-business-event"));
+ assertThatThrownBy(properties::validate).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void headerEncodingIsSecretFriendlyAndErrorsNeverEchoSecretInput() {
+ var properties = new BusinessLogsProperties();
+ properties.setHeaders("Authorization=Bearer%20fake%2Bsecret,x-tenant=example");
+ assertThat(properties.parsedHeaders()).containsEntry("Authorization", "Bearer fake+secret");
+ properties.setHeaders("Authorization=fake-secret%0Ainjected");
+ assertThatThrownBy(properties::parsedHeaders).isInstanceOf(IllegalStateException.class)
+ .hasMessageNotContaining("fake-secret").hasMessageNotContaining("injected");
+ properties.setHeaders(String.join(",", java.util.Collections.nCopies(17, "same=value")));
+ assertThatThrownBy(properties::parsedHeaders).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void emptyNamespaceFilterAndExplicitEmptyEventFilterBindWithoutDefaultDestination() {
+ runner.withPropertyValues("skillhub.observability.business-logs.enabled=true",
+ "skillhub.observability.business-logs.endpoint=http://127.0.0.1:4318/v1/logs",
+ "skillhub.observability.business-logs.namespace-allowlist=",
+ "skillhub.observability.business-logs.event-allowlist=")
+ .run(context -> {
+ assertThat(context).hasNotFailed();
+ assertThat(context.getBean(BusinessLogsProperties.class).getNamespaceAllowlist()).isEmpty();
+ assertThat(context.getBean(BusinessLogsProperties.class).getEventAllowlist()).isEmpty();
+ });
+ }
+}
diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsHttpTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsHttpTest.java
new file mode 100644
index 000000000..73749ae00
--- /dev/null
+++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsHttpTest.java
@@ -0,0 +1,89 @@
+package com.iflytek.skillhub.observability.business;
+
+import com.iflytek.skillhub.domain.event.BusinessEventSnapshot;
+import com.iflytek.skillhub.domain.event.SkillPublishedEvent;
+import com.iflytek.skillhub.domain.skill.SkillVisibility;
+import com.sun.net.httpserver.HttpServer;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporter;
+import io.opentelemetry.sdk.common.export.RetryPolicy;
+import io.opentelemetry.sdk.resources.Resource;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class BusinessLogsHttpTest {
+ @Test
+ void boundedRetryReusesIdenticalUidAndPayloadAndAuthenticationIsTransportOnly() throws Exception {
+ var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ List bodies = new CopyOnWriteArrayList<>();
+ List auth = new CopyOnWriteArrayList<>();
+ var requests = new AtomicInteger();
+ server.createContext("/v1/logs", exchange -> {
+ bodies.add(exchange.getRequestBody().readAllBytes());
+ auth.add(exchange.getRequestHeaders().getFirst("Authorization"));
+ exchange.sendResponseHeaders(requests.incrementAndGet() == 1 ? 503 : 200, -1);
+ exchange.close();
+ });
+ server.start();
+ var properties = new BusinessLogsProperties();
+ properties.setCompression("none");
+ var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry());
+ var exporter = OtlpHttpLogRecordExporter.builder()
+ .setEndpoint("http://127.0.0.1:" + server.getAddress().getPort() + "/v1/logs")
+ .setTimeout(Duration.ofMillis(500)).addHeader("Authorization", "Bearer fake-deployment-secret")
+ .setRetryPolicy(RetryPolicy.builder().setMaxAttempts(2).setInitialBackoff(Duration.ofMillis(10))
+ .setMaxBackoff(Duration.ofMillis(10)).build()).build();
+ try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) {
+ var snapshot = new BusinessEventSnapshot(7L, SkillVisibility.PUBLIC, "reviewer", "user", "review", Instant.now());
+ var record = new BusinessEventMapper(properties).map(new SkillPublishedEvent(1L, 2L, "reviewer", snapshot)).orElseThrow();
+ pipeline.emit(record, Context.root());
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(requests.get()).isEqualTo(2);
+ assertThat(bodies.get(0)).isEqualTo(bodies.get(1));
+ String payload = new String(bodies.getFirst(), java.nio.charset.StandardCharsets.ISO_8859_1);
+ assertThat(payload).contains(record.uid()).doesNotContain("fake-deployment-secret", "Authorization");
+ assertThat(auth).containsExactly("Bearer fake-deployment-secret", "Bearer fake-deployment-secret");
+ assertThat(metrics.exported.count()).isEqualTo(1);
+ assertThat(metrics.exportFailures.count()).isZero();
+ System.out.println("BUSINESS_LOGS retry: 503 then 200, 2 identical payloads, 1 stable uid; transport-only auth");
+ } finally { server.stop(0); }
+ }
+
+ @Test
+ void twoHundredPartialSuccessIsCountedAsSdkSuccessWithoutWholeBatchReplay() throws Exception {
+ var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ var requests = new AtomicInteger();
+ server.createContext("/v1/logs", exchange -> {
+ exchange.getRequestBody().readAllBytes();
+ requests.incrementAndGet();
+ // ExportLogsServiceResponse.partial_success { rejected_log_records: 1 }
+ byte[] body = {0x0a, 0x02, 0x08, 0x01};
+ exchange.getResponseHeaders().add("Content-Type", "application/x-protobuf");
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ exchange.close();
+ });
+ server.start();
+ var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry());
+ var properties = new BusinessLogsProperties();
+ var exporter = OtlpHttpLogRecordExporter.builder()
+ .setEndpoint("http://127.0.0.1:" + server.getAddress().getPort() + "/v1/logs").build();
+ try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) {
+ var snapshot = new BusinessEventSnapshot(7L, SkillVisibility.PUBLIC, "reviewer", "user", "review", Instant.now());
+ pipeline.emit(new BusinessEventMapper(properties).map(new SkillPublishedEvent(1L, 2L, "reviewer", snapshot)).orElseThrow(), Context.root());
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(requests.get()).isEqualTo(1);
+ assertThat(metrics.exported.count()).isEqualTo(1);
+ System.out.println("BUSINESS_LOGS partial-success: 200 rejected=1 body, SDK success=1, requests=1 (known 1.31 limitation)");
+ } finally { server.stop(0); }
+ }
+}
diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsProducerPostgresTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsProducerPostgresTest.java
new file mode 100644
index 000000000..cfbbfc845
--- /dev/null
+++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsProducerPostgresTest.java
@@ -0,0 +1,322 @@
+package com.iflytek.skillhub.observability.business;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iflytek.skillhub.domain.audit.AuditLogService;
+import com.iflytek.skillhub.domain.governance.GovernanceNotificationService;
+import com.iflytek.skillhub.domain.namespace.*;
+import com.iflytek.skillhub.domain.review.*;
+import com.iflytek.skillhub.domain.security.*;
+import com.iflytek.skillhub.domain.skill.*;
+import com.iflytek.skillhub.domain.skill.metadata.*;
+import com.iflytek.skillhub.domain.skill.service.*;
+import com.iflytek.skillhub.domain.skill.validation.*;
+import com.iflytek.skillhub.domain.user.UserAccount;
+import com.iflytek.skillhub.infra.jpa.*;
+import com.iflytek.skillhub.storage.ObjectMetadata;
+import com.iflytek.skillhub.storage.ObjectStorageService;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.resources.Resource;
+import java.time.Clock;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
+import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.testcontainers.containers.PostgreSQLContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.when;
+
+/** Real service proxies, JPA repositories, schema migrations, commit/rollback and analytics adapter. */
+@DataJpaTest(showSql = false)
+@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
+@ActiveProfiles("test")
+@Testcontainers
+@org.springframework.test.annotation.DirtiesContext(classMode = org.springframework.test.annotation.DirtiesContext.ClassMode.AFTER_CLASS)
+@Import({SkillOwnerAccessPolicy.class, JpaSkillRepositoryAdapter.class, SkillPublishService.class,
+ SkillReviewSubmitService.class, SkillGovernanceService.class, SkillVisibilityService.class,
+ SkillDownloadService.class, SkillSlugResolutionService.class, ReviewService.class, PromotionService.class,
+ ReviewPermissionChecker.class, BusinessLogsProducerPostgresTest.Configuration.class})
+@Transactional(propagation = Propagation.NOT_SUPPORTED)
+class BusinessLogsProducerPostgresTest {
+ @Container
+ static final PostgreSQLContainer POSTGRES = new PostgreSQLContainer<>("postgres:16-alpine")
+ .withTmpFs(Map.of("/var/lib/postgresql/data", "rw"))
+ .withCreateContainerCmdModifier(cmd -> cmd.withName("skillhub-1009-producers-postgres"));
+
+ @DynamicPropertySource
+ static void database(DynamicPropertyRegistry registry) {
+ registry.add("spring.datasource.url", POSTGRES::getJdbcUrl);
+ registry.add("spring.datasource.username", POSTGRES::getUsername);
+ registry.add("spring.datasource.password", POSTGRES::getPassword);
+ registry.add("spring.datasource.driver-class-name", () -> "org.postgresql.Driver");
+ registry.add("spring.jpa.database-platform", () -> "org.hibernate.dialect.PostgreSQLDialect");
+ registry.add("spring.flyway.enabled", () -> true);
+ registry.add("spring.jpa.hibernate.ddl-auto", () -> "validate");
+ }
+
+ @TestConfiguration
+ static class Configuration {
+ @Bean Clock clock() { return Clock.systemUTC(); }
+ @Bean ObjectMapper objectMapper() { return new ObjectMapper(); }
+ @Bean VisibilityChecker visibilityChecker() { return new VisibilityChecker(); }
+ @Bean BusinessLogsTransactionTest.CapturingExporter exporter() { return new BusinessLogsTransactionTest.CapturingExporter(); }
+ @Bean BusinessLogsMetrics metrics() { return new BusinessLogsMetrics(new SimpleMeterRegistry()); }
+ @Bean(destroyMethod = "close") BusinessLogsPipeline pipeline(BusinessLogsTransactionTest.CapturingExporter exporter, BusinessLogsMetrics metrics) {
+ return new BusinessLogsPipeline(new BusinessLogsProperties(), Resource.empty(), exporter, metrics);
+ }
+ @Bean BusinessEventListener listener(BusinessLogsPipeline pipeline, BusinessLogsMetrics metrics) {
+ return new BusinessEventListener(new BusinessEventMapper(new BusinessLogsProperties()), pipeline, metrics);
+ }
+ }
+
+ @MockBean SecurityScanService scanner;
+ @MockBean GovernanceNotificationService notifications;
+ @MockBean ReviewAvatarParticipant avatars;
+ @MockBean ObjectStorageService storage;
+ @MockBean SkillStorageDeletionCompensationService compensation;
+ @MockBean AuditLogService audit;
+ @MockBean SkillPackageValidator validator;
+ @MockBean SkillMetadataParser parser;
+ @MockBean PrePublishValidator prePublish;
+ @Autowired SkillPublishService publishing;
+ @Autowired SkillReviewSubmitService confirmations;
+ @Autowired ReviewService reviewing;
+ @Autowired PromotionService promoting;
+ @Autowired SkillGovernanceService governance;
+ @Autowired SkillVisibilityService visibility;
+ @Autowired SkillDownloadService downloads;
+ @Autowired SkillJpaRepository skills;
+ @Autowired SkillVersionJpaRepository versions;
+ @Autowired ReviewTaskJpaRepository reviews;
+ @Autowired PromotionRequestJpaRepository promotions;
+ @Autowired NamespaceJpaRepository namespaces;
+ @Autowired NamespaceMemberJpaRepository members;
+ @Autowired UserAccountJpaRepository users;
+ @Autowired SecurityAuditJpaRepository audits;
+ @Autowired PlatformTransactionManager transactions;
+ @Autowired BusinessLogsPipeline pipeline;
+ @Autowired BusinessLogsTransactionTest.CapturingExporter exporter;
+
+ @BeforeEach
+ void setup() {
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ exporter.records.clear();
+ when(scanner.isEnabled()).thenReturn(true);
+ when(validator.validate(anyList())).thenReturn(ValidationResult.pass());
+ when(prePublish.validate(any())).thenReturn(ValidationResult.pass());
+ when(parser.parse(anyString())).thenReturn(new SkillMetadata("business-probe", "Never export title", "3.0.0", "Never export body", Map.of()));
+ when(storage.exists(anyString())).thenReturn(true);
+ when(storage.getMetadata(anyString())).thenReturn(new ObjectMetadata(12, "application/zip", Instant.now()));
+ when(storage.generatePresignedUrl(anyString(), any(), anyString())).thenReturn("https://private.invalid/?signed=secret");
+ }
+
+ static Stream paths() {
+ return Stream.of("auto", "private", "review-approved", "review-rejected", "promotion-initial",
+ "promotion-update", "promotion-rejected", "archive", "unarchive", "visibility-narrow",
+ "visibility-expanded", "yank", "download").flatMap(path -> Stream.of(Arguments.of(path, false), Arguments.of(path, true)));
+ }
+
+ @ParameterizedTest(name = "{0}, rollback={1}")
+ @MethodSource("paths")
+ void producerFactIsExactlyDocumentedAndOnlyAfterRealCommit(String path, boolean rollback) {
+ Fixture fixture = fixture(path);
+ List expected = expected(path);
+ new TransactionTemplate(transactions).executeWithoutResult(status -> {
+ switch (path) {
+ case "auto" -> publishing.publishFromEntries(fixture.namespaceSlug(),
+ List.of(new PackageEntry("SKILL.md", "content".getBytes(java.nio.charset.StandardCharsets.UTF_8), 7, "text/markdown")),
+ fixture.owner(), SkillVisibility.PUBLIC, Set.of("SUPER_ADMIN"));
+ case "private" -> confirmations.confirmPublish(fixture.skillId(), fixture.uploadedId(), fixture.owner(), Map.of());
+ case "review-approved" -> reviewing.approveReview(fixture.reviewId(), fixture.reviewer(), "checked", Map.of(), Set.of("SUPER_ADMIN"));
+ case "review-rejected" -> reviewing.rejectReview(fixture.reviewId(), fixture.reviewer(), "private reason", Map.of(), Set.of("SUPER_ADMIN"));
+ case "promotion-initial", "promotion-update" -> promoting.approvePromotion(fixture.promotionId(), fixture.reviewer(), "checked", Set.of("SUPER_ADMIN"), Set.of());
+ case "promotion-rejected" -> promoting.rejectPromotion(fixture.promotionId(), fixture.reviewer(), "private reason", Set.of("SUPER_ADMIN"));
+ case "archive" -> governance.archiveSkill(fixture.skillId(), fixture.owner(), Map.of(), "not-exported-ip", "not-exported-ua", "private reason");
+ case "unarchive" -> governance.unarchiveSkill(fixture.skillId(), fixture.owner(), Map.of(), "not-exported-ip", "not-exported-ua");
+ case "visibility-narrow" -> visibility.change(fixture.skillId(), SkillVisibility.PRIVATE, fixture.owner(), Map.of(), Set.of());
+ case "visibility-expanded" -> reviewing.approveReview(fixture.reviewId(), fixture.reviewer(), "checked", Map.of(), Set.of("SUPER_ADMIN"));
+ case "yank" -> governance.yankVersion(fixture.publishedId(), fixture.reviewer(), "not-exported-ip", "not-exported-ua", "private reason");
+ case "download" -> assertThat(downloads.downloadVersion(fixture.namespaceSlug(), "business-probe", "1.0.0", null, Map.of()).presignedUrl()).contains("signed=secret");
+ default -> throw new AssertionError(path);
+ }
+ assertThat(exporter.records).isEmpty();
+ if (rollback) { status.setRollbackOnly(); }
+ });
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(exporter.records).extracting(r -> r.getAttributes().get(AttributeKey.stringKey("skillhub.event.name")))
+ .containsExactlyInAnyOrderElementsOf(rollback ? List.of() : expected);
+ if (!rollback) {
+ for (var record : exporter.records) {
+ assertThat(record.getAttributes().toString()).doesNotContain("private reason", "not-exported", "secret", "business-probe", "Never export");
+ String actor = record.getAttributes().get(AttributeKey.stringKey("user.id"));
+ if (path.equals("download")) { assertThat(actor).isNull(); }
+ else if (path.startsWith("review") || path.startsWith("promotion") || path.equals("visibility-expanded") || path.equals("yank")) {
+ assertThat(actor).isEqualTo(fixture.reviewer());
+ } else { assertThat(actor).isEqualTo(fixture.owner()); }
+ }
+ }
+ // Persistent state independently proves rollback, not just listener suppression.
+ if (path.equals("private")) {
+ assertThat(((SkillVersionRepository) versions).findById(fixture.uploadedId()).orElseThrow().getStatus())
+ .isEqualTo(rollback ? SkillVersionStatus.UPLOADED : SkillVersionStatus.PUBLISHED);
+ }
+ if (path.equals("review-approved")) {
+ assertThat(((ReviewTaskRepository) reviews).findById(fixture.reviewId()).orElseThrow().getStatus())
+ .isEqualTo(rollback ? ReviewTaskStatus.PENDING : ReviewTaskStatus.APPROVED);
+ }
+ System.out.printf("BUSINESS_LOGS producer=%s rollback=%s received=%d expected=%d%n", path, rollback, exporter.records.size(), rollback ? 0 : expected.size());
+ }
+
+ private List expected(String path) {
+ return switch (path) {
+ case "auto", "private" -> List.of("skill.publication.effective");
+ case "review-approved" -> List.of("skill.publication.effective", "skill.review.approved");
+ case "review-rejected" -> List.of("skill.review.rejected");
+ case "promotion-initial", "promotion-update" -> List.of("skill.publication.effective", "skill.promotion.approved");
+ case "promotion-rejected" -> List.of("skill.promotion.rejected");
+ case "archive", "unarchive" -> List.of("skill.status.changed");
+ case "visibility-narrow", "visibility-expanded" -> List.of("skill.visibility.changed");
+ case "yank" -> List.of("skill.version.yanked");
+ case "download" -> List.of("skill.download.prepared");
+ default -> throw new AssertionError(path);
+ };
+ }
+
+ static Stream unauthorizedPaths() {
+ return Stream.of("auto", "private", "review-approved", "review-rejected", "promotion-initial",
+ "promotion-update", "promotion-rejected", "archive", "unarchive", "visibility-narrow",
+ "visibility-expanded", "yank", "download");
+ }
+
+ @ParameterizedTest(name = "unauthorized {0}")
+ @MethodSource("unauthorizedPaths")
+ void deniedProducerPathsPersistNoMutationAndEmitNoFact(String path) {
+ Fixture fixture = fixture(path);
+ if (path.equals("download")) {
+ new TransactionTemplate(transactions).executeWithoutResult(status -> {
+ Skill skill = ((SkillRepository) skills).findById(fixture.skillId()).orElseThrow();
+ skill.setVisibility(SkillVisibility.PRIVATE);
+ skills.saveAndFlush(skill);
+ });
+ }
+ Skill before = ((SkillRepository) skills).findById(fixture.skillId()).orElseThrow();
+ org.assertj.core.api.Assertions.assertThatThrownBy(() -> {
+ switch (path) {
+ case "auto" -> publishing.publishFromEntries(fixture.namespaceSlug(),
+ List.of(new PackageEntry("SKILL.md", new byte[]{1}, 1, "text/markdown")),
+ fixture.reviewer(), SkillVisibility.PUBLIC, Set.of());
+ case "private" -> confirmations.confirmPublish(fixture.skillId(), fixture.uploadedId(), fixture.reviewer(), Map.of());
+ case "review-approved", "visibility-expanded" -> reviewing.approveReview(fixture.reviewId(), fixture.reviewer(), "checked", Map.of(), Set.of());
+ case "review-rejected" -> reviewing.rejectReview(fixture.reviewId(), fixture.reviewer(), "reason", Map.of(), Set.of());
+ case "promotion-initial", "promotion-update" -> promoting.approvePromotion(fixture.promotionId(), fixture.reviewer(), "checked", Set.of(), Set.of());
+ case "promotion-rejected" -> promoting.rejectPromotion(fixture.promotionId(), fixture.reviewer(), "reason", Set.of());
+ case "archive" -> governance.archiveSkill(fixture.skillId(), fixture.reviewer(), Map.of(), null, null, "reason");
+ case "unarchive" -> governance.unarchiveSkill(fixture.skillId(), fixture.reviewer(), Map.of(), null, null);
+ case "visibility-narrow" -> visibility.change(fixture.skillId(), SkillVisibility.PRIVATE, fixture.reviewer(), Map.of(), Set.of());
+ case "yank" -> governance.yankVersion(before, ((SkillVersionRepository) versions).findById(fixture.publishedId()).orElseThrow(),
+ fixture.reviewer(), Map.of(), null, null, "reason");
+ case "download" -> downloads.downloadVersion(fixture.namespaceSlug(), "business-probe", "1.0.0", null, Map.of());
+ default -> throw new AssertionError(path);
+ }
+ }).isInstanceOfAny(com.iflytek.skillhub.domain.shared.exception.DomainForbiddenException.class,
+ com.iflytek.skillhub.domain.shared.exception.DomainBadRequestException.class);
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(exporter.records).isEmpty();
+ Skill after = ((SkillRepository) skills).findById(fixture.skillId()).orElseThrow();
+ assertThat(after.getStatus()).isEqualTo(before.getStatus());
+ assertThat(after.getVisibility()).isEqualTo(before.getVisibility());
+ assertThat(((SkillVersionRepository) versions).findById(fixture.publishedId()).orElseThrow().getStatus()).isEqualTo(SkillVersionStatus.PUBLISHED);
+ if (fixture.reviewId() != null) {
+ assertThat(((ReviewTaskRepository) reviews).findById(fixture.reviewId()).orElseThrow().getStatus()).isEqualTo(ReviewTaskStatus.PENDING);
+ }
+ if (fixture.promotionId() != null) {
+ assertThat(((PromotionRequestRepository) promotions).findById(fixture.promotionId()).orElseThrow().getStatus()).isEqualTo(ReviewTaskStatus.PENDING);
+ }
+ System.out.printf("BUSINESS_LOGS denied producer=%s received=0; persisted state unchanged%n", path);
+ }
+
+ private Fixture fixture(String path) {
+ return new TransactionTemplate(transactions).execute(status -> {
+ String suffix = UUID.randomUUID().toString().substring(0, 8);
+ String owner = "business-owner-" + suffix;
+ String reviewer = "business-reviewer-" + suffix;
+ users.saveAndFlush(new UserAccount(owner, owner, null, null));
+ users.saveAndFlush(new UserAccount(reviewer, reviewer, null, null));
+ Namespace namespace = namespaces.saveAndFlush(new Namespace("business-" + suffix, "Never export namespace", owner));
+ members.saveAndFlush(new NamespaceMember(namespace.getId(), owner, NamespaceRole.OWNER));
+ SkillVisibility scope = path.equals("private") || path.equals("visibility-expanded") ? SkillVisibility.PRIVATE : SkillVisibility.PUBLIC;
+ Skill skill = skills.saveAndFlush(new Skill(namespace.getId(), "business-probe", owner, scope));
+ SkillVersion published = new SkillVersion(skill.getId(), "1.0.0", owner);
+ published.setStatus(SkillVersionStatus.PUBLISHED);
+ published.setBundleReady(true);
+ published.setDownloadReady(true);
+ published.setPublishedAt(Instant.now());
+ published = versions.saveAndFlush(published);
+ skill.setLatestVersionId(published.getId());
+ if (path.equals("unarchive")) { skill.setStatus(SkillStatus.ARCHIVED); }
+ skills.saveAndFlush(skill);
+ SkillVersion uploaded = new SkillVersion(skill.getId(), "2.0.0", owner);
+ uploaded.setStatus(path.startsWith("review") ? SkillVersionStatus.PENDING_REVIEW : SkillVersionStatus.UPLOADED);
+ uploaded = versions.saveAndFlush(uploaded);
+ Long reviewId = null;
+ if (path.startsWith("review")) {
+ reviewId = reviews.saveAndFlush(new ReviewTask(uploaded.getId(), skill.getId(), namespace.getId(), uploaded.getVersion(), owner)).getId();
+ }
+ if (path.equals("visibility-expanded")) {
+ SecurityAudit scan = new SecurityAudit(published.getId(), ScannerType.SKILL_SCANNER, "scan-" + suffix);
+ scan.setIsSafe(true); scan.setScannedAt(Instant.now()); audits.saveAndFlush(scan);
+ var task = ReviewTask.forVisibility(skill, SkillVisibility.PUBLIC, owner);
+ task.setVisibilityScanTaskId(scan.getTaskId());
+ reviewId = reviews.saveAndFlush(task).getId();
+ }
+ Long promotionId = null;
+ if (path.startsWith("promotion")) {
+ Namespace targetNamespace = new Namespace("global-" + suffix, "Never export global name", owner);
+ targetNamespace.setType(NamespaceType.GLOBAL);
+ targetNamespace = namespaces.saveAndFlush(targetNamespace);
+ PromotionRequest request = new PromotionRequest(skill.getId(), published.getId(), targetNamespace.getId(), owner);
+ if (path.equals("promotion-update")) {
+ Skill target = new Skill(targetNamespace.getId(), skill.getSlug(), owner, SkillVisibility.PUBLIC);
+ target.setSourceSkillId(skill.getId()); target = skills.saveAndFlush(target);
+ SkillVersion old = new SkillVersion(target.getId(), "0.9.0", owner);
+ old.setStatus(SkillVersionStatus.PUBLISHED); old = versions.saveAndFlush(old);
+ target.setLatestVersionId(old.getId()); skills.saveAndFlush(target);
+ PromotionRequest initial = new PromotionRequest(skill.getId(), published.getId(), targetNamespace.getId(), owner);
+ initial.setStatus(ReviewTaskStatus.APPROVED); initial.setTargetSkillId(target.getId()); initial.setTargetVersionId(old.getId());
+ promotions.saveAndFlush(initial);
+ request.setRequestKind(PromotionRequestKind.UPDATE); request.setTargetSkillId(target.getId());
+ }
+ promotionId = promotions.saveAndFlush(request).getId();
+ }
+ return new Fixture(skill.getId(), published.getId(), uploaded.getId(), reviewId, promotionId, namespace.getSlug(), owner, reviewer);
+ });
+ }
+
+ private record Fixture(Long skillId, Long publishedId, Long uploadedId, Long reviewId, Long promotionId,
+ String namespaceSlug, String owner, String reviewer) {}
+}
diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTransactionTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTransactionTest.java
new file mode 100644
index 000000000..bd07d0426
--- /dev/null
+++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTransactionTest.java
@@ -0,0 +1,234 @@
+package com.iflytek.skillhub.observability.business;
+
+import com.iflytek.skillhub.domain.event.*;
+import com.iflytek.skillhub.domain.namespace.Namespace;
+import com.iflytek.skillhub.domain.namespace.NamespaceRepository;
+import com.iflytek.skillhub.domain.skill.*;
+import com.iflytek.skillhub.domain.skill.service.SkillDownloadService;
+import com.iflytek.skillhub.domain.skill.service.SkillSlugResolutionService;
+import com.iflytek.skillhub.domain.shared.exception.DomainForbiddenException;
+import com.iflytek.skillhub.storage.ObjectStorageService;
+import com.iflytek.skillhub.storage.ObjectMetadata;
+import com.iflytek.skillhub.domain.skill.SkillStatus;
+import com.iflytek.skillhub.domain.skill.SkillVisibility;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.TraceFlags;
+import io.opentelemetry.api.trace.TraceState;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.logs.export.LogRecordExporter;
+import io.opentelemetry.sdk.resources.Resource;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DriverManagerDataSource;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.testcontainers.containers.PostgreSQLContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.*;
+
+@Testcontainers
+class BusinessLogsTransactionTest {
+ @Container
+ static final PostgreSQLContainer POSTGRES = new PostgreSQLContainer<>("postgres:16-alpine")
+ .withTmpFs(java.util.Map.of("/var/lib/postgresql/data", "rw"))
+ .withCreateContainerCmdModifier(cmd -> cmd.withName("skillhub-1009-business-transactions"));
+
+ private final BusinessEventSnapshot snapshot = new BusinessEventSnapshot(7L, SkillVisibility.PRIVATE,
+ "approver", "user", "review", Instant.parse("2026-10-01T00:00:00Z"));
+
+ @Test
+ void realPostgresCommitEmitsAndRollbackDoesNotForEveryIncludedFact() {
+ var dataSource = new DriverManagerDataSource(POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword());
+ var jdbc = new JdbcTemplate(dataSource);
+ jdbc.execute("create table if not exists analytics_transaction_probe (id int primary key)");
+ var transaction = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
+ var exporter = new CapturingExporter();
+ var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry());
+ var properties = new BusinessLogsProperties();
+ try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics);
+ var context = new AnnotationConfigApplicationContext()) {
+ context.registerBean(BusinessEventListener.class,
+ () -> new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics));
+ context.refresh();
+ var events = List.of(
+ new SkillPublishedEvent(1L, 2L, "approver", snapshot),
+ new SkillDownloadedEvent(1L, 2L, snapshot),
+ new ReviewApprovedEvent(3L, 1L, 2L, "approver", "owner", snapshot),
+ new ReviewRejectedEvent(3L, 1L, 2L, "approver", "owner", "secret", snapshot),
+ new PromotionApprovedEvent(4L, 1L, "approver", "owner", snapshot),
+ new PromotionRejectedEvent(4L, 1L, "approver", "owner", "secret", snapshot),
+ new SkillStatusChangedEvent(1L, SkillStatus.ACTIVE, SkillStatus.ARCHIVED, snapshot),
+ new SkillVisibilityChangedEvent(1L, SkillVisibility.PUBLIC, SkillVisibility.PRIVATE, snapshot),
+ new SkillVersionYankedEvent(1L, 2L, "approver", true, snapshot));
+ transaction.executeWithoutResult(status -> {
+ jdbc.update("insert into analytics_transaction_probe values (1) on conflict do nothing");
+ events.forEach(context::publishEvent);
+ assertThat(metrics.enqueued.count()).isZero();
+ });
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(exporter.records).hasSize(9);
+ assertThat(exporter.records).extracting(r -> r.getAttributes().get(AttributeKey.stringKey("skillhub.event.name")))
+ .containsExactlyInAnyOrderElementsOf(BusinessEventMapper.EVENT_NAMES);
+ transaction.executeWithoutResult(status -> {
+ jdbc.update("insert into analytics_transaction_probe values (2)");
+ events.forEach(context::publishEvent);
+ status.setRollbackOnly();
+ });
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(jdbc.queryForObject("select count(*) from analytics_transaction_probe where id=2", Integer.class)).isZero();
+ assertThat(exporter.records).hasSize(9);
+ context.publishEvent(new SkillDownloadedEvent(1L, 2L, snapshot));
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(exporter.records).hasSize(10);
+ assertThat(exporter.records.getLast().getTimestampEpochNanos()).isEqualTo(snapshot.occurredAt().getEpochSecond() * 1_000_000_000L);
+ System.out.println("BUSINESS_LOGS postgres: 9 commit facts, 0 rollback facts, 1 nontransactional fact");
+ }
+ }
+
+ @Test
+ void unsampledSpanCapturedBeforeCommitAndNoMdcOrBaggageExported() {
+ var dataSource = new DriverManagerDataSource(POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword());
+ var transaction = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
+ var exporter = new CapturingExporter();
+ var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry());
+ var properties = new BusinessLogsProperties();
+ var spanContext = SpanContext.create("12345678901234567890123456789012", "1234567890123456", TraceFlags.getDefault(), TraceState.getDefault());
+ try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) {
+ var listener = new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics);
+ transaction.executeWithoutResult(status -> {
+ try (var ignored = Span.wrap(spanContext).makeCurrent()) {
+ listener.onEvent(new SkillPublishedEvent(1L, 2L, "approver", snapshot));
+ }
+ assertThat(Span.current().getSpanContext().isValid()).isFalse();
+ });
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(exporter.records).hasSize(1);
+ assertThat(exporter.records.getFirst().getSpanContext()).isEqualTo(spanContext);
+ assertThat(exporter.records.getFirst().getSpanContext().isSampled()).isFalse();
+ }
+ }
+
+ @Test
+ void saturationDoesNotWaitForReceiverAndShutdownIsBounded() throws Exception {
+ var properties = new BusinessLogsProperties();
+ properties.setMaxQueueSize(2);
+ properties.setMaxExportBatchSize(1);
+ properties.setScheduleDelay(Duration.ofMillis(10));
+ properties.setShutdownTimeout(Duration.ofMillis(100));
+ var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry());
+ var entered = new CountDownLatch(1);
+ var pending = new CompletableResultCode();
+ var exporter = new CapturingExporter() {
+ @Override public CompletableResultCode export(Collection records) { entered.countDown(); return pending; }
+ @Override public CompletableResultCode shutdown() { pending.fail(); return CompletableResultCode.ofSuccess(); }
+ };
+ var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics);
+ var listener = new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics);
+ try {
+ listener.onEvent(new SkillPublishedEvent(1L, 2L, "approver", snapshot));
+ assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue();
+ long start = System.nanoTime();
+ for (int i = 0; i < 1000; i++) { listener.onEvent(new SkillPublishedEvent(1L, 2L, "approver", snapshot)); }
+ long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ assertThat(elapsed).isLessThan(1000);
+ assertThat(metrics.enqueued.count()).isEqualTo(2);
+ assertThat(metrics.queueFull.count()).isEqualTo(999);
+ start = System.nanoTime();
+ pipeline.close();
+ long shutdown = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ assertThat(shutdown).isLessThan(1000);
+ System.out.printf("BUSINESS_LOGS saturation: 1000 caller events in %d ms, drops=999, shutdown=%d ms%n", elapsed, shutdown);
+ } finally { pipeline.close(); pending.fail(); }
+ }
+
+ @Test
+ void mapperAndExporterExceptionsCannotEscapeBusinessThread() {
+ var properties = new BusinessLogsProperties();
+ var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry());
+ var exporter = new CapturingExporter() {
+ @Override public CompletableResultCode export(Collection records) { throw new IllegalStateException("must not be logged"); }
+ };
+ try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) {
+ var listener = new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics);
+ assertThatCode(() -> listener.onEvent(null)).doesNotThrowAnyException();
+ assertThat(metrics.errors.count()).isEqualTo(1);
+ assertThatCode(() -> listener.onEvent(new SkillPublishedEvent(1L, 2L, "approver", snapshot))).doesNotThrowAnyException();
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(metrics.exportFailures.count()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void actualNontransactionalDownloadProducerEmitsOnceAfterPermissionsAndNeverForReviewOrDeniedAccess() {
+ var properties = new BusinessLogsProperties();
+ var exporter = new CapturingExporter();
+ var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry());
+ try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics);
+ var context = new AnnotationConfigApplicationContext()) {
+ context.registerBean(BusinessEventListener.class,
+ () -> new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics));
+ context.refresh();
+ var namespaces = mock(NamespaceRepository.class);
+ var skills = mock(SkillRepository.class);
+ var versions = mock(SkillVersionRepository.class);
+ var storage = mock(ObjectStorageService.class);
+ var skill = new Skill(7L, "never-export-this-slug", "owner", SkillVisibility.PUBLIC);
+ org.springframework.test.util.ReflectionTestUtils.setField(skill, "id", 1L);
+ skill.setLatestVersionId(2L);
+ var version = new SkillVersion(1L, "1.0.0", "owner");
+ org.springframework.test.util.ReflectionTestUtils.setField(version, "id", 2L);
+ version.setStatus(SkillVersionStatus.PUBLISHED);
+ version.setDownloadReady(true);
+ var namespace = new Namespace("test", "Never export namespace name", "owner");
+ org.springframework.test.util.ReflectionTestUtils.setField(namespace, "id", 7L);
+ when(namespaces.findBySlug("test")).thenReturn(java.util.Optional.of(namespace));
+ when(skills.findByNamespaceIdAndSlug(7L, skill.getSlug())).thenReturn(List.of(skill));
+ when(versions.findById(2L)).thenReturn(java.util.Optional.of(version));
+ when(storage.exists("packages/1/2/bundle.zip")).thenReturn(true);
+ when(storage.getMetadata("packages/1/2/bundle.zip")).thenReturn(new ObjectMetadata(12, "application/zip", Instant.now()));
+ when(storage.generatePresignedUrl(anyString(), any(), anyString())).thenReturn("https://secret-signed-url.invalid/?token=not-exported");
+ var service = new SkillDownloadService(namespaces, skills, versions, mock(SkillVersionStatsRepository.class),
+ mock(SkillFileRepository.class), mock(SkillTagRepository.class), storage, new VisibilityChecker(),
+ context, new SkillSlugResolutionService(skills));
+ var response = service.downloadLatest("test", skill.getSlug(), null, java.util.Map.of());
+ assertThat(response.presignedUrl()).contains("secret-signed-url");
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(exporter.records).hasSize(1);
+ assertThat(exporter.records.getFirst().getAttributes().get(AttributeKey.stringKey("user.id"))).isNull();
+ assertThat(exporter.records.getFirst().getAttributes().toString()).doesNotContain("not-exported", skill.getSlug());
+ service.downloadReviewVersion(skill, version);
+ skill.setVisibility(SkillVisibility.PRIVATE);
+ assertThatThrownBy(() -> service.downloadLatest("test", skill.getSlug(), null, java.util.Map.of()))
+ .isInstanceOf(DomainForbiddenException.class);
+ assertThatThrownBy(() -> service.downloadLatest("test", skill.getSlug(), "stranger", java.util.Map.of()))
+ .isInstanceOf(DomainForbiddenException.class);
+ pipeline.forceFlush().join(5, TimeUnit.SECONDS);
+ assertThat(exporter.records).hasSize(1);
+ System.out.println("BUSINESS_LOGS real download service: anonymous prepared=1, review/anonymous-private/unauthorized-private=0");
+ }
+ }
+
+ static class CapturingExporter implements LogRecordExporter {
+ final List records = new CopyOnWriteArrayList<>();
+ @Override public CompletableResultCode export(Collection records) { this.records.addAll(records); return CompletableResultCode.ofSuccess(); }
+ @Override public CompletableResultCode flush() { return CompletableResultCode.ofSuccess(); }
+ @Override public CompletableResultCode shutdown() { return CompletableResultCode.ofSuccess(); }
+ }
+}