Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[GH-75] Too many put xxx into map log #76

Merged
merged 1 commit into from
Aug 20, 2024
Merged

[GH-75] Too many put xxx into map log #76

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions plugins/nf-float/src/main/com/memverge/nextflow/FloatBin.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import java.util.regex.Pattern
@Slf4j
class FloatBin {
private static final binName = 'float'
private static logged = false

/**
* This function checks if the float binary is available in the plugin
Expand Down Expand Up @@ -38,11 +39,11 @@ class FloatBin {
}
ret = targetDir.resolve(binName)
try {
log.info "try downloading $src to $ret"
log.info "[FLOAT] try downloading $src to $ret"
Global.download(src, ret)
ret.setExecutable(true)
} catch (Exception ex) {
log.warn("download ${binName} failed: ${ex.message}")
log.warn("[FLOAT] download ${binName} failed: ${ex.message}")
return Paths.get(binName)
}
}
Expand All @@ -67,11 +68,14 @@ class FloatBin {
for (String path : paths) {
def floatPath = Paths.get(path).resolve(binName)
if (Files.exists(floatPath)) {
log.info "found float binary in $path"
if (!logged) {
log.info "[FLOAT] found float binary in $path"
logged = true
}
return floatPath
}
}
log.info "${binName} binary not found"
log.warn "[FLOAT] ${binName} binary not found"
return null
}
}
18 changes: 10 additions & 8 deletions plugins/nf-float/src/main/com/memverge/nextflow/FloatConf.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import groovy.transform.CompileStatic
import groovy.util.logging.Slf4j
import nextflow.exception.AbortOperationException
import nextflow.io.BucketParser
import org.apache.commons.lang.StringUtils
import nextflow.processor.TaskId

/**
* @author Cedric Zhuang <[email protected]>
Expand Down Expand Up @@ -194,7 +194,7 @@ class FloatConf {
}

private static void warnDeprecated(String deprecated, String replacement) {
log.warn "[float] config option `$deprecated` " +
log.warn "[FLOAT] config option `$deprecated` " +
"is no longer supported, " +
"use `$replacement` instead"
}
Expand All @@ -213,11 +213,13 @@ class FloatConf {
}
}

List<String> getCliPrefix(String address = "") {
validate()
if (StringUtils.length(address) == 0) {
address = addresses[0]
List<String> getCliPrefix(TaskId id) {
if (id == null) {
id = new TaskId(0)
}
final address = addresses[id.intValue() % (addresses.size())]
validate()

def bin = FloatBin.get(address)
List<String> ret = [
bin.toString(),
Expand All @@ -232,8 +234,8 @@ class FloatConf {
return ret
}

String getCli(String address = "") {
return getCliPrefix(address).join(" ")
String getCli(TaskId id) {
return getCliPrefix(id).join(" ")
}
}

Expand Down
132 changes: 47 additions & 85 deletions plugins/nf-float/src/main/com/memverge/nextflow/FloatGridExecutor.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ import nextflow.util.ServiceName

import java.nio.file.Path
import java.nio.file.StandardCopyOption
import java.util.stream.Collectors

/**
* Float Executor with a shared file system
Expand All @@ -51,7 +50,7 @@ class FloatGridExecutor extends AbstractGridExecutor {

synchronized FloatJobs getFloatJobs() {
if (_floatJobs == null) {
_floatJobs = new FloatJobs(floatConf.addresses)
_floatJobs = new FloatJobs(getRunName())
}
return _floatJobs
}
Expand All @@ -72,7 +71,7 @@ class FloatGridExecutor extends AbstractGridExecutor {
}

protected String getHeaderScript(TaskRun task) {
log.info "[float] switch task ${task.id} to ${task.workDirStr}"
log.info "[FLOAT] switch task ${task.id} to ${task.workDirStr}"
floatJobs.setWorkDir(task.id, task.workDir)

final path = Escape.path(task.workDir)
Expand Down Expand Up @@ -115,11 +114,10 @@ class FloatGridExecutor extends AbstractGridExecutor {
}

private void syncFloatBin() {
final oc = floatJobs.getOc("")
def cmd = floatConf.getCliPrefix(oc)
def cmd = floatConf.getCliPrefix(null)
cmd << "release" << "sync"
def res = Global.execute(cmd)
log.info "[float] sync the float binary, $res"
log.info "[FLOAT] sync the float binary, $res"
}

private int getMemory(TaskRun task) {
Expand Down Expand Up @@ -169,19 +167,14 @@ class FloatGridExecutor extends AbstractGridExecutor {
return ret
}

private List<String> getCmdPrefixForJob(String floatJobID) {
final oc = floatJobs.getOc(floatJobID)
return floatConf.getCliPrefix(oc)
}

FloatJob getJob(TaskId taskId) {
def nfJobID = floatJobs.getNfJobID(taskId)
def job = floatJobs.nfJobID2job.get(nfJobID)
def job = floatJobs.getJob(taskId)
if (job == null) {
log.warn "[float] job not found for nextflow task $taskId"
log.warn "[FLOAT] job not found for nextflow task $taskId: $nfJobID"
return null
}
def cmd = getCmdPrefixForJob(job.floatJobID)
def cmd = floatConf.getCliPrefix(taskId)
cmd << 'show'
cmd << '-j'
cmd << job.floatJobID
Expand All @@ -193,28 +186,16 @@ class FloatGridExecutor extends AbstractGridExecutor {
job = FloatJob.parse(res.out)
job = floatJobs.updateJob(job)
} else {
log.warn "[float] failed to retrieve job status $nfJobID, " +
log.warn "[FLOAT] failed to retrieve job status $nfJobID, " +
"float: ${job.floatJobID}, details: ${res.out}"
}
} catch (Exception e) {
log.warn "[float] failed to retrieve job status $nfJobID, " +
log.warn "[FLOAT] failed to retrieve job status $nfJobID, " +
"float: ${job.floatJobID}", e
}
return job
}

/**
* Retrieve the prefix of the submit command.
* use round robin for multiple op-center
*
* @return
*/
private List<String> getSubmitCmdPrefix(Integer index) {
final addresses = floatConf.addresses
final address = addresses[index % (addresses.size())]
return floatConf.getCliPrefix(address)
}

String toLogStr(List<String> floatCmd) {
def ret = floatCmd.join(" ")
final toReplace = [
Expand All @@ -232,7 +213,7 @@ class FloatGridExecutor extends AbstractGridExecutor {
}

private static def warnDeprecated(String deprecated, String replacement) {
log.warn1 "[float] process `$deprecated` " +
log.warn1 "[FLOAT] process `$deprecated` " +
"is no longer supported, " +
"use $replacement instead"
}
Expand Down Expand Up @@ -264,7 +245,7 @@ class FloatGridExecutor extends AbstractGridExecutor {
volumes << floatConf.getInputVolume(src.uri)
}
def ret = volumes.unique() - ""
log.info "[float] volumes to mount for ${task.id}: ${toLogStr(ret)}"
log.info "[FLOAT] volumes to mount for ${task.id}: ${toLogStr(ret)}"
return ret
}

Expand All @@ -282,9 +263,17 @@ class FloatGridExecutor extends AbstractGridExecutor {
: [:]
}

String getRunName() {
// replace _ with - to avoid float cli error
return session.runName
.replaceAll("_", "-")
.toLowerCase()
}

private Map<String, String> getCustomTags(TaskRun task) {
final result = new LinkedHashMap<String, String>(10)
result[FloatConf.NF_JOB_ID] = floatJobs.getNfJobID(task.id)
def nfJobID = floatJobs.getNfJobID(task.id)
result[FloatConf.NF_JOB_ID] = nfJobID
result[FloatConf.NF_SESSION_ID] = "uuid-${session.uniqueId}".toString()
result[FloatConf.NF_TASK_NAME] = task.name
result[FloatConf.FLOAT_INPUT_SIZE] = getInputFileSize(task).toString()
Expand All @@ -300,8 +289,9 @@ class FloatGridExecutor extends AbstractGridExecutor {
result[FloatConf.NF_PROJECT_NAME] = projName
}
}
if (session.runName) {
result[FloatConf.NF_RUN_NAME] = session.runName
def runName = getRunName()
if (runName) {
result[FloatConf.NF_RUN_NAME] = runName
}
final resourceLabels = task.config.getResourceLabels()
if (resourceLabels)
Expand Down Expand Up @@ -351,7 +341,7 @@ class FloatGridExecutor extends AbstractGridExecutor {
} else {
log.info("got container image of the task ${container}")
}
def cmd = getSubmitCmdPrefix(task.index)
def cmd = floatConf.getCliPrefix(task.id)
cmd << 'submit'
getMountVols(task).forEach { cmd << '--dataVolume' << it }
cmd << '--image' << container
Expand All @@ -372,7 +362,8 @@ class FloatGridExecutor extends AbstractGridExecutor {
cmd << '--extraContainerOpts'
cmd << '--privileged'
}
getCustomTags(task).each { key, val ->
def tags = getCustomTags(task)
tags.each { key, val ->
cmd << '--customTag' << "${toTag(key)}:${toTag(val)}".toString()
}
if (task.config.getMachineType()) {
Expand All @@ -397,7 +388,7 @@ class FloatGridExecutor extends AbstractGridExecutor {
cmd << '--vmPolicy' << '[onDemand=true]'
}
cmd.addAll(getExtra(task))
log.info "[float] submit job: ${toLogStr(cmd)}"
log.info "[FLOAT] submit job: ${toLogStr(cmd)}"
return cmd
}

Expand Down Expand Up @@ -491,86 +482,57 @@ class FloatGridExecutor extends AbstractGridExecutor {
* @return The command line to be used to kill the specified job
*/
List<List<String>> killTaskCommands(def jobId) {
def jobIds
def floatJobIDs
if (jobId instanceof Collection) {
jobIds = jobId
floatJobIDs = jobId
} else {
jobIds = [jobId]
floatJobIDs = [jobId]
}
List<List<String>> ret = []
jobIds.forEach {
for (def it : floatJobIDs) {
def floatJobID = it.toString()
def cmd = getCmdPrefixForJob(floatJobID)
def taskID = floatJobs.getTaskID(floatJobID)
if (taskID == null) {
log.warn "[FLOAT] task id not found for float job id: $floatJobID"
return
}
def cmd = floatConf.getCliPrefix(taskID)
cmd << 'cancel'
cmd << '-j'
cmd << floatJobID
cmd << '-f'
log.info "[float] cancel job: ${toLogStr(cmd)}"
log.info "[FLOAT] cancel job: ${toLogStr(cmd)}"
ret.add(cmd)
}
return ret
}

private List<String> getCmdPrefix0() {
final addresses = floatConf.addresses
final address = addresses.first()
return floatConf.getCliPrefix(address)
}

@Override
protected List<String> getKillCommand() {
def cmd = getCmdPrefix0()
def cmd = floatConf.getCliPrefix(null)
cmd << 'cancel'
cmd << '-j'
log.info "[float] cancel job: ${toLogStr(cmd)}"
log.info "[FLOAT] cancel job: ${toLogStr(cmd)}"
return cmd
}

/**
* @return The status for all the scheduled and running jobs
*/
@Override
protected Map<String, QueueStatus> getQueueStatus0(queue) {
return queueStatus
}

Map<String, QueueStatus> getQueueStatus() {
final cmdMap = queueStatusCommands()
floatJobs.updateStatus(cmdMap)
log.debug "[float] collecting job status completes."
return nfJobID2Status
}

protected Map<String, List<String>> queueStatusCommands() {
return floatConf.addresses.stream().collect(
Collectors.toMap(
oc -> oc.toString(),
oc -> getQueueCmdOfOC(oc.toString())))
protected Map<String, QueueStatus> getQueueStatus0(_) {
log.debug "[FLOAT] collecting job status completes."
return toStatusMap(floatJobs.getNfJobID2Job())
}

@Override
protected List<String> queueStatusCommand(Object queue) {
return getQueueCmdOfOC()
return []
}

@Override
protected Map<String, QueueStatus> parseQueueStatus(String s) {
def stMap = floatJobs.parseQStatus(s)
return toStatusMap(stMap)
}

private List<String> getQueueCmdOfOC(String oc = "") {
def cmd = floatConf.getCliPrefix(oc)
cmd << 'list'
cmd << '--format'
cmd << 'json'
log.info "[float] query job status: ${toLogStr(cmd)}"
return cmd
}

private Map<String, QueueStatus> getNfJobID2Status() {
Map<String, FloatJob> stMap = floatJobs.getNfJobID2job()
return toStatusMap(stMap)
return getQueueStatus0(null)
}

private static Map<String, QueueStatus> toStatusMap(Map<String, FloatJob> stMap) {
Expand All @@ -585,10 +547,10 @@ class FloatGridExecutor extends AbstractGridExecutor {
FloatStatus getJobStatus(TaskRun task) {
def job = getJob(task.id)
if (!job) {
log.info "[float] task status unknown, job not found for ${task.id}"
log.info "[FLOAT] task status unknown, job not found for ${task.id}"
return FloatStatus.UNKNOWN
}
log.info "[float] task id: ${task.id}, nf-job-id: $job.nfJobID, " +
log.info "[FLOAT] task id: ${task.id}, nf-job-id: $job.nfJobID, " +
"float-job-id: $job.floatJobID, float status: $job.status"
return job.status
}
Expand Down
Loading
Loading