Move MTC validation to GKE orchestrator

Author: Neelabh94

cmd/job/submit.go

@@ -380,6 +380,11 @@ func validateGKENAPFlags() error {
 	if gkeNapProvisioning != "reservation" && gkeNapReservation != "" {
 		return fmt.Errorf("--gke-nap-reservation should only be provided when --gke-nap-provisioning=reservation")
 	}
+
+	if pathways.MTCEnabled {
+		// MTC Addon validation is now handled by the GKE Orchestrator preflight checks.
+	}
+
 	return nil
 }
 

pkg/orchestrator/gke/gke_job_orchestrator.go

@@ -32,6 +32,8 @@ import (
 	"strings"
 	"time"
 
+	container "google.golang.org/api/container/v1"
+
 	"github.com/google/safetext/yamltemplate"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -97,6 +99,12 @@ func (g *GKEOrchestrator) SubmitJob(job orchestrator.JobDefinition) error {
 		return err
 	}
 
+	if job.Pathways.MTCEnabled {
+		if err := g.checkMTCAddonEnabled(job.ProjectID, job.ClusterLocation, job.ClusterName); err != nil {
+			return err
+		}
+	}
+
 	if err := g.fetchClusterState(&job); err != nil {
 		return err
 	}
@@ -133,6 +141,25 @@ func (g *GKEOrchestrator) SubmitJob(job orchestrator.JobDefinition) error {
 	return nil
 }
 
+func (g *GKEOrchestrator) checkMTCAddonEnabled(projectID, location, clusterName string) error {
+	ctx := context.Background()
+	svc, err := container.NewService(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create cluster manager client: %v", err)
+	}
+
+	name := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", projectID, location, clusterName)
+	resp, err := svc.Projects.Locations.Clusters.Get(name).Context(ctx).Do()
+	if err != nil {
+		return fmt.Errorf("failed to get cluster details for MTC validation: %v", err)
+	}
+
+	if resp.AddonsConfig == nil || resp.AddonsConfig.HighScaleCheckpointingConfig == nil || !resp.AddonsConfig.HighScaleCheckpointingConfig.Enabled {
+		return fmt.Errorf("Multi-Tier Checkpointing (MTC) requires the HighScaleCheckpointing addon to be enabled on the target GKE cluster. Please update your cluster blueprint to set 'enable_multi_tier_checkpointing: true' and deploy the cluster before submitting jobs with --pathways-mtc-enabled")
+	}
+	return nil
+}
+
 // ListJobs retrieves a list of jobs in the GKE cluster.
 // It filters jobs based on the provided ListOptions.
 func (g *GKEOrchestrator) ListJobs(opts orchestrator.ListOptions) ([]orchestrator.JobStatus, error) {