feat(hatchery): region prerequisite (#5151)

Signed-off-by: Yvonnick Esnault <[email protected]>

docs/content/docs/concepts/requirement/_index.md

@@ -15,6 +15,7 @@ Requirement types:
 - [Service]({{< relref "/docs/concepts/requirement/requirement_service.md" >}})
 - [Memory]({{< relref "/docs/concepts/requirement/requirement_memory.md" >}})
 - [OS & Architecture]({{< relref "/docs/concepts/requirement/requirement_os_arch.md" >}})
+- [Region]({{< relref "/docs/concepts/requirement/requirement_region.md" >}})
 
 A [Job]({{< relref "/docs/concepts/job.md" >}}) will be executed by a **worker**.
 
@@ -26,3 +27,4 @@ You can set as many requirements as you want, following these rules:
 - Only one hostname can be set as requirement
 - Only one OS & Architecture requirement can be set at a time
 - Memory and Services requirements are available only on Docker models
+- Only one region can be set as requirement

docs/content/docs/concepts/requirement/requirement_region.md

@@ -0,0 +1,19 @@
+---
+title: "Region Requirement"
+weight: 4
+---
+
+The `Region` prerequisite allows you to require a worker to have access to a specific region.
+
+A `Region` can be configured on each hatchery. With a free text as `myregion` in hatchery configuration, 
+user can set a prerequisite 'region' with value `myregion` on CDS Job.
+
+Example of job configuration:
+```
+jobs:
+- job: build
+  requirements:
+  - region: myregion
+  steps:
+  ...
+```

engine/api/workermodel/registration.go

@@ -2,7 +2,6 @@ package workermodel
 
 import (
 	"context"
-	"errors"
 	"strconv"
 	"strings"
 	"time"
@@ -24,9 +23,7 @@ const (
 // setting flag need_registration to true in DB.
 func ComputeRegistrationNeeds(db gorp.SqlExecutor, allBinaryReqs sdk.RequirementList, reqs sdk.RequirementList) error {
 	log.Debug("ComputeRegistrationNeeds>")
-	var nbModelReq int
-	var nbOSArchReq int
-	var nbHostnameReq int
+	var nbModelReq, nbOSArchReq, nbHostnameReq, nbRegionReq int
 
 	for _, r := range reqs {
 		switch r.Type {
@@ -47,17 +44,22 @@ func ComputeRegistrationNeeds(db gorp.SqlExecutor, allBinaryReqs sdk.Requirement
 			nbModelReq++
 		case sdk.HostnameRequirement:
 			nbHostnameReq++
+		case sdk.RegionRequirement:
+			nbRegionReq++
 		}
 	}
 
 	if nbOSArchReq > 1 {
-		return sdk.NewError(sdk.ErrWrongRequest, errors.New("invalid os-architecture requirement usage"))
+		return sdk.NewErrorFrom(sdk.ErrWrongRequest, "invalid os-architecture requirement usage")
 	}
 	if nbModelReq > 1 {
-		return sdk.NewError(sdk.ErrWrongRequest, errors.New("invalid model requirement usage"))
+		return sdk.NewErrorFrom(sdk.ErrWrongRequest, "invalid model requirement usage")
 	}
 	if nbHostnameReq > 1 {
-		return sdk.NewError(sdk.ErrWrongRequest, errors.New("invalid hostname requirement usage"))
+		return sdk.NewErrorFrom(sdk.ErrWrongRequest, "invalid hostname requirement usage")
+	}
+	if nbRegionReq > 1 {
+		return sdk.NewErrorFrom(sdk.ErrWrongRequest, "invalid region requirement usage")
 	}
 
 	return nil

engine/hatchery/local/local.go

@@ -178,31 +178,6 @@ func (h *HatcheryLocal) getWorkerBinaryName() string {
 	return workerName
 }
 
-// checkCapabilities checks all requirements, foreach type binary, check if binary is on current host
-// returns an error "Exit status X" if current host misses one requirement
-func (h *HatcheryLocal) checkCapabilities(req []sdk.Requirement) ([]sdk.Requirement, error) {
-	var tmp map[string]sdk.Requirement
-
-	tmp = make(map[string]sdk.Requirement)
-	for _, r := range req {
-		ok, err := h.checkRequirement(r)
-		if err != nil {
-			return nil, err
-		}
-
-		if ok {
-			tmp[r.Name] = r
-		}
-	}
-
-	capa := make([]sdk.Requirement, 0, len(tmp))
-	for _, r := range tmp {
-		capa = append(capa, r)
-	}
-
-	return capa, nil
-}
-
 //Configuration returns Hatchery CommonConfiguration
 func (h *HatcheryLocal) Configuration() service.HatcheryCommonConfiguration {
 	return h.Config.HatcheryCommonConfiguration
@@ -350,6 +325,12 @@ func (h *HatcheryLocal) checkRequirement(r sdk.Requirement) (bool, error) {
 		return true, nil
 	case sdk.PluginRequirement:
 		return true, nil
+	case sdk.RegionRequirement:
+		if r.Value != h.Configuration().Provision.Region {
+			log.Debug("checkRequirement> job with region requirement: cannot spawn. hatchery-region:%s prerequisite:%s", h.Configuration().Provision.Region, r.Value)
+			return false, nil
+		}
+		return true, nil
 	case sdk.OSArchRequirement:
 		osarch := strings.Split(r.Value, "/")
 		if len(osarch) != 2 {

engine/hatchery/marathon/helper_test.go

@@ -1,10 +1,12 @@
 package marathon
 
 import (
+	"time"
+
 	"github.com/gambol99/go-marathon"
-	"github.com/ovh/cds/sdk/cdsclient"
 	"gopkg.in/h2non/gock.v1"
-	"time"
+
+	"github.com/ovh/cds/sdk/cdsclient"
 )
 
 type marathonJDD struct {

engine/hatchery/marathon/marathon.go

@@ -13,14 +13,13 @@ import (
 	"time"
 
 	"github.com/dgrijalva/jwt-go"
-	"github.com/ovh/cds/engine/service"
-
 	"github.com/gambol99/go-marathon"
 	"github.com/gorilla/mux"
 
 	"github.com/ovh/cds/engine/api"
 	"github.com/ovh/cds/engine/api/observability"
 	"github.com/ovh/cds/engine/api/services"
+	"github.com/ovh/cds/engine/service"
 	"github.com/ovh/cds/sdk"
 	"github.com/ovh/cds/sdk/cdsclient"
 	"github.com/ovh/cds/sdk/hatchery"

engine/service/types.go

@@ -40,12 +40,12 @@ type HatcheryCommonConfiguration struct {
 		MaxHeartbeatFailures int    `toml:"maxHeartbeatFailures" default:"10" comment:"Maximum allowed consecutives failures on heatbeat routine" json:"maxHeartbeatFailures"`
 	} `toml:"api" json:"api"`
 	Provision struct {
-		Disabled                  bool `toml:"disabled" default:"false" comment:"Disabled provisioning. Format:true or false" json:"disabled"`
-		RatioService              *int `toml:"ratioService" default:"50" commented:"true" comment:"Percent reserved for spawning worker with service requirement" json:"ratioService,omitempty" mapstructure:"ratioService"`
-		MaxWorker                 int  `toml:"maxWorker" default:"10" comment:"Maximum allowed simultaneous workers" json:"maxWorker"`
-		MaxConcurrentProvisioning int  `toml:"maxConcurrentProvisioning" default:"10" comment:"Maximum allowed simultaneous workers provisioning" json:"maxConcurrentProvisioning"`
-		MaxConcurrentRegistering  int  `toml:"maxConcurrentRegistering" default:"2" comment:"Maximum allowed simultaneous workers registering. -1 to disable registering on this hatchery" json:"maxConcurrentRegistering"`
-		RegisterFrequency         int  `toml:"registerFrequency" default:"60" comment:"Check if some worker model have to be registered each n Seconds" json:"registerFrequency"`
+		RatioService              *int   `toml:"ratioService" default:"50" commented:"true" comment:"Percent reserved for spawning worker with service requirement" json:"ratioService,omitempty" mapstructure:"ratioService"`
+		MaxWorker                 int    `toml:"maxWorker" default:"10" comment:"Maximum allowed simultaneous workers" json:"maxWorker"`
+		MaxConcurrentProvisioning int    `toml:"maxConcurrentProvisioning" default:"10" comment:"Maximum allowed simultaneous workers provisioning" json:"maxConcurrentProvisioning"`
+		MaxConcurrentRegistering  int    `toml:"maxConcurrentRegistering" default:"2" comment:"Maximum allowed simultaneous workers registering. -1 to disable registering on this hatchery" json:"maxConcurrentRegistering"`
+		RegisterFrequency         int    `toml:"registerFrequency" default:"60" comment:"Check if some worker model have to be registered each n Seconds" json:"registerFrequency"`
+		Region                    string `toml:"region" default:"" comment:"region of this hatchery - optional. With a free text as 'myregion', user can set a prerequisite 'region' with value 'myregion' on CDS Job" json:"region"`
 		WorkerLogsOptions         struct {
 			Graylog struct {
 				Host       string `toml:"host" comment:"Example: thot.ovh.com" json:"host"`

engine/worker/internal/requirement.go

@@ -26,6 +26,7 @@ var requirementCheckFuncs = map[string]func(w *CurrentWorker, r sdk.Requirement)
 	sdk.MemoryRequirement:        checkMemoryRequirement,
 	sdk.VolumeRequirement:        checkVolumeRequirement,
 	sdk.OSArchRequirement:        checkOSArchRequirement,
+	sdk.RegionRequirement:        checkRegionRequirement,
 }
 
 func checkRequirements(ctx context.Context, w *CurrentWorker, a *sdk.Action) (bool, []sdk.Requirement) {
@@ -209,6 +210,11 @@ func checkOSArchRequirement(w *CurrentWorker, r sdk.Requirement) (bool, error) {
 	return osarch[0] == strings.ToLower(sdk.GOOS) && osarch[1] == strings.ToLower(sdk.GOARCH), nil
 }
 
+// region is checked by hatchery only
+func checkRegionRequirement(w *CurrentWorker, r sdk.Requirement) (bool, error) {
+	return true, nil
+}
+
 // checkPluginDeployment returns true if current job:
 //  - is not linked to a deployment integration
 //  - is linked to a deployement integration, plugin well downloaded (in this func) and

sdk/exportentities/pipeline.go

@@ -60,6 +60,7 @@ type Requirement struct {
 	Service           ServiceRequirement `json:"service,omitempty" yaml:"service,omitempty"`
 	Memory            string             `json:"memory,omitempty" yaml:"memory,omitempty"`
 	OSArchRequirement string             `json:"os-architecture,omitempty" yaml:"os-architecture,omitempty"`
+	RegionRequirement string             `json:"region,omitempty" yaml:"region,omitempty"`
 }
 
 // ServiceRequirement represents an exported sdk.Requirement of type ServiceRequirement
@@ -151,6 +152,8 @@ func newRequirements(req []sdk.Requirement) []Requirement {
 			res = append(res, Requirement{Service: ServiceRequirement{Name: r.Name, Value: r.Value}})
 		case sdk.OSArchRequirement:
 			res = append(res, Requirement{OSArchRequirement: r.Value})
+		case sdk.RegionRequirement:
+			res = append(res, Requirement{RegionRequirement: r.Value})
 		case sdk.MemoryRequirement:
 			res = append(res, Requirement{Memory: r.Value})
 		}
@@ -222,6 +225,10 @@ func computeJobRequirements(req []Requirement) []sdk.Requirement {
 			name = r.OSArchRequirement
 			val = r.OSArchRequirement
 			tpe = sdk.OSArchRequirement
+		} else if r.RegionRequirement != "" {
+			name = "region"
+			val = r.RegionRequirement
+			tpe = sdk.RegionRequirement
 		} else if r.Plugin != "" {
 			name = r.Plugin
 			val = r.Plugin

sdk/exportentities/pipeline_test.go

@@ -41,6 +41,11 @@ var (
 										Type:  sdk.OSArchRequirement,
 										Value: "freebsd/amd64",
 									},
+									{
+										Name:  sdk.RegionRequirement,
+										Type:  sdk.RegionRequirement,
+										Value: "graxyz",
+									},
 								},
 								Actions: []sdk.Action{
 									{
@@ -542,6 +547,7 @@ jobs:
   requirements:
   - binary: git
   - os-archicture: freebsd/amd64
+  - region: graxyz
   steps:
   - gitClone:
       branch: '{{.git.branch}}'
@@ -568,7 +574,7 @@ jobs:
 	test.NoError(t, err)
 
 	assert.Len(t, p.Stages[0].Jobs[0].Action.Actions, 3)
-	assert.Len(t, p.Stages[0].Jobs[0].Action.Requirements, 2)
+	assert.Len(t, p.Stages[0].Jobs[0].Action.Requirements, 3)
 	assert.Equal(t, sdk.GitCloneAction, p.Stages[0].Jobs[0].Action.Actions[0].Name)
 	assert.Equal(t, sdk.ArtifactUpload, p.Stages[0].Jobs[0].Action.Actions[1].Name)
 	assert.Equal(t, sdk.ServeStaticFiles, p.Stages[0].Jobs[0].Action.Actions[2].Name)

sdk/hatchery/hatchery.go

@@ -278,11 +278,17 @@ func canRunJob(ctx context.Context, h Interface, j workerStarterRequest) bool {
 			return false
 		}
 
+		if r.Type == sdk.RegionRequirement && r.Value != h.Configuration().Provision.Region {
+			log.Debug("canRunJob> %d - job %d - job with region requirement: cannot spawn. hatchery-region:%s prerequisite:%s", j.timestamp, j.id, h.Configuration().Provision.Region, r.Value)
+			return false
+		}
+
 		// Skip others requirement as we can't check it
 		if r.Type == sdk.PluginRequirement || r.Type == sdk.ServiceRequirement || r.Type == sdk.MemoryRequirement {
-			log.Debug("canRunJob> %d - job %d - job with service, plugin, network or memory requirement. Skip these check as we can't checkt it on hatchery routine", j.timestamp, j.id)
+			log.Debug("canRunJob> %d - job %d - job with service, plugin, network or memory requirement. Skip these check as we can't check it on hatchery routine", j.timestamp, j.id)
 			continue
 		}
+
 	}
 	return h.CanSpawn(ctx, nil, j.id, j.requirements)
 }
@@ -293,18 +299,18 @@ const MemoryRegisterContainer int64 = 128
 
 func canRunJobWithModel(ctx context.Context, h InterfaceWithModels, j workerStarterRequest, model *sdk.Model) bool {
 	if model.Type != h.ModelType() {
-		log.Debug("canRunJob> model %s type:%s current hatchery modelType: %s", model.Name, model.Type, h.ModelType())
+		log.Debug("canRunJobWithModel> model %s type:%s current hatchery modelType: %s", model.Name, model.Type, h.ModelType())
 		return false
 	}
 
 	// If the model needs registration, don't spawn for now
 	if h.NeedRegistration(ctx, model) {
-		log.Debug("canRunJob> model %s needs registration", model.Name)
+		log.Debug("canRunJobWithModel> model %s needs registration", model.Name)
 		return false
 	}
 
 	if model.NbSpawnErr > 5 {
-		log.Warning(ctx, "canRunJob> Too many errors on spawn with model %s, please check this worker model", model.Name)
+		log.Warning(ctx, "canRunJobWithModel> Too many errors on spawn with model %s, please check this worker model", model.Name)
 		return false
 	}
 
@@ -317,7 +323,7 @@ func canRunJobWithModel(ctx context.Context, h InterfaceWithModels, j workerStar
 			}
 		}
 		if !checkGroup {
-			log.Debug("canRunJob> job %d - model %s attached to group %d can't run this job", j.id, model.Name, model.GroupID)
+			log.Debug("canRunJobWithModel> job %d - model %s attached to group %d can't run this job", j.id, model.Name, model.GroupID)
 			return false
 		}
 	}
@@ -333,7 +339,7 @@ func canRunJobWithModel(ctx context.Context, h InterfaceWithModels, j workerStar
 	}
 
 	if model.IsDeprecated && !containsModelRequirement {
-		log.Debug("canRunJob> %d - job %d - Cannot launch this model because it is deprecated", j.timestamp, j.id)
+		log.Debug("canRunJobWithModel> %d - job %d - Cannot launch this model because it is deprecated", j.timestamp, j.id)
 		return false
 	}
 
@@ -348,30 +354,35 @@ func canRunJobWithModel(ctx context.Context, h InterfaceWithModels, j workerStar
 			isSharedInfraModel := model.Group.Name == sdk.SharedInfraGroupName && modelName == model.Name
 			isSameName := modelName == model.Name // for backward compatibility with runs, if only the name match we considered that the model can be used, keep this condition until the workflow runs were not migrated.
 			if !isGroupModel && !isSharedInfraModel && !isSameName {
-				log.Debug("canRunJob> %d - job %d - model requirement r.Value(%s) do not match model.Name(%s) and model.Group(%s)", j.timestamp, j.id, strings.Split(r.Value, " ")[0], model.Name, model.Group.Name)
+				log.Debug("canRunJobWithModel> %d - job %d - model requirement r.Value(%s) do not match model.Name(%s) and model.Group(%s)", j.timestamp, j.id, strings.Split(r.Value, " ")[0], model.Name, model.Group.Name)
 				return false
 			}
 		}
 
 		// service and memory requirements are only supported by docker model
 		if model.Type != sdk.Docker && (r.Type == sdk.ServiceRequirement || r.Type == sdk.MemoryRequirement) {
-			log.Debug("canRunJob> %d - job %d - job with service requirement or memory requirement: only for model docker. current model:%s", j.timestamp, j.id, model.Type)
+			log.Debug("canRunJobWithModel> %d - job %d - job with service requirement or memory requirement: only for model docker. current model:%s", j.timestamp, j.id, model.Type)
 			return false
 		}
 
 		if r.Type == sdk.NetworkAccessRequirement && !sdk.CheckNetworkAccessRequirement(r) {
-			log.Debug("canRunJob> %d - job %d - network requirement failed: %v", j.timestamp, j.id, r.Value)
+			log.Debug("canRunJobWithModel> %d - job %d - network requirement failed: %v", j.timestamp, j.id, r.Value)
 			return false
 		}
 
 		// Skip other requirement as we can't check it
 		if r.Type == sdk.PluginRequirement || r.Type == sdk.ServiceRequirement || r.Type == sdk.MemoryRequirement {
-			log.Debug("canRunJob> %d - job %d - job with service, plugin, network or memory requirement. Skip these check as we can't check it on hatchery routine", j.timestamp, j.id)
+			log.Debug("canRunJobWithModel> %d - job %d - job with service, plugin, network or memory requirement. Skip these check as we can't check it on hatchery routine", j.timestamp, j.id)
 			continue
 		}
 
 		if r.Type == sdk.OSArchRequirement && model.RegisteredOS != "" && model.RegisteredArch != "" && r.Value != (model.RegisteredOS+"/"+model.RegisteredArch) {
-			log.Debug("canRunJob> %d - job %d - job with OSArch requirement: cannot spawn on this OSArch. current model: %s/%s", j.timestamp, j.id, model.RegisteredOS, model.RegisteredArch)
+			log.Debug("canRunJobWithModel> %d - job %d - job with OSArch requirement: cannot spawn on this OSArch. current model: %s/%s", j.timestamp, j.id, model.RegisteredOS, model.RegisteredArch)
+			return false
+		}
+
+		if r.Type == sdk.RegionRequirement && r.Value != h.Configuration().Provision.Region {
+			log.Debug("canRunJobWithModel> %d - job %d - job with region requirement: cannot spawn. hatchery-region:%s prerequisite:%s", j.timestamp, j.id, h.Configuration().Provision.Region, r.Value)
 			return false
 		}
 
@@ -387,7 +398,7 @@ func canRunJobWithModel(ctx context.Context, h InterfaceWithModels, j workerStar
 				}
 
 				if !found {
-					log.Debug("canRunJob> %d - job %d - model(%s) does not have binary %s(%s) for this job.", j.timestamp, j.id, model.Name, r.Name, r.Value)
+					log.Debug("canRunJobWithModel> %d - job %d - model(%s) does not have binary %s(%s) for this job.", j.timestamp, j.id, model.Name, r.Name, r.Value)
 					return false
 				}
 			}