adnanhemani commented on code in PR #4836: URL: https://github.com/apache/polaris/pull/4836#discussion_r3463648051
########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); + + for (PolarisOtelAttribute attribute : PolarisOtelAttribute.values()) { Review Comment: **Performance: `Enum.values()` clones the backing array on every event.** Per the JLS, `Enum.values()` returns a fresh defensive copy of the backing array each call — here a 34-element `PolarisOtelAttribute[]` per `onEvent()` invocation. Rather, we should cache `private static final PolarisOtelAttribute[] VALUES = PolarisOtelAttribute.values()` and iterate over that instead. ########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); + + for (PolarisOtelAttribute attribute : PolarisOtelAttribute.values()) { + attribute.forward(this, attributes, event); + } + + return attributes.build(); + } + + private enum PolarisOtelAttribute { + CATALOG_NAME(string(CATALOG_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_NAME)), + NAMESPACE(string(NAMESPACE_ATTRIBUTE_NAME, EventAttributes.NAMESPACE, Namespace::toString)), + NAMESPACE_FQN(string(NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_FQN)), + PARENT_NAMESPACE_FQN( + string(PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.PARENT_NAMESPACE_FQN)), + TABLE_NAME(string(TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.TABLE_NAME)), + TABLE_IDENTIFIER(string(TABLE_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.TABLE_IDENTIFIER)), + DERIVED_TABLE_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedTableIdentifier(event) + .ifPresent( + tableIdentifier -> + attributes.put(TABLE_IDENTIFIER_ATTRIBUTE_NAME, tableIdentifier))), + VIEW_NAME(string(VIEW_NAME_ATTRIBUTE_NAME, EventAttributes.VIEW_NAME)), + VIEW_IDENTIFIER(string(VIEW_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.VIEW_IDENTIFIER)), + DERIVED_VIEW_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedViewIdentifier(event) + .ifPresent( + viewIdentifier -> + attributes.put(VIEW_IDENTIFIER_ATTRIBUTE_NAME, viewIdentifier))), + PRINCIPAL_NAME(string(PRINCIPAL_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_NAME)), + PRINCIPAL_ROLE_NAME( + string(PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_ROLE_NAME)), + CATALOG_ROLE_NAME(string(CATALOG_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_ROLE_NAME)), + PRIVILEGE(string(PRIVILEGE_ATTRIBUTE_NAME, EventAttributes.PRIVILEGE, value -> value.name())), + CASCADE(booleanAttribute(CASCADE_ATTRIBUTE_NAME, EventAttributes.CASCADE)), + WAREHOUSE(string(WAREHOUSE_ATTRIBUTE_NAME, EventAttributes.WAREHOUSE)), + ACCESS_DELEGATION_MODE( + string(ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME, EventAttributes.ACCESS_DELEGATION_MODE)), + IF_NONE_MATCH(string(IF_NONE_MATCH_ATTRIBUTE_NAME, EventAttributes.IF_NONE_MATCH_STRING)), + SNAPSHOTS(string(SNAPSHOTS_ATTRIBUTE_NAME, EventAttributes.SNAPSHOTS)), + PURGE_REQUESTED( + booleanAttribute(PURGE_REQUESTED_ATTRIBUTE_NAME, EventAttributes.PURGE_REQUESTED)), + TASK_ENTITY_ID(longAttribute(TASK_ENTITY_ID_ATTRIBUTE_NAME, EventAttributes.TASK_ENTITY_ID)), + TASK_ATTEMPT(longAttribute(TASK_ATTEMPT_ATTRIBUTE_NAME, EventAttributes.TASK_ATTEMPT)), + TASK_SUCCESS(booleanAttribute(TASK_SUCCESS_ATTRIBUTE_NAME, EventAttributes.TASK_SUCCESS)), + HTTP_METHOD(string(HTTP_METHOD_ATTRIBUTE_NAME, EventAttributes.HTTP_METHOD)), + REQUEST_URI(string(REQUEST_URI_ATTRIBUTE_NAME, EventAttributes.REQUEST_URI)), + NAMESPACE_NAME(string(NAMESPACE_NAME_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_NAME)), + GENERIC_TABLE_NAME( + string(GENERIC_TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.GENERIC_TABLE_NAME)), + POLICY_NAME(string(POLICY_NAME_ATTRIBUTE_NAME, EventAttributes.POLICY_NAME)), + POLICY_TYPE(string(POLICY_TYPE_ATTRIBUTE_NAME, EventAttributes.POLICY_TYPE)), + TARGET_NAME(string(TARGET_NAME_ATTRIBUTE_NAME, EventAttributes.TARGET_NAME)), + DETACH_ALL(booleanAttribute(DETACH_ALL_ATTRIBUTE_NAME, EventAttributes.DETACH_ALL)), + GRANT_RESOURCE((listener, attributes, event) -> listener.putGrantResource(attributes, event)), + ADD_GRANT_REQUEST(json(ADD_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.ADD_GRANT_REQUEST)), + REVOKE_GRANT_REQUEST( + json(REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.REVOKE_GRANT_REQUEST)); + + private final AttributeForwarder forwarder; + + PolarisOtelAttribute(AttributeForwarder forwarder) { + this.forwarder = forwarder; + } + + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event) { + forwarder.forward(listener, attributes, event); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return string(logAttributeName, eventAttributeKey, Object::toString); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey, Function<T, String> mapper) { + return (listener, attributes, event) -> + listener.putStringAttribute( + attributes, event, logAttributeName, eventAttributeKey, mapper); + } + + private static AttributeForwarder booleanAttribute( + String logAttributeName, AttributeKey<Boolean> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putBooleanAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T extends Number> AttributeForwarder longAttribute( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putLongAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T> AttributeForwarder json( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putJsonAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + } + + @FunctionalInterface + private interface AttributeForwarder { + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event); + } + + private <T> void putStringAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey, + Function<T, String> mapper) { + event + .attributes() + .get(eventAttributeKey) + .map(mapper) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putBooleanAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<Boolean> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(booleanKey(logAttributeName), value)); + } + + private <T extends Number> void putLongAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(longKey(logAttributeName), value.longValue())); + } + + private <T> void putJsonAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putGrantResource(AttributesBuilder attributes, PolarisEvent event) { + Optional<GrantResource> grantResource = event.attributes().get(EventAttributes.GRANT_RESOURCE); + grantResource + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(GRANT_RESOURCE_ATTRIBUTE_NAME, value)); + grantResource + .map(GrantResource::getType) + .map(Enum::name) Review Comment: **Bug: `GrantResource` type case mismatch between the two grant attributes.** `Enum::name` returns the Java constant name — `"TABLE"`, `"CATALOG_ROLE"`, etc. But `TypeEnum` has `@JsonValue` on `toString()` returning lowercase values (`"table"`, `"catalog_role"`), so the JSON blob in `polaris.grant.resource` contains `"type":"table"` while `polaris.grant.resource.type` contains `"TABLE"`. Any consumer correlating both attributes will see inconsistent case on every grant event. Fix: replace `Enum::name` with `Object::toString`. ########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); + + for (PolarisOtelAttribute attribute : PolarisOtelAttribute.values()) { + attribute.forward(this, attributes, event); + } + + return attributes.build(); + } + + private enum PolarisOtelAttribute { + CATALOG_NAME(string(CATALOG_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_NAME)), + NAMESPACE(string(NAMESPACE_ATTRIBUTE_NAME, EventAttributes.NAMESPACE, Namespace::toString)), + NAMESPACE_FQN(string(NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_FQN)), + PARENT_NAMESPACE_FQN( + string(PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.PARENT_NAMESPACE_FQN)), + TABLE_NAME(string(TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.TABLE_NAME)), + TABLE_IDENTIFIER(string(TABLE_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.TABLE_IDENTIFIER)), + DERIVED_TABLE_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedTableIdentifier(event) + .ifPresent( + tableIdentifier -> + attributes.put(TABLE_IDENTIFIER_ATTRIBUTE_NAME, tableIdentifier))), + VIEW_NAME(string(VIEW_NAME_ATTRIBUTE_NAME, EventAttributes.VIEW_NAME)), + VIEW_IDENTIFIER(string(VIEW_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.VIEW_IDENTIFIER)), + DERIVED_VIEW_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedViewIdentifier(event) + .ifPresent( + viewIdentifier -> + attributes.put(VIEW_IDENTIFIER_ATTRIBUTE_NAME, viewIdentifier))), + PRINCIPAL_NAME(string(PRINCIPAL_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_NAME)), + PRINCIPAL_ROLE_NAME( + string(PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_ROLE_NAME)), + CATALOG_ROLE_NAME(string(CATALOG_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_ROLE_NAME)), + PRIVILEGE(string(PRIVILEGE_ATTRIBUTE_NAME, EventAttributes.PRIVILEGE, value -> value.name())), + CASCADE(booleanAttribute(CASCADE_ATTRIBUTE_NAME, EventAttributes.CASCADE)), + WAREHOUSE(string(WAREHOUSE_ATTRIBUTE_NAME, EventAttributes.WAREHOUSE)), + ACCESS_DELEGATION_MODE( + string(ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME, EventAttributes.ACCESS_DELEGATION_MODE)), + IF_NONE_MATCH(string(IF_NONE_MATCH_ATTRIBUTE_NAME, EventAttributes.IF_NONE_MATCH_STRING)), + SNAPSHOTS(string(SNAPSHOTS_ATTRIBUTE_NAME, EventAttributes.SNAPSHOTS)), + PURGE_REQUESTED( + booleanAttribute(PURGE_REQUESTED_ATTRIBUTE_NAME, EventAttributes.PURGE_REQUESTED)), + TASK_ENTITY_ID(longAttribute(TASK_ENTITY_ID_ATTRIBUTE_NAME, EventAttributes.TASK_ENTITY_ID)), + TASK_ATTEMPT(longAttribute(TASK_ATTEMPT_ATTRIBUTE_NAME, EventAttributes.TASK_ATTEMPT)), + TASK_SUCCESS(booleanAttribute(TASK_SUCCESS_ATTRIBUTE_NAME, EventAttributes.TASK_SUCCESS)), + HTTP_METHOD(string(HTTP_METHOD_ATTRIBUTE_NAME, EventAttributes.HTTP_METHOD)), + REQUEST_URI(string(REQUEST_URI_ATTRIBUTE_NAME, EventAttributes.REQUEST_URI)), + NAMESPACE_NAME(string(NAMESPACE_NAME_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_NAME)), + GENERIC_TABLE_NAME( + string(GENERIC_TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.GENERIC_TABLE_NAME)), + POLICY_NAME(string(POLICY_NAME_ATTRIBUTE_NAME, EventAttributes.POLICY_NAME)), + POLICY_TYPE(string(POLICY_TYPE_ATTRIBUTE_NAME, EventAttributes.POLICY_TYPE)), + TARGET_NAME(string(TARGET_NAME_ATTRIBUTE_NAME, EventAttributes.TARGET_NAME)), + DETACH_ALL(booleanAttribute(DETACH_ALL_ATTRIBUTE_NAME, EventAttributes.DETACH_ALL)), + GRANT_RESOURCE((listener, attributes, event) -> listener.putGrantResource(attributes, event)), + ADD_GRANT_REQUEST(json(ADD_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.ADD_GRANT_REQUEST)), + REVOKE_GRANT_REQUEST( + json(REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.REVOKE_GRANT_REQUEST)); + + private final AttributeForwarder forwarder; + + PolarisOtelAttribute(AttributeForwarder forwarder) { + this.forwarder = forwarder; + } + + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event) { + forwarder.forward(listener, attributes, event); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return string(logAttributeName, eventAttributeKey, Object::toString); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey, Function<T, String> mapper) { + return (listener, attributes, event) -> + listener.putStringAttribute( + attributes, event, logAttributeName, eventAttributeKey, mapper); + } + + private static AttributeForwarder booleanAttribute( + String logAttributeName, AttributeKey<Boolean> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putBooleanAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T extends Number> AttributeForwarder longAttribute( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putLongAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T> AttributeForwarder json( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putJsonAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + } + + @FunctionalInterface + private interface AttributeForwarder { + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event); + } + + private <T> void putStringAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey, + Function<T, String> mapper) { + event + .attributes() + .get(eventAttributeKey) + .map(mapper) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); Review Comment: **Performance: `stringKey()`/`booleanKey()`/`longKey()` allocate a new `AttributeKey` object on every event.** `InternalAttributeKeyImpl.create()` calls `new InternalAttributeKeyImpl<>()` unconditionally — confirmed by bytecode, no cache or intern pool. The [OTel SDK docs](https://javadoc.io/doc/io.opentelemetry/opentelemetry-api/1.0.1/io/opentelemetry/api/trace/Span.html) explicitly say: *"It is strongly recommended to pre-allocate your keys."* With ~25 present attributes per event, each `onEvent()` allocates 25+ fresh key objects. The same issue affects the `put(String, String)` convenience overloads used in `toLogAttributes()` lines 134–148, which internally call `stringKey()` too. Fix: declare all attribute keys as `private static final io.opentelemetry.api.common.AttributeKey<?>` fields alongside the existing `static final String` name constants, initialized once at class load, and pass those directly to `attributes.put(key, value)`. ########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); + + for (PolarisOtelAttribute attribute : PolarisOtelAttribute.values()) { + attribute.forward(this, attributes, event); + } + + return attributes.build(); + } + + private enum PolarisOtelAttribute { + CATALOG_NAME(string(CATALOG_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_NAME)), + NAMESPACE(string(NAMESPACE_ATTRIBUTE_NAME, EventAttributes.NAMESPACE, Namespace::toString)), + NAMESPACE_FQN(string(NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_FQN)), + PARENT_NAMESPACE_FQN( + string(PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.PARENT_NAMESPACE_FQN)), + TABLE_NAME(string(TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.TABLE_NAME)), + TABLE_IDENTIFIER(string(TABLE_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.TABLE_IDENTIFIER)), + DERIVED_TABLE_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedTableIdentifier(event) + .ifPresent( + tableIdentifier -> + attributes.put(TABLE_IDENTIFIER_ATTRIBUTE_NAME, tableIdentifier))), + VIEW_NAME(string(VIEW_NAME_ATTRIBUTE_NAME, EventAttributes.VIEW_NAME)), + VIEW_IDENTIFIER(string(VIEW_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.VIEW_IDENTIFIER)), + DERIVED_VIEW_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedViewIdentifier(event) + .ifPresent( + viewIdentifier -> + attributes.put(VIEW_IDENTIFIER_ATTRIBUTE_NAME, viewIdentifier))), + PRINCIPAL_NAME(string(PRINCIPAL_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_NAME)), + PRINCIPAL_ROLE_NAME( + string(PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_ROLE_NAME)), + CATALOG_ROLE_NAME(string(CATALOG_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_ROLE_NAME)), + PRIVILEGE(string(PRIVILEGE_ATTRIBUTE_NAME, EventAttributes.PRIVILEGE, value -> value.name())), + CASCADE(booleanAttribute(CASCADE_ATTRIBUTE_NAME, EventAttributes.CASCADE)), + WAREHOUSE(string(WAREHOUSE_ATTRIBUTE_NAME, EventAttributes.WAREHOUSE)), + ACCESS_DELEGATION_MODE( + string(ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME, EventAttributes.ACCESS_DELEGATION_MODE)), + IF_NONE_MATCH(string(IF_NONE_MATCH_ATTRIBUTE_NAME, EventAttributes.IF_NONE_MATCH_STRING)), + SNAPSHOTS(string(SNAPSHOTS_ATTRIBUTE_NAME, EventAttributes.SNAPSHOTS)), + PURGE_REQUESTED( + booleanAttribute(PURGE_REQUESTED_ATTRIBUTE_NAME, EventAttributes.PURGE_REQUESTED)), + TASK_ENTITY_ID(longAttribute(TASK_ENTITY_ID_ATTRIBUTE_NAME, EventAttributes.TASK_ENTITY_ID)), + TASK_ATTEMPT(longAttribute(TASK_ATTEMPT_ATTRIBUTE_NAME, EventAttributes.TASK_ATTEMPT)), + TASK_SUCCESS(booleanAttribute(TASK_SUCCESS_ATTRIBUTE_NAME, EventAttributes.TASK_SUCCESS)), + HTTP_METHOD(string(HTTP_METHOD_ATTRIBUTE_NAME, EventAttributes.HTTP_METHOD)), + REQUEST_URI(string(REQUEST_URI_ATTRIBUTE_NAME, EventAttributes.REQUEST_URI)), + NAMESPACE_NAME(string(NAMESPACE_NAME_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_NAME)), + GENERIC_TABLE_NAME( + string(GENERIC_TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.GENERIC_TABLE_NAME)), + POLICY_NAME(string(POLICY_NAME_ATTRIBUTE_NAME, EventAttributes.POLICY_NAME)), + POLICY_TYPE(string(POLICY_TYPE_ATTRIBUTE_NAME, EventAttributes.POLICY_TYPE)), + TARGET_NAME(string(TARGET_NAME_ATTRIBUTE_NAME, EventAttributes.TARGET_NAME)), + DETACH_ALL(booleanAttribute(DETACH_ALL_ATTRIBUTE_NAME, EventAttributes.DETACH_ALL)), + GRANT_RESOURCE((listener, attributes, event) -> listener.putGrantResource(attributes, event)), + ADD_GRANT_REQUEST(json(ADD_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.ADD_GRANT_REQUEST)), + REVOKE_GRANT_REQUEST( + json(REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.REVOKE_GRANT_REQUEST)); + + private final AttributeForwarder forwarder; + + PolarisOtelAttribute(AttributeForwarder forwarder) { + this.forwarder = forwarder; + } + + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event) { + forwarder.forward(listener, attributes, event); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return string(logAttributeName, eventAttributeKey, Object::toString); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey, Function<T, String> mapper) { + return (listener, attributes, event) -> + listener.putStringAttribute( + attributes, event, logAttributeName, eventAttributeKey, mapper); + } + + private static AttributeForwarder booleanAttribute( + String logAttributeName, AttributeKey<Boolean> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putBooleanAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T extends Number> AttributeForwarder longAttribute( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putLongAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T> AttributeForwarder json( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putJsonAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + } + + @FunctionalInterface + private interface AttributeForwarder { + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event); + } + + private <T> void putStringAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey, + Function<T, String> mapper) { + event + .attributes() + .get(eventAttributeKey) + .map(mapper) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putBooleanAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<Boolean> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(booleanKey(logAttributeName), value)); + } + + private <T extends Number> void putLongAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(longKey(logAttributeName), value.longValue())); + } + + private <T> void putJsonAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putGrantResource(AttributesBuilder attributes, PolarisEvent event) { + Optional<GrantResource> grantResource = event.attributes().get(EventAttributes.GRANT_RESOURCE); + grantResource + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(GRANT_RESOURCE_ATTRIBUTE_NAME, value)); + grantResource + .map(GrantResource::getType) + .map(Enum::name) + .ifPresent(value -> attributes.put(GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME, value)); + } + + private Optional<String> toJsonString(Object value) { + try { + return Optional.of(objectMapper.writeValueAsString(value)); + } catch (JsonProcessingException e) { + LOGGER.debug("Could not serialize Polaris event attribute {}", value, e); Review Comment: An error that fails deserialization and therefore drops information should probably be at a `WARN` level instead of `DEBUG`. ########## runtime/service/src/test/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListenerTest.java: ########## @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SAMPLED_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_ROLES_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ADD_GRANT_REQUEST_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_ROLE_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_CATEGORY_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRINCIPAL_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRIVILEGE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PURGE_REQUESTED_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REALM_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REQUEST_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_IDENTIFIER_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_NAME_ATTRIBUTE_NAME; +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.logs.export.LogRecordExporter; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.catalog.Namespace; +import org.apache.polaris.core.admin.model.AddGrantRequest; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.core.admin.model.TableGrant; +import org.apache.polaris.core.admin.model.TablePrivilege; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.core.entity.PolarisPrivilege; +import org.apache.polaris.service.events.EventAttributeMap; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.PolarisEventType; +import org.junit.jupiter.api.Test; + +class OpenTelemetryEventListenerTest { + private static final String TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736"; + private static final String SPAN_ID = "00f067aa0ba902b7"; + + @Test + void shouldEmitCreateTableEventAttributes() { + CapturingLogRecordExporter exporter = new CapturingLogRecordExporter(); + OpenTelemetryEventListener listener = createListener(exporter); + + listener.onEvent( + new PolarisEvent( + PolarisEventType.AFTER_CREATE_TABLE, + metadata(), + new EventAttributeMap() + .put(EventAttributes.CATALOG_NAME, "test_catalog") + .put(EventAttributes.NAMESPACE, Namespace.of("test_namespace")) + .put(EventAttributes.TABLE_NAME, "test_table") + .put(EventAttributes.PURGE_REQUESTED, true))); + + LogRecordData record = exporter.records().getFirst(); + + assertThat(record.getBodyValue().asString()).isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getEventName()).isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getSeverity()).isEqualTo(Severity.INFO); + assertThat(record.getSpanContext().getTraceId()).isEqualTo(TRACE_ID); + assertThat(record.getSpanContext().getSpanId()).isEqualTo(SPAN_ID); + assertThat(record.getAttributes().get(stringKey(EVENT_TYPE_ATTRIBUTE_NAME))) + .isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getAttributes().get(stringKey(EVENT_CATEGORY_ATTRIBUTE_NAME))) + .isEqualTo("TABLE"); + assertThat(record.getAttributes().get(stringKey(REALM_ID_ATTRIBUTE_NAME))) + .isEqualTo("test_realm"); + assertThat(record.getAttributes().get(stringKey(REQUEST_ID_ATTRIBUTE_NAME))) + .isEqualTo("request-1"); + assertThat(record.getAttributes().get(stringKey(ACTOR_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_user"); + assertThat(record.getAttributes().get(stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME))) + .containsExactlyInAnyOrder("role1", "role2"); + assertThat(record.getAttributes().get(stringKey(CATALOG_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_catalog"); + assertThat(record.getAttributes().get(stringKey(TABLE_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_table"); + assertThat(record.getAttributes().get(stringKey(TABLE_IDENTIFIER_ATTRIBUTE_NAME))) Review Comment: **No test for the `derivedTableIdentifier` guard.** `derivedTableIdentifier` returns empty when `TABLE_IDENTIFIER` is already present in the attribute map, preventing a double-write. There is no test that puts both `TABLE_IDENTIFIER` and `NAMESPACE + TABLE_NAME` into the map and verifies the pre-set identifier wins. Without this, a refactor that removes or reorders the guard would go undetected. ########## runtime/service/src/test/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListenerTest.java: ########## @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SAMPLED_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_ROLES_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ADD_GRANT_REQUEST_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_ROLE_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_CATEGORY_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRINCIPAL_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRIVILEGE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PURGE_REQUESTED_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REALM_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REQUEST_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_IDENTIFIER_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_NAME_ATTRIBUTE_NAME; +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.logs.export.LogRecordExporter; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.catalog.Namespace; +import org.apache.polaris.core.admin.model.AddGrantRequest; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.core.admin.model.TableGrant; +import org.apache.polaris.core.admin.model.TablePrivilege; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.core.entity.PolarisPrivilege; +import org.apache.polaris.service.events.EventAttributeMap; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.PolarisEventType; +import org.junit.jupiter.api.Test; + +class OpenTelemetryEventListenerTest { + private static final String TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736"; + private static final String SPAN_ID = "00f067aa0ba902b7"; + + @Test + void shouldEmitCreateTableEventAttributes() { + CapturingLogRecordExporter exporter = new CapturingLogRecordExporter(); + OpenTelemetryEventListener listener = createListener(exporter); + + listener.onEvent( + new PolarisEvent( + PolarisEventType.AFTER_CREATE_TABLE, + metadata(), + new EventAttributeMap() + .put(EventAttributes.CATALOG_NAME, "test_catalog") + .put(EventAttributes.NAMESPACE, Namespace.of("test_namespace")) + .put(EventAttributes.TABLE_NAME, "test_table") + .put(EventAttributes.PURGE_REQUESTED, true))); + + LogRecordData record = exporter.records().getFirst(); Review Comment: somewhat nit - but would like to see this throughout the file: **Missing record count assertion before `getFirst()`.** Each test calls `exporter.records().getFirst()` without first asserting exactly one record was emitted. If `onEvent` silently produces nothing (e.g., a bad `SdkLoggerProvider` setup, an unhandled exception inside `toLogAttributes`), this throws `NoSuchElementException` with no message about what failed. Prefer `assertThat(exporter.records()).hasSize(1)` before reading the first element. ########## runtime/service/src/test/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListenerTest.java: ########## @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SAMPLED_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_ROLES_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ADD_GRANT_REQUEST_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_ROLE_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_CATEGORY_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRINCIPAL_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRIVILEGE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PURGE_REQUESTED_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REALM_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REQUEST_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_IDENTIFIER_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_NAME_ATTRIBUTE_NAME; +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.logs.export.LogRecordExporter; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.catalog.Namespace; +import org.apache.polaris.core.admin.model.AddGrantRequest; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.core.admin.model.TableGrant; +import org.apache.polaris.core.admin.model.TablePrivilege; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.core.entity.PolarisPrivilege; +import org.apache.polaris.service.events.EventAttributeMap; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.PolarisEventType; +import org.junit.jupiter.api.Test; + +class OpenTelemetryEventListenerTest { + private static final String TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736"; + private static final String SPAN_ID = "00f067aa0ba902b7"; + + @Test + void shouldEmitCreateTableEventAttributes() { + CapturingLogRecordExporter exporter = new CapturingLogRecordExporter(); + OpenTelemetryEventListener listener = createListener(exporter); + + listener.onEvent( + new PolarisEvent( + PolarisEventType.AFTER_CREATE_TABLE, + metadata(), + new EventAttributeMap() + .put(EventAttributes.CATALOG_NAME, "test_catalog") + .put(EventAttributes.NAMESPACE, Namespace.of("test_namespace")) + .put(EventAttributes.TABLE_NAME, "test_table") + .put(EventAttributes.PURGE_REQUESTED, true))); + + LogRecordData record = exporter.records().getFirst(); + + assertThat(record.getBodyValue().asString()).isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getEventName()).isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getSeverity()).isEqualTo(Severity.INFO); + assertThat(record.getSpanContext().getTraceId()).isEqualTo(TRACE_ID); + assertThat(record.getSpanContext().getSpanId()).isEqualTo(SPAN_ID); + assertThat(record.getAttributes().get(stringKey(EVENT_TYPE_ATTRIBUTE_NAME))) + .isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getAttributes().get(stringKey(EVENT_CATEGORY_ATTRIBUTE_NAME))) + .isEqualTo("TABLE"); + assertThat(record.getAttributes().get(stringKey(REALM_ID_ATTRIBUTE_NAME))) + .isEqualTo("test_realm"); + assertThat(record.getAttributes().get(stringKey(REQUEST_ID_ATTRIBUTE_NAME))) + .isEqualTo("request-1"); + assertThat(record.getAttributes().get(stringKey(ACTOR_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_user"); + assertThat(record.getAttributes().get(stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME))) + .containsExactlyInAnyOrder("role1", "role2"); + assertThat(record.getAttributes().get(stringKey(CATALOG_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_catalog"); + assertThat(record.getAttributes().get(stringKey(TABLE_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_table"); + assertThat(record.getAttributes().get(stringKey(TABLE_IDENTIFIER_ATTRIBUTE_NAME))) + .isEqualTo("test_namespace.test_table"); + assertThat(record.getAttributes().get(booleanKey(PURGE_REQUESTED_ATTRIBUTE_NAME))).isTrue(); + } + + @Test + void shouldEmitPrincipalEventAttributesWithoutOverwritingActor() { + CapturingLogRecordExporter exporter = new CapturingLogRecordExporter(); + OpenTelemetryEventListener listener = createListener(exporter); + + listener.onEvent( + new PolarisEvent( + PolarisEventType.AFTER_GET_PRINCIPAL, + metadata(), + new EventAttributeMap().put(EventAttributes.PRINCIPAL_NAME, "target_principal"))); + + LogRecordData record = exporter.records().getFirst(); + + assertThat(record.getAttributes().get(stringKey(ACTOR_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_user"); + assertThat(record.getAttributes().get(stringKey(PRINCIPAL_NAME_ATTRIBUTE_NAME))) + .isEqualTo("target_principal"); + } + + @Test + void shouldEmitGrantEventAttributes() { + CapturingLogRecordExporter exporter = new CapturingLogRecordExporter(); + OpenTelemetryEventListener listener = createListener(exporter); + TableGrant grant = + new TableGrant( + List.of("test_namespace"), + "test_table", + TablePrivilege.TABLE_WRITE_DATA, + GrantResource.TypeEnum.TABLE); + + listener.onEvent( + new PolarisEvent( + PolarisEventType.AFTER_ADD_GRANT_TO_CATALOG_ROLE, + metadata(), + new EventAttributeMap() + .put(EventAttributes.CATALOG_NAME, "test_catalog") + .put(EventAttributes.CATALOG_ROLE_NAME, "test_catalog_role") + .put(EventAttributes.PRIVILEGE, PolarisPrivilege.TABLE_WRITE_DATA) + .put(EventAttributes.GRANT_RESOURCE, grant))); + + LogRecordData record = exporter.records().getFirst(); + + assertThat(record.getAttributes().get(stringKey(EVENT_TYPE_ATTRIBUTE_NAME))) + .isEqualTo("AFTER_ADD_GRANT_TO_CATALOG_ROLE"); + assertThat(record.getAttributes().get(stringKey(CATALOG_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_catalog"); + assertThat(record.getAttributes().get(stringKey(CATALOG_ROLE_NAME_ATTRIBUTE_NAME))) + .isEqualTo("test_catalog_role"); + assertThat(record.getAttributes().get(stringKey(PRIVILEGE_ATTRIBUTE_NAME))) + .isEqualTo("TABLE_WRITE_DATA"); + assertThat(record.getAttributes().get(stringKey(GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME))) + .isEqualTo("TABLE"); + assertThat(record.getAttributes().get(stringKey(GRANT_RESOURCE_ATTRIBUTE_NAME))) + .contains("test_namespace", "test_table", "TABLE_WRITE_DATA"); Review Comment: **Loose substring check will not catch serialization regressions.** `.contains("test_namespace", "test_table", "TABLE_WRITE_DATA")` on a `String` only verifies substring presence in the raw JSON. A field rename in `TableGrant`'s serialized output (e.g., `tableNamespace` → `namespace`) or a new JSON format would still pass. Consider deserializing the JSON with `ObjectMapper` and asserting typed field values, or at minimum asserting that the field *keys* appear alongside the values. ########## runtime/service/src/test/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListenerTest.java: ########## @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SAMPLED_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ACTOR_ROLES_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.ADD_GRANT_REQUEST_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.CATALOG_ROLE_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_CATEGORY_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.EVENT_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRINCIPAL_NAME_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PRIVILEGE_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.PURGE_REQUESTED_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REALM_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.REQUEST_ID_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_IDENTIFIER_ATTRIBUTE_NAME; +import static org.apache.polaris.service.events.listeners.opentelemetry.OpenTelemetryEventListener.TABLE_NAME_ATTRIBUTE_NAME; +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.logs.export.LogRecordExporter; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.catalog.Namespace; +import org.apache.polaris.core.admin.model.AddGrantRequest; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.core.admin.model.TableGrant; +import org.apache.polaris.core.admin.model.TablePrivilege; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.core.entity.PolarisPrivilege; +import org.apache.polaris.service.events.EventAttributeMap; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.PolarisEventType; +import org.junit.jupiter.api.Test; + +class OpenTelemetryEventListenerTest { + private static final String TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736"; + private static final String SPAN_ID = "00f067aa0ba902b7"; + + @Test + void shouldEmitCreateTableEventAttributes() { + CapturingLogRecordExporter exporter = new CapturingLogRecordExporter(); + OpenTelemetryEventListener listener = createListener(exporter); + + listener.onEvent( + new PolarisEvent( + PolarisEventType.AFTER_CREATE_TABLE, + metadata(), + new EventAttributeMap() + .put(EventAttributes.CATALOG_NAME, "test_catalog") + .put(EventAttributes.NAMESPACE, Namespace.of("test_namespace")) + .put(EventAttributes.TABLE_NAME, "test_table") + .put(EventAttributes.PURGE_REQUESTED, true))); + + LogRecordData record = exporter.records().getFirst(); + + assertThat(record.getBodyValue().asString()).isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getEventName()).isEqualTo("AFTER_CREATE_TABLE"); + assertThat(record.getSeverity()).isEqualTo(Severity.INFO); + assertThat(record.getSpanContext().getTraceId()).isEqualTo(TRACE_ID); Review Comment: **No coverage for `toOpenTelemetryContext` failure paths.** The happy path (valid trace/span IDs) is covered here, but three branches are not tested: 1. Empty OTel context map → expect `record.getSpanContext().isValid()` to be false. 2. A one-char `traceFlags` string → exercises the `StringIndexOutOfBoundsException` bug that escapes the `catch (IllegalArgumentException)`. 3. Malformed `traceId`/`spanId` → exercises the `spanContext.isValid() == false` branch and the WARN log path. In general, there are no "sad path" tests. I can't consider this PR as complete/maintainable without those. ########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); + + for (PolarisOtelAttribute attribute : PolarisOtelAttribute.values()) { + attribute.forward(this, attributes, event); + } + + return attributes.build(); + } + + private enum PolarisOtelAttribute { + CATALOG_NAME(string(CATALOG_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_NAME)), + NAMESPACE(string(NAMESPACE_ATTRIBUTE_NAME, EventAttributes.NAMESPACE, Namespace::toString)), + NAMESPACE_FQN(string(NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_FQN)), + PARENT_NAMESPACE_FQN( + string(PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.PARENT_NAMESPACE_FQN)), + TABLE_NAME(string(TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.TABLE_NAME)), + TABLE_IDENTIFIER(string(TABLE_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.TABLE_IDENTIFIER)), + DERIVED_TABLE_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedTableIdentifier(event) + .ifPresent( + tableIdentifier -> + attributes.put(TABLE_IDENTIFIER_ATTRIBUTE_NAME, tableIdentifier))), + VIEW_NAME(string(VIEW_NAME_ATTRIBUTE_NAME, EventAttributes.VIEW_NAME)), + VIEW_IDENTIFIER(string(VIEW_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.VIEW_IDENTIFIER)), + DERIVED_VIEW_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedViewIdentifier(event) + .ifPresent( + viewIdentifier -> + attributes.put(VIEW_IDENTIFIER_ATTRIBUTE_NAME, viewIdentifier))), + PRINCIPAL_NAME(string(PRINCIPAL_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_NAME)), + PRINCIPAL_ROLE_NAME( + string(PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_ROLE_NAME)), + CATALOG_ROLE_NAME(string(CATALOG_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_ROLE_NAME)), + PRIVILEGE(string(PRIVILEGE_ATTRIBUTE_NAME, EventAttributes.PRIVILEGE, value -> value.name())), + CASCADE(booleanAttribute(CASCADE_ATTRIBUTE_NAME, EventAttributes.CASCADE)), + WAREHOUSE(string(WAREHOUSE_ATTRIBUTE_NAME, EventAttributes.WAREHOUSE)), + ACCESS_DELEGATION_MODE( + string(ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME, EventAttributes.ACCESS_DELEGATION_MODE)), + IF_NONE_MATCH(string(IF_NONE_MATCH_ATTRIBUTE_NAME, EventAttributes.IF_NONE_MATCH_STRING)), + SNAPSHOTS(string(SNAPSHOTS_ATTRIBUTE_NAME, EventAttributes.SNAPSHOTS)), + PURGE_REQUESTED( + booleanAttribute(PURGE_REQUESTED_ATTRIBUTE_NAME, EventAttributes.PURGE_REQUESTED)), + TASK_ENTITY_ID(longAttribute(TASK_ENTITY_ID_ATTRIBUTE_NAME, EventAttributes.TASK_ENTITY_ID)), + TASK_ATTEMPT(longAttribute(TASK_ATTEMPT_ATTRIBUTE_NAME, EventAttributes.TASK_ATTEMPT)), + TASK_SUCCESS(booleanAttribute(TASK_SUCCESS_ATTRIBUTE_NAME, EventAttributes.TASK_SUCCESS)), + HTTP_METHOD(string(HTTP_METHOD_ATTRIBUTE_NAME, EventAttributes.HTTP_METHOD)), + REQUEST_URI(string(REQUEST_URI_ATTRIBUTE_NAME, EventAttributes.REQUEST_URI)), + NAMESPACE_NAME(string(NAMESPACE_NAME_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_NAME)), + GENERIC_TABLE_NAME( + string(GENERIC_TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.GENERIC_TABLE_NAME)), + POLICY_NAME(string(POLICY_NAME_ATTRIBUTE_NAME, EventAttributes.POLICY_NAME)), + POLICY_TYPE(string(POLICY_TYPE_ATTRIBUTE_NAME, EventAttributes.POLICY_TYPE)), + TARGET_NAME(string(TARGET_NAME_ATTRIBUTE_NAME, EventAttributes.TARGET_NAME)), + DETACH_ALL(booleanAttribute(DETACH_ALL_ATTRIBUTE_NAME, EventAttributes.DETACH_ALL)), + GRANT_RESOURCE((listener, attributes, event) -> listener.putGrantResource(attributes, event)), + ADD_GRANT_REQUEST(json(ADD_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.ADD_GRANT_REQUEST)), + REVOKE_GRANT_REQUEST( + json(REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.REVOKE_GRANT_REQUEST)); + + private final AttributeForwarder forwarder; + + PolarisOtelAttribute(AttributeForwarder forwarder) { + this.forwarder = forwarder; + } + + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event) { + forwarder.forward(listener, attributes, event); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return string(logAttributeName, eventAttributeKey, Object::toString); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey, Function<T, String> mapper) { + return (listener, attributes, event) -> + listener.putStringAttribute( + attributes, event, logAttributeName, eventAttributeKey, mapper); + } + + private static AttributeForwarder booleanAttribute( + String logAttributeName, AttributeKey<Boolean> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putBooleanAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T extends Number> AttributeForwarder longAttribute( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putLongAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T> AttributeForwarder json( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putJsonAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + } + + @FunctionalInterface + private interface AttributeForwarder { + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event); + } + + private <T> void putStringAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey, + Function<T, String> mapper) { + event + .attributes() + .get(eventAttributeKey) + .map(mapper) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putBooleanAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<Boolean> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(booleanKey(logAttributeName), value)); + } + + private <T extends Number> void putLongAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(longKey(logAttributeName), value.longValue())); + } + + private <T> void putJsonAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putGrantResource(AttributesBuilder attributes, PolarisEvent event) { + Optional<GrantResource> grantResource = event.attributes().get(EventAttributes.GRANT_RESOURCE); + grantResource + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(GRANT_RESOURCE_ATTRIBUTE_NAME, value)); + grantResource + .map(GrantResource::getType) + .map(Enum::name) + .ifPresent(value -> attributes.put(GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME, value)); + } + + private Optional<String> toJsonString(Object value) { + try { + return Optional.of(objectMapper.writeValueAsString(value)); + } catch (JsonProcessingException e) { + LOGGER.debug("Could not serialize Polaris event attribute {}", value, e); + return Optional.empty(); + } + } + + private Optional<String> derivedTableIdentifier(PolarisEvent event) { Review Comment: **Reuse: `derivedTableIdentifier`/`derivedViewIdentifier` duplicate private logic already in `PolarisPersistenceEventListener`.** `PolarisPersistenceEventListener.resolveNamespaceAndTableName` (private static) does the identical `Namespace + tableName → TableIdentifier.of(...).toString()` derivation. Because it is private, this listener copies the logic. A third listener needing this will add a third copy. Fix: extract to a package-visible static utility method so all listeners can share it. ########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); Review Comment: Why are the `openTelemetryContext` variables both here and in the `toOpenTelemetryContext` function? Also, if returns back with an empty return value, it is still possible that the `openTelemetryContext` variables will be returned here - which I don't believe is correct. A consumer reading these variables may believe valid trace correlation exists when it does not. ########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); + + for (PolarisOtelAttribute attribute : PolarisOtelAttribute.values()) { + attribute.forward(this, attributes, event); + } + + return attributes.build(); + } + + private enum PolarisOtelAttribute { + CATALOG_NAME(string(CATALOG_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_NAME)), + NAMESPACE(string(NAMESPACE_ATTRIBUTE_NAME, EventAttributes.NAMESPACE, Namespace::toString)), + NAMESPACE_FQN(string(NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_FQN)), + PARENT_NAMESPACE_FQN( + string(PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.PARENT_NAMESPACE_FQN)), + TABLE_NAME(string(TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.TABLE_NAME)), + TABLE_IDENTIFIER(string(TABLE_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.TABLE_IDENTIFIER)), + DERIVED_TABLE_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedTableIdentifier(event) + .ifPresent( + tableIdentifier -> + attributes.put(TABLE_IDENTIFIER_ATTRIBUTE_NAME, tableIdentifier))), + VIEW_NAME(string(VIEW_NAME_ATTRIBUTE_NAME, EventAttributes.VIEW_NAME)), + VIEW_IDENTIFIER(string(VIEW_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.VIEW_IDENTIFIER)), + DERIVED_VIEW_IDENTIFIER( + (listener, attributes, event) -> + listener + .derivedViewIdentifier(event) + .ifPresent( + viewIdentifier -> + attributes.put(VIEW_IDENTIFIER_ATTRIBUTE_NAME, viewIdentifier))), + PRINCIPAL_NAME(string(PRINCIPAL_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_NAME)), + PRINCIPAL_ROLE_NAME( + string(PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.PRINCIPAL_ROLE_NAME)), + CATALOG_ROLE_NAME(string(CATALOG_ROLE_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_ROLE_NAME)), + PRIVILEGE(string(PRIVILEGE_ATTRIBUTE_NAME, EventAttributes.PRIVILEGE, value -> value.name())), + CASCADE(booleanAttribute(CASCADE_ATTRIBUTE_NAME, EventAttributes.CASCADE)), + WAREHOUSE(string(WAREHOUSE_ATTRIBUTE_NAME, EventAttributes.WAREHOUSE)), + ACCESS_DELEGATION_MODE( + string(ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME, EventAttributes.ACCESS_DELEGATION_MODE)), + IF_NONE_MATCH(string(IF_NONE_MATCH_ATTRIBUTE_NAME, EventAttributes.IF_NONE_MATCH_STRING)), + SNAPSHOTS(string(SNAPSHOTS_ATTRIBUTE_NAME, EventAttributes.SNAPSHOTS)), + PURGE_REQUESTED( + booleanAttribute(PURGE_REQUESTED_ATTRIBUTE_NAME, EventAttributes.PURGE_REQUESTED)), + TASK_ENTITY_ID(longAttribute(TASK_ENTITY_ID_ATTRIBUTE_NAME, EventAttributes.TASK_ENTITY_ID)), + TASK_ATTEMPT(longAttribute(TASK_ATTEMPT_ATTRIBUTE_NAME, EventAttributes.TASK_ATTEMPT)), + TASK_SUCCESS(booleanAttribute(TASK_SUCCESS_ATTRIBUTE_NAME, EventAttributes.TASK_SUCCESS)), + HTTP_METHOD(string(HTTP_METHOD_ATTRIBUTE_NAME, EventAttributes.HTTP_METHOD)), + REQUEST_URI(string(REQUEST_URI_ATTRIBUTE_NAME, EventAttributes.REQUEST_URI)), + NAMESPACE_NAME(string(NAMESPACE_NAME_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_NAME)), + GENERIC_TABLE_NAME( + string(GENERIC_TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.GENERIC_TABLE_NAME)), + POLICY_NAME(string(POLICY_NAME_ATTRIBUTE_NAME, EventAttributes.POLICY_NAME)), + POLICY_TYPE(string(POLICY_TYPE_ATTRIBUTE_NAME, EventAttributes.POLICY_TYPE)), + TARGET_NAME(string(TARGET_NAME_ATTRIBUTE_NAME, EventAttributes.TARGET_NAME)), + DETACH_ALL(booleanAttribute(DETACH_ALL_ATTRIBUTE_NAME, EventAttributes.DETACH_ALL)), + GRANT_RESOURCE((listener, attributes, event) -> listener.putGrantResource(attributes, event)), + ADD_GRANT_REQUEST(json(ADD_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.ADD_GRANT_REQUEST)), + REVOKE_GRANT_REQUEST( + json(REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME, EventAttributes.REVOKE_GRANT_REQUEST)); + + private final AttributeForwarder forwarder; + + PolarisOtelAttribute(AttributeForwarder forwarder) { + this.forwarder = forwarder; + } + + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event) { + forwarder.forward(listener, attributes, event); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return string(logAttributeName, eventAttributeKey, Object::toString); + } + + private static <T> AttributeForwarder string( + String logAttributeName, AttributeKey<T> eventAttributeKey, Function<T, String> mapper) { + return (listener, attributes, event) -> + listener.putStringAttribute( + attributes, event, logAttributeName, eventAttributeKey, mapper); + } + + private static AttributeForwarder booleanAttribute( + String logAttributeName, AttributeKey<Boolean> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putBooleanAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T extends Number> AttributeForwarder longAttribute( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putLongAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + + private static <T> AttributeForwarder json( + String logAttributeName, AttributeKey<T> eventAttributeKey) { + return (listener, attributes, event) -> + listener.putJsonAttribute(attributes, event, logAttributeName, eventAttributeKey); + } + } + + @FunctionalInterface + private interface AttributeForwarder { + void forward( + OpenTelemetryEventListener listener, AttributesBuilder attributes, PolarisEvent event); + } + + private <T> void putStringAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey, + Function<T, String> mapper) { + event + .attributes() + .get(eventAttributeKey) + .map(mapper) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putBooleanAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<Boolean> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(booleanKey(logAttributeName), value)); + } + + private <T extends Number> void putLongAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .ifPresent(value -> attributes.put(longKey(logAttributeName), value.longValue())); + } + + private <T> void putJsonAttribute( + AttributesBuilder attributes, + PolarisEvent event, + String logAttributeName, + AttributeKey<T> eventAttributeKey) { + event + .attributes() + .get(eventAttributeKey) + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(stringKey(logAttributeName), value)); + } + + private void putGrantResource(AttributesBuilder attributes, PolarisEvent event) { + Optional<GrantResource> grantResource = event.attributes().get(EventAttributes.GRANT_RESOURCE); + grantResource + .flatMap(this::toJsonString) + .ifPresent(value -> attributes.put(GRANT_RESOURCE_ATTRIBUTE_NAME, value)); + grantResource + .map(GrantResource::getType) + .map(Enum::name) + .ifPresent(value -> attributes.put(GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME, value)); + } + + private Optional<String> toJsonString(Object value) { + try { + return Optional.of(objectMapper.writeValueAsString(value)); + } catch (JsonProcessingException e) { + LOGGER.debug("Could not serialize Polaris event attribute {}", value, e); + return Optional.empty(); + } + } + + private Optional<String> derivedTableIdentifier(PolarisEvent event) { + if (event.attributes().contains(EventAttributes.TABLE_IDENTIFIER)) { + return Optional.empty(); + } + Optional<Namespace> namespace = event.attributes().get(EventAttributes.NAMESPACE); + Optional<String> tableName = event.attributes().get(EventAttributes.TABLE_NAME); + if (namespace.isPresent() && tableName.isPresent()) { + return Optional.of(TableIdentifier.of(namespace.get(), tableName.get()).toString()); + } + return Optional.empty(); + } + + private Optional<String> derivedViewIdentifier(PolarisEvent event) { + if (event.attributes().contains(EventAttributes.VIEW_IDENTIFIER)) { + return Optional.empty(); + } + Optional<Namespace> namespace = event.attributes().get(EventAttributes.NAMESPACE); + Optional<String> viewName = event.attributes().get(EventAttributes.VIEW_NAME); + if (namespace.isPresent() && viewName.isPresent()) { + return Optional.of(TableIdentifier.of(namespace.get(), viewName.get()).toString()); + } + return Optional.empty(); + } + + private Optional<Context> toOpenTelemetryContext(PolarisEvent event) { + Map<String, String> contextValues = event.metadata().openTelemetryContext(); + String traceId = contextValues.get(OPEN_TELEMETRY_TRACE_ID_KEY); + String spanId = contextValues.get(OPEN_TELEMETRY_SPAN_ID_KEY); + if (traceId == null || spanId == null) { + return Optional.empty(); + } + + try { + TraceFlags traceFlags = + Optional.ofNullable(contextValues.get(OPEN_TELEMETRY_TRACE_FLAGS_KEY)) + .map(flags -> TraceFlags.fromHex(flags, 0)) Review Comment: **Defensive bug: `TraceFlags.fromHex` throws `StringIndexOutOfBoundsException` for short strings, which escapes the `catch (IllegalArgumentException)`.** `fromHex` calls `src.charAt(srcOffset)` and `src.charAt(srcOffset + 1)`. A zero- or one-char string throws `StringIndexOutOfBoundsException`, which extends `IndexOutOfBoundsException` — a sibling of `IllegalArgumentException`, not a subtype — so it escapes this catch block entirely and crashes event delivery. Not reachable via `PolarisEventMetadataFactory` (which always produces a 2-char hex string from `TraceFlags.asHex()`), but reachable from any test or code that manually constructs `PolarisEventMetadata` with a partial context map. Fix: add a length guard before calling `fromHex`, or widen the catch to also cover `IndexOutOfBoundsException`. ########## runtime/service/src/main/java/org/apache/polaris/service/events/listeners/opentelemetry/OpenTelemetryEventListener.java: ########## @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.polaris.service.events.listeners.opentelemetry; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringArrayKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_SPAN_ID_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_FLAGS_KEY; +import static org.apache.polaris.service.events.PolarisEventMetadata.OPEN_TELEMETRY_TRACE_ID_KEY; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.logs.Logger; +import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Context; +import io.smallrye.common.annotation.Identifier; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.polaris.core.admin.model.GrantResource; +import org.apache.polaris.service.events.AttributeKey; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventMetadata; +import org.apache.polaris.service.events.listeners.PolarisEventListener; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +@Identifier("opentelemetry") +public class OpenTelemetryEventListener implements PolarisEventListener { + static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.polaris.events"; + + static final String EVENT_TYPE_ATTRIBUTE_NAME = "polaris.event.type"; + static final String EVENT_CATEGORY_ATTRIBUTE_NAME = "polaris.event.category"; + static final String REALM_ID_ATTRIBUTE_NAME = "polaris.realm.id"; + static final String REQUEST_ID_ATTRIBUTE_NAME = "polaris.request.id"; + static final String ACTOR_NAME_ATTRIBUTE_NAME = "polaris.actor.name"; + static final String ACTOR_ROLES_ATTRIBUTE_NAME = "polaris.actor.roles"; + static final String PRINCIPAL_NAME_ATTRIBUTE_NAME = "polaris.principal.name"; + + static final String CATALOG_NAME_ATTRIBUTE_NAME = "polaris.catalog.name"; + static final String NAMESPACE_ATTRIBUTE_NAME = "polaris.namespace"; + static final String NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.namespace.fqn"; + static final String PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME = "polaris.parent_namespace.fqn"; + static final String TABLE_NAME_ATTRIBUTE_NAME = "polaris.table.name"; + static final String TABLE_IDENTIFIER_ATTRIBUTE_NAME = "polaris.table.identifier"; + static final String VIEW_NAME_ATTRIBUTE_NAME = "polaris.view.name"; + static final String VIEW_IDENTIFIER_ATTRIBUTE_NAME = "polaris.view.identifier"; + static final String PRINCIPAL_ROLE_NAME_ATTRIBUTE_NAME = "polaris.principal_role.name"; + static final String CATALOG_ROLE_NAME_ATTRIBUTE_NAME = "polaris.catalog_role.name"; + static final String PRIVILEGE_ATTRIBUTE_NAME = "polaris.privilege"; + static final String GRANT_RESOURCE_ATTRIBUTE_NAME = "polaris.grant.resource"; + static final String GRANT_RESOURCE_TYPE_ATTRIBUTE_NAME = "polaris.grant.resource.type"; + static final String ADD_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.add_request"; + static final String REVOKE_GRANT_REQUEST_ATTRIBUTE_NAME = "polaris.grant.revoke_request"; + static final String CASCADE_ATTRIBUTE_NAME = "polaris.cascade"; + static final String WAREHOUSE_ATTRIBUTE_NAME = "polaris.warehouse"; + static final String ACCESS_DELEGATION_MODE_ATTRIBUTE_NAME = "polaris.access_delegation_mode"; + static final String IF_NONE_MATCH_ATTRIBUTE_NAME = "polaris.if_none_match"; + static final String SNAPSHOTS_ATTRIBUTE_NAME = "polaris.snapshots"; + static final String PURGE_REQUESTED_ATTRIBUTE_NAME = "polaris.purge_requested"; + static final String TASK_ENTITY_ID_ATTRIBUTE_NAME = "polaris.task.entity_id"; + static final String TASK_ATTEMPT_ATTRIBUTE_NAME = "polaris.task.attempt"; + static final String TASK_SUCCESS_ATTRIBUTE_NAME = "polaris.task.success"; + static final String HTTP_METHOD_ATTRIBUTE_NAME = "polaris.http.method"; + static final String REQUEST_URI_ATTRIBUTE_NAME = "polaris.http.request_uri"; + static final String NAMESPACE_NAME_ATTRIBUTE_NAME = "polaris.namespace.name"; + static final String GENERIC_TABLE_NAME_ATTRIBUTE_NAME = "polaris.generic_table.name"; + static final String POLICY_NAME_ATTRIBUTE_NAME = "polaris.policy.name"; + static final String POLICY_TYPE_ATTRIBUTE_NAME = "polaris.policy.type"; + static final String TARGET_NAME_ATTRIBUTE_NAME = "polaris.target.name"; + static final String DETACH_ALL_ATTRIBUTE_NAME = "polaris.detach_all"; + + private static final org.slf4j.Logger LOGGER = + LoggerFactory.getLogger(OpenTelemetryEventListener.class); + + private final Logger openTelemetryLogger; + private final ObjectMapper objectMapper; + + @Inject + public OpenTelemetryEventListener(OpenTelemetry openTelemetry, ObjectMapper objectMapper) { + this.openTelemetryLogger = + openTelemetry.getLogsBridge().loggerBuilder(INSTRUMENTATION_SCOPE_NAME).build(); + this.objectMapper = objectMapper; + } + + @Override + public void onEvent(PolarisEvent event) { + var logRecordBuilder = + openTelemetryLogger + .logRecordBuilder() + .setTimestamp(event.metadata().timestamp()) + .setSeverity(Severity.INFO) + .setSeverityText("INFO") + .setEventName(event.type().name()) + .setBody(event.type().name()) + .setAllAttributes(toLogAttributes(event)); + toOpenTelemetryContext(event).ifPresent(logRecordBuilder::setContext); + logRecordBuilder.emit(); + } + + private Attributes toLogAttributes(PolarisEvent event) { + AttributesBuilder attributes = Attributes.builder(); + attributes.put(EVENT_TYPE_ATTRIBUTE_NAME, event.type().name()); + attributes.put(EVENT_CATEGORY_ATTRIBUTE_NAME, event.type().category().name()); + + PolarisEventMetadata metadata = event.metadata(); + attributes.put(REALM_ID_ATTRIBUTE_NAME, metadata.realmId()); + metadata + .requestId() + .ifPresent(requestId -> attributes.put(REQUEST_ID_ATTRIBUTE_NAME, requestId)); + metadata + .user() + .ifPresent( + principal -> { + attributes.put(ACTOR_NAME_ATTRIBUTE_NAME, principal.getName()); + attributes.put( + stringArrayKey(ACTOR_ROLES_ATTRIBUTE_NAME), List.copyOf(principal.getRoles())); + }); + metadata.openTelemetryContext().forEach(attributes::put); + + for (PolarisOtelAttribute attribute : PolarisOtelAttribute.values()) { + attribute.forward(this, attributes, event); + } + + return attributes.build(); + } + + private enum PolarisOtelAttribute { + CATALOG_NAME(string(CATALOG_NAME_ATTRIBUTE_NAME, EventAttributes.CATALOG_NAME)), + NAMESPACE(string(NAMESPACE_ATTRIBUTE_NAME, EventAttributes.NAMESPACE, Namespace::toString)), + NAMESPACE_FQN(string(NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.NAMESPACE_FQN)), + PARENT_NAMESPACE_FQN( + string(PARENT_NAMESPACE_FQN_ATTRIBUTE_NAME, EventAttributes.PARENT_NAMESPACE_FQN)), + TABLE_NAME(string(TABLE_NAME_ATTRIBUTE_NAME, EventAttributes.TABLE_NAME)), + TABLE_IDENTIFIER(string(TABLE_IDENTIFIER_ATTRIBUTE_NAME, EventAttributes.TABLE_IDENTIFIER)), + DERIVED_TABLE_IDENTIFIER( Review Comment: nit: maybe we shouldn't have this logic here - and create the derived table identifier during the event emission time itself... would love to see if anyone else has thoughts on this as well. Not sure that putting this logic here is the right place for it. I understand if we don't want to have scope creep here, but maybe put a TODO here to fix this eventually. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
