Add basic strategy deployment controller

pkg/cmd/server/origin/master.go

@@ -33,6 +33,7 @@ import (
 	osclient "github.com/openshift/origin/pkg/client"
 	cmdutil "github.com/openshift/origin/pkg/cmd/util"
 	deploycontrollerfactory "github.com/openshift/origin/pkg/deploy/controller/factory"
+	deployconfiggenerator "github.com/openshift/origin/pkg/deploy/generator"
 	deployregistry "github.com/openshift/origin/pkg/deploy/registry/deploy"
 	deployconfigregistry "github.com/openshift/origin/pkg/deploy/registry/deployconfig"
 	deployetcd "github.com/openshift/origin/pkg/deploy/registry/etcd"
@@ -114,6 +115,12 @@ func (c *MasterConfig) RunAPI(m APIInstaller) {
 	deployEtcd := deployetcd.New(c.EtcdHelper)
 	routeEtcd := routeetcd.New(c.EtcdHelper)
 
+	deployConfigGenerator := &deployconfiggenerator.DeploymentConfigGenerator{
+		DeploymentInterface:       deployEtcd,
+		DeploymentConfigInterface: deployEtcd,
+		ImageRepositoryInterface:  imageRegistry,
+	}
+
 	// initialize OpenShift API
 	storage := map[string]apiserver.RESTStorage{
 		"builds":       buildregistry.NewREST(buildRegistry),
@@ -124,8 +131,9 @@ func (c *MasterConfig) RunAPI(m APIInstaller) {
 		"imageRepositories":       imagerepository.NewREST(imageRegistry),
 		"imageRepositoryMappings": imagerepositorymapping.NewREST(imageRegistry, imageRegistry),
 
-		"deployments":       deployregistry.NewREST(deployEtcd),
-		"deploymentConfigs": deployconfigregistry.NewREST(deployEtcd),
+		"deployments":               deployregistry.NewREST(deployEtcd),
+		"deploymentConfigs":         deployconfigregistry.NewREST(deployEtcd),
+		"generateDeploymentConfigs": deployconfiggenerator.NewREST(deployConfigGenerator, v1beta1.Codec),
 
 		"templateConfigs": template.NewStorage(),
 
@@ -233,6 +241,16 @@ func (c *MasterConfig) RunDeploymentController() {
 	controller.Run()
 }
 
+func (c *MasterConfig) RunBasicDeploymentController() {
+	factory := deploycontrollerfactory.BasicDeploymentControllerFactory{
+		Client:     c.OSClient,
+		KubeClient: c.KubeClient,
+	}
+
+	controller := factory.Create()
+	controller.Run()
+}
+
 func (c *MasterConfig) RunDeploymentConfigController() {
 	factory := deploycontrollerfactory.DeploymentConfigControllerFactory{c.OSClient}
 	controller := factory.Create()

pkg/cmd/server/start.go

@@ -177,6 +177,7 @@ func NewCommandStartServer(name string) *cobra.Command {
 				osmaster.RunAssetServer()
 				osmaster.RunBuildController()
 				osmaster.RunDeploymentController()
+				osmaster.RunBasicDeploymentController()
 				osmaster.RunDeploymentConfigController()
 				osmaster.RunConfigChangeController()
 				osmaster.RunDeploymentImageChangeTriggerController()

pkg/deploy/api/validation/validation.go

@@ -27,10 +27,12 @@ func validateDeploymentStrategy(strategy *deployapi.DeploymentStrategy) errors.E
 		result = append(result, errors.NewFieldRequired("Type", ""))
 	}
 
-	if strategy.CustomPod == nil {
-		result = append(result, errors.NewFieldRequired("CustomPod", nil))
-	} else {
-		result = append(result, validateCustomPodStrategy(strategy.CustomPod).Prefix("CustomPod")...)
+	if strategy.Type == "customPod" {
+		if strategy.CustomPod == nil {
+			result = append(result, errors.NewFieldRequired("CustomPod", nil))
+		} else {
+			result = append(result, validateCustomPodStrategy(strategy.CustomPod).Prefix("CustomPod")...)
+		}
 	}
 
 	return result

pkg/deploy/controller/basic_deployment_controller.go

@@ -0,0 +1,128 @@
+package controller
+
+import (
+  "github.com/golang/glog"
+
+  kapi "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+  "github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+  "github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+  deployapi "github.com/openshift/origin/pkg/deploy/api"
+)
+
+type BasicDeploymentController struct {
+  DeploymentInterface            bdcDeploymentInterface
+  ReplicationControllerInterface bdcReplicationControllerInterface
+  NextDeployment                 func() *deployapi.Deployment
+}
+
+type bdcDeploymentInterface interface {
+  UpdateDeployment(ctx kapi.Context, deployment *deployapi.Deployment) (*deployapi.Deployment, error)
+}
+
+type bdcReplicationControllerInterface interface {
+  ListReplicationControllers(ctx kapi.Context, selector labels.Selector) (*kapi.ReplicationControllerList, error)
+  GetReplicationController(ctx kapi.Context, id string) (*kapi.ReplicationController, error)
+  CreateReplicationController(ctx kapi.Context, ctrl *kapi.ReplicationController) (*kapi.ReplicationController, error)
+  UpdateReplicationController(ctx kapi.Context, ctrl *kapi.ReplicationController) (*kapi.ReplicationController, error)
+  DeleteReplicationController(ctx kapi.Context, id string) error
+}
+
+func (dc *BasicDeploymentController) Run() {
+  go util.Forever(func() { dc.HandleDeployment() }, 0)
+}
+
+func (dc *BasicDeploymentController) HandleDeployment() error {
+  deployment := dc.NextDeployment()
+  ctx := kapi.WithNamespace(kapi.NewContext(), deployment.Namespace)
+
+  nextStatus := deployment.Status
+  switch deployment.Status {
+  case deployapi.DeploymentStatusNew:
+    nextStatus = dc.handleNew(ctx, deployment)
+  }
+
+  // persist any status change
+  if deployment.Status != nextStatus {
+    deployment.Status = nextStatus
+    glog.Infof("Saving deployment %v status: %v", deployment.ID, deployment.Status)
+    if _, err := dc.DeploymentInterface.UpdateDeployment(ctx, deployment); err != nil {
+      glog.Errorf("Received error while saving deployment %v: %v", deployment.ID, err)
+      return err
+    }
+  }
+
+  return nil
+}
+
+func (dc *BasicDeploymentController) handleNew(ctx kapi.Context, deployment *deployapi.Deployment) deployapi.DeploymentStatus {
+  var replicationControllers *kapi.ReplicationControllerList
+  var err error
+
+  configID, hasConfigID := deployment.Labels[deployapi.DeploymentConfigIDLabel]
+  if hasConfigID {
+    selector, _ := labels.ParseSelector(deployapi.DeploymentConfigIDLabel + "=" + configID)
+    replicationControllers, err = dc.ReplicationControllerInterface.ListReplicationControllers(ctx, selector)
+    if err != nil {
+      glog.Infof("Unable to get list of replication controllers for previous deploymentConfig %s: %v\n", configID, err)
+      return deployapi.DeploymentStatusFailed
+    }
+  }
+
+  controller := &kapi.ReplicationController{
+    DesiredState: deployment.ControllerTemplate,
+    Labels:       map[string]string{deployapi.DeploymentConfigIDLabel: configID, "deploymentID": deployment.ID},
+  }
+
+  if controller.DesiredState.PodTemplate.Labels == nil {
+    controller.DesiredState.PodTemplate.Labels = make(map[string]string)
+  }
+
+  controller.DesiredState.PodTemplate.Labels[deployapi.DeploymentConfigIDLabel] = configID
+  controller.DesiredState.PodTemplate.Labels["deploymentID"] = deployment.ID
+
+  glog.Infof("Creating replicationController for deployment %s", deployment.ID)
+  if _, err := dc.ReplicationControllerInterface.CreateReplicationController(ctx, controller); err != nil {
+    glog.Infof("An error occurred creating the replication controller for deployment %s: %v", deployment.ID, err)
+    return deployapi.DeploymentStatusFailed
+  }
+
+  allReplControllersProcessed := true
+  // For this simple deploy, remove previous replication controllers
+  for _, rc := range replicationControllers.Items {
+    configID, _ := deployment.Labels[deployapi.DeploymentConfigIDLabel]
+    glog.Infof("Stopping replication controller for previous deploymentConfig %s: %v", configID, rc.ID)
+
+    replicationController, err := dc.ReplicationControllerInterface.GetReplicationController(ctx, rc.ID)
+    if err != nil {
+      glog.Infof("Unable to get replication controller %s for previous deploymentConfig %s: %#v\n", rc.ID, configID, err)
+      allReplControllersProcessed = false
+      continue
+    }
+
+    replicationController.DesiredState.Replicas = 0
+    glog.Infof("Settings Replicas=0 for replicationController %s for previous deploymentConfig %s", rc.ID, configID)
+    if _, err := dc.ReplicationControllerInterface.UpdateReplicationController(ctx, replicationController); err != nil {
+      glog.Infof("Unable to stop replication controller %s for previous deploymentConfig %s: %#v\n", rc.ID, configID, err)
+      allReplControllersProcessed = false
+      continue
+    }
+  }
+
+  for _, rc := range replicationControllers.Items {
+    configID, _ := deployment.Labels[deployapi.DeploymentConfigIDLabel]
+    glog.Infof("Deleting replication controller %s for previous deploymentConfig %s", rc.ID, configID)
+    err := dc.ReplicationControllerInterface.DeleteReplicationController(ctx, rc.ID)
+    if err != nil {
+      glog.Infof("Unable to remove replication controller %s for previous deploymentConfig %s:%#v\n", rc.ID, configID, err)
+      allReplControllersProcessed = false
+      continue
+    }
+  }
+
+  if allReplControllersProcessed {
+    return deployapi.DeploymentStatusComplete
+  } else {
+    return deployapi.DeploymentStatusFailed
+  }
+}

pkg/deploy/controller/config_change_controller.go

@@ -43,14 +43,26 @@ func (dc *ConfigChangeController) HandleDeploymentConfig() error {
   }
 
   if !hasChangeTrigger {
+    glog.Infof("Config has no change trigger; skipping")
+    return nil
+  }
+
+  if config.LatestVersion == 0 {
+    glog.Info("Ignoring config change with LatestVersion=0")
     return nil
   }
 
   latestDeploymentId := deployutil.LatestDeploymentIDForConfig(config)
   obj, exists := dc.DeploymentStore.Get(latestDeploymentId)
 
-  if !exists || !deployutil.PodTemplatesEqual(config.Template.ControllerTemplate.PodTemplate,
+  if !exists {
+    glog.Info("Ignoring config change due to lack of existing deployment")
+    return nil
+  }
+
+  if !deployutil.PodTemplatesEqual(config.Template.ControllerTemplate.PodTemplate,
     obj.(*deployapi.Deployment).ControllerTemplate.PodTemplate) {
+    glog.Info("Detected change to existing deployment")
     ctx := kapi.WithNamespace(kapi.NewContext(), config.Namespace)
     newConfig, err := dc.DeploymentConfigInterface.GenerateDeploymentConfig(ctx, config.ID)
     if err != nil {
@@ -65,5 +77,7 @@ func (dc *ConfigChangeController) HandleDeploymentConfig() error {
     }
   }
 
+  glog.Info("How did I get here?")
+
   return nil
 }

pkg/deploy/controller/deployment_config_controller.go

@@ -58,6 +58,10 @@ func (c *DeploymentConfigController) shouldDeploy(ctx kapi.Context, config *depl
 	}
 
 	deployment, err := c.latestDeploymentForConfig(ctx, config)
+	if deployment != nil {
+		return false, nil
+	}
+
 	if err != nil {
 		if errors.IsNotFound(err) {
 			glog.Infof("Should deploy config %s because there's no latest deployment", config.ID)
@@ -68,7 +72,8 @@ func (c *DeploymentConfigController) shouldDeploy(ctx kapi.Context, config *depl
 		}
 	}
 
-	return !deployutil.PodTemplatesEqual(deployment.ControllerTemplate.PodTemplate, config.Template.ControllerTemplate.PodTemplate), nil
+	// TODO: what state would this represent?
+	return false, nil
 }
 
 // TODO: reduce code duplication between trigger and config controllers

pkg/deploy/controller/factory/factory.go

@@ -21,8 +21,9 @@ type DeploymentControllerFactory struct {
 }
 
 func (factory *DeploymentControllerFactory) Create() *controller.DeploymentController {
+  field := labels.SelectorFromSet(labels.Set{"Strategy": "customPod"})
   queue := cache.NewFIFO()
-  cache.NewReflector(&deploymentLW{factory.Client}, &deployapi.Deployment{}, queue).Run()
+  cache.NewReflector(&deploymentLW{client: factory.Client, field: field}, &deployapi.Deployment{}, queue).Run()
 
   return &controller.DeploymentController{
     DeploymentInterface: factory.Client,
@@ -34,6 +35,25 @@ func (factory *DeploymentControllerFactory) Create() *controller.DeploymentContr
   }
 }
 
+type BasicDeploymentControllerFactory struct {
+  Client     *osclient.Client
+  KubeClient *kclient.Client
+}
+
+func (factory *BasicDeploymentControllerFactory) Create() *controller.BasicDeploymentController {
+  field := labels.SelectorFromSet(labels.Set{"Strategy": "basic"})
+  queue := cache.NewFIFO()
+  cache.NewReflector(&deploymentLW{client: factory.Client, field: field}, &deployapi.Deployment{}, queue).Run()
+
+  return &controller.BasicDeploymentController{
+    DeploymentInterface:            factory.Client,
+    ReplicationControllerInterface: factory.KubeClient,
+    NextDeployment: func() *deployapi.Deployment {
+      return queue.Pop().(*deployapi.Deployment)
+    },
+  }
+}
+
 type DeploymentConfigControllerFactory struct {
   Client *osclient.Client
 }
@@ -59,7 +79,7 @@ func (factory *ConfigChangeControllerFactory) Create() *controller.ConfigChangeC
   cache.NewReflector(&deploymentConfigLW{factory.Client}, &deployapi.DeploymentConfig{}, queue).Run()
 
   store := cache.NewStore()
-  cache.NewReflector(&deploymentLW{factory.Client}, &deployapi.Deployment{}, store).Run()
+  cache.NewReflector(&deploymentLW{client: factory.Client, field: labels.Everything()}, &deployapi.Deployment{}, store).Run()
 
   return &controller.ConfigChangeController{
     DeploymentConfigInterface: factory.Client,
@@ -92,14 +112,35 @@ func (factory *ImageChangeControllerFactory) Create() *controller.ImageChangeCon
 
 type deploymentLW struct {
   client osclient.Interface
+  field  labels.Selector
 }
 
 func (lw *deploymentLW) List() (runtime.Object, error) {
-  return lw.client.ListDeployments(kapi.NewContext(), labels.Everything())
+  // TODO: remove this filtering once the List interfaces support field selectors
+  results := &deployapi.DeploymentList{}
+  filtered := []deployapi.Deployment{}
+  deployments, err := lw.client.ListDeployments(kapi.NewContext(), labels.Everything())
+
+  if err != nil {
+    return nil, err
+  }
+
+  for _, deployment := range deployments.Items {
+    fields := labels.Set{
+      "ID":       deployment.ID,
+      "Strategy": deployment.Strategy.Type,
+    }
+    if lw.field.Matches(fields) {
+      filtered = append(filtered, deployment)
+    }
+  }
+
+  results.Items = filtered
+  return results, nil
 }
 
 func (lw *deploymentLW) Watch(resourceVersion uint64) (watch.Interface, error) {
-  return lw.client.WatchDeployments(kapi.NewContext(), labels.Everything(), labels.Everything(), 0)
+  return lw.client.WatchDeployments(kapi.NewContext(), lw.field, labels.Everything(), 0)
 }
 
 type deploymentConfigLW struct {

pkg/deploy/registry/deploy/rest.go

@@ -108,7 +108,8 @@ func (s *REST) Update(ctx kubeapi.Context, obj runtime.Object) (<-chan runtime.O
 func (s *REST) Watch(ctx kubeapi.Context, label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
 	return s.registry.WatchDeployments(resourceVersion, func(deployment *deployapi.Deployment) bool {
 		fields := labels.Set{
-			"ID": deployment.ID,
+			"ID":       deployment.ID,
+			"Strategy": deployment.Strategy.Type,
 		}
 		return label.Matches(labels.Set(deployment.Labels)) && field.Matches(fields)
 	})

pkg/deploy/registry/deployconfig/rest.go

@@ -5,11 +5,13 @@ import (
 
 	"code.google.com/p/go-uuid/uuid"
 	kubeapi "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	kubeerrors "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 	deployapi "github.com/openshift/origin/pkg/deploy/api"
+	validation "github.com/openshift/origin/pkg/deploy/api/validation"
 )
 
 // REST is an implementation of RESTStorage for the api server.
@@ -75,7 +77,9 @@ func (s *REST) Create(ctx kubeapi.Context, obj runtime.Object) (<-chan runtime.O
 		deploymentConfig.ID = uuid.NewUUID().String()
 	}
 
-	//TODO: Add validation
+	if errs := validation.ValidateDeploymentConfig(deploymentConfig); len(errs) > 0 {
+		return nil, kubeerrors.NewInvalid("deploymentConfig", deploymentConfig.ID, errs)
+	}
 
 	return apiserver.MakeAsync(func() (runtime.Object, error) {
 		err := s.registry.CreateDeploymentConfig(deploymentConfig)