Skip to content

K
k3s
Commit a999c0fb authored by k8s-merge-robot's avatar k8s-merge-robot
Browse files
Options

Merge pull request #13943 from mesosphere/static-pods

Auto commit by PR queue bot
parents b250006e 3d3577b9
Hide whitespace changes
Inline Side-by-side
Showing with 908 additions and 458 deletions
cluster/mesos/docker/deploy-addons.sh
View file @ a999c0fb
...@@ -52,8 +52,9 @@ function deploy_ui { ...@@ -52,8 +52,9 @@ function deploy_ui {
"${kubectl}" create -f "${KUBE_ROOT}/cluster/addons/kube-ui/kube-ui-svc.yaml" "${kubectl}" create -f "${KUBE_ROOT}/cluster/addons/kube-ui/kube-ui-svc.yaml"
} }
# create the kube-system namespace # create the kube-system and static-pods namespaces
"${kubectl}" create -f "${KUBE_ROOT}/cluster/mesos/docker/kube-system-ns.yaml" "${kubectl}" create -f "${KUBE_ROOT}/cluster/mesos/docker/kube-system-ns.yaml"
"${kubectl}" create -f "${KUBE_ROOT}/cluster/mesos/docker/static-pods-ns.yaml"
if [ "${ENABLE_CLUSTER_DNS}" == true ]; then if [ "${ENABLE_CLUSTER_DNS}" == true ]; then
cluster::mesos::docker::run_in_temp_dir 'k8sm-dns' 'deploy_dns' cluster::mesos::docker::run_in_temp_dir 'k8sm-dns' 'deploy_dns'
......
cluster/mesos/docker/docker-compose.yml
View file @ a999c0fb
...@@ -142,6 +142,7 @@ scheduler: ...@@ -142,6 +142,7 @@ scheduler:
--cluster-domain=cluster.local --cluster-domain=cluster.local
--mesos-executor-cpus=1.0 --mesos-executor-cpus=1.0
--mesos-sandbox-overlay=/opt/sandbox-overlay.tar.gz --mesos-sandbox-overlay=/opt/sandbox-overlay.tar.gz
--static-pods-config=/opt/static-pods
--v=4 --v=4
--executor-logv=4 --executor-logv=4
--profiling=true --profiling=true
...@@ -149,6 +150,8 @@ scheduler: ...@@ -149,6 +150,8 @@ scheduler:
- etcd - etcd
- mesosmaster1 - mesosmaster1
- apiserver - apiserver
volumes:
- ./static-pod.json:/opt/static-pods/static-pod.json
keygen: keygen:
image: mesosphere/kubernetes-mesos-keygen image: mesosphere/kubernetes-mesos-keygen
command: command:
......
cluster/mesos/docker/static-pod.json 0 → 100644
View file @ a999c0fb
{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"name": "busybox",
"namespace": "static-pods"
},
"spec": {
"containers": [
{
"image": "busybox",
"command": [
"sh",
"-c",
"exec tail -f /dev/null"
],
"imagePullPolicy": "IfNotPresent",
"name": "busybox"
}
],
"restartPolicy": "Always"
}
}
cluster/mesos/docker/static-pods-ns.yaml 0 → 100644
View file @ a999c0fb
---
apiVersion: v1
kind: Namespace
metadata:
name: static-pods
labels:
name: static-pods
contrib/mesos/pkg/archive/zip.go deleted 100644 → 0
View file @ b250006e
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 archive
import (
"archive/zip"
"bytes"
"fmt"
"io"
"os"
"path"
"path/filepath"
)
// ZipWalker returns a filepath.WalkFunc that adds every filesystem node
// to the given *zip.Writer.
func ZipWalker(zw *zip.Writer) filepath.WalkFunc {
var base string
return func(path string, info os.FileInfo, err error) error {
if base == "" {
base = path
}
header, err := zip.FileInfoHeader(info)
if err != nil {
return err
}
if header.Name, err = filepath.Rel(base, path); err != nil {
return err
} else if info.IsDir() {
header.Name = header.Name + string(filepath.Separator)
} else {
header.Method = zip.Deflate
}
w, err := zw.CreateHeader(header)
if err != nil {
return err
}
if info.IsDir() {
return nil
}
f, err := os.Open(path)
if err != nil {
return err
}
_, err = io.Copy(w, f)
f.Close()
return err
}
}
// Create a zip of all files in a directory recursively, return a byte array and
// the number of files archived.
func ZipDir(path string) ([]byte, []string, error) {
var buf bytes.Buffer
zw := zip.NewWriter(&buf)
zipWalker := ZipWalker(zw)
paths := []string{}
err := filepath.Walk(path, filepath.WalkFunc(func(path string, info os.FileInfo, err error) error {
if !info.IsDir() {
paths = append(paths, path)
}
return zipWalker(path, info, err)
}))
if err != nil {
return nil, nil, err
} else if err = zw.Close(); err != nil {
return nil, nil, err
}
return buf.Bytes(), paths, nil
}
// UnzipDir unzips all files from a given zip byte array into a given directory.
// The directory is created if it does not exist yet.
func UnzipDir(data []byte, destPath string) error {
// open zip
zr, err := zip.NewReader(bytes.NewReader(data), int64(len(data)))
if err != nil {
return fmt.Errorf("Unzip archive read error: %v", err)
}
for _, file := range zr.File {
// skip directories
if file.FileInfo().IsDir() {
continue
}
// open file
rc, err := file.Open()
defer rc.Close()
if err != nil {
return fmt.Errorf("Unzip file read error: %v", err)
}
// make sure the directory of the file exists, otherwise create
destPath := filepath.Clean(filepath.Join(destPath, file.Name))
destBasedir := path.Dir(destPath)
err = os.MkdirAll(destBasedir, 0755)
if err != nil {
return fmt.Errorf("Unzip mkdir error: %v", err)
}
// create file
f, err := os.Create(destPath)
if err != nil {
return fmt.Errorf("Unzip file creation error: %v", err)
}
defer f.Close()
// write file
if _, err := io.Copy(f, rc); err != nil {
return fmt.Errorf("Unzip file write error: %v", err)
}
}
return nil
}
contrib/mesos/pkg/archive/zip_test.go deleted 100644 → 0
View file @ b250006e
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 archive
import (
"archive/zip"
"bytes"
"io/ioutil"
"os"
"path/filepath"
"testing"
)
func TestZipWalker(t *testing.T) {
dir, err := ioutil.TempDir(os.TempDir(), "")
if err != nil {
t.Fatal(err)
}
tree := map[string]string{"a/b/c": "12345", "a/b/d": "54321", "a/e": "00000"}
for path, content := range tree {
path = filepath.Join(dir, path)
if err := os.MkdirAll(filepath.Dir(path), os.ModeTemporary|0700); err != nil {
t.Fatal(err)
} else if err = ioutil.WriteFile(path, []byte(content), 0700); err != nil {
t.Fatal(err)
}
}
var buf bytes.Buffer
zw := zip.NewWriter(&buf)
if err := filepath.Walk(dir, ZipWalker(zw)); err != nil {
t.Fatal(err)
} else if err = zw.Close(); err != nil {
t.Fatal(err)
}
zr, err := zip.NewReader(bytes.NewReader(buf.Bytes()), int64(buf.Len()))
if err != nil {
t.Fatal(err)
}
for _, file := range zr.File {
if rc, err := file.Open(); err != nil {
t.Fatal(err)
} else if got, err := ioutil.ReadAll(rc); err != nil {
t.Error(err)
} else if want := []byte(tree[file.Name]); !bytes.Equal(got, want) {
t.Errorf("%s\ngot: %s\nwant: %s", file.Name, got, want)
}
}
}
contrib/mesos/pkg/executor/executor.go
View file @ a999c0fb
...@@ -30,9 +30,9 @@ import ( ...@@ -30,9 +30,9 @@ import (
bindings "github.com/mesos/mesos-go/executor" bindings "github.com/mesos/mesos-go/executor"
mesos "github.com/mesos/mesos-go/mesosproto" mesos "github.com/mesos/mesos-go/mesosproto"
mutil "github.com/mesos/mesos-go/mesosutil" mutil "github.com/mesos/mesos-go/mesosutil"
"k8s.io/kubernetes/contrib/mesos/pkg/archive"
"k8s.io/kubernetes/contrib/mesos/pkg/executor/messages" "k8s.io/kubernetes/contrib/mesos/pkg/executor/messages"
"k8s.io/kubernetes/contrib/mesos/pkg/node" "k8s.io/kubernetes/contrib/mesos/pkg/node"
"k8s.io/kubernetes/contrib/mesos/pkg/podutil"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/meta" "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/meta"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
unversionedapi "k8s.io/kubernetes/pkg/api/unversioned" unversionedapi "k8s.io/kubernetes/pkg/api/unversioned"
...@@ -205,11 +205,19 @@ func (k *Executor) isDone() bool { ...@@ -205,11 +205,19 @@ func (k *Executor) isDone() bool {
} }
} }
// sendPodUpdate assumes that caller is holding state lock; returns true when update is sent otherwise false // sendPodsSnapshot assumes that caller is holding state lock; returns true when update is sent otherwise false
func (k *Executor) sendPodUpdate(u *kubetypes.PodUpdate) bool { func (k *Executor) sendPodsSnapshot() bool {
if k.isDone() { if k.isDone() {
return false return false
} }
snapshot := make([]*api.Pod, 0, len(k.pods))
for _, v := range k.pods {
snapshot = append(snapshot, v)
}
u := &kubetypes.PodUpdate{
Op: kubetypes.SET,
Pods: snapshot,
}
k.updateChan <- *u k.updateChan <- *u
return true return true
} }
...@@ -227,7 +235,10 @@ func (k *Executor) Registered(driver bindings.ExecutorDriver, ...@@ -227,7 +235,10 @@ func (k *Executor) Registered(driver bindings.ExecutorDriver,
} }
if executorInfo != nil && executorInfo.Data != nil { if executorInfo != nil && executorInfo.Data != nil {
k.initializeStaticPodsSource(executorInfo.Data) err := k.initializeStaticPodsSource(slaveInfo.GetHostname(), executorInfo.Data)
if err != nil {
log.Errorf("failed to initialize static pod configuration: %v", err)
}
} }
if slaveInfo != nil { if slaveInfo != nil {
...@@ -240,10 +251,7 @@ func (k *Executor) Registered(driver bindings.ExecutorDriver, ...@@ -240,10 +251,7 @@ func (k *Executor) Registered(driver bindings.ExecutorDriver,
// emit an empty update to allow the mesos "source" to be marked as seen // emit an empty update to allow the mesos "source" to be marked as seen
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
k.sendPodUpdate(&kubetypes.PodUpdate{ k.sendPodsSnapshot()
Pods: []*api.Pod{},
Op: kubetypes.SET,
})
if slaveInfo != nil && k.nodeInfos != nil { if slaveInfo != nil && k.nodeInfos != nil {
k.nodeInfos <- nodeInfo(slaveInfo, executorInfo) // leave it behind the upper lock to avoid panics k.nodeInfos <- nodeInfo(slaveInfo, executorInfo) // leave it behind the upper lock to avoid panics
...@@ -280,13 +288,15 @@ func (k *Executor) Reregistered(driver bindings.ExecutorDriver, slaveInfo *mesos ...@@ -280,13 +288,15 @@ func (k *Executor) Reregistered(driver bindings.ExecutorDriver, slaveInfo *mesos
} }
// initializeStaticPodsSource unzips the data slice into the static-pods directory // initializeStaticPodsSource unzips the data slice into the static-pods directory
func (k *Executor) initializeStaticPodsSource(data []byte) { func (k *Executor) initializeStaticPodsSource(hostname string, data []byte) error {
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) // annotate the pod with BindingHostKey so that the scheduler will ignore the pod
if err != nil { // once it appears in the pod registry. the stock kubelet sets the pod host in order
log.Errorf("Failed to extract static pod config: %v", err) // to accomplish the same; we do this because the k8sm scheduler works differently.
return annotator := podutil.Annotator(map[string]string{
} meta.BindingHostKey: hostname,
})
return podutil.WriteToDir(annotator.Do(podutil.Gunzip(data)), k.staticPodsConfigPath)
} }
// Disconnected is called when the executor is disconnected from the slave. // Disconnected is called when the executor is disconnected from the slave.
...@@ -390,10 +400,7 @@ func (k *Executor) handleChangedApiserverPod(pod *api.Pod) { ...@@ -390,10 +400,7 @@ func (k *Executor) handleChangedApiserverPod(pod *api.Pod) {
oldPod.DeletionTimestamp = pod.DeletionTimestamp oldPod.DeletionTimestamp = pod.DeletionTimestamp
oldPod.DeletionGracePeriodSeconds = pod.DeletionGracePeriodSeconds oldPod.DeletionGracePeriodSeconds = pod.DeletionGracePeriodSeconds
k.sendPodUpdate(&kubetypes.PodUpdate{ k.sendPodsSnapshot()
Op: kubetypes.UPDATE,
Pods: []*api.Pod{oldPod},
})
} }
} }
} }
...@@ -557,18 +564,15 @@ func (k *Executor) launchTask(driver bindings.ExecutorDriver, taskId string, pod ...@@ -557,18 +564,15 @@ func (k *Executor) launchTask(driver bindings.ExecutorDriver, taskId string, pod
//TODO(jdef) check for duplicate pod name, if found send TASK_ERROR //TODO(jdef) check for duplicate pod name, if found send TASK_ERROR
// send the new pod to the kubelet which will spin it up // send the new pod to the kubelet which will spin it up
ok := k.sendPodUpdate(&kubetypes.PodUpdate{ task.podName = podFullName
Op: kubetypes.ADD, k.pods[podFullName] = pod
Pods: []*api.Pod{pod}, ok := k.sendPodsSnapshot()
})
if !ok { if !ok {
task.podName = ""
delete(k.pods, podFullName)
return // executor is terminating, cancel launch return // executor is terminating, cancel launch
} }
// mark task as sent by setting the podName and register the sent pod
task.podName = podFullName
k.pods[podFullName] = pod
// From here on, we need to delete containers associated with the task upon // From here on, we need to delete containers associated with the task upon
// it going into a terminal state. // it going into a terminal state.
...@@ -752,7 +756,7 @@ func (k *Executor) removePodTask(driver bindings.ExecutorDriver, tid, reason str ...@@ -752,7 +756,7 @@ func (k *Executor) removePodTask(driver bindings.ExecutorDriver, tid, reason str
k.resetSuicideWatch(driver) k.resetSuicideWatch(driver)
pid := task.podName pid := task.podName
pod, found := k.pods[pid] _, found := k.pods[pid]
if !found { if !found {
log.Warningf("Cannot remove unknown pod %v for task %v", pid, tid) log.Warningf("Cannot remove unknown pod %v for task %v", pid, tid)
} else { } else {
...@@ -760,10 +764,7 @@ func (k *Executor) removePodTask(driver bindings.ExecutorDriver, tid, reason str ...@@ -760,10 +764,7 @@ func (k *Executor) removePodTask(driver bindings.ExecutorDriver, tid, reason str
delete(k.pods, pid) delete(k.pods, pid)
// tell the kubelet to remove the pod // tell the kubelet to remove the pod
k.sendPodUpdate(&kubetypes.PodUpdate{ k.sendPodsSnapshot()
Op: kubetypes.REMOVE,
Pods: []*api.Pod{pod},
})
} }
// TODO(jdef): ensure that the update propagates, perhaps return a signal chan? // TODO(jdef): ensure that the update propagates, perhaps return a signal chan?
k.sendStatus(driver, newStatus(mutil.NewTaskID(tid), state, reason)) k.sendStatus(driver, newStatus(mutil.NewTaskID(tid), state, reason))
......
contrib/mesos/pkg/executor/executor_test.go
View file @ a999c0fb
...@@ -17,14 +17,13 @@ limitations under the License. ...@@ -17,14 +17,13 @@ limitations under the License.
package executor package executor
import ( import (
"archive/zip"
"bytes"
"fmt" "fmt"
"io/ioutil" "io/ioutil"
"net" "net"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"os" "os"
"path/filepath"
"reflect" "reflect"
"sync" "sync"
"sync/atomic" "sync/atomic"
...@@ -33,6 +32,7 @@ import ( ...@@ -33,6 +32,7 @@ import (
assertext "k8s.io/kubernetes/contrib/mesos/pkg/assert" assertext "k8s.io/kubernetes/contrib/mesos/pkg/assert"
"k8s.io/kubernetes/contrib/mesos/pkg/executor/messages" "k8s.io/kubernetes/contrib/mesos/pkg/executor/messages"
"k8s.io/kubernetes/contrib/mesos/pkg/podutil"
kmruntime "k8s.io/kubernetes/contrib/mesos/pkg/runtime" kmruntime "k8s.io/kubernetes/contrib/mesos/pkg/runtime"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask" "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/podtask"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
...@@ -41,7 +41,6 @@ import ( ...@@ -41,7 +41,6 @@ import (
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
client "k8s.io/kubernetes/pkg/client/unversioned" client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/kubelet" "k8s.io/kubernetes/pkg/kubelet"
kconfig "k8s.io/kubernetes/pkg/kubelet/config"
"k8s.io/kubernetes/pkg/kubelet/dockertools" "k8s.io/kubernetes/pkg/kubelet/dockertools"
kubetypes "k8s.io/kubernetes/pkg/kubelet/types" kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
...@@ -49,7 +48,6 @@ import ( ...@@ -49,7 +48,6 @@ import (
"k8s.io/kubernetes/pkg/watch" "k8s.io/kubernetes/pkg/watch"
"github.com/mesos/mesos-go/mesosproto" "github.com/mesos/mesos-go/mesosproto"
"github.com/mesos/mesos-go/mesosutil"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock" "github.com/stretchr/testify/mock"
) )
...@@ -246,123 +244,79 @@ func TestExecutorLaunchAndKillTask(t *testing.T) { ...@@ -246,123 +244,79 @@ func TestExecutorLaunchAndKillTask(t *testing.T) {
// TestExecutorStaticPods test that the ExecutorInfo.data is parsed // TestExecutorStaticPods test that the ExecutorInfo.data is parsed
// as a zip archive with pod definitions. // as a zip archive with pod definitions.
func TestExecutorStaticPods(t *testing.T) { func TestExecutorInitializeStaticPodsSource(t *testing.T) {
// create some zip with static pod definition // create some zip with static pod definition
var buf bytes.Buffer givenPodsDir, err := ioutil.TempDir("/tmp", "executor-givenpods")
zw := zip.NewWriter(&buf) assert.NoError(t, err)
createStaticPodFile := func(fileName, id, name string) { defer os.RemoveAll(givenPodsDir)
w, err := zw.Create(fileName)
assert.NoError(t, err) var wg sync.WaitGroup
spod := `{ reportErrors := func(errCh <-chan error) {
"apiVersion": "v1", wg.Add(1)
"kind": "Pod", go func() {
"metadata": { defer wg.Done()
"name": "%v", for err := range errCh {
"labels": { "name": "foo", "cluster": "bar" } t.Error(err)
},
"spec": {
"containers": [{
"name": "%v",
"image": "library/nginx",
"ports": [{ "containerPort": 80, "name": "http" }],
"livenessProbe": {
"enabled": true,
"type": "http",
"initialDelaySeconds": 30,
"httpGet": { "path": "/", "port": 80 }
} }
}] }()
} }
}`
_, err = w.Write([]byte(fmt.Sprintf(spod, id, name))) createStaticPodFile := func(fileName, name string) {
spod := `{
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"name": "%v",
"namespace": "staticpods",
"labels": { "name": "foo", "cluster": "bar" }
},
"spec": {
"containers": [{
"name": "%v",
"image": "library/nginx",
"ports": [{ "containerPort": 80, "name": "http" }]
}]
}
}`
destfile := filepath.Join(givenPodsDir, fileName)
err = os.MkdirAll(filepath.Dir(destfile), 0770)
assert.NoError(t, err)
err = ioutil.WriteFile(destfile, []byte(fmt.Sprintf(spod, name, name)), 0660)
assert.NoError(t, err) assert.NoError(t, err)
} }
createStaticPodFile("spod.json", "spod-id-01", "spod-01")
createStaticPodFile("spod2.json", "spod-id-02", "spod-02")
createStaticPodFile("dir/spod.json", "spod-id-03", "spod-03") // same file name as first one to check for overwriting
expectedStaticPodsNum := 2 // subdirectories are ignored by FileSource, hence only 2 createStaticPodFile("spod.json", "spod-01")
createStaticPodFile("spod2.json", "spod-02")
createStaticPodFile("dir/spod.json", "spod-03") // same file name as first one to check for overwriting
staticpods, errs := podutil.ReadFromDir(givenPodsDir)
reportErrors(errs)
err := zw.Close() gzipped, err := podutil.Gzip(staticpods)
assert.NoError(t, err) assert.NoError(t, err)
// create fake apiserver expectedStaticPodsNum := 2 // subdirectories are ignored by FileSource, hence only 2
testApiServer := NewTestServer(t, api.NamespaceDefault, nil)
defer testApiServer.server.Close()
// temporary directory which is normally located in the executor sandbox // temporary directory which is normally located in the executor sandbox
staticPodsConfigPath, err := ioutil.TempDir("/tmp", "executor-k8sm-archive") staticPodsConfigPath, err := ioutil.TempDir("/tmp", "executor-k8sm-archive")
assert.NoError(t, err) assert.NoError(t, err)
defer os.RemoveAll(staticPodsConfigPath) defer os.RemoveAll(staticPodsConfigPath)
mockDriver := &MockExecutorDriver{} executor := &Executor{
config := Config{ staticPodsConfigPath: staticPodsConfigPath,
Docker: dockertools.ConnectToDockerOrDie("fake://"),
Updates: make(chan kubetypes.PodUpdate, 1), // allow kube-executor source to proceed past init
NodeInfos: make(chan NodeInfo, 1),
APIClient: client.NewOrDie(&client.Config{
Host: testApiServer.server.URL,
Version: testapi.Default.Version(),
}),
PodStatusFunc: func(pod *api.Pod) (*api.PodStatus, error) {
return &api.PodStatus{
ContainerStatuses: []api.ContainerStatus{
{
Name: "foo",
State: api.ContainerState{
Running: &api.ContainerStateRunning{},
},
},
},
Phase: api.PodRunning,
}, nil
},
StaticPodsConfigPath: staticPodsConfigPath,
PodLW: &NewMockPodsListWatch(api.PodList{}).ListWatch,
} }
executor := New(config)
// register static pod source // extract the pods into staticPodsConfigPath
hostname := "h1" hostname := "h1"
fileSourceUpdates := make(chan interface{}, 1024) err = executor.initializeStaticPodsSource(hostname, gzipped)
kconfig.NewSourceFile(staticPodsConfigPath, hostname, 1*time.Second, fileSourceUpdates) assert.NoError(t, err)
// create ExecutorInfo with static pod zip in data field actualpods, errs := podutil.ReadFromDir(staticPodsConfigPath)
executorInfo := mesosutil.NewExecutorInfo( reportErrors(errs)
mesosutil.NewExecutorID("ex1"),
mesosutil.NewCommandInfo("k8sm-executor"),
)
executorInfo.Data = buf.Bytes()
// start the executor with the static pod data list := podutil.List(actualpods)
executor.Init(mockDriver) assert.NotNil(t, list)
executor.Registered(mockDriver, executorInfo, nil, nil) assert.Equal(t, expectedStaticPodsNum, len(list.Items))
wg.Wait()
// wait for static pod to start
seenPods := map[string]struct{}{}
timeout := time.After(util.ForeverTestTimeout)
defer mockDriver.AssertExpectations(t)
for {
// filter by PodUpdate type
select {
case <-timeout:
t.Fatalf("Executor should send pod updates for %v pods, only saw %v", expectedStaticPodsNum, len(seenPods))
case update, ok := <-fileSourceUpdates:
if !ok {
return
}
podUpdate, ok := update.(kubetypes.PodUpdate)
if !ok {
continue
}
for _, pod := range podUpdate.Pods {
seenPods[pod.Name] = struct{}{}
}
if len(seenPods) == expectedStaticPodsNum {
return
}
}
}
} }
// TestExecutorFrameworkMessage ensures that the executor is able to // TestExecutorFrameworkMessage ensures that the executor is able to
......
contrib/mesos/pkg/executor/service/podsource.go 0 → 100644
View file @ a999c0fb
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 service
import (
"fmt"
log "github.com/golang/glog"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/client/cache"
kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
)
const (
// if we don't use this source then the kubelet will do funny, mirror things. we alias
// this here for convenience. see the docs for sourceMesos for additional explanation.
// @see ConfigSourceAnnotationKey
mesosSource = kubetypes.ApiserverSource
)
// sourceMesos merges pods from mesos, and mirror pods from the apiserver. why?
// (a) can't have two sources with the same name;
// (b) all sources, other than ApiserverSource are considered static/mirror
// sources, and;
// (c) kubelet wants to see mirror pods reflected in a non-static source.
//
// Mesos pods must appear to come from apiserver due to (b), while reflected
// static pods (mirror pods) must appear to come from apiserver due to (c).
//
// The only option I could think of was creating a source that merges the pod
// streams. I don't like it. But I could think of anything else, other than
// starting to hack up the kubelet's understanding of mirror/static pod
// sources (ouch!)
type sourceMesos struct {
sourceFinished chan struct{} // sourceFinished closes when mergeAndForward exits
out chan<- interface{} // out is the sink for merged pod snapshots
mirrorPods chan []*api.Pod // mirrorPods communicates snapshots of the current set of mirror pods
execUpdates <-chan kubetypes.PodUpdate // execUpdates receives snapshots of the current set of mesos pods
}
// newSourceMesos creates a pod config source that merges pod updates from
// mesos (via execUpdates), and mirror pod updates from the apiserver (via
// podWatch) writing the merged update stream to the out chan. It is expected
// that execUpdates will only ever contain SET operations. The source takes
// ownership of the sourceFinished chan, closing it when the source terminates.
// Source termination happens when the execUpdates chan is closed and fully
// drained of updates.
func newSourceMesos(
sourceFinished chan struct{},
execUpdates <-chan kubetypes.PodUpdate,
out chan<- interface{},
podWatch *cache.ListWatch,
) {
source := &sourceMesos{
sourceFinished: sourceFinished,
mirrorPods: make(chan []*api.Pod),
execUpdates: execUpdates,
out: out,
}
// reflect changes from the watch into a chan, filtered to include only mirror pods (have an ConfigMirrorAnnotationKey attr)
cache.NewReflector(podWatch, &api.Pod{}, cache.NewUndeltaStore(source.send, cache.MetaNamespaceKeyFunc), 0).RunUntil(sourceFinished)
go source.mergeAndForward()
}
func (source *sourceMesos) send(objs []interface{}) {
var mirrors []*api.Pod
for _, o := range objs {
p := o.(*api.Pod)
if _, ok := p.Annotations[kubetypes.ConfigMirrorAnnotationKey]; ok {
mirrors = append(mirrors, p)
}
}
select {
case <-source.sourceFinished:
case source.mirrorPods <- mirrors:
}
}
func (source *sourceMesos) mergeAndForward() {
// execUpdates will be closed by the executor on shutdown
defer close(source.sourceFinished)
var (
mirrors = []*api.Pod{}
pods = []*api.Pod{}
)
eventLoop:
for {
select {
case m := <-source.mirrorPods:
mirrors = m[:]
u := kubetypes.PodUpdate{
Op: kubetypes.SET,
Pods: append(m, pods...),
Source: mesosSource,
}
log.V(3).Infof("mirror update, sending snapshot of size %d", len(u.Pods))
source.out <- u
case u, ok := <-source.execUpdates:
if !ok {
break eventLoop
}
if u.Op != kubetypes.SET {
panic(fmt.Sprintf("unexpected Op type: %v", u.Op))
}
pods = u.Pods[:]
u.Pods = append(u.Pods, mirrors...)
u.Source = mesosSource
log.V(3).Infof("pods update, sending snapshot of size %d", len(u.Pods))
source.out <- u
}
}
log.V(2).Infoln("mesos pod source terminating normally")
}
contrib/mesos/pkg/executor/service/service.go
View file @ a999c0fb
...@@ -42,12 +42,6 @@ import ( ...@@ -42,12 +42,6 @@ import (
kubetypes "k8s.io/kubernetes/pkg/kubelet/types" kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
) )
const (
// if we don't use this source then the kubelet will do funny, mirror things.
// @see ConfigSourceAnnotationKey
MESOS_CFG_SOURCE = kubetypes.ApiserverSource
)
type KubeletExecutorServer struct { type KubeletExecutorServer struct {
*app.KubeletServer *app.KubeletServer
SuicideTimeout time.Duration SuicideTimeout time.Duration
...@@ -79,8 +73,14 @@ func (s *KubeletExecutorServer) AddFlags(fs *pflag.FlagSet) { ...@@ -79,8 +73,14 @@ func (s *KubeletExecutorServer) AddFlags(fs *pflag.FlagSet) {
fs.DurationVar(&s.LaunchGracePeriod, "mesos-launch-grace-period", s.LaunchGracePeriod, "Launch grace period after which launching tasks will be cancelled. Zero disables launch cancellation.") fs.DurationVar(&s.LaunchGracePeriod, "mesos-launch-grace-period", s.LaunchGracePeriod, "Launch grace period after which launching tasks will be cancelled. Zero disables launch cancellation.")
} }
func (s *KubeletExecutorServer) runExecutor(execUpdates chan<- kubetypes.PodUpdate, nodeInfos chan<- executor.NodeInfo, kubeletFinished <-chan struct{}, func (s *KubeletExecutorServer) runExecutor(
staticPodsConfigPath string, apiclient *client.Client) error { execUpdates chan<- kubetypes.PodUpdate,
nodeInfos chan<- executor.NodeInfo,
kubeletFinished <-chan struct{},
staticPodsConfigPath string,
apiclient *client.Client,
podLW *cache.ListWatch,
) error {
exec := executor.New(executor.Config{ exec := executor.New(executor.Config{
Updates: execUpdates, Updates: execUpdates,
APIClient: apiclient, APIClient: apiclient,
...@@ -111,10 +111,8 @@ func (s *KubeletExecutorServer) runExecutor(execUpdates chan<- kubetypes.PodUpda ...@@ -111,10 +111,8 @@ func (s *KubeletExecutorServer) runExecutor(execUpdates chan<- kubetypes.PodUpda
return status, nil return status, nil
}, },
StaticPodsConfigPath: staticPodsConfigPath, StaticPodsConfigPath: staticPodsConfigPath,
PodLW: cache.NewListWatchFromClient(apiclient, "pods", api.NamespaceAll, PodLW: podLW,
fields.OneTermEqualSelector(client.PodHost, s.HostnameOverride), NodeInfos: nodeInfos,
),
NodeInfos: nodeInfos,
}) })
// initialize driver and initialize the executor with it // initialize driver and initialize the executor with it
...@@ -141,8 +139,14 @@ func (s *KubeletExecutorServer) runExecutor(execUpdates chan<- kubetypes.PodUpda ...@@ -141,8 +139,14 @@ func (s *KubeletExecutorServer) runExecutor(execUpdates chan<- kubetypes.PodUpda
return nil return nil
} }
func (s *KubeletExecutorServer) runKubelet(execUpdates <-chan kubetypes.PodUpdate, nodeInfos <-chan executor.NodeInfo, kubeletDone chan<- struct{}, func (s *KubeletExecutorServer) runKubelet(
staticPodsConfigPath string, apiclient *client.Client) error { execUpdates <-chan kubetypes.PodUpdate,
nodeInfos <-chan executor.NodeInfo,
kubeletDone chan<- struct{},
staticPodsConfigPath string,
apiclient *client.Client,
podLW *cache.ListWatch,
) error {
kcfg, err := s.UnsecuredKubeletConfig() kcfg, err := s.UnsecuredKubeletConfig()
if err == nil { if err == nil {
// apply Messo specific settings // apply Messo specific settings
...@@ -199,17 +203,8 @@ func (s *KubeletExecutorServer) runKubelet(execUpdates <-chan kubetypes.PodUpdat ...@@ -199,17 +203,8 @@ func (s *KubeletExecutorServer) runKubelet(execUpdates <-chan kubetypes.PodUpdat
} }
}() }()
// create main pod source // create main pod source, it will close executorDone when the executor updates stop flowing
updates := kcfg.PodConfig.Channel(MESOS_CFG_SOURCE) newSourceMesos(executorDone, execUpdates, kcfg.PodConfig.Channel(mesosSource), podLW)
go func() {
// execUpdates will be closed by the executor on shutdown
defer close(executorDone)
for u := range execUpdates {
u.Source = MESOS_CFG_SOURCE
updates <- u
}
}()
// create static-pods directory file source // create static-pods directory file source
log.V(2).Infof("initializing static pods source factory, configured at path %q", staticPodsConfigPath) log.V(2).Infof("initializing static pods source factory, configured at path %q", staticPodsConfigPath)
...@@ -257,14 +252,18 @@ func (s *KubeletExecutorServer) Run(hks hyperkube.Interface, _ []string) error { ...@@ -257,14 +252,18 @@ func (s *KubeletExecutorServer) Run(hks hyperkube.Interface, _ []string) error {
return fmt.Errorf("cannot create API client: %v", err) return fmt.Errorf("cannot create API client: %v", err)
} }
pw := cache.NewListWatchFromClient(apiclient, "pods", api.NamespaceAll,
fields.OneTermEqualSelector(client.PodHost, s.HostnameOverride),
)
// start executor // start executor
err = s.runExecutor(execUpdates, nodeInfos, kubeletFinished, staticPodsConfigPath, apiclient) err = s.runExecutor(execUpdates, nodeInfos, kubeletFinished, staticPodsConfigPath, apiclient, pw)
if err != nil { if err != nil {
return err return err
} }
// start kubelet, blocking // start kubelet, blocking
return s.runKubelet(execUpdates, nodeInfos, kubeletFinished, staticPodsConfigPath, apiclient) return s.runKubelet(execUpdates, nodeInfos, kubeletFinished, staticPodsConfigPath, apiclient, pw)
} }
func defaultBindingAddress() string { func defaultBindingAddress() string {
......
contrib/mesos/pkg/archive/doc.go → contrib/mesos/pkg/podutil/doc.go
View file @ a999c0fb
...@@ -14,6 +14,6 @@ See the License for the specific language governing permissions and ...@@ -14,6 +14,6 @@ See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
// Package archive provides utilities to archive and unarchive filesystem // podutil contains utilities for reading, writing and filtering streams
// hierarchies. // and lists of api.Pod objects.
package archive package podutil
contrib/mesos/pkg/podutil/filters.go 0 → 100644
View file @ a999c0fb
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 podutil
import (
log "github.com/golang/glog"
"k8s.io/kubernetes/pkg/api"
)
type defaultFunc func(pod *api.Pod) error
type FilterFunc func(pod *api.Pod) (bool, error)
type Filters []FilterFunc
// Annotate safely copies annotation metadata from kv to meta.Annotations.
func Annotate(meta *api.ObjectMeta, kv map[string]string) {
//TODO(jdef) this func probably belong in an "apiutil" package, but we don't
//have much to put there right now so it can just live here.
if meta.Annotations == nil {
meta.Annotations = make(map[string]string)
}
for k, v := range kv {
meta.Annotations[k] = v
}
}
// Annotator returns a filter that copies annotations from map m into a pod
func Annotator(m map[string]string) FilterFunc {
return FilterFunc(func(pod *api.Pod) (bool, error) {
Annotate(&pod.ObjectMeta, m)
return true, nil
})
}
// Stream returns a chan of pods that yields each pod from the given list.
// No pods are yielded if err is non-nil.
func Stream(list *api.PodList, err error) <-chan *api.Pod {
out := make(chan *api.Pod)
go func() {
defer close(out)
if err != nil {
log.Errorf("failed to obtain pod list: %v", err)
return
}
for _, pod := range list.Items {
pod := pod
out <- &pod
}
}()
return out
}
func (filter FilterFunc) Do(in <-chan *api.Pod) <-chan *api.Pod {
out := make(chan *api.Pod)
go func() {
defer close(out)
for pod := range in {
if ok, err := filter(pod); err != nil {
log.Errorf("pod failed selection: %v", err)
} else if ok {
out <- pod
}
}
}()
return out
}
// List reads every pod from the pods chan and returns them all in an api.PodList
func List(pods <-chan *api.Pod) *api.PodList {
list := &api.PodList{}
for p := range pods {
list.Items = append(list.Items, *p)
}
return list
}
contrib/mesos/pkg/podutil/gzip.go 0 → 100644
View file @ a999c0fb
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 podutil
import (
"bytes"
"compress/gzip"
"fmt"
"io/ioutil"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/v1"
)
func Gzip(pods <-chan *api.Pod) ([]byte, error) {
return gzipList(List(pods))
}
func gzipList(list *api.PodList) ([]byte, error) {
raw, err := v1.Codec.Encode(list)
if err != nil {
return nil, err
}
zipped := &bytes.Buffer{}
zw := gzip.NewWriter(zipped)
_, err = bytes.NewBuffer(raw).WriteTo(zw)
if err != nil {
return nil, err
}
err = zw.Close()
if err != nil {
return nil, err
}
return zipped.Bytes(), nil
}
func Gunzip(gzipped []byte) <-chan *api.Pod {
return Stream(gunzipList(gzipped))
}
func gunzipList(gzipped []byte) (*api.PodList, error) {
zr, err := gzip.NewReader(bytes.NewReader(gzipped))
if err != nil {
return nil, err
}
defer zr.Close()
raw, err := ioutil.ReadAll(zr)
if err != nil {
return nil, err
}
obj, err := api.Scheme.Decode(raw)
if err != nil {
return nil, err
}
podlist, ok := obj.(*api.PodList)
if !ok {
return nil, fmt.Errorf("expected *api.PodList instead of %T", obj)
}
return podlist, nil
}
contrib/mesos/pkg/podutil/gzip_test.go 0 → 100644
View file @ a999c0fb
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 podutil
import (
"testing"
"github.com/stretchr/testify/assert"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/v1"
)
func TestGzipList(t *testing.T) {
// pod spec defaults are written during deserialization, this is what we
// expect them to be
period := int64(v1.DefaultTerminationGracePeriodSeconds)
defaultSpec := api.PodSpec{
DNSPolicy: api.DNSClusterFirst,
RestartPolicy: api.RestartPolicyAlways,
TerminationGracePeriodSeconds: &period,
SecurityContext: new(api.PodSecurityContext),
}
list := &api.PodList{
Items: []api.Pod{
{
ObjectMeta: api.ObjectMeta{
Name: "foo",
Namespace: "bar",
},
},
{
ObjectMeta: api.ObjectMeta{
Name: "qax",
Namespace: "lkj",
},
},
},
}
amap := map[string]string{
"crazy": "horse",
}
annotator := Annotator(amap)
raw, err := Gzip(annotator.Do(Stream(list, nil)))
assert.NoError(t, err)
list2, err := gunzipList(raw)
assert.NoError(t, err)
list.Items[0].Spec = defaultSpec
list.Items[0].Annotations = amap
list.Items[1].Spec = defaultSpec
list.Items[1].Annotations = amap
assert.True(t, api.Semantic.DeepEqual(*list, *list2), "expected %+v instead of %+v", *list, *list2)
}
contrib/mesos/pkg/podutil/io.go 0 → 100644
View file @ a999c0fb
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 podutil
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
log "github.com/golang/glog"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/meta"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/api/validation"
utilyaml "k8s.io/kubernetes/pkg/util/yaml"
)
func WriteToDir(pods <-chan *api.Pod, destDir string) error {
err := os.MkdirAll(destDir, 0660)
if err != nil {
return err
}
for p := range pods {
filename, ok := p.Annotations[meta.StaticPodFilenameKey]
if !ok {
log.Warningf("skipping static pod %s/%s that had no filename", p.Namespace, p.Name)
continue
}
raw, err := v1.Codec.Encode(p)
if err != nil {
log.Errorf("failed to encode static pod as v1 object: %v", err)
continue
}
destfile := filepath.Join(destDir, filename)
err = ioutil.WriteFile(destfile, raw, 0660)
if err != nil {
log.Errorf("failed to write static pod file %q: %v", destfile, err)
}
log.V(1).Infof("wrote static pod %s/%s to %s", p.Namespace, p.Name, destfile)
}
return nil
}
func ReadFromDir(dirpath string) (<-chan *api.Pod, <-chan error) {
pods := make(chan *api.Pod)
errors := make(chan error)
go func() {
defer close(pods)
defer close(errors)
files, err := ioutil.ReadDir(dirpath)
if err != nil {
errors <- fmt.Errorf("error scanning static pods directory: %q: %v", dirpath, err)
return
}
for _, f := range files {
if f.IsDir() || f.Size() == 0 {
continue
}
filename := filepath.Join(dirpath, f.Name())
log.V(1).Infof("reading static pod conf from file %q", filename)
data, err := ioutil.ReadFile(filename)
if err != nil {
errors <- fmt.Errorf("failed to read static pod file: %q: %v", filename, err)
continue
}
parsed, pod, err := tryDecodeSinglePod(data)
if !parsed {
if err != nil {
errors <- fmt.Errorf("error parsing static pod file %q: %v", filename, err)
}
continue
}
if err != nil {
errors <- fmt.Errorf("error validating static pod file %q: %v", filename, err)
continue
}
Annotate(&pod.ObjectMeta, map[string]string{meta.StaticPodFilenameKey: f.Name()})
pods <- pod
}
}()
return pods, errors
}
// tryDecodeSinglePod was copied from pkg/kubelet/config/common.go v1.0.5
func tryDecodeSinglePod(data []byte) (parsed bool, pod *api.Pod, err error) {
// JSON is valid YAML, so this should work for everything.
json, err := utilyaml.ToJSON(data)
if err != nil {
return false, nil, err
}
obj, err := api.Scheme.Decode(json)
if err != nil {
return false, pod, err
}
// Check whether the object could be converted to single pod.
if _, ok := obj.(*api.Pod); !ok {
err = fmt.Errorf("invalid pod: %+v", obj)
return false, pod, err
}
newPod := obj.(*api.Pod)
if errs := validation.ValidatePod(newPod); len(errs) > 0 {
err = fmt.Errorf("invalid pod: %v", errs)
return true, pod, err
}
return true, newPod, nil
}
contrib/mesos/pkg/scheduler/components/framework/framework.go
View file @ a999c0fb
...@@ -547,7 +547,6 @@ func (k *framework) SlaveLost(driver bindings.SchedulerDriver, slaveId *mesos.Sl ...@@ -547,7 +547,6 @@ func (k *framework) SlaveLost(driver bindings.SchedulerDriver, slaveId *mesos.Sl
// ExecutorLost is called when some executor is lost. // ExecutorLost is called when some executor is lost.
func (k *framework) ExecutorLost(driver bindings.SchedulerDriver, executorId *mesos.ExecutorID, slaveId *mesos.SlaveID, status int) { func (k *framework) ExecutorLost(driver bindings.SchedulerDriver, executorId *mesos.ExecutorID, slaveId *mesos.SlaveID, status int) {
log.Infof("Executor %v of slave %v is lost, status: %v\n", executorId, slaveId, status) log.Infof("Executor %v of slave %v is lost, status: %v\n", executorId, slaveId, status)
// TODO(yifan): Restart any unfinished tasks of the executor.
} }
// Error is called when there is an unrecoverable error in the scheduler or scheduler driver. // Error is called when there is an unrecoverable error in the scheduler or scheduler driver.
......
contrib/mesos/pkg/scheduler/meta/annotations.go
View file @ a999c0fb
...@@ -30,4 +30,5 @@ const ( ...@@ -30,4 +30,5 @@ const (
PortNameMappingKeyPrefix = "k8s.mesosphere.io/portName_" PortNameMappingKeyPrefix = "k8s.mesosphere.io/portName_"
PortNameMappingKeyFormat = PortNameMappingKeyPrefix + "%s_%s" PortNameMappingKeyFormat = PortNameMappingKeyPrefix + "%s_%s"
ContainerPortKeyFormat = "k8s.mesosphere.io/containerPort_%s_%s_%d" ContainerPortKeyFormat = "k8s.mesosphere.io/containerPort_%s_%s_%d"
StaticPodFilenameKey = "k8s.mesosphere.io/staticPodFilename"
) )
contrib/mesos/pkg/scheduler/scheduler.go
View file @ a999c0fb
...@@ -33,6 +33,5 @@ type Scheduler interface { ...@@ -33,6 +33,5 @@ type Scheduler interface {
Reconcile(t *podtask.T) Reconcile(t *podtask.T)
KillTask(id string) error KillTask(id string) error
LaunchTask(t *podtask.T) error LaunchTask(t *podtask.T) error
Run(done <-chan struct{}) Run(done <-chan struct{})
} }
contrib/mesos/pkg/scheduler/service/service.go
View file @ a999c0fb
...@@ -18,7 +18,6 @@ package service ...@@ -18,7 +18,6 @@ package service
import ( import (
"bufio" "bufio"
"encoding/json"
"errors" "errors"
"fmt" "fmt"
"io/ioutil" "io/ioutil"
...@@ -47,11 +46,11 @@ import ( ...@@ -47,11 +46,11 @@ import (
"github.com/spf13/pflag" "github.com/spf13/pflag"
"golang.org/x/net/context" "golang.org/x/net/context"
"k8s.io/kubernetes/contrib/mesos/pkg/archive"
"k8s.io/kubernetes/contrib/mesos/pkg/election" "k8s.io/kubernetes/contrib/mesos/pkg/election"
execcfg "k8s.io/kubernetes/contrib/mesos/pkg/executor/config" execcfg "k8s.io/kubernetes/contrib/mesos/pkg/executor/config"
"k8s.io/kubernetes/contrib/mesos/pkg/hyperkube" "k8s.io/kubernetes/contrib/mesos/pkg/hyperkube"
minioncfg "k8s.io/kubernetes/contrib/mesos/pkg/minion/config" minioncfg "k8s.io/kubernetes/contrib/mesos/pkg/minion/config"
"k8s.io/kubernetes/contrib/mesos/pkg/podutil"
"k8s.io/kubernetes/contrib/mesos/pkg/profile" "k8s.io/kubernetes/contrib/mesos/pkg/profile"
"k8s.io/kubernetes/contrib/mesos/pkg/runtime" "k8s.io/kubernetes/contrib/mesos/pkg/runtime"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components" "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/components"
...@@ -76,6 +75,9 @@ import ( ...@@ -76,6 +75,9 @@ import (
"k8s.io/kubernetes/pkg/master/ports" "k8s.io/kubernetes/pkg/master/ports"
etcdstorage "k8s.io/kubernetes/pkg/storage/etcd" etcdstorage "k8s.io/kubernetes/pkg/storage/etcd"
"k8s.io/kubernetes/pkg/tools" "k8s.io/kubernetes/pkg/tools"
// lock to this API version, compilation will fail when this becomes unsupported
_ "k8s.io/kubernetes/pkg/api/v1"
) )
const ( const (
...@@ -432,46 +434,7 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E ...@@ -432,46 +434,7 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
} }
// Check for staticPods // Check for staticPods
var staticPodCPUs, staticPodMem float64 data, staticPodCPUs, staticPodMem := s.prepareStaticPods()
if s.staticPodsConfigPath != "" {
bs, paths, err := archive.ZipDir(s.staticPodsConfigPath)
if err != nil {
return nil, nil, err
}
// try to read pod files and sum resources
// TODO(sttts): don't terminate when static pods are broken, but skip them
// TODO(sttts): add a directory watch and tell running executors about updates
for _, podPath := range paths {
podJson, err := ioutil.ReadFile(podPath)
if err != nil {
return nil, nil, fmt.Errorf("error reading static pod spec: %v", err)
}
pod := api.Pod{}
err = json.Unmarshal(podJson, &pod)
if err != nil {
return nil, nil, fmt.Errorf("error parsing static pod spec at %v: %v", podPath, err)
}
_, cpu, _, err := mresource.LimitPodCPU(&pod, s.defaultContainerCPULimit)
if err != nil {
return nil, nil, fmt.Errorf("cannot derive cpu limit for static pod: %v", podPath)
}
_, mem, _, err := mresource.LimitPodMem(&pod, s.defaultContainerMemLimit)
if err != nil {
return nil, nil, fmt.Errorf("cannot derive memory limit for static pod: %v", podPath)
}
log.V(2).Infof("reserving %.2f cpu shares and %.2f MB of memory to static pod %s", cpu, mem, pod.Name)
staticPodCPUs += float64(cpu)
staticPodMem += float64(mem)
}
// pass zipped pod spec to executor
execInfo.Data = bs
}
execInfo.Resources = []*mesos.Resource{ execInfo.Resources = []*mesos.Resource{
mutil.NewScalarResource("cpus", float64(s.mesosExecutorCPUs)+staticPodCPUs), mutil.NewScalarResource("cpus", float64(s.mesosExecutorCPUs)+staticPodCPUs),
...@@ -483,10 +446,43 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E ...@@ -483,10 +446,43 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
ehash := hashExecutorInfo(execInfo) ehash := hashExecutorInfo(execInfo)
eid := uid.New(ehash, execcfg.DefaultInfoID) eid := uid.New(ehash, execcfg.DefaultInfoID)
execInfo.ExecutorId = &mesos.ExecutorID{Value: proto.String(eid.String())} execInfo.ExecutorId = &mesos.ExecutorID{Value: proto.String(eid.String())}
execInfo.Data = data
return execInfo, eid, nil return execInfo, eid, nil
} }
func (s *SchedulerServer) prepareStaticPods() (data []byte, staticPodCPUs, staticPodMem float64) {
// TODO(sttts): add a directory watch and tell running executors about updates
if s.staticPodsConfigPath == "" {
return
}
entries, errCh := podutil.ReadFromDir(s.staticPodsConfigPath)
go func() {
// we just skip file system errors for now, do our best to gather
// as many static pod specs as we can.
for err := range errCh {
log.Errorln(err.Error())
}
}()
// validate cpu and memory limits, tracking the running totals in staticPod{CPUs,Mem}
validateResourceLimits := StaticPodValidator(
s.defaultContainerCPULimit,
s.defaultContainerMemLimit,
&staticPodCPUs,
&staticPodMem)
zipped, err := podutil.Gzip(validateResourceLimits.Do(entries))
if err != nil {
log.Errorf("failed to generate static pod data: %v", err)
staticPodCPUs, staticPodMem = 0, 0
} else {
data = zipped
}
return
}
// TODO(jdef): hacked from kubelet/server/server.go // TODO(jdef): hacked from kubelet/server/server.go
// TODO(k8s): replace this with clientcmd // TODO(k8s): replace this with clientcmd
func (s *SchedulerServer) createAPIServerClient() (*client.Client, error) { func (s *SchedulerServer) createAPIServerClient() (*client.Client, error) {
......
contrib/mesos/pkg/scheduler/service/service_test.go
View file @ a999c0fb
// +build unit_test
/* /*
Copyright 2015 The Kubernetes Authors All rights reserved. Copyright 2015 The Kubernetes Authors All rights reserved.
...@@ -19,15 +17,9 @@ limitations under the License. ...@@ -19,15 +17,9 @@ limitations under the License.
package service package service
import ( import (
"archive/zip"
"bytes"
"io/ioutil"
"os"
"path/filepath"
"testing" "testing"
"time" "time"
"k8s.io/kubernetes/contrib/mesos/pkg/archive"
mresource "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/resource" mresource "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/resource"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
...@@ -124,42 +116,3 @@ func Test_DefaultResourceLimits(t *testing.T) { ...@@ -124,42 +116,3 @@ func Test_DefaultResourceLimits(t *testing.T) {
assert.Equal(s.defaultContainerCPULimit, mresource.DefaultDefaultContainerCPULimit) assert.Equal(s.defaultContainerCPULimit, mresource.DefaultDefaultContainerCPULimit)
assert.Equal(s.defaultContainerMemLimit, mresource.DefaultDefaultContainerMemLimit) assert.Equal(s.defaultContainerMemLimit, mresource.DefaultDefaultContainerMemLimit)
} }
func Test_StaticPods(t *testing.T) {
assert := assert.New(t)
// create static pods config files, spod1 on toplevel and spod2 in a directory "dir"
staticPodsConfigPath, err := ioutil.TempDir(os.TempDir(), "executor-k8sm-archive")
assert.NoError(err)
defer os.RemoveAll(staticPodsConfigPath)
spod1, err := os.Create(filepath.Join(staticPodsConfigPath, "spod1.json"))
assert.NoError(err)
_, err = spod1.WriteString("content1")
assert.NoError(err)
err = os.Mkdir(filepath.Join(staticPodsConfigPath, "dir"), 0755)
assert.NoError(err)
spod2, err := os.Create(filepath.Join(staticPodsConfigPath, "dir", "spod2.json"))
assert.NoError(err)
_, err = spod2.WriteString("content2")
assert.NoError(err)
// archive config files
data, paths, err := archive.ZipDir(staticPodsConfigPath)
assert.NoError(err)
assert.Equal(2, len(paths))
// unarchive config files
zr, err := zip.NewReader(bytes.NewReader(data), int64(len(data)))
assert.NoError(err)
fileNames := []string{}
for _, f := range zr.File {
if !f.FileInfo().IsDir() {
fileNames = append(fileNames, f.Name)
}
}
assert.Contains(fileNames, "spod1.json")
assert.Contains(fileNames, "dir/spod2.json")
}
contrib/mesos/pkg/scheduler/service/validation.go 0 → 100644
View file @ a999c0fb
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 service
import (
log "github.com/golang/glog"
"k8s.io/kubernetes/contrib/mesos/pkg/podutil"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/resource"
"k8s.io/kubernetes/pkg/api"
)
// StaticPodValidator discards a pod if we can't calculate resource limits for it.
func StaticPodValidator(
defaultContainerCPULimit resource.CPUShares,
defaultContainerMemLimit resource.MegaBytes,
accumCPU, accumMem *float64,
) podutil.FilterFunc {
return podutil.FilterFunc(func(pod *api.Pod) (bool, error) {
_, cpu, _, err := resource.LimitPodCPU(pod, defaultContainerCPULimit)
if err != nil {
return false, err
}
_, mem, _, err := resource.LimitPodMem(pod, defaultContainerMemLimit)
if err != nil {
return false, err
}
log.V(2).Infof("reserving %.2f cpu shares and %.2f MB of memory to static pod %s/%s", cpu, mem, pod.Namespace, pod.Name)
*accumCPU += float64(cpu)
*accumMem += float64(mem)
return true, nil
})
}
contrib/mesos/pkg/scheduler/service/validation_test.go 0 → 100644
View file @ a999c0fb
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed 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 service_test
import (
"testing"
"github.com/stretchr/testify/assert"
"k8s.io/kubernetes/contrib/mesos/pkg/podutil"
mresource "k8s.io/kubernetes/contrib/mesos/pkg/scheduler/resource"
"k8s.io/kubernetes/contrib/mesos/pkg/scheduler/service"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/resource"
)
func TestStaticPodValidator(t *testing.T) {
// test within limits
tests := []struct {
// given
pods <-chan *api.Pod
// wants
podcount int
cputot float64
memtot float64
}{
// test: valid, pod specifies limits for ALL containers
{
pods: pods(pod(
podName("foo", "bar"),
containers(
container(resourceLimits(10, 20)), // min is 32
container(resourceLimits(30, 40)),
container(resourceLimits(50, 60)),
),
)),
podcount: 1,
cputot: 90,
memtot: 132,
},
// test: valid, multiple pods, specify limits for ALL containers
{
pods: pods(
pod(
podName("foo", "bar"),
containers(
container(resourceLimits(10, 20)), // min is 32
container(resourceLimits(30, 40)),
container(resourceLimits(50, 60)),
),
),
pod(
podName("kjh", "jkk"),
containers(
container(resourceLimits(15, 25)), // min is 32
container(resourceLimits(35, 45)),
container(resourceLimits(55, 65)),
),
),
),
podcount: 2,
cputot: 195,
memtot: 274,
},
// test: no limits on CT in first pod so it's rejected
{
pods: pods(
pod(
podName("foo", "bar"),
containers(
container(resourceLimits(10, 20)), // min is 32
container(), // min is 0.01, 32
container(resourceLimits(50, 60)),
),
),
pod(
podName("wza", "wer"),
containers(
container(resourceLimits(10, 20)), // min is 32
container(resourceLimits(30, 40)),
container(resourceLimits(50, 60)),
),
),
),
podcount: 2,
cputot: 60.01 + 90,
memtot: 124 + 132,
},
}
for i, tc := range tests {
var cpu, mem float64
f := service.StaticPodValidator(0, 0, &cpu, &mem)
list := podutil.List(f.Do(tc.pods))
assert.Equal(t, tc.podcount, len(list.Items), "test case #%d: expected %d pods instead of %d", i, tc.podcount, len(list.Items))
assert.EqualValues(t, tc.cputot, cpu, "test case #%d: expected %f total cpu instead of %f", i, tc.cputot, cpu)
assert.EqualValues(t, tc.memtot, mem, "test case #%d: expected %f total mem instead of %f", i, tc.memtot, mem)
}
}
type podOpt func(*api.Pod)
type ctOpt func(*api.Container)
func pods(pods ...*api.Pod) <-chan *api.Pod {
ch := make(chan *api.Pod, len(pods))
for _, x := range pods {
ch <- x
}
close(ch)
return ch
}
func pod(opts ...podOpt) *api.Pod {
p := &api.Pod{}
for _, x := range opts {
x(p)
}
return p
}
func container(opts ...ctOpt) (c api.Container) {
for _, x := range opts {
x(&c)
}
return
}
func containers(ct ...api.Container) podOpt {
return podOpt(func(p *api.Pod) {
p.Spec.Containers = ct
})
}
func resourceLimits(cpu mresource.CPUShares, mem mresource.MegaBytes) ctOpt {
return ctOpt(func(c *api.Container) {
if c.Resources.Limits == nil {
c.Resources.Limits = make(api.ResourceList)
}
c.Resources.Limits[api.ResourceCPU] = *resource.NewMilliQuantity(int64(float64(cpu)*1000.0), resource.DecimalSI)
c.Resources.Limits[api.ResourceMemory] = *resource.NewQuantity(int64(float64(mem)*1024.0*1024.0), resource.BinarySI)
})
}
func podName(ns, name string) podOpt {
return podOpt(func(p *api.Pod) {
p.Namespace = ns
p.Name = name
})
}
test/e2e/mesos.go
View file @ a999c0fb
...@@ -17,12 +17,15 @@ limitations under the License. ...@@ -17,12 +17,15 @@ limitations under the License.
package e2e package e2e
import ( import (
"fmt"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/fields"
"k8s.io/kubernetes/pkg/labels" "k8s.io/kubernetes/pkg/labels"
"k8s.io/kubernetes/pkg/util"
. "github.com/onsi/ginkgo" . "github.com/onsi/ginkgo"
. "github.com/onsi/gomega" . "github.com/onsi/gomega"
"k8s.io/kubernetes/pkg/fields"
) )
var _ = Describe("Mesos", func() { var _ = Describe("Mesos", func() {
...@@ -50,4 +53,17 @@ var _ = Describe("Mesos", func() { ...@@ -50,4 +53,17 @@ var _ = Describe("Mesos", func() {
} }
Expect(len(addr)).NotTo(Equal("")) Expect(len(addr)).NotTo(Equal(""))
}) })
It("starts static pods on every node in the mesos cluster", func() {
client := framework.Client
expectNoError(allNodesReady(client, util.ForeverTestTimeout), "all nodes ready")
nodelist, err := client.Nodes().List(labels.Everything(), fields.Everything())
expectNoError(err, "nodes fetched from apiserver")
const ns = "static-pods"
numpods := len(nodelist.Items)
expectNoError(waitForPodsRunningReady(ns, numpods, util.ForeverTestTimeout),
fmt.Sprintf("number of static pods in namespace %s is %d", ns, numpods))
})
}) })
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment