diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/ActorType.java b/api/all/src/main/java/io/opentelemetry/api/audit/ActorType.java new file mode 100644 index 00000000000..c6458f170d0 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/ActorType.java @@ -0,0 +1,23 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +/** + * Classifies the kind of entity that performed an auditable action. + * + * @see AuditRecordBuilder#setActorType(ActorType) + */ +public enum ActorType { + + /** A human user, identified by a user account. */ + USER, + + /** An automated service, daemon, or service account. */ + SERVICE, + + /** The operating system or a privileged system component. */ + SYSTEM +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/AuditDeliveryException.java b/api/all/src/main/java/io/opentelemetry/api/audit/AuditDeliveryException.java new file mode 100644 index 00000000000..8b30f2c9009 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/AuditDeliveryException.java @@ -0,0 +1,25 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +/** + * Hard-error thrown by {@link AuditRecordBuilder#emit()} when the audit sink cannot be reached and + * all retries are exhausted. This is an unchecked exception so that audit-logging call sites remain + * clean, but callers SHOULD catch it and escalate the failure through their incident-management + * process. + */ +public final class AuditDeliveryException extends RuntimeException { + + private static final long serialVersionUID = 1L; + + public AuditDeliveryException(String message) { + super(message); + } + + public AuditDeliveryException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/AuditLogger.java b/api/all/src/main/java/io/opentelemetry/api/audit/AuditLogger.java new file mode 100644 index 00000000000..e345e56e7f9 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/AuditLogger.java @@ -0,0 +1,25 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +import javax.annotation.concurrent.ThreadSafe; + +/** + * The entry point for emitting audit records. + * + *

Obtain an {@link AuditRecordBuilder} via {@link #auditRecordBuilder()}, populate all required + * fields, and call {@link AuditRecordBuilder#emit()} to deliver the record to the audit sink. + * + *

Unlike {@link io.opentelemetry.api.logs.Logger}, this interface does not expose an + * {@code isEnabled} check: audit records are ALWAYS emitted. Dropping audit records is prohibited + * by the audit logging specification. + */ +@ThreadSafe +public interface AuditLogger { + + /** Returns an {@link AuditRecordBuilder} for constructing and emitting an audit record. */ + AuditRecordBuilder auditRecordBuilder(); +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/AuditLoggerBuilder.java b/api/all/src/main/java/io/opentelemetry/api/audit/AuditLoggerBuilder.java new file mode 100644 index 00000000000..b2ea7437c76 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/AuditLoggerBuilder.java @@ -0,0 +1,28 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +/** + * Builder for creating named {@link AuditLogger} instances. + * + *

The {@code name} (set on the owning {@link AuditProvider}) is stored as a diagnostic label on + * the logger. Unlike {@link io.opentelemetry.api.logs.LoggerBuilder}, the name is NOT mapped to an + * OTLP {@code InstrumentationScope}. + */ +public interface AuditLoggerBuilder { + + /** + * Sets the schema URL to be recorded on emitted {@link AuditRecordBuilder}s for semantic + * convention versioning. + */ + AuditLoggerBuilder setSchemaUrl(String schemaUrl); + + /** Sets the version of the component or library that is emitting audit records. */ + AuditLoggerBuilder setInstrumentationVersion(String instrumentationVersion); + + /** Returns the configured {@link AuditLogger}. */ + AuditLogger build(); +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/AuditProvider.java b/api/all/src/main/java/io/opentelemetry/api/audit/AuditProvider.java new file mode 100644 index 00000000000..125174064da --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/AuditProvider.java @@ -0,0 +1,48 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +import javax.annotation.concurrent.ThreadSafe; + +/** + * The entry point of the Audit Logging API. Provides named {@link AuditLogger} instances. + * + *

The provider is expected to be accessed from a central place. Use {@link + * GlobalAuditProvider#get()} to obtain the globally registered instance, or create an {@link + * AuditProvider} directly via the SDK. + * + *

When no SDK is installed, {@link #noop()} returns an {@link AuditProvider} whose loggers emit + * no-op receipts without error. + */ +@ThreadSafe +public interface AuditProvider { + + /** + * Gets or creates a named {@link AuditLogger} instance. + * + * @param name A string identifying the component or subsystem emitting audit records (for example + * {@code "com.example.auth"}). MUST NOT be empty. + */ + default AuditLogger get(String name) { + return auditLoggerBuilder(name).build(); + } + + /** + * Creates an {@link AuditLoggerBuilder} for a named {@link AuditLogger}. + * + * @param name A string identifying the component or subsystem emitting audit records. MUST NOT be + * empty. + */ + AuditLoggerBuilder auditLoggerBuilder(String name); + + /** + * Returns a no-op {@link AuditProvider} whose loggers return no-op {@link AuditReceipt}s + * immediately without error. + */ + static AuditProvider noop() { + return DefaultAuditProvider.getInstance(); + } +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/AuditReceipt.java b/api/all/src/main/java/io/opentelemetry/api/audit/AuditReceipt.java new file mode 100644 index 00000000000..74ec7a4c3b6 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/AuditReceipt.java @@ -0,0 +1,87 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +import javax.annotation.concurrent.Immutable; + +/** + * Proof-of-delivery returned by {@link AuditLogger} once the audit sink has persisted the record. + * + *

The {@code recordId} echoes the caller's {@link AuditRecordBuilder#setRecordId(String)}. + * {@code integrityHash} is the SHA-256 of the record as written by the sink. {@code + * sinkTimestampEpochNanos} is the nanosecond UNIX epoch at which the sink persisted the record. + */ +@Immutable +public final class AuditReceipt { + + private final String recordId; + private final String integrityHash; + private final long sinkTimestampEpochNanos; + + private AuditReceipt(String recordId, String integrityHash, long sinkTimestampEpochNanos) { + this.recordId = recordId; + this.integrityHash = integrityHash; + this.sinkTimestampEpochNanos = sinkTimestampEpochNanos; + } + + /** Creates an {@link AuditReceipt} with the given fields. */ + public static AuditReceipt create( + String recordId, String integrityHash, long sinkTimestampEpochNanos) { + return new AuditReceipt(recordId, integrityHash, sinkTimestampEpochNanos); + } + + /** Returns the {@code RecordId} echoed from the corresponding {@link AuditRecordBuilder}. */ + public String recordId() { + return recordId; + } + + /** + * Returns the SHA-256 hex digest of the canonical serialization of the {@code AuditRecord} as + * persisted by the audit sink. + */ + public String integrityHash() { + return integrityHash; + } + + /** Returns the nanosecond UNIX epoch at which the audit sink persisted the record. */ + public long sinkTimestampEpochNanos() { + return sinkTimestampEpochNanos; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (!(obj instanceof AuditReceipt)) { + return false; + } + AuditReceipt other = (AuditReceipt) obj; + return recordId.equals(other.recordId) + && integrityHash.equals(other.integrityHash) + && sinkTimestampEpochNanos == other.sinkTimestampEpochNanos; + } + + @Override + public int hashCode() { + int result = recordId.hashCode(); + result = 31 * result + integrityHash.hashCode(); + result = 31 * result + Long.hashCode(sinkTimestampEpochNanos); + return result; + } + + @Override + public String toString() { + return "AuditReceipt{" + + "recordId=" + + recordId + + ", integrityHash=" + + integrityHash + + ", sinkTimestampEpochNanos=" + + sinkTimestampEpochNanos + + "}"; + } +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/AuditRecordBuilder.java b/api/all/src/main/java/io/opentelemetry/api/audit/AuditRecordBuilder.java new file mode 100644 index 00000000000..9879ce78111 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/AuditRecordBuilder.java @@ -0,0 +1,163 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Value; +import java.time.Instant; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; + +/** + * Used to construct and emit {@link AuditReceipt}-returning audit records from an {@link + * AuditLogger}. + * + *

Obtain an {@link AuditLogger#auditRecordBuilder()}, set all required and desired optional + * fields, then call {@link #emit()} which blocks until the audit sink acknowledges the record and + * returns an {@link AuditReceipt} as proof-of-delivery. + * + *

Unlike {@link io.opentelemetry.api.logs.LogRecordBuilder}, {@code emit()} returns a non-void + * {@link AuditReceipt} and MUST NOT silently drop the record. An exception is raised if the sink + * cannot be reached within the configured timeout. + */ +public interface AuditRecordBuilder { + + // ── Required fields ────────────────────────────────────────────────────── + + /** + * Sets the caller-generated unique identifier for this record. If not set, the SDK MUST generate + * a UUID v4. The value MUST remain stable across retries of the same event. + */ + AuditRecordBuilder setRecordId(String recordId); + + /** + * Sets the epoch timestamp (event time) using the given value and unit. + * + *

This field is required. It represents the time at which the auditable action occurred. + */ + AuditRecordBuilder setTimestamp(long timestamp, TimeUnit unit); + + /** Sets the epoch timestamp (event time) using the given {@link Instant}. */ + AuditRecordBuilder setTimestamp(Instant instant); + + /** + * Sets the semantic name that uniquely identifies the type of audit event, e.g. {@code + * "user.login.success"}. MUST be non-empty and stable across releases. + */ + AuditRecordBuilder setEventName(String eventName); + + /** + * Sets the identity of the entity that performed the auditable action. + * + *

MAY be a structured value. If the actor cannot be determined, set to a sentinel such as + * {@code "anonymous"}. + */ + AuditRecordBuilder setActor(Value actor); + + /** Convenience overload of {@link #setActor(Value)} accepting a plain string. */ + default AuditRecordBuilder setActor(String actor) { + return setActor(Value.of(actor)); + } + + /** Sets the type of the actor. */ + AuditRecordBuilder setActorType(ActorType actorType); + + /** + * Sets the verb that describes what the actor did, e.g. {@code "LOGIN"}, {@code "READ"}, {@code + * "DELETE"}. MUST be non-empty and stable across releases. + */ + AuditRecordBuilder setAction(String action); + + /** Sets the result of the auditable action. */ + AuditRecordBuilder setOutcome(Outcome outcome); + + // ── Optional fields ─────────────────────────────────────────────────────── + + /** + * Sets the epoch observed-timestamp using the given value and unit. If not set, the SDK MUST set + * this to the wall-clock time at the moment {@link #emit()} is called. + */ + AuditRecordBuilder setObservedTimestamp(long timestamp, TimeUnit unit); + + /** Sets the epoch observed-timestamp using the given {@link Instant}. */ + AuditRecordBuilder setObservedTimestamp(Instant instant); + + /** Sets the schema version of the audit payload, e.g. {@code "1.0.0"}. */ + AuditRecordBuilder setSchemaVersion(String schemaVersion); + + /** + * Sets the object upon which the action was performed, e.g. a file path, database row, or + * structured resource descriptor. + */ + AuditRecordBuilder setTargetResource(Value targetResource); + + /** Sets the source network address of the auditable action, e.g. {@code "203.0.113.42"}. */ + AuditRecordBuilder setSourceIp(String sourceIp); + + /** Sets free-form additional information about the audit event. */ + AuditRecordBuilder setBody(Value body); + + /** Convenience overload of {@link #setBody(Value)} accepting a plain string. */ + default AuditRecordBuilder setBody(String body) { + return setBody(Value.of(body)); + } + + /** + * Sets an attribute on this record. If the record already contains a mapping for the key, the + * old value is replaced. + * + *

Providing a {@code null} value is a no-op and does not remove previously set values. + */ + AuditRecordBuilder setAttribute(AttributeKey key, @Nullable T value); + + /** + * Sets an asymmetric digital signature over the canonical serialization of this record and the + * algorithm used (e.g. {@code "ES256"}). MUST NOT be set together with {@link + * #setHmac(byte[], String)}. + */ + AuditRecordBuilder setSignature(byte[] signature, String algorithm); + + /** + * Sets the DER-encoded X.509 public-key certificate corresponding to the signing key. Only + * meaningful when {@link #setSignature(byte[], String)} is also set. + */ + AuditRecordBuilder setCertificate(byte[] certificate); + + /** + * Sets a symmetric HMAC over the canonical serialization of this record and the algorithm used + * (e.g. {@code "HMAC-SHA256"}). MUST NOT be set together with {@link + * #setSignature(byte[], String)}. + */ + AuditRecordBuilder setHmac(byte[] hmac, String algorithm); + + /** + * Sets the monotonically increasing sequence number for hash-chain continuity. When set, + * receivers can detect gaps that indicate lost or deleted records. + */ + AuditRecordBuilder setSequenceNo(long sequenceNo); + + /** + * Sets the {@code IntegrityHash} of the immediately preceding record in the same audit stream, + * enabling hash-chain validation. + */ + AuditRecordBuilder setPrevHash(String prevHash); + + // ── Terminal ────────────────────────────────────────────────────────────── + + /** + * Emits the audit record and blocks until the audit sink acknowledges receipt. + * + *

Returns an {@link AuditReceipt} containing the sink-assigned {@code RecordId}, {@code + * IntegrityHash}, and {@code SinkTimestamp}. + * + *

If the sink cannot be reached within the configured timeout and the retry budget is + * exhausted, this method MUST throw a runtime exception and MUST NOT return silently. + * + * @throws AuditDeliveryException if the audit sink cannot be reached and all retries are + * exhausted + */ + AuditReceipt emit(); +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/DefaultAuditLogger.java b/api/all/src/main/java/io/opentelemetry/api/audit/DefaultAuditLogger.java new file mode 100644 index 00000000000..162d4288323 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/DefaultAuditLogger.java @@ -0,0 +1,140 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Value; +import java.time.Instant; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; + +class DefaultAuditLogger implements AuditLogger { + + private static final AuditLogger INSTANCE = new DefaultAuditLogger(); + private static final AuditRecordBuilder NOOP_BUILDER = new NoopAuditRecordBuilder(); + private static final AuditReceipt NOOP_RECEIPT = AuditReceipt.create("", "", 0); + + private DefaultAuditLogger() {} + + static AuditLogger getInstance() { + return INSTANCE; + } + + @Override + public AuditRecordBuilder auditRecordBuilder() { + return NOOP_BUILDER; + } + + private static final class NoopAuditRecordBuilder implements AuditRecordBuilder { + + private NoopAuditRecordBuilder() {} + + @Override + public AuditRecordBuilder setRecordId(String recordId) { + return this; + } + + @Override + public AuditRecordBuilder setTimestamp(long timestamp, TimeUnit unit) { + return this; + } + + @Override + public AuditRecordBuilder setTimestamp(Instant instant) { + return this; + } + + @Override + public AuditRecordBuilder setEventName(String eventName) { + return this; + } + + @Override + public AuditRecordBuilder setActor(Value actor) { + return this; + } + + @Override + public AuditRecordBuilder setActorType(ActorType actorType) { + return this; + } + + @Override + public AuditRecordBuilder setAction(String action) { + return this; + } + + @Override + public AuditRecordBuilder setOutcome(Outcome outcome) { + return this; + } + + @Override + public AuditRecordBuilder setObservedTimestamp(long timestamp, TimeUnit unit) { + return this; + } + + @Override + public AuditRecordBuilder setObservedTimestamp(Instant instant) { + return this; + } + + @Override + public AuditRecordBuilder setSchemaVersion(String schemaVersion) { + return this; + } + + @Override + public AuditRecordBuilder setTargetResource(Value targetResource) { + return this; + } + + @Override + public AuditRecordBuilder setSourceIp(String sourceIp) { + return this; + } + + @Override + public AuditRecordBuilder setBody(Value body) { + return this; + } + + @Override + public AuditRecordBuilder setAttribute(AttributeKey key, @Nullable T value) { + return this; + } + + @Override + public AuditRecordBuilder setSignature(byte[] signature, String algorithm) { + return this; + } + + @Override + public AuditRecordBuilder setCertificate(byte[] certificate) { + return this; + } + + @Override + public AuditRecordBuilder setHmac(byte[] hmac, String algorithm) { + return this; + } + + @Override + public AuditRecordBuilder setSequenceNo(long sequenceNo) { + return this; + } + + @Override + public AuditRecordBuilder setPrevHash(String prevHash) { + return this; + } + + @Override + public AuditReceipt emit() { + return NOOP_RECEIPT; + } + } +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/DefaultAuditProvider.java b/api/all/src/main/java/io/opentelemetry/api/audit/DefaultAuditProvider.java new file mode 100644 index 00000000000..e42810c6627 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/DefaultAuditProvider.java @@ -0,0 +1,41 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +class DefaultAuditProvider implements AuditProvider { + + private static final AuditProvider INSTANCE = new DefaultAuditProvider(); + private static final AuditLoggerBuilder NOOP_BUILDER = new NoopAuditLoggerBuilder(); + + private DefaultAuditProvider() {} + + static AuditProvider getInstance() { + return INSTANCE; + } + + @Override + public AuditLoggerBuilder auditLoggerBuilder(String name) { + return NOOP_BUILDER; + } + + private static class NoopAuditLoggerBuilder implements AuditLoggerBuilder { + + @Override + public AuditLoggerBuilder setSchemaUrl(String schemaUrl) { + return this; + } + + @Override + public AuditLoggerBuilder setInstrumentationVersion(String instrumentationVersion) { + return this; + } + + @Override + public AuditLogger build() { + return DefaultAuditLogger.getInstance(); + } + } +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/GlobalAuditProvider.java b/api/all/src/main/java/io/opentelemetry/api/audit/GlobalAuditProvider.java new file mode 100644 index 00000000000..e5dbdf2fcac --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/GlobalAuditProvider.java @@ -0,0 +1,49 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +import java.util.concurrent.atomic.AtomicReference; + +/** + * Global singleton holder for the process-wide {@link AuditProvider}. + * + *

In most applications there is only one {@link AuditProvider}. {@link #set(AuditProvider)} + * SHOULD be called once, early in the application lifecycle (for example, in the same place where + * the OpenTelemetry SDK is initialised). + * + *

If no provider is registered, {@link #get()} returns the no-op provider from {@link + * AuditProvider#noop()}. + */ +public final class GlobalAuditProvider { + + private static final AtomicReference globalProvider = + new AtomicReference<>(AuditProvider.noop()); + + private GlobalAuditProvider() {} + + /** Returns the globally registered {@link AuditProvider}, or the no-op instance if none set. */ + public static AuditProvider get() { + return globalProvider.get(); + } + + /** + * Sets the globally registered {@link AuditProvider}. + * + * @param auditProvider the provider to register; MUST NOT be null + * @throws IllegalArgumentException if {@code auditProvider} is null + */ + public static void set(AuditProvider auditProvider) { + if (auditProvider == null) { + throw new IllegalArgumentException("auditProvider must not be null"); + } + globalProvider.set(auditProvider); + } + + /** Resets the global provider to the no-op implementation. Intended for use in tests only. */ + public static void resetForTest() { + globalProvider.set(AuditProvider.noop()); + } +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/Outcome.java b/api/all/src/main/java/io/opentelemetry/api/audit/Outcome.java new file mode 100644 index 00000000000..7f9e891d8f7 --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/Outcome.java @@ -0,0 +1,23 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.api.audit; + +/** + * The result of an auditable action. + * + * @see AuditRecordBuilder#setOutcome(Outcome) + */ +public enum Outcome { + + /** The action completed successfully. */ + SUCCESS, + + /** The action was attempted but did not complete successfully. */ + FAILURE, + + /** The outcome could not be determined at the time of emission. */ + UNKNOWN +} diff --git a/api/all/src/main/java/io/opentelemetry/api/audit/package-info.java b/api/all/src/main/java/io/opentelemetry/api/audit/package-info.java new file mode 100644 index 00000000000..67bbe297c0b --- /dev/null +++ b/api/all/src/main/java/io/opentelemetry/api/audit/package-info.java @@ -0,0 +1,10 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +/** OpenTelemetry Audit Logging API. */ +@ParametersAreNonnullByDefault +package io.opentelemetry.api.audit; + +import javax.annotation.ParametersAreNonnullByDefault; diff --git a/exporters/otlp/audit/build.gradle.kts b/exporters/otlp/audit/build.gradle.kts new file mode 100644 index 00000000000..d03630a23c2 --- /dev/null +++ b/exporters/otlp/audit/build.gradle.kts @@ -0,0 +1,18 @@ +plugins { + id("otel.java-conventions") + id("otel.publish-conventions") + id("otel.animalsniffer-conventions") +} + +description = "OpenTelemetry OTLP Audit Exporter" +otelJava.moduleName.set("io.opentelemetry.exporter.otlp.audit") + +dependencies { + api(project(":sdk:audit")) + api(project(":sdk:logs")) + implementation(project(":exporters:otlp:common")) + implementation(project(":exporters:sender:okhttp")) + + testImplementation(project(":exporters:otlp:testing-internal")) + testImplementation("com.linecorp.armeria:armeria-junit5") +} diff --git a/exporters/otlp/audit/gradle.properties b/exporters/otlp/audit/gradle.properties new file mode 100644 index 00000000000..4476ae57e31 --- /dev/null +++ b/exporters/otlp/audit/gradle.properties @@ -0,0 +1 @@ +otel.release=alpha diff --git a/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/AuditLogRecordDataAdapter.java b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/AuditLogRecordDataAdapter.java new file mode 100644 index 00000000000..0be32c753ec --- /dev/null +++ b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/AuditLogRecordDataAdapter.java @@ -0,0 +1,147 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.exporter.otlp.http.audit; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.sdk.audit.AuditRecordData; +import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.resources.Resource; +import javax.annotation.Nullable; + +/** + * Adapts an {@link AuditRecordData} to the {@link LogRecordData} interface so that the existing + * OTLP log marshaling infrastructure can serialize audit records to the {@code + * ExportLogsServiceRequest} protobuf message. + * + *

Mappings per the Audit Logging specification: + * + *

+ */ +final class AuditLogRecordDataAdapter implements LogRecordData { + + private static final String ATTR_RECORD_ID = "audit.record_id"; + private static final String ATTR_ACTOR = "audit.actor"; + private static final String ATTR_ACTOR_TYPE = "audit.actor_type"; + private static final String ATTR_ACTION = "audit.action"; + private static final String ATTR_OUTCOME = "audit.outcome"; + private static final String ATTR_TARGET_RESOURCE = "audit.target_resource"; + private static final String ATTR_SOURCE_IP = "audit.source_ip"; + private static final String ATTR_SCHEMA_VERSION = "audit.schema_version"; + private static final String ATTR_SEQUENCE_NO = "audit.sequence_no"; + private static final String ATTR_PREV_HASH = "audit.prev_hash"; + + private final AuditRecordData audit; + private final Attributes mergedAttributes; + + AuditLogRecordDataAdapter(AuditRecordData audit) { + this.audit = audit; + this.mergedAttributes = buildAttributes(audit); + } + + private static Attributes buildAttributes(AuditRecordData a) { + AttributesBuilder b = Attributes.builder(); + // Mandatory audit fields as attributes + b.put(AttributeKey.stringKey(ATTR_RECORD_ID), a.getRecordId()); + b.put(AttributeKey.stringKey(ATTR_ACTOR), a.getActor().asString()); + b.put(AttributeKey.stringKey(ATTR_ACTOR_TYPE), a.getActorType().name()); + b.put(AttributeKey.stringKey(ATTR_ACTION), a.getAction()); + b.put(AttributeKey.stringKey(ATTR_OUTCOME), a.getOutcome().name()); + // Optional audit fields + if (a.getTargetResource() != null) { + b.put(AttributeKey.stringKey(ATTR_TARGET_RESOURCE), a.getTargetResource().asString()); + } + if (a.getSourceIp() != null) { + b.put(AttributeKey.stringKey(ATTR_SOURCE_IP), a.getSourceIp()); + } + if (a.getSchemaVersion() != null) { + b.put(AttributeKey.stringKey(ATTR_SCHEMA_VERSION), a.getSchemaVersion()); + } + if (a.getSequenceNo() != 0) { + b.put(AttributeKey.longKey(ATTR_SEQUENCE_NO), a.getSequenceNo()); + } + if (a.getPrevHash() != null) { + b.put(AttributeKey.stringKey(ATTR_PREV_HASH), a.getPrevHash()); + } + // User-supplied attributes (merged last so they can override if needed) + a.getAttributes() + .forEach( + (key, value) -> { + @SuppressWarnings("unchecked") + AttributeKey castKey = (AttributeKey) key; + b.put(castKey, value); + }); + return b.build(); + } + + @Override + public Resource getResource() { + return audit.getResource(); + } + + /** Audit records do not use instrumentation scope; always returns empty. */ + @Override + public InstrumentationScopeInfo getInstrumentationScopeInfo() { + return InstrumentationScopeInfo.empty(); + } + + @Override + public long getTimestampEpochNanos() { + return audit.getTimestampEpochNanos(); + } + + @Override + public long getObservedTimestampEpochNanos() { + return audit.getObservedTimestampEpochNanos(); + } + + @Override + public SpanContext getSpanContext() { + return SpanContext.getInvalid(); + } + + /** Audit records do not use severity; always returns {@code null}. */ + @Override + @Nullable + public Severity getSeverity() { + return null; + } + + @Override + @Nullable + public String getSeverityText() { + return null; + } + + @Override + @Nullable + public io.opentelemetry.api.common.Value getBodyValue() { + return audit.getBody(); + } + + @Override + public Attributes getAttributes() { + return mergedAttributes; + } + + @Override + public int getTotalAttributeCount() { + return mergedAttributes.size(); + } + + @Override + public String getEventName() { + return audit.getEventName(); + } +} diff --git a/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/OtlpHttpAuditRecordExporter.java b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/OtlpHttpAuditRecordExporter.java new file mode 100644 index 00000000000..ab5d78976f5 --- /dev/null +++ b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/OtlpHttpAuditRecordExporter.java @@ -0,0 +1,153 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.exporter.otlp.http.audit; + +import static java.util.Objects.requireNonNull; + +import io.opentelemetry.api.audit.AuditReceipt; +import io.opentelemetry.exporter.internal.http.HttpExporter; +import io.opentelemetry.exporter.internal.http.HttpExporterBuilder; +import io.opentelemetry.exporter.internal.otlp.logs.LogsRequestMarshaler; +import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent; +import io.opentelemetry.sdk.audit.AuditExportResult; +import io.opentelemetry.sdk.audit.AuditRecordData; +import io.opentelemetry.sdk.audit.AuditRecordExporter; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.common.internal.ComponentId; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.StringJoiner; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import javax.annotation.concurrent.ThreadSafe; + +/** + * Exports {@link AuditRecordData}s using OTLP/HTTP to the dedicated {@code /v1/audit} endpoint. + * + *

Audit records are serialized as OTLP {@code LogRecord} protobuf messages (reusing the + * {@code ExportLogsServiceRequest} envelope) with mandatory audit fields stored as attributes. + * + *

The OTLP receiver MUST NOT respond with {@code partial_success}; any partial-success response + * is treated as a hard failure and all records in the batch are retained for retry. + * + *

Create via {@link #builder()} or {@link #getDefault()}. + */ +@ThreadSafe +public final class OtlpHttpAuditRecordExporter implements AuditRecordExporter { + + static final String DEFAULT_ENDPOINT = "http://localhost:4318/v1/audit"; + private static final ComponentId COMPONENT_ID = + ComponentId.generateLazy("otlp_http_audit_exporter"); + + private final HttpExporterBuilder builder; + private final HttpExporter delegate; + + OtlpHttpAuditRecordExporter(HttpExporterBuilder builder, HttpExporter delegate) { + this.builder = builder; + this.delegate = delegate; + } + + /** Returns a new {@link OtlpHttpAuditRecordExporter} with default configuration. */ + public static OtlpHttpAuditRecordExporter getDefault() { + return builder().build(); + } + + /** Returns a new {@link OtlpHttpAuditRecordExporterBuilder}. */ + public static OtlpHttpAuditRecordExporterBuilder builder() { + return new OtlpHttpAuditRecordExporterBuilder(); + } + + /** + * Exports the given audit records to the configured OTLP {@code /v1/audit} endpoint. + * + *

Audit records are adapted to OTLP {@code LogRecord}s via {@link + * AuditLogRecordDataAdapter}. The {@code InstrumentationScope} is left empty and {@code + * SeverityNumber} is unset per the audit logging specification. + * + *

Returns synthetic {@link AuditReceipt}s on success. The {@code IntegrityHash} field is + * empty in this implementation; a future OTLP response extension will carry the sink-computed + * hash. + * + *

Any {@code partial_success} response is treated as a hard failure (all records are + * returned in the failure result for retry). + */ + @Override + public AuditExportResult export(Collection records) { + if (records.isEmpty()) { + return AuditExportResult.success(java.util.Collections.emptyList()); + } + + // Adapt AuditRecordData → LogRecordData for marshaling + List adapted = new ArrayList<>(records.size()); + for (AuditRecordData record : records) { + adapted.add(new AuditLogRecordDataAdapter(record)); + } + + // Serialize as ExportLogsServiceRequest and send to /v1/audit + LogsRequestMarshaler marshaler = LogsRequestMarshaler.create(adapted); + AtomicReference failureCause = new AtomicReference<>(); + AtomicBoolean succeeded = new AtomicBoolean(false); + CountDownLatch latch = new CountDownLatch(1); + + CompletableResultCode result = delegate.export(marshaler, adapted.size()); + result.whenComplete( + () -> { + if (result.isSuccess()) { + succeeded.set(true); + } else { + failureCause.set(result.getFailureThrowable()); + } + latch.countDown(); + }); + + try { + // Block until export completes (audit records must be acknowledged) + latch.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return AuditExportResult.failure(e); + } + + if (latch.getCount() > 0) { + return AuditExportResult.failure( + new io.opentelemetry.api.audit.AuditDeliveryException( + "OTLP export timed out waiting for /v1/audit acknowledgement")); + } + + if (!succeeded.get()) { + Throwable cause = failureCause.get(); + return cause != null ? AuditExportResult.failure(cause) : AuditExportResult.failure(); + } + + // Synthesize receipts (IntegrityHash populated by sink in future OTLP extension) + List receipts = new ArrayList<>(records.size()); + for (AuditRecordData record : records) { + receipts.add(AuditReceipt.create(record.getRecordId(), "", 0)); + } + return AuditExportResult.success(receipts); + } + + @Override + public CompletableResultCode flush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return delegate.shutdown(); + } + + @Override + public String toString() { + StringJoiner joiner = new StringJoiner(", ", "OtlpHttpAuditRecordExporter{", "}"); + joiner.add(builder.toString(false)); + return joiner.toString(); + } +} diff --git a/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/OtlpHttpAuditRecordExporterBuilder.java b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/OtlpHttpAuditRecordExporterBuilder.java new file mode 100644 index 00000000000..fe10357ca5b --- /dev/null +++ b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/OtlpHttpAuditRecordExporterBuilder.java @@ -0,0 +1,99 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.exporter.otlp.http.audit; + +import static io.opentelemetry.api.internal.Utils.checkArgument; +import static java.util.Objects.requireNonNull; + +import io.opentelemetry.exporter.internal.http.HttpExporterBuilder; +import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent; +import io.opentelemetry.sdk.common.export.RetryPolicy; +import io.opentelemetry.sdk.common.internal.ComponentId; +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import javax.net.ssl.SSLContext; +import javax.net.ssl.X509TrustManager; + +/** + * Builder for {@link OtlpHttpAuditRecordExporter}. + * + *

Defaults: endpoint {@code http://localhost:4318/v1/audit}, 10 s timeout. + */ +public final class OtlpHttpAuditRecordExporterBuilder { + + private static final ComponentId COMPONENT_ID = + ComponentId.generateLazy("otlp_http_audit_exporter"); + + private final HttpExporterBuilder delegate; + + OtlpHttpAuditRecordExporterBuilder() { + this.delegate = + new HttpExporterBuilder(COMPONENT_ID, OtlpHttpAuditRecordExporter.DEFAULT_ENDPOINT); + OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders); + } + + OtlpHttpAuditRecordExporterBuilder(HttpExporterBuilder delegate) { + this.delegate = delegate; + } + + /** Sets the maximum time to wait for the collector to process an exported batch. */ + public OtlpHttpAuditRecordExporterBuilder setTimeout(long timeout, TimeUnit unit) { + requireNonNull(unit, "unit"); + checkArgument(timeout >= 0, "timeout must be non-negative"); + return setTimeout(Duration.ofNanos(unit.toNanos(timeout))); + } + + /** Sets the maximum time to wait for the collector to process an exported batch. */ + public OtlpHttpAuditRecordExporterBuilder setTimeout(Duration timeout) { + requireNonNull(timeout, "timeout"); + delegate.setTimeout(timeout); + return this; + } + + /** Sets the OTLP endpoint URL. Defaults to {@code http://localhost:4318/v1/audit}. */ + public OtlpHttpAuditRecordExporterBuilder setEndpoint(String endpoint) { + requireNonNull(endpoint, "endpoint"); + delegate.setEndpoint(endpoint); + return this; + } + + /** Adds a constant HTTP header sent with every request. */ + public OtlpHttpAuditRecordExporterBuilder addHeader(String key, String value) { + delegate.addConstantHeaders(key, value); + return this; + } + + /** Adds constant HTTP headers sent with every request. */ + public OtlpHttpAuditRecordExporterBuilder setHeaders(Map headers) { + headers.forEach(delegate::addConstantHeaders); + return this; + } + + /** Configures TLS for the OTLP connection. */ + public OtlpHttpAuditRecordExporterBuilder setSslContext( + SSLContext sslContext, X509TrustManager trustManager) { + requireNonNull(sslContext, "sslContext"); + requireNonNull(trustManager, "trustManager"); + delegate.setSslContext(sslContext, trustManager); + return this; + } + + /** + * Sets the retry policy. Audit records MUST NOT be silently dropped on retry exhaustion; the + * exporter will surface a hard error instead. + */ + public OtlpHttpAuditRecordExporterBuilder setRetryPolicy(@Nullable RetryPolicy retryPolicy) { + delegate.setRetryPolicy(retryPolicy); + return this; + } + + /** Builds and returns the configured {@link OtlpHttpAuditRecordExporter}. */ + public OtlpHttpAuditRecordExporter build() { + return new OtlpHttpAuditRecordExporter(delegate, delegate.build()); + } +} diff --git a/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/package-info.java b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/package-info.java new file mode 100644 index 00000000000..003db11e912 --- /dev/null +++ b/exporters/otlp/audit/src/main/java/io/opentelemetry/exporter/otlp/http/audit/package-info.java @@ -0,0 +1,10 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +/** OTLP Audit Logging exporter. */ +@ParametersAreNonnullByDefault +package io.opentelemetry.exporter.otlp.http.audit; + +import javax.annotation.ParametersAreNonnullByDefault; diff --git a/sdk-extensions/autoconfigure-spi/src/main/java/io/opentelemetry/sdk/autoconfigure/spi/audit/ConfigurableAuditRecordExporterProvider.java b/sdk-extensions/autoconfigure-spi/src/main/java/io/opentelemetry/sdk/autoconfigure/spi/audit/ConfigurableAuditRecordExporterProvider.java new file mode 100644 index 00000000000..73de13ad513 --- /dev/null +++ b/sdk-extensions/autoconfigure-spi/src/main/java/io/opentelemetry/sdk/autoconfigure/spi/audit/ConfigurableAuditRecordExporterProvider.java @@ -0,0 +1,32 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.autoconfigure.spi.audit; + +import io.opentelemetry.sdk.audit.AuditRecordExporter; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; + +/** + * A service provider interface (SPI) for providing audit record exporters that can be used with + * the autoconfigured SDK. If the {@code otel.audit.exporter} property contains a value equal to + * what is returned by {@link #getName()}, the exporter returned by {@link + * #createExporter(ConfigProperties)} will be enabled and added to the audit pipeline. + * + *

This SPI is at {@code Development} stability; the interface may change in future releases. + */ +public interface ConfigurableAuditRecordExporterProvider { + + /** + * Returns an {@link AuditRecordExporter} that can be registered to the audit pipeline by + * providing the property value specified by {@link #getName()}. + */ + AuditRecordExporter createExporter(ConfigProperties config); + + /** + * Returns the name of this exporter, which can be specified with the {@code otel.audit.exporter} + * property to enable it. The name returned MUST NOT be the same as any other exporter name. + */ + String getName(); +} diff --git a/sdk-extensions/autoconfigure-spi/src/main/java/io/opentelemetry/sdk/autoconfigure/spi/audit/package-info.java b/sdk-extensions/autoconfigure-spi/src/main/java/io/opentelemetry/sdk/autoconfigure/spi/audit/package-info.java new file mode 100644 index 00000000000..78ed13d9040 --- /dev/null +++ b/sdk-extensions/autoconfigure-spi/src/main/java/io/opentelemetry/sdk/autoconfigure/spi/audit/package-info.java @@ -0,0 +1,16 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Java SPI (Service Provider Interface) for implementing extensions to SDK autoconfiguration of + * audit logging. + * + *

This package is at {@code Development} stability; the interfaces may change in future + * releases. + */ +@ParametersAreNonnullByDefault +package io.opentelemetry.sdk.autoconfigure.spi.audit; + +import javax.annotation.ParametersAreNonnullByDefault; diff --git a/sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/AuditExporterConfiguration.java b/sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/AuditExporterConfiguration.java new file mode 100644 index 00000000000..6130eab6c0c --- /dev/null +++ b/sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/AuditExporterConfiguration.java @@ -0,0 +1,67 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.autoconfigure; + +import io.opentelemetry.sdk.audit.AuditRecordExporter; +import io.opentelemetry.sdk.autoconfigure.internal.NamedSpiManager; +import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException; +import io.opentelemetry.sdk.autoconfigure.spi.audit.ConfigurableAuditRecordExporterProvider; +import io.opentelemetry.sdk.autoconfigure.spi.internal.DefaultConfigProperties; +import java.io.Closeable; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +final class AuditExporterConfiguration { + + private static final String EXPORTER_NONE = "none"; + + static Map configureAuditRecordExporters( + ConfigProperties config, SpiHelper spiHelper, List closeables) { + Set exporterNames = DefaultConfigProperties.getSet(config, "otel.audit.exporter"); + + if (exporterNames.contains(EXPORTER_NONE)) { + if (exporterNames.size() > 1) { + throw new ConfigurationException( + "otel.audit.exporter contains " + EXPORTER_NONE + " along with other exporters"); + } + return Collections.emptyMap(); + } + + if (exporterNames.isEmpty()) { + exporterNames = Collections.singleton("otlp"); + } + + NamedSpiManager spiManager = + auditExporterSpiManager(config, spiHelper); + + Map map = new HashMap<>(); + for (String name : exporterNames) { + AuditRecordExporter exporter = spiManager.getByName(name); + if (exporter == null) { + throw new ConfigurationException("Unrecognized value for otel.audit.exporter: " + name); + } + closeables.add(exporter); + map.put(name, exporter); + } + return Collections.unmodifiableMap(map); + } + + static NamedSpiManager auditExporterSpiManager( + ConfigProperties config, SpiHelper spiHelper) { + return spiHelper.loadConfigurable( + ConfigurableAuditRecordExporterProvider.class, + ConfigurableAuditRecordExporterProvider::getName, + ConfigurableAuditRecordExporterProvider::createExporter, + config); + } + + private AuditExporterConfiguration() {} +} diff --git a/sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/AuditProviderConfiguration.java b/sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/AuditProviderConfiguration.java new file mode 100644 index 00000000000..69a3ca2c78a --- /dev/null +++ b/sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/AuditProviderConfiguration.java @@ -0,0 +1,55 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.autoconfigure; + +import io.opentelemetry.api.audit.GlobalAuditProvider; +import io.opentelemetry.sdk.audit.AuditRecordExporter; +import io.opentelemetry.sdk.audit.AuditRecordProcessor; +import io.opentelemetry.sdk.audit.SdkAuditProvider; +import io.opentelemetry.sdk.audit.SdkAuditProviderBuilder; +import io.opentelemetry.sdk.audit.export.SimpleAuditRecordProcessor; +import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.resources.Resource; +import java.io.Closeable; +import java.util.List; +import java.util.Map; + +/** + * Configures an {@link SdkAuditProvider} from autoconfiguration properties and registers it as + * the global {@link io.opentelemetry.api.audit.AuditProvider}. + * + *

The property {@code otel.audit.exporter} controls which exporter is used (default: {@code + * otlp}). Use {@code otel.audit.exporter=none} to disable audit logging. + */ +final class AuditProviderConfiguration { + + static SdkAuditProvider configureAuditProvider( + Resource resource, + ConfigProperties config, + SpiHelper spiHelper, + List closeables) { + + Map exportersByName = + AuditExporterConfiguration.configureAuditRecordExporters(config, spiHelper, closeables); + + SdkAuditProviderBuilder builder = SdkAuditProvider.builder().setResource(resource); + + // Each configured exporter gets its own SimpleAuditRecordProcessor in the chain. + for (AuditRecordExporter exporter : exportersByName.values()) { + AuditRecordProcessor processor = SimpleAuditRecordProcessor.create(exporter); + closeables.add(processor); + builder.addAuditRecordProcessor(processor); + } + + SdkAuditProvider provider = builder.build(); + closeables.add(provider); + GlobalAuditProvider.set(provider); + return provider; + } + + private AuditProviderConfiguration() {} +} diff --git a/sdk/all/build.gradle.kts b/sdk/all/build.gradle.kts index 3a19758f313..fc3979f1e78 100644 --- a/sdk/all/build.gradle.kts +++ b/sdk/all/build.gradle.kts @@ -16,6 +16,7 @@ dependencies { api(project(":sdk:trace")) api(project(":sdk:metrics")) api(project(":sdk:logs")) + api(project(":sdk:audit")) compileOnly(project(":api:incubator")) diff --git a/sdk/audit/build.gradle.kts b/sdk/audit/build.gradle.kts new file mode 100644 index 00000000000..fb81b3580c5 --- /dev/null +++ b/sdk/audit/build.gradle.kts @@ -0,0 +1,18 @@ +plugins { + id("otel.java-conventions") + id("otel.publish-conventions") + id("otel.animalsniffer-conventions") +} + +description = "OpenTelemetry Audit Logging SDK" +otelJava.moduleName.set("io.opentelemetry.sdk.audit") + +dependencies { + api(project(":api:all")) + api(project(":sdk:common")) + + annotationProcessor("com.google.auto.value:auto-value") + + testImplementation("org.awaitility:awaitility") + testImplementation("com.google.guava:guava") +} diff --git a/sdk/audit/gradle.properties b/sdk/audit/gradle.properties new file mode 100644 index 00000000000..4476ae57e31 --- /dev/null +++ b/sdk/audit/gradle.properties @@ -0,0 +1 @@ +otel.release=alpha diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditExportResult.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditExportResult.java new file mode 100644 index 00000000000..4b122fce7ec --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditExportResult.java @@ -0,0 +1,66 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.AuditReceipt; +import java.util.Collections; +import java.util.List; + +/** + * The synchronous result of an {@link AuditRecordExporter#export} call. + * + *

On success, {@link #getReceipts()} contains one {@link AuditReceipt} per exported record in + * the same order as the input collection. On failure, the list is empty and {@link #isSuccess()} + * returns {@code false}. + */ +public final class AuditExportResult { + + private final boolean success; + private final List receipts; + private final Throwable failure; + + private AuditExportResult(boolean success, List receipts, Throwable failure) { + this.success = success; + this.receipts = receipts; + this.failure = failure; + } + + /** Creates a successful result with the given receipts. */ + public static AuditExportResult success(List receipts) { + return new AuditExportResult(true, Collections.unmodifiableList(receipts), null); + } + + /** Creates a failure result with the given cause. */ + public static AuditExportResult failure(Throwable cause) { + return new AuditExportResult(false, Collections.emptyList(), cause); + } + + /** Creates a failure result without a specific cause. */ + public static AuditExportResult failure() { + return new AuditExportResult(false, Collections.emptyList(), null); + } + + /** Returns {@code true} if all records were successfully acknowledged by the audit sink. */ + public boolean isSuccess() { + return success; + } + + /** + * Returns the {@link AuditReceipt}s returned by the audit sink, one per exported record. Empty + * if {@link #isSuccess()} is {@code false}. + */ + public List getReceipts() { + return receipts; + } + + /** + * Returns the cause of the failure, or {@code null} if the failure has no associated throwable + * or if the export succeeded. + */ + public Throwable getFailure() { + return failure; + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordData.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordData.java new file mode 100644 index 00000000000..452c4c004a0 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordData.java @@ -0,0 +1,112 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.ActorType; +import io.opentelemetry.api.audit.Outcome; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.Value; +import io.opentelemetry.sdk.resources.Resource; +import javax.annotation.Nullable; + +/** + * Immutable representation of an audit record for use by processors and exporters. + * + *

Instances are created internally by the SDK when {@link + * io.opentelemetry.api.audit.AuditRecordBuilder#emit()} is called. + */ +public interface AuditRecordData { + + /** Returns the {@link Resource} of the emitting service. */ + Resource getResource(); + + /** + * Returns the diagnostic name of the {@link io.opentelemetry.api.audit.AuditLogger} that emitted + * this record (for example {@code "com.example.auth"}). + */ + String getLoggerName(); + + /** Returns the optional version of the emitting component, or {@code null} if not set. */ + @Nullable + String getLoggerVersion(); + + /** Returns the optional schema URL, or {@code null} if not set. */ + @Nullable + String getSchemaUrl(); + + /** Returns the caller-generated unique identifier for this record. Never null or empty. */ + String getRecordId(); + + /** Returns the event time as nanoseconds since the UNIX epoch (UTC). */ + long getTimestampEpochNanos(); + + /** Returns the SDK observation time as nanoseconds since the UNIX epoch (UTC). */ + long getObservedTimestampEpochNanos(); + + /** Returns the semantic name of the audit event, e.g. {@code "user.login.success"}. */ + String getEventName(); + + /** Returns the identity of the actor that performed the action. */ + Value getActor(); + + /** Returns the type of the actor. */ + ActorType getActorType(); + + /** Returns the action verb, e.g. {@code "LOGIN"}, {@code "DELETE"}. */ + String getAction(); + + /** Returns the outcome of the action. */ + Outcome getOutcome(); + + /** Returns the target resource of the action, or {@code null} if not set. */ + @Nullable + Value getTargetResource(); + + /** Returns the source IP address, or {@code null} if not set. */ + @Nullable + String getSourceIp(); + + /** Returns the free-form body, or {@code null} if not set. */ + @Nullable + Value getBody(); + + /** Returns the attributes attached to this record (never null; may be empty). */ + Attributes getAttributes(); + + /** Returns the optional digital signature bytes, or {@code null} if not set. */ + @Nullable + byte[] getSignature(); + + /** Returns the signature algorithm, or {@code null} if {@link #getSignature()} is not set. */ + @Nullable + String getAlgorithm(); + + /** Returns the DER-encoded X.509 certificate, or {@code null} if not set. */ + @Nullable + byte[] getCertificate(); + + /** Returns the HMAC bytes, or {@code null} if not set. */ + @Nullable + byte[] getHmac(); + + /** Returns the HMAC algorithm, or {@code null} if {@link #getHmac()} is not set. */ + @Nullable + String getHmacAlgorithm(); + + /** + * Returns the monotonic sequence number for hash-chain continuity, or {@code 0} if not set. + */ + long getSequenceNo(); + + /** Returns the {@code IntegrityHash} of the preceding record for hash-chain linking, or {@code + * null} if not set. */ + @Nullable + String getPrevHash(); + + /** Returns the schema version of the audit payload, or {@code null} if not set. */ + @Nullable + String getSchemaVersion(); +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordExporter.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordExporter.java new file mode 100644 index 00000000000..6155dde0210 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordExporter.java @@ -0,0 +1,56 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.sdk.common.CompletableResultCode; +import java.io.Closeable; +import java.util.Collection; +import java.util.concurrent.TimeUnit; + +/** + * Transmits {@link AuditRecordData}s to the configured audit sink. + * + *

Implementations MUST document their concurrency characteristics. {@link #export} MUST NOT be + * called concurrently on the same instance. + * + *

The audit logging specification prohibits partial success: if the receiver cannot process one + * or more records, the entire batch MUST be rejected. Implementations MUST treat a partial-success + * response from the OTLP receiver as a hard failure and retry the full batch. + */ +public interface AuditRecordExporter extends Closeable { + + /** + * Exports the given collection of {@link AuditRecordData}s to the audit sink. + * + *

MUST NOT be called concurrently on the same exporter instance. MUST NOT block indefinitely; + * the exporter MUST time out within the configured export timeout. + * + * @param records the records to export; the collection MUST NOT be mutated after this call + * @return an {@link AuditExportResult} containing one {@link + * io.opentelemetry.api.audit.AuditReceipt} per record on success, or a failure result + */ + AuditExportResult export(Collection records); + + /** + * Requests that any internally buffered records be exported immediately. + * + * @return a result indicating whether the flush succeeded + */ + CompletableResultCode flush(); + + /** + * Shuts down this exporter. On the first call, flushes any buffered records and releases all + * resources. Subsequent calls are no-ops. + * + * @return a result indicating whether the shutdown succeeded + */ + CompletableResultCode shutdown(); + + @Override + default void close() { + shutdown().join(10, TimeUnit.SECONDS); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordProcessor.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordProcessor.java new file mode 100644 index 00000000000..23ee5ec7242 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/AuditRecordProcessor.java @@ -0,0 +1,79 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; +import java.io.Closeable; +import java.util.Arrays; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import javax.annotation.concurrent.ThreadSafe; + +/** + * Interface for hooking into the audit record pipeline for enrichment and forwarding. + * + *

Processors MUST only add attributes to records (enrichment). They MUST NOT remove mandatory + * fields, filter records, aggregate records, or introduce sampling. Processors that would remove + * or filter records are rejected at configuration time by {@link SdkAuditProvider}. + */ +@ThreadSafe +public interface AuditRecordProcessor extends Closeable { + + /** + * Returns a composite {@link AuditRecordProcessor} that delegates to all given processors in + * order. + */ + static AuditRecordProcessor composite(AuditRecordProcessor... processors) { + return composite(Arrays.asList(processors)); + } + + /** + * Returns a composite {@link AuditRecordProcessor} that delegates to all given processors in + * order. + */ + static AuditRecordProcessor composite(Iterable processors) { + List list = new ArrayList<>(); + for (AuditRecordProcessor p : processors) { + list.add(p); + } + if (list.isEmpty()) { + return NoopAuditRecordProcessor.getInstance(); + } + if (list.size() == 1) { + return list.get(0); + } + return MultiAuditRecordProcessor.create(list); + } + + /** + * Called synchronously on the calling thread after the record has been enqueued and before the + * {@link io.opentelemetry.api.audit.AuditReceipt} is returned to the caller. + * + *

Implementations MAY enrich {@code record} by adding attributes. They MUST NOT block + * indefinitely. + * + * @param context the ambient {@link Context} at the time of {@code emit()} + * @param record the mutable record; enrichment only + */ + void onEmit(Context context, ReadWriteAuditRecord record); + + /** Shuts down this processor, flushing all buffered records. */ + default CompletableResultCode shutdown() { + return forceFlush(); + } + + /** Requests that all buffered records be exported as soon as possible. */ + default CompletableResultCode forceFlush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + default void close() { + shutdown().join(10, TimeUnit.SECONDS); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/MultiAuditRecordProcessor.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/MultiAuditRecordProcessor.java new file mode 100644 index 00000000000..f61fd8f9285 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/MultiAuditRecordProcessor.java @@ -0,0 +1,50 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; +import java.util.ArrayList; +import java.util.List; + +/** Composite {@link AuditRecordProcessor} that delegates to multiple processors in order. */ +final class MultiAuditRecordProcessor implements AuditRecordProcessor { + + private final List processors; + + private MultiAuditRecordProcessor(List processors) { + this.processors = processors; + } + + static MultiAuditRecordProcessor create(List processors) { + return new MultiAuditRecordProcessor(new ArrayList<>(processors)); + } + + @Override + public void onEmit(Context context, ReadWriteAuditRecord record) { + for (AuditRecordProcessor processor : processors) { + processor.onEmit(context, record); + } + } + + @Override + public CompletableResultCode shutdown() { + List results = new ArrayList<>(processors.size()); + for (AuditRecordProcessor processor : processors) { + results.add(processor.shutdown()); + } + return CompletableResultCode.ofAll(results); + } + + @Override + public CompletableResultCode forceFlush() { + List results = new ArrayList<>(processors.size()); + for (AuditRecordProcessor processor : processors) { + results.add(processor.forceFlush()); + } + return CompletableResultCode.ofAll(results); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/NoopAuditRecordProcessor.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/NoopAuditRecordProcessor.java new file mode 100644 index 00000000000..4d4e04f34bf --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/NoopAuditRecordProcessor.java @@ -0,0 +1,34 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; + +/** No-op {@link AuditRecordProcessor} returned when no processors are registered. */ +final class NoopAuditRecordProcessor implements AuditRecordProcessor { + + private static final AuditRecordProcessor INSTANCE = new NoopAuditRecordProcessor(); + + private NoopAuditRecordProcessor() {} + + static AuditRecordProcessor getInstance() { + return INSTANCE; + } + + @Override + public void onEmit(Context context, ReadWriteAuditRecord record) {} + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode forceFlush() { + return CompletableResultCode.ofSuccess(); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/ReadWriteAuditRecord.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/ReadWriteAuditRecord.java new file mode 100644 index 00000000000..dce7cf62390 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/ReadWriteAuditRecord.java @@ -0,0 +1,63 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.AuditReceipt; +import io.opentelemetry.api.audit.ActorType; +import io.opentelemetry.api.audit.Outcome; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Value; +import javax.annotation.Nullable; + +/** + * Mutable view of an {@link AuditRecordData} passed to {@link AuditRecordProcessor#onEmit}. + * + *

Processors MAY enrich the record by calling {@link #setAttribute}. They MUST NOT modify the + * mandatory audit fields ({@code EventName}, {@code Actor}, {@code ActorType}, {@code Action}, + * {@code Outcome}): those are exposed as read-only accessors. + * + *

The {@link #setReceipt}/{@link #getReceipt} pair is used internally by the SDK to thread the + * {@link AuditReceipt} returned by the exporter back to the calling {@code emit()} invocation. + */ +public interface ReadWriteAuditRecord { + + /** + * Adds or replaces an attribute on this record. A {@code null} value is a no-op. + * + *

MUST NOT be called to modify mandatory fields; only additional enrichment attributes are + * permitted. + */ + ReadWriteAuditRecord setAttribute(AttributeKey key, @Nullable T value); + + /** + * Stores the {@link AuditReceipt} returned by the exporter. Called by the SDK after a successful + * export. + */ + void setReceipt(AuditReceipt receipt); + + /** Returns the {@link AuditReceipt} set by the exporter, or {@code null} if not yet set. */ + @Nullable + AuditReceipt getReceipt(); + + /** Snapshots this record into an immutable {@link AuditRecordData} for export. */ + AuditRecordData toAuditRecordData(); + + // ── Read-only accessors for mandatory fields ────────────────────────────── + + String getRecordId(); + + long getTimestampEpochNanos(); + + String getEventName(); + + Value getActor(); + + ActorType getActorType(); + + String getAction(); + + Outcome getOutcome(); +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditLogger.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditLogger.java new file mode 100644 index 00000000000..7bf006588ae --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditLogger.java @@ -0,0 +1,28 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.AuditLogger; +import io.opentelemetry.api.audit.AuditRecordBuilder; +import javax.annotation.concurrent.ThreadSafe; + +/** SDK implementation of {@link AuditLogger}. */ +@ThreadSafe +final class SdkAuditLogger implements AuditLogger { + + private final SdkAuditProvider provider; + private final SdkAuditProvider.AuditLoggerKey key; + + SdkAuditLogger(SdkAuditProvider provider, SdkAuditProvider.AuditLoggerKey key) { + this.provider = provider; + this.key = key; + } + + @Override + public AuditRecordBuilder auditRecordBuilder() { + return new SdkAuditRecordBuilder(provider, key); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditLoggerBuilder.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditLoggerBuilder.java new file mode 100644 index 00000000000..260c511aa39 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditLoggerBuilder.java @@ -0,0 +1,42 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.AuditLogger; +import io.opentelemetry.api.audit.AuditLoggerBuilder; +import javax.annotation.Nullable; + +/** SDK implementation of {@link AuditLoggerBuilder}. */ +final class SdkAuditLoggerBuilder implements AuditLoggerBuilder { + + private final SdkAuditProvider provider; + private final String name; + @Nullable private String version; + @Nullable private String schemaUrl; + + SdkAuditLoggerBuilder(SdkAuditProvider provider, String name) { + this.provider = provider; + this.name = name; + } + + @Override + public SdkAuditLoggerBuilder setSchemaUrl(String schemaUrl) { + this.schemaUrl = schemaUrl; + return this; + } + + @Override + public SdkAuditLoggerBuilder setInstrumentationVersion(String instrumentationVersion) { + this.version = instrumentationVersion; + return this; + } + + @Override + public AuditLogger build() { + return provider.getOrCreateLogger( + new SdkAuditProvider.AuditLoggerKey(name, version, schemaUrl)); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditProvider.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditProvider.java new file mode 100644 index 00000000000..18ed766b883 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditProvider.java @@ -0,0 +1,179 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.AuditDeliveryException; +import io.opentelemetry.api.audit.AuditLogger; +import io.opentelemetry.api.audit.AuditLoggerBuilder; +import io.opentelemetry.api.audit.AuditProvider; +import io.opentelemetry.sdk.common.Clock; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.resources.Resource; +import java.io.Closeable; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.logging.Level; +import java.util.logging.Logger; +import javax.annotation.Nullable; + +/** + * SDK implementation of {@link AuditProvider}. + * + *

Create via {@link #builder()}. The provider maintains a dedicated queue and exporter pipeline + * that is completely independent of the Log signal pipeline. + * + *

The provider intentionally does NOT expose sampler configuration. Any attempt to configure + * sampling on the audit pipeline is a configuration error and will be rejected. + */ +public final class SdkAuditProvider implements AuditProvider, Closeable { + + private static final Logger logger = Logger.getLogger(SdkAuditProvider.class.getName()); + + private final Resource resource; + private final AuditRecordProcessor processor; + private final Clock clock; + private final ConcurrentHashMap loggerRegistry = + new ConcurrentHashMap<>(); + private final AtomicBoolean isShutdown = new AtomicBoolean(false); + + SdkAuditProvider(Resource resource, List processors, Clock clock) { + this.resource = resource; + this.processor = AuditRecordProcessor.composite(processors); + this.clock = clock; + } + + /** Returns a new {@link SdkAuditProviderBuilder}. */ + public static SdkAuditProviderBuilder builder() { + return new SdkAuditProviderBuilder(); + } + + @Override + public AuditLoggerBuilder auditLoggerBuilder(String name) { + if (isShutdown.get()) { + throw new AuditDeliveryException( + "AuditProvider has been shut down; cannot create new AuditLoggers"); + } + if (name == null || name.isEmpty()) { + logger.log( + Level.WARNING, + "AuditProvider.auditLoggerBuilder() called with null or empty name; using 'unknown'"); + name = "unknown"; + } + return new SdkAuditLoggerBuilder(this, name); + } + + /** Returns the {@link SdkAuditLogger} for the given key, creating it if necessary. */ + SdkAuditLogger getOrCreateLogger(AuditLoggerKey key) { + return loggerRegistry.computeIfAbsent(key, k -> new SdkAuditLogger(this, k)); + } + + Resource getResource() { + return resource; + } + + AuditRecordProcessor getProcessor() { + return processor; + } + + Clock getClock() { + return clock; + } + + boolean isShutdown() { + return isShutdown.get(); + } + + /** + * Shuts down this provider. Calls {@link #forceFlush()} then shuts down all registered + * processors. + * + * @return a result indicating whether shutdown succeeded + */ + public CompletableResultCode shutdown() { + if (!isShutdown.compareAndSet(false, true)) { + return CompletableResultCode.ofSuccess(); + } + CompletableResultCode result = new CompletableResultCode(); + CompletableResultCode flushResult = forceFlush(); + flushResult.whenComplete( + () -> { + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.whenComplete( + () -> { + if (!flushResult.isSuccess() || !shutdownResult.isSuccess()) { + result.fail(); + } else { + result.succeed(); + } + }); + }); + return result; + } + + /** + * Forces all buffered audit records to be exported. + * + * @return a result indicating whether the flush succeeded + */ + public CompletableResultCode forceFlush() { + if (isShutdown.get()) { + return CompletableResultCode.ofSuccess(); + } + return processor.forceFlush(); + } + + @Override + public void close() { + shutdown().join(10, TimeUnit.SECONDS); + } + + // ── Inner key type ──────────────────────────────────────────────────────── + + static final class AuditLoggerKey { + + private final String name; + @Nullable private final String version; + @Nullable private final String schemaUrl; + + AuditLoggerKey(String name, @Nullable String version, @Nullable String schemaUrl) { + this.name = name; + this.version = version; + this.schemaUrl = schemaUrl; + } + + String getName() { + return name; + } + + @Nullable + String getVersion() { + return version; + } + + @Nullable + String getSchemaUrl() { + return schemaUrl; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) return true; + if (!(obj instanceof AuditLoggerKey)) return false; + AuditLoggerKey other = (AuditLoggerKey) obj; + return name.equals(other.name) + && Objects.equals(version, other.version) + && Objects.equals(schemaUrl, other.schemaUrl); + } + + @Override + public int hashCode() { + return Objects.hash(name, version, schemaUrl); + } + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditProviderBuilder.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditProviderBuilder.java new file mode 100644 index 00000000000..9f0fae27abb --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditProviderBuilder.java @@ -0,0 +1,61 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.sdk.common.Clock; +import io.opentelemetry.sdk.resources.Resource; +import java.util.ArrayList; +import java.util.List; + +/** Builder for {@link SdkAuditProvider}. */ +public final class SdkAuditProviderBuilder { + + private Resource resource = Resource.getDefault(); + private Clock clock = Clock.getDefault(); + private final List processors = new ArrayList<>(); + + SdkAuditProviderBuilder() {} + + /** + * Sets the {@link Resource} to be associated with all audit records emitted by this provider. + */ + public SdkAuditProviderBuilder setResource(Resource resource) { + if (resource == null) { + throw new NullPointerException("resource"); + } + this.resource = resource; + return this; + } + + /** Sets the {@link Clock} used for {@code ObservedTimestamp} generation. */ + public SdkAuditProviderBuilder setClock(Clock clock) { + if (clock == null) { + throw new NullPointerException("clock"); + } + this.clock = clock; + return this; + } + + /** + * Adds an {@link AuditRecordProcessor} to the pipeline. Processors are invoked in the order + * they are added. + * + *

The last processor in the chain is responsible for forwarding records to the exporter and + * setting the {@link io.opentelemetry.api.audit.AuditReceipt} on the record. + */ + public SdkAuditProviderBuilder addAuditRecordProcessor(AuditRecordProcessor processor) { + if (processor == null) { + throw new NullPointerException("processor"); + } + processors.add(processor); + return this; + } + + /** Builds and returns the configured {@link SdkAuditProvider}. */ + public SdkAuditProvider build() { + return new SdkAuditProvider(resource, processors, clock); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditRecordBuilder.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditRecordBuilder.java new file mode 100644 index 00000000000..8af4da2892e --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditRecordBuilder.java @@ -0,0 +1,260 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.ActorType; +import io.opentelemetry.api.audit.AuditDeliveryException; +import io.opentelemetry.api.audit.AuditReceipt; +import io.opentelemetry.api.audit.AuditRecordBuilder; +import io.opentelemetry.api.audit.Outcome; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Value; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.internal.AttributesMap; +import java.time.Instant; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; + +/** SDK implementation of {@link AuditRecordBuilder}. */ +final class SdkAuditRecordBuilder implements AuditRecordBuilder { + + private final SdkAuditProvider provider; + private final SdkAuditProvider.AuditLoggerKey loggerKey; + + // Required fields + @Nullable private String recordId; + private long timestampEpochNanos; + @Nullable private String eventName; + @Nullable private Value actor; + @Nullable private ActorType actorType; + @Nullable private String action; + @Nullable private Outcome outcome; + + // Optional fields + private long observedTimestampEpochNanos; + @Nullable private String schemaVersion; + @Nullable private Value targetResource; + @Nullable private String sourceIp; + @Nullable private Value body; + @Nullable private AttributesMap attributes; + @Nullable private byte[] signature; + @Nullable private String algorithm; + @Nullable private byte[] certificate; + @Nullable private byte[] hmac; + @Nullable private String hmacAlgorithm; + private long sequenceNo; + @Nullable private String prevHash; + + SdkAuditRecordBuilder(SdkAuditProvider provider, SdkAuditProvider.AuditLoggerKey loggerKey) { + this.provider = provider; + this.loggerKey = loggerKey; + } + + @Override + public SdkAuditRecordBuilder setRecordId(String recordId) { + this.recordId = recordId; + return this; + } + + @Override + public SdkAuditRecordBuilder setTimestamp(long timestamp, TimeUnit unit) { + this.timestampEpochNanos = unit.toNanos(timestamp); + return this; + } + + @Override + public SdkAuditRecordBuilder setTimestamp(Instant instant) { + this.timestampEpochNanos = + TimeUnit.SECONDS.toNanos(instant.getEpochSecond()) + instant.getNano(); + return this; + } + + @Override + public SdkAuditRecordBuilder setEventName(String eventName) { + this.eventName = eventName; + return this; + } + + @Override + public SdkAuditRecordBuilder setActor(Value actor) { + this.actor = actor; + return this; + } + + @Override + public SdkAuditRecordBuilder setActorType(ActorType actorType) { + this.actorType = actorType; + return this; + } + + @Override + public SdkAuditRecordBuilder setAction(String action) { + this.action = action; + return this; + } + + @Override + public SdkAuditRecordBuilder setOutcome(Outcome outcome) { + this.outcome = outcome; + return this; + } + + @Override + public SdkAuditRecordBuilder setObservedTimestamp(long timestamp, TimeUnit unit) { + this.observedTimestampEpochNanos = unit.toNanos(timestamp); + return this; + } + + @Override + public SdkAuditRecordBuilder setObservedTimestamp(Instant instant) { + this.observedTimestampEpochNanos = + TimeUnit.SECONDS.toNanos(instant.getEpochSecond()) + instant.getNano(); + return this; + } + + @Override + public SdkAuditRecordBuilder setSchemaVersion(String schemaVersion) { + this.schemaVersion = schemaVersion; + return this; + } + + @Override + public SdkAuditRecordBuilder setTargetResource(Value targetResource) { + this.targetResource = targetResource; + return this; + } + + @Override + public SdkAuditRecordBuilder setSourceIp(String sourceIp) { + this.sourceIp = sourceIp; + return this; + } + + @Override + public SdkAuditRecordBuilder setBody(Value body) { + this.body = body; + return this; + } + + @Override + public SdkAuditRecordBuilder setAttribute(AttributeKey key, @Nullable T value) { + if (key == null || value == null) { + return this; + } + if (attributes == null) { + attributes = AttributesMap.create(128, Integer.MAX_VALUE); + } + attributes.put(key, value); + return this; + } + + @Override + public SdkAuditRecordBuilder setSignature(byte[] signature, String algorithm) { + this.signature = signature; + this.algorithm = algorithm; + return this; + } + + @Override + public SdkAuditRecordBuilder setCertificate(byte[] certificate) { + this.certificate = certificate; + return this; + } + + @Override + public SdkAuditRecordBuilder setHmac(byte[] hmac, String algorithm) { + this.hmac = hmac; + this.hmacAlgorithm = algorithm; + return this; + } + + @Override + public SdkAuditRecordBuilder setSequenceNo(long sequenceNo) { + this.sequenceNo = sequenceNo; + return this; + } + + @Override + public SdkAuditRecordBuilder setPrevHash(String prevHash) { + this.prevHash = prevHash; + return this; + } + + @Override + public AuditReceipt emit() { + if (provider.isShutdown()) { + throw new AuditDeliveryException( + "AuditProvider has been shut down; cannot emit audit records"); + } + + // Step 1: Generate RecordId if absent + if (recordId == null || recordId.isEmpty()) { + recordId = UUID.randomUUID().toString(); + } + + // Step 2: Set ObservedTimestamp if absent + if (observedTimestampEpochNanos == 0) { + observedTimestampEpochNanos = provider.getClock().now(); + } + + // Step 3: Validate required fields + validateRequired("Timestamp", timestampEpochNanos != 0, "Timestamp must be set"); + validateRequired("EventName", eventName != null && !eventName.isEmpty(), "EventName must be set and non-empty"); + validateRequired("Actor", actor != null, "Actor must be set"); + validateRequired("ActorType", actorType != null, "ActorType must be set"); + validateRequired("Action", action != null && !action.isEmpty(), "Action must be set and non-empty"); + validateRequired("Outcome", outcome != null, "Outcome must be set"); + + // Step 4+5: Create the mutable record and pass it through all processors. + // Transfer ownership of the attributes map to the record (builder must not be reused). + AttributesMap recordAttributes = this.attributes; + this.attributes = null; + SdkReadWriteAuditRecord rwRecord = + new SdkReadWriteAuditRecord( + provider.getResource(), + loggerKey.getName(), + loggerKey.getVersion(), + loggerKey.getSchemaUrl(), + recordId, + timestampEpochNanos, + observedTimestampEpochNanos, + eventName, + actor, + actorType, + action, + outcome, + targetResource, + sourceIp, + body, + recordAttributes, + signature, + algorithm, + certificate, + hmac, + hmacAlgorithm, + sequenceNo, + prevHash, + schemaVersion); + + provider.getProcessor().onEmit(Context.current(), rwRecord); + + // Step 6+7: Retrieve and return the receipt set by the exporter-wrapping processor + AuditReceipt receipt = rwRecord.getReceipt(); + if (receipt == null) { + throw new AuditDeliveryException( + "Audit pipeline returned no receipt; ensure an AuditRecordExporter is configured"); + } + return receipt; + } + + private static void validateRequired(String field, boolean condition, String message) { + if (!condition) { + throw new IllegalArgumentException( + "AuditRecord validation failed for field '" + field + "': " + message); + } + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditRecordData.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditRecordData.java new file mode 100644 index 00000000000..413bc703adf --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkAuditRecordData.java @@ -0,0 +1,75 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import com.google.auto.value.AutoValue; +import io.opentelemetry.api.audit.ActorType; +import io.opentelemetry.api.audit.Outcome; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.Value; +import io.opentelemetry.sdk.resources.Resource; +import javax.annotation.Nullable; + +/** Immutable AutoValue implementation of {@link AuditRecordData}. */ +@AutoValue +public abstract class SdkAuditRecordData implements AuditRecordData { + + SdkAuditRecordData() {} + + /** Creates a new {@link SdkAuditRecordData}. */ + @SuppressWarnings("TooManyParameters") + public static SdkAuditRecordData create( + Resource resource, + String loggerName, + @Nullable String loggerVersion, + @Nullable String schemaUrl, + String recordId, + long timestampEpochNanos, + long observedTimestampEpochNanos, + String eventName, + Value actor, + ActorType actorType, + String action, + Outcome outcome, + @Nullable Value targetResource, + @Nullable String sourceIp, + @Nullable Value body, + Attributes attributes, + @Nullable byte[] signature, + @Nullable String algorithm, + @Nullable byte[] certificate, + @Nullable byte[] hmac, + @Nullable String hmacAlgorithm, + long sequenceNo, + @Nullable String prevHash, + @Nullable String schemaVersion) { + return new AutoValue_SdkAuditRecordData( + resource, + loggerName, + loggerVersion, + schemaUrl, + recordId, + timestampEpochNanos, + observedTimestampEpochNanos, + eventName, + actor, + actorType, + action, + outcome, + targetResource, + sourceIp, + body, + attributes, + signature, + algorithm, + certificate, + hmac, + hmacAlgorithm, + sequenceNo, + prevHash, + schemaVersion); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkReadWriteAuditRecord.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkReadWriteAuditRecord.java new file mode 100644 index 00000000000..206c66bc37a --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/SdkReadWriteAuditRecord.java @@ -0,0 +1,211 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit; + +import io.opentelemetry.api.audit.AuditReceipt; +import io.opentelemetry.api.audit.ActorType; +import io.opentelemetry.api.audit.Outcome; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.Value; +import io.opentelemetry.api.internal.GuardedBy; +import io.opentelemetry.sdk.common.internal.AttributesMap; +import io.opentelemetry.sdk.resources.Resource; +import javax.annotation.Nullable; +import javax.annotation.concurrent.ThreadSafe; + +/** + * Mutable view of an audit record passed to {@link AuditRecordProcessor#onEmit}. Processors MAY + * enrich the record by adding attributes but MUST NOT modify the mandatory fields. + */ +@ThreadSafe +final class SdkReadWriteAuditRecord implements ReadWriteAuditRecord { + + private final Resource resource; + private final String loggerName; + @Nullable private final String loggerVersion; + @Nullable private final String schemaUrl; + private final String recordId; + private final long timestampEpochNanos; + private final long observedTimestampEpochNanos; + private final String eventName; + private final Value actor; + private final ActorType actorType; + private final String action; + private final Outcome outcome; + @Nullable private final Value targetResource; + @Nullable private final String sourceIp; + @Nullable private final Value body; + @Nullable private final byte[] signature; + @Nullable private final String algorithm; + @Nullable private final byte[] certificate; + @Nullable private final byte[] hmac; + @Nullable private final String hmacAlgorithm; + private final long sequenceNo; + @Nullable private final String prevHash; + @Nullable private final String schemaVersion; + + private final Object lock = new Object(); + + @GuardedBy("lock") + @Nullable + private AttributesMap attributes; + + @GuardedBy("lock") + @Nullable + private AuditReceipt receipt; + + @SuppressWarnings("TooManyParameters") + SdkReadWriteAuditRecord( + Resource resource, + String loggerName, + @Nullable String loggerVersion, + @Nullable String schemaUrl, + String recordId, + long timestampEpochNanos, + long observedTimestampEpochNanos, + String eventName, + Value actor, + ActorType actorType, + String action, + Outcome outcome, + @Nullable Value targetResource, + @Nullable String sourceIp, + @Nullable Value body, + @Nullable AttributesMap attributes, + @Nullable byte[] signature, + @Nullable String algorithm, + @Nullable byte[] certificate, + @Nullable byte[] hmac, + @Nullable String hmacAlgorithm, + long sequenceNo, + @Nullable String prevHash, + @Nullable String schemaVersion) { + this.resource = resource; + this.loggerName = loggerName; + this.loggerVersion = loggerVersion; + this.schemaUrl = schemaUrl; + this.recordId = recordId; + this.timestampEpochNanos = timestampEpochNanos; + this.observedTimestampEpochNanos = observedTimestampEpochNanos; + this.eventName = eventName; + this.actor = actor; + this.actorType = actorType; + this.action = action; + this.outcome = outcome; + this.targetResource = targetResource; + this.sourceIp = sourceIp; + this.body = body; + this.attributes = attributes; + this.signature = signature; + this.algorithm = algorithm; + this.certificate = certificate; + this.hmac = hmac; + this.hmacAlgorithm = hmacAlgorithm; + this.sequenceNo = sequenceNo; + this.prevHash = prevHash; + this.schemaVersion = schemaVersion; + } + + @Override + public ReadWriteAuditRecord setAttribute(AttributeKey key, @Nullable T value) { + if (key == null || value == null) { + return this; + } + synchronized (lock) { + if (attributes == null) { + attributes = AttributesMap.create(128, Integer.MAX_VALUE); + } + attributes.put(key, value); + } + return this; + } + + @Override + public void setReceipt(AuditReceipt receipt) { + synchronized (lock) { + this.receipt = receipt; + } + } + + @Override + @Nullable + public AuditReceipt getReceipt() { + synchronized (lock) { + return receipt; + } + } + + @Override + public AuditRecordData toAuditRecordData() { + final Attributes frozenAttributes; + synchronized (lock) { + frozenAttributes = attributes != null ? attributes.immutableCopy() : Attributes.empty(); + } + return SdkAuditRecordData.create( + resource, + loggerName, + loggerVersion, + schemaUrl, + recordId, + timestampEpochNanos, + observedTimestampEpochNanos, + eventName, + actor, + actorType, + action, + outcome, + targetResource, + sourceIp, + body, + frozenAttributes, + signature, + algorithm, + certificate, + hmac, + hmacAlgorithm, + sequenceNo, + prevHash, + schemaVersion); + } + + // ── Read accessors for processors ───────────────────────────────────────── + + @Override + public String getRecordId() { + return recordId; + } + + @Override + public long getTimestampEpochNanos() { + return timestampEpochNanos; + } + + @Override + public String getEventName() { + return eventName; + } + + @Override + public Value getActor() { + return actor; + } + + @Override + public ActorType getActorType() { + return actorType; + } + + @Override + public String getAction() { + return action; + } + + @Override + public Outcome getOutcome() { + return outcome; + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/BatchAuditRecordProcessor.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/BatchAuditRecordProcessor.java new file mode 100644 index 00000000000..095f3e8ab8f --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/BatchAuditRecordProcessor.java @@ -0,0 +1,257 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit.export; + +import io.opentelemetry.api.audit.AuditDeliveryException; +import io.opentelemetry.api.audit.AuditReceipt; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.audit.AuditExportResult; +import io.opentelemetry.sdk.audit.AuditRecordData; +import io.opentelemetry.sdk.audit.AuditRecordExporter; +import io.opentelemetry.sdk.audit.AuditRecordProcessor; +import io.opentelemetry.sdk.audit.ReadWriteAuditRecord; +import io.opentelemetry.sdk.common.CompletableResultCode; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * An {@link AuditRecordProcessor} that batches records for efficient export while still blocking + * the calling thread until the batch containing the record is acknowledged by the audit sink. + * + *

Records are never dropped when the queue is full: the calling thread is blocked (back-pressure + * is applied to the application) until there is space in the queue or the configured timeout + * elapses. + * + *

Build via {@link BatchAuditRecordProcessorBuilder}: + * + * 

{@code
+ * BatchAuditRecordProcessor processor = BatchAuditRecordProcessor.builder(exporter)
+ *     .setMaxQueueSize(4096)
+ *     .setScheduledDelayMillis(1000)
+ *     .build();
+ * }
+ */ +public final class BatchAuditRecordProcessor implements AuditRecordProcessor { + + private static final Logger logger = + Logger.getLogger(BatchAuditRecordProcessor.class.getName()); + + private final AuditRecordExporter exporter; + private final int maxExportBatchSize; + private final long exportTimeoutMillis; + private final long scheduledDelayMillis; + private final int maxRetryCount; + private final long initialBackoffMillis; + + // Queue of pending records. BlockingQueue to apply back-pressure when full. + private final BlockingQueue queue; + private final int maxQueueSize; + + private final AtomicBoolean isShutdown = new AtomicBoolean(false); + private final AtomicReference flushRequested = new AtomicReference<>(); + private final BlockingQueue signal = new ArrayBlockingQueue<>(1); + + private final Thread worker; + + BatchAuditRecordProcessor( + AuditRecordExporter exporter, + int maxQueueSize, + int maxExportBatchSize, + long scheduledDelayMillis, + long exportTimeoutMillis, + int maxRetryCount, + long initialBackoffMillis) { + this.exporter = exporter; + this.maxQueueSize = maxQueueSize; + this.maxExportBatchSize = maxExportBatchSize; + this.scheduledDelayMillis = scheduledDelayMillis; + this.exportTimeoutMillis = exportTimeoutMillis; + this.maxRetryCount = maxRetryCount; + this.initialBackoffMillis = initialBackoffMillis; + this.queue = new ArrayBlockingQueue<>(maxQueueSize); + this.worker = new Thread(new Worker(), "otel-audit-batch-worker"); + this.worker.setDaemon(true); + this.worker.start(); + } + + /** Returns a new {@link BatchAuditRecordProcessorBuilder} for the given exporter. */ + public static BatchAuditRecordProcessorBuilder builder(AuditRecordExporter exporter) { + return new BatchAuditRecordProcessorBuilder(exporter); + } + + @Override + public void onEmit(Context context, ReadWriteAuditRecord record) { + if (isShutdown.get()) { + throw new AuditDeliveryException( + "BatchAuditRecordProcessor has been shut down; refusing to emit record"); + } + CompletableFuture future = new CompletableFuture<>(); + PendingRecord pending = new PendingRecord(record.toAuditRecordData(), future); + // Block until there is space in the queue (back-pressure as per spec) + boolean offered = false; + try { + offered = queue.offer(pending, exportTimeoutMillis, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AuditDeliveryException("Interrupted while enqueuing audit record", e); + } + if (!offered) { + throw new AuditDeliveryException( + "Audit record queue is full and back-pressure timeout elapsed; record rejected"); + } + // Signal the worker in case it is waiting for work + signal.offer(Boolean.TRUE); + // Block until the worker exports the batch and completes the future + try { + AuditReceipt receipt = future.get(exportTimeoutMillis, TimeUnit.MILLISECONDS); + record.setReceipt(receipt); + } catch (TimeoutException e) { + future.cancel(false); + throw new AuditDeliveryException( + "Timed out waiting for audit export acknowledgement after " + exportTimeoutMillis + "ms", + e); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + if (cause instanceof AuditDeliveryException) { + throw (AuditDeliveryException) cause; + } + throw new AuditDeliveryException("Audit export failed", cause); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AuditDeliveryException("Interrupted while waiting for audit export", e); + } + } + + @Override + public CompletableResultCode shutdown() { + if (!isShutdown.compareAndSet(false, true)) { + return CompletableResultCode.ofSuccess(); + } + CompletableResultCode result = forceFlush(); + result.whenComplete( + () -> { + worker.interrupt(); + exporter.shutdown(); + }); + return result; + } + + @Override + public CompletableResultCode forceFlush() { + CompletableResultCode flushResult = new CompletableResultCode(); + if (flushRequested.compareAndSet(null, flushResult)) { + signal.offer(Boolean.TRUE); + } + CompletableResultCode existing = flushRequested.get(); + return existing != null ? existing : flushResult; + } + + // ── Worker ──────────────────────────────────────────────────────────────── + + private final class Worker implements Runnable { + + @Override + public void run() { + List batch = new ArrayList<>(maxExportBatchSize); + while (!isShutdown.get()) { + try { + // Wait for work or the scheduled delay + signal.poll(scheduledDelayMillis, TimeUnit.MILLISECONDS); + exportBatch(batch); + CompletableResultCode flush = flushRequested.getAndSet(null); + if (flush != null) { + flush.succeed(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + // Drain remaining records on shutdown + exportBatch(batch); + } + + private void exportBatch(List batch) { + batch.clear(); + queue.drainTo(batch, maxExportBatchSize); + if (batch.isEmpty()) { + return; + } + List records = new ArrayList<>(batch.size()); + for (PendingRecord p : batch) { + records.add(p.data); + } + AuditExportResult result = exportWithRetry(records); + if (result.isSuccess()) { + List receipts = result.getReceipts(); + for (int i = 0; i < batch.size(); i++) { + AuditReceipt receipt = i < receipts.size() ? receipts.get(i) : null; + if (receipt != null) { + batch.get(i).future.complete(receipt); + } else { + batch.get(i).future.completeExceptionally( + new AuditDeliveryException("Exporter returned no receipt for record " + i)); + } + } + } else { + AuditDeliveryException ex = + new AuditDeliveryException( + "Audit export failed after " + maxRetryCount + " retries", + result.getFailure()); + for (PendingRecord p : batch) { + p.future.completeExceptionally(ex); + } + } + } + + private AuditExportResult exportWithRetry(List records) { + long backoff = initialBackoffMillis; + for (int attempt = 0; attempt <= maxRetryCount; attempt++) { + AuditExportResult result = exporter.export(records); + if (result.isSuccess()) { + return result; + } + if (attempt < maxRetryCount) { + logger.log( + Level.WARNING, + "Audit export attempt {0} failed; retrying in {1}ms", + new Object[] {attempt + 1, backoff}); + try { + Thread.sleep(backoff); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return AuditExportResult.failure(e); + } + backoff = Math.min(backoff * 2, exportTimeoutMillis); + } + } + return AuditExportResult.failure( + new AuditDeliveryException("Max retries exhausted: " + maxRetryCount)); + } + } + + // ── Holder ──────────────────────────────────────────────────────────────── + + private static final class PendingRecord { + final AuditRecordData data; + final CompletableFuture future; + + PendingRecord(AuditRecordData data, CompletableFuture future) { + this.data = data; + this.future = future; + } + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/BatchAuditRecordProcessorBuilder.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/BatchAuditRecordProcessorBuilder.java new file mode 100644 index 00000000000..e5b781d6a32 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/BatchAuditRecordProcessorBuilder.java @@ -0,0 +1,99 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit.export; + +import static java.util.Objects.requireNonNull; + +import io.opentelemetry.sdk.audit.AuditRecordExporter; + +/** Builder for {@link BatchAuditRecordProcessor}. */ +public final class BatchAuditRecordProcessorBuilder { + + static final int DEFAULT_MAX_QUEUE_SIZE = 2048; + static final int DEFAULT_MAX_EXPORT_BATCH_SIZE = 512; + static final long DEFAULT_SCHEDULED_DELAY_MILLIS = 5_000; + static final long DEFAULT_EXPORT_TIMEOUT_MILLIS = 30_000; + static final int DEFAULT_MAX_RETRY_COUNT = 5; + static final long DEFAULT_INITIAL_BACKOFF_MILLIS = 1_000; + + private final AuditRecordExporter exporter; + private int maxQueueSize = DEFAULT_MAX_QUEUE_SIZE; + private int maxExportBatchSize = DEFAULT_MAX_EXPORT_BATCH_SIZE; + private long scheduledDelayMillis = DEFAULT_SCHEDULED_DELAY_MILLIS; + private long exportTimeoutMillis = DEFAULT_EXPORT_TIMEOUT_MILLIS; + private int maxRetryCount = DEFAULT_MAX_RETRY_COUNT; + private long initialBackoffMillis = DEFAULT_INITIAL_BACKOFF_MILLIS; + + BatchAuditRecordProcessorBuilder(AuditRecordExporter exporter) { + this.exporter = requireNonNull(exporter, "exporter"); + } + + /** + * Sets the maximum number of records held in the queue before back-pressure is applied to the + * calling thread. Default: {@value #DEFAULT_MAX_QUEUE_SIZE}. + */ + public BatchAuditRecordProcessorBuilder setMaxQueueSize(int maxQueueSize) { + this.maxQueueSize = maxQueueSize; + return this; + } + + /** + * Sets the maximum number of records per exported batch. Default: {@value + * #DEFAULT_MAX_EXPORT_BATCH_SIZE}. + */ + public BatchAuditRecordProcessorBuilder setMaxExportBatchSize(int maxExportBatchSize) { + this.maxExportBatchSize = maxExportBatchSize; + return this; + } + + /** + * Sets the delay in milliseconds between two consecutive exports when the batch is not full. + * Default: {@value #DEFAULT_SCHEDULED_DELAY_MILLIS} ms. + */ + public BatchAuditRecordProcessorBuilder setScheduledDelayMillis(long scheduledDelayMillis) { + this.scheduledDelayMillis = scheduledDelayMillis; + return this; + } + + /** + * Sets the maximum time in milliseconds allowed for a single export call before it is considered + * a failure. Default: {@value #DEFAULT_EXPORT_TIMEOUT_MILLIS} ms. + */ + public BatchAuditRecordProcessorBuilder setExportTimeoutMillis(long exportTimeoutMillis) { + this.exportTimeoutMillis = exportTimeoutMillis; + return this; + } + + /** + * Sets the maximum number of export retry attempts before surfacing a hard error. Default: + * {@value #DEFAULT_MAX_RETRY_COUNT}. + */ + public BatchAuditRecordProcessorBuilder setMaxRetryCount(int maxRetryCount) { + this.maxRetryCount = maxRetryCount; + return this; + } + + /** + * Sets the initial back-off delay in milliseconds between retries; doubled on each attempt. + * Default: {@value #DEFAULT_INITIAL_BACKOFF_MILLIS} ms. + */ + public BatchAuditRecordProcessorBuilder setInitialBackoffMillis(long initialBackoffMillis) { + this.initialBackoffMillis = initialBackoffMillis; + return this; + } + + /** Builds and returns the configured {@link BatchAuditRecordProcessor}. */ + public BatchAuditRecordProcessor build() { + return new BatchAuditRecordProcessor( + exporter, + maxQueueSize, + maxExportBatchSize, + scheduledDelayMillis, + exportTimeoutMillis, + maxRetryCount, + initialBackoffMillis); + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/InMemoryAuditRecordExporter.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/InMemoryAuditRecordExporter.java new file mode 100644 index 00000000000..9e79537c637 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/InMemoryAuditRecordExporter.java @@ -0,0 +1,90 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit.export; + +import io.opentelemetry.api.audit.AuditReceipt; +import io.opentelemetry.sdk.audit.AuditExportResult; +import io.opentelemetry.sdk.audit.AuditRecordData; +import io.opentelemetry.sdk.audit.AuditRecordExporter; +import io.opentelemetry.sdk.common.CompletableResultCode; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * An in-memory {@link AuditRecordExporter} for use in tests. + * + *

Stores all exported {@link AuditRecordData}s in a list and returns synthetic {@link + * AuditReceipt}s with empty integrity hashes (since there is no real audit sink). + * + * 

{@code
+ * InMemoryAuditRecordExporter exporter = InMemoryAuditRecordExporter.create();
+ * SdkAuditProvider provider = SdkAuditProvider.builder()
+ *     .addAuditRecordProcessor(SimpleAuditRecordProcessor.create(exporter))
+ *     .build();
+ * // ...
+ * List records = exporter.getFinishedAuditRecords();
+ * }
+ */ +public final class InMemoryAuditRecordExporter implements AuditRecordExporter { + + private final List finishedRecords = new ArrayList<>(); + private final AtomicBoolean isShutdown = new AtomicBoolean(false); + private final Object lock = new Object(); + + private InMemoryAuditRecordExporter() {} + + /** Creates a new {@link InMemoryAuditRecordExporter}. */ + public static InMemoryAuditRecordExporter create() { + return new InMemoryAuditRecordExporter(); + } + + @Override + public AuditExportResult export(Collection records) { + if (isShutdown.get()) { + return AuditExportResult.failure(new IllegalStateException("Exporter has been shut down")); + } + List receipts = new ArrayList<>(records.size()); + synchronized (lock) { + for (AuditRecordData record : records) { + finishedRecords.add(record); + // Synthetic receipt: echo the recordId; integrity hash is empty (no real sink) + receipts.add(AuditReceipt.create(record.getRecordId(), "", 0)); + } + } + return AuditExportResult.success(receipts); + } + + @Override + public CompletableResultCode flush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + isShutdown.set(true); + return CompletableResultCode.ofSuccess(); + } + + /** + * Returns an unmodifiable snapshot of all exported {@link AuditRecordData}s in the order they + * were exported. + */ + public List getFinishedAuditRecords() { + synchronized (lock) { + return Collections.unmodifiableList(new ArrayList<>(finishedRecords)); + } + } + + /** Clears the list of finished records. */ + public void reset() { + synchronized (lock) { + finishedRecords.clear(); + } + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/SimpleAuditRecordProcessor.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/SimpleAuditRecordProcessor.java new file mode 100644 index 00000000000..c1ad684df82 --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/export/SimpleAuditRecordProcessor.java @@ -0,0 +1,117 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.sdk.audit.export; + +import static java.util.Objects.requireNonNull; + +import io.opentelemetry.api.audit.AuditDeliveryException; +import io.opentelemetry.api.audit.AuditReceipt; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.audit.AuditExportResult; +import io.opentelemetry.sdk.audit.AuditRecordData; +import io.opentelemetry.sdk.audit.AuditRecordExporter; +import io.opentelemetry.sdk.audit.AuditRecordProcessor; +import io.opentelemetry.sdk.audit.ReadWriteAuditRecord; +import io.opentelemetry.sdk.common.CompletableResultCode; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * An {@link AuditRecordProcessor} that passes each {@link AuditRecordData} directly to the + * configured {@link AuditRecordExporter} synchronously on the calling thread. + * + *

This is the default processor for synchronous {@code emit()} calls. It guarantees that + * {@code emit()} blocks until the exporter has acknowledged the record and returns the {@link + * AuditReceipt} from the sink. + * + *

For high-volume scenarios, consider {@link BatchAuditRecordProcessor}. + */ +public final class SimpleAuditRecordProcessor implements AuditRecordProcessor { + + private static final Logger logger = + Logger.getLogger(SimpleAuditRecordProcessor.class.getName()); + + private final AuditRecordExporter exporter; + private final Object exporterLock = new Object(); + private final AtomicBoolean isShutdown = new AtomicBoolean(false); + + private SimpleAuditRecordProcessor(AuditRecordExporter exporter) { + this.exporter = exporter; + } + + /** + * Creates a new {@link SimpleAuditRecordProcessor} that synchronously exports to the given + * {@link AuditRecordExporter}. + */ + public static SimpleAuditRecordProcessor create(AuditRecordExporter exporter) { + requireNonNull(exporter, "exporter"); + return new SimpleAuditRecordProcessor(exporter); + } + + @Override + public void onEmit(Context context, ReadWriteAuditRecord record) { + if (isShutdown.get()) { + throw new AuditDeliveryException( + "SimpleAuditRecordProcessor has been shut down; refusing to emit record"); + } + List batch = Collections.singletonList(record.toAuditRecordData()); + AuditExportResult result; + synchronized (exporterLock) { + result = exporter.export(batch); + } + if (!result.isSuccess()) { + Throwable cause = result.getFailure(); + String msg = "Audit record export failed"; + if (cause != null) { + throw new AuditDeliveryException(msg, cause); + } + throw new AuditDeliveryException(msg); + } + List receipts = result.getReceipts(); + if (receipts.isEmpty()) { + throw new AuditDeliveryException( + "Exporter returned success but no AuditReceipt; check exporter implementation"); + } + record.setReceipt(receipts.get(0)); + } + + @Override + public CompletableResultCode shutdown() { + if (isShutdown.getAndSet(true)) { + return CompletableResultCode.ofSuccess(); + } + CompletableResultCode result = new CompletableResultCode(); + CompletableResultCode shutdownResult = exporter.shutdown(); + shutdownResult.whenComplete( + () -> { + if (shutdownResult.isSuccess()) { + result.succeed(); + } else { + logger.log(Level.WARNING, "Exporter failed to shut down cleanly"); + result.fail(); + } + }); + return result; + } + + @Override + public CompletableResultCode forceFlush() { + return exporter.flush(); + } + + /** Returns the configured {@link AuditRecordExporter}. */ + public AuditRecordExporter getExporter() { + return exporter; + } + + @Override + public String toString() { + return "SimpleAuditRecordProcessor{exporter=" + exporter + '}'; + } +} diff --git a/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/package-info.java b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/package-info.java new file mode 100644 index 00000000000..ae58a6cbf1e --- /dev/null +++ b/sdk/audit/src/main/java/io/opentelemetry/sdk/audit/package-info.java @@ -0,0 +1,10 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +/** OpenTelemetry Audit Logging SDK. */ +@ParametersAreNonnullByDefault +package io.opentelemetry.sdk.audit; + +import javax.annotation.ParametersAreNonnullByDefault; diff --git a/settings.gradle.kts b/settings.gradle.kts index b87a127857e..7eaf99aae99 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -44,6 +44,7 @@ include(":exporters:sender:okhttp") include(":exporters:logging") include(":exporters:logging-otlp") include(":exporters:otlp:all") +include(":exporters:otlp:audit") include(":exporters:otlp:common") include(":exporters:otlp:profiles") include(":exporters:otlp:testing-internal") @@ -59,6 +60,7 @@ include(":opencensus-shim") include(":opentracing-shim") include(":perf-harness") include(":sdk:all") +include(":sdk:audit") include(":sdk:common") include(":sdk:logs") include(":sdk:metrics")