Changeset 878

Timestamp:

Aug 22, 2018, 3:24:22 PM (6 years ago)

Author:

djay

Message:

Add FinalizeHPC1 service to use sacct command rather than scontrol.

Location:

branches/prototype-v0/zoo-project/zoo-services/utils/hpc

Files:

• cgi-env/FinalizeHPC1.zcfg (added)
• service.c (modified) (2 diffs)

branches/prototype-v0/zoo-project/zoo-services/utils/hpc/service.c

@@ -82 +82 @@
       return SERVICE_FAILED;
     }
-    unlink(flenv);
-    free(flenv);
-
+    
     SSHCON *test=ssh_connect(conf);
-    if(test==NULL){
-      setMapInMaps(conf,"lenv","message",_("Unable to connect using through ssh."));
-      return SERVICE_FAILED;
-    }
+    /*if(test==NULL){
+      setMapInMaps(conf,"lenv","message",_("Unable to connect using ssh."));
+      return SERVICE_FAILED;
+      }*/
 
     char *logPath=(char*)malloc((strlen(tmpPath->value)+strlen(jobid->value)+12)*sizeof(char));
@@ -201 +199 @@
     close(fd);
     setOutputValue(outputs,"Result",(char*)"\"FinalizeHPC run successfully\"",32);
+    unlink(flenv);
+    free(flenv);
 
     return SERVICE_SUCCEEDED;
   }
 
+
+  /**
+   * FinalizeHPC1 ZOO Service :
+   * This service is used to inform a ZOO-Kernel waiting for the end of the
+   * execution of a HPC service
+   *
+   * format="AllocCPUS"; for i in $(sacct -e) ; do format="$format,$i"; done; format="$(echo $format | sed "s:AllocCPUS,::")" ; echo $format; sacct --format=$format -p | grep "997f-11e8-9f78-0050569320d2" 
+   *
+   * AllocCPUS,AllocGRES,AllocNodes,AllocTRES,Account,AssocID,AveCPU,AveCPUFreq,AveDiskRead,AveDiskWrite,AvePages,AveRSS,AveVMSize,BlockID,Cluster,Comment,ConsumedEnergy,ConsumedEnergyRaw,CPUTime,CPUTimeRAW,DerivedExitCode,Elapsed,Eligible,End,ExitCode,GID,Group,JobID,JobIDRaw,JobName,Layout,MaxDiskRead,MaxDiskReadNode,MaxDiskReadTask,MaxDiskWrite,MaxDiskWriteNode,MaxDiskWriteTask,MaxPages,MaxPagesNode,MaxPagesTask,MaxRSS,MaxRSSNode,MaxRSSTask,MaxVMSize,MaxVMSizeNode,MaxVMSizeTask,MinCPU,MinCPUNode,MinCPUTask,NCPUS,NNodes,NodeList,NTasks,Priority,Partition,QOS,QOSRAW,ReqCPUFreq,ReqCPUFreqMin,ReqCPUFreqMax,ReqCPUFreqGov,ReqCPUS,ReqGRES,ReqMem,ReqNodes,ReqTRES,Reservation,ReservationId,Reserved,ResvCPU,ResvCPURAW,Start,State,Submit,Suspended,SystemCPU,Timelimit,TotalCPU,UID,User,UserCPU,WCKey,WCKeyID
+   * 28||1|cpu=28,node=1|geosud|258|||||||||cluster||||00:00:56|56|0:0|00:00:02|2018-08-06T15:48:13|2018-08-06T15:48:16|0:0|1019|geosud|883299|883299|ZOO-Project_5bd1c32b-997f-11e8-9f78-0050569320d2_GSDBandMath_6_2_005||||||||||||||||||||28|1|muse044||4294360886|defq|qos_geosud|20|Unknown|Unknown|Unknown|Unknown|1||0n|1|cpu=1,node=1|||00:00:01|00:00:01|1|2018-08-06T15:48:14|COMPLETED|2018-08-06T15:48:13|00:00:00||UNLIMITED|00:00:00|1229|geosudwps|||0|
+   * 
+   */
+  ZOO_DLL_EXPORT int FinalizeHPC1(maps*& conf,maps*& inputs,maps*& outputs){
+    // Retrieve the jobid corresponding to the identifier generated by SLURM
+    // by reading the file generated when running the SBATCH file
+    map* jobid=getMapFromMaps(inputs,"jobid","value");
+    struct sockaddr_un addr;
+    char buf[100]="3";
+    int fd,rc=NULL;
+    int i=0;
+    map* usid=getMapFromMaps(conf,"lenv","usid");
+    map* tmpPath=getMapFromMaps(conf,"main","tmpPath");
+
+    char *flenv =
+      (char *) malloc ((strlen (tmpPath->value) + 
+                        strlen (jobid->value) + 12) * sizeof (char));
+    sprintf (flenv, "%s/%s_lenv.cfg", tmpPath->value, jobid->value);
+    maps* m = (maps *) malloc (MAPS_SIZE);
+    m->child=NULL;
+    m->next=NULL;
+    map* configId=NULL;
+
+    
+    if(conf_read(flenv, m) != 2){
+      configId=getMapFromMaps(m,"lenv","configId");
+      setMapInMaps(conf,"lenv","configId",configId->value);
+    }else{
+      setMapInMaps(conf,"lenv","message",_("Unable to read the lenv section file of the requested jobid"));
+      return SERVICE_FAILED;
+    }
+
+    SSHCON *test=ssh_connect(conf);
+    /*if(test==NULL){
+      setMapInMaps(conf,"lenv","message",_("Unable to connect using ssh."));
+      return SERVICE_FAILED;
+      }*/
+
+    char *logPath=(char*)malloc((strlen(tmpPath->value)+strlen(jobid->value)+12)*sizeof(char));
+    sprintf(logPath,"%s/exec_out_%s",tmpPath->value,jobid->value);
+    struct stat f_status;
+    int ts=stat(logPath, &f_status);
+    char* fcontent = NULL;
+    if(ts==0) {
+      fcontent=(char*)malloc(sizeof(char)*(f_status.st_size+1));
+      FILE* f=fopen(logPath,"rb");
+      fread(fcontent,f_status.st_size,1,f);
+      int fsize=f_status.st_size;
+      fcontent[fsize]=0;
+      fclose(f);
+    }else{
+      setMapInMaps(conf,"lenv","message",_("No service with this jobid can be found"));
+      return SERVICE_FAILED;
+    }
+    free(logPath);
+    // Run scontrol to check if the service execution ended.
+    // Store all the informations returned by scontrol command as a cfg file to
+    // be parsed back by the ZOO-Kernel waiting for the execution of the remote
+    // service
+    maps* tmpMaps=createMaps("henv");
+    
+    map* tmpMap=getMapFromMaps(conf,configId->value,"remote_command_opt");
+    char* command=(char*)malloc((126+strlen(tmpMap->value))*sizeof(char));
+    sprintf(command,"sacct --format=%s -p | grep \"%s\" | sed \"s:||:|None|:g;s:||:|:g\"",tmpMap->value,jobid->value);    
+    if(ssh_exec(conf,command,ssh_get_cnt(conf))==0){
+      free(command);
+      setMapInMaps(conf,"lenv","message",_("Failed to run scontrol remotely"));
+      // TODO: check status in db and if available continue in other case return SERVICE_FAILED
+      return SERVICE_FAILED;
+    }else{
+      free(command);
+      logPath=(char*)malloc((strlen(tmpPath->value)+strlen(usid->value)+11)*sizeof(char));
+      sprintf(logPath,"%s/exec_out_%s",tmpPath->value,usid->value);
+      int ts=stat(logPath, &f_status);
+      if(ts==0) {
+        fcontent=(char*)malloc(sizeof(char)*(f_status.st_size+1));
+        FILE* f=fopen(logPath,"rb");
+        fread(fcontent,f_status.st_size,1,f);
+        int fsize=f_status.st_size;
+        fcontent[fsize]=0;
+        fclose(f);
+        free(logPath);
+        char *token, *saveptr;
+        char *token1, *saveptr1;
+        token = strtok_r (tmpMap->value, ",", &saveptr);
+        token1 = strtok_r (fcontent, "|", &saveptr1);
+        while (token != NULL) {
+          fprintf(stderr,"%s %d %s \n",__FILE__,__LINE__,token);
+          fflush(stderr);
+          fprintf(stderr,"%s %d %s %s \n",__FILE__,__LINE__,token,token1);
+          fflush(stderr);
+          if(token1 != NULL){
+            if(tmpMaps->content==NULL)
+              tmpMaps->content=createMap(token,token1);
+            else
+              addToMap(tmpMaps->content,token,token1);
+          }
+          token = strtok_r (NULL, ",", &saveptr);
+          token1 = strtok_r (NULL, "|", &saveptr1);
+        }
+      }else{
+        setMapInMaps(conf,"lenv","message",_("Unable to access the downloaded execution log file"));
+        return SERVICE_FAILED;
+      }
+    }
+    logPath=(char*)malloc((strlen(tmpPath->value)+strlen(jobid->value)+15)*sizeof(char));
+    sprintf(logPath,"%s/exec_status_%s",tmpPath->value,jobid->value);
+    dumpMapsToFile(tmpMaps,logPath,0);
+    char *sname=(char*)malloc((strlen(tmpPath->value)+strlen(jobid->value)+21));
+    sprintf(sname,"%s/.wait_socket_%s.sock",tmpPath->value,jobid->value);
+    if ( (fd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
+      perror("socket error");
+      setMapInMaps(conf,"lenv","message",_("Socket error"));
+      return SERVICE_FAILED;
+    }
+    memset(&addr, 0, sizeof(addr));
+    addr.sun_family = AF_UNIX;
+    strncpy(addr.sun_path, sname, sizeof(addr.sun_path)-1);
+    if (connect(fd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
+      perror("connect error");
+      setMapInMaps(conf,"lenv","message",_("Unable to connect"));
+      return SERVICE_FAILED;
+    }
+    if (write(fd, "3", 1) != rc) {
+      if (rc < 0) {
+        perror("write error");
+        setMapInMaps(conf,"lenv","message",_("Unable to announce the successful execution of the HPC service"));
+        close(fd);
+        return SERVICE_FAILED;
+      }
+    }
+    close(fd);
+    unlink(flenv);
+    free(flenv);
+    setOutputValue(outputs,"Result",(char*)"\"FinalizeHPC run successfully\"",32);
+
+    return SERVICE_SUCCEEDED;
+  }
+  
 }