feat(#255): make the namespace of workers configurable for k8s cloud (#…

…259)

cloud/k8s.go

@@ -180,7 +180,17 @@ func (cloud *K8SCloud) CanProvision(quota Quota) (bool, error) {
 
 // Provision returns a worker if the cloud can provison
 func (cloud *K8SCloud) Provision(id string, wopts WorkerOptions) (Worker, error) {
-	can, err := cloud.CanProvision(wopts.Quota)
+	var cp *K8SCloud
+	// If specify the namespace for worker in worker options, new a cloud pointer and set its namespace.
+	if len(wopts.Namespace) != 0 {
+		nc := *cloud
+		cp = &nc
+		cp.namespace = wopts.Namespace
+	} else {
+		cp = cloud
+	}
+
+	can, err := cp.CanProvision(wopts.Quota)
 	if err != nil {
 		return nil, err
 	}
@@ -194,7 +204,7 @@ func (cloud *K8SCloud) Provision(id string, wopts WorkerOptions) (Worker, error)
 	Privileged := true
 	pod := &apiv1.Pod{
 		ObjectMeta: apiv1.ObjectMeta{
-			Namespace: cloud.namespace,
+			Namespace: cp.namespace,
 			Name:      name,
 			Labels: map[string]string{
 				"cyclone":    "worker",
@@ -222,7 +232,7 @@ func (cloud *K8SCloud) Provision(id string, wopts WorkerOptions) (Worker, error)
 	// }
 
 	worker := &K8SPodWorker{
-		K8SCloud: cloud,
+		K8SCloud: cp,
 		pod:      pod,
 	}
 

cloud/worker.go

@@ -218,6 +218,9 @@ type WorkerOptions struct {
 	WorkerEnvs *WorkerEnvs
 
 	Quota Quota
+
+	// Namespace represents the k8s namespace where to create worker, only works for k8s cloud provider.
+	Namespace string
 }
 
 // NewWorkerOptions creates a new WorkerOptions with default value

event/event.go

@@ -147,6 +147,7 @@ func generateAbortedStateLog(log string) string {
 func createServiceHandler(event *api.Event) error {
 	logdog.Infof("create service handler")
 	opts := workerOptions.DeepCopy()
+	opts.Namespace = getNamespaceFromEvent(event)
 	worker, err := CloudController.Provision(string(event.EventID), opts)
 	if err != nil {
 		return err
@@ -235,6 +236,7 @@ func createVersionHandler(event *api.Event) error {
 
 	opts := workerOptions.DeepCopy()
 	opts.Quota = Resource2Quota(event.Version.BuildResource, opts.Quota)
+	opts.Namespace = getNamespaceFromEvent(event)
 	worker, err := CloudController.Provision(string(event.EventID), opts)
 	if err != nil {
 		return err
@@ -396,3 +398,19 @@ func Resource2Quota(resource api.BuildResource, def cloud.Quota) cloud.Quota {
 
 	return quota
 }
+
+// getNamespaceFromEvent gets the namespace from event data for worker. Will return empty string if can not get it.
+func getNamespaceFromEvent(event *api.Event) string {
+	if event == nil {
+		log.Error("Can not get namespace from data as event is nil")
+		return ""
+	}
+
+	if v, e := event.Data["namespace"]; e {
+		if sv, ok := v.(string); ok {
+			return sv
+		}
+	}
+
+	return ""
+}

event/operations.go

@@ -49,7 +49,12 @@ func SendCreateServiceEvent(service *api.Service) error {
 		Data:      map[string]interface{}{"Token": project.SCM.Token},
 	}
 
-	// log.Infof("send create service event: %v", event)
+	// Set the namespace for worker in k8s cloud.
+	if project.Worker != nil && len(project.Worker.Namespace) != 0 {
+		event.Data["namespace"] = project.Worker.Namespace
+	}
+
+	log.Infof("send create service event: %v", event)
 
 	etcdClient := etcd.GetClient()
 	jsEvent, err := json.Marshal(&event)
@@ -99,7 +104,12 @@ func SendCreateVersionEvent(service *api.Service, version *api.Version) error {
 		Status: api.EventStatusPending,
 	}
 
-	// log.Infof("send create version event: %v", event)
+	// Set the namespace for worker in k8s cloud.
+	if project.Worker != nil && len(project.Worker.Namespace) != 0 {
+		event.Data["namespace"] = project.Worker.Namespace
+	}
+
+	log.Infof("send create version event: %v", event)
 
 	etcdClient := etcd.GetClient()
 	jsEvent, err := json.Marshal(&event)

pkg/api/types.go

@@ -29,10 +29,16 @@ type Project struct {
 	Description    string     `bson:"description,omitempty" json:"description,omitempty" description:"description of the project"`
 	Owner          string     `bson:"owner,omitempty" json:"owner,omitempty" description:"owner of the project"`
 	SCM            *SCMConfig `bson:"scm,omitempty" json:"scm,omitempty" description:"scm config of the project"`
+	Worker         *Worker    `bson:"worker,omitempty" json:"worker,omitempty" description:"worker config of the project"`
 	CreationTime   time.Time  `bson:"creationTime,omitempty" json:"creationTime,omitempty" description:"creation time of the project"`
 	LastUpdateTime time.Time  `bson:"lastUpdateTime,omitempty" json:"lastUpdateTime,omitempty" description:"last update time of the project"`
 }
 
+// Worker represents the config of worker for the pipelines of the project.
+type Worker struct {
+	Namespace string `bson:"namespace,omitempty" json:"namespace,omitempty" description:"k8s namespace to create the worker"`
+}
+
 // Pipeline represents a set of configs to describe the workflow of CI/CD.
 type Pipeline struct {
 	ID          string `bson:"_id,omitempty" json:"id,omitempty" description:"id of the pipeline"`

pkg/server/manager/project.go

@@ -117,6 +117,8 @@ func (m *projectManager) UpdateProject(projectName string, newProject *api.Proje
 		project.Owner = newProject.Owner
 	}
 
+	project.Worker = newProject.Worker
+
 	if err = m.dataStore.UpdateProject(project); err != nil {
 		return nil, err
 	}