Skip to content

Commit f9d8834

Browse files
jagednabhi18av
jagedn
and
abhi18av
authored
add Constraints (Node and Attr) (#66)
* implement ConstraintNodeSpec Signed-off-by: Jorge Aguilera <[email protected]> * implement ConstraintAttrSpec Signed-off-by: Jorge Aguilera <[email protected]> * Constraints refactor to introduce a specific models package (#68) * refactor the package organization to create a separate package for models * Refactor the test folder and add license headers * move contraintsbuilder to model Signed-off-by: Jorge Aguilera <[email protected]> * refactor and new "raw" dsl to include open constraints Signed-off-by: Jorge Aguilera <[email protected]> * iterate upon the constraint config [ci skip] * fix nomadlab typos Signed-off-by: Jorge Aguilera <[email protected]> * use cloud cache in sun-nomadlab [ci skip] --------- Signed-off-by: Jorge Aguilera <[email protected]> Co-authored-by: Abhinav Sharma <[email protected]> Co-authored-by: Abhinav Sharma <[email protected]>
1 parent 6ee5725 commit f9d8834

20 files changed

+908
-62
lines changed

‎gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
1-
version=0.1.1
1+
version=0.1.2
22
github_organization=nextflow-io

‎plugins/nf-nomad/src/main/nextflow/nomad/config/NomadJobOpts.groovy‎

Lines changed: 35 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,10 @@ package nextflow.nomad.config
1919

2020
import groovy.transform.CompileStatic
2121
import groovy.util.logging.Slf4j
22+
import nextflow.nomad.models.JobAffinity
23+
import nextflow.nomad.models.JobConstraint
24+
import nextflow.nomad.models.JobConstraints
25+
import nextflow.nomad.models.JobVolume
2226

2327

2428
/**
@@ -37,9 +41,11 @@ class NomadJobOpts{
3741
String region
3842
String namespace
3943
String dockerVolume
40-
VolumeSpec[] volumeSpec
41-
AffinitySpec affinitySpec
42-
ConstraintSpec constraintSpec
44+
JobVolume[] volumeSpec
45+
JobAffinity affinitySpec
46+
JobConstraint constraintSpec
47+
48+
JobConstraints constraintsSpec
4349

4450
NomadJobOpts(Map nomadJobOpts, Map<String,String> env=null){
4551
assert nomadJobOpts!=null
@@ -69,12 +75,13 @@ class NomadJobOpts{
6975
this.volumeSpec = parseVolumes(nomadJobOpts)
7076
this.affinitySpec = parseAffinity(nomadJobOpts)
7177
this.constraintSpec = parseConstraint(nomadJobOpts)
78+
this.constraintsSpec = parseConstraints(nomadJobOpts)
7279
}
7380

74-
VolumeSpec[] parseVolumes(Map nomadJobOpts){
75-
List<VolumeSpec> ret = []
81+
JobVolume[] parseVolumes(Map nomadJobOpts){
82+
List<JobVolume> ret = []
7683
if( nomadJobOpts.volume && nomadJobOpts.volume instanceof Closure){
77-
def volumeSpec = new VolumeSpec()
84+
def volumeSpec = new JobVolume()
7885
def closure = (nomadJobOpts.volume as Closure)
7986
def clone = closure.rehydrate(volumeSpec, closure.owner, closure.thisObject)
8087
clone.resolveStrategy = Closure.DELEGATE_FIRST
@@ -86,7 +93,7 @@ class NomadJobOpts{
8693
if( nomadJobOpts.volumes && nomadJobOpts.volumes instanceof List){
8794
nomadJobOpts.volumes.each{ closure ->
8895
if( closure instanceof Closure){
89-
def volumeSpec = new VolumeSpec()
96+
def volumeSpec = new JobVolume()
9097
def clone = closure.rehydrate(volumeSpec, closure.owner, closure.thisObject)
9198
clone.resolveStrategy = Closure.DELEGATE_FIRST
9299
clone()
@@ -105,12 +112,13 @@ class NomadJobOpts{
105112
throw new IllegalArgumentException("No more than a workdir volume allowed")
106113
}
107114

108-
return ret as VolumeSpec[]
115+
return ret as JobVolume[]
109116
}
110117

111-
AffinitySpec parseAffinity(Map nomadJobOpts) {
118+
JobAffinity parseAffinity(Map nomadJobOpts) {
112119
if (nomadJobOpts.affinity && nomadJobOpts.affinity instanceof Closure) {
113-
def affinitySpec = new AffinitySpec()
120+
log.info "affinity config will be deprecated, use affinities closure instead"
121+
def affinitySpec = new JobAffinity()
114122
def closure = (nomadJobOpts.affinity as Closure)
115123
def clone = closure.rehydrate(affinitySpec, closure.owner, closure.thisObject)
116124
clone.resolveStrategy = Closure.DELEGATE_FIRST
@@ -122,9 +130,10 @@ class NomadJobOpts{
122130
}
123131
}
124132

125-
ConstraintSpec parseConstraint(Map nomadJobOpts){
133+
JobConstraint parseConstraint(Map nomadJobOpts){
126134
if (nomadJobOpts.constraint && nomadJobOpts.constraint instanceof Closure) {
127-
def constraintSpec = new ConstraintSpec()
135+
log.info "constraint config will be deprecated, use constraints closure instead"
136+
def constraintSpec = new JobConstraint()
128137
def closure = (nomadJobOpts.constraint as Closure)
129138
def clone = closure.rehydrate(constraintSpec, closure.owner, closure.thisObject)
130139
clone.resolveStrategy = Closure.DELEGATE_FIRST
@@ -135,4 +144,18 @@ class NomadJobOpts{
135144
null
136145
}
137146
}
147+
148+
JobConstraints parseConstraints(Map nomadJobOpts){
149+
if (nomadJobOpts.constraints && nomadJobOpts.constraints instanceof Closure) {
150+
def constraintsSpec = new JobConstraints()
151+
def closure = (nomadJobOpts.constraints as Closure)
152+
def clone = closure.rehydrate(constraintsSpec, closure.owner, closure.thisObject)
153+
clone.resolveStrategy = Closure.DELEGATE_FIRST
154+
clone()
155+
constraintsSpec.validate()
156+
constraintsSpec
157+
}else{
158+
null
159+
}
160+
}
138161
}

‎plugins/nf-nomad/src/main/nextflow/nomad/executor/NomadService.groovy‎

Lines changed: 33 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,10 @@ import groovy.util.logging.Slf4j
2222
import io.nomadproject.client.ApiClient
2323
import io.nomadproject.client.api.JobsApi
2424
import io.nomadproject.client.model.*
25+
import nextflow.nomad.models.ConstraintsBuilder
26+
import nextflow.nomad.models.JobConstraints
2527
import nextflow.nomad.config.NomadConfig
26-
import nextflow.nomad.config.VolumeSpec
28+
import nextflow.nomad.models.JobVolume
2729
import nextflow.processor.TaskRun
2830
import nextflow.util.MemoryUnit
2931
import nextflow.exception.ProcessSubmitException
@@ -135,7 +137,7 @@ class NomadService implements Closeable{
135137
if( config.jobOpts().volumeSpec ) {
136138
taskGroup.volumes = [:]
137139
config.jobOpts().volumeSpec.eachWithIndex { volumeSpec , idx->
138-
if (volumeSpec && volumeSpec.type == VolumeSpec.VOLUME_CSI_TYPE) {
140+
if (volumeSpec && volumeSpec.type == JobVolume.VOLUME_CSI_TYPE) {
139141
taskGroup.volumes["vol_${idx}".toString()] = new VolumeRequest(
140142
type: volumeSpec.type,
141143
source: volumeSpec.name,
@@ -145,7 +147,7 @@ class NomadService implements Closeable{
145147
)
146148
}
147149

148-
if (volumeSpec && volumeSpec.type == VolumeSpec.VOLUME_HOST_TYPE) {
150+
if (volumeSpec && volumeSpec.type == JobVolume.VOLUME_HOST_TYPE) {
149151
taskGroup.volumes["vol_${idx}".toString()] = new VolumeRequest(
150152
type: volumeSpec.type,
151153
source: volumeSpec.name,
@@ -182,7 +184,8 @@ class NomadService implements Closeable{
182184

183185
volumes(task, taskDef, workingDir)
184186
affinity(task, taskDef)
185-
constrains(task, taskDef)
187+
constraint(task, taskDef)
188+
constraints(task, taskDef)
186189

187190
return taskDef
188191
}
@@ -233,7 +236,7 @@ class NomadService implements Closeable{
233236
taskDef
234237
}
235238

236-
protected Task constrains(TaskRun task, Task taskDef){
239+
protected Task constraint(TaskRun task, Task taskDef){
237240
if( config.jobOpts().constraintSpec ){
238241
def constraint = new Constraint()
239242
if(config.jobOpts().constraintSpec.attribute){
@@ -251,8 +254,32 @@ class NomadService implements Closeable{
251254
taskDef
252255
}
253256

257+
protected Task constraints(TaskRun task, Task taskDef){
258+
def constraints = [] as List<Constraint>
259+
260+
if( config.jobOpts().constraintsSpec ){
261+
def list = ConstraintsBuilder.constraintsSpecToList(config.jobOpts().constraintsSpec)
262+
constraints.addAll(list)
263+
}
264+
265+
if( task.processor?.config?.get(TaskDirectives.CONSTRAINTS) &&
266+
task.processor?.config?.get(TaskDirectives.CONSTRAINTS) instanceof Closure) {
267+
Closure closure = task.processor?.config?.get(TaskDirectives.CONSTRAINTS) as Closure
268+
JobConstraints constraintsSpec = JobConstraints.parse(closure)
269+
def list = ConstraintsBuilder.constraintsSpecToList(constraintsSpec)
270+
constraints.addAll(list)
271+
}
272+
273+
if( constraints.size()) {
274+
taskDef.constraints(constraints)
275+
}
276+
taskDef
277+
}
278+
279+
280+
254281
protected Job assignDatacenters(TaskRun task, Job job){
255-
def datacenters = task.processor?.config?.get("datacenters")
282+
def datacenters = task.processor?.config?.get(TaskDirectives.DATACENTERS)
256283
if( datacenters ){
257284
if( datacenters instanceof List<String>) {
258285
job.datacenters( datacenters as List<String>)

‎plugins/nf-nomad/src/main/nextflow/nomad/executor/TaskDirectives.groovy‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,10 @@ class TaskDirectives {
44

55
public static final String DATACENTERS = "datacenters"
66

7+
public static final String CONSTRAINTS = "constraints"
8+
79
public static final List<String> ALL = [
8-
DATACENTERS
10+
DATACENTERS,
11+
CONSTRAINTS
912
]
1013
}

‎plugins/nf-nomad/src/main/nextflow/nomad/models/ConstraintsBuilder.groovy‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
package nextflow.nomad.models
2+
3+
import io.nomadproject.client.model.Constraint
4+
import nextflow.nomad.models.JobConstraintsAttr
5+
import nextflow.nomad.models.JobConstraintsNode
6+
import nextflow.nomad.models.JobConstraints
7+
8+
class ConstraintsBuilder {
9+
10+
static List<Constraint> constraintsSpecToList(JobConstraints spec){
11+
def constraints = [] as List<Constraint>
12+
if( spec?.nodeSpecs ){
13+
def nodes = spec.nodeSpecs
14+
?.collect({ nodeConstraints(it)})
15+
?.flatten() as List<Constraint>
16+
constraints.addAll(nodes)
17+
}
18+
if( spec?.attrSpecs ){
19+
def nodes = spec.attrSpecs
20+
?.collect({ attrConstraints(it)})
21+
?.flatten() as List<Constraint>
22+
constraints.addAll(nodes)
23+
}
24+
return constraints
25+
}
26+
27+
protected static List<Constraint> nodeConstraints(JobConstraintsNode nodeSpec){
28+
def ret = nodeSpec.raws?.collect{ triple->
29+
return new Constraint()
30+
.ltarget('${'+triple.left+'}')
31+
.operand(triple.middle)
32+
.rtarget(triple.right)
33+
} as List<Constraint>
34+
ret
35+
}
36+
37+
protected static List<Constraint> attrConstraints(JobConstraintsAttr nodeSpec) {
38+
def ret = nodeSpec.raws?.collect{ triple->
39+
return new Constraint()
40+
.ltarget('${'+triple.left+'}')
41+
.operand(triple.middle)
42+
.rtarget(triple.right)
43+
} as List<Constraint>
44+
ret
45+
}
46+
47+
}

‎plugins/nf-nomad/src/main/nextflow/nomad/config/AffinitySpec.groovy‎ renamed to ‎plugins/nf-nomad/src/main/nextflow/nomad/models/JobAffinity.groovy‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,13 @@
1515
* limitations under the License.
1616
*/
1717

18-
package nextflow.nomad.config
18+
package nextflow.nomad.models
1919
/**
2020
* Nomad Job Affinity Spec
2121
*
2222
* @author Jorge Aguilera <[email protected]>
2323
*/
24-
class AffinitySpec{
24+
class JobAffinity {
2525

2626
private String attribute
2727
private String operator
@@ -44,22 +44,22 @@ class AffinitySpec{
4444
return weight
4545
}
4646

47-
AffinitySpec attribute(String attribute){
47+
JobAffinity attribute(String attribute){
4848
this.attribute=attribute
4949
this
5050
}
5151

52-
AffinitySpec operator(String operator){
52+
JobAffinity operator(String operator){
5353
this.operator = operator
5454
this
5555
}
5656

57-
AffinitySpec value(String value){
57+
JobAffinity value(String value){
5858
this.value = value
5959
this
6060
}
6161

62-
AffinitySpec weight(int weight){
62+
JobAffinity weight(int weight){
6363
this.weight = weight
6464
this
6565
}

‎plugins/nf-nomad/src/main/nextflow/nomad/config/ConstraintSpec.groovy‎ renamed to ‎plugins/nf-nomad/src/main/nextflow/nomad/models/JobConstraint.groovy‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,14 @@
1515
* limitations under the License.
1616
*/
1717

18-
package nextflow.nomad.config
18+
package nextflow.nomad.models
1919
/**
2020
* Nomad Job Constraint Spec
2121
*
2222
* @author Jorge Aguilera <[email protected]>
2323
*/
2424

25-
class ConstraintSpec {
25+
class JobConstraint {
2626

2727
private String attribute
2828
private String operator
@@ -40,17 +40,17 @@ class ConstraintSpec {
4040
return value
4141
}
4242

43-
ConstraintSpec attribute(String attribute){
43+
JobConstraint attribute(String attribute){
4444
this.attribute=attribute
4545
this
4646
}
4747

48-
ConstraintSpec operator(String operator){
48+
JobConstraint operator(String operator){
4949
this.operator = operator
5050
this
5151
}
5252

53-
ConstraintSpec value(String value){
53+
JobConstraint value(String value){
5454
this.value = value
5555
this
5656
}

‎plugins/nf-nomad/src/main/nextflow/nomad/models/JobConstraints.groovy‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
/*
2+
* Copyright 2023-, Stellenbosch University, South Africa
3+
* Copyright 2024, Evaluacion y Desarrollo de Negocios, Spain
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package nextflow.nomad.models
19+
20+
/**
21+
* Nomad Job Constraint Spec
22+
*
23+
* @author Jorge Aguilera <[email protected]>
24+
*/
25+
26+
class JobConstraints {
27+
28+
List<JobConstraintsNode> nodeSpecs = []
29+
List<JobConstraintsAttr> attrSpecs = []
30+
31+
JobConstraints node(@DelegatesTo(JobConstraintsNode)Closure closure){
32+
JobConstraintsNode constraintSpec = new JobConstraintsNode()
33+
def clone = closure.rehydrate(constraintSpec, closure.owner, closure.thisObject)
34+
clone.resolveStrategy = Closure.DELEGATE_FIRST
35+
clone()
36+
nodeSpecs << constraintSpec
37+
this
38+
}
39+
40+
JobConstraints attr(@DelegatesTo(JobConstraintsAttr)Closure closure){
41+
JobConstraintsAttr constraintSpec = new JobConstraintsAttr()
42+
def clone = closure.rehydrate(constraintSpec, closure.owner, closure.thisObject)
43+
clone.resolveStrategy = Closure.DELEGATE_FIRST
44+
clone()
45+
attrSpecs << constraintSpec
46+
this
47+
}
48+
49+
void validate(){
50+
51+
}
52+
53+
static JobConstraints parse(@DelegatesTo(JobConstraints)Closure closure){
54+
JobConstraints constraintsSpec = new JobConstraints()
55+
def clone = closure.rehydrate(constraintsSpec, closure.owner, closure.thisObject)
56+
clone.resolveStrategy = Closure.DELEGATE_FIRST
57+
clone()
58+
constraintsSpec.validate()
59+
constraintsSpec
60+
}
61+
}

0 commit comments

Comments
 (0)