Merge branch 'main' of https://github.com/nirupama-dev/scheduler-jupyter-plugin into bug-vertex-listing-pagination

Author: nirupama-dev

scheduler_jupyter_plugin/commons/constants.py

@@ -32,6 +32,7 @@
 TAGS = "scheduler_jupyter_plugin"
 VERTEX_STORAGE_BUCKET = "vertex-schedules"
 UTF8 = "utf-8"
+PAYLOAD_JSON_FILE_PATH = "payload.json"
 
 # Composer environment name restrictions are documented here:
 #  https://cloud.google.com/composer/docs/reference/rest/v1/projects.locations.environments#resource:-environment

scheduler_jupyter_plugin/services/executor.py

@@ -21,6 +21,7 @@
 from google.api_core.exceptions import NotFound
 import google.oauth2.credentials as oauth2
 import aiofiles
+import json
 
 import aiohttp
 import pendulum
@@ -35,6 +36,7 @@
     PACKAGE_NAME,
     WRAPPER_PAPPERMILL_FILE,
     UTF8,
+    PAYLOAD_JSON_FILE_PATH,
 )
 from scheduler_jupyter_plugin.models.models import DescribeJob
 from scheduler_jupyter_plugin.services import airflow
@@ -290,8 +292,7 @@ async def check_package_in_env(self, composer_environment_name):
             else:
                 decoded_output = stdout.decode(UTF8)
                 installed_packages = set(
-                    line.split()[0].lower()
-                    for line in decoded_output.splitlines()[2:]
+                    line.split()[0].lower() for line in decoded_output.splitlines()[2:]
                 )
                 for package in packages:
                     if package.lower() not in installed_packages:
@@ -303,7 +304,6 @@ async def check_package_in_env(self, composer_environment_name):
             self.log.exception(f"Error checking packages: {error}")
             raise IOError(f"Error checking packages: {error}")
 
-
     async def install_to_composer_environment(
         self, local_kernel, composer_environment_name, packages_to_install
     ):
@@ -320,9 +320,7 @@ async def install_to_composer_environment(
                         stderr=subprocess.PIPE,
                         shell=True,
                     )
-                    install_stdout, install_stderr = (
-                        install_process.communicate()
-                    )
+                    install_stdout, install_stderr = install_process.communicate()
                     if install_process.returncode == 0:
                         self.log.info(f"{package} installed successfully.")
                     else:
@@ -338,6 +336,12 @@ async def install_to_composer_environment(
             self.log.exception(f"error installing {package}: {install_stderr}")
             return {"error": str(e)}
 
+    def create_payload(self, file_path, project_id, region, input_data):
+        payload = {"projectId": project_id, "region": region, "job": input_data}
+
+        with open(file_path, "w") as f:
+            json.dump(payload, f, indent=4)
+
     async def execute(self, input_data, project_id=None, region_id=None):
         try:
             job = DescribeJob(**input_data)
@@ -354,7 +358,9 @@ async def execute(self, input_data, project_id=None, region_id=None):
 
             if job.packages_to_install != None:
                 install_packages = await self.install_to_composer_environment(
-                    job.local_kernel, job.composer_environment_name, job.packages_to_install
+                    job.local_kernel,
+                    job.composer_environment_name,
+                    job.packages_to_install,
                 )
             if install_packages and install_packages.get("error"):
                 raise Exception(install_packages)
@@ -374,12 +380,25 @@ async def execute(self, input_data, project_id=None, region_id=None):
                 print(
                     f"The file gs://{gcs_dag_bucket}/{wrapper_pappermill_file_path} does not exist."
                 )
+            # uploading input file while creating the job
             if not job.input_filename.startswith(GCS):
                 await self.upload_to_gcs(
                     gcs_dag_bucket,
                     file_path=f"./{job.input_filename}",
                     destination_dir=f"dataproc-notebooks/{job_name}/input_notebooks",
                 )
+            # creating a json file for payload
+            self.create_payload(
+                PAYLOAD_JSON_FILE_PATH, project_id, region_id, input_data
+            )
+
+            # uploading payload JSON file to GCS
+            await self.upload_to_gcs(
+                gcs_dag_bucket,
+                file_path=PAYLOAD_JSON_FILE_PATH,
+                destination_dir=f"dataproc-notebooks/{job_name}/dag_details",
+            )
+
             file_path = self.prepare_dag(job, gcs_dag_bucket, dag_file)
             await self.upload_to_gcs(
                 gcs_dag_bucket, file_path=file_path, destination_dir="dags"

scheduler_jupyter_plugin/services/vertex.py

@@ -14,6 +14,7 @@
 
 
 import aiohttp
+import json
 from cron_descriptor import get_description
 
 import google.oauth2.credentials as oauth2
@@ -88,10 +89,17 @@ async def upload_to_gcs(self, bucket_name, file_path, job_name):
         blob = bucket.blob(blob_name)
         blob.upload_from_filename(file_path)
 
+        # creating json file containing the input file path
+        metadata = {"inputFilePath": f"gs://{bucket_name}/{blob_name}"}
+        json_file_name = f"{job_name}.json"
+
+        with open(json_file_name, "w") as f:
+            json.dump(metadata, f, indent=4)
+
         # uploading json file containing the input file path
-        json_blob_name = f"{job_name}/{job_name}.json"
+        json_blob_name = f"{job_name}/{json_file_name}"
         json_blob = bucket.blob(json_blob_name)
-        json_blob.upload_from_string(f"gs://{bucket_name}/{blob_name}")
+        json_blob.upload_from_filename(json_file_name)
 
         self.log.info(f"File {input_notebook} uploaded to gcs successfully")
         return blob_name
@@ -248,10 +256,10 @@ async def list_uiconfig(self, region_id):
     async def list_schedules(self, region_id, page_size=100, next_page_token=None):
         try:
             result = {}
-            
+
             if next_page_token:
                 api_endpoint = f"https://{region_id}-aiplatform.googleapis.com/v1/projects/{self.project_id}/locations/{region_id}/schedules?orderBy=createTime desc&pageToken={next_page_token}&pageSize={page_size}"
-                
+
             else:
                 api_endpoint = f"https://{region_id}-aiplatform.googleapis.com/v1/projects/{self.project_id}/locations/{region_id}/schedules?orderBy=createTime desc&pageSize={page_size}"
 

src/controls/RegionDropdown.tsx

@@ -48,16 +48,22 @@ export function RegionDropdown(props: Props) {
     loaderRegion,
     regionsList
   } = props;
-  const regions = useRegion(projectId);
+  let regionStrList: string[] = [];
 
-  const regionStrList = useMemo(
-    () => regions.map(region => region.name),
-    [regions]
-  );
+  if (!regionsList) {
+    const regions = useRegion(projectId);
+
+    regionStrList = useMemo(
+      () => regions.map(region => region.name),
+      [regions]
+    );
+  }
 
   return (
     <Autocomplete
-      value={region}
+      value={
+        regionsList ? (regionsList?.includes(region) ? region : '') : region
+      }
       options={regionsList ? regionsList : regionStrList}
       onChange={(_, value) => onRegionChange(value ?? '')}
       PaperComponent={(props: PaperProps) => <Paper elevation={8} {...props} />}

src/jobs/LabelProperties.tsx

@@ -162,7 +162,9 @@ function LabelProperties({
     } else if (
       buttonText !== 'ADD LABEL' &&
       (labelDetail.length === 0 ||
-        labelDetail[labelDetail.length - 1].split(':')[0].length > 0)
+        (labelDetail[labelDetail.length - 1].split(':')[0].length > 0 &&
+          labelDetail[labelDetail.length - 1].split(':')[1].length > 0)) &&
+      duplicateKeyError === -1
     ) {
       return 'job-add-property-button';
     } else {
@@ -334,7 +336,8 @@ function LabelProperties({
             if (!isDisabled) {
               if (
                 labelDetail.length === 0 ||
-                labelDetail[labelDetail.length - 1].split(':')[0].length > 0
+                (labelDetail[labelDetail.length - 1].split(':')[0].length > 0 &&
+                  labelDetail[labelDetail.length - 1].split(':')[1].length > 0)
               ) {
                 handleAddLabel(e);
               }
@@ -344,7 +347,8 @@ function LabelProperties({
           }}
         >
           {labelDetail.length === 0 ||
-          labelDetail[labelDetail.length - 1].split(':')[0].length > 0 ? (
+          (labelDetail[labelDetail.length - 1].split(':')[0].length > 0 &&
+            labelDetail[labelDetail.length - 1].split(':')[1].length > 0) ? (
             <iconPlus.react
               tag="div"
               className="icon-black logo-alignment-style"
@@ -358,7 +362,8 @@ function LabelProperties({
           <span
             className={
               labelDetail.length === 0 ||
-              labelDetail[labelDetail.length - 1].split(':')[0].length > 0
+              (labelDetail[labelDetail.length - 1].split(':')[0].length > 0 &&
+                labelDetail[labelDetail.length - 1].split(':')[1].length > 0)
                 ? 'job-edit-text'
                 : 'job-edit-text-disabled'
             }

src/scheduler/NotebookScheduler.tsx

@@ -182,13 +182,22 @@ const NotebookSchedulerComponent = ({
               />
             </div>
             {jobNameSelected === '' && !editMode && (
-              <ErrorMessage message="Job name is required" />
+              <ErrorMessage
+                message="Job name is required"
+                showIcon={notebookSelector === 'composer' ? true : false}
+              />
             )}
             {jobNameSpecialValidation && jobNameValidation && !editMode && (
-              <ErrorMessage message="Name must contain only letters, numbers, hyphens, and underscores" />
+              <ErrorMessage
+                message="Name must contain only letters, numbers, hyphens, and underscores"
+                showIcon={notebookSelector === 'composer' ? true : false}
+              />
             )}
             {!jobNameUniqueValidation && !editMode && (
-              <ErrorMessage message="Job name must be unique for the selected environment" />
+              <ErrorMessage
+                message="Job name must be unique for the selected environment"
+                showIcon={notebookSelector === 'composer' ? true : false}
+              />
             )}
 
             <div className="create-scheduler-form-element-input-file">

src/scheduler/composer/ListNotebookScheduler.tsx

@@ -487,12 +487,14 @@ function listNotebookScheduler({
       );
     } else if (cell.column.Header === 'Job Name') {
       return (
-        <td
-          {...cell.getCellProps()}
-          className="clusters-table-data"
-          onClick={() => handleDagIdSelection(composerSelectedList, cell.value)}
-        >
-          {cell.value}
+        <td {...cell.getCellProps()} className="clusters-table-data">
+          <span
+            onClick={() =>
+              handleDagIdSelection(composerSelectedList, cell.value)
+            }
+          >
+            {cell.value}
+          </span>
         </td>
       );
     } else {
@@ -551,6 +553,10 @@ function listNotebookScheduler({
   }, []);
 
   useEffect(() => {
+    if (composerList.length === 0) {
+      setComposerSelectedList('');
+      setDagList([]);
+    }
     if (
       composerList.length > 0 &&
       backselectedEnvironment === '' &&