somandal commented on code in PR #16869:
URL: https://github.com/apache/pinot/pull/16869#discussion_r2372955732
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java:
##########
@@ -269,6 +269,12 @@ public class BrokerMeter implements AbstractMetrics.Meter {
public static final BrokerMeter RLS_FILTERS_APPLIED =
create("RLS_FILTERS_APPLIED", "queries", false);
+ // Audit logging metrics
+ public static final BrokerMeter AUDIT_REQUEST_FAILURES =
create("AUDIT_REQUEST_FAILURES", "failures", true);
+ public static final BrokerMeter AUDIT_RESPONSE_FAILURES =
create("AUDIT_RESPONSE_FAILURES", "failures", true);
+ public static final BrokerMeter AUDIT_REQUEST_PAYLOAD_TRUNCATED =
create("AUDIT_REQUEST_PAYLOAD_TRUNCATED",
Review Comment:
nit: is the unit type as "truncated" a good unit? maybe call it something
else? same for controller meter
##########
pinot-common/src/main/java/org/apache/pinot/common/audit/AuditLogFilter.java:
##########
@@ -109,20 +122,30 @@ public void filter(ContainerRequestContext
requestContext, ContainerResponseCont
if (requestId == null) {
return;
}
- try {
- long durationMs = (System.nanoTime() - auditContext.getStartTimeNanos())
/ 1_000_000;
-
- final AuditEvent auditEvent = new AuditEvent().setRequestId(requestId)
- .setTimestamp(Instant.now().toString())
- .setResponseCode(responseContext.getStatus())
- .setDurationMs(durationMs)
- .setEndpoint(requestContext.getUriInfo().getPath())
- .setMethod(requestContext.getMethod());
+ long durationMs = (System.nanoTime() - auditContext.getStartTimeNanos()) /
1_000_000;
Review Comment:
nit: let's use a constant for 1_000_000
also, is it possible to use `TimeUnit.NANOSECONDS.toMillis()` here as well,
instead of dividing? better to use a consistent mechanism to do this conversion
##########
pinot-common/src/main/java/org/apache/pinot/common/audit/AuditLogFilter.java:
##########
@@ -109,20 +122,30 @@ public void filter(ContainerRequestContext
requestContext, ContainerResponseCont
if (requestId == null) {
return;
}
- try {
- long durationMs = (System.nanoTime() - auditContext.getStartTimeNanos())
/ 1_000_000;
-
- final AuditEvent auditEvent = new AuditEvent().setRequestId(requestId)
- .setTimestamp(Instant.now().toString())
- .setResponseCode(responseContext.getStatus())
- .setDurationMs(durationMs)
- .setEndpoint(requestContext.getUriInfo().getPath())
- .setMethod(requestContext.getMethod());
+ long durationMs = (System.nanoTime() - auditContext.getStartTimeNanos()) /
1_000_000;
+ final AuditEvent auditEvent = new AuditEvent().setRequestId(requestId)
+ .setTimestamp(Instant.now().toString())
+ .setResponseCode(responseContext.getStatus())
+ .setDurationMs(durationMs)
+ .setEndpoint(requestContext.getUriInfo().getPath())
+ .setMethod(requestContext.getMethod());
+
+ AuditLogger.auditLog(auditEvent);
+ }
- AuditLogger.auditLog(auditEvent);
+ private void measure(Runnable operation, AuditMetrics.AuditTimer timer,
AuditMetrics.AuditMeter failureMeter) {
+ long startTime = System.nanoTime();
+ try {
+ operation.run();
} catch (Exception e) {
// Graceful degradation: Never let audit logging failures affect the
main response
- LOG.warn("Failed to process audit logging for response", e);
+ // logging the failure meter provides additional context if
request/response
+ LOG.warn("Failed to process audit logging. Incrementing {}",
failureMeter, e);
+ _auditMetrics.addMeteredGlobalValue(failureMeter, 1L);
+ // DO NOT escalate the exception.
Review Comment:
nit: is this comment really needed? you've already explained why you don't
throw an exception
##########
pinot-common/src/main/java/org/apache/pinot/common/audit/AuditMetrics.java:
##########
@@ -0,0 +1,165 @@
+/**
+ * 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.pinot.common.audit;
+
+import java.util.concurrent.TimeUnit;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.apache.pinot.common.metrics.AbstractMetrics;
+import org.apache.pinot.common.metrics.BrokerMeter;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.metrics.BrokerTimer;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.common.metrics.ControllerTimer;
+import org.apache.pinot.spi.services.ServiceRole;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.util.Objects.requireNonNull;
+
+
+/**
+ * Abstraction layer for audit metrics that handles the complexity of routing
metrics
+ * to the appropriate component-specific metrics (Controller vs Broker).
+ * This class provides a clean API for audit components to record metrics
without
+ * needing to know about the underlying component type.
+ */
+@Singleton
+public class AuditMetrics {
+ private static final Logger LOG =
LoggerFactory.getLogger(AuditMetrics.class);
+
+ private final TimerRecorder _timerRecorder;
+ private final MeterRecorder _meterRecorder;
+
+ /**
+ * Creates an AuditMetrics instance that will route metrics to the
appropriate component.
+ *
+ * @param delegate The component-specific metrics instance
(ControllerMetrics or BrokerMetrics)
+ * @param serviceRole The service role (CONTROLLER, BROKER, etc.)
+ */
+ @Inject
+ public AuditMetrics(AbstractMetrics<?, ?, ?, ?> delegate, ServiceRole
serviceRole) {
+ requireNonNull(delegate, "Component metrics cannot be null");
+ requireNonNull(serviceRole, "Service role cannot be null");
+
+ // One-time setup of delegation logic based on service role
+ switch (serviceRole) {
+ case CONTROLLER:
+ final ControllerMetrics controllerMetrics = (ControllerMetrics)
delegate;
+ _timerRecorder = (timer, durationMs) ->
controllerMetrics.addTimedValue(timer.getControllerTimer(), durationMs,
+ TimeUnit.MILLISECONDS);
+ _meterRecorder = (meter, count) ->
controllerMetrics.addMeteredGlobalValue(meter.getControllerMeter(), count);
+ break;
+ case BROKER:
+ final BrokerMetrics brokerMetrics = (BrokerMetrics) delegate;
+ _timerRecorder = (timer, durationMs) ->
brokerMetrics.addTimedValue(timer.getBrokerTimer(), durationMs,
+ TimeUnit.MILLISECONDS);
+ _meterRecorder = (meter, count) ->
brokerMetrics.addMeteredGlobalValue(meter.getBrokerMeter(), count);
+ break;
+ default:
+ LOG.warn("Audit not supported for service role: {}", serviceRole);
Review Comment:
could this just throw an exception instead to make it explicit? when support
is added for other serviceRoles, this can be updated then.
we anyways control how this is instantiated, right?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
