yarn-operator: only start yarn node syncer on leader (#53)

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

charts/hadoop-yarn/templates/hadoop-configmap.yaml

@@ -67,18 +67,21 @@ data:
  cat >> $HADOOP_HOME/etc/hadoop/yarn-site.xml <<- EOM
  <property>
  <name>yarn.nodemanager.resource.memory-mb</name>
- <value>${NM_INIT_MEMORY_MB:-2048}</value>
+ <value>${NM_INIT_MEMORY_MB:-1024}</value>
  </property>
  <property>
  <name>yarn.nodemanager.resource.cpu-vcores</name>
- <value>${NM_INIT_CPU_CORES:-2}</value>
+ <value>${NM_INIT_CPU_CORES:-1}</value>
  </property>
  <property>
  <name>yarn.nodemanager.address</name>
  <value>${HOSTNAME}:8041</value>
  </property>
  EOM
 
+ # annotate nm id on pod
+ kubectl annotate pod -n ${POD_NAMESPACE} ${POD_NAME} yarn.hadoop.apache.org/node-id=${HOSTNAME}:8041
+
  echo '</configuration>' >> $HADOOP_HOME/etc/hadoop/yarn-site.xml
 
  # Wait with timeout for resourcemanager
@@ -89,7 +92,6 @@ data:
  echo "$0: Timeout waiting for $TMP_URL, exiting."
  exit 1
  fi
-
  fi
 
  # ------------------------------------------------------

charts/hadoop-yarn/templates/rbac.yaml

@@ -0,0 +1,36 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+ name: yarn-nodemanager
+ namespace: {{ .Values.installation.namespace }}
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+ name: yarn-nodemanager-role
+ namespace: {{ .Values.installation.namespace }}
+rules:
+ - apiGroups:
+ - ""
+ resources:
+ - pods
+ verbs:
+ - patch
+ - update
+ - get
+ - list
+ - watch
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+ name: yarn-nodemanager-rolebinding
+ namespace : {{ .Values.installation.namespace }}
+roleRef:
+ apiGroup: rbac.authorization.k8s.io
+ kind: Role
+ name: yarn-nodemanager-role
+subjects:
+ - kind: ServiceAccount
+ name: yarn-nodemanager
+ namespace: {{ .Values.installation.namespace }}

charts/hadoop-yarn/templates/yarn-nm-statefulset.yaml

@@ -42,6 +42,7 @@ spec:
  app.kubernetes.io/name: {{ include "hadoop-yarn.name" . }}
  app.kubernetes.io/instance: {{ .Release.Name }}
  app.kubernetes.io/component: node-manager
+ serviceAccountName: yarn-nodemanager
  setHostnameAsFQDN: true
  terminationGracePeriodSeconds: 0
  dnsPolicy: ClusterFirst
@@ -93,9 +94,17 @@ spec:
  - name: YARN_ROLE
  value: yarn-nm
  - name: NM_INIT_CPU_CORES
- value: "4"
+ value: "{{ .Values.yarn.nodeManager.initCPUVCores }}"
  - name: NM_INIT_MEMORY_MB
- value: "4096"
+ value: "{{ .Values.yarn.nodeManager.initMemoryMB }}"
+ - name: POD_NAMESPACE
+ valueFrom:
+ fieldRef:
+ fieldPath: metadata.namespace
+ - name: POD_NAME
+ valueFrom:
+ fieldRef:
+ fieldPath: metadata.name
  volumeMounts:
  - name: hadoop-config
  mountPath: /tmp/hadoop-config

charts/hadoop-yarn/values.yaml

@@ -4,7 +4,7 @@ installation:
 image:
  repository: registry.cn-hangzhou.aliyuncs.com/koordinator-sh/apache-hadoop
  tag: 3.3.2-v1.0
- pullPolicy: IfNotPresent
+ pullPolicy: Always
 
 # The version of the hadoop libraries being used in the image.
 hadoopVersion: 3.3.2
@@ -50,6 +50,10 @@ yarn:
 
  serviceName: node-manager
 
+ # initial cpu and memory of nm reported
+ initCPUVCores: 1
+ initMemoryMB: 1024
+
  config:
  yarnSite:
  cgroupsHierarchy: /kubepods.slice/kubepods-besteffort.slice/hadoop-yarn

docker/hadoop-yarn.dockerfile

@@ -1,6 +1,6 @@
 FROM alpine:3.14 as BUILDER
 
-ENV HADOOP_VERSION 3.3.2
+ENV HADOOP_VERSION 3.3.3
 ENV SPARK_VERSION 3.3.3
 
 RUN apk update \
@@ -14,7 +14,7 @@ RUN curl -s -o /tmp/spark.tgz https://mirrors.aliyun.com/apache/spark/spark-${SP
 
 FROM openjdk:8
 
-ENV HADOOP_VERSION 3.3.2
+ENV HADOOP_VERSION 3.3.3
 ENV SPARK_VERSION 3.3.3
 ENV SPARK_HOME=/opt/spark
 ENV HADOOP_HOME=/opt/hadoop
@@ -29,6 +29,14 @@ ENV HADOOP_COMMON_HOME=${HADOOP_HOME} \
 COPY --from=BUILDER /opt/hadoop-${HADOOP_VERSION} ${HADOOP_HOME}
 COPY --from=BUILDER /opt/spark-${SPARK_VERSION}-bin-hadoop3 ${SPARK_HOME}
 
-RUN apt-get update && apt-get install -y dnsutils
+RUN apt-get update && apt-get install -y apt-transport-https
+RUN curl https://mirrors.aliyun.com/kubernetes/apt/doc/apt-key.gpg | apt-key add -
+RUN echo 'deb https://mirrors.aliyun.com/kubernetes/apt/ kubernetes-xenial main' >> /etc/apt/sources.list.d/kubernetes.list
+#RUN cat <<EOF >/etc/apt/sources.list.d/kubernetes.list \
+# deb https://mirrors.aliyun.com/kubernetes/apt/ kubernetes-xenial main \
+# EOF
+
+RUN apt-get update
+RUN apt-get install -y kubectl dnsutils
 
 WORKDIR $HADOOP_HOME

pkg/controller/noderesource/resource_sync_controller.go

@@ -82,13 +82,13 @@ func (r *YARNResourceSyncReconciler) Reconcile(ctx context.Context, req reconcil
  // TODO exclude batch pod requested
  batchCPU, batchMemory, err := getNodeBatchResource(node)
  if err != nil {
- return ctrl.Result{}, err
+ return ctrl.Result{Requeue: true}, err
  }
  klog.V(4).Infof("get node batch resource cpu: %d, memory: %d, name: %s", batchCPU.Value(), batchMemory.Value(), node.Name)
 
  vcores, memoryMB := calculate(batchCPU, batchMemory)
 
- // TODO control update frequency
+ // TODO control update frequency by ignore unnecessary node update event
  if err := r.updateYARNNodeResource(yarnNode, vcores, memoryMB); err != nil {
  klog.Warningf("update batch resource to yarn node %+v failed, k8s node name: %s, error %v", yarnNode, node.Name, err)
  return ctrl.Result{Requeue: true}, err
@@ -98,7 +98,7 @@ func (r *YARNResourceSyncReconciler) Reconcile(ctx context.Context, req reconcil
 
  core, mb, err := r.getYARNNodeAllocatedResource(yarnNode)
  if err != nil {
- return reconcile.Result{}, err
+ return reconcile.Result{Requeue: true}, err
  }
  if err := r.updateYARNAllocatedResource(node, core, mb); err != nil {
  klog.Warningf("failed to update yarn allocated resource for node %v, error %v", node.Name, err)
@@ -171,7 +171,7 @@ func Add(mgr ctrl.Manager) error {
  return err
  }
  yarnNodesSyncer := cache.NewNodesSyncer(clients)
- go yarnNodesSyncer.Sync()
+
  coll := yarnmetrics.NewYarnMetricCollector(yarnNodesSyncer)
  if err = metrics.Registry.Register(coll); err != nil {
  return err
@@ -181,10 +181,15 @@ func Add(mgr ctrl.Manager) error {
  yarnClients: clients,
  yarnNodeCache: yarnNodesSyncer,
  }
+
+ if err := mgr.Add(yarnNodesSyncer); err != nil {
+ return err
+ }
  return r.SetupWithManager(mgr)
 }
 
 func (r *YARNResourceSyncReconciler) SetupWithManager(mgr ctrl.Manager) error {
+ // TODO use source.Channel to handle yarn node requested update event
  return ctrl.NewControllerManagedBy(mgr).
  For(&corev1.Node{}).
  Named(Name).

pkg/yarn/cache/nodes_syncer.go

@@ -17,6 +17,7 @@ limitations under the License.
 package cache
 
 import (
+ "context"
  "fmt"
  "sync"
  "time"
@@ -67,19 +68,25 @@ func (r *NodesSyncer) getKey(yarnNodeName string, yarnNodePort int32) string {
  return fmt.Sprintf("%s-%d", yarnNodeName, yarnNodePort)
 }
 
-func (r *NodesSyncer) Sync() {
+func (r *NodesSyncer) Start(ctx context.Context) error {
  t := time.NewTicker(syncInterval)
  debug := time.NewTicker(syncInterval * 10)
- for {
- select {
- case <-t.C:
- if err := r.syncYARNNodeAllocatedResource(); err != nil {
- klog.Errorf("sync yarn node allocated resource failed, error: %v", err)
+ go func() {
+ for {
+ select {
+ case <-t.C:
+ if err := r.syncYARNNodeAllocatedResource(); err != nil {
+ klog.Errorf("sync yarn node allocated resource failed, error: %v", err)
+ }
+ case <-debug.C:
+ r.debug()
+ case <-ctx.Done():
+ klog.V(1).Infof("stop node syncer")
+ return
  }
- case <-debug.C:
- r.debug()
  }
- }
+ }()
+ return nil
 }
 
 func (r *NodesSyncer) debug() {