[improve][broker] Add authorization operation metrics

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/metrics/AuthorizationMetrics.java

@@ -0,0 +1,83 @@
+/*
+ * 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.pulsar.broker.authorization.metrics;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.LongCounter;
+import io.prometheus.client.Counter;
+
+public class AuthorizationMetrics {
+    public static final String AUTHORIZATION_OPERATIONS_METRIC_NAME = "pulsar_authorization_operations_total";
+    public static final String AUTHORIZATION_COUNTER_METRIC_NAME = "pulsar.authorization.operation.count";
+    public static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.pulsar.authorization";
+    public static final String RESULT_SUCCESS = "success";
+    public static final String RESULT_FAILURE = "failure";
+    public static final String RESULT_ERROR = "error";
+    public static final String RESOURCE_TYPE_SUPERUSER = "superuser";
+    public static final String RESOURCE_TYPE_TENANT_ADMIN = "tenant_admin";
+    public static final String RESOURCE_TYPE_TENANT = "tenant";
+    public static final String RESOURCE_TYPE_BROKER = "broker";
+    public static final String RESOURCE_TYPE_CLUSTER = "cluster";
+    public static final String RESOURCE_TYPE_CLUSTER_POLICY = "cluster_policy";
+    public static final String RESOURCE_TYPE_NAMESPACE = "namespace";
+    public static final String RESOURCE_TYPE_NAMESPACE_POLICY = "namespace_policy";
+    public static final String RESOURCE_TYPE_TOPIC = "topic";
+    public static final String RESOURCE_TYPE_TOPIC_POLICY = "topic_policy";
+    public static final AttributeKey<String> RESOURCE_TYPE_KEY =
+            AttributeKey.stringKey("pulsar.authorization.resource.type");
+    public static final AttributeKey<String> OPERATION_KEY = AttributeKey.stringKey("pulsar.authorization.operation");
+    public static final AttributeKey<String> RESULT_KEY = AttributeKey.stringKey("pulsar.authorization.result");
+
+    private static final Counter authorizationOperations = Counter.build()
+            .name(AUTHORIZATION_OPERATIONS_METRIC_NAME)
+            .help("Pulsar authorization operations")
+            .labelNames("resource_type", "operation", "result")
+            .register();
+
+    private final LongCounter authorizationCounter;
+
+    public AuthorizationMetrics(OpenTelemetry openTelemetry) {
+        var meter = openTelemetry.getMeter(INSTRUMENTATION_SCOPE_NAME);
+        authorizationCounter = meter.counterBuilder(AUTHORIZATION_COUNTER_METRIC_NAME)
+                .setDescription("The number of authorization operations")
+                .setUnit("{operation}")
+                .build();
+    }
+
+    public void recordSuccess(String resourceType, String operation) {
+        record(resourceType, operation, RESULT_SUCCESS);
+    }
+
+    public void recordFailure(String resourceType, String operation) {
+        record(resourceType, operation, RESULT_FAILURE);
+    }
+
+    public void recordError(String resourceType, String operation) {
+        record(resourceType, operation, RESULT_ERROR);
+    }
+
+    private void record(String resourceType, String operation, String result) {
+        authorizationOperations.labels(resourceType, operation, result).inc();
+        authorizationCounter.add(1, Attributes.of(RESOURCE_TYPE_KEY, resourceType,
+                OPERATION_KEY, operation,
+                RESULT_KEY, result));
+    }
+}

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/metrics/package-info.java

@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+/**
+ * Authorization metrics.
+ */
+package org.apache.pulsar.broker.authorization.metrics;

pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/AuthorizationServiceTest.java

@@ -20,9 +20,13 @@
 
 import static org.testng.AssertJUnit.assertFalse;
 import static org.testng.AssertJUnit.assertTrue;
+import static org.testng.AssertJUnit.fail;
+import io.prometheus.client.CollectorRegistry;
 import java.util.HashSet;
+import java.util.concurrent.ExecutionException;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authorization.metrics.AuthorizationMetrics;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.NamespaceOperation;
@@ -132,4 +136,68 @@ public void testTopicPolicyOperationAsync(String role, String originalRole, bool
                 PolicyName.ALL, PolicyOperation.READ, originalRole, role, null).get();
         checkResult(shouldPass, isAuthorized);
     }
+
+    @Test
+    public void testAuthorizationFailureMetricForTopicOperation() throws Exception {
+        double before = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                TopicOperation.PRODUCE.name().toLowerCase(), AuthorizationMetrics.RESULT_FAILURE);
+        boolean isAuthorized = authorizationService.allowTopicOperationAsync(TopicName.get("topic"),
+                TopicOperation.PRODUCE, null, "fail.client", null).get();
+        double after = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                TopicOperation.PRODUCE.name().toLowerCase(), AuthorizationMetrics.RESULT_FAILURE);
+
+        assertFalse(isAuthorized);
+        assertTrue(after - before == 1.0d);
+    }
+
+    @Test
+    public void testAuthorizationFailureMetricForInvalidOriginalPrincipal() throws Exception {
+        double before = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_NAMESPACE,
+                NamespaceOperation.PACKAGES.name().toLowerCase(), AuthorizationMetrics.RESULT_FAILURE);
+        boolean isAuthorized = authorizationService.allowNamespaceOperationAsync(NamespaceName.get("public/default"),
+                NamespaceOperation.PACKAGES, "pass.client", "pass.not-proxy", null).get();
+        double after = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_NAMESPACE,
+                NamespaceOperation.PACKAGES.name().toLowerCase(), AuthorizationMetrics.RESULT_FAILURE);
+
+        assertFalse(isAuthorized);
+        assertTrue(after - before == 1.0d);
+    }
+
+    @Test
+    public void testAuthorizationSuccessMetricForTopicOperation() throws Exception {
+        double before = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                TopicOperation.PRODUCE.name().toLowerCase(), AuthorizationMetrics.RESULT_SUCCESS);
+        boolean isAuthorized = authorizationService.allowTopicOperationAsync(TopicName.get("topic"),
+                TopicOperation.PRODUCE, null, "pass.client", null).get();
+        double after = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                TopicOperation.PRODUCE.name().toLowerCase(), AuthorizationMetrics.RESULT_SUCCESS);
+
+        assertTrue(isAuthorized);
+        assertTrue(after - before == 1.0d);
+    }
+
+    @Test
+    public void testAuthorizationErrorMetricForTopicOperation() throws Exception {
+        double before = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                TopicOperation.PRODUCE.name().toLowerCase(), AuthorizationMetrics.RESULT_ERROR);
+        try {
+            authorizationService.allowTopicOperationAsync(TopicName.get("topic"),
+                    TopicOperation.PRODUCE, null, "error.client", null).get();
+            fail("Expected authorization provider error");
+        } catch (ExecutionException e) {
+            // Expected.
+        }
+        double after = getAuthorizationOperations(AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                TopicOperation.PRODUCE.name().toLowerCase(), AuthorizationMetrics.RESULT_ERROR);
+
+        assertTrue(after - before == 1.0d);
+    }
+
+    private double getAuthorizationOperations(String resourceType, String operation, String result) {
+        Double sample = CollectorRegistry.defaultRegistry.getSampleValue(
+                AuthorizationMetrics.AUTHORIZATION_OPERATIONS_METRIC_NAME,
+                new String[] {"resource_type", "operation", "result"},
+                new String[] {resourceType, operation, result});
+        return sample == null ? 0.0d : sample;
+    }
 }

pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MockAuthorizationProvider.java

@@ -38,6 +38,9 @@
 public class MockAuthorizationProvider implements AuthorizationProvider {
 
     private CompletableFuture<Boolean> shouldPass(String role) {
+        if (role != null && role.startsWith("error")) {
+            return CompletableFuture.failedFuture(new RuntimeException("Authorization provider error"));
+        }
         return CompletableFuture.completedFuture(role != null && role.startsWith("pass"));
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -374,7 +374,7 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
 
         this.statsUpdater = new SingleThreadNonConcurrentFixedRateScheduler("pulsar-stats-updater");
         this.authorizationService = new AuthorizationService(
-                pulsar.getConfiguration(), pulsar().getPulsarResources());
+                pulsar.getConfiguration(), pulsar().getPulsarResources(), pulsar.getOpenTelemetry().getOpenTelemetry());
         this.entryFilterProvider = new EntryFilterProvider(pulsar.getConfiguration());
 
         pulsar.getLocalMetadataStore().registerListener(this::handleMetadataChanges);

pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/OpenTelemetryAuthorizationStatsTest.java

@@ -0,0 +1,127 @@
+/*
+ * 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.pulsar.broker.stats;
+
+import static org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil.assertMetricLongSumValue;
+import static org.testng.AssertJUnit.fail;
+import io.opentelemetry.api.common.Attributes;
+import java.util.HashSet;
+import java.util.concurrent.ExecutionException;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authorization.AuthorizationService;
+import org.apache.pulsar.broker.authorization.MockAuthorizationProvider;
+import org.apache.pulsar.broker.authorization.metrics.AuthorizationMetrics;
+import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.broker.testcontext.PulsarTestContext;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.NamespaceOperation;
+import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class OpenTelemetryAuthorizationStatsTest extends BrokerTestBase {
+
+    private AuthorizationService authorizationService;
+
+    @BeforeMethod(alwaysRun = true)
+    @Override
+    protected void setup() throws Exception {
+        super.baseSetup();
+
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setAuthorizationEnabled(true);
+        conf.setAuthorizationProvider(MockAuthorizationProvider.class.getName());
+        HashSet<String> proxyRoles = new HashSet<>();
+        proxyRoles.add("pass.proxy");
+        proxyRoles.add("fail.proxy");
+        conf.setProxyRoles(proxyRoles);
+        authorizationService = new AuthorizationService(conf, null, pulsar.getOpenTelemetry().getOpenTelemetry());
+    }
+
+    @AfterMethod(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Override
+    protected void customizeMainPulsarTestContextBuilder(PulsarTestContext.Builder builder) {
+        super.customizeMainPulsarTestContextBuilder(builder);
+        builder.enableOpenTelemetry(true);
+    }
+
+    @Test
+    public void testAuthorizationSuccess() throws Exception {
+        authorizationService.allowTopicOperationAsync(TopicName.get("topic"),
+                TopicOperation.PRODUCE, null, "pass.client", null).get();
+
+        assertMetricLongSumValue(pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics(),
+                AuthorizationMetrics.AUTHORIZATION_COUNTER_METRIC_NAME,
+                Attributes.of(AuthorizationMetrics.RESOURCE_TYPE_KEY, AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                        AuthorizationMetrics.OPERATION_KEY, "produce",
+                        AuthorizationMetrics.RESULT_KEY, AuthorizationMetrics.RESULT_SUCCESS),
+                1);
+    }
+
+    @Test
+    public void testAuthorizationFailure() throws Exception {
+        authorizationService.allowTopicOperationAsync(TopicName.get("topic"),
+                TopicOperation.PRODUCE, null, "fail.client", null).get();
+
+        assertMetricLongSumValue(pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics(),
+                AuthorizationMetrics.AUTHORIZATION_COUNTER_METRIC_NAME,
+                Attributes.of(AuthorizationMetrics.RESOURCE_TYPE_KEY, AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                        AuthorizationMetrics.OPERATION_KEY, "produce",
+                        AuthorizationMetrics.RESULT_KEY, AuthorizationMetrics.RESULT_FAILURE),
+                1);
+    }
+
+    @Test
+    public void testAuthorizationFailureForInvalidOriginalPrincipal() throws Exception {
+        authorizationService.allowNamespaceOperationAsync(NamespaceName.get("public/default"),
+                NamespaceOperation.PACKAGES, "pass.client", "pass.not-proxy", null).get();
+
+        assertMetricLongSumValue(pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics(),
+                AuthorizationMetrics.AUTHORIZATION_COUNTER_METRIC_NAME,
+                Attributes.of(AuthorizationMetrics.RESOURCE_TYPE_KEY, AuthorizationMetrics.RESOURCE_TYPE_NAMESPACE,
+                        AuthorizationMetrics.OPERATION_KEY, "packages",
+                        AuthorizationMetrics.RESULT_KEY, AuthorizationMetrics.RESULT_FAILURE),
+                1);
+    }
+
+    @Test
+    public void testAuthorizationError() throws Exception {
+        try {
+            authorizationService.allowTopicOperationAsync(TopicName.get("topic"),
+                    TopicOperation.PRODUCE, null, "error.client", null).get();
+            fail("Expected authorization provider error");
+        } catch (ExecutionException e) {
+            // Expected.
+        }
+
+        assertMetricLongSumValue(pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics(),
+                AuthorizationMetrics.AUTHORIZATION_COUNTER_METRIC_NAME,
+                Attributes.of(AuthorizationMetrics.RESOURCE_TYPE_KEY, AuthorizationMetrics.RESOURCE_TYPE_TOPIC,
+                        AuthorizationMetrics.OPERATION_KEY, "produce",
+                        AuthorizationMetrics.RESULT_KEY, AuthorizationMetrics.RESULT_ERROR),
+                1);
+    }
+}