fix/reconciler → develop: Fix reconciler issues

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)

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 {

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?

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)
 		}
 	}
 }