HADOOP-19737: ABFS: Add metrics to identify improvements with read and write aggressiveness #8056
+1,793
−149
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
manika137
reviewed
Nov 19, 2025
| configuration1.set(FS_AZURE_ENABLE_PREFETCH_REQUEST_PRIORITY, "true"); | ||
|
|
||
| Configuration configuration2 = new Configuration(getRawConfiguration()); | ||
| //use the default value for the config: false |
Contributor
There was a problem hiding this comment.
we can remove the comment as well. The default traffic priority value is true now
manika137
reviewed
Nov 19, 2025
| if (lastTime == ZERO) { | ||
| lastCpuTime = cpuTime; | ||
| lastTime = now; | ||
| return 0.0; // first call has no previous data |
Contributor
Author
There was a problem hiding this comment.
removed this piece of code
manika137
reviewed
Nov 19, 2025
| @Test | ||
| public void verifyWriteRequestOfBufferSizeAndClose() throws Exception { | ||
|
|
||
| AbfsCounters abfsCounters = Mockito.spy(new AbfsCountersImpl(new URI("abcd"))); |
Contributor
There was a problem hiding this comment.
we can have a constant for URI "abcd"
Contributor
There was a problem hiding this comment.
if creating mock client handlers, client, abfsrestop, spying abfscounters is repeated- then we could have it as a helper method maybe
Contributor
Author
There was a problem hiding this comment.
if creating mock client handlers, client, abfsrestop, spying abfscounters is repeated- then we could have it as a helper method maybe -> was existing code will take it up later
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
Contributor
anujmodi2021
left a comment
There was a problem hiding this comment.
Partial Review. Will complete in some time
| @@ -0,0 +1,95 @@ | |||
| /** | |||
Contributor
There was a problem hiding this comment.
Do we need 2 searate classes for these?
All the metrics seems to be common in both.
Contributor
There was a problem hiding this comment.
Also these are not only threads but memory related metrics as well.
We can rename to something like: ABFSResourceUtilizationMetrics
Contributor
Author
There was a problem hiding this comment.
We can create a single class but in future, there may be some metrics not relevant to both of them, hence for separation of concerns felt it would be better to have 2 separate classes
| abfsUriQueryBuilder, cachedSasToken); | ||
|
|
||
| // Retrieve the read thread pool metrics from the ABFS counters. | ||
| AbfsReadThreadPoolMetrics metrics = getAbfsCounters() |
Contributor
There was a problem hiding this comment.
Common code in Blob and Dfs client Can be a common method in base class.
|
|
||
| /** | ||
| * Private constructor to prevent instantiation as this needs to be singleton. | ||
| * Initializes a new instance of {@code ReadBufferManagerV2} for the given ABFS client. |
Contributor
There was a problem hiding this comment.
All the clients in JVM ae supposed to use same ReadBufferManager. This comment seems wrong
| */ | ||
| private ReadBufferManagerV2() { | ||
| private ReadBufferManagerV2(AbfsClient abfsClient) { | ||
| this.abfsClient = abfsClient; |
Contributor
There was a problem hiding this comment.
Why do we need client here?
IMO we should not pass client to RBM.
Any client related work should happen in AbfsInputStream how its happening today.
Contributor
Author
There was a problem hiding this comment.
Passed the abfscounters directly instead of client
| <exclude>**/azurebfs/ITestSmallWriteOptimization.java</exclude> | ||
| <exclude>**/azurebfs/ITestAbfsStreamStatistics*.java</exclude> | ||
| <exclude>**/azurebfs/services/ITestReadBufferManager.java</exclude> | ||
| <exclude>**/azurebfs/WriteThreadPoolSizeManager.java</exclude> |
| * is initialized and not already monitoring. | ||
| */ | ||
| private void initializeMonitoringIfNeeded() { | ||
| if (poolSizeManager != null && !poolSizeManager.isMonitoringStarted()) { |
Contributor
There was a problem hiding this comment.
Do we need to repeat this check inside synchronized block as well?
What if 2 threads check together and both try to startCPUMonitoring?
| @@ -0,0 +1,165 @@ | |||
| /** | |||
Contributor
There was a problem hiding this comment.
Same here.
A lot ff redundant code can be combined.
| AppendRequestParameters reqParams, | ||
| TracingContext tracingContext) throws IOException { | ||
| TracingContext tracingContextAppend = new TracingContext(tracingContext); | ||
| // Fetches write thread pool metrics from the ABFS client and adds them to the tracing context. |
Contributor
There was a problem hiding this comment.
Common code Can be moved to base IngressHandler
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
anujmodi2021
requested changes
Nov 24, 2025
| @@ -0,0 +1,95 @@ | |||
| /** | |||
Contributor
There was a problem hiding this comment.
All of these metrics might be a overhead for some customer who does not wish to use new design for read and write and they might not wish to get these metrics.
If not already done, can we put this change behind a config so that these metrics comes only when enabled.
Contributor
Author
There was a problem hiding this comment.
already taken care of in AbfsClient class
| * Sets the metric results string used for tracing or logging. | ||
| * @param metricResults the formatted metric data to store. | ||
| */ | ||
| public void setMetricResults(final String metricResults) { |
Contributor
There was a problem hiding this comment.
These are only for Thread Pool metrics right?
Let's rename accordingly.
| // If metricHeader is present, append it to the client request ID header for tracing | ||
| if (!metricHeader.equals(EMPTY_STRING)) { | ||
| httpOperation.setRequestProperty(HttpHeaderConfigurations.X_MS_FECLIENT_METRICS, metricHeader); | ||
| httpOperation.setRequestProperty(HttpHeaderConfigurations.X_MS_CLIENT_REQUEST_ID, header + COLON + metricHeader); |
Contributor
There was a problem hiding this comment.
This would cause disparity in schema. With v2 we have made every field mandatory. If metrics are not there we should simply keep empty string but colon should always be there.
| abfsUriQueryBuilder, cachedSasToken); | ||
|
|
||
| // Retrieve the read thread pool metrics from the ABFS counters. | ||
| AbfsReadThreadPoolMetrics metrics = getAbfsCounters() |
Contributor
There was a problem hiding this comment.
Does getAbfsReadThreadPoolMetrics() returns aggregated metrics?
What defines the period over which these metrics are aggregated?
Contributor
Author
There was a problem hiding this comment.
So, these metrics are pushed only when thread pool size changes and whenever toString() method is called whatever we have aggregated till then is pushed and then they are reinitialized. There is no concept of a period for these metrics.
|
|
||
| AbfsReadFooterMetrics getAbfsReadFooterMetrics(); | ||
|
|
||
| void initializeReadMetrics(); |
Contributor
There was a problem hiding this comment.
Nit: rename to accomodate the fact that these are resource management related.
| * @return the highest JVM CPU utilization percentage recorded | ||
| */ | ||
| @VisibleForTesting | ||
| public double getMaxCpuUtilization() { |
Contributor
There was a problem hiding this comment.
Nit: maxJvmCPUUtilization
| V1("v1", 13); | ||
| V1("v1", 13), | ||
| /** | ||
| * Version 1 of the tracing header, which includes a version prefix and has 13 permanent fields. |
Contributor
There was a problem hiding this comment.
Nit: Edit javadoc to reflect v2
| * Schema: version:clientCorrelationId:clientRequestId:fileSystemId | ||
| * :primaryRequestId:streamId:opType:retryHeader:ingressHandler | ||
| * :position:operatedBlobCount:operationSpecificHeader:httpOperationHeader | ||
| * :networkLibrary:operationMetrics |
| * Schema: version:clientCorrelationId:clientRequestId:fileSystemId | ||
| * :primaryRequestId:streamId:opType:retryHeader:ingressHandler | ||
| * :position:operatedBlobCount:operationSpecificHeader:httpOperationHeader | ||
| * :networkLibrary:operationMetrics |
Contributor
There was a problem hiding this comment.
nit: instaed of operationMetrics, use resourceUtilizationMetrics
| */ | ||
| private long getAvailableHeapMemory() { | ||
| @VisibleForTesting | ||
| public long getAvailableHeapMemory() { |
Contributor
There was a problem hiding this comment.
Common methods between read and write.
Can be moved to some util class
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Introduces new performance metrics in the ABFS driver to monitor and evaluate the effectiveness of read and write aggressiveness tuning. These metrics help in understanding how thread pool behavior, CPU utilization, and heap availability impact overall I/O throughput and latency. By capturing detailed statistics such as active thread count, pool size, and system resource utilization, this enhancement enables data-driven analysis of optimizations made to improve ABFS read and write performance under varying workloads.