yarn-operator: add yarn application client

Signed-off-by: 佑祎 <[email protected]>

config/manager/configmap.yaml

@@ -10,6 +10,10 @@ data:
             <name>yarn.resourcemanager.admin.address</name>
             <value>0.0.0.0:8033</value>
         </property>
+        <property>
+            <name>yarn.resourcemanager.address</name>
+            <value>0.0.0.0:8032</value>
+        </property>
         <property>
           <name>yarn.resourcemanager.ha.enabled</name>
           <value>true</value>
@@ -30,6 +34,17 @@ data:
             <name>yarn.resourcemanager.admin.address.rm3</name>
             <value>0.0.0.0:8033</value>
         </property>
+        <property>
+            <name>yarn.resourcemanager.address.rm1</name>
+            <value>0.0.0.0:8032</value>
+        </property>
+        <property>
+            <name>yarn.resourcemanager.address.rm2</name>
+            <value>0.0.0.0:8032</value>
+        </property>
+        <property>
+            <name>yarn.resourcemanager.address.rm3</name>
+            <value>0.0.0.0:8032</value>
     </configuration>
   core-site.xml: |
     <configuration>

hack/generate-yarn.sh

@@ -72,7 +72,7 @@ function generate_import_files_pkg_map() {
   # server/yarn_server_resourcemanager_service_protos.proto
   import_paths=("$( grep -E "import \".*.proto\";" ${file_path} | grep -Eo "\".*\"" | sed "s/\"//g" )")
   if [ ! -z "${import_paths}" ]; then
-    for import_path in "${import_paths[@]}"
+    for import_path in ${import_paths[@]}
     do
         import_file_name="$( echo ${import_path} | grep -Eo "[a-z,_,A-Z]*.proto$" )" # yarn_server_resourcemanager_service_protos.proto
         import_file_path="$( find ${GOYARN_API_PATH} -name ${import_file_name} )" # ./hack/../pkg/yarn/apis/proto/hadoopyarn/server/yarn_server_resourcemanager_service_protos.proto

pkg/controller/noderesource/resource_sync_controller.go

@@ -47,7 +47,7 @@ const (
 
 type YARNResourceSyncReconciler struct {
 	client.Client
-	yarnClient *yarnclient.YARNClient
+	yarnClient *yarnclient.YarnClient
 }
 
 func (r *YARNResourceSyncReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
@@ -91,7 +91,7 @@ func (r *YARNResourceSyncReconciler) Reconcile(ctx context.Context, req reconcil
 }
 
 func Add(mgr ctrl.Manager) error {
-	yarnClient, err := yarnclient.CreateYARNClient()
+	yarnClient, err := yarnclient.CreateYarnClient()
 	if err != nil {
 		return err
 	}

pkg/yarn/apis/proto/hadoopyarn/applicationclient_protocol.proto

@@ -0,0 +1,71 @@
+/**
+ * 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.
+ */
+
+/**
+ * These .proto interfaces are public and stable.
+ * Please see https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/Compatibility.html
+ * for what changes are allowed for a *stable* .proto interface.
+ */
+
+syntax = "proto2";
+option java_package = "org.apache.hadoop.yarn.proto";
+option java_outer_classname = "ApplicationClientProtocol";
+option java_generic_services = true;
+option java_generate_equals_and_hash = true;
+package hadoop.yarn;
+
+import "Security.proto";
+import "yarn_service_protos.proto";
+
+service ApplicationClientProtocolService {
+  rpc getNewApplication (GetNewApplicationRequestProto) returns (GetNewApplicationResponseProto);
+  rpc getApplicationReport (GetApplicationReportRequestProto) returns (GetApplicationReportResponseProto);
+  rpc submitApplication (SubmitApplicationRequestProto) returns (SubmitApplicationResponseProto);
+  rpc failApplicationAttempt (FailApplicationAttemptRequestProto) returns (FailApplicationAttemptResponseProto);
+  rpc forceKillApplication (KillApplicationRequestProto) returns (KillApplicationResponseProto);
+  rpc getClusterMetrics (GetClusterMetricsRequestProto) returns (GetClusterMetricsResponseProto);
+  rpc getApplications (GetApplicationsRequestProto) returns (GetApplicationsResponseProto);
+  rpc getClusterNodes (GetClusterNodesRequestProto) returns (GetClusterNodesResponseProto);
+  rpc getQueueInfo (GetQueueInfoRequestProto) returns (GetQueueInfoResponseProto);
+  rpc getQueueUserAcls (GetQueueUserAclsInfoRequestProto) returns (GetQueueUserAclsInfoResponseProto);
+  rpc getDelegationToken(hadoop.common.GetDelegationTokenRequestProto) returns (hadoop.common.GetDelegationTokenResponseProto);
+  rpc renewDelegationToken(hadoop.common.RenewDelegationTokenRequestProto) returns (hadoop.common.RenewDelegationTokenResponseProto);
+  rpc cancelDelegationToken(hadoop.common.CancelDelegationTokenRequestProto) returns (hadoop.common.CancelDelegationTokenResponseProto);
+  rpc moveApplicationAcrossQueues(MoveApplicationAcrossQueuesRequestProto) returns (MoveApplicationAcrossQueuesResponseProto);
+  rpc getApplicationAttemptReport (GetApplicationAttemptReportRequestProto) returns (GetApplicationAttemptReportResponseProto);
+  rpc getApplicationAttempts (GetApplicationAttemptsRequestProto) returns (GetApplicationAttemptsResponseProto);
+  rpc getContainerReport (GetContainerReportRequestProto) returns (GetContainerReportResponseProto);
+  rpc getContainers (GetContainersRequestProto) returns (GetContainersResponseProto);
+  rpc getNewReservation (GetNewReservationRequestProto) returns (GetNewReservationResponseProto);
+  rpc submitReservation (ReservationSubmissionRequestProto) returns (ReservationSubmissionResponseProto);
+  rpc updateReservation (ReservationUpdateRequestProto) returns (ReservationUpdateResponseProto);
+  rpc deleteReservation (ReservationDeleteRequestProto) returns (ReservationDeleteResponseProto);
+  rpc listReservations (ReservationListRequestProto) returns (ReservationListResponseProto);
+  rpc getNodeToLabels (GetNodesToLabelsRequestProto) returns (GetNodesToLabelsResponseProto);
+  rpc getLabelsToNodes (GetLabelsToNodesRequestProto) returns (GetLabelsToNodesResponseProto);
+  rpc getClusterNodeLabels (GetClusterNodeLabelsRequestProto) returns (GetClusterNodeLabelsResponseProto);
+  rpc updateApplicationPriority (UpdateApplicationPriorityRequestProto) returns (UpdateApplicationPriorityResponseProto);
+  rpc signalToContainer(SignalContainerRequestProto) returns (SignalContainerResponseProto);
+  rpc updateApplicationTimeouts (UpdateApplicationTimeoutsRequestProto) returns (UpdateApplicationTimeoutsResponseProto);
+  rpc getResourceProfiles(GetAllResourceProfilesRequestProto) returns (GetAllResourceProfilesResponseProto);
+  rpc getResourceProfile(GetResourceProfileRequestProto) returns (GetResourceProfileResponseProto);
+  rpc getResourceTypeInfo(GetAllResourceTypeInfoRequestProto) returns (GetAllResourceTypeInfoResponseProto);
+  rpc getClusterNodeAttributes (GetClusterNodeAttributesRequestProto) returns (GetClusterNodeAttributesResponseProto);
+  rpc getAttributesToNodes (GetAttributesToNodesRequestProto) returns (GetAttributesToNodesResponseProto);
+  rpc getNodesToAttributes (GetNodesToAttributesRequestProto) returns (GetNodesToAttributesResponseProto);
+}