github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rename KubernetesExecutor -> KubernetesMesosExecutor

Browse Source
This commit is contained in:
Dr. Stefan Schimanski
2015-10-25 10:27:54 -07:00
parent 12165cd6e0
commit 866a17d6d8
2 changed files with 31 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
contrib/mesos/pkg/executor/executor.go
View File

@@ -99,7 +99,7 @@ type NodeInfo struct {
// KubernetesExecutor is an mesos executor that runs pods // KubernetesExecutor is an mesos executor that runs pods
// in a minion machine. // in a minion machine.
type KubernetesExecutor struct { type KubernetesMesosExecutor struct {
updateChan chan<- kubetypes.PodUpdate // sent to the kubelet, closed on shutdown updateChan chan<- kubetypes.PodUpdate // sent to the kubelet, closed on shutdown
state stateType state stateType
tasks map[string]*kuberTask tasks map[string]*kuberTask
@@ -136,13 +136,13 @@ type Config struct {
NodeInfos chan<- NodeInfo NodeInfos chan<- NodeInfo
} }
func (k *KubernetesExecutor) isConnected() bool { func (k *KubernetesMesosExecutor) isConnected() bool {
return connectedState == (&k.state).get() return connectedState == (&k.state).get()
} }
// New creates a new kubernetes executor. // New creates a new kubernetes executor.
func New(config Config) *KubernetesExecutor { func New(config Config) *KubernetesMesosExecutor {
k := &KubernetesExecutor{ k := &KubernetesMesosExecutor{
updateChan: config.Updates, updateChan: config.Updates,
state: disconnectedState, state: disconnectedState,
tasks: make(map[string]*kuberTask), tasks: make(map[string]*kuberTask),
@@ -187,7 +187,7 @@ func New(config Config) *KubernetesExecutor {
return k return k
} }
func (k *KubernetesExecutor) Init(driver bindings.ExecutorDriver) { func (k *KubernetesMesosExecutor) Init(driver bindings.ExecutorDriver) {
k.killKubeletContainers() k.killKubeletContainers()
k.resetSuicideWatch(driver) k.resetSuicideWatch(driver)
@@ -196,7 +196,7 @@ func (k *KubernetesExecutor) Init(driver bindings.ExecutorDriver) {
//TODO(jdef) monitor kubeletFinished and shutdown if it happens //TODO(jdef) monitor kubeletFinished and shutdown if it happens
} }
func (k *KubernetesExecutor) isDone() bool { func (k *KubernetesMesosExecutor) isDone() bool {
select { select {
case <-k.terminate: case <-k.terminate:
return true return true
@@ -206,7 +206,7 @@ func (k *KubernetesExecutor) isDone() bool {
} }
// sendPodUpdate assumes that caller is holding state lock; returns true when update is sent otherwise false // sendPodUpdate assumes that caller is holding state lock; returns true when update is sent otherwise false
func (k *KubernetesExecutor) sendPodUpdate(u *kubetypes.PodUpdate) bool { func (k *KubernetesMesosExecutor) sendPodUpdate(u *kubetypes.PodUpdate) bool {
if k.isDone() { if k.isDone() {
return false return false
} }
@@ -215,7 +215,7 @@ func (k *KubernetesExecutor) sendPodUpdate(u *kubetypes.PodUpdate) bool {
} }
// Registered is called when the executor is successfully registered with the slave. // Registered is called when the executor is successfully registered with the slave.
func (k *KubernetesExecutor) Registered(driver bindings.ExecutorDriver, func (k *KubernetesMesosExecutor) Registered(driver bindings.ExecutorDriver,
executorInfo *mesos.ExecutorInfo, frameworkInfo *mesos.FrameworkInfo, slaveInfo *mesos.SlaveInfo) { executorInfo *mesos.ExecutorInfo, frameworkInfo *mesos.FrameworkInfo, slaveInfo *mesos.SlaveInfo) {
if k.isDone() { if k.isDone() {
return return
@@ -252,7 +252,7 @@ func (k *KubernetesExecutor) Registered(driver bindings.ExecutorDriver,
// Reregistered is called when the executor is successfully re-registered with the slave. // Reregistered is called when the executor is successfully re-registered with the slave.
// This can happen when the slave fails over. // This can happen when the slave fails over.
func (k *KubernetesExecutor) Reregistered(driver bindings.ExecutorDriver, slaveInfo *mesos.SlaveInfo) { func (k *KubernetesMesosExecutor) Reregistered(driver bindings.ExecutorDriver, slaveInfo *mesos.SlaveInfo) {
if k.isDone() { if k.isDone() {
return return
} }
@@ -280,7 +280,7 @@ func (k *KubernetesExecutor) Reregistered(driver bindings.ExecutorDriver, slaveI
} }
// initializeStaticPodsSource unzips the data slice into the static-pods directory // initializeStaticPodsSource unzips the data slice into the static-pods directory
func (k *KubernetesExecutor) initializeStaticPodsSource(data []byte) { func (k *KubernetesMesosExecutor) initializeStaticPodsSource(data []byte) {
log.V(2).Infof("extracting static pods config to %s", k.staticPodsConfigPath) log.V(2).Infof("extracting static pods config to %s", k.staticPodsConfigPath)
err := archive.UnzipDir(data, k.staticPodsConfigPath) err := archive.UnzipDir(data, k.staticPodsConfigPath)
if err != nil { if err != nil {
@@ -290,7 +290,7 @@ func (k *KubernetesExecutor) initializeStaticPodsSource(data []byte) {
} }
// Disconnected is called when the executor is disconnected from the slave. // Disconnected is called when the executor is disconnected from the slave.
func (k *KubernetesExecutor) Disconnected(driver bindings.ExecutorDriver) { func (k *KubernetesMesosExecutor) Disconnected(driver bindings.ExecutorDriver) {
if k.isDone() { if k.isDone() {
return return
} }
@@ -306,7 +306,7 @@ func (k *KubernetesExecutor) Disconnected(driver bindings.ExecutorDriver) {
// is running, but the binding is not recorded in the Kubernetes store yet. // is running, but the binding is not recorded in the Kubernetes store yet.
// This function is invoked to tell the executor to record the binding in the // This function is invoked to tell the executor to record the binding in the
// Kubernetes store and start the pod via the Kubelet. // Kubernetes store and start the pod via the Kubelet.
func (k *KubernetesExecutor) LaunchTask(driver bindings.ExecutorDriver, taskInfo *mesos.TaskInfo) { func (k *KubernetesMesosExecutor) LaunchTask(driver bindings.ExecutorDriver, taskInfo *mesos.TaskInfo) {
if k.isDone() { if k.isDone() {
return return
} }
@@ -356,7 +356,7 @@ func (k *KubernetesExecutor) LaunchTask(driver bindings.ExecutorDriver, taskInfo
go k.launchTask(driver, taskId, pod) go k.launchTask(driver, taskId, pod)
} }
func (k *KubernetesExecutor) handleChangedApiserverPod(pod *api.Pod) { func (k *KubernetesMesosExecutor) handleChangedApiserverPod(pod *api.Pod) {
// exclude "pre-scheduled" pods which have a NodeName set to this node without being scheduled already // exclude "pre-scheduled" pods which have a NodeName set to this node without being scheduled already
taskId := pod.Annotations[meta.TaskIdKey] taskId := pod.Annotations[meta.TaskIdKey]
if taskId == "" { if taskId == "" {
@@ -402,7 +402,7 @@ func (k *KubernetesExecutor) handleChangedApiserverPod(pod *api.Pod) {
// a timer that, upon expiration, causes this executor to commit suicide. // a timer that, upon expiration, causes this executor to commit suicide.
// this implementation runs asynchronously. callers that wish to wait for the // this implementation runs asynchronously. callers that wish to wait for the
// reset to complete may wait for the returned signal chan to close. // reset to complete may wait for the returned signal chan to close.
func (k *KubernetesExecutor) resetSuicideWatch(driver bindings.ExecutorDriver) <-chan struct{} { func (k *KubernetesMesosExecutor) resetSuicideWatch(driver bindings.ExecutorDriver) <-chan struct{} {
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
defer close(ch) defer close(ch)
@@ -432,7 +432,7 @@ func (k *KubernetesExecutor) resetSuicideWatch(driver bindings.ExecutorDriver) <
return ch return ch
} }
func (k *KubernetesExecutor) attemptSuicide(driver bindings.ExecutorDriver, abort <-chan struct{}) { func (k *KubernetesMesosExecutor) attemptSuicide(driver bindings.ExecutorDriver, abort <-chan struct{}) {
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
@@ -464,7 +464,7 @@ func (k *KubernetesExecutor) attemptSuicide(driver bindings.ExecutorDriver, abor
} }
// async continuation of LaunchTask // async continuation of LaunchTask
func (k *KubernetesExecutor) launchTask(driver bindings.ExecutorDriver, taskId string, pod *api.Pod) { func (k *KubernetesMesosExecutor) launchTask(driver bindings.ExecutorDriver, taskId string, pod *api.Pod) {
deleteTask := func() { deleteTask := func() {
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
@@ -588,7 +588,7 @@ func (k *KubernetesExecutor) launchTask(driver bindings.ExecutorDriver, taskId s
go k._launchTask(driver, taskId, podFullName, psf) go k._launchTask(driver, taskId, podFullName, psf)
} }
func (k *KubernetesExecutor) _launchTask(driver bindings.ExecutorDriver, taskId, podFullName string, psf podStatusFunc) { func (k *KubernetesMesosExecutor) _launchTask(driver bindings.ExecutorDriver, taskId, podFullName string, psf podStatusFunc) {
expired := make(chan struct{}) expired := make(chan struct{})
@@ -669,7 +669,7 @@ reportLost:
k.reportLostTask(driver, taskId, messages.LaunchTaskFailed) k.reportLostTask(driver, taskId, messages.LaunchTaskFailed)
} }
func (k *KubernetesExecutor) __launchTask(driver bindings.ExecutorDriver, taskId, podFullName string, psf podStatusFunc) { func (k *KubernetesMesosExecutor) __launchTask(driver bindings.ExecutorDriver, taskId, podFullName string, psf podStatusFunc) {
// TODO(nnielsen): Monitor health of pod and report if lost. // TODO(nnielsen): Monitor health of pod and report if lost.
// Should we also allow this to fail a couple of times before reporting lost? // Should we also allow this to fail a couple of times before reporting lost?
// What if the docker daemon is restarting and we can't connect, but it's // What if the docker daemon is restarting and we can't connect, but it's
@@ -692,7 +692,7 @@ func (k *KubernetesExecutor) __launchTask(driver bindings.ExecutorDriver, taskId
// whether the pod is running. It will only return false if the task is still registered and the pod is // whether the pod is running. It will only return false if the task is still registered and the pod is
// registered in Docker. Otherwise it returns true. If there's still a task record on file, but no pod // registered in Docker. Otherwise it returns true. If there's still a task record on file, but no pod
// in Docker, then we'll also send a TASK_LOST event. // in Docker, then we'll also send a TASK_LOST event.
func (k *KubernetesExecutor) checkForLostPodTask(driver bindings.ExecutorDriver, taskId string, isKnownPod func() bool) bool { func (k *KubernetesMesosExecutor) checkForLostPodTask(driver bindings.ExecutorDriver, taskId string, isKnownPod func() bool) bool {
// TODO (jdefelice) don't send false alarms for deleted pods (KILLED tasks) // TODO (jdefelice) don't send false alarms for deleted pods (KILLED tasks)
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
@@ -716,7 +716,7 @@ func (k *KubernetesExecutor) checkForLostPodTask(driver bindings.ExecutorDriver,
} }
// KillTask is called when the executor receives a request to kill a task. // KillTask is called when the executor receives a request to kill a task.
func (k *KubernetesExecutor) KillTask(driver bindings.ExecutorDriver, taskId *mesos.TaskID) { func (k *KubernetesMesosExecutor) KillTask(driver bindings.ExecutorDriver, taskId *mesos.TaskID) {
if k.isDone() { if k.isDone() {
return return
} }
@@ -735,14 +735,14 @@ func (k *KubernetesExecutor) KillTask(driver bindings.ExecutorDriver, taskId *me
// Reports a lost task to the slave and updates internal task and pod tracking state. // Reports a lost task to the slave and updates internal task and pod tracking state.
// Assumes that the caller is locking around pod and task state. // Assumes that the caller is locking around pod and task state.
func (k *KubernetesExecutor) reportLostTask(driver bindings.ExecutorDriver, tid, reason string) { func (k *KubernetesMesosExecutor) reportLostTask(driver bindings.ExecutorDriver, tid, reason string) {
k.removePodTask(driver, tid, reason, mesos.TaskState_TASK_LOST) k.removePodTask(driver, tid, reason, mesos.TaskState_TASK_LOST)
} }
// deletes the pod and task associated with the task identified by tid and sends a task // deletes the pod and task associated with the task identified by tid and sends a task
// status update to mesos. also attempts to reset the suicide watch. // status update to mesos. also attempts to reset the suicide watch.
// Assumes that the caller is locking around pod and task state. // Assumes that the caller is locking around pod and task state.
func (k *KubernetesExecutor) removePodTask(driver bindings.ExecutorDriver, tid, reason string, state mesos.TaskState) { func (k *KubernetesMesosExecutor) removePodTask(driver bindings.ExecutorDriver, tid, reason string, state mesos.TaskState) {
task, ok := k.tasks[tid] task, ok := k.tasks[tid]
if !ok { if !ok {
log.V(1).Infof("Failed to remove task, unknown task %v\n", tid) log.V(1).Infof("Failed to remove task, unknown task %v\n", tid)
@@ -770,7 +770,7 @@ func (k *KubernetesExecutor) removePodTask(driver bindings.ExecutorDriver, tid,
} }
// FrameworkMessage is called when the framework sends some message to the executor // FrameworkMessage is called when the framework sends some message to the executor
func (k *KubernetesExecutor) FrameworkMessage(driver bindings.ExecutorDriver, message string) { func (k *KubernetesMesosExecutor) FrameworkMessage(driver bindings.ExecutorDriver, message string) {
if k.isDone() { if k.isDone() {
return return
} }
@@ -798,14 +798,14 @@ func (k *KubernetesExecutor) FrameworkMessage(driver bindings.ExecutorDriver, me
} }
// Shutdown is called when the executor receives a shutdown request. // Shutdown is called when the executor receives a shutdown request.
func (k *KubernetesExecutor) Shutdown(driver bindings.ExecutorDriver) { func (k *KubernetesMesosExecutor) Shutdown(driver bindings.ExecutorDriver) {
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
k.doShutdown(driver) k.doShutdown(driver)
} }
// assumes that caller has obtained state lock // assumes that caller has obtained state lock
func (k *KubernetesExecutor) doShutdown(driver bindings.ExecutorDriver) { func (k *KubernetesMesosExecutor) doShutdown(driver bindings.ExecutorDriver) {
defer func() { defer func() {
log.Errorf("exiting with unclean shutdown: %v", recover()) log.Errorf("exiting with unclean shutdown: %v", recover())
if k.exitFunc != nil { if k.exitFunc != nil {
@@ -859,7 +859,7 @@ func (k *KubernetesExecutor) doShutdown(driver bindings.ExecutorDriver) {
} }
// Destroy existing k8s containers // Destroy existing k8s containers
func (k *KubernetesExecutor) killKubeletContainers() { func (k *KubernetesMesosExecutor) killKubeletContainers() {
if containers, err := dockertools.GetKubeletDockerContainers(k.dockerClient, true); err == nil { if containers, err := dockertools.GetKubeletDockerContainers(k.dockerClient, true); err == nil {
opts := docker.RemoveContainerOptions{ opts := docker.RemoveContainerOptions{
RemoveVolumes: true, RemoveVolumes: true,
@@ -878,7 +878,7 @@ func (k *KubernetesExecutor) killKubeletContainers() {
} }
// Error is called when some error happens. // Error is called when some error happens.
func (k *KubernetesExecutor) Error(driver bindings.ExecutorDriver, message string) { func (k *KubernetesMesosExecutor) Error(driver bindings.ExecutorDriver, message string) {
log.Errorln(message) log.Errorln(message)
} }
@@ -890,7 +890,7 @@ func newStatus(taskId *mesos.TaskID, state mesos.TaskState, message string) *mes
} }
} }
func (k *KubernetesExecutor) sendStatus(driver bindings.ExecutorDriver, status *mesos.TaskStatus) { func (k *KubernetesMesosExecutor) sendStatus(driver bindings.ExecutorDriver, status *mesos.TaskStatus) {
select { select {
case <-k.terminate: case <-k.terminate:
default: default:
@@ -898,7 +898,7 @@ func (k *KubernetesExecutor) sendStatus(driver bindings.ExecutorDriver, status *
} }
} }
func (k *KubernetesExecutor) sendFrameworkMessage(driver bindings.ExecutorDriver, msg string) { func (k *KubernetesMesosExecutor) sendFrameworkMessage(driver bindings.ExecutorDriver, msg string) {
select { select {
case <-k.terminate: case <-k.terminate:
default: default:
@@ -906,7 +906,7 @@ func (k *KubernetesExecutor) sendFrameworkMessage(driver bindings.ExecutorDriver
} }
} }
func (k *KubernetesExecutor) sendLoop() { func (k *KubernetesMesosExecutor) sendLoop() {
defer log.V(1).Info("sender loop exiting") defer log.V(1).Info("sender loop exiting")
for { for {
select { select {

2
contrib/mesos/pkg/executor/mock_test.go
View File

@@ -66,7 +66,7 @@ func (m *MockExecutorDriver) SendFrameworkMessage(msg string) (mesosproto.Status
return args.Get(0).(mesosproto.Status), args.Error(1) return args.Get(0).(mesosproto.Status), args.Error(1)
} }
func NewTestKubernetesExecutor() (*KubernetesExecutor, chan kubetypes.PodUpdate) { func NewTestKubernetesExecutor() (*KubernetesMesosExecutor, chan kubetypes.PodUpdate) {
updates := make(chan kubetypes.PodUpdate, 1024) updates := make(chan kubetypes.PodUpdate, 1024)
return New(Config{ return New(Config{
Docker: dockertools.ConnectToDockerOrDie("fake://"), Docker: dockertools.ConnectToDockerOrDie("fake://"),