You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by st...@apache.org on 2020/04/22 16:24:54 UTC
[hadoop] branch branch-3.3 updated (54a64e5 -> f53ded6)
This is an automated email from the ASF dual-hosted git repository.
stevel pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git.
from 54a64e5 HADOOP-16944. Use Yetus 0.12.0 in GitHub PR (#1917)
new f74a571 HADOOP-16910 : ABFS Streams to update FileSystem.Statistics counters on IO. (#1918). Contributed by Mehakmeet Singh.
new 98fdbb8 HADOOP-16965. Refactor abfs stream configuration. (#1956)
new f53ded6 HADOOP-16922. ABFS: Change User-Agent header (#1938)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../hadoop/fs/azurebfs/AbfsConfiguration.java | 18 +-
.../hadoop/fs/azurebfs/AzureBlobFileSystem.java | 4 +-
.../fs/azurebfs/AzureBlobFileSystemStore.java | 56 ++++--
.../fs/azurebfs/constants/AbfsHttpConstants.java | 3 +
.../fs/azurebfs/constants/ConfigurationKeys.java | 2 +
.../constants/FileSystemConfigurations.java | 5 +
.../hadoop/fs/azurebfs/services/AbfsClient.java | 70 +++++--
.../fs/azurebfs/services/AbfsInputStream.java | 32 ++-
.../azurebfs/services/AbfsInputStreamContext.java | 70 +++++++
.../fs/azurebfs/services/AbfsOutputStream.java | 32 ++-
.../azurebfs/services/AbfsOutputStreamContext.java | 68 +++++++
.../{AuthType.java => AbfsStreamContext.java} | 10 +-
.../fs/azurebfs/AbstractAbfsTestWithTimeout.java | 59 ++++++
.../fs/azurebfs/ITestAbfsStreamStatistics.java | 157 +++++++++++++++
.../fs/azurebfs/ITestAzureBlobFileSystemOauth.java | 11 +-
.../fs/azurebfs/services/TestAbfsClient.java | 224 +++++++++++++++++----
16 files changed, 711 insertions(+), 110 deletions(-)
create mode 100644 hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStreamContext.java
create mode 100644 hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStreamContext.java
copy hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/{AuthType.java => AbfsStreamContext.java} (85%)
create mode 100644 hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsStreamStatistics.java
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[hadoop] 02/03: HADOOP-16965. Refactor abfs stream configuration.
(#1956)
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stevel pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 98fdbb820ea180a3c7b94a4b09ed96858a803d58
Author: Mukund Thakur <mu...@users.noreply.github.com>
AuthorDate: Tue Apr 21 21:57:29 2020 +0530
HADOOP-16965. Refactor abfs stream configuration. (#1956)
Contributed by Mukund Thakur.
---
.../fs/azurebfs/AzureBlobFileSystemStore.java | 30 +++++++---
.../fs/azurebfs/services/AbfsInputStream.java | 20 +++----
.../azurebfs/services/AbfsInputStreamContext.java | 70 ++++++++++++++++++++++
.../fs/azurebfs/services/AbfsOutputStream.java | 19 +++---
.../azurebfs/services/AbfsOutputStreamContext.java | 68 +++++++++++++++++++++
.../fs/azurebfs/services/AbfsStreamContext.java | 26 ++++++++
6 files changed, 204 insertions(+), 29 deletions(-)
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
index a330da4..6b194a4 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
@@ -81,7 +81,9 @@ import org.apache.hadoop.fs.azurebfs.services.AbfsAclHelper;
import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
import org.apache.hadoop.fs.azurebfs.services.AbfsHttpOperation;
import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
+import org.apache.hadoop.fs.azurebfs.services.AbfsInputStreamContext;
import org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream;
+import org.apache.hadoop.fs.azurebfs.services.AbfsOutputStreamContext;
import org.apache.hadoop.fs.azurebfs.services.AbfsPermission;
import org.apache.hadoop.fs.azurebfs.services.AbfsRestOperation;
import org.apache.hadoop.fs.azurebfs.services.AuthType;
@@ -415,12 +417,18 @@ public class AzureBlobFileSystemStore implements Closeable {
statistics,
AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path),
0,
- abfsConfiguration.getWriteBufferSize(),
- abfsConfiguration.isFlushEnabled(),
- abfsConfiguration.isOutputStreamFlushDisabled());
+ populateAbfsOutputStreamContext());
}
}
+ private AbfsOutputStreamContext populateAbfsOutputStreamContext() {
+ return new AbfsOutputStreamContext()
+ .withWriteBufferSize(abfsConfiguration.getWriteBufferSize())
+ .enableFlush(abfsConfiguration.isFlushEnabled())
+ .disableOutputStreamFlush(abfsConfiguration.isOutputStreamFlushDisabled())
+ .build();
+ }
+
public void createDirectory(final Path path, final FsPermission permission, final FsPermission umask)
throws AzureBlobFileSystemException {
try (AbfsPerfInfo perfInfo = startTracking("createDirectory", "createPath")) {
@@ -466,11 +474,19 @@ public class AzureBlobFileSystemStore implements Closeable {
// Add statistics for InputStream
return new AbfsInputStream(client, statistics,
AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path), contentLength,
- abfsConfiguration.getReadBufferSize(), abfsConfiguration.getReadAheadQueueDepth(),
- abfsConfiguration.getTolerateOobAppends(), eTag);
+ populateAbfsInputStreamContext(),
+ eTag);
}
}
+ private AbfsInputStreamContext populateAbfsInputStreamContext() {
+ return new AbfsInputStreamContext()
+ .withReadBufferSize(abfsConfiguration.getReadBufferSize())
+ .withReadAheadQueueDepth(abfsConfiguration.getReadAheadQueueDepth())
+ .withTolerateOobAppends(abfsConfiguration.getTolerateOobAppends())
+ .build();
+ }
+
public OutputStream openFileForWrite(final Path path, final FileSystem.Statistics statistics, final boolean overwrite) throws
AzureBlobFileSystemException {
try (AbfsPerfInfo perfInfo = startTracking("openFileForWrite", "getPathStatus")) {
@@ -502,9 +518,7 @@ public class AzureBlobFileSystemStore implements Closeable {
statistics,
AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path),
offset,
- abfsConfiguration.getWriteBufferSize(),
- abfsConfiguration.isFlushEnabled(),
- abfsConfiguration.isOutputStreamFlushDisabled());
+ populateAbfsOutputStreamContext());
}
}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
index 0c06014..05c093a 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
@@ -61,21 +61,19 @@ public class AbfsInputStream extends FSInputStream implements CanUnbuffer,
private boolean closed = false;
public AbfsInputStream(
- final AbfsClient client,
- final Statistics statistics,
- final String path,
- final long contentLength,
- final int bufferSize,
- final int readAheadQueueDepth,
- final boolean tolerateOobAppends,
- final String eTag) {
+ final AbfsClient client,
+ final Statistics statistics,
+ final String path,
+ final long contentLength,
+ final AbfsInputStreamContext abfsInputStreamContext,
+ final String eTag) {
this.client = client;
this.statistics = statistics;
this.path = path;
this.contentLength = contentLength;
- this.bufferSize = bufferSize;
- this.readAheadQueueDepth = (readAheadQueueDepth >= 0) ? readAheadQueueDepth : Runtime.getRuntime().availableProcessors();
- this.tolerateOobAppends = tolerateOobAppends;
+ this.bufferSize = abfsInputStreamContext.getReadBufferSize();
+ this.readAheadQueueDepth = abfsInputStreamContext.getReadAheadQueueDepth();
+ this.tolerateOobAppends = abfsInputStreamContext.isTolerateOobAppends();
this.eTag = eTag;
this.readAheadEnabled = true;
}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStreamContext.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStreamContext.java
new file mode 100644
index 0000000..cba7191
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStreamContext.java
@@ -0,0 +1,70 @@
+/**
+ * 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.hadoop.fs.azurebfs.services;
+
+/**
+ * Class to hold extra input stream configs.
+ */
+public class AbfsInputStreamContext extends AbfsStreamContext {
+
+ private int readBufferSize;
+
+ private int readAheadQueueDepth;
+
+ private boolean tolerateOobAppends;
+
+ public AbfsInputStreamContext() {
+ }
+
+ public AbfsInputStreamContext withReadBufferSize(final int readBufferSize) {
+ this.readBufferSize = readBufferSize;
+ return this;
+ }
+
+ public AbfsInputStreamContext withReadAheadQueueDepth(
+ final int readAheadQueueDepth) {
+ this.readAheadQueueDepth = (readAheadQueueDepth >= 0)
+ ? readAheadQueueDepth
+ : Runtime.getRuntime().availableProcessors();
+ return this;
+ }
+
+ public AbfsInputStreamContext withTolerateOobAppends(
+ final boolean tolerateOobAppends) {
+ this.tolerateOobAppends = tolerateOobAppends;
+ return this;
+ }
+
+ public AbfsInputStreamContext build() {
+ // Validation of parameters to be done here.
+ return this;
+ }
+
+ public int getReadBufferSize() {
+ return readBufferSize;
+ }
+
+ public int getReadAheadQueueDepth() {
+ return readAheadQueueDepth;
+ }
+
+ public boolean isTolerateOobAppends() {
+ return tolerateOobAppends;
+ }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java
index e943169..f29cc4a 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java
@@ -82,23 +82,22 @@ public class AbfsOutputStream extends OutputStream implements Syncable, StreamCa
private final Statistics statistics;
public AbfsOutputStream(
- final AbfsClient client,
- final Statistics statistics,
- final String path,
- final long position,
- final int bufferSize,
- final boolean supportFlush,
- final boolean disableOutputStreamFlush) {
+ final AbfsClient client,
+ final Statistics statistics,
+ final String path,
+ final long position,
+ AbfsOutputStreamContext abfsOutputStreamContext) {
this.client = client;
this.statistics = statistics;
this.path = path;
this.position = position;
this.closed = false;
- this.supportFlush = supportFlush;
- this.disableOutputStreamFlush = disableOutputStreamFlush;
+ this.supportFlush = abfsOutputStreamContext.isEnableFlush();
+ this.disableOutputStreamFlush = abfsOutputStreamContext
+ .isDisableOutputStreamFlush();
this.lastError = null;
this.lastFlushOffset = 0;
- this.bufferSize = bufferSize;
+ this.bufferSize = abfsOutputStreamContext.getWriteBufferSize();
this.buffer = byteBufferPool.getBuffer(false, bufferSize).array();
this.bufferIndex = 0;
this.writeOperations = new ConcurrentLinkedDeque<>();
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStreamContext.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStreamContext.java
new file mode 100644
index 0000000..0be97c5
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStreamContext.java
@@ -0,0 +1,68 @@
+/**
+ * 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.hadoop.fs.azurebfs.services;
+
+/**
+ * Class to hold extra output stream configs.
+ */
+public class AbfsOutputStreamContext extends AbfsStreamContext {
+
+ private int writeBufferSize;
+
+ private boolean enableFlush;
+
+ private boolean disableOutputStreamFlush;
+
+ public AbfsOutputStreamContext() {
+ }
+
+ public AbfsOutputStreamContext withWriteBufferSize(
+ final int writeBufferSize) {
+ this.writeBufferSize = writeBufferSize;
+ return this;
+ }
+
+ public AbfsOutputStreamContext enableFlush(final boolean enableFlush) {
+ this.enableFlush = enableFlush;
+ return this;
+ }
+
+ public AbfsOutputStreamContext disableOutputStreamFlush(
+ final boolean disableOutputStreamFlush) {
+ this.disableOutputStreamFlush = disableOutputStreamFlush;
+ return this;
+ }
+
+ public AbfsOutputStreamContext build() {
+ // Validation of parameters to be done here.
+ return this;
+ }
+
+ public int getWriteBufferSize() {
+ return writeBufferSize;
+ }
+
+ public boolean isEnableFlush() {
+ return enableFlush;
+ }
+
+ public boolean isDisableOutputStreamFlush() {
+ return disableOutputStreamFlush;
+ }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsStreamContext.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsStreamContext.java
new file mode 100644
index 0000000..ee77f59
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsStreamContext.java
@@ -0,0 +1,26 @@
+/**
+ * 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.hadoop.fs.azurebfs.services;
+
+/**
+ * Base stream configuration class which is going
+ * to store common configs among input and output streams.
+ */
+public abstract class AbfsStreamContext {
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[hadoop] 01/03: HADOOP-16910 : ABFS Streams to update
FileSystem.Statistics counters on IO. (#1918). Contributed by Mehakmeet
Singh.
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stevel pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit f74a571fdfe184613817b17f0884dce613ce4b4f
Author: Mehakmeet Singh <me...@gmail.com>
AuthorDate: Tue Mar 31 18:19:09 2020 +0530
HADOOP-16910 : ABFS Streams to update FileSystem.Statistics counters on IO. (#1918). Contributed by Mehakmeet Singh.
---
.../hadoop/fs/azurebfs/AzureBlobFileSystem.java | 4 +-
.../fs/azurebfs/AzureBlobFileSystemStore.java | 38 ++---
.../fs/azurebfs/services/AbfsInputStream.java | 12 ++
.../fs/azurebfs/services/AbfsOutputStream.java | 15 ++
.../fs/azurebfs/AbstractAbfsTestWithTimeout.java | 59 ++++++++
.../fs/azurebfs/ITestAbfsStreamStatistics.java | 157 +++++++++++++++++++++
.../fs/azurebfs/ITestAzureBlobFileSystemOauth.java | 11 +-
7 files changed, 275 insertions(+), 21 deletions(-)
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java
index 8eda2f3..4ddc2e3 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java
@@ -188,7 +188,7 @@ public class AzureBlobFileSystem extends FileSystem {
Path qualifiedPath = makeQualified(f);
try {
- OutputStream outputStream = abfsStore.createFile(qualifiedPath, overwrite,
+ OutputStream outputStream = abfsStore.createFile(qualifiedPath, statistics, overwrite,
permission == null ? FsPermission.getFileDefault() : permission, FsPermission.getUMask(getConf()));
return new FSDataOutputStream(outputStream, statistics);
} catch(AzureBlobFileSystemException ex) {
@@ -250,7 +250,7 @@ public class AzureBlobFileSystem extends FileSystem {
Path qualifiedPath = makeQualified(f);
try {
- OutputStream outputStream = abfsStore.openFileForWrite(qualifiedPath, false);
+ OutputStream outputStream = abfsStore.openFileForWrite(qualifiedPath, statistics, false);
return new FSDataOutputStream(outputStream, statistics);
} catch(AzureBlobFileSystemException ex) {
checkException(f, ex);
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
index bff0e45..a330da4 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java
@@ -51,6 +51,8 @@ import java.util.Set;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
@@ -97,8 +99,6 @@ import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.http.client.utils.URIBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_EQUALS;
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_FORWARD_SLASH;
@@ -391,8 +391,10 @@ public class AzureBlobFileSystemStore implements Closeable {
}
}
- public OutputStream createFile(final Path path, final boolean overwrite, final FsPermission permission,
- final FsPermission umask) throws AzureBlobFileSystemException {
+ public OutputStream createFile(final Path path,
+ final FileSystem.Statistics statistics,
+ final boolean overwrite, final FsPermission permission,
+ final FsPermission umask) throws AzureBlobFileSystemException {
try (AbfsPerfInfo perfInfo = startTracking("createFile", "createPath")) {
boolean isNamespaceEnabled = getIsNamespaceEnabled();
LOG.debug("createFile filesystem: {} path: {} overwrite: {} permission: {} umask: {} isNamespaceEnabled: {}",
@@ -409,12 +411,13 @@ public class AzureBlobFileSystemStore implements Closeable {
perfInfo.registerResult(op.getResult()).registerSuccess(true);
return new AbfsOutputStream(
- client,
- AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path),
- 0,
- abfsConfiguration.getWriteBufferSize(),
- abfsConfiguration.isFlushEnabled(),
- abfsConfiguration.isOutputStreamFlushDisabled());
+ client,
+ statistics,
+ AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path),
+ 0,
+ abfsConfiguration.getWriteBufferSize(),
+ abfsConfiguration.isFlushEnabled(),
+ abfsConfiguration.isOutputStreamFlushDisabled());
}
}
@@ -468,7 +471,7 @@ public class AzureBlobFileSystemStore implements Closeable {
}
}
- public OutputStream openFileForWrite(final Path path, final boolean overwrite) throws
+ public OutputStream openFileForWrite(final Path path, final FileSystem.Statistics statistics, final boolean overwrite) throws
AzureBlobFileSystemException {
try (AbfsPerfInfo perfInfo = startTracking("openFileForWrite", "getPathStatus")) {
LOG.debug("openFileForWrite filesystem: {} path: {} overwrite: {}",
@@ -495,12 +498,13 @@ public class AzureBlobFileSystemStore implements Closeable {
perfInfo.registerSuccess(true);
return new AbfsOutputStream(
- client,
- AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path),
- offset,
- abfsConfiguration.getWriteBufferSize(),
- abfsConfiguration.isFlushEnabled(),
- abfsConfiguration.isOutputStreamFlushDisabled());
+ client,
+ statistics,
+ AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path),
+ offset,
+ abfsConfiguration.getWriteBufferSize(),
+ abfsConfiguration.isFlushEnabled(),
+ abfsConfiguration.isOutputStreamFlushDisabled());
}
}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
index 8dc3b8f..0c06014 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java
@@ -101,6 +101,7 @@ public class AbfsInputStream extends FSInputStream implements CanUnbuffer,
int currentLen = len;
int lastReadBytes;
int totalReadBytes = 0;
+ incrementReadOps();
do {
lastReadBytes = readOneBlock(b, currentOff, currentLen);
if (lastReadBytes > 0) {
@@ -201,6 +202,7 @@ public class AbfsInputStream extends FSInputStream implements CanUnbuffer,
// try reading from buffers first
receivedBytes = ReadBufferManager.getBufferManager().getBlock(this, position, length, b);
if (receivedBytes > 0) {
+ incrementReadOps();
return receivedBytes;
}
@@ -236,6 +238,7 @@ public class AbfsInputStream extends FSInputStream implements CanUnbuffer,
try (AbfsPerfInfo perfInfo = new AbfsPerfInfo(tracker, "readRemote", "read")) {
op = client.read(path, position, b, offset, length, tolerateOobAppends ? "*" : eTag);
perfInfo.registerResult(op.getResult()).registerSuccess(true);
+ incrementReadOps();
} catch (AzureBlobFileSystemException ex) {
if (ex instanceof AbfsRestOperationException) {
AbfsRestOperationException ere = (AbfsRestOperationException) ex;
@@ -253,6 +256,15 @@ public class AbfsInputStream extends FSInputStream implements CanUnbuffer,
}
/**
+ * Increment Read Operations.
+ */
+ private void incrementReadOps() {
+ if (statistics != null) {
+ statistics.incrementReadOps(1);
+ }
+ }
+
+ /**
* Seek to given position in stream.
* @param n position to seek to
* @throws IOException if there is an error
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java
index 7e9746d..e943169 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsOutputStream.java
@@ -39,6 +39,7 @@ import com.google.common.base.Preconditions;
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
import org.apache.hadoop.io.ElasticByteBufferPool;
+import org.apache.hadoop.fs.FileSystem.Statistics;
import org.apache.hadoop.fs.FSExceptionMessages;
import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.fs.Syncable;
@@ -78,14 +79,18 @@ public class AbfsOutputStream extends OutputStream implements Syncable, StreamCa
private final ElasticByteBufferPool byteBufferPool
= new ElasticByteBufferPool();
+ private final Statistics statistics;
+
public AbfsOutputStream(
final AbfsClient client,
+ final Statistics statistics,
final String path,
final long position,
final int bufferSize,
final boolean supportFlush,
final boolean disableOutputStreamFlush) {
this.client = client;
+ this.statistics = statistics;
this.path = path;
this.position = position;
this.closed = false;
@@ -181,6 +186,16 @@ public class AbfsOutputStream extends OutputStream implements Syncable, StreamCa
writableBytes = bufferSize - bufferIndex;
}
+ incrementWriteOps();
+ }
+
+ /**
+ * Increment Write Operations.
+ */
+ private void incrementWriteOps() {
+ if (statistics != null) {
+ statistics.incrementWriteOps(1);
+ }
}
/**
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsTestWithTimeout.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsTestWithTimeout.java
index fee90ab..0485422 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsTestWithTimeout.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/AbstractAbfsTestWithTimeout.java
@@ -17,12 +17,19 @@
*/
package org.apache.hadoop.fs.azurebfs;
+import java.io.IOException;
+
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.rules.TestName;
import org.junit.rules.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.Path;
import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.TEST_TIMEOUT;
@@ -31,6 +38,9 @@ import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.TEST
* This class does not attempt to bind to Azure.
*/
public class AbstractAbfsTestWithTimeout extends Assert {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractAbfsTestWithTimeout.class);
+
/**
* The name of the current method.
*/
@@ -67,4 +77,53 @@ public class AbstractAbfsTestWithTimeout extends Assert {
protected int getTestTimeoutMillis() {
return TEST_TIMEOUT;
}
+
+ /**
+ * Describe a test in the logs.
+ *
+ * @param text text to print
+ * @param args arguments to format in the printing
+ */
+ protected void describe(String text, Object... args) {
+ LOG.info("\n\n{}: {}\n",
+ methodName.getMethodName(),
+ String.format(text, args));
+ }
+
+ /**
+ * Validate Contents written on a file in Abfs.
+ *
+ * @param fs AzureBlobFileSystem
+ * @param path Path of the file
+ * @param originalByteArray original byte array
+ * @return if content is validated true else, false
+ * @throws IOException
+ */
+ protected boolean validateContent(AzureBlobFileSystem fs, Path path,
+ byte[] originalByteArray)
+ throws IOException {
+ int pos = 0;
+ int lenOfOriginalByteArray = originalByteArray.length;
+
+ try (FSDataInputStream in = fs.open(path)) {
+ byte valueOfContentAtPos = (byte) in.read();
+
+ while (valueOfContentAtPos != -1 && pos < lenOfOriginalByteArray) {
+ if (originalByteArray[pos] != valueOfContentAtPos) {
+ assertEquals("Mismatch in content validation at position {}", pos,
+ originalByteArray[pos], valueOfContentAtPos);
+ return false;
+ }
+ valueOfContentAtPos = (byte) in.read();
+ pos++;
+ }
+ if (valueOfContentAtPos != -1) {
+ assertEquals("Expected end of file", -1, valueOfContentAtPos);
+ return false;
+ }
+ return true;
+ }
+
+ }
+
}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsStreamStatistics.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsStreamStatistics.java
new file mode 100644
index 0000000..b749f49
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsStreamStatistics.java
@@ -0,0 +1,157 @@
+/**
+ * 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.hadoop.fs.azurebfs;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
+
+/**
+ * Test Abfs Stream.
+ */
+
+public class ITestAbfsStreamStatistics extends AbstractAbfsIntegrationTest {
+ public ITestAbfsStreamStatistics() throws Exception {
+ }
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ITestAbfsStreamStatistics.class);
+
+ private static final int LARGE_NUMBER_OF_OPS = 999999;
+
+ /***
+ * Testing {@code incrementReadOps()} in class {@code AbfsInputStream} and
+ * {@code incrementWriteOps()} in class {@code AbfsOutputStream}.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testAbfsStreamOps() throws Exception {
+ describe("Test to see correct population of read and write operations in "
+ + "Abfs");
+
+ final AzureBlobFileSystem fs = getFileSystem();
+ Path smallOperationsFile = new Path("testOneReadWriteOps");
+ Path largeOperationsFile = new Path("testLargeReadWriteOps");
+ FileSystem.Statistics statistics = fs.getFsStatistics();
+ String testReadWriteOps = "test this";
+ statistics.reset();
+
+ //Test for zero write operation
+ assertReadWriteOps("write", 0, statistics.getWriteOps());
+
+ //Test for zero read operation
+ assertReadWriteOps("read", 0, statistics.getReadOps());
+
+ FSDataOutputStream outForOneOperation = null;
+ FSDataInputStream inForOneOperation = null;
+ try {
+ outForOneOperation = fs.create(smallOperationsFile);
+ statistics.reset();
+ outForOneOperation.write(testReadWriteOps.getBytes());
+
+ //Test for a single write operation
+ assertReadWriteOps("write", 1, statistics.getWriteOps());
+
+ //Flushing output stream to see content to read
+ outForOneOperation.hflush();
+ inForOneOperation = fs.open(smallOperationsFile);
+ statistics.reset();
+ int result = inForOneOperation.read(testReadWriteOps.getBytes(), 0,
+ testReadWriteOps.getBytes().length);
+
+ LOG.info("Result of Read operation : {}", result);
+ /*
+ Testing if 2 read_ops value is coming after reading full content from a
+ file (3 if anything to read from Buffer too).
+ Reason: read() call gives read_ops=1,
+ reading from AbfsClient(http GET) gives read_ops=2.
+ */
+ assertReadWriteOps("read", 2, statistics.getReadOps());
+
+ } finally {
+ IOUtils.cleanupWithLogger(LOG, inForOneOperation,
+ outForOneOperation);
+ }
+
+ //Validating if content is being written in the smallOperationsFile
+ assertTrue("Mismatch in content validation",
+ validateContent(fs, smallOperationsFile,
+ testReadWriteOps.getBytes()));
+
+ FSDataOutputStream outForLargeOperations = null;
+ FSDataInputStream inForLargeOperations = null;
+ StringBuilder largeOperationsValidationString = new StringBuilder();
+ try {
+ outForLargeOperations = fs.create(largeOperationsFile);
+ statistics.reset();
+ int largeValue = LARGE_NUMBER_OF_OPS;
+ for (int i = 0; i < largeValue; i++) {
+ outForLargeOperations.write(testReadWriteOps.getBytes());
+
+ //Creating the String for content Validation
+ largeOperationsValidationString.append(testReadWriteOps);
+ }
+ LOG.info("Number of bytes of Large data written: {}",
+ largeOperationsValidationString.toString().getBytes().length);
+
+ //Test for 1000000 write operations
+ assertReadWriteOps("write", largeValue, statistics.getWriteOps());
+
+ inForLargeOperations = fs.open(largeOperationsFile);
+ for (int i = 0; i < largeValue; i++) {
+ inForLargeOperations
+ .read(testReadWriteOps.getBytes(), 0,
+ testReadWriteOps.getBytes().length);
+ }
+
+ //Test for 1000000 read operations
+ assertReadWriteOps("read", largeValue, statistics.getReadOps());
+
+ } finally {
+ IOUtils.cleanupWithLogger(LOG, inForLargeOperations,
+ outForLargeOperations);
+ }
+ //Validating if content is being written in largeOperationsFile
+ assertTrue("Mismatch in content validation",
+ validateContent(fs, largeOperationsFile,
+ largeOperationsValidationString.toString().getBytes()));
+
+ }
+
+ /**
+ * Generic method to assert both Read an write operations.
+ *
+ * @param operation what operation is being asserted
+ * @param expectedValue value which is expected
+ * @param actualValue value which is actual
+ */
+
+ private void assertReadWriteOps(String operation, long expectedValue,
+ long actualValue) {
+ assertEquals("Mismatch in " + operation + " operations", expectedValue,
+ actualValue);
+ }
+}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemOauth.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemOauth.java
index 533f471..e517f68 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemOauth.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemOauth.java
@@ -22,18 +22,21 @@ import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
-import org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys;
import org.junit.Assume;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys;
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
import org.apache.hadoop.fs.azurebfs.services.AuthType;
+import org.apache.hadoop.io.IOUtils;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_OAUTH_CLIENT_ID;
import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_OAUTH_CLIENT_SECRET;
@@ -52,6 +55,8 @@ public class ITestAzureBlobFileSystemOauth extends AbstractAbfsIntegrationTest{
private static final Path FILE_PATH = new Path("/testFile");
private static final Path EXISTED_FILE_PATH = new Path("/existedFile");
private static final Path EXISTED_FOLDER_PATH = new Path("/existedFolder");
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ITestAbfsStreamStatistics.class);
public ITestAzureBlobFileSystemOauth() throws Exception {
Assume.assumeTrue(this.getAuthType() == AuthType.OAuth);
@@ -143,9 +148,11 @@ public class ITestAzureBlobFileSystemOauth extends AbstractAbfsIntegrationTest{
// TEST WRITE FILE
try {
- abfsStore.openFileForWrite(EXISTED_FILE_PATH, true);
+ abfsStore.openFileForWrite(EXISTED_FILE_PATH, fs.getFsStatistics(), true);
} catch (AbfsRestOperationException e) {
assertEquals(AzureServiceErrorCode.AUTHORIZATION_PERMISSION_MISS_MATCH, e.getErrorCode());
+ } finally {
+ IOUtils.cleanupWithLogger(LOG, abfsStore);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[hadoop] 03/03: HADOOP-16922. ABFS: Change User-Agent header (#1938)
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stevel pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit f53ded618539a409d9b8efa041b96457f4e873c2
Author: bilaharith <52...@users.noreply.github.com>
AuthorDate: Tue Apr 21 22:07:40 2020 +0530
HADOOP-16922. ABFS: Change User-Agent header (#1938)
Contributed by Bilahari T H.
---
.../hadoop/fs/azurebfs/AbfsConfiguration.java | 18 +-
.../fs/azurebfs/constants/AbfsHttpConstants.java | 3 +
.../fs/azurebfs/constants/ConfigurationKeys.java | 2 +
.../constants/FileSystemConfigurations.java | 5 +
.../hadoop/fs/azurebfs/services/AbfsClient.java | 70 +++++--
.../fs/azurebfs/services/TestAbfsClient.java | 224 +++++++++++++++++----
6 files changed, 261 insertions(+), 61 deletions(-)
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java
index 61fe3d8..084b558 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java
@@ -168,9 +168,17 @@ public class AbfsConfiguration{
private boolean enableAutoThrottling;
@StringConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_USER_AGENT_PREFIX_KEY,
- DefaultValue = "")
+ DefaultValue = DEFAULT_FS_AZURE_USER_AGENT_PREFIX)
private String userAgentId;
+ @StringConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_CLUSTER_NAME,
+ DefaultValue = DEFAULT_VALUE_UNKNOWN)
+ private String clusterName;
+
+ @StringConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_CLUSTER_TYPE,
+ DefaultValue = DEFAULT_VALUE_UNKNOWN)
+ private String clusterType;
+
@BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_ENABLE_DELEGATION_TOKEN,
DefaultValue = DEFAULT_ENABLE_DELEGATION_TOKEN)
private boolean enableDelegationToken;
@@ -479,6 +487,14 @@ public class AbfsConfiguration{
return this.userAgentId;
}
+ public String getClusterName() {
+ return this.clusterName;
+ }
+
+ public String getClusterType() {
+ return this.clusterType;
+ }
+
public DelegatingSSLSocketFactory.SSLChannelMode getPreferredSSLFactoryOption() {
return getEnum(FS_AZURE_SSL_CHANNEL_MODE_KEY, DEFAULT_FS_AZURE_SSL_CHANNEL_MODE);
}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java
index c6ade9c..42dc923 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java
@@ -42,10 +42,13 @@ public final class AbfsHttpConstants {
public static final String DEFAULT_TIMEOUT = "90";
public static final String TOKEN_VERSION = "2";
+ public static final String JAVA_VENDOR = "java.vendor";
public static final String JAVA_VERSION = "java.version";
public static final String OS_NAME = "os.name";
public static final String OS_VERSION = "os.version";
+ public static final String OS_ARCH = "os.arch";
+ public static final String APN_VERSION = "APN/1.0";
public static final String CLIENT_VERSION = "Azure Blob FS/" + VersionInfo.getVersion();
// Abfs Http Verb
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java
index a63e953..3d8b987 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java
@@ -62,6 +62,8 @@ public final class ConfigurationKeys {
* Default value of this config is true. **/
public static final String FS_AZURE_DISABLE_OUTPUTSTREAM_FLUSH = "fs.azure.disable.outputstream.flush";
public static final String FS_AZURE_USER_AGENT_PREFIX_KEY = "fs.azure.user.agent.prefix";
+ public static final String FS_AZURE_CLUSTER_NAME = "fs.azure.cluster.name";
+ public static final String FS_AZURE_CLUSTER_TYPE = "fs.azure.cluster.type";
public static final String FS_AZURE_SSL_CHANNEL_MODE_KEY = "fs.azure.ssl.channel.mode";
/** Provides a config to enable/disable the checkAccess API.
* By default this will be
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java
index c6b308e..5c27d84 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java
@@ -22,6 +22,8 @@ import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.security.ssl.DelegatingSSLSocketFactory;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
+
/**
* Responsible to keep all the Azure Blob File System related configurations.
*/
@@ -71,5 +73,8 @@ public final class FileSystemConfigurations {
public static final boolean DEFAULT_ENABLE_CHECK_ACCESS = false;
public static final boolean DEFAULT_ABFS_LATENCY_TRACK = false;
+ public static final String DEFAULT_FS_AZURE_USER_AGENT_PREFIX = EMPTY_STRING;
+ public static final String DEFAULT_VALUE_UNKNOWN = "UNKNOWN";
+
private FileSystemConfigurations() {}
}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java
index 6e1de68..6bacde4 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java
@@ -679,32 +679,60 @@ public class AbfsClient implements Closeable {
@VisibleForTesting
String initializeUserAgent(final AbfsConfiguration abfsConfiguration,
- final String sslProviderName) {
+ final String sslProviderName) {
+
StringBuilder sb = new StringBuilder();
- sb.append("(JavaJRE ");
+
+ sb.append(APN_VERSION);
+ sb.append(SINGLE_WHITE_SPACE);
+ sb.append(CLIENT_VERSION);
+ sb.append(SINGLE_WHITE_SPACE);
+
+ sb.append("(");
+
+ sb.append(System.getProperty(JAVA_VENDOR)
+ .replaceAll(SINGLE_WHITE_SPACE, EMPTY_STRING));
+ sb.append(SINGLE_WHITE_SPACE);
+ sb.append("JavaJRE");
+ sb.append(SINGLE_WHITE_SPACE);
sb.append(System.getProperty(JAVA_VERSION));
- sb.append("; ");
- sb.append(
- System.getProperty(OS_NAME).replaceAll(SINGLE_WHITE_SPACE, EMPTY_STRING));
- sb.append(" ");
+ sb.append(SEMICOLON);
+ sb.append(SINGLE_WHITE_SPACE);
+
+ sb.append(System.getProperty(OS_NAME)
+ .replaceAll(SINGLE_WHITE_SPACE, EMPTY_STRING));
+ sb.append(SINGLE_WHITE_SPACE);
sb.append(System.getProperty(OS_VERSION));
- if (sslProviderName != null && !sslProviderName.isEmpty()) {
- sb.append("; ");
- sb.append(sslProviderName);
- }
- String tokenProviderField =
- ExtensionHelper.getUserAgentSuffix(tokenProvider, "");
- if (!tokenProviderField.isEmpty()) {
- sb.append("; ").append(tokenProviderField);
- }
+ sb.append(FORWARD_SLASH);
+ sb.append(System.getProperty(OS_ARCH));
+ sb.append(SEMICOLON);
+
+ appendIfNotEmpty(sb, sslProviderName, true);
+ appendIfNotEmpty(sb,
+ ExtensionHelper.getUserAgentSuffix(tokenProvider, EMPTY_STRING), true);
+
+ sb.append(SINGLE_WHITE_SPACE);
+ sb.append(abfsConfiguration.getClusterName());
+ sb.append(FORWARD_SLASH);
+ sb.append(abfsConfiguration.getClusterType());
+
sb.append(")");
- final String userAgentComment = sb.toString();
- String customUserAgentId = abfsConfiguration.getCustomUserAgentPrefix();
- if (customUserAgentId != null && !customUserAgentId.isEmpty()) {
- return String.format(Locale.ROOT, CLIENT_VERSION + " %s %s",
- userAgentComment, customUserAgentId);
+
+ appendIfNotEmpty(sb, abfsConfiguration.getCustomUserAgentPrefix(), false);
+
+ return String.format(Locale.ROOT, sb.toString());
+ }
+
+ private void appendIfNotEmpty(StringBuilder sb, String regEx,
+ boolean shouldAppendSemiColon) {
+ if (regEx == null || regEx.trim().isEmpty()) {
+ return;
+ }
+ sb.append(SINGLE_WHITE_SPACE);
+ sb.append(regEx);
+ if (shouldAppendSemiColon) {
+ sb.append(SEMICOLON);
}
- return String.format(Locale.ROOT, CLIENT_VERSION + " %s", userAgentComment);
}
@VisibleForTesting
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java
index deca8b3..724f57d 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsClient.java
@@ -18,11 +18,11 @@
package org.apache.hadoop.fs.azurebfs.services;
+import java.io.IOException;
+import java.net.MalformedURLException;
import java.net.URL;
import java.util.regex.Pattern;
-import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
-import org.junit.Assert;
import org.junit.Test;
import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
@@ -30,7 +30,24 @@ import org.apache.hadoop.fs.azurebfs.oauth2.AccessTokenProvider;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys;
import org.apache.hadoop.security.ssl.DelegatingSSLSocketFactory;
-import org.apache.hadoop.util.VersionInfo;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.APN_VERSION;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CLIENT_VERSION;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.JAVA_VENDOR;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.JAVA_VERSION;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.OS_ARCH;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.OS_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.OS_VERSION;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.SEMICOLON;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.SINGLE_WHITE_SPACE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_CLUSTER_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_CLUSTER_TYPE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_VALUE_UNKNOWN;
+import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.TEST_CONFIGURATION_FILE_NAME;
/**
* Test useragent of abfs client.
@@ -38,60 +55,189 @@ import org.apache.hadoop.util.VersionInfo;
*/
public final class TestAbfsClient {
- private final String accountName = "bogusAccountName.dfs.core.windows.net";
+ private static final String ACCOUNT_NAME = "bogusAccountName.dfs.core.windows.net";
+ private static final String FS_AZURE_USER_AGENT_PREFIX = "Partner Service";
+
+ private final Pattern userAgentStringPattern;
+
+ public TestAbfsClient(){
+ StringBuilder regEx = new StringBuilder();
+ regEx.append("^");
+ regEx.append(APN_VERSION);
+ regEx.append(SINGLE_WHITE_SPACE);
+ regEx.append(CLIENT_VERSION);
+ regEx.append(SINGLE_WHITE_SPACE);
+ regEx.append("\\(");
+ regEx.append(System.getProperty(JAVA_VENDOR)
+ .replaceAll(SINGLE_WHITE_SPACE, EMPTY_STRING));
+ regEx.append(SINGLE_WHITE_SPACE);
+ regEx.append("JavaJRE");
+ regEx.append(SINGLE_WHITE_SPACE);
+ regEx.append(System.getProperty(JAVA_VERSION));
+ regEx.append(SEMICOLON);
+ regEx.append(SINGLE_WHITE_SPACE);
+ regEx.append(System.getProperty(OS_NAME)
+ .replaceAll(SINGLE_WHITE_SPACE, EMPTY_STRING));
+ regEx.append(SINGLE_WHITE_SPACE);
+ regEx.append(System.getProperty(OS_VERSION));
+ regEx.append(FORWARD_SLASH);
+ regEx.append(System.getProperty(OS_ARCH));
+ regEx.append(SEMICOLON);
+ regEx.append("([a-zA-Z].*; )?"); // Regex for sslProviderName
+ regEx.append("([a-zA-Z].*; )?"); // Regex for tokenProvider
+ regEx.append(" ?");
+ regEx.append(".+"); // cluster name
+ regEx.append(FORWARD_SLASH);
+ regEx.append(".+"); // cluster type
+ regEx.append("\\)");
+ regEx.append("( .*)?"); // Regex for user agent prefix
+ regEx.append("$");
+ this.userAgentStringPattern = Pattern.compile(regEx.toString());
+ }
- private void validateUserAgent(String expectedPattern,
- URL baseUrl,
- AbfsConfiguration config,
- boolean includeSSLProvider)
- throws AzureBlobFileSystemException {
- AbfsClient client = new AbfsClient(baseUrl, null,
+ private String getUserAgentString(AbfsConfiguration config,
+ boolean includeSSLProvider) throws MalformedURLException {
+ AbfsClient client = new AbfsClient(new URL("https://azure.com"), null,
config, null, (AccessTokenProvider) null, null);
String sslProviderName = null;
if (includeSSLProvider) {
- sslProviderName = DelegatingSSLSocketFactory.getDefaultFactory().getProviderName();
+ sslProviderName = DelegatingSSLSocketFactory.getDefaultFactory()
+ .getProviderName();
}
- String userAgent = client.initializeUserAgent(config, sslProviderName);
- Pattern pattern = Pattern.compile(expectedPattern);
- Assert.assertTrue("Incorrect User Agent String",
- pattern.matcher(userAgent).matches());
+ return client.initializeUserAgent(config, sslProviderName);
}
@Test
- public void verifyUnknownUserAgent() throws Exception {
- String clientVersion = "Azure Blob FS/" + VersionInfo.getVersion();
- String expectedUserAgentPattern = String.format(clientVersion
- + " %s", "\\(JavaJRE ([^\\)]+)\\)");
+ public void verifybBasicInfo() throws Exception {
final Configuration configuration = new Configuration();
- configuration.unset(ConfigurationKeys.FS_AZURE_USER_AGENT_PREFIX_KEY);
- AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration, accountName);
- validateUserAgent(expectedUserAgentPattern, new URL("http://azure.com"),
- abfsConfiguration, false);
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ verifybBasicInfo(getUserAgentString(abfsConfiguration, false));
+ }
+
+ private void verifybBasicInfo(String userAgentStr) {
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string [" + userAgentStr
+ + "] should be of the pattern: " + this.userAgentStringPattern.pattern())
+ .matches(this.userAgentStringPattern)
+ .describedAs("User-Agent string should contain java vendor")
+ .contains(System.getProperty(JAVA_VENDOR)
+ .replaceAll(SINGLE_WHITE_SPACE, EMPTY_STRING))
+ .describedAs("User-Agent string should contain java version")
+ .contains(System.getProperty(JAVA_VERSION))
+ .describedAs("User-Agent string should contain OS name")
+ .contains(System.getProperty(OS_NAME)
+ .replaceAll(SINGLE_WHITE_SPACE, EMPTY_STRING))
+ .describedAs("User-Agent string should contain OS version")
+ .contains(System.getProperty(OS_VERSION))
+ .describedAs("User-Agent string should contain OS arch")
+ .contains(System.getProperty(OS_ARCH));
}
@Test
- public void verifyUserAgent() throws Exception {
- String clientVersion = "Azure Blob FS/" + VersionInfo.getVersion();
- String expectedUserAgentPattern = String.format(clientVersion
- + " %s", "\\(JavaJRE ([^\\)]+)\\) Partner Service");
+ public void verifyUserAgentPrefix()
+ throws IOException, IllegalAccessException {
final Configuration configuration = new Configuration();
- configuration.set(ConfigurationKeys.FS_AZURE_USER_AGENT_PREFIX_KEY, "Partner Service");
- AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration, accountName);
- validateUserAgent(expectedUserAgentPattern, new URL("http://azure.com"),
- abfsConfiguration, false);
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ configuration.set(ConfigurationKeys.FS_AZURE_USER_AGENT_PREFIX_KEY, FS_AZURE_USER_AGENT_PREFIX);
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ String userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should contain " + FS_AZURE_USER_AGENT_PREFIX)
+ .contains(FS_AZURE_USER_AGENT_PREFIX);
+
+ configuration.unset(ConfigurationKeys.FS_AZURE_USER_AGENT_PREFIX_KEY);
+ abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should not contain " + FS_AZURE_USER_AGENT_PREFIX)
+ .doesNotContain(FS_AZURE_USER_AGENT_PREFIX);
}
@Test
- public void verifyUserAgentWithSSLProvider() throws Exception {
- String clientVersion = "Azure Blob FS/" + VersionInfo.getVersion();
- String expectedUserAgentPattern = String.format(clientVersion
- + " %s", "\\(JavaJRE ([^\\)]+)\\) Partner Service");
+ public void verifyUserAgentWithoutSSLProvider() throws Exception {
final Configuration configuration = new Configuration();
- configuration.set(ConfigurationKeys.FS_AZURE_USER_AGENT_PREFIX_KEY, "Partner Service");
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
configuration.set(ConfigurationKeys.FS_AZURE_SSL_CHANNEL_MODE_KEY,
DelegatingSSLSocketFactory.SSLChannelMode.Default_JSSE.name());
- AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration, accountName);
- validateUserAgent(expectedUserAgentPattern, new URL("https://azure.com"),
- abfsConfiguration, true);
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ String userAgentStr = getUserAgentString(abfsConfiguration, true);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should contain sslProvider")
+ .contains(DelegatingSSLSocketFactory.getDefaultFactory().getProviderName());
+
+ userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should not contain sslProvider")
+ .doesNotContain(DelegatingSSLSocketFactory.getDefaultFactory().getProviderName());
}
+
+ @Test
+ public void verifyUserAgentClusterName() throws Exception {
+ final String clusterName = "testClusterName";
+ final Configuration configuration = new Configuration();
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ configuration.set(FS_AZURE_CLUSTER_NAME, clusterName);
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ String userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should contain cluster name")
+ .contains(clusterName);
+
+ configuration.unset(FS_AZURE_CLUSTER_NAME);
+ abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should not contain cluster name")
+ .doesNotContain(clusterName)
+ .describedAs("User-Agent string should contain UNKNOWN as cluster name config is absent")
+ .contains(DEFAULT_VALUE_UNKNOWN);
+ }
+
+ @Test
+ public void verifyUserAgentClusterType() throws Exception {
+ final String clusterType = "testClusterType";
+ final Configuration configuration = new Configuration();
+ configuration.addResource(TEST_CONFIGURATION_FILE_NAME);
+ configuration.set(FS_AZURE_CLUSTER_TYPE, clusterType);
+ AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ String userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should contain cluster type")
+ .contains(clusterType);
+
+ configuration.unset(FS_AZURE_CLUSTER_TYPE);
+ abfsConfiguration = new AbfsConfiguration(configuration,
+ ACCOUNT_NAME);
+ userAgentStr = getUserAgentString(abfsConfiguration, false);
+
+ verifybBasicInfo(userAgentStr);
+ assertThat(userAgentStr)
+ .describedAs("User-Agent string should not contain cluster type")
+ .doesNotContain(clusterType)
+ .describedAs("User-Agent string should contain UNKNOWN as cluster type config is absent")
+ .contains(DEFAULT_VALUE_UNKNOWN);
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org