Skip to content

Commit d60262f

Browse files
hchaverrihchaverri
authored
HADOOP-18167. Add metrics to track delegation token secret manager op… (#4092)
* HADOOP-18167. Add metrics to track delegation token secret manager operations
1 parent f1e5f8e commit d60262f

File tree

2 files changed

+252
-5
lines changed

2 files changed

+252
-5
lines changed

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

Lines changed: 116 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,17 @@
3636

3737
import org.apache.hadoop.classification.InterfaceAudience;
3838
import org.apache.hadoop.classification.InterfaceStability;
39+
import org.apache.hadoop.fs.statistics.DurationTracker;
40+
import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
41+
import org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding;
42+
import org.apache.hadoop.fs.statistics.impl.IOStatisticsStore;
3943
import org.apache.hadoop.io.Text;
44+
import org.apache.hadoop.metrics2.annotation.Metric;
45+
import org.apache.hadoop.metrics2.annotation.Metrics;
46+
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
47+
import org.apache.hadoop.metrics2.lib.MetricsRegistry;
48+
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
49+
import org.apache.hadoop.metrics2.lib.MutableRate;
4050
import org.apache.hadoop.metrics2.util.Metrics2Util.NameValuePair;
4151
import org.apache.hadoop.metrics2.util.Metrics2Util.TopN;
4252
import org.apache.hadoop.security.AccessControlException;
@@ -47,6 +57,7 @@
4757
import org.apache.hadoop.util.Time;
4858

4959
import org.apache.hadoop.util.Preconditions;
60+
import org.apache.hadoop.util.functional.InvocationRaisingIOE;
5061
import org.slf4j.Logger;
5162
import org.slf4j.LoggerFactory;
5263

@@ -96,6 +107,10 @@ private String formatTokenId(TokenIdent id) {
96107
* Access to currentKey is protected by this object lock
97108
*/
98109
private DelegationKey currentKey;
110+
/**
111+
* Metrics to track token management operations.
112+
*/
113+
private DelegationTokenSecretManagerMetrics metrics;
99114

100115
private long keyUpdateInterval;
101116
private long tokenMaxLifetime;
@@ -134,6 +149,7 @@ public AbstractDelegationTokenSecretManager(long delegationKeyUpdateInterval,
134149
this.tokenRenewInterval = delegationTokenRenewInterval;
135150
this.tokenRemoverScanInterval = delegationTokenRemoverScanInterval;
136151
this.storeTokenTrackingId = false;
152+
this.metrics = DelegationTokenSecretManagerMetrics.create();
137153
}
138154

139155
/** should be called before this object is used */
@@ -430,14 +446,14 @@ protected synchronized byte[] createPassword(TokenIdent identifier) {
430446
DelegationTokenInformation tokenInfo = new DelegationTokenInformation(now
431447
+ tokenRenewInterval, password, getTrackingIdIfEnabled(identifier));
432448
try {
433-
storeToken(identifier, tokenInfo);
449+
metrics.trackStoreToken(() -> storeToken(identifier, tokenInfo));
434450
} catch (IOException ioe) {
435451
LOG.error("Could not store token " + formatTokenId(identifier) + "!!",
436452
ioe);
437453
}
438454
return password;
439455
}
440-
456+
441457

442458

443459
/**
@@ -555,7 +571,7 @@ public synchronized long renewToken(Token<TokenIdent> token,
555571
throw new InvalidToken("Renewal request for unknown token "
556572
+ formatTokenId(id));
557573
}
558-
updateToken(id, info);
574+
metrics.trackUpdateToken(() -> updateToken(id, info));
559575
return renewTime;
560576
}
561577

@@ -591,8 +607,10 @@ public synchronized TokenIdent cancelToken(Token<TokenIdent> token,
591607
if (info == null) {
592608
throw new InvalidToken("Token not found " + formatTokenId(id));
593609
}
594-
removeTokenForOwnerStats(id);
595-
removeStoredToken(id);
610+
metrics.trackRemoveToken(() -> {
611+
removeTokenForOwnerStats(id);
612+
removeStoredToken(id);
613+
});
596614
return id;
597615
}
598616

@@ -825,4 +843,97 @@ protected void syncTokenOwnerStats() {
825843
addTokenForOwnerStats(id);
826844
}
827845
}
846+
847+
protected DelegationTokenSecretManagerMetrics getMetrics() {
848+
return metrics;
849+
}
850+
851+
/**
852+
* DelegationTokenSecretManagerMetrics tracks token management operations
853+
* and publishes them through the metrics interfaces.
854+
*/
855+
@Metrics(about="Delegation token secret manager metrics", context="token")
856+
static class DelegationTokenSecretManagerMetrics implements DurationTrackerFactory {
857+
private static final Logger LOG = LoggerFactory.getLogger(
858+
DelegationTokenSecretManagerMetrics.class);
859+
860+
final static String STORE_TOKEN_STAT = "storeToken";
861+
final static String UPDATE_TOKEN_STAT = "updateToken";
862+
final static String REMOVE_TOKEN_STAT = "removeToken";
863+
final static String TOKEN_FAILURE_STAT = "tokenFailure";
864+
865+
private final MetricsRegistry registry;
866+
private final IOStatisticsStore ioStatistics;
867+
868+
@Metric("Rate of storage of delegation tokens and latency (milliseconds)")
869+
private MutableRate storeToken;
870+
@Metric("Rate of update of delegation tokens and latency (milliseconds)")
871+
private MutableRate updateToken;
872+
@Metric("Rate of removal of delegation tokens and latency (milliseconds)")
873+
private MutableRate removeToken;
874+
@Metric("Counter of delegation tokens operation failures")
875+
private MutableCounterLong tokenFailure;
876+
877+
static DelegationTokenSecretManagerMetrics create() {
878+
return DefaultMetricsSystem.instance().register(new DelegationTokenSecretManagerMetrics());
879+
}
880+
881+
DelegationTokenSecretManagerMetrics() {
882+
ioStatistics = IOStatisticsBinding.iostatisticsStore()
883+
.withDurationTracking(STORE_TOKEN_STAT, UPDATE_TOKEN_STAT, REMOVE_TOKEN_STAT)
884+
.withCounters(TOKEN_FAILURE_STAT)
885+
.build();
886+
registry = new MetricsRegistry("DelegationTokenSecretManagerMetrics");
887+
LOG.debug("Initialized {}", registry);
888+
}
889+
890+
public void trackStoreToken(InvocationRaisingIOE invocation) throws IOException {
891+
trackInvocation(invocation, STORE_TOKEN_STAT, storeToken);
892+
}
893+
894+
public void trackUpdateToken(InvocationRaisingIOE invocation) throws IOException {
895+
trackInvocation(invocation, UPDATE_TOKEN_STAT, updateToken);
896+
}
897+
898+
public void trackRemoveToken(InvocationRaisingIOE invocation) throws IOException {
899+
trackInvocation(invocation, REMOVE_TOKEN_STAT, removeToken);
900+
}
901+
902+
public void trackInvocation(InvocationRaisingIOE invocation, String statistic,
903+
MutableRate metric) throws IOException {
904+
try {
905+
long start = Time.monotonicNow();
906+
IOStatisticsBinding.trackDurationOfInvocation(this, statistic, invocation);
907+
metric.add(Time.monotonicNow() - start);
908+
} catch (Exception ex) {
909+
tokenFailure.incr();
910+
throw ex;
911+
}
912+
}
913+
914+
@Override
915+
public DurationTracker trackDuration(String key, long count) {
916+
return ioStatistics.trackDuration(key, count);
917+
}
918+
919+
protected MutableRate getStoreToken() {
920+
return storeToken;
921+
}
922+
923+
protected MutableRate getUpdateToken() {
924+
return updateToken;
925+
}
926+
927+
protected MutableRate getRemoveToken() {
928+
return removeToken;
929+
}
930+
931+
protected MutableCounterLong getTokenFailure() {
932+
return tokenFailure;
933+
}
934+
935+
protected IOStatisticsStore getIoStatistics() {
936+
return ioStatistics;
937+
}
938+
}
828939
}

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

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,12 @@
3030
import java.util.List;
3131
import java.util.Map;
3232

33+
import java.util.concurrent.Callable;
34+
import org.apache.hadoop.fs.statistics.IOStatisticAssertions;
35+
import org.apache.hadoop.fs.statistics.MeanStatistic;
36+
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
37+
import org.apache.hadoop.metrics2.lib.MutableRate;
38+
import org.apache.hadoop.test.LambdaTestUtils;
3339
import org.junit.Assert;
3440

3541
import org.apache.hadoop.io.DataInputBuffer;
@@ -155,6 +161,55 @@ public DelegationKey getKey(TestDelegationTokenIdentifier id) {
155161
return allKeys.get(id.getMasterKeyId());
156162
}
157163
}
164+
165+
public static class TestFailureDelegationTokenSecretManager
166+
extends TestDelegationTokenSecretManager {
167+
private boolean throwError = false;
168+
private long errorSleepMillis;
169+
170+
public TestFailureDelegationTokenSecretManager(long errorSleepMillis) {
171+
super(24*60*60*1000, 10*1000, 1*1000, 60*60*1000);
172+
this.errorSleepMillis = errorSleepMillis;
173+
}
174+
175+
public void setThrowError(boolean throwError) {
176+
this.throwError = throwError;
177+
}
178+
179+
private void sleepAndThrow() throws IOException {
180+
try {
181+
Thread.sleep(errorSleepMillis);
182+
throw new IOException("Test exception");
183+
} catch (InterruptedException e) {
184+
}
185+
}
186+
187+
@Override
188+
protected void storeNewToken(TestDelegationTokenIdentifier ident, long renewDate)
189+
throws IOException {
190+
if (throwError) {
191+
sleepAndThrow();
192+
}
193+
super.storeNewToken(ident, renewDate);
194+
}
195+
196+
@Override
197+
protected void removeStoredToken(TestDelegationTokenIdentifier ident) throws IOException {
198+
if (throwError) {
199+
sleepAndThrow();
200+
}
201+
super.removeStoredToken(ident);
202+
}
203+
204+
@Override
205+
protected void updateStoredToken(TestDelegationTokenIdentifier ident, long renewDate)
206+
throws IOException {
207+
if (throwError) {
208+
sleepAndThrow();
209+
}
210+
super.updateStoredToken(ident, renewDate);
211+
}
212+
}
158213

159214
public static class TokenSelector extends
160215
AbstractDelegationTokenSelector<TestDelegationTokenIdentifier>{
@@ -579,4 +634,85 @@ public void testEmptyToken() throws IOException {
579634
assertEquals(token1, token2);
580635
assertEquals(token1.encodeToUrlString(), token2.encodeToUrlString());
581636
}
637+
638+
@Test
639+
public void testDelegationTokenSecretManagerMetrics() throws Exception {
640+
TestDelegationTokenSecretManager dtSecretManager =
641+
new TestDelegationTokenSecretManager(24*60*60*1000,
642+
10*1000, 1*1000, 60*60*1000);
643+
try {
644+
dtSecretManager.startThreads();
645+
646+
final Token<TestDelegationTokenIdentifier> token = callAndValidateMetrics(
647+
dtSecretManager, dtSecretManager.getMetrics().getStoreToken(), "storeToken",
648+
() -> generateDelegationToken(dtSecretManager, "SomeUser", "JobTracker"), 1);
649+
650+
callAndValidateMetrics(dtSecretManager, dtSecretManager.getMetrics().getUpdateToken(),
651+
"updateToken", () -> dtSecretManager.renewToken(token, "JobTracker"), 1);
652+
653+
callAndValidateMetrics(dtSecretManager, dtSecretManager.getMetrics().getRemoveToken(),
654+
"removeToken", () -> dtSecretManager.cancelToken(token, "JobTracker"), 1);
655+
} finally {
656+
dtSecretManager.stopThreads();
657+
}
658+
}
659+
660+
@Test
661+
public void testDelegationTokenSecretManagerMetricsFailures() throws Exception {
662+
int errorSleepMillis = 200;
663+
TestFailureDelegationTokenSecretManager dtSecretManager =
664+
new TestFailureDelegationTokenSecretManager(errorSleepMillis);
665+
666+
try {
667+
dtSecretManager.startThreads();
668+
669+
final Token<TestDelegationTokenIdentifier> token =
670+
generateDelegationToken(dtSecretManager, "SomeUser", "JobTracker");
671+
672+
dtSecretManager.setThrowError(true);
673+
674+
callAndValidateFailureMetrics(dtSecretManager, "storeToken", 1, 1, false,
675+
errorSleepMillis,
676+
() -> generateDelegationToken(dtSecretManager, "SomeUser", "JobTracker"));
677+
678+
callAndValidateFailureMetrics(dtSecretManager, "updateToken", 1, 2, true,
679+
errorSleepMillis, () -> dtSecretManager.renewToken(token, "JobTracker"));
680+
681+
callAndValidateFailureMetrics(dtSecretManager, "removeToken", 1, 3, true,
682+
errorSleepMillis, () -> dtSecretManager.cancelToken(token, "JobTracker"));
683+
} finally {
684+
dtSecretManager.stopThreads();
685+
}
686+
}
687+
688+
private <T> T callAndValidateMetrics(TestDelegationTokenSecretManager dtSecretManager,
689+
MutableRate metric, String statName, Callable<T> callable, int expectedCount)
690+
throws Exception {
691+
MeanStatistic stat = IOStatisticAssertions.lookupMeanStatistic(
692+
dtSecretManager.getMetrics().getIoStatistics(), statName + ".mean");
693+
assertEquals(expectedCount - 1, metric.lastStat().numSamples());
694+
assertEquals(expectedCount - 1, stat.getSamples());
695+
T returnedObject = callable.call();
696+
assertEquals(expectedCount, metric.lastStat().numSamples());
697+
assertEquals(expectedCount, stat.getSamples());
698+
return returnedObject;
699+
}
700+
701+
private <T> void callAndValidateFailureMetrics(TestDelegationTokenSecretManager dtSecretManager,
702+
String statName, int expectedStatCount, int expectedMetricCount, boolean expectError,
703+
int errorSleepMillis, Callable<T> callable) throws Exception {
704+
MutableCounterLong counter = dtSecretManager.getMetrics().getTokenFailure();
705+
MeanStatistic failureStat = IOStatisticAssertions.lookupMeanStatistic(
706+
dtSecretManager.getMetrics().getIoStatistics(), statName + ".failures.mean");
707+
assertEquals(expectedMetricCount - 1, counter.value());
708+
assertEquals(expectedStatCount - 1, failureStat.getSamples());
709+
if (expectError) {
710+
LambdaTestUtils.intercept(IOException.class, callable);
711+
} else {
712+
callable.call();
713+
}
714+
assertEquals(expectedMetricCount, counter.value());
715+
assertEquals(expectedStatCount, failureStat.getSamples());
716+
assertTrue(failureStat.getSum() >= errorSleepMillis);
717+
}
582718
}

0 commit comments

Comments
 (0)