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.

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

Add Google Batch NOT_FOUND error management #5690

Merged
merged 27 commits into from
Feb 12, 2025
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
e60d4b9
Add not_found error management
jorgee Jan 21, 2025
959dd2b
Revert to old job status check for standard tasks and use task status…
jorgee Jan 29, 2025
9fdae13
belongsToArray renamed to isChild
jorgee Jan 30, 2025
8a99b03
fix NullPointerExceptions when retry task because of staging error
jorgee Jan 30, 2025
ae292d6
review changes
jorgee Jan 31, 2025
37bfe77
Merge branch 'master' into 5422-not_found-error-on-google-batch
pditommaso Feb 4, 2025
9f811de
Include remove of task in hasmap when completed
jorgee Jan 29, 2025
070932b
fix isChild missing rename
jorgee Jan 30, 2025
1ee152a
fix rebase issue
jorgee Feb 6, 2025
e78bbef
Merge branch 'master' into 5422-not_found-error-on-google-batch
pditommaso Feb 10, 2025
3abaaf9
Update plugins/nf-google/src/main/nextflow/cloud/google/batch/client/…
pditommaso Feb 10, 2025
20b19a7
Update plugins/nf-google/src/main/nextflow/cloud/google/batch/client/…
pditommaso Feb 10, 2025
195ba32
Merge pull request #5723 from nextflow-io/5422-alternavite-task-arrays
pditommaso Feb 10, 2025
15672c7
Fix bugs with workflow outputs (#5502)
bentsherman Feb 10, 2025
7860c91
Fix CI script
pditommaso Feb 10, 2025
3d98495
Bump netty-common:4.1.118.Final
pditommaso Feb 10, 2025
04cd0c2
Improve CI tests collection
pditommaso Feb 10, 2025
846cd42
Fix typo [ci skip]
pditommaso Feb 10, 2025
3e03114
Task array improve
pditommaso Feb 10, 2025
edb4c9f
Update modules/nextflow/src/main/groovy/nextflow/processor/TaskProces…
pditommaso Feb 11, 2025
5cf9c44
Merge pull request #5776 from nextflow-io/5422-not_found-error-on-goo…
pditommaso Feb 11, 2025
f878f9d
Merge branch 'master' into 5422-not_found-error-on-google-batch
pditommaso Feb 11, 2025
40c5f58
Minor changes
pditommaso Feb 11, 2025
ccbdca2
Fix failing test
pditommaso Feb 11, 2025
436ac3f
Comment in ProcessStageException [ci skip]
jorgee Feb 12, 2025
915636b
Amend comment [ci skip]
pditommaso Feb 12, 2025
6eff10a
Fix typos [ci skip]
jorgee Feb 12, 2025
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
Original file line number Diff line number Diff line change
Expand Up @@ -24,5 +24,5 @@ import groovy.transform.InheritConstructors
* @author Paolo Di Tommaso <[email protected]>
*/
@InheritConstructors
class ProcessStageException extends ProcessException implements ShowOnlyExceptionMessage {
class ProcessStageException extends ProcessUnrecoverableException implements ShowOnlyExceptionMessage {
pditommaso marked this conversation as resolved.
Show resolved Hide resolved
}
Original file line number Diff line number Diff line change
Expand Up @@ -331,7 +331,7 @@ class FilePorter {
// remove the target file that could be have partially downloaded
cleanup(stagePath)
// check if a stage/download retry is allowed
if( count++ < maxRetries && e !instanceof NoSuchFileException && e !instanceof InterruptedIOException && !Thread.currentThread().isInterrupted() ) {
if( count++ < maxRetries && e !instanceof NoSuchFileException && (e instanceof SocketTimeoutException || e !instanceof InterruptedIOException) && !Thread.currentThread().isInterrupted() ) {
pditommaso marked this conversation as resolved.
Show resolved Hide resolved
def message = "Unable to stage foreign file: ${filePath.toUriString()} (try ${count} of ${maxRetries}) -- Cause: $e.message"
log.isDebugEnabled() ? log.warn(message, e) : log.warn(message)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,10 @@

package nextflow.cloud.google.batch

import com.google.api.gax.rpc.NotFoundException
import com.google.cloud.batch.v1.JobStatus
import com.google.cloud.batch.v1.Task

import java.nio.file.Path
import java.util.regex.Pattern

Expand Down Expand Up @@ -91,6 +95,11 @@ class GoogleBatchTaskHandler extends TaskHandler implements FusionAwareTask {
*/
private String uid

/**
* Flag to indicate if task belong to an TaskRunArray
*/
private boolean isChild
pditommaso marked this conversation as resolved.
Show resolved Hide resolved

/**
* Task state assigned by Google Batch service
*/
Expand Down Expand Up @@ -180,19 +189,20 @@ class GoogleBatchTaskHandler extends TaskHandler implements FusionAwareTask {
log.debug "[GOOGLE BATCH] Process `${task.lazyName()}` submitted > job=$jobId; uid=$uid; work-dir=${task.getWorkDirStr()}"
}

protected void updateStatus(String jobId, String taskId, String uid) {
protected void updateStatus(String jobId, String taskId, String uid, boolean isChild = false) {
if( task instanceof TaskArrayRun ) {
// update status for children
for( int i=0; i<task.children.size(); i++ ) {
final handler = task.children[i] as GoogleBatchTaskHandler
final arrayTaskId = executor.getArrayTaskId(jobId, i)
handler.updateStatus(jobId, arrayTaskId, uid)
handler.updateStatus(jobId, arrayTaskId, uid, true)
}
}
else {
this.jobId = jobId
this.taskId = taskId
this.uid = uid
this.isChild = isChild
this.status = TaskStatus.SUBMITTED
}
}
Expand Down Expand Up @@ -451,49 +461,91 @@ class GoogleBatchTaskHandler extends TaskHandler implements FusionAwareTask {
* @return Retrieve the submitted task state
*/
protected String getTaskState() {
if (isChild) {
return getStateFromTaskStatus()
} else {
return getStateFromJobStatus()
}
}

protected String getStateFromTaskStatus() {
final tasks = client.listTasks(jobId)
if( !tasks.iterator().hasNext() ) {
// if there are no tasks checks the job status
return checkJobStatus()
return getStateFromJobStatus()
}
final now = System.currentTimeMillis()
final delta = now - timestamp;
if( !taskState || delta >= 1_000) {
final status = client.getTaskStatus(jobId, taskId)
final newState = status?.state as String
if( newState ) {
log.trace "[GOOGLE BATCH] Get job=$jobId task=$taskId state=$newState"
taskState = newState
timestamp = now
try {
final status = client.getTaskStatus(jobId, taskId)
inspectTaskStatus(status)
}catch (NotFoundException e) {
manageNotFound(tasks)
}
if( newState == 'PENDING' ) {
final eventsCount = status.getStatusEventsCount()
final lastEvent = eventsCount > 0 ? status.getStatusEvents(eventsCount - 1) : null
if( lastEvent?.getDescription()?.contains('CODE_GCE_QUOTA_EXCEEDED') )
log.warn1 "Batch job cannot be run: ${lastEvent.getDescription()}"
}
return taskState
}

protected String getStateFromJobStatus() {
final now = System.currentTimeMillis()
final delta = now - timestamp;
if( !taskState || delta >= 1_000) {
final status = client.getJobStatus(jobId)
inspectJobStatus(status)
}
return taskState
}

private void inspectTaskStatus(com.google.cloud.batch.v1.TaskStatus status) {
final newState = status?.state as String
if (newState) {
log.trace "[GOOGLE BATCH] Get job=$jobId task=$taskId state=$newState"
taskState = newState
timestamp = System.currentTimeMillis()
}
if (newState == 'PENDING') {
final eventsCount = status.getStatusEventsCount()
final lastEvent = eventsCount > 0 ? status.getStatusEvents(eventsCount - 1) : null
if (lastEvent?.getDescription()?.contains('CODE_GCE_QUOTA_EXCEEDED'))
log.warn1 "Batch job cannot be run: ${lastEvent.getDescription()}"
}
}

protected String manageNotFound( Iterable<Task> tasks) {
// If task is array, check if the in the task list
for (Task t in tasks) {
if (t.name == client.generateTaskName(jobId, taskId)) {
inspectTaskStatus(t.status)
return taskState
}
}
// if not array or it task is not in the list, check job status.
final status = client.getJobStatus(jobId)
inspectJobStatus(status)
return taskState
}

protected String checkJobStatus() {
final jobStatus = client.getJobStatus(jobId)
final newState = jobStatus?.state as String
protected String inspectJobStatus(JobStatus status) {
final newState = status?.state as String
if (newState) {
log.trace "[GOOGLE BATCH] Get job=$jobId state=$newState"
taskState = newState
timestamp = System.currentTimeMillis()
if (newState == "FAILED") {
noTaskJobfailure = true
}
return taskState
} else {
return "PENDING"
}
if (newState == 'SCHEDULED') {
final eventsCount = status.getStatusEventsCount()
final lastEvent = eventsCount > 0 ? status.getStatusEvents(eventsCount - 1) : null
if (lastEvent?.getDescription()?.contains('CODE_GCE_QUOTA_EXCEEDED'))
log.warn1 "Batch job cannot be run: ${lastEvent.getDescription()}"
}
}

static private final List<String> RUNNING_OR_COMPLETED = ['RUNNING', 'SUCCEEDED', 'FAILED']
static private final List<String> RUNNING_OR_COMPLETED = ['RUNNING', 'SUCCEEDED', 'FAILED', 'DELETION_IN_PROGRESS']

static private final List<String> COMPLETED = ['SUCCEEDED', 'FAILED']
static private final List<String> COMPLETED = ['SUCCEEDED', 'FAILED', 'DELETION_IN_PROGRESS']

@Override
boolean checkIfRunning() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ class BatchClient {
}

Task describeTask(String jobId, String taskId) {
final name = TaskName.of(projectId, location, jobId, 'group0', taskId)
final name = generateTaskName(jobId, taskId)
return apply(()-> batchServiceClient.getTask(name))
}

Expand Down Expand Up @@ -141,6 +141,10 @@ class BatchClient {
return location
}

String generateTaskName(String jobId, String taskId) {
TaskName.of(projectId, location, jobId, 'group0', taskId)
}

/**
* Creates a retry policy using the configuration specified by {@link BatchRetryConfig}
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,11 @@

package nextflow.cloud.google.batch

import com.google.api.gax.grpc.GrpcStatusCode
import com.google.api.gax.rpc.NotFoundException
import com.google.cloud.batch.v1.JobStatus
import com.google.cloud.batch.v1.Task
import io.grpc.Status

import java.nio.file.Path

Expand Down Expand Up @@ -596,7 +599,7 @@ class GoogleBatchTaskHandlerTest extends Specification {
builder.build()
}

def 'should check job status when no tasks in job '() {
def 'should check job status when no tasks in task array '() {

given:
def jobId = 'job-id'
Expand All @@ -605,7 +608,7 @@ class GoogleBatchTaskHandlerTest extends Specification {
def task = Mock(TaskRun) {
lazyName() >> 'foo (1)'
}
def handler = Spy(new GoogleBatchTaskHandler(jobId: jobId, taskId: taskId, client: client, task: task))
def handler = Spy(new GoogleBatchTaskHandler(jobId: jobId, taskId: taskId, client: client, task: task, isChild: true))
final message = 'Job failed when Batch tries to schedule it: Batch Error: code - CODE_MACHINE_TYPE_NOT_FOUND'
when:
client.listTasks(jobId) >>> [new LinkedList<Task>(), new LinkedList<Task>()]
Expand All @@ -615,8 +618,40 @@ class GoogleBatchTaskHandlerTest extends Specification {
makeJobStatus(JobStatus.State.FAILED, message)
]
then:
handler.getTaskState() == "PENDING"
handler.getTaskState() == null
handler.getTaskState() == "FAILED"
handler.getJobError().message == message
}

def 'should manage not found when getting task state in task array'() {
given:
def jobId = '1'
def taskId = '1'
def client = Mock(BatchClient)
def task = Mock(TaskRun) {
lazyName() >> 'foo (1)'
}
def handler = Spy(new GoogleBatchTaskHandler(jobId: jobId, taskId: taskId, client: client, task: task, isChild: true))

when:
client.generateTaskName(jobId, taskId) >> "$jobId/group0/$taskId"
//Force errors
client.getTaskStatus(jobId, taskId) >> { throw new NotFoundException(new Exception("Error"), GrpcStatusCode.of(Status.Code.NOT_FOUND), false) }
client.listTasks(jobId) >> TASK_LIST
client.getJobStatus(jobId) >> makeJobStatus(JOB_STATUS, "")
then:
handler.getTaskState() == EXPECTED

where:
EXPECTED | JOB_STATUS | TASK_LIST
"FAILED" | JobStatus.State.FAILED | {[ makeTask("1/group0/2", TaskStatus.State.PENDING), makeTask("1/group0/3", TaskStatus.State.PENDING) ].iterator() } // Task not in the list, get from job
"SUCCEEDED" | JobStatus.State.FAILED | {[ makeTask("1/group0/1", TaskStatus.State.SUCCEEDED), makeTask("1/group0/2", TaskStatus.State.PENDING)].iterator() } //Task in the list, get from task status
}

def makeTask(String name, TaskStatus.State state){
Task.newBuilder().setName(name)
.setStatus(TaskStatus.newBuilder().setState(state).build())
.build()

}
}