mirror of
https://github.com/iflytek/skillhub.git
synced 2026-11-14 09:01:16 +00:00
feat(observability): export opt-in business Logs over OTLP
Signed-off-by: 千乘妍 (Xiaoyaner) <258399167+xiaoyaner0201@users.noreply.github.com>
This commit is contained in:
千乘妍 (Xiaoyaner)
parent
76ea586719
commit
244ae51ac5
17 changed files with 1508 additions and 0 deletions
20
docs/examples/business-event-logs/collector.yaml
Normal file
20
docs/examples/business-event-logs/collector.yaml
Normal file
|
|
@ -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]
|
||||
14
docs/examples/business-event-logs/compose.yaml
Normal file
14
docs/examples/business-event-logs/compose.yaml
Normal file
|
|
@ -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
|
||||
|
||||
|
|
@ -59,6 +59,10 @@
|
|||
<groupId>io.opentelemetry</groupId>
|
||||
<artifactId>opentelemetry-exporter-otlp</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.opentelemetry</groupId>
|
||||
<artifactId>opentelemetry-sdk-logs</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.skywalking</groupId>
|
||||
<artifactId>apm-toolkit-logback-1.x</artifactId>
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String> 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<String> SOURCES = Set.of("auto_publish", "private_confirm", "review", "promotion",
|
||||
"download", "governance", "visibility");
|
||||
private final Set<String> events;
|
||||
private final Set<Long> namespaces;
|
||||
|
||||
BusinessEventMapper(BusinessLogsProperties properties) {
|
||||
events = Set.copyOf(properties.getEventAllowlist());
|
||||
namespaces = Set.copyOf(properties.getNamespaceAllowlist());
|
||||
}
|
||||
|
||||
public Optional<MappedRecord> 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) {}
|
||||
}
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<LogRecordData> 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(); }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String> eventAllowlist = BusinessEventMapper.EVENT_NAMES;
|
||||
private Set<Long> 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<String, String> parsedHeaders() {
|
||||
try {
|
||||
if (headers == null || headers.isBlank()) { return Map.of(); }
|
||||
if (headers.length() > 8192) { throw new IllegalArgumentException(); }
|
||||
Map<String, String> 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<String> getEventAllowlist() { return eventAllowlist; }
|
||||
public void setEventAllowlist(Set<String> eventAllowlist) { this.eventAllowlist = eventAllowlist; }
|
||||
public Set<Long> getNamespaceAllowlist() { return namespaceAllowlist; }
|
||||
public void setNamespaceAllowlist(Set<Long> namespaceAllowlist) { this.namespaceAllowlist = namespaceAllowlist; }
|
||||
}
|
||||
|
|
@ -0,0 +1,2 @@
|
|||
/** Opt-in, independent, lossy business analytics adapter; never an audit or application-log bridge. */
|
||||
package com.iflytek.skillhub.observability.business;
|
||||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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.<SkillBusinessEvent>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<JsonNode> records = records(raw);
|
||||
assertThat(records).hasSize(27);
|
||||
var names = new HashSet<String>();
|
||||
var uids = new HashSet<String>();
|
||||
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<JsonNode> records(String raw) throws Exception {
|
||||
List<JsonNode> 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<String, JsonNode> attributes(JsonNode record) {
|
||||
var result = new java.util.HashMap<String, JsonNode>();
|
||||
for (JsonNode attr : record.path("attributes")) { result.put(attr.path("key").asText(), attr.path("value").elements().next()); }
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -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<byte[]> bodies = new CopyOnWriteArrayList<>();
|
||||
List<String> 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); }
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Arguments> 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<String> 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<String> 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<String> 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) {}
|
||||
}
|
||||
|
|
@ -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.<SkillBusinessEvent>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<LogRecordData> 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<LogRecordData> 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<LogRecordData> records = new CopyOnWriteArrayList<>();
|
||||
@Override public CompletableResultCode export(Collection<LogRecordData> records) { this.records.addAll(records); return CompletableResultCode.ofSuccess(); }
|
||||
@Override public CompletableResultCode flush() { return CompletableResultCode.ofSuccess(); }
|
||||
@Override public CompletableResultCode shutdown() { return CompletableResultCode.ofSuccess(); }
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue