HADOOP-18167. Add metrics to track delegation token secret manager op…

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/AbstractDelegationTokenSecretManager.java

@@ -36,7 +36,16 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.statistics.impl.IOStatisticsStore;
 import org.apache.hadoop.io.Text;
+import org.apache.hadoop.metrics2.annotation.Metric;
+import org.apache.hadoop.metrics2.annotation.Metrics;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.metrics2.lib.MetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableCounterLong;
+import org.apache.hadoop.metrics2.lib.MutableRate;
 import org.apache.hadoop.metrics2.util.Metrics2Util.NameValuePair;
 import org.apache.hadoop.metrics2.util.Metrics2Util.TopN;
 import org.apache.hadoop.security.AccessControlException;
@@ -50,6 +59,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.*;
+
+
 @InterfaceAudience.Public
 @InterfaceStability.Evolving
 public abstract 
@@ -96,6 +108,10 @@ private String formatTokenId(TokenIdent id) {
    * Access to currentKey is protected by this object lock
    */
   private DelegationKey currentKey;
+  /**
+   * Metrics to track token management operations
+   */
+  protected DelegationTokenSecretManagerMetrics metrics;
 
   private long keyUpdateInterval;
   private long tokenMaxLifetime;
@@ -134,6 +150,7 @@ public AbstractDelegationTokenSecretManager(long delegationKeyUpdateInterval,
     this.tokenRenewInterval = delegationTokenRenewInterval;
     this.tokenRemoverScanInterval = delegationTokenRemoverScanInterval;
     this.storeTokenTrackingId = false;
+    this.metrics = DelegationTokenSecretManagerMetrics.create();
   }
 
   /** should be called before this object is used */
@@ -429,11 +446,18 @@ protected synchronized byte[] createPassword(TokenIdent identifier) {
     byte[] password = createPassword(identifier.getBytes(), currentKey.getKey());
     DelegationTokenInformation tokenInfo = new DelegationTokenInformation(now
         + tokenRenewInterval, password, getTrackingIdIfEnabled(identifier));
+    long start = Time.monotonicNow();
+    boolean success = false;
     try {
       storeToken(identifier, tokenInfo);
+      success = true;
     } catch (IOException ioe) {
       LOG.error("Could not store token " + formatTokenId(identifier) + "!!",
           ioe);
+    } finally {
+      if (metrics != null) {
+        metrics.addStoreToken(success, Time.monotonicNow() - start);
+      }
     }
     return password;
   }
@@ -555,7 +579,16 @@ public synchronized long renewToken(Token<TokenIdent> token,
       throw new InvalidToken("Renewal request for unknown token "
           + formatTokenId(id));
     }
-    updateToken(id, info);
+    long start = Time.monotonicNow();
+    boolean success = false;
+    try {
+      updateToken(id, info);
+      success = true;
+    } finally {
+      if (metrics != null) {
+        metrics.addUpdateToken(success, Time.monotonicNow() - start);
+      }
+    }
     return renewTime;
   }
 
@@ -591,8 +624,17 @@ public synchronized TokenIdent cancelToken(Token<TokenIdent> token,
     if (info == null) {
       throw new InvalidToken("Token not found " + formatTokenId(id));
     }
-    removeTokenForOwnerStats(id);
-    removeStoredToken(id);
+    long start = Time.monotonicNow();
+    boolean success = false;
+    try {
+      removeTokenForOwnerStats(id);
+      removeStoredToken(id);
+      success = true;
+    } finally {
+      if (metrics != null) {
+        metrics.addRemoveToken(success, Time.monotonicNow() - start);
+      }
+    }
     return id;
   }
 
@@ -825,4 +867,78 @@ protected void syncTokenOwnerStats() {
       addTokenForOwnerStats(id);
     }
   }
+
+  /**
+   * DelegationTokenSecretManagerMetrics tracks token management operations
+   * and publishes them through the metrics interfaces.
+   */
+  @Metrics(about="Delegation token secret manager metrics", context="token")
+  static class DelegationTokenSecretManagerMetrics implements IOStatisticsSource {
+    private static final Logger LOG = LoggerFactory.getLogger(DelegationTokenSecretManagerMetrics.class);
+
+    final static String STORE_TOKEN_STAT = "storeToken";
+    final static String UPDATE_TOKEN_STAT = "updateToken";
+    final static String REMOVE_TOKEN_STAT = "removeToken";
+    final static String TOKEN_FAILURE_STAT = "tokenFailure";
+
+    final MetricsRegistry registry;
+    final IOStatisticsStore ioStatistics;
+
+    @Metric("Rate of storage of delegation tokens and latency (milliseconds)")
+    MutableRate storeToken;
+    @Metric("Rate of update of delegation tokens and latency (milliseconds)")
+    MutableRate updateToken;
+    @Metric("Rate of removal of delegation tokens and latency (milliseconds)")
+    MutableRate removeToken;
+    @Metric("Counter of delegation tokens operation failures")
+    MutableCounterLong tokenFailure;
+
+    static DelegationTokenSecretManagerMetrics create() {
+      return DefaultMetricsSystem.instance().register(new DelegationTokenSecretManagerMetrics());
+    }
+
+    public DelegationTokenSecretManagerMetrics() {
+      ioStatistics = iostatisticsStore()
+          .withDurationTracking(STORE_TOKEN_STAT, UPDATE_TOKEN_STAT, REMOVE_TOKEN_STAT)
+          .withCounters(TOKEN_FAILURE_STAT)
+          .build();
+      registry = new MetricsRegistry("DelegationTokenSecretManagerMetrics");
+      LOG.debug("Initialized {}", registry);
+    }
+
+    public void addStoreToken(boolean success, long value) {
+      if (success) {
+        storeToken.add(value);
+        ioStatistics.addTimedOperation(STORE_TOKEN_STAT, value);
+      } else {
+        tokenFailure.incr();
+        ioStatistics.incrementCounter(TOKEN_FAILURE_STAT);
+      }
+    }
+
+    public void addUpdateToken(boolean success, long value) {
+      if (success) {
+        updateToken.add(value);
+        ioStatistics.addTimedOperation(UPDATE_TOKEN_STAT, value);
+      } else {
+        tokenFailure.incr();
+        ioStatistics.incrementCounter(TOKEN_FAILURE_STAT);
+      }
+    }
+
+    public void addRemoveToken(boolean success, long value) {
+      if (success) {
+        removeToken.add(value);
+        ioStatistics.addTimedOperation(REMOVE_TOKEN_STAT, value);
+      } else {
+        tokenFailure.incr();
+        ioStatistics.incrementCounter(TOKEN_FAILURE_STAT);
+      }
+    }
+
+    @Override
+    public IOStatistics getIOStatistics() {
+      return ioStatistics;
+    }
+  }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestDelegationToken.java

@@ -155,6 +155,42 @@ public DelegationKey getKey(TestDelegationTokenIdentifier id) {
       return allKeys.get(id.getMasterKeyId());
     }
   }
+
+  public static class TestFailureDelegationTokenSecretManager extends TestDelegationTokenSecretManager {
+    private boolean throwError = false;
+
+    public TestFailureDelegationTokenSecretManager() {
+      super(24*60*60*1000, 10*1000, 1*1000, 3600000);
+    }
+
+    public void setThrowError(boolean throwError) {
+      this.throwError = throwError;
+    }
+
+    @Override
+    protected void storeNewToken(TestDelegationTokenIdentifier ident, long renewDate) throws IOException {
+      if (throwError) {
+        throw new IOException("Test exception");
+      }
+      super.storeNewToken(ident, renewDate);
+    }
+
+    @Override
+    protected void removeStoredToken(TestDelegationTokenIdentifier ident) throws IOException {
+      if (throwError) {
+        throw new IOException("Test exception");
+      }
+      super.removeStoredToken(ident);
+    }
+
+    @Override
+    protected void updateStoredToken(TestDelegationTokenIdentifier ident, long renewDate) throws IOException {
+      if (throwError) {
+        throw new IOException("Test exception");
+      }
+      super.updateStoredToken(ident, renewDate);
+    }
+  }
 
   public static class TokenSelector extends 
   AbstractDelegationTokenSelector<TestDelegationTokenIdentifier>{
@@ -579,4 +615,65 @@ public void testEmptyToken() throws IOException {
     assertEquals(token1, token2);
     assertEquals(token1.encodeToUrlString(), token2.encodeToUrlString());
   }
+
+  @Test
+  public void testDelegationTokenSecretManagerMetrics() throws Exception {
+    TestDelegationTokenSecretManager dtSecretManager =
+        new TestDelegationTokenSecretManager(24*60*60*1000,
+            10*1000,1*1000,3600000);
+    try {
+      dtSecretManager.startThreads();
+
+      Assert.assertEquals(0, dtSecretManager.metrics.storeToken.lastStat().numSamples());
+      final Token<TestDelegationTokenIdentifier> token =
+          generateDelegationToken(dtSecretManager, "SomeUser", "JobTracker");
+      Assert.assertEquals(1, dtSecretManager.metrics.storeToken.lastStat().numSamples());
+
+      Assert.assertEquals(0, dtSecretManager.metrics.updateToken.lastStat().numSamples());
+      dtSecretManager.renewToken(token, "JobTracker");
+      Assert.assertEquals(1, dtSecretManager.metrics.updateToken.lastStat().numSamples());
+
+      Assert.assertEquals(0, dtSecretManager.metrics.removeToken.lastStat().numSamples());
+      dtSecretManager.cancelToken(token, "JobTracker");
+      Assert.assertEquals(1, dtSecretManager.metrics.removeToken.lastStat().numSamples());
+    } finally {
+      dtSecretManager.stopThreads();
+    }
+  }
+
+  @Test
+  public void testDelegationTokenSecretManagerMetricsFailures() throws Exception {
+    TestFailureDelegationTokenSecretManager dtSecretManager = new TestFailureDelegationTokenSecretManager();
+
+    try {
+      dtSecretManager.startThreads();
+
+      final Token<TestDelegationTokenIdentifier> token =
+          generateDelegationToken(dtSecretManager, "SomeUser", "JobTracker");
+
+      dtSecretManager.setThrowError(true);
+
+      Assert.assertEquals(0, dtSecretManager.metrics.tokenFailure.value());
+      generateDelegationToken(dtSecretManager, "SomeUser", "JobTracker");
+      Assert.assertEquals(1, dtSecretManager.metrics.tokenFailure.value());
+
+      try {
+        dtSecretManager.renewToken(token, "JobTracker");
+        Assert.fail("Expected exception");
+      } catch (Exception ex) {
+        // Expected exception
+      }
+      Assert.assertEquals(2, dtSecretManager.metrics.tokenFailure.value());
+
+      try {
+        dtSecretManager.cancelToken(token, "JobTracker");
+        Assert.fail("Expected exception");
+      } catch (Exception ex) {
+        // Expected exception
+      }
+      Assert.assertEquals(3, dtSecretManager.metrics.tokenFailure.value());
+    } finally {
+      dtSecretManager.stopThreads();
+    }
+  }
 }