[fix][broker] Fix the reason label of authentication metrics (#20030)

(cherry picked from commit 2b515ff)
apache · May 8, 2023 · 80dcea7 · 80dcea7
1 parent 7da7873
commit 80dcea7
Show file tree

Hide file tree

Showing 8 changed files with 101 additions and 27 deletions.
diff --git a/...z/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderAthenz.java b/...z/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderAthenz.java
@@ -43,6 +43,15 @@ public class AuthenticationProviderAthenz implements AuthenticationProvider {
     private List<String> domainNameList = null;
     private int allowedOffset = 30;
 
+    public enum ErrorCode {
+        UNKNOWN,
+        NO_CLIENT,
+        NO_TOKEN,
+        NO_PUBLIC_KEY,
+        DOMAIN_MISMATCH,
+        INVALID_TOKEN,
+    }
+
     @Override
     public void initialize(ServiceConfiguration config) throws IOException {
         String domainNames;
@@ -81,11 +90,13 @@ public String getAuthMethodName() {
     public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
         SocketAddress clientAddress;
         String roleToken;
+        ErrorCode errorCode = ErrorCode.UNKNOWN;
         try {
 
             if (authData.hasDataFromPeer()) {
                 clientAddress = authData.getPeerAddress();
             } else {
+                errorCode = ErrorCode.NO_CLIENT;
                 throw new AuthenticationException("Authentication data source does not have a client address");
             }
 
@@ -94,13 +105,16 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
             } else if (authData.hasDataFromHttp()) {
                 roleToken = authData.getHttpHeader(AuthZpeClient.ZPE_TOKEN_HDR);
             } else {
+                errorCode = ErrorCode.NO_TOKEN;
                 throw new AuthenticationException("Authentication data source does not have a role token");
             }
 
             if (roleToken == null) {
+                errorCode = ErrorCode.NO_TOKEN;
                 throw new AuthenticationException("Athenz token is null, can't authenticate");
             }
             if (roleToken.isEmpty()) {
+                errorCode = ErrorCode.NO_TOKEN;
                 throw new AuthenticationException("Athenz RoleToken is empty, Server is Using Athenz Authentication");
             }
             if (log.isDebugEnabled()) {
@@ -110,6 +124,7 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
             RoleToken token = new RoleToken(roleToken);
 
             if (!domainNameList.contains(token.getDomain())) {
+                errorCode = ErrorCode.DOMAIN_MISMATCH;
                 throw new AuthenticationException(
                         String.format("Athenz RoleToken Domain mismatch, Expected: %s, Found: %s",
                                 domainNameList.toString(), token.getDomain()));
@@ -120,6 +135,7 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
                 PublicKey ztsPublicKey = AuthZpeClient.getZtsPublicKey(token.getKeyId());
 
                 if (ztsPublicKey == null) {
+                    errorCode = ErrorCode.NO_PUBLIC_KEY;
                     throw new AuthenticationException("Unable to retrieve ZTS Public Key");
                 }
 
@@ -128,13 +144,13 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
                     AuthenticationMetrics.authenticateSuccess(getClass().getSimpleName(), getAuthMethodName());
                     return token.getPrincipal();
                 } else {
+                    errorCode = ErrorCode.INVALID_TOKEN;
                     throw new AuthenticationException(
                             String.format("Athenz Role Token Not Authenticated from Client: %s", clientAddress));
                 }
             }
         } catch (AuthenticationException exception) {
-            AuthenticationMetrics.authenticateFailure(getClass().getSimpleName(), getAuthMethodName(),
-                    exception.getMessage());
+            incrementFailureMetric(errorCode);
             throw exception;
         }
     }

diff --git a/...-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java b/...-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProvider.java
@@ -26,6 +26,7 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.metrics.AuthenticationMetrics;
 import org.apache.pulsar.common.api.AuthData;
 
 /**
@@ -80,4 +81,9 @@ default AuthenticationState newAuthState(AuthData authData,
     default boolean authenticateHttpRequest(HttpServletRequest request, HttpServletResponse response) throws Exception {
         throw new AuthenticationException("Not supported");
     }
+
+    default void incrementFailureMetric(Enum<?> errorCode) {
+        AuthenticationMetrics.authenticateFailure(getClass().getSimpleName(), getAuthMethodName(), errorCode);
+    }
+
 }
diff --git a/...on/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java b/...on/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderBasic.java
@@ -47,6 +47,14 @@ public class AuthenticationProviderBasic implements AuthenticationProvider {
     private static final String CONF_PULSAR_PROPERTY_KEY = "basicAuthConf";
     private Map<String, String> users;
 
+    private enum ErrorCode {
+        UNKNOWN,
+        EMPTY_AUTH_DATA,
+        INVALID_HEADER,
+        INVALID_AUTH_DATA,
+        INVALID_TOKEN,
+    }
+
     @Override
     public void close() throws IOException {
         // noop
@@ -105,9 +113,10 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
         String userId = authParams.getUserId();
         String password = authParams.getPassword();
         String msg = "Unknown user or invalid password";
-
+        ErrorCode errorCode = ErrorCode.UNKNOWN;
         try {
             if (users.get(userId) == null) {
+                errorCode = ErrorCode.INVALID_AUTH_DATA;
                 throw new AuthenticationException(msg);
             }
 
@@ -118,15 +127,16 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
                 List<String> splitEncryptedPassword = Arrays.asList(encryptedPassword.split("\\$"));
                 if (splitEncryptedPassword.size() != 4 || !encryptedPassword
                         .equals(Md5Crypt.apr1Crypt(password.getBytes(), splitEncryptedPassword.get(2)))) {
+                    errorCode = ErrorCode.INVALID_TOKEN;
                     throw new AuthenticationException(msg);
                 }
                 // For crypt algorithm
             } else if (!encryptedPassword.equals(Crypt.crypt(password.getBytes(), encryptedPassword.substring(0, 2)))) {
+                errorCode = ErrorCode.INVALID_TOKEN;
                 throw new AuthenticationException(msg);
             }
         } catch (AuthenticationException exception) {
-            AuthenticationMetrics.authenticateFailure(getClass().getSimpleName(), getAuthMethodName(),
-                    exception.getMessage());
+            incrementFailureMetric(errorCode);
             throw exception;
         }
         AuthenticationMetrics.authenticateSuccess(getClass().getSimpleName(), getAuthMethodName());
@@ -145,24 +155,29 @@ public AuthParams(AuthenticationDataSource authData) throws AuthenticationExcept
                 String rawAuthToken = authData.getHttpHeader(HTTP_HEADER_NAME);
                 // parsing and validation
                 if (StringUtils.isBlank(rawAuthToken) || !rawAuthToken.toUpperCase().startsWith("BASIC ")) {
+                    incrementFailureMetric(ErrorCode.INVALID_HEADER);
                     throw new AuthenticationException("Authentication token has to be started with \"Basic \"");
                 }
                 String[] splitRawAuthToken = rawAuthToken.split(" ");
                 if (splitRawAuthToken.length != 2) {
+                    incrementFailureMetric(ErrorCode.INVALID_HEADER);
                     throw new AuthenticationException("Base64 encoded token is not found");
                 }
 
                 try {
                     authParams = new String(Base64.getDecoder().decode(splitRawAuthToken[1]));
                 } catch (Exception e) {
+                    incrementFailureMetric(ErrorCode.INVALID_HEADER);
                     throw new AuthenticationException("Base64 decoding is failure: " + e.getMessage());
                 }
             } else {
+                incrementFailureMetric(ErrorCode.EMPTY_AUTH_DATA);
                 throw new AuthenticationException("Authentication data source does not have data");
             }
 
             String[] parsedAuthParams = authParams.split(":");
             if (parsedAuthParams.length != 2) {
+                incrementFailureMetric(ErrorCode.INVALID_AUTH_DATA);
                 throw new AuthenticationException("Base64 decoded params are invalid");
             }
 

diff --git a/...mon/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderList.java b/...mon/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderList.java
@@ -43,9 +43,15 @@ private interface AuthProcessor<T, W> {
 
     }
 
+    private enum ErrorCode {
+        UNKNOWN,
+        AUTH_REQUIRED,
+    }
+
     static <T, W> T applyAuthProcessor(List<W> processors, AuthProcessor<T, W> authFunc)
         throws AuthenticationException {
         AuthenticationException authenticationException = null;
+        String errorCode = ErrorCode.UNKNOWN.name();
         for (W ap : processors) {
             try {
                 return authFunc.apply(ap);
@@ -55,19 +61,19 @@ static <T, W> T applyAuthProcessor(List<W> processors, AuthProcessor<T, W> authF
                 }
                 // Store the exception so we can throw it later instead of a generic one
                 authenticationException = ae;
+                errorCode = ap.getClass().getSimpleName() + "-INVALID-AUTH";
             }
         }
 
         if (null == authenticationException) {
             AuthenticationMetrics.authenticateFailure(
                     AuthenticationProviderList.class.getSimpleName(),
-                    "authentication-provider-list", "Authentication required");
+                    "authentication-provider-list", ErrorCode.AUTH_REQUIRED);
             throw new AuthenticationException("Authentication required");
         } else {
-            AuthenticationMetrics.authenticateFailure(AuthenticationProviderList.class.getSimpleName(),
-                    "authentication-provider-list",
-                    authenticationException.getMessage() != null
-                            ? authenticationException.getMessage() : "Authentication required");
+            AuthenticationMetrics.authenticateFailure(
+                    AuthenticationProviderList.class.getSimpleName(),
+                    "authentication-provider-list", errorCode);
             throw authenticationException;
         }
 

diff --git a/...mmon/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTls.java b/...mmon/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderTls.java
@@ -27,6 +27,12 @@
 
 public class AuthenticationProviderTls implements AuthenticationProvider {
 
+    private enum ErrorCode {
+        UNKNOWN,
+        INVALID_CERTS,
+        INVALID_CN, // invalid common name
+    }
+
     @Override
     public void close() throws IOException {
         // noop
@@ -45,6 +51,7 @@ public String getAuthMethodName() {
     @Override
     public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
         String commonName = null;
+        ErrorCode errorCode = ErrorCode.UNKNOWN;
         try {
             if (authData.hasDataFromTls()) {
                 /**
@@ -72,6 +79,7 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
                 // CN=Steve Kille,O=Isode Limited,C=GB
                 Certificate[] certs = authData.getTlsCertificates();
                 if (null == certs) {
+                    errorCode = ErrorCode.INVALID_CERTS;
                     throw new AuthenticationException("Failed to get TLS certificates from client");
                 }
                 String distinguishedName = ((X509Certificate) certs[0]).getSubjectX500Principal().getName();
@@ -85,12 +93,12 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
             }
 
             if (commonName == null) {
+                errorCode = ErrorCode.INVALID_CN;
                 throw new AuthenticationException("Client unable to authenticate with TLS certificate");
             }
             AuthenticationMetrics.authenticateSuccess(getClass().getSimpleName(), getAuthMethodName());
         } catch (AuthenticationException exception) {
-            AuthenticationMetrics.authenticateFailure(getClass().getSimpleName(), getAuthMethodName(),
-                    exception.getMessage());
+            incrementFailureMetric(errorCode);
             throw exception;
         }
         return commonName;

diff --git a/...on/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java b/...on/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderToken.java
@@ -99,6 +99,12 @@ public class AuthenticationProviderToken implements AuthenticationProvider {
     private String confTokenAudienceClaimSettingName;
     private String confTokenAudienceSettingName;
 
+    public enum ErrorCode {
+        INVALID_AUTH_DATA,
+        INVALID_TOKEN,
+        INVALID_AUDIENCES,
+    }
+
     @Override
     public void close() throws IOException {
         // noop
@@ -145,19 +151,18 @@ public String getAuthMethodName() {
 
     @Override
     public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
+        String token;
         try {
             // Get Token
-            String token;
             token = getToken(authData);
-            // Parse Token by validating
-            String role = getPrincipal(authenticateToken(token));
-            AuthenticationMetrics.authenticateSuccess(getClass().getSimpleName(), getAuthMethodName());
-            return role;
         } catch (AuthenticationException exception) {
-            AuthenticationMetrics.authenticateFailure(getClass().getSimpleName(), getAuthMethodName(),
-                    exception.getMessage());
+            incrementFailureMetric(ErrorCode.INVALID_AUTH_DATA);
             throw exception;
         }
+        // Parse Token by validating
+        String role = getPrincipal(authenticateToken(token));
+        AuthenticationMetrics.authenticateSuccess(getClass().getSimpleName(), getAuthMethodName());
+        return role;
     }
 
     @Override
@@ -209,16 +214,19 @@ private Jwt<?, Claims> authenticateToken(final String token) throws Authenticati
                     List<String> audiences = (List<String>) object;
                     // audience not contains this broker, throw exception.
                     if (audiences.stream().noneMatch(audienceInToken -> audienceInToken.equals(audience))) {
-                        throw new AuthenticationException("Audiences in token: [" + String.join(", ", audiences)
-                                                          + "] not contains this broker: " + audience);
+                        incrementFailureMetric(ErrorCode.INVALID_AUDIENCES);
+                        throw new AuthenticationException("Audiences in token: ["
+                                + String.join(", ", audiences) + "] not contains this broker: " + audience);
                     }
                 } else if (object instanceof String) {
                     if (!object.equals(audience)) {
-                        throw new AuthenticationException("Audiences in token: [" + object
-                                                          + "] not contains this broker: " + audience);
+                        incrementFailureMetric(ErrorCode.INVALID_AUDIENCES);
+                        throw new AuthenticationException(
+                                "Audiences in token: [" + object + "] not contains this broker: " + audience);
                     }
                 } else {
                     // should not reach here.
+                    incrementFailureMetric(ErrorCode.INVALID_AUDIENCES);
                     throw new AuthenticationException("Audiences in token is not in expected format: " + object);
                 }
             }
@@ -232,6 +240,7 @@ private Jwt<?, Claims> authenticateToken(final String token) throws Authenticati
             if (e instanceof ExpiredJwtException) {
                 expiredTokenMetrics.inc();
             }
+            incrementFailureMetric(ErrorCode.INVALID_TOKEN);
             throw new AuthenticationException("Failed to authentication token: " + e.getMessage());
         }
     }

diff --git a/.../src/main/java/org/apache/pulsar/broker/authentication/metrics/AuthenticationMetrics.java b/.../src/main/java/org/apache/pulsar/broker/authentication/metrics/AuthenticationMetrics.java
@@ -43,11 +43,27 @@ public static void authenticateSuccess(String providerName, String authMethod) {
 
     /**
      * Log authenticate failure event to the authentication metrics.
+     *
+     * This method is deprecated due to the label "reason" is a potential infinite value.
+     * @deprecated See {@link #authenticateFailure(String, String, Enum)} ()}
+     *
      * @param providerName The short class name of the provider
      * @param authMethod Authentication method name.
      * @param reason Failure reason.
      */
+    @Deprecated
     public static void authenticateFailure(String providerName, String authMethod, String reason) {
         authFailuresMetrics.labels(providerName, authMethod, reason).inc();
     }
+
+    /**
+     * Log authenticate failure event to the authentication metrics.
+     * @param providerName The short class name of the provider
+     * @param authMethod Authentication method name.
+     * @param errorCode Error code.
+     */
+    public static void authenticateFailure(String providerName, String authMethod, Enum<?> errorCode) {
+        authFailuresMetrics.labels(providerName, authMethod, errorCode.name()).inc();
+    }
+
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -1141,15 +1141,12 @@ public void testAuthMetrics() throws IOException, AuthenticationException {
         conf.setProperties(properties);
         provider.initialize(conf);
 
-        String authExceptionMessage = "";
-
         try {
             provider.authenticate(new AuthenticationDataSource() {
             });
             fail("Should have failed");
         } catch (AuthenticationException e) {
             // expected, no credential passed
-            authExceptionMessage = e.getMessage();
         }
 
         String token = AuthTokenUtils.createToken(secretKey, "subject", Optional.empty());
@@ -1186,7 +1183,8 @@ public String getCommandData() {
         boolean haveFailed = false;
         for (Metric metric : cm) {
             if (Objects.equals(metric.tags.get("auth_method"), "token")
-                    && Objects.equals(metric.tags.get("reason"), authExceptionMessage)
+                    && Objects.equals(metric.tags.get("reason"),
+                    AuthenticationProviderToken.ErrorCode.INVALID_AUTH_DATA.name())
                     && Objects.equals(metric.tags.get("provider_name"), provider.getClass().getSimpleName())) {
                 haveFailed = true;
             }