Merge pull request #9077 from mesosphere/staticPodsUpstream

Add static pod support to mesos scheduler and executor.

contrib/mesos/pkg/scheduler/plugin_test.go

@@ -379,12 +379,16 @@ func TestPlugin_LifeCycle(t *testing.T) {
 	testApiServer := NewTestServer(t, api.NamespaceDefault, podListWatch)
 	defer testApiServer.server.Close()
 
+	// create executor with some data for static pods if set
+	executor := util.NewExecutorInfo(
+		util.NewExecutorID("executor-id"),
+		util.NewCommandInfo("executor-cmd"),
+	)
+	executor.Data = []byte{0, 1, 2}
+
 	// create scheduler
 	testScheduler := New(Config{
-		Executor: util.NewExecutorInfo(
-			util.NewExecutorID("executor-id"),
-			util.NewCommandInfo("executor-cmd"),
-		),
+		Executor:     executor,
 		Client:       client.NewOrDie(&client.Config{Host: testApiServer.server.URL, Version: testapi.Version()}),
 		ScheduleFunc: FCFSScheduleFunc,
 		Schedcfg:     *schedcfg.CreateDefaultConfig(),
@@ -477,6 +481,9 @@ func TestPlugin_LifeCycle(t *testing.T) {
 		testScheduler.StatusUpdate(mockDriver, newTaskStatusForTask(launchedTask, mesos.TaskState_TASK_STAGING))
 		testScheduler.StatusUpdate(mockDriver, newTaskStatusForTask(launchedTask, mesos.TaskState_TASK_RUNNING))
 
+		// check that ExecutorInfo.data has the static pod data
+		assert.Len(launchedTask.Executor.Data, 3)
+
 		// report back that the task has been lost
 		mockDriver.AssertNumberOfCalls(t, "SendFrameworkMessage", 0)
 		testScheduler.StatusUpdate(mockDriver, newTaskStatusForTask(launchedTask, mesos.TaskState_TASK_LOST))

contrib/mesos/pkg/scheduler/podtask/pod_task.go

@@ -33,8 +33,8 @@ import (
 )
 
 const (
-	containerCpus = 0.25 // initial CPU allocated for executor
-	containerMem  = 64   // initial MB of memory allocated for executor
+	DefaultContainerCpus = 0.25 // initial CPU allocated for executor
+	DefaultContainerMem  = 64   // initial MB of memory allocated for executor
 )
 
 type StateType int
@@ -164,8 +164,8 @@ func (t *T) FillFromDetails(details *mesos.Offer) error {
 
 	t.Spec = Spec{
 		SlaveID: details.GetSlaveId().GetValue(),
-		CPU:     containerCpus,
-		Memory:  containerMem,
+		CPU:     DefaultContainerCpus,
+		Memory:  DefaultContainerMem,
 	}
 
 	if mapping, err := t.mapper.Generate(t, details); err != nil {
@@ -238,7 +238,7 @@ func (t *T) AcceptOffer(offer *mesos.Offer) bool {
 	// resource allocation and management.
 	//
 	// TODO(jdef): remove hardcoded values and make use of actual pod resource settings
-	if (cpus < containerCpus) || (mem < containerMem) {
+	if (cpus < DefaultContainerCpus) || (mem < DefaultContainerMem) {
 		log.V(3).Infof("not enough resources: cpus: %f mem: %f", cpus, mem)
 		return false
 	}

contrib/mesos/pkg/scheduler/service/service.go

@@ -31,6 +31,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/archive"
 	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/election"
 	execcfg "github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/executor/config"
 	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/hyperkube"
@@ -41,6 +42,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/scheduler/ha"
 	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/scheduler/meta"
 	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/scheduler/metrics"
+	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/scheduler/podtask"
 	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/scheduler/uid"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/clientauth"
@@ -116,6 +118,7 @@ type SchedulerServer struct {
 	KubeletHostNetworkSources     string
 	KubeletSyncFrequency          time.Duration
 	KubeletNetworkPluginName      string
+	StaticPodsConfigPath          string
 
 	executable  string // path to the binary running this service
 	client      *client.Client
@@ -174,6 +177,7 @@ func (s *SchedulerServer) addCoreFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&s.AllowPrivileged, "allow-privileged", s.AllowPrivileged, "If true, allow privileged containers.")
 	fs.StringVar(&s.ClusterDomain, "cluster-domain", s.ClusterDomain, "Domain for this cluster.  If set, kubelet will configure all containers to search this domain in addition to the host's search domains")
 	fs.Var(&s.ClusterDNS, "cluster-dns", "IP address for a cluster DNS server. If set, kubelet will configure all containers to use this for DNS resolution in addition to the host's DNS servers")
+	fs.StringVar(&s.StaticPodsConfigPath, "static-pods-config", s.StaticPodsConfigPath, "Path for specification of static pods. Path should point to dir containing the staticPods configuration files. Defaults to none.")
 
 	fs.StringVar(&s.MesosMaster, "mesos-master", s.MesosMaster, "Location of the Mesos master. The format is a comma-delimited list of of hosts like zk://host1:port,host2:port/mesos. If using ZooKeeper, pay particular attention to the leading zk:// and trailing /mesos! If not using ZooKeeper, standard URLs like http://localhost are also acceptable.")
 	fs.StringVar(&s.MesosUser, "mesos-user", s.MesosUser, "Mesos user for this framework, defaults to root.")
@@ -353,6 +357,25 @@ func (s *SchedulerServer) prepareExecutorInfo(hks hyperkube.Interface) (*mesos.E
 		Source:  proto.String(execcfg.DefaultInfoSource),
 	}
 
+	// Check for staticPods
+	if s.StaticPodsConfigPath != "" {
+		bs, numberStaticPods, err := archive.ZipDir(s.StaticPodsConfigPath)
+		if err != nil {
+			return nil, nil, err
+		}
+		info.Data = bs
+
+		// Adjust the resource accounting for the executor.
+		// Currently each podTask accounts the default amount of resources.
+		// TODO(joerg84) adapt to actual resources specified by pods.
+		log.Infof("Detected %d staticPods in Configuration.", numberStaticPods)
+
+		info.Resources = []*mesos.Resource{
+			mutil.NewScalarResource("cpus", float64(numberStaticPods)*podtask.DefaultContainerCpus),
+			mutil.NewScalarResource("mem", float64(numberStaticPods)*podtask.DefaultContainerMem),
+		}
+	}
+
 	// calculate ExecutorInfo hash to be used for validating compatibility
 	// of ExecutorInfo's generated by other HA schedulers.
 	ehash := hashExecutorInfo(info)

contrib/mesos/pkg/scheduler/service/service_test.go

@@ -19,8 +19,16 @@ limitations under the License.
 package service
 
 import (
+	"archive/zip"
+	"bytes"
+	"io/ioutil"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
+
+	"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/archive"
+	"github.com/stretchr/testify/assert"
 )
 
 type fakeSchedulerProcess struct {
@@ -106,3 +114,42 @@ func Test_awaitFailoverDoneFailover(t *testing.T) {
 		t.Fatalf("expected call to failover handler")
 	}
 }
+
+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, fileNum, err := archive.ZipDir(staticPodsConfigPath)
+	assert.NoError(err)
+	assert.Equal(2, fileNum)
+
+	// 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")
+}