diff --git a/compute/kubernetes/backend.go b/compute/kubernetes/backend.go index c0dfc241f..edad383e5 100644 --- a/compute/kubernetes/backend.go +++ b/compute/kubernetes/backend.go @@ -125,7 +125,6 @@ func (b *Backend) Close() { // Submit creates both the PVC and the worker job with better error handling func (b *Backend) Submit(ctx context.Context, task *tes.Task, config *config.Config) error { err := b.createResources(task, config) - b.log.Debug("Error creating resources", "error", err, "task ID", task.Id) if err != nil { b.log.Error("Error creating resources, writing SystemError event", "error", err, "task ID", task.Id) @@ -331,14 +330,14 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea if !disableCleanup { jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{}) if err != nil { - b.log.Error("backlog cleanup: listing jobs", err) + b.log.Error("Reconciler", "backlog cleanup listing jobs", err) } else { for _, j := range jobs.Items { s := j.Status if s.Succeeded > 0 || s.Failed > 0 { - b.log.Debug("backlog cleanup: deleting job", "taskID", j.Name) + b.log.Debug("Reconciler", "cleaning resources from completed task", "taskID", j.Name) if err := b.cleanResources(ctx, j.Name); err != nil { - b.log.Error("backlog cleanup: failed to clean resources", "taskID", j.Name, "error", err) + b.log.Error("Reconciler", "failed to clean resources", "taskID", j.Name, "error", err) } } } @@ -354,15 +353,15 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea case <-ctx.Done(): return case <-ticker.C: - // List ALL current Kubernetes Jobs // Bug: If K8s Job is not created by the time reconciler runs, then the TES Task itself will be prematurely marked as SYSTEM_ERROR jobs, err := b.client.BatchV1().Jobs(b.conf.Kubernetes.JobsNamespace).List(ctx, metav1.ListOptions{}) if err != nil { - b.log.Error("reconcile: listing jobs", err) + b.log.Error("listing jobs error", "error", err) continue } + b.log.Debug("Number of K8s Jobs", len(jobs.Items)) k8sJobs := make(map[string]*v1.Job) for i := range jobs.Items { k8sJobs[jobs.Items[i].Name] = &jobs.Items[i] @@ -371,6 +370,7 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea // List non-terminal tasks from Funnel's database states := []tes.State{tes.State_QUEUED, tes.State_INITIALIZING, tes.State_RUNNING} for _, s := range states { + b.log.Debug("State", s.String()) pageToken := "" for { lresp, err := b.database.ListTasks(ctx, &tes.ListTasksRequest{ @@ -379,7 +379,7 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea PageToken: pageToken, }) if err != nil { - b.log.Error("reconcile: listing non-terminal tasks from Funnel DB", err) + b.log.Error("listing non-terminal tasks from Funnel DB", "error", err) break } pageToken = lresp.NextPageToken @@ -389,28 +389,28 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea taskID := task.Id // Check for Orphaned Task (Job Missing) - // if _, exists := k8sJobs[taskID]; !exists { + if _, exists := k8sJobs[taskID]; !exists { - // b.log.Debug("reconcile: orphaned task found, marking as SYSTEM_ERROR", "taskID", taskID) + b.log.Debug("TES Task not found in K8s Jobs, updating to SYSTEM_ERROR", "taskID", taskID, "Current state", task.State.String()) - // b.event.WriteEvent(ctx, events.NewState(taskID, tes.SystemError)) + b.event.WriteEvent(ctx, events.NewState(taskID, tes.SystemError)) - // b.event.WriteEvent( - // ctx, - // events.NewSystemLog( - // taskID, 0, 0, "error", - // "DEBUG: Kubernetes Worker Job not found! Submission failed or external deletion.", - // nil, - // ), - // ) - // continue - // } + b.event.WriteEvent( + ctx, + events.NewSystemLog( + taskID, 0, 0, "error", + "DEBUG: Kubernetes Worker Job not found! Submission failed or external deletion.", + nil, + ), + ) + continue + } // If the job exists, check its current status (Active, Succeeded, Failed) j := k8sJobs[taskID] // Remove from map to ensure only orphaned checks are done above - // delete(k8sJobs, taskID) + delete(k8sJobs, taskID) if j == nil { continue @@ -426,7 +426,7 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea if disableCleanup { continue } - b.log.Debug("reconcile: cleaning up successful job", "taskID", jobName) + b.log.Debug("Reconciler", "Cleaning up successful job", "taskID", jobName) // Delete resources if err := b.cleanResources(ctx, jobName); err != nil { @@ -440,10 +440,10 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea continue } - b.log.Debug("reconcile: writing system error event for failed job", "taskID", jobName) + b.log.Debug("Reconciler", "Writing system error event for failed job", "taskID", jobName) conds, err := json.Marshal(status.Conditions) if err != nil { - b.log.Error("reconcile: marshal failed job conditions", "taskID", jobName, "error", err) + b.log.Error("Reconciler", "marshal failed job conditions", "taskID", jobName, "error", err) } b.event.WriteEvent(ctx, events.NewState(jobName, tes.SystemError)) @@ -461,9 +461,9 @@ func (b *Backend) reconcile(ctx context.Context, rate time.Duration, disableClea continue } - b.log.Debug("reconcile: cleaning up failed job", "taskID", jobName) + b.log.Debug("Reconciler", "Cleaning up failed job", "taskID", jobName) if err := b.cleanResources(ctx, jobName); err != nil { - b.log.Error("failed to clean resources", "taskID", jobName, "error", err) + b.log.Error("Reconciler", "failed to clean resources", "taskID", jobName, "error", err) continue } delete(failedJobEvents, jobName) diff --git a/database/postgres/tes.go b/database/postgres/tes.go index c3fd34f67..b9f2aa623 100644 --- a/database/postgres/tes.go +++ b/database/postgres/tes.go @@ -86,6 +86,13 @@ func (db *Postgres) ListTasks(ctx context.Context, req *tes.ListTasksRequest) (* var whereClauses []string paramCount := 1 + // State filter + if req.State != tes.State_UNKNOWN { + whereClauses = append(whereClauses, fmt.Sprintf("state = $%d", paramCount)) + args = append(args, req.State.String()) + paramCount++ + } + // Name prefix filter if req.NamePrefix != "" { whereClauses = append(whereClauses, fmt.Sprintf("data ->> 'name' LIKE $%d", paramCount)) @@ -107,7 +114,7 @@ func (db *Postgres) ListTasks(ctx context.Context, req *tes.ListTasksRequest) (* whereClauses = append(whereClauses, fmt.Sprintf("data -> 'tags' ? $%d", paramCount)) args = append(args, k) } else { - // Check if tag value equals: `data -> 'tags' ->> 'key' = 'value'` + // Check if tag key equals value: `data -> 'tags' ->> 'key' = 'value'` whereClauses = append(whereClauses, fmt.Sprintf("data -> 'tags' ->> $%d = $%d", paramCount, paramCount+1)) args = append(args, k, v) paramCount++ @@ -122,16 +129,16 @@ func (db *Postgres) ListTasks(ctx context.Context, req *tes.ListTasksRequest) (* paramCount++ } - whereClause := "" + where := "" if len(whereClauses) > 0 { - whereClause = "WHERE " + strings.Join(whereClauses, " AND ") + where = "WHERE " + strings.Join(whereClauses, " AND ") } - orderByClause := "ORDER BY creation_time DESC, id DESC" - limitClause := fmt.Sprintf("LIMIT $%d", paramCount) + order := "ORDER BY creation_time DESC, id DESC" + limit := fmt.Sprintf("LIMIT $%d", paramCount) args = append(args, pageSize) - selectSQL := fmt.Sprintf("SELECT data FROM tasks %s %s %s", whereClause, orderByClause, limitClause) + selectSQL := fmt.Sprintf("SELECT data FROM tasks %s %s %s", where, order, limit) rows, err := db.client.Query(ctx, selectSQL, args...) if err != nil { diff --git a/storage/generic_s3.go b/storage/generic_s3.go index 412727255..4665cf998 100644 --- a/storage/generic_s3.go +++ b/storage/generic_s3.go @@ -216,26 +216,6 @@ func download(ctx context.Context, client *minio.Client, bucket, objectPath, fil } defer outFile.Close() - // Output the downloaded file contents for debugging - content, err := os.ReadFile(outFile.Name()) - if err != nil { - logger.Debug("Error reading file", "filePath", outFile.Name(), "error", err) - } - logger.Debug("genericS3: file contents", "filePath", outFile.Name(), "content", string(content)) - - // Write the content to the file - err = os.WriteFile(outFile.Name(), []byte("Hello, Go file writing!"), 0644) - if err != nil { - logger.Debug("Error writing to file", "filePath", outFile.Name(), "error", err) - } - - // Output the downloaded file contents for debugging - content, err = os.ReadFile(outFile.Name()) - if err != nil { - logger.Debug("Error reading file", "filePath", outFile.Name(), "error", err) - } - logger.Debug("genericS3: file contents", "filePath", outFile.Name(), "content", string(content)) - // Step 3: Copy the contents // TODO: Add stack trace (or simply more verbose logging) here... // Can we add stack traces for all errors in Funnel? diff --git a/worker/kubernetes.go b/worker/kubernetes.go index 692072806..6f6c26f95 100644 --- a/worker/kubernetes.go +++ b/worker/kubernetes.go @@ -197,11 +197,18 @@ func waitForPodRunning(ctx context.Context, namespace string, podName string, ti return nil, fmt.Errorf("timed out waiting for pod %s to be in running state", podName) case <-ticker.C: pod, err := clientset.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) + if err != nil { - return nil, fmt.Errorf("getting pod %s: %v", podName, err) + log.Printf("Still waiting for pod %s to be created... (error: %v)", podName, err) + continue } - return pod, nil + if pod.Status.Phase == corev1.PodRunning || + pod.Status.Phase == corev1.PodSucceeded || + pod.Status.Phase == corev1.PodFailed { + return pod, nil + } + log.Printf("Pod %s exists but is in phase %s, waiting...", podName, pod.Status.Phase) } } }