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 su...@apache.org on 2018/11/20 02:32:39 UTC
[1/2] hadoop git commit: YARN-8953. [CSI] CSI driver adaptor module
support in NodeManager. Contributed by Weiwei Yang.
Repository: hadoop
Updated Branches:
refs/heads/trunk 27ffec7ba -> 5fb14e063
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestCsiAdaptorService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestCsiAdaptorService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestCsiAdaptorService.java
new file mode 100644
index 0000000..128240d
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestCsiAdaptorService.java
@@ -0,0 +1,256 @@
+/**
+ * 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.yarn.csi.adaptor;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import csi.v0.Csi;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.service.ServiceStateException;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
+import org.apache.hadoop.yarn.api.impl.pb.client.CsiAdaptorProtocolPBClientImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ValidateVolumeCapabilitiesRequestPBImpl;
+import org.apache.hadoop.yarn.client.NMProxy;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.csi.client.CsiClient;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+
+import static org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.AccessMode.MULTI_NODE_MULTI_WRITER;
+import static org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.VolumeType.FILE_SYSTEM;
+
+/**
+ * UT for {@link CsiAdaptorProtocolService}.
+ */
+public class TestCsiAdaptorService {
+
+ private static File testRoot = null;
+ private static String domainSocket = null;
+
+ @BeforeClass
+ public static void setUp() throws IOException {
+ testRoot = GenericTestUtils.getTestDir("csi-test");
+ File socketPath = new File(testRoot, "csi.sock");
+ FileUtils.forceMkdirParent(socketPath);
+ domainSocket = "unix://" + socketPath.getAbsolutePath();
+ }
+
+ @AfterClass
+ public static void tearDown() throws IOException {
+ if (testRoot != null) {
+ FileUtils.deleteDirectory(testRoot);
+ }
+ }
+
+ @Test
+ public void testValidateVolume() throws IOException, YarnException {
+ ServerSocket ss = new ServerSocket(0);
+ ss.close();
+ InetSocketAddress address = new InetSocketAddress(ss.getLocalPort());
+ Configuration conf = new Configuration();
+ conf.setSocketAddr(
+ YarnConfiguration.NM_CSI_ADAPTOR_PREFIX + "test-driver.address",
+ address);
+ CsiAdaptorProtocolService service =
+ new CsiAdaptorProtocolService("test-driver", domainSocket);
+
+ // inject a fake CSI client
+ // this client validates if the ValidateVolumeCapabilitiesRequest
+ // is integrity, and then reply a fake response
+ service.setCsiClient(new CsiClient() {
+ @Override
+ public Csi.GetPluginInfoResponse getPluginInfo() {
+ return Csi.GetPluginInfoResponse.newBuilder()
+ .setName("test-plugin")
+ .setVendorVersion("0.1")
+ .build();
+ }
+
+ @Override
+ public Csi.ValidateVolumeCapabilitiesResponse validateVolumeCapabilities(
+ Csi.ValidateVolumeCapabilitiesRequest request) {
+ // validate we get all info from the request
+ Assert.assertEquals("volume-id-0000123", request.getVolumeId());
+ Assert.assertEquals(1, request.getVolumeCapabilitiesCount());
+ Assert.assertEquals(Csi.VolumeCapability.AccessMode
+ .newBuilder().setModeValue(5).build(),
+ request.getVolumeCapabilities(0).getAccessMode());
+ Assert.assertTrue(request.getVolumeCapabilities(0).hasMount());
+ Assert.assertEquals(2, request.getVolumeCapabilities(0)
+ .getMount().getMountFlagsCount());
+ Assert.assertTrue(request.getVolumeCapabilities(0)
+ .getMount().getMountFlagsList().contains("mountFlag1"));
+ Assert.assertTrue(request.getVolumeCapabilities(0)
+ .getMount().getMountFlagsList().contains("mountFlag2"));
+ Assert.assertEquals(2, request.getVolumeAttributesCount());
+ Assert.assertEquals("v1", request.getVolumeAttributesMap().get("k1"));
+ Assert.assertEquals("v2", request.getVolumeAttributesMap().get("k2"));
+ // return a fake result
+ return Csi.ValidateVolumeCapabilitiesResponse.newBuilder()
+ .setSupported(false)
+ .setMessage("this is a test")
+ .build();
+ }
+ });
+
+ service.init(conf);
+ service.start();
+
+ try (CsiAdaptorProtocolPBClientImpl client =
+ new CsiAdaptorProtocolPBClientImpl(1L, address, new Configuration())) {
+ ValidateVolumeCapabilitiesRequest request =
+ ValidateVolumeCapabilitiesRequestPBImpl
+ .newInstance("volume-id-0000123",
+ ImmutableList.of(
+ new ValidateVolumeCapabilitiesRequest
+ .VolumeCapability(
+ MULTI_NODE_MULTI_WRITER, FILE_SYSTEM,
+ ImmutableList.of("mountFlag1", "mountFlag2"))),
+ ImmutableMap.of("k1", "v1", "k2", "v2"));
+
+ ValidateVolumeCapabilitiesResponse response = client
+ .validateVolumeCapacity(request);
+
+ Assert.assertEquals(false, response.isSupported());
+ Assert.assertEquals("this is a test", response.getResponseMessage());
+ } finally {
+ service.stop();
+ }
+ }
+
+ @Test
+ public void testValidateVolumeWithNMProxy() throws Exception {
+ ServerSocket ss = new ServerSocket(0);
+ ss.close();
+ InetSocketAddress address = new InetSocketAddress(ss.getLocalPort());
+ Configuration conf = new Configuration();
+ conf.setSocketAddr(
+ YarnConfiguration.NM_CSI_ADAPTOR_PREFIX + "test-driver.address",
+ address);
+ CsiAdaptorProtocolService service =
+ new CsiAdaptorProtocolService("test-driver", domainSocket);
+
+ // inject a fake CSI client
+ // this client validates if the ValidateVolumeCapabilitiesRequest
+ // is integrity, and then reply a fake response
+ service.setCsiClient(new CsiClient() {
+ @Override
+ public Csi.GetPluginInfoResponse getPluginInfo() {
+ return Csi.GetPluginInfoResponse.newBuilder()
+ .setName("test-plugin")
+ .setVendorVersion("0.1")
+ .build();
+ }
+
+ @Override
+ public Csi.ValidateVolumeCapabilitiesResponse validateVolumeCapabilities(
+ Csi.ValidateVolumeCapabilitiesRequest request) {
+ // validate we get all info from the request
+ Assert.assertEquals("volume-id-0000123", request.getVolumeId());
+ Assert.assertEquals(1, request.getVolumeCapabilitiesCount());
+ Assert.assertEquals(Csi.VolumeCapability.AccessMode
+ .newBuilder().setModeValue(5).build(),
+ request.getVolumeCapabilities(0).getAccessMode());
+ Assert.assertTrue(request.getVolumeCapabilities(0).hasMount());
+ Assert.assertEquals(2, request.getVolumeCapabilities(0)
+ .getMount().getMountFlagsCount());
+ Assert.assertTrue(request.getVolumeCapabilities(0)
+ .getMount().getMountFlagsList().contains("mountFlag1"));
+ Assert.assertTrue(request.getVolumeCapabilities(0)
+ .getMount().getMountFlagsList().contains("mountFlag2"));
+ Assert.assertEquals(2, request.getVolumeAttributesCount());
+ Assert.assertEquals("v1", request.getVolumeAttributesMap().get("k1"));
+ Assert.assertEquals("v2", request.getVolumeAttributesMap().get("k2"));
+ // return a fake result
+ return Csi.ValidateVolumeCapabilitiesResponse.newBuilder()
+ .setSupported(false)
+ .setMessage("this is a test")
+ .build();
+ }
+ });
+
+ service.init(conf);
+ service.start();
+
+ YarnRPC rpc = YarnRPC.create(conf);
+ UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
+ CsiAdaptorProtocol adaptorClient = NMProxy
+ .createNMProxy(conf, CsiAdaptorProtocol.class, currentUser, rpc,
+ NetUtils.createSocketAddrForHost("localhost", ss.getLocalPort()));
+ ValidateVolumeCapabilitiesRequest request =
+ ValidateVolumeCapabilitiesRequestPBImpl
+ .newInstance("volume-id-0000123",
+ ImmutableList.of(new ValidateVolumeCapabilitiesRequest
+ .VolumeCapability(
+ MULTI_NODE_MULTI_WRITER, FILE_SYSTEM,
+ ImmutableList.of("mountFlag1", "mountFlag2"))),
+ ImmutableMap.of("k1", "v1", "k2", "v2"));
+
+ ValidateVolumeCapabilitiesResponse response = adaptorClient
+ .validateVolumeCapacity(request);
+ Assert.assertEquals(false, response.isSupported());
+ Assert.assertEquals("this is a test", response.getResponseMessage());
+
+ service.stop();
+ }
+
+ @Test (expected = ServiceStateException.class)
+ public void testMissingConfiguration() {
+ Configuration conf = new Configuration();
+ CsiAdaptorProtocolService service =
+ new CsiAdaptorProtocolService("test-driver", domainSocket);
+ service.init(conf);
+ }
+
+ @Test (expected = ServiceStateException.class)
+ public void testInvalidServicePort() {
+ Configuration conf = new Configuration();
+ conf.set(YarnConfiguration.NM_CSI_ADAPTOR_PREFIX
+ + "test-driver-0001.address",
+ "0.0.0.0:-100"); // this is an invalid address
+ CsiAdaptorProtocolService service =
+ new CsiAdaptorProtocolService("test-driver-0001", domainSocket);
+ service.init(conf);
+ }
+
+ @Test (expected = ServiceStateException.class)
+ public void testInvalidHost() {
+ Configuration conf = new Configuration();
+ conf.set(YarnConfiguration.NM_CSI_ADAPTOR_PREFIX
+ + "test-driver-0001.address",
+ "192.0.1:8999"); // this is an invalid ip address
+ CsiAdaptorProtocolService service =
+ new CsiAdaptorProtocolService("test-driver-0001", domainSocket);
+ service.init(conf);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestGetPluginInfoRequestResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestGetPluginInfoRequestResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestGetPluginInfoRequestResponse.java
new file mode 100644
index 0000000..f1734c8
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestGetPluginInfoRequestResponse.java
@@ -0,0 +1,66 @@
+/**
+ * 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.yarn.csi.adaptor;
+
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetPluginInfoRequestPBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetPluginInfoResponsePBImpl;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Verify the integrity of GetPluginInfoRequest and GetPluginInfoResponse.
+ */
+public class TestGetPluginInfoRequestResponse {
+
+ @Test
+ public void testGetPluginInfoRequestPBRecord() {
+ CsiAdaptorProtos.GetPluginInfoRequest requestProto =
+ CsiAdaptorProtos.GetPluginInfoRequest.newBuilder().build();
+ GetPluginInfoRequestPBImpl pbImpl =
+ new GetPluginInfoRequestPBImpl(requestProto);
+ Assert.assertNotNull(pbImpl);
+ Assert.assertEquals(requestProto, pbImpl.getProto());
+ }
+
+ @Test
+ public void testGetPluginInfoResponsePBRecord() {
+ CsiAdaptorProtos.GetPluginInfoResponse responseProto =
+ CsiAdaptorProtos.GetPluginInfoResponse.newBuilder()
+ .setName("test-driver")
+ .setVendorVersion("1.0.1")
+ .build();
+
+ GetPluginInfoResponsePBImpl pbImpl =
+ new GetPluginInfoResponsePBImpl(responseProto);
+ Assert.assertEquals("test-driver", pbImpl.getDriverName());
+ Assert.assertEquals("1.0.1", pbImpl.getVersion());
+ Assert.assertEquals(responseProto, pbImpl.getProto());
+
+ GetPluginInfoResponse pbImpl2 = GetPluginInfoResponsePBImpl
+ .newInstance("test-driver", "1.0.1");
+ Assert.assertEquals("test-driver", pbImpl2.getDriverName());
+ Assert.assertEquals("1.0.1", pbImpl2.getVersion());
+
+ CsiAdaptorProtos.GetPluginInfoResponse proto =
+ ((GetPluginInfoResponsePBImpl) pbImpl2).getProto();
+ Assert.assertEquals("test-driver", proto.getName());
+ Assert.assertEquals("1.0.1", proto.getVendorVersion());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityRequest.java
new file mode 100644
index 0000000..303cfc4
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityRequest.java
@@ -0,0 +1,113 @@
+/**
+ * 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.yarn.csi.adaptor;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.VolumeCapability;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ValidateVolumeCapabilitiesRequestPBImpl;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos.VolumeCapability.AccessMode;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos.VolumeCapability.VolumeType;
+import org.apache.hadoop.yarn.proto.YarnProtos;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.AccessMode.MULTI_NODE_MULTI_WRITER;
+import static org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.VolumeType.FILE_SYSTEM;
+
+/**
+ * UT for message exchanges.
+ */
+public class TestValidateVolumeCapabilityRequest {
+
+ @Test
+ public void testPBRecord() {
+ CsiAdaptorProtos.VolumeCapability vcProto =
+ CsiAdaptorProtos.VolumeCapability.newBuilder()
+ .setAccessMode(AccessMode.MULTI_NODE_MULTI_WRITER)
+ .setVolumeType(VolumeType.FILE_SYSTEM)
+ .addMountFlags("flag0")
+ .addMountFlags("flag1")
+ .build();
+
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest requestProto =
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest.newBuilder()
+ .setVolumeId("volume-id-0000001")
+ .addVolumeCapabilities(vcProto)
+ .addVolumeAttributes(YarnProtos.StringStringMapProto
+ .newBuilder().setKey("attr0")
+ .setValue("value0")
+ .build())
+ .addVolumeAttributes(YarnProtos.StringStringMapProto
+ .newBuilder().setKey("attr1")
+ .setValue("value1")
+ .build())
+ .build();
+
+ ValidateVolumeCapabilitiesRequestPBImpl request =
+ new ValidateVolumeCapabilitiesRequestPBImpl(requestProto);
+
+ Assert.assertEquals("volume-id-0000001", request.getVolumeId());
+ Assert.assertEquals(2, request.getVolumeAttributes().size());
+ Assert.assertEquals("value0", request.getVolumeAttributes().get("attr0"));
+ Assert.assertEquals("value1", request.getVolumeAttributes().get("attr1"));
+ Assert.assertEquals(1, request.getVolumeCapabilities().size());
+ VolumeCapability vc =
+ request.getVolumeCapabilities().get(0);
+ Assert.assertEquals(MULTI_NODE_MULTI_WRITER, vc.getAccessMode());
+ Assert.assertEquals(FILE_SYSTEM, vc.getVolumeType());
+ Assert.assertEquals(2, vc.getMountFlags().size());
+
+ Assert.assertEquals(requestProto, request.getProto());
+ }
+
+ @Test
+ public void testNewInstance() {
+ ValidateVolumeCapabilitiesRequest pbImpl =
+ ValidateVolumeCapabilitiesRequestPBImpl
+ .newInstance("volume-id-0000123",
+ ImmutableList.of(
+ new VolumeCapability(
+ MULTI_NODE_MULTI_WRITER, FILE_SYSTEM,
+ ImmutableList.of("mountFlag1", "mountFlag2"))),
+ ImmutableMap.of("k1", "v1", "k2", "v2"));
+
+ Assert.assertEquals("volume-id-0000123", pbImpl.getVolumeId());
+ Assert.assertEquals(1, pbImpl.getVolumeCapabilities().size());
+ Assert.assertEquals(FILE_SYSTEM,
+ pbImpl.getVolumeCapabilities().get(0).getVolumeType());
+ Assert.assertEquals(MULTI_NODE_MULTI_WRITER,
+ pbImpl.getVolumeCapabilities().get(0).getAccessMode());
+ Assert.assertEquals(2, pbImpl.getVolumeAttributes().size());
+
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest proto =
+ ((ValidateVolumeCapabilitiesRequestPBImpl) pbImpl).getProto();
+ Assert.assertEquals("volume-id-0000123", proto.getVolumeId());
+ Assert.assertEquals(1, proto.getVolumeCapabilitiesCount());
+ Assert.assertEquals(AccessMode.MULTI_NODE_MULTI_WRITER,
+ proto.getVolumeCapabilities(0).getAccessMode());
+ Assert.assertEquals(VolumeType.FILE_SYSTEM,
+ proto.getVolumeCapabilities(0).getVolumeType());
+ Assert.assertEquals(2, proto.getVolumeCapabilities(0)
+ .getMountFlagsCount());
+ Assert.assertEquals(2, proto.getVolumeCapabilities(0)
+ .getMountFlagsList().size());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityResponse.java
new file mode 100644
index 0000000..97f116a
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/TestValidateVolumeCapabilityResponse.java
@@ -0,0 +1,61 @@
+/**
+ * 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.yarn.csi.adaptor;
+
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ValidateVolumeCapabilitiesResponsePBImpl;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * UT for message exchanges.
+ */
+public class TestValidateVolumeCapabilityResponse {
+
+ @Test
+ public void testPBRecord() {
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse proto =
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse.newBuilder()
+ .setSupported(true)
+ .setMessage("capability is supported")
+ .build();
+
+ ValidateVolumeCapabilitiesResponsePBImpl pbImpl =
+ new ValidateVolumeCapabilitiesResponsePBImpl(proto);
+
+ Assert.assertEquals(true, pbImpl.isSupported());
+ Assert.assertEquals("capability is supported", pbImpl.getResponseMessage());
+ Assert.assertEquals(proto, pbImpl.getProto());
+ }
+
+ @Test
+ public void testNewInstance() {
+ ValidateVolumeCapabilitiesResponse pbImpl =
+ ValidateVolumeCapabilitiesResponsePBImpl
+ .newInstance(false, "capability not supported");
+ Assert.assertEquals(false, pbImpl.isSupported());
+ Assert.assertEquals("capability not supported",
+ pbImpl.getResponseMessage());
+
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse proto =
+ ((ValidateVolumeCapabilitiesResponsePBImpl) pbImpl).getProto();
+ Assert.assertEquals(false, proto.getSupported());
+ Assert.assertEquals("capability not supported", proto.getMessage());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java
new file mode 100644
index 0000000..ecc7fc2
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/test/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * This package contains UT classes for CSI adaptor.
+ */
+package org.apache.hadoop.yarn.csi.adaptor;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/volume/csi/CsiAdaptorClientProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/volume/csi/CsiAdaptorClientProtocol.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/volume/csi/CsiAdaptorClientProtocol.java
deleted file mode 100644
index b894d4e..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/volume/csi/CsiAdaptorClientProtocol.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/**
- * 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.yarn.server.volume.csi;
-
-import org.apache.hadoop.classification.InterfaceAudience.Private;
-import org.apache.hadoop.classification.InterfaceStability.Unstable;
-import org.apache.hadoop.yarn.server.volume.csi.exception.VolumeException;
-
-/**
- * Protocol for the CSI adaptor.
- */
-@Private
-@Unstable
-public interface CsiAdaptorClientProtocol {
-
- void validateVolume() throws VolumeException;
-
- void controllerPublishVolume() throws VolumeException;
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/CsiAdaptorClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/CsiAdaptorClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/CsiAdaptorClient.java
deleted file mode 100644
index 043e7ae..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/CsiAdaptorClient.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/**
- * 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.yarn.server.resourcemanager.volume.csi;
-
-import org.apache.hadoop.yarn.server.volume.csi.exception.VolumeException;
-import org.apache.hadoop.yarn.server.volume.csi.CsiAdaptorClientProtocol;
-
-/**
- * Client talks to CSI adaptor.
- */
-public class CsiAdaptorClient implements CsiAdaptorClientProtocol {
-
- @Override
- public void validateVolume() throws VolumeException {
- // TODO
- }
-
- @Override public void controllerPublishVolume() throws VolumeException {
- // TODO
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManager.java
index 5f2669d..32563cb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManager.java
@@ -17,13 +17,12 @@
*/
package org.apache.hadoop.yarn.server.resourcemanager.volume.csi;
-import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle.Volume;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.provisioner.VolumeProvisioningResults;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.provisioner.VolumeProvisioningTask;
-import org.apache.hadoop.yarn.server.volume.csi.CsiAdaptorClientProtocol;
import java.util.concurrent.ScheduledFuture;
@@ -40,12 +39,8 @@ public interface VolumeManager {
/**
* @return all known volumes and their states.
*/
- @VisibleForTesting
VolumeStates getVolumeStates();
- @VisibleForTesting
- void setClient(CsiAdaptorClientProtocol client);
-
/**
* Start to supervise on a volume.
* @param volume
@@ -60,4 +55,20 @@ public interface VolumeManager {
*/
ScheduledFuture<VolumeProvisioningResults> schedule(
VolumeProvisioningTask volumeProvisioningTask, int delaySecond);
+
+ /**
+ * Register a csi-driver-adaptor to the volume manager.
+ * @param driverName
+ * @param client
+ */
+ void registerCsiDriverAdaptor(String driverName, CsiAdaptorProtocol client);
+
+ /**
+ * Returns the csi-driver-adaptor client from cache by the given driver name.
+ * If the client is not found, null is returned.
+ * @param driverName
+ * @return a csi-driver-adaptor client working for given driver or null
+ * if the adaptor could not be found.
+ */
+ CsiAdaptorProtocol getAdaptorByDriverName(String driverName);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManagerImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManagerImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManagerImpl.java
index 5252f53..839d1bc 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManagerImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/VolumeManagerImpl.java
@@ -18,16 +18,28 @@
package org.apache.hadoop.yarn.server.resourcemanager.volume.csi;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.service.AbstractService;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.client.NMProxy;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle.Volume;
-import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle.VolumeImpl;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.provisioner.VolumeProvisioningResults;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.provisioner.VolumeProvisioningTask;
-import org.apache.hadoop.yarn.server.volume.csi.CsiAdaptorClientProtocol;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -43,20 +55,84 @@ public class VolumeManagerImpl extends AbstractService
private final VolumeStates volumeStates;
private ScheduledExecutorService provisioningExecutor;
- private CsiAdaptorClientProtocol adaptorClient;
+ private Map<String, CsiAdaptorProtocol> csiAdaptorMap;
private final static int PROVISIONING_TASK_THREAD_POOL_SIZE = 10;
public VolumeManagerImpl() {
super(VolumeManagerImpl.class.getName());
this.volumeStates = new VolumeStates();
+ this.csiAdaptorMap = new ConcurrentHashMap<>();
this.provisioningExecutor = Executors
.newScheduledThreadPool(PROVISIONING_TASK_THREAD_POOL_SIZE);
- this.adaptorClient = new CsiAdaptorClient();
+ }
+
+ // Init the CSI adaptor cache according to the configuration.
+ // user only needs to configure a list of adaptor addresses,
+ // this method extracts each address and init an adaptor client,
+ // then proceed with a hand-shake by calling adaptor's getPluginInfo
+ // method to retrieve the driver info. If the driver can be resolved,
+ // it is then added to the cache. Note, we don't allow two drivers
+ // specified with same driver-name even version is different.
+ private void initCsiAdaptorCache(
+ final Map<String, CsiAdaptorProtocol> adaptorMap, Configuration conf)
+ throws IOException, YarnException {
+ LOG.info("Initializing cache for csi-driver-adaptors");
+ String[] addresses =
+ conf.getStrings(YarnConfiguration.NM_CSI_ADAPTOR_ADDRESSES);
+ if (addresses != null && addresses.length > 0) {
+ for (String addr : addresses) {
+ LOG.info("Found csi-driver-adaptor socket address: " + addr);
+ InetSocketAddress address = NetUtils.createSocketAddr(addr);
+ YarnRPC rpc = YarnRPC.create(conf);
+ UserGroupInformation currentUser =
+ UserGroupInformation.getCurrentUser();
+ CsiAdaptorProtocol adaptorClient = NMProxy
+ .createNMProxy(conf, CsiAdaptorProtocol.class, currentUser, rpc,
+ address);
+ // Attempt to resolve the driver by contacting to
+ // the diver's identity service on the given address.
+ // If the call failed, the initialization is also failed
+ // in order running into inconsistent state.
+ LOG.info("Retrieving info from csi-driver-adaptor on address " + addr);
+ GetPluginInfoResponse response =
+ adaptorClient.getPluginInfo(GetPluginInfoRequest.newInstance());
+ if (!Strings.isNullOrEmpty(response.getDriverName())) {
+ String driverName = response.getDriverName();
+ if (adaptorMap.containsKey(driverName)) {
+ throw new YarnException(
+ "Duplicate driver adaptor found," + " driver name: "
+ + driverName);
+ }
+ adaptorMap.put(driverName, adaptorClient);
+ LOG.info("CSI Adaptor added to the cache, adaptor name: " + driverName
+ + ", driver version: " + response.getVersion());
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns a CsiAdaptorProtocol client by the given driver name,
+ * returns null if no adaptor is found for the driver, that means
+ * the driver has not registered to the volume manager yet enhance not valid.
+ * @param driverName the name of the driver
+ * @return CsiAdaptorProtocol client or null if driver not registered
+ */
+ public CsiAdaptorProtocol getAdaptorByDriverName(String driverName) {
+ return csiAdaptorMap.get(driverName);
+ }
+
+ @VisibleForTesting
+ @Override
+ public void registerCsiDriverAdaptor(String driverName,
+ CsiAdaptorProtocol client) {
+ this.csiAdaptorMap.put(driverName, client);
}
@Override
protected void serviceInit(Configuration conf) throws Exception {
+ initCsiAdaptorCache(csiAdaptorMap, conf);
super.serviceInit(conf);
}
@@ -82,18 +158,11 @@ public class VolumeManagerImpl extends AbstractService
// volume already exists
return volumeStates.getVolume(volume.getVolumeId());
} else {
- // add the volume and set the client
- ((VolumeImpl) volume).setClient(adaptorClient);
this.volumeStates.addVolumeIfAbsent(volume);
return volume;
}
}
- @VisibleForTesting
- public void setClient(CsiAdaptorClientProtocol client) {
- this.adaptorClient = client;
- }
-
@Override
public ScheduledFuture<VolumeProvisioningResults> schedule(
VolumeProvisioningTask volumeProvisioningTask,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/Volume.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/Volume.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/Volume.java
index 68e89b0..83501ac 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/Volume.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/Volume.java
@@ -19,9 +19,11 @@ package org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.event.VolumeEvent;
import org.apache.hadoop.yarn.server.volume.csi.VolumeId;
+import org.apache.hadoop.yarn.server.volume.csi.VolumeMetaData;
/**
* Major volume interface at RM's view, it maintains the volume states and
@@ -34,4 +36,10 @@ public interface Volume extends EventHandler<VolumeEvent> {
VolumeState getVolumeState();
VolumeId getVolumeId();
+
+ VolumeMetaData getVolumeMeta();
+
+ CsiAdaptorProtocol getClient();
+
+ void setClient(CsiAdaptorProtocol client);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/VolumeImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/VolumeImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/VolumeImpl.java
index 2515047..82a4acb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/VolumeImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/lifecycle/VolumeImpl.java
@@ -18,10 +18,15 @@
package org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.CsiAdaptorClient;
-import org.apache.hadoop.yarn.server.volume.csi.CsiAdaptorClientProtocol;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.VolumeCapability;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.event.VolumeEvent;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.event.VolumeEventType;
import org.apache.hadoop.yarn.state.InvalidStateTransitionException;
@@ -30,13 +35,16 @@ import org.apache.hadoop.yarn.state.StateMachine;
import org.apache.hadoop.yarn.state.StateMachineFactory;
import org.apache.hadoop.yarn.server.volume.csi.VolumeId;
import org.apache.hadoop.yarn.server.volume.csi.VolumeMetaData;
-import org.apache.hadoop.yarn.server.volume.csi.exception.VolumeException;
+import java.io.IOException;
import java.util.EnumSet;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import static org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.AccessMode.SINGLE_NODE_READER_ONLY;
+import static org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.VolumeType.FILE_SYSTEM;
+
/**
* This class maintains the volume states and state transition
* according to the CSI volume lifecycle. Volume states are stored in
@@ -54,7 +62,7 @@ public class VolumeImpl implements Volume {
private final VolumeId volumeId;
private final VolumeMetaData volumeMeta;
- private CsiAdaptorClientProtocol client;
+ private CsiAdaptorProtocol adaptorClient;
public VolumeImpl(VolumeMetaData volumeMeta) {
ReadWriteLock lock = new ReentrantReadWriteLock();
@@ -63,16 +71,21 @@ public class VolumeImpl implements Volume {
this.volumeId = volumeMeta.getVolumeId();
this.volumeMeta = volumeMeta;
this.stateMachine = createVolumeStateFactory().make(this);
- this.client = new CsiAdaptorClient();
}
@VisibleForTesting
- public void setClient(CsiAdaptorClientProtocol client) {
- this.client = client;
+ public void setClient(CsiAdaptorProtocol csiAdaptorClient) {
+ this.adaptorClient = csiAdaptorClient;
}
- public CsiAdaptorClientProtocol getClient() {
- return this.client;
+ @Override
+ public CsiAdaptorProtocol getClient() {
+ return this.adaptorClient;
+ }
+
+ @Override
+ public VolumeMetaData getVolumeMeta() {
+ return this.volumeMeta;
}
private StateMachineFactory<VolumeImpl, VolumeState,
@@ -135,9 +148,20 @@ public class VolumeImpl implements Volume {
VolumeEvent volumeEvent) {
try {
// this call could cross node, we should keep the message tight
- volume.getClient().validateVolume();
- return VolumeState.VALIDATED;
- } catch (VolumeException e) {
+ // TODO we should parse the capability from volume resource spec
+ VolumeCapability capability = new VolumeCapability(
+ SINGLE_NODE_READER_ONLY, FILE_SYSTEM,
+ ImmutableList.of());
+ ValidateVolumeCapabilitiesRequest request =
+ ValidateVolumeCapabilitiesRequest
+ .newInstance(volume.getVolumeId().getId(),
+ ImmutableList.of(capability),
+ ImmutableMap.of());
+ ValidateVolumeCapabilitiesResponse response = volume.getClient()
+ .validateVolumeCapacity(request);
+ return response.isSupported() ? VolumeState.VALIDATED
+ : VolumeState.UNAVAILABLE;
+ } catch (YarnException | IOException e) {
LOG.warn("Got exception while calling the CSI adaptor", e);
return VolumeState.UNAVAILABLE;
}
@@ -150,14 +174,8 @@ public class VolumeImpl implements Volume {
@Override
public VolumeState transition(VolumeImpl volume,
VolumeEvent volumeEvent) {
- try {
- // this call could cross node, we should keep the message tight
- volume.getClient().controllerPublishVolume();
- return VolumeState.NODE_READY;
- } catch (VolumeException e) {
- LOG.warn("Got exception while calling the CSI adaptor", e);
- return volume.getVolumeState();
- }
+ // this call could cross node, we should keep the message tight
+ return VolumeState.NODE_READY;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/processor/VolumeAMSProcessor.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/processor/VolumeAMSProcessor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/processor/VolumeAMSProcessor.java
index f275768..814634a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/processor/VolumeAMSProcessor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/processor/VolumeAMSProcessor.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.yarn.server.resourcemanager.volume.csi.processor;
import org.apache.hadoop.yarn.ams.ApplicationMasterServiceContext;
import org.apache.hadoop.yarn.ams.ApplicationMasterServiceProcessor;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
@@ -29,6 +30,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceInformation;
import org.apache.hadoop.yarn.api.records.SchedulingRequest;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.VolumeManager;
@@ -142,8 +144,21 @@ public class VolumeAMSProcessor implements ApplicationMasterServiceProcessor {
* @param metaData
* @return volume
*/
- private Volume checkAndGetVolume(VolumeMetaData metaData) {
+ private Volume checkAndGetVolume(VolumeMetaData metaData)
+ throws InvalidVolumeException {
Volume toAdd = new VolumeImpl(metaData);
+ CsiAdaptorProtocol adaptor = volumeManager
+ .getAdaptorByDriverName(metaData.getDriverName());
+ if (adaptor == null) {
+ throw new InvalidVolumeException("It seems for the driver name"
+ + " specified in the volume " + metaData.getDriverName()
+ + " ,there is no matched driver-adaptor can be found. "
+ + "Is the driver probably registered? Please check if"
+ + " adaptors service addresses defined in "
+ + YarnConfiguration.NM_CSI_ADAPTOR_ADDRESSES
+ + " are correct and services are started.");
+ }
+ toAdd.setClient(adaptor);
return this.volumeManager.addOrGetVolume(toAdd);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeLifecycle.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeLifecycle.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeLifecycle.java
index 18c23e8..a94f508 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeLifecycle.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeLifecycle.java
@@ -18,9 +18,11 @@
package org.apache.hadoop.yarn.server.resourcemanager.volume.csi;
import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.event.ControllerPublishVolumeEvent;
-import org.apache.hadoop.yarn.server.volume.csi.CsiAdaptorClientProtocol;
-import org.apache.hadoop.yarn.server.volume.csi.exception.InvalidVolumeException;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.event.ValidateVolumeEvent;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle.VolumeImpl;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle.VolumeState;
@@ -29,8 +31,8 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
+import java.io.IOException;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.mockito.Mockito.*;
@@ -40,7 +42,13 @@ import static org.mockito.Mockito.*;
public class TestVolumeLifecycle {
@Test
- public void testValidation() throws InvalidVolumeException {
+ public void testValidation() throws YarnException, IOException {
+ CsiAdaptorProtocol mockedClient = Mockito
+ .mock(CsiAdaptorProtocol.class);
+ doReturn(ValidateVolumeCapabilitiesResponse.newInstance(true, ""))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
+
VolumeImpl volume = (VolumeImpl) VolumeBuilder.newBuilder()
.volumeId("test_vol_00000001")
.maxCapability(5L)
@@ -48,6 +56,7 @@ public class TestVolumeLifecycle {
.mountPoint("/path/to/mount")
.driverName("test-driver-name")
.build();
+ volume.setClient(mockedClient);
Assert.assertEquals(VolumeState.NEW, volume.getVolumeState());
volume.handle(new ValidateVolumeEvent(volume));
@@ -55,16 +64,19 @@ public class TestVolumeLifecycle {
}
@Test
- public void testValidationFailure() throws VolumeException {
+ public void testVolumeCapacityNotSupported() throws Exception {
+ CsiAdaptorProtocol mockedClient = Mockito
+ .mock(CsiAdaptorProtocol.class);
+
VolumeImpl volume = (VolumeImpl) VolumeBuilder
.newBuilder().volumeId("test_vol_00000001").build();
- CsiAdaptorClientProtocol mockedClient = Mockito
- .mock(CsiAdaptorClientProtocol.class);
volume.setClient(mockedClient);
// NEW -> UNAVAILABLE
// Simulate a failed API call to the adaptor
- doThrow(new VolumeException("failed")).when(mockedClient).validateVolume();
+ doReturn(ValidateVolumeCapabilitiesResponse.newInstance(false, ""))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
volume.handle(new ValidateVolumeEvent(volume));
try {
@@ -80,47 +92,62 @@ public class TestVolumeLifecycle {
}
@Test
- public void testValidated() throws InvalidVolumeException {
- AtomicInteger validatedTimes = new AtomicInteger(0);
+ public void testValidationFailure() throws YarnException, IOException {
+ CsiAdaptorProtocol mockedClient = Mockito
+ .mock(CsiAdaptorProtocol.class);
+ doThrow(new VolumeException("fail"))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
+
+ VolumeImpl volume = (VolumeImpl) VolumeBuilder
+ .newBuilder().volumeId("test_vol_00000001").build();
+ volume.setClient(mockedClient);
+
+ // NEW -> UNAVAILABLE
+ // Simulate a failed API call to the adaptor
+ doThrow(new VolumeException("failed"))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
+ volume.handle(new ValidateVolumeEvent(volume));
+ }
+
+ @Test
+ public void testValidated() throws YarnException, IOException {
VolumeImpl volume = (VolumeImpl) VolumeBuilder
.newBuilder().volumeId("test_vol_00000001").build();
- CsiAdaptorClientProtocol mockedClient = new CsiAdaptorClientProtocol() {
- @Override
- public void validateVolume() {
- validatedTimes.incrementAndGet();
- }
-
- @Override
- public void controllerPublishVolume() {
- // do nothing
- }
- };
+ CsiAdaptorProtocol mockedClient = Mockito.mock(CsiAdaptorProtocol.class);
// The client has a count to memorize how many times being called
volume.setClient(mockedClient);
// NEW -> VALIDATED
+ doReturn(ValidateVolumeCapabilitiesResponse.newInstance(true, ""))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
Assert.assertEquals(VolumeState.NEW, volume.getVolumeState());
volume.handle(new ValidateVolumeEvent(volume));
Assert.assertEquals(VolumeState.VALIDATED, volume.getVolumeState());
- Assert.assertEquals(1, validatedTimes.get());
+ verify(mockedClient, times(1))
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
// VALIDATED -> VALIDATED
volume.handle(new ValidateVolumeEvent(volume));
Assert.assertEquals(VolumeState.VALIDATED, volume.getVolumeState());
- Assert.assertEquals(1, validatedTimes.get());
+ verify(mockedClient, times(1))
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
}
@Test
- public void testUnavailableState() throws VolumeException {
+ public void testUnavailableState() throws YarnException, IOException {
VolumeImpl volume = (VolumeImpl) VolumeBuilder
.newBuilder().volumeId("test_vol_00000001").build();
- CsiAdaptorClientProtocol mockedClient = Mockito
- .mock(CsiAdaptorClientProtocol.class);
+ CsiAdaptorProtocol mockedClient = Mockito
+ .mock(CsiAdaptorProtocol.class);
volume.setClient(mockedClient);
// NEW -> UNAVAILABLE
- doThrow(new VolumeException("failed")).when(mockedClient)
- .validateVolume();
+ doThrow(new VolumeException("failed"))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
Assert.assertEquals(VolumeState.NEW, volume.getVolumeState());
volume.handle(new ValidateVolumeEvent(volume));
Assert.assertEquals(VolumeState.UNAVAILABLE, volume.getVolumeState());
@@ -130,23 +157,26 @@ public class TestVolumeLifecycle {
Assert.assertEquals(VolumeState.UNAVAILABLE, volume.getVolumeState());
// UNAVAILABLE -> VALIDATED
- doNothing().when(mockedClient).validateVolume();
+ doReturn(ValidateVolumeCapabilitiesResponse.newInstance(true, ""))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
volume.setClient(mockedClient);
volume.handle(new ValidateVolumeEvent(volume));
Assert.assertEquals(VolumeState.VALIDATED, volume.getVolumeState());
}
@Test
- public void testPublishUnavailableVolume() throws VolumeException {
+ public void testPublishUnavailableVolume() throws YarnException, IOException {
VolumeImpl volume = (VolumeImpl) VolumeBuilder
.newBuilder().volumeId("test_vol_00000001").build();
- CsiAdaptorClientProtocol mockedClient = Mockito
- .mock(CsiAdaptorClientProtocol.class);
+ CsiAdaptorProtocol mockedClient = Mockito
+ .mock(CsiAdaptorProtocol.class);
volume.setClient(mockedClient);
// NEW -> UNAVAILABLE (on validateVolume)
- doThrow(new VolumeException("failed")).when(mockedClient)
- .validateVolume();
+ doThrow(new VolumeException("failed"))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
Assert.assertEquals(VolumeState.NEW, volume.getVolumeState());
volume.handle(new ValidateVolumeEvent(volume));
Assert.assertEquals(VolumeState.UNAVAILABLE, volume.getVolumeState());
@@ -154,7 +184,7 @@ public class TestVolumeLifecycle {
// UNAVAILABLE -> UNAVAILABLE (on publishVolume)
volume.handle(new ControllerPublishVolumeEvent(volume));
// controller publish is not called since the state is UNAVAILABLE
- verify(mockedClient, times(0)).controllerPublishVolume();
+ // verify(mockedClient, times(0)).controllerPublishVolume();
// state remains to UNAVAILABLE
Assert.assertEquals(VolumeState.UNAVAILABLE, volume.getVolumeState());
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeProcessor.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeProcessor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeProcessor.java
index d6f9d92..cee8fdf 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeProcessor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/volume/csi/TestVolumeProcessor.java
@@ -20,8 +20,11 @@ package org.apache.hadoop.yarn.server.resourcemanager.volume.csi;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
import org.apache.hadoop.yarn.api.protocolrecords.ResourceTypes;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceInformation;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
@@ -40,7 +43,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.Capacity
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle.Volume;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.lifecycle.VolumeState;
import org.apache.hadoop.yarn.server.resourcemanager.volume.csi.processor.VolumeAMSProcessor;
-import org.apache.hadoop.yarn.server.volume.csi.CsiAdaptorClientProtocol;
import org.apache.hadoop.yarn.server.volume.csi.CsiConstants;
import org.apache.hadoop.yarn.server.volume.csi.VolumeId;
import org.apache.hadoop.yarn.server.volume.csi.exception.InvalidVolumeException;
@@ -57,6 +59,10 @@ import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+
/**
* Test cases for volume processor.
*/
@@ -91,6 +97,7 @@ public class TestVolumeProcessor {
conf.set(CapacitySchedulerConfiguration.PREFIX
+ CapacitySchedulerConfiguration.ROOT + ".default.ordering-policy",
"fair");
+ // this is required to enable volume processor
conf.set(YarnConfiguration.RM_APPLICATION_MASTER_SERVICE_PROCESSORS,
VolumeAMSProcessor.class.getName());
mgr = new NullRMNodeLabelsManager();
@@ -155,6 +162,17 @@ public class TestVolumeProcessor {
.schedulingRequests(Arrays.asList(sc))
.build();
+ // inject adaptor client for testing
+ CsiAdaptorProtocol mockedClient = Mockito
+ .mock(CsiAdaptorProtocol.class);
+ rm.getRMContext().getVolumeManager()
+ .registerCsiDriverAdaptor("hostpath", mockedClient);
+
+ // simulate validation succeed
+ doReturn(ValidateVolumeCapabilitiesResponse.newInstance(true, ""))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
+
am1.allocate(ar);
VolumeStates volumeStates =
rm.getRMContext().getVolumeManager().getVolumeStates();
@@ -212,12 +230,14 @@ public class TestVolumeProcessor {
RMApp app1 = rm.submitApp(1 * GB, "app", "user", null, "default");
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, mockNMS[0]);
- CsiAdaptorClientProtocol mockedClient = Mockito
- .mock(CsiAdaptorClientProtocol.class);
+ CsiAdaptorProtocol mockedClient = Mockito
+ .mock(CsiAdaptorProtocol.class);
// inject adaptor client
- rm.getRMContext().getVolumeManager().setClient(mockedClient);
- Mockito.doThrow(new VolumeException("failed")).when(mockedClient)
- .validateVolume();
+ rm.getRMContext().getVolumeManager()
+ .registerCsiDriverAdaptor("hostpath", mockedClient);
+ doThrow(new VolumeException("failed"))
+ .when(mockedClient)
+ .validateVolumeCapacity(any(ValidateVolumeCapabilitiesRequest.class));
Resource resource = Resource.newInstance(1024, 1);
ResourceInformation volumeResource = ResourceInformation
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[2/2] hadoop git commit: YARN-8953. [CSI] CSI driver adaptor module
support in NodeManager. Contributed by Weiwei Yang.
Posted by su...@apache.org.
YARN-8953. [CSI] CSI driver adaptor module support in NodeManager. Contributed by Weiwei Yang.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/5fb14e06
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/5fb14e06
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/5fb14e06
Branch: refs/heads/trunk
Commit: 5fb14e0635f850c63e009f954b8539a401d0f90a
Parents: 27ffec7
Author: Sunil G <su...@apache.org>
Authored: Tue Nov 20 08:02:11 2018 +0530
Committer: Sunil G <su...@apache.org>
Committed: Tue Nov 20 08:02:11 2018 +0530
----------------------------------------------------------------------
.../hadoop-yarn/hadoop-yarn-api/pom.xml | 2 +
.../apache/hadoop/yarn/api/CsiAdaptorPB.java | 31 +++
.../hadoop/yarn/api/CsiAdaptorProtocol.java | 39 +++
.../protocolrecords/GetPluginInfoRequest.java | 30 +++
.../protocolrecords/GetPluginInfoResponse.java | 43 ++++
.../ValidateVolumeCapabilitiesRequest.java | 117 +++++++++
.../ValidateVolumeCapabilitiesResponse.java | 46 ++++
.../hadoop/yarn/conf/YarnConfiguration.java | 12 +
.../src/main/proto/YarnCsiAdaptor.proto | 34 +++
.../src/main/proto/yarn_csi_adaptor.proto | 69 +++++
.../client/CsiAdaptorProtocolPBClientImpl.java | 91 +++++++
.../CsiAdaptorProtocolPBServiceImpl.java | 75 ++++++
.../impl/pb/GetPluginInfoRequestPBImpl.java | 60 +++++
.../impl/pb/GetPluginInfoResponsePBImpl.java | 84 ++++++
...ValidateVolumeCapabilitiesRequestPBImpl.java | 121 +++++++++
...alidateVolumeCapabilitiesResponsePBImpl.java | 87 +++++++
.../src/main/resources/yarn-default.xml | 13 +
.../hadoop-yarn/hadoop-yarn-csi/pom.xml | 12 +
.../csi/adaptor/CsiAdaptorProtocolService.java | 122 +++++++++
.../hadoop/yarn/csi/adaptor/package-info.java | 21 ++
.../hadoop/yarn/csi/client/CsiClient.java | 4 +
.../hadoop/yarn/csi/client/CsiClientImpl.java | 11 +
.../GetPluginInfoResponseProtoTranslator.java | 44 ++++
.../yarn/csi/translator/ProtoTranslator.java | 49 ++++
.../csi/translator/ProtoTranslatorFactory.java | 66 +++++
...olumeCapabilitiesRequestProtoTranslator.java | 93 +++++++
...lumeCapabilitiesResponseProtoTranslator.java | 48 ++++
.../yarn/csi/translator/package-info.java | 21 ++
.../hadoop/yarn/csi/utils/ConfigUtils.java | 61 +++++
.../yarn/csi/adaptor/TestCsiAdaptorService.java | 256 +++++++++++++++++++
.../TestGetPluginInfoRequestResponse.java | 66 +++++
.../TestValidateVolumeCapabilityRequest.java | 113 ++++++++
.../TestValidateVolumeCapabilityResponse.java | 61 +++++
.../hadoop/yarn/csi/adaptor/package-info.java | 21 ++
.../volume/csi/CsiAdaptorClientProtocol.java | 34 ---
.../volume/csi/CsiAdaptorClient.java | 36 ---
.../volume/csi/VolumeManager.java | 23 +-
.../volume/csi/VolumeManagerImpl.java | 91 ++++++-
.../volume/csi/lifecycle/Volume.java | 8 +
.../volume/csi/lifecycle/VolumeImpl.java | 58 +++--
.../csi/processor/VolumeAMSProcessor.java | 17 +-
.../volume/csi/TestVolumeLifecycle.java | 100 +++++---
.../volume/csi/TestVolumeProcessor.java | 32 ++-
43 files changed, 2273 insertions(+), 149 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
index 4f11ad6..832e98c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
@@ -124,6 +124,8 @@
<include>server/application_history_server.proto</include>
<include>client_SCM_protocol.proto</include>
<include>server/SCM_Admin_protocol.proto</include>
+ <include>yarn_csi_adaptor.proto</include>
+ <include>YarnCsiAdaptor.proto</include>
</includes>
</source>
</configuration>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorPB.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorPB.java
new file mode 100644
index 0000000..2a7a114
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorPB.java
@@ -0,0 +1,31 @@
+/**
+ * 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.yarn.api;
+
+import org.apache.hadoop.ipc.ProtocolInfo;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtocol;
+
+/**
+ * Interface for the CSI adaptor protocol.
+ */
+@ProtocolInfo(
+ protocolName = "CsiAdaptorPB",
+ protocolVersion = 1)
+public interface CsiAdaptorPB extends
+ CsiAdaptorProtocol.CsiAdaptorProtocolService.BlockingInterface {
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorProtocol.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorProtocol.java
new file mode 100644
index 0000000..0822163
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/CsiAdaptorProtocol.java
@@ -0,0 +1,39 @@
+/**
+ * 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.yarn.api;
+
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+import java.io.IOException;
+
+/**
+ * CSI adaptor delegates all the calls from YARN to a CSI driver.
+ */
+public interface CsiAdaptorProtocol {
+
+ GetPluginInfoResponse getPluginInfo(GetPluginInfoRequest request)
+ throws YarnException, IOException;
+
+ ValidateVolumeCapabilitiesResponse validateVolumeCapacity(
+ ValidateVolumeCapabilitiesRequest request) throws YarnException,
+ IOException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoRequest.java
new file mode 100644
index 0000000..a32c93c
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoRequest.java
@@ -0,0 +1,30 @@
+/**
+ * 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.yarn.api.protocolrecords;
+
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * Get plugin info request.
+ */
+public abstract class GetPluginInfoRequest {
+
+ public static GetPluginInfoRequest newInstance() {
+ return Records.newRecord(GetPluginInfoRequest.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoResponse.java
new file mode 100644
index 0000000..9577200
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/GetPluginInfoResponse.java
@@ -0,0 +1,43 @@
+/**
+ * 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.yarn.api.protocolrecords;
+
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * Get plugin info response.
+ */
+public abstract class GetPluginInfoResponse {
+
+ public static GetPluginInfoResponse newInstance(
+ String driverName, String version) {
+ GetPluginInfoResponse response =
+ Records.newRecord(GetPluginInfoResponse.class);
+ response.setDriverName(driverName);
+ response.setVersion(version);
+ return response;
+ }
+
+ public abstract void setDriverName(String driverName);
+
+ public abstract String getDriverName();
+
+ public abstract void setVersion(String version);
+
+ public abstract String getVersion();
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesRequest.java
new file mode 100644
index 0000000..0ab1b1f
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesRequest.java
@@ -0,0 +1,117 @@
+/**
+ * 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.yarn.api.protocolrecords;
+
+import org.apache.hadoop.yarn.util.Records;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * YARN internal message used to validate volume capabilities
+ * with a CSI driver controller plugin.
+ */
+public abstract class ValidateVolumeCapabilitiesRequest {
+
+ /**
+ * Volume access mode.
+ */
+ public enum AccessMode {
+ UNKNOWN,
+ SINGLE_NODE_WRITER,
+ SINGLE_NODE_READER_ONLY,
+ MULTI_NODE_READER_ONLY,
+ MULTI_NODE_SINGLE_WRITER,
+ MULTI_NODE_MULTI_WRITER,
+ }
+
+ /**
+ * Volume type.
+ */
+ public enum VolumeType {
+ BLOCK,
+ FILE_SYSTEM
+ }
+
+ /**
+ * Volume capability.
+ */
+ public static class VolumeCapability {
+
+ private AccessMode mode;
+ private VolumeType type;
+ private List<String> flags;
+
+ public VolumeCapability(AccessMode accessMode, VolumeType volumeType,
+ List<String> mountFlags) {
+ this.mode = accessMode;
+ this.type = volumeType;
+ this.flags = mountFlags;
+ }
+
+ public AccessMode getAccessMode() {
+ return mode;
+ }
+
+ public VolumeType getVolumeType() {
+ return type;
+ }
+
+ public List<String> getMountFlags() {
+ return flags;
+ }
+ }
+
+ public static ValidateVolumeCapabilitiesRequest newInstance(
+ String volumeId, List<VolumeCapability> volumeCapabilities,
+ Map<String, String> volumeAttributes) {
+ ValidateVolumeCapabilitiesRequest
+ request =
+ Records.newRecord(
+ ValidateVolumeCapabilitiesRequest.class);
+ request.setVolumeId(volumeId);
+ request.setVolumeAttributes(volumeAttributes);
+ for (VolumeCapability capability : volumeCapabilities) {
+ request.addVolumeCapability(capability);
+ }
+ return request;
+ }
+
+ public static ValidateVolumeCapabilitiesRequest newInstance(
+ String volumeId, Map<String, String> volumeAttributes) {
+ ValidateVolumeCapabilitiesRequest
+ request =
+ Records.newRecord(
+ ValidateVolumeCapabilitiesRequest.class);
+ request.setVolumeId(volumeId);
+ request.setVolumeAttributes(volumeAttributes);
+ return request;
+ }
+
+ public abstract void setVolumeId(String volumeId);
+
+ public abstract String getVolumeId();
+
+ public abstract void setVolumeAttributes(Map<String, String> attributes);
+
+ public abstract Map<String, String> getVolumeAttributes();
+
+ public abstract void addVolumeCapability(VolumeCapability volumeCapability);
+
+ public abstract List<VolumeCapability> getVolumeCapabilities();
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesResponse.java
new file mode 100644
index 0000000..a72ba0a
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/ValidateVolumeCapabilitiesResponse.java
@@ -0,0 +1,46 @@
+/**
+ * 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.yarn.api.protocolrecords;
+
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * YARN internal message used to represent the response of
+ * volume capabilities validation with a CSI driver controller plugin.
+ */
+public abstract class ValidateVolumeCapabilitiesResponse {
+
+ public static ValidateVolumeCapabilitiesResponse newInstance(
+ boolean supported, String responseMessage) {
+ ValidateVolumeCapabilitiesResponse
+ record =
+ Records.newRecord(
+ ValidateVolumeCapabilitiesResponse.class);
+ record.setResponseMessage(responseMessage);
+ record.setSupported(supported);
+ return record;
+ }
+
+ public abstract void setSupported(boolean supported);
+
+ public abstract boolean isSupported();
+
+ public abstract void setResponseMessage(String responseMessage);
+
+ public abstract String getResponseMessage();
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index e88d594..2f25284 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -3428,6 +3428,18 @@ public class YarnConfiguration extends Configuration {
false;
////////////////////////////////
+ // CSI Volume configs
+ ////////////////////////////////
+ /**
+ * One or more socket addresses for csi-adaptor.
+ * Multiple addresses are delimited by ",".
+ */
+ public static final String NM_CSI_ADAPTOR_PREFIX =
+ NM_PREFIX + "csi-driver-adaptor.";
+ public static final String NM_CSI_ADAPTOR_ADDRESSES =
+ NM_CSI_ADAPTOR_PREFIX + "addresses";
+
+ ////////////////////////////////
// Other Configs
////////////////////////////////
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/YarnCsiAdaptor.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/YarnCsiAdaptor.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/YarnCsiAdaptor.proto
new file mode 100644
index 0000000..9dcb8a7
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/YarnCsiAdaptor.proto
@@ -0,0 +1,34 @@
+/**
+ * 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.
+ */
+
+option java_package = "org.apache.hadoop.yarn.proto";
+option java_outer_classname = "CsiAdaptorProtocol";
+option java_generic_services = true;
+option java_generate_equals_and_hash = true;
+package hadoop.yarn;
+
+import "yarn_csi_adaptor.proto";
+
+service CsiAdaptorProtocolService {
+
+ rpc getPluginInfo (GetPluginInfoRequest)
+ returns (GetPluginInfoResponse);
+
+ rpc validateVolumeCapacity (ValidateVolumeCapabilitiesRequest)
+ returns (ValidateVolumeCapabilitiesResponse);
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_csi_adaptor.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_csi_adaptor.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_csi_adaptor.proto
new file mode 100644
index 0000000..c9adbea
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_csi_adaptor.proto
@@ -0,0 +1,69 @@
+/**
+ * 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.
+ */
+option java_package = "org.apache.hadoop.yarn.proto";
+option java_outer_classname = "CsiAdaptorProtos";
+option java_generate_equals_and_hash = true;
+package hadoop.yarn;
+
+import "yarn_protos.proto";
+
+message ValidateVolumeCapabilitiesRequest {
+ required string volume_id = 1;
+ repeated VolumeCapability volume_capabilities = 2;
+ repeated StringStringMapProto volume_attributes = 3;
+}
+
+message ValidateVolumeCapabilitiesResponse {
+ // True if the Plugin supports the specified capabilities for the
+ // given volume. This field is REQUIRED.
+ required bool supported = 1;
+
+ // Message to the CO if `supported` above is false. This field is
+ // OPTIONAL.
+ // An empty string is equal to an unspecified field value.
+ optional string message = 2;
+}
+
+message VolumeCapability {
+ enum VolumeType {
+ BLOCK = 0;
+ FILE_SYSTEM = 1;
+ }
+
+ enum AccessMode {
+ UNKNOWN = 0;
+ SINGLE_NODE_WRITER = 1;
+ SINGLE_NODE_READER_ONLY = 2;
+ MULTI_NODE_READER_ONLY = 3;
+ MULTI_NODE_SINGLE_WRITER = 4;
+ MULTI_NODE_MULTI_WRITER = 5;
+ }
+
+ required VolumeType volume_type = 1;
+ required AccessMode access_mode = 2;
+ repeated string mount_flags = 3;
+}
+
+message GetPluginInfoRequest {
+ // Intentionally empty.
+}
+
+message GetPluginInfoResponse {
+ required string name = 1;
+ required string vendor_version = 2;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/CsiAdaptorProtocolPBClientImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/CsiAdaptorProtocolPBClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/CsiAdaptorProtocolPBClientImpl.java
new file mode 100644
index 0000000..2e10f72
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/CsiAdaptorProtocolPBClientImpl.java
@@ -0,0 +1,91 @@
+/**
+ * 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.yarn.api.impl.pb.client;
+
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.yarn.api.CsiAdaptorPB;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetPluginInfoRequestPBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetPluginInfoResponsePBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ValidateVolumeCapabilitiesRequestPBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ValidateVolumeCapabilitiesResponsePBImpl;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+/**
+ * CSI adaptor client implementation.
+ */
+public class CsiAdaptorProtocolPBClientImpl
+ implements CsiAdaptorProtocol, Closeable {
+
+ private final CsiAdaptorPB proxy;
+
+ public CsiAdaptorProtocolPBClientImpl(long clientVersion,
+ InetSocketAddress addr, Configuration conf) throws IOException {
+ RPC.setProtocolEngine(conf, CsiAdaptorPB.class, ProtobufRpcEngine.class);
+ this.proxy = RPC.getProxy(CsiAdaptorPB.class, clientVersion, addr, conf);
+ }
+
+ @Override
+ public GetPluginInfoResponse getPluginInfo(
+ GetPluginInfoRequest request) throws YarnException, IOException {
+ CsiAdaptorProtos.GetPluginInfoRequest requestProto =
+ ((GetPluginInfoRequestPBImpl) request).getProto();
+ try {
+ return new GetPluginInfoResponsePBImpl(
+ proxy.getPluginInfo(null, requestProto));
+ } catch (ServiceException e) {
+ RPCUtil.unwrapAndThrowException(e);
+ return null;
+ }
+ }
+
+ @Override
+ public ValidateVolumeCapabilitiesResponse validateVolumeCapacity(
+ ValidateVolumeCapabilitiesRequest request)
+ throws YarnException, IOException {
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest requestProto =
+ ((ValidateVolumeCapabilitiesRequestPBImpl) request).getProto();
+ try {
+ return new ValidateVolumeCapabilitiesResponsePBImpl(
+ proxy.validateVolumeCapacity(null, requestProto));
+ } catch (ServiceException e) {
+ RPCUtil.unwrapAndThrowException(e);
+ return null;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if(this.proxy != null) {
+ RPC.stopProxy(this.proxy);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/CsiAdaptorProtocolPBServiceImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/CsiAdaptorProtocolPBServiceImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/CsiAdaptorProtocolPBServiceImpl.java
new file mode 100644
index 0000000..9a19435
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/CsiAdaptorProtocolPBServiceImpl.java
@@ -0,0 +1,75 @@
+/**
+ * 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.yarn.api.impl.pb.service;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.yarn.api.CsiAdaptorPB;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetPluginInfoRequestPBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.GetPluginInfoResponsePBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ValidateVolumeCapabilitiesRequestPBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ValidateVolumeCapabilitiesResponsePBImpl;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+
+import java.io.IOException;
+
+/**
+ * CSI adaptor server side implementation, this is hosted on a node manager.
+ */
+public class CsiAdaptorProtocolPBServiceImpl implements CsiAdaptorPB {
+
+ private final CsiAdaptorProtocol real;
+ public CsiAdaptorProtocolPBServiceImpl(CsiAdaptorProtocol impl) {
+ this.real = impl;
+ }
+
+ @Override
+ public CsiAdaptorProtos.GetPluginInfoResponse getPluginInfo(
+ RpcController controller, CsiAdaptorProtos.GetPluginInfoRequest request)
+ throws ServiceException {
+ try {
+ GetPluginInfoRequest req =
+ new GetPluginInfoRequestPBImpl(request);
+ GetPluginInfoResponse response = real.getPluginInfo(req);
+ return ((GetPluginInfoResponsePBImpl) response).getProto();
+ } catch (YarnException | IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+ @Override
+ public CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse
+ validateVolumeCapacity(RpcController controller,
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest request)
+ throws ServiceException {
+ try {
+ ValidateVolumeCapabilitiesRequestPBImpl req =
+ new ValidateVolumeCapabilitiesRequestPBImpl(request);
+ ValidateVolumeCapabilitiesResponse response =
+ real.validateVolumeCapacity(req);
+ return ((ValidateVolumeCapabilitiesResponsePBImpl) response).getProto();
+ } catch (YarnException | IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoRequestPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoRequestPBImpl.java
new file mode 100644
index 0000000..0d1c2e5
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoRequestPBImpl.java
@@ -0,0 +1,60 @@
+/**
+ * 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.yarn.api.protocolrecords.impl.pb;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoRequest;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+
+/**
+ * Get plugin info request protobuf impl.
+ */
+public class GetPluginInfoRequestPBImpl extends GetPluginInfoRequest {
+
+ private CsiAdaptorProtos.GetPluginInfoRequest.Builder builder;
+
+ public GetPluginInfoRequestPBImpl(
+ CsiAdaptorProtos.GetPluginInfoRequest requestProto) {
+ this.builder = requestProto.toBuilder();
+ }
+
+ public GetPluginInfoRequestPBImpl() {
+ this.builder = CsiAdaptorProtos.GetPluginInfoRequest.newBuilder();
+ }
+
+ public CsiAdaptorProtos.GetPluginInfoRequest getProto() {
+ Preconditions.checkNotNull(builder);
+ return builder.build();
+ }
+
+ @Override
+ public int hashCode() {
+ return getProto().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null) {
+ return false;
+ }
+ if (other.getClass().isAssignableFrom(this.getClass())) {
+ return this.getProto().equals(this.getClass().cast(other).getProto());
+ }
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoResponsePBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoResponsePBImpl.java
new file mode 100644
index 0000000..141fd6d
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/GetPluginInfoResponsePBImpl.java
@@ -0,0 +1,84 @@
+/**
+ * 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.yarn.api.protocolrecords.impl.pb;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+
+/**
+ * Get plugin info response protobuf impl.
+ */
+public class GetPluginInfoResponsePBImpl extends GetPluginInfoResponse {
+
+ private CsiAdaptorProtos.GetPluginInfoResponse.Builder builder;
+
+ public GetPluginInfoResponsePBImpl(
+ CsiAdaptorProtos.GetPluginInfoResponse responseProto) {
+ this.builder = responseProto.toBuilder();
+ }
+
+ public GetPluginInfoResponsePBImpl() {
+ this.builder = CsiAdaptorProtos.GetPluginInfoResponse.newBuilder();
+ }
+
+ @Override
+ public void setDriverName(String driverName) {
+ Preconditions.checkNotNull(builder);
+ builder.setName(driverName);
+ }
+
+ @Override
+ public String getDriverName() {
+ Preconditions.checkNotNull(builder);
+ return builder.getName();
+ }
+
+ @Override
+ public void setVersion(String version) {
+ Preconditions.checkNotNull(builder);
+ builder.setVendorVersion(version);
+ }
+
+ @Override
+ public String getVersion() {
+ Preconditions.checkNotNull(builder);
+ return builder.getVendorVersion();
+ }
+
+ public CsiAdaptorProtos.GetPluginInfoResponse getProto() {
+ Preconditions.checkNotNull(builder);
+ return builder.build();
+ }
+
+ @Override
+ public int hashCode() {
+ return getProto().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null) {
+ return false;
+ }
+ if (other.getClass().isAssignableFrom(this.getClass())) {
+ return this.getProto().equals(this.getClass().cast(other).getProto());
+ }
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesRequestPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesRequestPBImpl.java
new file mode 100644
index 0000000..14bd89d
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesRequestPBImpl.java
@@ -0,0 +1,121 @@
+/**
+ * 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.yarn.api.protocolrecords.impl.pb;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * PB wrapper for CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest.
+ */
+public class ValidateVolumeCapabilitiesRequestPBImpl extends
+ ValidateVolumeCapabilitiesRequest {
+
+ private CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest.Builder builder;
+
+ public ValidateVolumeCapabilitiesRequestPBImpl(
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest proto) {
+ this.builder = proto.toBuilder();
+ }
+
+ public ValidateVolumeCapabilitiesRequestPBImpl() {
+ this.builder = CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest
+ .newBuilder();
+ }
+
+ @Override
+ public String getVolumeId() {
+ Preconditions.checkNotNull(builder);
+ return builder.getVolumeId();
+ }
+
+ @Override
+ public void setVolumeAttributes(Map<String, String> attributes) {
+ Preconditions.checkNotNull(builder);
+ builder.addAllVolumeAttributes(ProtoUtils.convertToProtoFormat(attributes));
+ }
+
+ @Override
+ public void setVolumeId(String volumeId) {
+ Preconditions.checkNotNull(builder);
+ builder.setVolumeId(volumeId);
+ }
+
+ @Override
+ public void addVolumeCapability(VolumeCapability volumeCapability) {
+ Preconditions.checkNotNull(builder);
+ CsiAdaptorProtos.VolumeCapability vc =
+ CsiAdaptorProtos.VolumeCapability.newBuilder()
+ .setAccessMode(CsiAdaptorProtos.VolumeCapability.AccessMode
+ .valueOf(volumeCapability.getAccessMode().ordinal()))
+ .setVolumeType(CsiAdaptorProtos.VolumeCapability.VolumeType
+ .valueOf(volumeCapability.getVolumeType().ordinal()))
+ .addAllMountFlags(volumeCapability.getMountFlags())
+ .build();
+ builder.addVolumeCapabilities(vc);
+ }
+
+ @Override
+ public List<VolumeCapability> getVolumeCapabilities() {
+ Preconditions.checkNotNull(builder);
+ List<VolumeCapability> caps = new ArrayList<>(
+ builder.getVolumeCapabilitiesCount());
+ builder.getVolumeCapabilitiesList().forEach(capability -> {
+ VolumeCapability vc = new VolumeCapability(
+ AccessMode.valueOf(capability.getAccessMode().name()),
+ VolumeType.valueOf(capability.getVolumeType().name()),
+ capability.getMountFlagsList());
+ caps.add(vc);
+ });
+ return caps;
+ }
+
+ @Override
+ public Map<String, String> getVolumeAttributes() {
+ Preconditions.checkNotNull(builder);
+ return ProtoUtils.convertStringStringMapProtoListToMap(
+ builder.getVolumeAttributesList());
+ }
+
+ public CsiAdaptorProtos.ValidateVolumeCapabilitiesRequest getProto() {
+ Preconditions.checkNotNull(builder);
+ return builder.build();
+ }
+
+ @Override
+ public int hashCode() {
+ return getProto().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null) {
+ return false;
+ }
+ if (other.getClass().isAssignableFrom(this.getClass())) {
+ return this.getProto().equals(this.getClass().cast(other).getProto());
+ }
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesResponsePBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesResponsePBImpl.java
new file mode 100644
index 0000000..aa33ab7
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/ValidateVolumeCapabilitiesResponsePBImpl.java
@@ -0,0 +1,87 @@
+/**
+ * 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.yarn.api.protocolrecords.impl.pb;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.proto.CsiAdaptorProtos;
+
+/**
+ * PB wrapper for CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse.
+ */
+public class ValidateVolumeCapabilitiesResponsePBImpl
+ extends ValidateVolumeCapabilitiesResponse {
+
+ private CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse.Builder builder;
+
+ public ValidateVolumeCapabilitiesResponsePBImpl() {
+ this.builder = CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse
+ .newBuilder();
+ }
+
+ public ValidateVolumeCapabilitiesResponsePBImpl(
+ CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse response) {
+ this.builder = response.toBuilder();
+ }
+
+ @Override
+ public void setSupported(boolean supported) {
+ Preconditions.checkNotNull(builder);
+ this.builder.setSupported(supported);
+ }
+
+ @Override
+ public boolean isSupported() {
+ Preconditions.checkNotNull(builder);
+ return builder.getSupported();
+ }
+
+ @Override
+ public void setResponseMessage(String message) {
+ Preconditions.checkNotNull(builder);
+ this.builder.setMessage(message);
+ }
+
+ @Override
+ public String getResponseMessage() {
+ Preconditions.checkNotNull(builder);
+ return this.builder.getMessage();
+ }
+
+ @Override
+ public int hashCode() {
+ return getProto().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null) {
+ return false;
+ }
+ if (other.getClass().isAssignableFrom(this.getClass())) {
+ return this.getProto().equals(this.getClass().cast(other).getProto());
+ }
+ return false;
+ }
+
+ public CsiAdaptorProtos.ValidateVolumeCapabilitiesResponse getProto() {
+ Preconditions.checkNotNull(builder);
+ return builder.build();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index f5493bc..e7a0e14 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -4037,4 +4037,17 @@
<name>yarn.node-attribute.fs-store.impl.class</name>
<value>org.apache.hadoop.yarn.server.resourcemanager.nodelabels.FileSystemNodeAttributeStore</value>
</property>
+
+ <!-- CSI configuration -->
+ <property>
+ <description>
+ CSI driver adaptor addresses on a node manager.
+ This configuration will be loaded by the resource manager to initiate
+ a client for each adaptor in order to communicate with CSI drivers.
+ Note, these addresses should be mapped to the adaptor addresses which
+ runs the controller plugin.
+ </description>
+ <name>yarn.nodemanager.csi-driver-adaptor.addresses</name>
+ <value></value>
+ </property>
</configuration>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/pom.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/pom.xml
index 41f5098..27d8452 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/pom.xml
@@ -84,6 +84,18 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-api</artifactId>
+ </dependency>
+ <dependency>
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
<scope>compile</scope>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/CsiAdaptorProtocolService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/CsiAdaptorProtocolService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/CsiAdaptorProtocolService.java
new file mode 100644
index 0000000..f94275f
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/CsiAdaptorProtocolService.java
@@ -0,0 +1,122 @@
+/**
+ * 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.yarn.csi.adaptor;
+
+import com.google.common.annotations.VisibleForTesting;
+import csi.v0.Csi;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.service.AbstractService;
+import org.apache.hadoop.yarn.api.CsiAdaptorProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.csi.client.CsiClient;
+import org.apache.hadoop.yarn.csi.client.CsiClientImpl;
+import org.apache.hadoop.yarn.csi.translator.ProtoTranslatorFactory;
+import org.apache.hadoop.yarn.csi.utils.ConfigUtils;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+/**
+ * This is a Hadoop RPC server, we uses the Hadoop RPC framework here
+ * because we need to stick to the security model current Hadoop supports.
+ */
+public class CsiAdaptorProtocolService extends AbstractService
+ implements CsiAdaptorProtocol {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(CsiAdaptorProtocolService.class);
+
+ private Server server;
+ private InetSocketAddress adaptorServiceAddress;
+ private CsiClient csiClient;
+ private String csiDriverName;
+
+ public CsiAdaptorProtocolService(String driverName,
+ String domainSocketPath) {
+ super(CsiAdaptorProtocolService.class.getName());
+ this.csiClient = new CsiClientImpl(domainSocketPath);
+ this.csiDriverName = driverName;
+ }
+
+ @VisibleForTesting
+ public void setCsiClient(CsiClient client) {
+ this.csiClient = client;
+ }
+
+ @Override
+ protected void serviceInit(Configuration conf) throws Exception {
+ adaptorServiceAddress = ConfigUtils
+ .getCsiAdaptorAddressForDriver(csiDriverName, conf);
+ super.serviceInit(conf);
+ }
+
+ @Override
+ protected void serviceStart() throws Exception {
+ Configuration conf = getConfig();
+ YarnRPC rpc = YarnRPC.create(conf);
+ this.server = rpc.getServer(
+ CsiAdaptorProtocol.class,
+ this, adaptorServiceAddress, conf, null, 1);
+ this.server.start();
+ LOG.info("{} started, listening on address: {}",
+ CsiAdaptorProtocolService.class.getName(),
+ adaptorServiceAddress.toString());
+ super.serviceStart();
+ }
+
+ @Override
+ protected void serviceStop() throws Exception {
+ if (this.server != null) {
+ this.server.stop();
+ }
+ super.serviceStop();
+ }
+
+ @Override
+ public GetPluginInfoResponse getPluginInfo(
+ GetPluginInfoRequest request) throws YarnException, IOException {
+ Csi.GetPluginInfoResponse response = csiClient.getPluginInfo();
+ return ProtoTranslatorFactory.getTranslator(
+ GetPluginInfoResponse.class, Csi.GetPluginInfoResponse.class)
+ .convertFrom(response);
+ }
+
+ @Override
+ public ValidateVolumeCapabilitiesResponse validateVolumeCapacity(
+ ValidateVolumeCapabilitiesRequest request) throws YarnException,
+ IOException {
+ Csi.ValidateVolumeCapabilitiesRequest req = ProtoTranslatorFactory
+ .getTranslator(ValidateVolumeCapabilitiesRequest.class,
+ Csi.ValidateVolumeCapabilitiesRequest.class)
+ .convertTo(request);
+ Csi.ValidateVolumeCapabilitiesResponse response =
+ csiClient.validateVolumeCapabilities(req);
+ return ProtoTranslatorFactory.getTranslator(
+ ValidateVolumeCapabilitiesResponse.class,
+ Csi.ValidateVolumeCapabilitiesResponse.class)
+ .convertFrom(response);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java
new file mode 100644
index 0000000..919aab5
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/adaptor/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * This package contains CSI adaptor classes.
+ */
+package org.apache.hadoop.yarn.csi.adaptor;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClient.java
index 5bb9ce9..d31c0c9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClient.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.yarn.csi.client;
+import csi.v0.Csi;
import csi.v0.Csi.GetPluginInfoResponse;
import java.io.IOException;
@@ -36,4 +37,7 @@ public interface CsiClient {
* @throws IOException when unable to get plugin info from the driver.
*/
GetPluginInfoResponse getPluginInfo() throws IOException;
+
+ Csi.ValidateVolumeCapabilitiesResponse validateVolumeCapabilities(
+ Csi.ValidateVolumeCapabilitiesRequest request) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClientImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClientImpl.java
index 58dd292..5b3d2e2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClientImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/client/CsiClientImpl.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.yarn.csi.client;
+import csi.v0.Csi;
import csi.v0.Csi.GetPluginInfoRequest;
import csi.v0.Csi.GetPluginInfoResponse;
import org.apache.hadoop.yarn.csi.utils.GrpcHelper;
@@ -48,4 +49,14 @@ public class CsiClientImpl implements CsiClient {
return client.createIdentityBlockingStub().getPluginInfo(request);
}
}
+
+ @Override
+ public Csi.ValidateVolumeCapabilitiesResponse validateVolumeCapabilities(
+ Csi.ValidateVolumeCapabilitiesRequest request) throws IOException {
+ try (CsiGrpcClient client = CsiGrpcClient.newBuilder()
+ .setDomainSocketAddress(address).build()) {
+ return client.createControllerBlockingStub()
+ .validateVolumeCapabilities(request);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/GetPluginInfoResponseProtoTranslator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/GetPluginInfoResponseProtoTranslator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/GetPluginInfoResponseProtoTranslator.java
new file mode 100644
index 0000000..c4f042e
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/GetPluginInfoResponseProtoTranslator.java
@@ -0,0 +1,44 @@
+/**
+ * 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.yarn.csi.translator;
+
+import csi.v0.Csi;
+import org.apache.hadoop.yarn.api.protocolrecords.GetPluginInfoResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+/**
+ * Protobuf message translator for GetPluginInfoResponse and
+ * Csi.GetPluginInfoResponse.
+ */
+public class GetPluginInfoResponseProtoTranslator implements
+ ProtoTranslator<GetPluginInfoResponse, Csi.GetPluginInfoResponse> {
+
+ @Override public Csi.GetPluginInfoResponse convertTo(
+ GetPluginInfoResponse messageA) throws YarnException {
+ return Csi.GetPluginInfoResponse.newBuilder()
+ .setName(messageA.getDriverName())
+ .setVendorVersion(messageA.getVersion())
+ .build();
+ }
+
+ @Override public GetPluginInfoResponse convertFrom(
+ Csi.GetPluginInfoResponse messageB) throws YarnException {
+ return GetPluginInfoResponse.newInstance(messageB.getName(),
+ messageB.getVendorVersion());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslator.java
new file mode 100644
index 0000000..93cb441
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslator.java
@@ -0,0 +1,49 @@
+/**
+ * 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.yarn.csi.translator;
+
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+/**
+ * ProtoTranslator converts a YARN side message to CSI proto message
+ * and vice versa. Each CSI proto message should have a corresponding
+ * YARN side message implementation, and a transformer to convert them
+ * one to the other. This layer helps we to hide CSI spec messages
+ * from YARN components.
+ *
+ * @param <A> YARN side internal messages
+ * @param <B> CSI proto messages
+ */
+public interface ProtoTranslator<A, B> {
+
+ /**
+ * Convert message from type A to type B.
+ * @param messageA
+ * @return messageB
+ * @throws YarnException
+ */
+ B convertTo(A messageA) throws YarnException;
+
+ /**
+ * Convert message from type B to type A.
+ * @param messageB
+ * @return messageA
+ * @throws YarnException
+ */
+ A convertFrom(B messageB) throws YarnException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslatorFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslatorFactory.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslatorFactory.java
new file mode 100644
index 0000000..5eb76ff
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ProtoTranslatorFactory.java
@@ -0,0 +1,66 @@
+/**
+ * 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.yarn.csi.translator;
+
+import csi.v0.Csi;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+
+/**
+ * Factory class to get desired proto transformer instance.
+ */
+public final class ProtoTranslatorFactory {
+
+ private ProtoTranslatorFactory() {
+ // hide constructor for the factory class
+ }
+
+ /**
+ * Get a {@link ProtoTranslator} based on the given input message
+ * types. If the type is not supported, a IllegalArgumentException
+ * will be thrown. When adding more transformers to this factory class,
+ * note each transformer works exactly for one message to another
+ * (and vice versa). For each type of the message, make sure there is
+ * a corresponding unit test added, such as
+ * TestValidateVolumeCapabilitiesRequest.
+ *
+ * @param yarnProto yarn proto message
+ * @param csiProto CSI proto message
+ * @param <A> yarn proto message
+ * @param <B> CSI proto message
+ * @throws IllegalArgumentException
+ * when given types are not supported
+ * @return
+ * a proto message transformer that transforms
+ * YARN internal proto message to CSI
+ */
+ public static <A, B> ProtoTranslator<A, B> getTranslator(
+ Class<A> yarnProto, Class<B> csiProto) {
+ if (yarnProto == ValidateVolumeCapabilitiesRequest.class
+ && csiProto == Csi.ValidateVolumeCapabilitiesRequest.class) {
+ return new ValidateVolumeCapabilitiesRequestProtoTranslator();
+ } else if (yarnProto == ValidateVolumeCapabilitiesResponse.class
+ && csiProto == Csi.ValidateVolumeCapabilitiesResponse.class) {
+ return new ValidationVolumeCapabilitiesResponseProtoTranslator();
+ }
+ throw new IllegalArgumentException("A problem is found while processing"
+ + " proto message translating. Unexpected message types,"
+ + " no transformer is found can handle the transformation from type "
+ + yarnProto.getName() + " <-> " + csiProto.getName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidateVolumeCapabilitiesRequestProtoTranslator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidateVolumeCapabilitiesRequestProtoTranslator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidateVolumeCapabilitiesRequestProtoTranslator.java
new file mode 100644
index 0000000..a74c47a
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidateVolumeCapabilitiesRequestProtoTranslator.java
@@ -0,0 +1,93 @@
+/**
+ * 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.yarn.csi.translator;
+
+import csi.v0.Csi;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.VolumeCapability;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesRequest.VolumeType;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Proto message translator for ValidateVolumeCapabilitiesRequest.
+ * @param <A> ValidateVolumeCapabilitiesRequest
+ * @param <B> Csi.ValidateVolumeCapabilitiesRequest
+ */
+public class ValidateVolumeCapabilitiesRequestProtoTranslator<A, B>
+ implements ProtoTranslator<ValidateVolumeCapabilitiesRequest,
+ Csi.ValidateVolumeCapabilitiesRequest> {
+
+ @Override
+ public Csi.ValidateVolumeCapabilitiesRequest convertTo(
+ ValidateVolumeCapabilitiesRequest request) throws YarnException {
+ Csi.ValidateVolumeCapabilitiesRequest.Builder buidler =
+ Csi.ValidateVolumeCapabilitiesRequest.newBuilder();
+ buidler.setVolumeId(request.getVolumeId());
+ if (request.getVolumeCapabilities() != null
+ && request.getVolumeCapabilities().size() > 0) {
+ buidler.putAllVolumeAttributes(request.getVolumeAttributes());
+ }
+ for (VolumeCapability cap :
+ request.getVolumeCapabilities()) {
+ Csi.VolumeCapability.AccessMode accessMode =
+ Csi.VolumeCapability.AccessMode.newBuilder()
+ .setModeValue(cap.getAccessMode().ordinal())
+ .build();
+ Csi.VolumeCapability.MountVolume mountVolume =
+ Csi.VolumeCapability.MountVolume.newBuilder()
+ .addAllMountFlags(cap.getMountFlags())
+ .build();
+ Csi.VolumeCapability capability =
+ Csi.VolumeCapability.newBuilder()
+ .setAccessMode(accessMode)
+ .setMount(mountVolume)
+ .build();
+ buidler.addVolumeCapabilities(capability);
+ }
+ return buidler.build();
+ }
+
+ @Override
+ public ValidateVolumeCapabilitiesRequest convertFrom(
+ Csi.ValidateVolumeCapabilitiesRequest request) throws YarnException {
+ ValidateVolumeCapabilitiesRequest result = ValidateVolumeCapabilitiesRequest
+ .newInstance(request.getVolumeId(), request.getVolumeAttributesMap());
+ for (Csi.VolumeCapability csiCap :
+ request.getVolumeCapabilitiesList()) {
+ ValidateVolumeCapabilitiesRequest.AccessMode mode =
+ ValidateVolumeCapabilitiesRequest.AccessMode
+ .valueOf(csiCap.getAccessMode().getMode().name());
+ if (!csiCap.hasMount()) {
+ throw new YarnException("Invalid request,"
+ + " mount is not found in the request.");
+ }
+ List<String> mountFlags = new ArrayList<>();
+ for (int i=0; i<csiCap.getMount().getMountFlagsCount(); i++) {
+ mountFlags.add(csiCap.getMount().getMountFlags(i));
+ }
+ VolumeCapability capability = new VolumeCapability(mode,
+ VolumeType.FILE_SYSTEM, mountFlags);
+ result.addVolumeCapability(capability);
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidationVolumeCapabilitiesResponseProtoTranslator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidationVolumeCapabilitiesResponseProtoTranslator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidationVolumeCapabilitiesResponseProtoTranslator.java
new file mode 100644
index 0000000..fd42712
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/ValidationVolumeCapabilitiesResponseProtoTranslator.java
@@ -0,0 +1,48 @@
+/**
+ * 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.yarn.csi.translator;
+
+import csi.v0.Csi;
+import org.apache.hadoop.yarn.api.protocolrecords.ValidateVolumeCapabilitiesResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+/**
+ * Proto message translator for ValidateVolumeCapabilitiesResponse.
+ * @param <A> ValidateVolumeCapabilitiesResponse
+ * @param <B> Csi.ValidateVolumeCapabilitiesResponse
+ */
+public class ValidationVolumeCapabilitiesResponseProtoTranslator<A, B>
+ implements ProtoTranslator<ValidateVolumeCapabilitiesResponse,
+ Csi.ValidateVolumeCapabilitiesResponse> {
+
+ @Override
+ public Csi.ValidateVolumeCapabilitiesResponse convertTo(
+ ValidateVolumeCapabilitiesResponse response) throws YarnException {
+ return Csi.ValidateVolumeCapabilitiesResponse.newBuilder()
+ .setSupported(response.isSupported())
+ .setMessage(response.getResponseMessage())
+ .build();
+ }
+
+ @Override
+ public ValidateVolumeCapabilitiesResponse convertFrom(
+ Csi.ValidateVolumeCapabilitiesResponse response) throws YarnException {
+ return ValidateVolumeCapabilitiesResponse.newInstance(
+ response.getSupported(), response.getMessage());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/package-info.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/package-info.java
new file mode 100644
index 0000000..c0964cd
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/translator/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * This package contains classes for protocol translation between YARN and CSI.
+ */
+package org.apache.hadoop.yarn.csi.translator;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb14e06/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/utils/ConfigUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/utils/ConfigUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/utils/ConfigUtils.java
new file mode 100644
index 0000000..77e6955
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-csi/src/main/java/org/apache/hadoop/yarn/csi/utils/ConfigUtils.java
@@ -0,0 +1,61 @@
+/**
+ * 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.yarn.csi.utils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Utility class to load configurations.
+ */
+public final class ConfigUtils {
+
+ private ConfigUtils() {
+ // Hide constructor for utility class.
+ }
+ /**
+ * Resolve the CSI adaptor address for a CSI driver from configuration.
+ * Expected configuration property name is
+ * yarn.nodemanager.csi-driver-adaptor.${driverName}.address.
+ * @param driverName
+ * @param conf
+ * @return adaptor service address
+ * @throws YarnException
+ */
+ public static InetSocketAddress getCsiAdaptorAddressForDriver(
+ String driverName, Configuration conf) throws YarnException {
+ String configName = YarnConfiguration.NM_CSI_ADAPTOR_PREFIX
+ + driverName + ".address";
+ String errorMessage = "Failed to load CSI adaptor address for driver "
+ + driverName + ", configuration property " + configName
+ + " is not defined or invalid.";
+ try {
+ InetSocketAddress address = conf
+ .getSocketAddr(configName, null, -1);
+ if (address == null) {
+ throw new YarnException(errorMessage);
+ }
+ return address;
+ } catch (IllegalArgumentException e) {
+ throw new YarnException(errorMessage);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org