Skip to content

Commit db79c43

Browse files
pditommasojorgee
pditommaso
and
jorgee
authored
Add serde interfaces (#5893) [ci skip]
Signed-off-by: Paolo Di Tommaso <[email protected]> Signed-off-by: jorgee <[email protected]> Co-authored-by: jorgee <[email protected]>
1 parent c04a58f commit db79c43

32 files changed

+1149
-231
lines changed

‎modules/nextflow/src/test/groovy/nextflow/cli/CmdCidTest.groovy‎

Lines changed: 32 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -17,20 +17,22 @@
1717

1818
package nextflow.cli
1919

20-
import groovy.json.JsonOutput
21-
2220
import java.nio.file.Files
2321

2422
import nextflow.SysEnv
2523
import nextflow.dag.MermaidHtmlRenderer
2624
import nextflow.data.cid.CidHistoryRecord
2725
import nextflow.data.cid.CidStoreFactory
26+
import nextflow.data.cid.model.Checksum
27+
import nextflow.data.cid.model.Parameter
28+
import nextflow.data.cid.model.TaskOutput
29+
import nextflow.data.cid.model.TaskRun
30+
import nextflow.data.cid.model.WorkflowOutput
31+
import nextflow.data.cid.serde.CidEncoder
2832
import nextflow.plugin.Plugins
29-
3033
import org.junit.Rule
3134
import spock.lang.Specification
3235
import test.OutputCapture
33-
3436
/**
3537
* CLI cid Tests
3638
*
@@ -130,16 +132,12 @@ class CmdCidTest extends Specification {
130132
def launcher = Mock(Launcher){
131133
getOptions() >> new CliOptions(config: [configFile.toString()])
132134
}
133-
134-
def recordEntry = JsonOutput.prettyPrint('{"type":"WorkflowOutput",' +
135-
'"path":"/path/to/file",' +
136-
'"checksum":"45372qe",' +
137-
'"source":"cid://123987/file.bam",' +
138-
'"size": 1234,' +
139-
'"createdAt": 123456789,' +
140-
'"modifiedAt": 123456789,' +
141-
'"annotations":null}')
142-
cidFile.text = recordEntry
135+
def encoder = new CidEncoder().withPrettyPrint(true)
136+
def entry = new WorkflowOutput("path/to/file",new Checksum("45372qe","nextflow","standard"),
137+
"cid://123987/file.bam", 1234, 123456789, 123456789, null)
138+
def jsonSer = encoder.encode(entry)
139+
def expectedOutput = jsonSer
140+
cidFile.text = jsonSer
143141
when:
144142
def cidCmd = new CmdCid(launcher: launcher, args: ["show", "cid://12345"])
145143
cidCmd.run()
@@ -151,8 +149,8 @@ class CmdCidTest extends Specification {
151149
.findResults { line -> !line.contains('plugin') ? line : null }
152150

153151
then:
154-
stdout.size() == recordEntry.readLines().size()
155-
stdout.join('\n') == recordEntry
152+
stdout.size() == expectedOutput.readLines().size()
153+
stdout.join('\n') == expectedOutput
156154

157155
cleanup:
158156
folder?.deleteDir()
@@ -204,31 +202,24 @@ class CmdCidTest extends Specification {
204202
Files.createDirectories(cidFile3.parent)
205203
Files.createDirectories(cidFile4.parent)
206204
Files.createDirectories(cidFile5.parent)
207-
208-
def recordEntry = JsonOutput.prettyPrint('{"type":"WorkflowOutput",' +
209-
'"path":"/path/to/file","checksum":"45372qe","source":"cid://123987/file.bam",' +
210-
'"size": 1234,"createdAt": 123456789, "modifiedAt": 123456789,"annotations":null}')
211-
cidFile.text = recordEntry
212-
recordEntry = JsonOutput.prettyPrint('{"type":"TaskOutput",' +
213-
'"path":"/path/to/file","checksum":"45372qe","source":"cid://123987",' +
214-
'"size": 1234,"createdAt": 123456789,"modifiedAt": 123456789,"annotations":null}')
215-
cidFile2.text = recordEntry
216-
recordEntry = JsonOutput.prettyPrint('{"type":"TaskRun",' +
217-
'"sessionId":"u345-2346-1stw2", "name":"foo","code":"abcde2345",' +
218-
'"inputs": [{"type": "ValueInParam","name": "sample_id","value": "ggal_gut"},' +
219-
'{"type": "FileInParam","name": "reads","value": ["cid://45678/output.txt"]}],' +
220-
'"container": null,"conda": null,"spack": null,"architecture": null,' +
221-
'"globalVars": {},"binEntries": [],"annotations":null}')
222-
cidFile3.text = recordEntry
223-
recordEntry = JsonOutput.prettyPrint('{"type":"TaskOutput",' +
224-
'"path":"/path/to/file","checksum":"45372qe","source":"cid://45678",' +
225-
'"size": 1234,"createdAt": 123456789,"modifiedAt": 123456789,"annotations":null}')
226-
cidFile4.text = recordEntry
227-
recordEntry = JsonOutput.prettyPrint('{"type":"TaskRun",' +
228-
'"sessionId":"u345-2346-1stw2", "name":"bar","code":"abfs2556",' +
229-
'"inputs": null,"container": null,"conda": null,"spack": null,"architecture": null,' +
230-
'"globalVars": {},"binEntries": [],"annotations":null}')
231-
cidFile5.text = recordEntry
205+
def encoder = new CidEncoder()
206+
def entry = new WorkflowOutput("path/to/file",new Checksum("45372qe","nextflow","standard"),
207+
"cid://123987/file.bam", 1234, 123456789, 123456789, null)
208+
cidFile.text = encoder.encode(entry)
209+
entry = new TaskOutput("path/to/file",new Checksum("45372qe","nextflow","standard"),
210+
"cid://123987", 1234, 123456789, 123456789, null)
211+
cidFile2.text = encoder.encode(entry)
212+
entry = new TaskRun("u345-2346-1stw2", "foo", new Checksum("abcde2345","nextflow","standard"),
213+
[new Parameter( "ValueInParam", "sample_id","ggal_gut"),
214+
new Parameter("FileInParam","reads",["cid://45678/output.txt"])],
215+
null, null, null, null, [:],[], null)
216+
cidFile3.text = encoder.encode(entry)
217+
entry = new TaskOutput("path/to/file",new Checksum("45372qe","nextflow","standard"),
218+
"cid://45678", 1234, 123456789, 123456789, null)
219+
cidFile4.text = encoder.encode(entry)
220+
entry = new TaskRun("u345-2346-1stw2", "bar", new Checksum("abfs2556","nextflow","standard"),
221+
null,null, null, null, null, [:],[], null)
222+
cidFile5.text = encoder.encode(entry)
232223
final network = """flowchart BT
233224
cid://12345/file.bam@{shape: document, label: "cid://12345/file.bam"}
234225
cid://123987/file.bam@{shape: document, label: "cid://123987/file.bam"}

‎modules/nf-cid/src/main/nextflow/data/cid/CidObserver.groovy‎

Lines changed: 37 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -17,40 +17,38 @@
1717

1818
package nextflow.data.cid
1919

20+
import static nextflow.data.cid.fs.CidPath.*
21+
22+
import java.nio.file.Files
23+
import java.nio.file.Path
24+
import java.nio.file.attribute.BasicFileAttributes
25+
26+
import groovy.transform.CompileStatic
2027
import groovy.util.logging.Slf4j
28+
import nextflow.Session
2129
import nextflow.data.cid.model.Checksum
2230
import nextflow.data.cid.model.DataPath
2331
import nextflow.data.cid.model.Parameter
24-
import nextflow.data.cid.model.WorkflowResults
32+
import nextflow.data.cid.model.TaskOutput
2533
import nextflow.data.cid.model.Workflow
34+
import nextflow.data.cid.model.WorkflowOutput
35+
import nextflow.data.cid.model.WorkflowResults
2636
import nextflow.data.cid.model.WorkflowRun
37+
import nextflow.data.cid.serde.CidEncoder
2738
import nextflow.file.FileHelper
2839
import nextflow.file.FileHolder
40+
import nextflow.processor.TaskHandler
41+
import nextflow.processor.TaskRun
2942
import nextflow.script.ScriptMeta
3043
import nextflow.script.params.DefaultInParam
3144
import nextflow.script.params.FileInParam
32-
import nextflow.script.params.InParam
33-
import nextflow.util.PathNormalizer
34-
import nextflow.util.TestOnly
35-
36-
import java.nio.file.Files
37-
import java.nio.file.Path
38-
import java.nio.file.attribute.BasicFileAttributes
39-
40-
import groovy.json.JsonOutput
41-
import groovy.transform.CompileStatic
42-
import nextflow.Session
43-
import nextflow.data.cid.model.DataType
44-
import nextflow.data.cid.model.Output
45-
import nextflow.processor.TaskHandler
46-
import nextflow.processor.TaskRun
4745
import nextflow.script.params.FileOutParam
46+
import nextflow.script.params.InParam
4847
import nextflow.trace.TraceObserver
4948
import nextflow.trace.TraceRecord
5049
import nextflow.util.CacheHelper
51-
52-
import static nextflow.data.cid.fs.CidPath.CID_PROT
53-
50+
import nextflow.util.PathNormalizer
51+
import nextflow.util.TestOnly
5452
/**
5553
* Observer to write the generated workflow metadata in a CID store.
5654
*
@@ -65,6 +63,7 @@ class CidObserver implements TraceObserver {
6563
private Session session
6664
private WorkflowResults workflowResults
6765
private Map<String,String> outputsStoreDirCid = new HashMap<String,String>(10)
66+
private CidEncoder encoder = new CidEncoder()
6867

6968
CidObserver(Session session, CidStore store){
7069
this.session = session
@@ -83,18 +82,17 @@ class CidObserver implements TraceObserver {
8382
void onFlowBegin() {
8483
this.executionHash = storeWorkflowRun()
8584
workflowResults = new WorkflowResults(
86-
DataType.WorkflowResults,
8785
"$CID_PROT${executionHash}",
88-
new ArrayList<Parameter>())
86+
new ArrayList<String>())
8987
this.store.getHistoryLog().updateRunCid(session.uniqueId, "${CID_PROT}${this.executionHash}")
9088
}
9189

9290
@Override
9391
void onFlowComplete(){
9492
if (this.workflowResults){
95-
final content = JsonOutput.prettyPrint(JsonOutput.toJson(workflowResults))
96-
final wfResultsHash = CacheHelper.hasher(content).hash().toString()
97-
this.store.save(wfResultsHash, content)
93+
final json = encoder.encode(workflowResults)
94+
final wfResultsHash = CacheHelper.hasher(json).hash().toString()
95+
this.store.save(wfResultsHash, workflowResults)
9896
this.store.getHistoryLog().updateResultsCid(session.uniqueId, "${CID_PROT}${wfResultsHash}")
9997
}
10098
}
@@ -121,23 +119,21 @@ class CidObserver implements TraceObserver {
121119
}
122120
}
123121
final workflow = new Workflow(
124-
DataType.Workflow,
125122
mainScript,
126123
otherScripts,
127124
session.workflowMetadata.repository,
128125
session.workflowMetadata.commitId
129126
)
130127
final value = new WorkflowRun(
131-
DataType.WorkflowRun,
132128
workflow,
133129
session.uniqueId.toString(),
134130
session.runName,
135131
getNormalizedParams(session.params, normalizer)
136132
)
137133

138-
final content = JsonOutput.prettyPrint(JsonOutput.toJson(value))
139-
final executionHash = CacheHelper.hasher(content).hash().toString()
140-
store.save(executionHash, content)
134+
final json = encoder.encode(value)
135+
final executionHash = CacheHelper.hasher(json).hash().toString()
136+
store.save(executionHash, value)
141137
return executionHash
142138
}
143139

@@ -184,7 +180,6 @@ class CidObserver implements TraceObserver {
184180
final codeChecksum = new Checksum(CacheHelper.hasher(session.stubRun ? task.stubSource: task.source).hash().toString(),
185181
"nextflow", CacheHelper.HashMode.DEFAULT().toString().toLowerCase())
186182
final value = new nextflow.data.cid.model.TaskRun(
187-
DataType.TaskRun,
188183
session.uniqueId.toString(),
189184
task.getName(),
190185
codeChecksum,
@@ -203,7 +198,7 @@ class CidObserver implements TraceObserver {
203198

204199
// store in the underlying persistence
205200
final key = task.hash.toString()
206-
store.save(key, JsonOutput.prettyPrint(JsonOutput.toJson(value)))
201+
store.save(key, value)
207202
return key
208203
}
209204

@@ -215,15 +210,14 @@ class CidObserver implements TraceObserver {
215210
final key = cid.toString()
216211
final checksum = new Checksum( CacheHelper.hasher(path).hash().toString(),
217212
"nextflow", CacheHelper.HashMode.DEFAULT().toString().toLowerCase() )
218-
final value = new Output(
219-
DataType.TaskOutput,
213+
final value = new TaskOutput(
220214
path.toUriString(),
221215
checksum,
222216
"$CID_PROT$task.hash",
223217
attrs.size(),
224218
attrs.creationTime().toMillis(),
225219
attrs.lastModifiedTime().toMillis())
226-
store.save(key, JsonOutput.prettyPrint(JsonOutput.toJson(value)))
220+
store.save(key, value)
227221
} catch (Throwable e) {
228222
log.warn("Exception storing CID output $path for task ${task.name}. ${e.getLocalizedMessage()}")
229223
}
@@ -273,20 +267,20 @@ class CidObserver implements TraceObserver {
273267
CacheHelper.HashMode.DEFAULT().toString().toLowerCase()
274268
)
275269
final rel = getWorkflowRelative(destination)
276-
final key = "$executionHash/${rel}"
270+
final key = "$executionHash/${rel}" as String
277271
final sourceReference = getSourceReference(source)
278272
final attrs = readAttributes(destination)
279-
final value = new Output(
280-
DataType.WorkflowOutput,
273+
final value = new WorkflowOutput(
281274
destination.toUriString(),
282275
checksum,
283276
sourceReference,
284277
attrs.size(),
285278
attrs.creationTime().toMillis(),
286279
attrs.lastModifiedTime().toMillis())
287-
store.save(key, JsonOutput.prettyPrint(JsonOutput.toJson(value)))
288-
workflowResults.outputs.add("${CID_PROT}${key}")
289-
} catch (Throwable e) {
280+
store.save(key, value)
281+
workflowResults.outputs.add("${CID_PROT}${key}".toString())
282+
}
283+
catch (Throwable e) {
290284
log.warn("Exception storing CID output $destination for workflow ${executionHash}.", e)
291285
}
292286
}
@@ -315,16 +309,15 @@ class CidObserver implements TraceObserver {
315309
final rel = getWorkflowRelative(destination)
316310
final key = "$executionHash/${rel}"
317311
final attrs = readAttributes(destination)
318-
final value = new Output(
319-
DataType.WorkflowOutput,
312+
final value = new WorkflowOutput(
320313
destination.toUriString(),
321314
checksum,
322315
"${CID_PROT}${executionHash}".toString(),
323316
attrs.size(),
324317
attrs.creationTime().toMillis(),
325318
attrs.lastModifiedTime().toMillis())
326-
store.save(key, JsonOutput.prettyPrint(JsonOutput.toJson(value)))
327-
workflowResults.outputs.add("${CID_PROT}${key}")
319+
store.save(key, value)
320+
workflowResults.outputs.add("${CID_PROT}${key}" as String)
328321
}catch (Throwable e) {
329322
log.warn("Exception storing CID output $destination for workflow ${executionHash}. ${e.getLocalizedMessage()}")
330323
}

‎modules/nf-cid/src/main/nextflow/data/cid/CidStore.groovy‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,8 @@
1717

1818
package nextflow.data.cid
1919

20-
2120
import groovy.transform.CompileStatic
21+
import nextflow.data.cid.serde.CidSerializable
2222
import nextflow.data.config.DataConfig
2323
/**
2424
* Interface for the CID store
@@ -38,14 +38,14 @@ interface CidStore extends Closeable {
3838
* @param key Entry key.
3939
* @param value Entry object.
4040
*/
41-
void save(String key, Object value)
41+
void save(String key, CidSerializable value)
4242

4343
/**
4444
* Load an entry for a given CID key.
4545
* @param key CID key.
4646
* @return entry value, or null if key does not exists
4747
*/
48-
Object load(String key)
48+
CidSerializable load(String key)
4949

5050
/**
5151
* Get the {@link CidHistoryLog} object associated to the CidStore.

‎modules/nf-cid/src/main/nextflow/data/cid/DefaultCidStore.groovy‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -17,17 +17,17 @@
1717

1818
package nextflow.data.cid
1919

20-
import nextflow.file.FileHelper
21-
import nextflow.util.TestOnly
22-
2320
import java.nio.file.Files
2421
import java.nio.file.Path
2522

2623
import groovy.transform.CompileStatic
2724
import groovy.util.logging.Slf4j
25+
import nextflow.data.cid.serde.CidEncoder
26+
import nextflow.data.cid.serde.CidSerializable
2827
import nextflow.data.config.DataConfig
2928
import nextflow.exception.AbortOperationException
30-
29+
import nextflow.file.FileHelper
30+
import nextflow.util.TestOnly
3131
/**
3232
* Default Implementation for the a CID store.
3333
*
@@ -44,10 +44,13 @@ class DefaultCidStore implements CidStore {
4444
private Path metaLocation
4545
private Path location
4646
private CidHistoryLog historyLog
47+
private CidEncoder encoder
48+
4749

4850
DefaultCidStore open(DataConfig config) {
4951
location = toLocationPath(config.store.location)
5052
metaLocation = location.resolve(METADATA_PATH)
53+
encoder = new CidEncoder()
5154
if( !Files.exists(metaLocation) && !Files.createDirectories(metaLocation) ) {
5255
throw new AbortOperationException("Unable to create CID store directory: $metaLocation")
5356
}
@@ -62,19 +65,19 @@ class DefaultCidStore implements CidStore {
6265
}
6366

6467
@Override
65-
void save(String key, Object value) {
68+
void save(String key, CidSerializable value) {
6669
final path = metaLocation.resolve("$key/$METADATA_FILE")
6770
Files.createDirectories(path.parent)
6871
log.debug "Save CID file path: $path"
69-
path.text = value
72+
path.text = encoder.encode(value)
7073
}
7174

7275
@Override
73-
Object load(String key) {
76+
CidSerializable load(String key) {
7477
final path = metaLocation.resolve("$key/$METADATA_FILE")
7578
log.debug("Loading from path $path")
7679
if (path.exists())
77-
return path.text
80+
return encoder.decode(path.text)
7881
log.debug("File for key $key not found")
7982
return null
8083
}

0 commit comments

Comments
 (0)