+ * The protocol between a NodeManager's
+ * SharedCacheUploadService and the
+ * SharedCacheManager.
+ *
+ * The method used by the NodeManager's SharedCacheUploadService
+ * to notify the shared cache manager of a newly cached resource.
+ *
+ * The SharedCacheManager responds with whether or not the
+ * NodeManager should delete the uploaded file.
+ *
+ * The method used by the NodeManager's SharedCacheUploadService
+ * to request whether a resource can be uploaded.
+ *
+ * The SharedCacheManager responds with whether or not the
+ * NodeManager can upload the file.
+ *
+ * The request from the NodeManager to the SharedCacheManager that
+ * requests whether it can upload a resource in the shared cache.
+ *
key of the resource that would be uploaded to the
+ * shared cache.
+ *
+ * @return key
+ */
+ public abstract String getResourceKey();
+
+ /**
+ * Set the key of the resource that would be uploaded to the
+ * shared cache.
+ *
+ * @param key unique identifier for the resource
+ */
+ public abstract void setResourceKey(String key);
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderCanUploadResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderCanUploadResponse.java
new file mode 100644
index 00000000000..5fb4626963e
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderCanUploadResponse.java
@@ -0,0 +1,52 @@
+/**
+ * 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.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+
+/**
+ * + * The response from the SharedCacheManager to the NodeManager that indicates + * whether the NodeManager can upload the resource to the shared cache. If it is + * not accepted by SCM, the NodeManager should not upload it to the shared + * cache. + *
+ */ +@Private +@Unstable +public abstract class SCMUploaderCanUploadResponse { + + /** + * Get whether or not the node manager can upload the resource to the shared + * cache. + * + * @return boolean True if the resource can be uploaded, false otherwise. + */ + public abstract boolean getUploadable(); + + /** + * Set whether or not the node manager can upload the resource to the shared + * cache. + * + * @param b True if the resource can be uploaded, false otherwise. + */ + public abstract void setUploadable(boolean b); + +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderNotifyRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderNotifyRequest.java new file mode 100644 index 00000000000..c72453c568c --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderNotifyRequest.java @@ -0,0 +1,67 @@ +/** + * 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.api.protocolrecords; + +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.classification.InterfaceStability.Unstable; + +/** + *
+ * The request from the NodeManager to the SharedCacheManager that
+ * notifies that a resource has been uploaded to the shared cache. The
+ * SharedCacheManager may reject the resource for various reasons,
+ * in which case the NodeManager should remove it from the shared cache.
+ *
key of the resource that was just uploaded to the
+ * shared cache.
+ *
+ * @return key
+ */
+ public abstract String getResourceKey();
+
+ /**
+ * Set the key of the resource that was just uploaded to the
+ * shared cache.
+ *
+ * @param key unique identifier for the resource
+ */
+ public abstract void setResourceKey(String key);
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderNotifyResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderNotifyResponse.java
new file mode 100644
index 00000000000..83d7913ddcf
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/SCMUploaderNotifyResponse.java
@@ -0,0 +1,51 @@
+/**
+ * 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.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+
+/**
+ * + * The response from the SharedCacheManager to the NodeManager that indicates + * whether the NodeManager needs to delete the cached resource it was sending + * the notification for. + *
+ */ +@Private +@Unstable +public abstract class SCMUploaderNotifyResponse { + + /** + * Get whether or not the shared cache manager has accepted the notified + * resource (i.e. the uploaded file should remain in the cache). + * + * @return boolean True if the resource has been accepted, false otherwise. + */ + public abstract boolean getAccepted(); + + /** + * Set whether or not the shared cache manager has accepted the notified + * resource (i.e. the uploaded file should remain in the cache). + * + * @param b True if the resource has been accepted, false otherwise. + */ + public abstract void setAccepted(boolean b); + +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderCanUploadRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderCanUploadRequestPBImpl.java new file mode 100644 index 00000000000..d350fb97dc6 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderCanUploadRequestPBImpl.java @@ -0,0 +1,78 @@ +/** + * 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.api.protocolrecords.impl.pb; + +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderCanUploadRequestProto; +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderCanUploadRequestProtoOrBuilder; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderCanUploadRequest; + +public class SCMUploaderCanUploadRequestPBImpl + extends SCMUploaderCanUploadRequest { + SCMUploaderCanUploadRequestProto proto = + SCMUploaderCanUploadRequestProto.getDefaultInstance(); + SCMUploaderCanUploadRequestProto.Builder builder = null; + boolean viaProto = false; + + public SCMUploaderCanUploadRequestPBImpl() { + builder = SCMUploaderCanUploadRequestProto.newBuilder(); + } + + public SCMUploaderCanUploadRequestPBImpl( + SCMUploaderCanUploadRequestProto proto) { + this.proto = proto; + viaProto = true; + } + + public SCMUploaderCanUploadRequestProto getProto() { + mergeLocalToProto(); + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + @Override + public String getResourceKey() { + SCMUploaderCanUploadRequestProtoOrBuilder p = viaProto ? proto : builder; + return (p.hasResourceKey()) ? p.getResourceKey() : null; + } + + @Override + public void setResourceKey(String key) { + maybeInitBuilder(); + if (key == null) { + builder.clearResourceKey(); + return; + } + builder.setResourceKey(key); + } + + private void mergeLocalToProto() { + if (viaProto) + maybeInitBuilder(); + proto = builder.build(); + viaProto = true; + } + + private void maybeInitBuilder() { + if (viaProto || builder == null) { + builder = SCMUploaderCanUploadRequestProto.newBuilder(proto); + } + viaProto = false; + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderCanUploadResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderCanUploadResponsePBImpl.java new file mode 100644 index 00000000000..3f44e2a009f --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderCanUploadResponsePBImpl.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.server.api.protocolrecords.impl.pb; + +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderCanUploadResponseProto; +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderCanUploadResponseProtoOrBuilder; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderCanUploadResponse; + +public class SCMUploaderCanUploadResponsePBImpl + extends SCMUploaderCanUploadResponse { + SCMUploaderCanUploadResponseProto proto = + SCMUploaderCanUploadResponseProto.getDefaultInstance(); + SCMUploaderCanUploadResponseProto.Builder builder = null; + boolean viaProto = false; + + public SCMUploaderCanUploadResponsePBImpl() { + builder = SCMUploaderCanUploadResponseProto.newBuilder(); + } + + public SCMUploaderCanUploadResponsePBImpl( + SCMUploaderCanUploadResponseProto proto) { + this.proto = proto; + viaProto = true; + } + + public SCMUploaderCanUploadResponseProto getProto() { + mergeLocalToProto(); + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + @Override + public boolean getUploadable() { + SCMUploaderCanUploadResponseProtoOrBuilder p = viaProto ? proto : builder; + // Default to true, when in doubt allow the upload + return (p.hasUploadable()) ? p.getUploadable() : true; + } + + @Override + public void setUploadable(boolean b) { + maybeInitBuilder(); + builder.setUploadable(b); + } + + private void mergeLocalToProto() { + if (viaProto) + maybeInitBuilder(); + proto = builder.build(); + viaProto = true; + } + + private void maybeInitBuilder() { + if (viaProto || builder == null) { + builder = SCMUploaderCanUploadResponseProto.newBuilder(proto); + } + viaProto = false; + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderNotifyRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderNotifyRequestPBImpl.java new file mode 100644 index 00000000000..9b52b115c00 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderNotifyRequestPBImpl.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.server.api.protocolrecords.impl.pb; + +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderNotifyRequestProto; +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderNotifyRequestProtoOrBuilder; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderNotifyRequest; + +public class SCMUploaderNotifyRequestPBImpl extends SCMUploaderNotifyRequest { + SCMUploaderNotifyRequestProto proto = + SCMUploaderNotifyRequestProto.getDefaultInstance(); + SCMUploaderNotifyRequestProto.Builder builder = null; + boolean viaProto = false; + + public SCMUploaderNotifyRequestPBImpl() { + builder = SCMUploaderNotifyRequestProto.newBuilder(); + } + + public SCMUploaderNotifyRequestPBImpl( + SCMUploaderNotifyRequestProto proto) { + this.proto = proto; + viaProto = true; + } + + public SCMUploaderNotifyRequestProto getProto() { + mergeLocalToProto(); + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + @Override + public String getResourceKey() { + SCMUploaderNotifyRequestProtoOrBuilder p = viaProto ? proto : builder; + return (p.hasResourceKey()) ? p.getResourceKey() : null; + } + + @Override + public void setResourceKey(String key) { + maybeInitBuilder(); + if (key == null) { + builder.clearResourceKey(); + return; + } + builder.setResourceKey(key); + } + + @Override + public String getFileName() { + SCMUploaderNotifyRequestProtoOrBuilder p = viaProto ? proto : builder; + return (p.hasFilename()) ? p.getFilename() : null; + } + + @Override + public void setFilename(String filename) { + maybeInitBuilder(); + if (filename == null) { + builder.clearFilename(); + return; + } + builder.setFilename(filename); + } + + private void mergeLocalToProto() { + if (viaProto) + maybeInitBuilder(); + proto = builder.build(); + viaProto = true; + } + + private void maybeInitBuilder() { + if (viaProto || builder == null) { + builder = SCMUploaderNotifyRequestProto.newBuilder(proto); + } + viaProto = false; + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderNotifyResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderNotifyResponsePBImpl.java new file mode 100644 index 00000000000..c899bbe6296 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/SCMUploaderNotifyResponsePBImpl.java @@ -0,0 +1,73 @@ +/** + * 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.api.protocolrecords.impl.pb; + +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderNotifyResponseProto; +import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.SCMUploaderNotifyResponseProtoOrBuilder; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderNotifyResponse; + +public class SCMUploaderNotifyResponsePBImpl extends SCMUploaderNotifyResponse { + SCMUploaderNotifyResponseProto proto = + SCMUploaderNotifyResponseProto.getDefaultInstance(); + SCMUploaderNotifyResponseProto.Builder builder = null; + boolean viaProto = false; + + public SCMUploaderNotifyResponsePBImpl() { + builder = SCMUploaderNotifyResponseProto.newBuilder(); + } + + public SCMUploaderNotifyResponsePBImpl(SCMUploaderNotifyResponseProto proto) { + this.proto = proto; + viaProto = true; + } + + public SCMUploaderNotifyResponseProto getProto() { + mergeLocalToProto(); + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + @Override + public boolean getAccepted() { + SCMUploaderNotifyResponseProtoOrBuilder p = viaProto ? proto : builder; + // Default to true, when in doubt just leave the file in the cache + return (p.hasAccepted()) ? p.getAccepted() : true; + } + + @Override + public void setAccepted(boolean b) { + maybeInitBuilder(); + builder.setAccepted(b); + } + + private void mergeLocalToProto() { + if (viaProto) + maybeInitBuilder(); + proto = builder.build(); + viaProto = true; + } + + private void maybeInitBuilder() { + if (viaProto || builder == null) { + builder = SCMUploaderNotifyResponseProto.newBuilder(proto); + } + viaProto = false; + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/SCMUploader.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/SCMUploader.proto new file mode 100644 index 00000000000..227842269e1 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/SCMUploader.proto @@ -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. + */ + +option java_package = "org.apache.hadoop.yarn.proto"; +option java_outer_classname = "SCMUploaderProtocol"; +option java_generic_services = true; +option java_generate_equals_and_hash = true; +package hadoop.yarn; + +import "yarn_server_common_service_protos.proto"; + +service SCMUploaderProtocolService { + rpc notify(SCMUploaderNotifyRequestProto) returns (SCMUploaderNotifyResponseProto); + rpc canUpload(SCMUploaderCanUploadRequestProto) returns (SCMUploaderCanUploadResponseProto); +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto index f2d01ada027..91473c5a2de 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto @@ -75,4 +75,21 @@ message NMContainerStatusProto { optional string diagnostics = 5 [default = "N/A"]; optional int32 container_exit_status = 6; optional int64 creation_time = 7; -} \ No newline at end of file +} + +message SCMUploaderNotifyRequestProto { + optional string resource_key = 1; + optional string filename = 2; +} + +message SCMUploaderNotifyResponseProto { + optional bool accepted = 1; +} + +message SCMUploaderCanUploadRequestProto { + optional string resource_key = 1; +} + +message SCMUploaderCanUploadResponseProto { + optional bool uploadable = 1; +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java index 3fdb5882103..ab50727b8dd 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java @@ -67,6 +67,10 @@ public class SharedCacheManager extends CompositeService { CleanerService cs = createCleanerService(store); addService(cs); + SharedCacheUploaderService nms = + createNMCacheUploaderSCMProtocolService(store); + addService(nms); + // init metrics DefaultMetricsSystem.initialize("SharedCacheManager"); JvmMetrics.initSingleton("SharedCacheManager", null); @@ -97,6 +101,11 @@ public class SharedCacheManager extends CompositeService { return new CleanerService(store); } + private SharedCacheUploaderService + createNMCacheUploaderSCMProtocolService(SCMStore store) { + return new SharedCacheUploaderService(store); + } + @Override protected void serviceStop() throws Exception { diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheUploaderService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheUploaderService.java new file mode 100644 index 00000000000..f9494386b02 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheUploaderService.java @@ -0,0 +1,140 @@ +/** + * 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.sharedcachemanager; + +import java.io.IOException; +import java.net.InetSocketAddress; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.ipc.Server; +import org.apache.hadoop.service.AbstractService; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.factories.RecordFactory; +import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider; +import org.apache.hadoop.yarn.ipc.YarnRPC; +import org.apache.hadoop.yarn.server.api.SCMUploaderProtocol; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderCanUploadRequest; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderCanUploadResponse; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderNotifyRequest; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderNotifyResponse; +import org.apache.hadoop.yarn.server.sharedcachemanager.metrics.SharedCacheUploaderMetrics; +import org.apache.hadoop.yarn.server.sharedcachemanager.store.SCMStore; + +/** + * This service handles all rpc calls from the NodeManager uploader to the + * shared cache manager. + */ +public class SharedCacheUploaderService extends AbstractService + implements SCMUploaderProtocol { + private final RecordFactory recordFactory = RecordFactoryProvider + .getRecordFactory(null); + + private Server server; + InetSocketAddress bindAddress; + private final SCMStore store; + private SharedCacheUploaderMetrics metrics; + + public SharedCacheUploaderService(SCMStore store) { + super(SharedCacheUploaderService.class.getName()); + this.store = store; + } + + @Override + protected void serviceInit(Configuration conf) throws Exception { + this.bindAddress = getBindAddress(conf); + + super.serviceInit(conf); + } + + InetSocketAddress getBindAddress(Configuration conf) { + return conf.getSocketAddr(YarnConfiguration.SCM_UPLOADER_SERVER_ADDRESS, + YarnConfiguration.DEFAULT_SCM_UPLOADER_SERVER_ADDRESS, + YarnConfiguration.DEFAULT_SCM_UPLOADER_SERVER_PORT); + } + + @Override + protected void serviceStart() throws Exception { + Configuration conf = getConfig(); + this.metrics = SharedCacheUploaderMetrics.initSingleton(conf); + + YarnRPC rpc = YarnRPC.create(conf); + this.server = + rpc.getServer(SCMUploaderProtocol.class, this, bindAddress, + conf, null, // Secret manager null for now (security not supported) + conf.getInt(YarnConfiguration.SCM_UPLOADER_SERVER_THREAD_COUNT, + YarnConfiguration.DEFAULT_SCM_UPLOADER_SERVER_THREAD_COUNT)); + + // TODO (YARN-2774): Enable service authorization + + this.server.start(); + bindAddress = + conf.updateConnectAddr(YarnConfiguration.SCM_UPLOADER_SERVER_ADDRESS, + server.getListenerAddress()); + + super.serviceStart(); + } + + @Override + protected void serviceStop() throws Exception { + if (this.server != null) { + this.server.stop(); + this.server = null; + } + + super.serviceStop(); + } + + @Override + public SCMUploaderNotifyResponse notify(SCMUploaderNotifyRequest request) + throws YarnException, IOException { + SCMUploaderNotifyResponse response = + recordFactory.newRecordInstance(SCMUploaderNotifyResponse.class); + + // TODO (YARN-2774): proper security/authorization needs to be implemented + + String filename = + store.addResource(request.getResourceKey(), request.getFileName()); + + boolean accepted = filename.equals(request.getFileName()); + + if (accepted) { + this.metrics.incAcceptedUploads(); + } else { + this.metrics.incRejectedUploads(); + } + + response.setAccepted(accepted); + + return response; + } + + @Override + public SCMUploaderCanUploadResponse canUpload( + SCMUploaderCanUploadRequest request) throws YarnException, IOException { + // TODO (YARN-2781): we may want to have a more flexible policy of + // instructing the node manager to upload only if it meets a certain + // criteria + // until then we return true for now + SCMUploaderCanUploadResponse response = + recordFactory.newRecordInstance(SCMUploaderCanUploadResponse.class); + response.setUploadable(true); + return response; + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/metrics/SharedCacheUploaderMetrics.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/metrics/SharedCacheUploaderMetrics.java new file mode 100644 index 00000000000..6fd816f29a6 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/metrics/SharedCacheUploaderMetrics.java @@ -0,0 +1,105 @@ +/** + * 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.sharedcachemanager.metrics; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.classification.InterfaceStability.Evolving; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.metrics2.MetricsSystem; +import org.apache.hadoop.metrics2.annotation.Metric; +import org.apache.hadoop.metrics2.annotation.Metrics; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.hadoop.metrics2.lib.MetricsRegistry; +import org.apache.hadoop.metrics2.lib.MutableCounterLong; + +/** + * This class is for maintaining shared cache uploader requests metrics + * and publishing them through the metrics interfaces. + */ +@Private +@Evolving +@Metrics(about="shared cache upload metrics", context="yarn") +public class SharedCacheUploaderMetrics { + + static final Log LOG = + LogFactory.getLog(SharedCacheUploaderMetrics.class); + final MetricsRegistry registry; + + SharedCacheUploaderMetrics() { + registry = new MetricsRegistry("SharedCacheUploaderRequests"); + LOG.debug("Initialized "+ registry); + } + + enum Singleton { + INSTANCE; + + SharedCacheUploaderMetrics impl; + + synchronized SharedCacheUploaderMetrics init(Configuration conf) { + if (impl == null) { + impl = create(); + } + return impl; + } + } + + public static SharedCacheUploaderMetrics + initSingleton(Configuration conf) { + return Singleton.INSTANCE.init(conf); + } + + public static SharedCacheUploaderMetrics getInstance() { + SharedCacheUploaderMetrics topMetrics = Singleton.INSTANCE.impl; + if (topMetrics == null) + throw new IllegalStateException( + "The SharedCacheUploaderMetrics singleton instance is not" + + "initialized. Have you called init first?"); + return topMetrics; + } + + static SharedCacheUploaderMetrics create() { + MetricsSystem ms = DefaultMetricsSystem.instance(); + + SharedCacheUploaderMetrics metrics = + new SharedCacheUploaderMetrics(); + ms.register("SharedCacheUploaderRequests", null, metrics); + return metrics; + } + + @Metric("Number of accepted uploads") MutableCounterLong acceptedUploads; + @Metric("Number of rejected uploads") MutableCounterLong rejectedUploads; + + /** + * One accepted upload event + */ + public void incAcceptedUploads() { + acceptedUploads.incr(); + } + + /** + * One rejected upload event + */ + public void incRejectedUploads() { + rejectedUploads.incr(); + } + + public long getAcceptedUploads() { return acceptedUploads.value(); } + public long getRejectUploads() { return rejectedUploads.value(); } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/test/java/org/apache/hadoop/yarn/server/sharedcachemanager/TestSharedCacheUploaderService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/test/java/org/apache/hadoop/yarn/server/sharedcachemanager/TestSharedCacheUploaderService.java new file mode 100644 index 00000000000..1cb0663c4ef --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/test/java/org/apache/hadoop/yarn/server/sharedcachemanager/TestSharedCacheUploaderService.java @@ -0,0 +1,188 @@ +/** + * 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.sharedcachemanager; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; + +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.Collection; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.ipc.RPC; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.factories.RecordFactory; +import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider; +import org.apache.hadoop.yarn.ipc.YarnRPC; +import org.apache.hadoop.yarn.server.api.SCMUploaderProtocol; +import org.apache.hadoop.yarn.server.api.protocolrecords.SCMUploaderNotifyRequest; +import org.apache.hadoop.yarn.server.sharedcachemanager.metrics.SharedCacheUploaderMetrics; +import org.apache.hadoop.yarn.server.sharedcachemanager.store.InMemorySCMStore; +import org.apache.hadoop.yarn.server.sharedcachemanager.store.SCMStore; +import org.apache.hadoop.yarn.server.sharedcachemanager.store.SharedCacheResourceReference; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + + +/** + * Basic unit tests for the NodeManger to SCM Protocol Service. + */ +public class TestSharedCacheUploaderService { + private static File testDir = null; + + @BeforeClass + public static void setupTestDirs() throws IOException { + testDir = new File("target", + TestSharedCacheUploaderService.class.getCanonicalName()); + testDir.delete(); + testDir.mkdirs(); + testDir = testDir.getAbsoluteFile(); + } + + @AfterClass + public static void cleanupTestDirs() throws IOException { + if (testDir != null) { + testDir.delete(); + } + } + + private SharedCacheUploaderService service; + private SCMUploaderProtocol proxy; + private SCMStore store; + private final RecordFactory recordFactory = RecordFactoryProvider + .getRecordFactory(null); + + @Before + public void startUp() { + Configuration conf = new Configuration(); + conf.set(YarnConfiguration.SCM_STORE_CLASS, + InMemorySCMStore.class.getName()); + conf.set(YarnConfiguration.SHARED_CACHE_ROOT, testDir.getPath()); + AppChecker appChecker = mock(AppChecker.class); + store = new InMemorySCMStore(appChecker); + store.init(conf); + store.start(); + + service = new SharedCacheUploaderService(store); + service.init(conf); + service.start(); + + YarnRPC rpc = YarnRPC.create(new Configuration()); + + InetSocketAddress scmAddress = + conf.getSocketAddr(YarnConfiguration.SCM_UPLOADER_SERVER_ADDRESS, + YarnConfiguration.DEFAULT_SCM_UPLOADER_SERVER_ADDRESS, + YarnConfiguration.DEFAULT_SCM_UPLOADER_SERVER_PORT); + + proxy = + (SCMUploaderProtocol) rpc.getProxy( + SCMUploaderProtocol.class, scmAddress, conf); + } + + @After + public void cleanUp() { + if (store != null) { + store.stop(); + } + + if (service != null) { + service.stop(); + } + + if (proxy != null) { + RPC.stopProxy(proxy); + } + } + + @Test + public void testNotify_noEntry() throws Exception { + long accepted = + SharedCacheUploaderMetrics.getInstance().getAcceptedUploads(); + + SCMUploaderNotifyRequest request = + recordFactory.newRecordInstance(SCMUploaderNotifyRequest.class); + request.setResourceKey("key1"); + request.setFilename("foo.jar"); + assertTrue(proxy.notify(request).getAccepted()); + Collection