Spark on Kubernetes 问题记录

executor pod 残留

问题描述

使用 Client 模式启动 Spark 任务，删除 Driver Pod 后，Executor Pods 并没有被删除，而是进入 Error 状态残留在集群中
问题原因

Driver 收到 Delete 指令后，偶现 Executor 与 Driver 通信提前断连，Executor 无人回收。

解决方案

配置 spark.kubernetes.driver.pod.name 参数，参数值为 Driver Pod 的实际名称。Spark Scheduler 将会创建带有 Pod ownerReference 信息的 Executor。
https://spark.apache.org/docs/latest/running-on-kubernetes.html#cluster-mode

Pod Garbage Collection

// repo: spark-parent_2.12 tag: v2.4.5 commit: cee4ecbb16
// Add a OwnerReference to the given resources making the driver pod an owner of them so when
  // the driver pod is deleted, the resources are garbage collected.
  private def addDriverOwnerReference(driverPod: Pod, resources: Seq[HasMetadata]): Unit = {
    val driverPodOwnerReference = new OwnerReferenceBuilder()
      .withName(driverPod.getMetadata.getName)
      .withApiVersion(driverPod.getApiVersion)
      .withUid(driverPod.getMetadata.getUid)
      .withKind(driverPod.getKind)
      .withController(true)
      .build()
    resources.foreach { resource =>
      val originalMetadata = resource.getMetadata
      originalMetadata.setOwnerReferences(Collections.singletonList(driverPodOwnerReference))
    }
  }

ownerReference 意味 Driver 与 Executor 建立了级联关系。Driver 为 Executor 的拥有者，反之 Executor 是 Driver 的附属。在默认显式垃圾回收下，当删除 Driver 时，Executor 也会被级联垃圾回收。

https://kubernetes.io/zh/docs/concepts/workloads/controllers/garbage-collection/

ownerReferences

Driver 中使用 downWard Api，将 Pod Name 通过环境变量方式进行暴露。

downward api

Executor Pod outofcpu

问题描述

Spark 任务启动后，有部分 Executor Pod 进入 outofcpu 状态并不断重复创建销毁
问题原因

开始玩 Spark 任务时，使用的是 default-scheduler，TFJob 使用 Kube-Batch，存在多调度器共存场景，有可能发生 mutil-Scheduler Node Allocated 与实际资源数据不一致情况
解决方案

根本上解决此问题需要两种任务使用同一个调度器，保证数据一致性。目前有两种方案，具体细节如下
- Patch Spark 3.0.0 相关逻辑
  Spark 3.0.0 版本将支持配置 spark.kubernetes.executor.scheduler.name 参数，指定 Executor 使用自定义调度器
  
  https://github.com/apache/spark/blob/master/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala#L177
目前内部使用 Spark 2.4.4 版本，可 Patch 此特性重新编译 spark-2.4.4-mogu。

https://github.com/apache/spark/pull/26088
- Admission Mutating Webhook
  平台侧收到 Spark 任务创建请求后，根据 Executor 数量创建对应的 podGroup
  
  Kubernetes 收到 Driver 发出的 Executor 创建请求后，Admission Mutating Webhook 拦截此类 Pod Create 请求，识别 Pod 是否拥有 Executor Labels，以修改 schedulerName 名称为 Kube-Batch
  
  并根据 Driver Pod Name 为 Executor 添加 metadata.annotations[scheduling.k8s.io/group-name]，使用 Batch 能力进行调度。

Executor Pod cpuSet Failed

问题描述

Spark 任务创建后，某些 Executor Pod 创建未成功，显示 Internal PreStartContainer hook failed: not enough cpus available to satisfy request

问题原因

现场日志形如下图

kubelet

Kubelet 在调用 StartContainer 前，将执行 PreStartContainer

// Kubernetes tag: v1.18.3, coomit: 2e7996e3e27 
// pkg/kubelet/kuberuntime/kuberuntime_container.go:176
containerID, err := m.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)
	if err != nil {
		s, _ := grpcstatus.FromError(err)
		m.recordContainerEvent(pod, container, containerID, v1.EventTypeWarning, events.FailedToCreateContainer, "Error: %v", s.Message())
		return s.Message(), ErrCreateContainer
	}
	err = m.internalLifecycle.PreStartContainer(pod, container, containerID)
	if err != nil {
		s, _ := grpcstatus.FromError(err)
		m.recordContainerEvent(pod, container, containerID, v1.EventTypeWarning, events.FailedToStartContainer, "Internal PreStartContainer hook failed: %v", s.Message())
		return s.Message(), ErrPreStartHook
	}
	m.recordContainerEvent(pod, container, containerID, v1.EventTypeNormal, events.CreatedContainer, fmt.Sprintf("Created container %s", container.Name))

	if ref != nil {
		m.containerRefManager.SetRef(kubecontainer.ContainerID{
			Type: m.runtimeName,
			ID:   containerID,
		}, ref)
	}

	// Step 3: start the container.
	err = m.runtimeService.StartContainer(containerID)
	if err != nil {
		s, _ := grpcstatus.FromError(err)
		m.recordContainerEvent(pod, container, containerID, v1.EventTypeWarning, events.FailedToStartContainer, "Error: %v", s.Message())
		return s.Message(), kubecontainer.ErrRunContainer
	}
	m.recordContainerEvent(pod, container, containerID, v1.EventTypeNormal, events.StartedContainer, fmt.Sprintf("Started container %s", container.Name))

PreStartContainer 中将调用 cpuManager.AddContainer，依次对所有 containner 的 CPU 拓扑结构进行更新

// Kubernetes tag: v1.18.3, coomit: 2e7996e3e27 
// pkg/kubelet/cm/internal_container_lifecycle.go:40
func (i *internalContainerLifecycleImpl) PreStartContainer(pod *v1.Pod, container *v1.Container, containerID string) error {
	if i.cpuManager != nil {
		err := i.cpuManager.AddContainer(pod, container, containerID)
		if err != nil {
			return err
		}
	}
	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.TopologyManager) {
		err := i.topologyManager.AddContainer(pod, containerID)
		if err != nil {
			return err
		}
	}
	return nil
}

cpu_manager_state 拓扑信息因为某些复杂原因没有同步正确更新，导致逻辑上部分 CPU 仍绑在某个container上，导致后续 cpuset container 无法正常创建

解决方案

社区中有一个 PR 解决了cpu_manager 脏数据残留问题，已 Cherry pick release-1.13 和 release-1.12

主要逻辑是在 cpuManager reconcile 协程中进行对账操作，兜底清理 cpu_manager_state 脏数据

升级 Kubelet 版本能彻底解决此问题

https://github.com/kubernetes/kubernetes/pull/68619

cpu_manager_state