fix: updating return codes

all.sas

@@ -229,6 +229,32 @@ options noquotelenmax;
 %mend;
 
 /** @endcond *//**
+  @file
+  @brief Checks whether a fileref exists
+  @details You can probably do without this macro as it is just a one liner.
+  Mainly it is here as a convenient way to remember the syntax!
+
+  @param fref the fileref to detect
+
+  @return output Returns 1 if found and 0 if not found.  Note - it is possible
+  that the fileref is found, but the file does not (yet) exist. If you need
+  to test for this, you may as well use the fileref function directly.
+
+  @version 8
+  @author [Allan Bowe](https://www.linkedin.com/in/allanbowe/)
+**/
+
+%macro mf_existfileref(fref
+)/*/STORE SOURCE*/;
+
+  %if %sysfunc(fileref(&fref))=0 %then %do;
+    1
+  %end;
+  %else %do;
+    0
+  %end;
+
+%mend;/**
   @file
   @brief Checks if a variable exists in a data set.
   @details Returns 0 if the variable does NOT exist, and return the position of
@@ -577,11 +603,18 @@ options noquotelenmax;
   @brief Adds custom quotes / delimiters to a  delimited string
   @details Can be used in open code, eg as follows:
 
-    %put %mf_getquotedstr(blah   blah  blah);
+      %put %mf_getquotedstr(blah   blah  blah);
 
   which returns:
 > 'blah','blah','blah'
 
+  Alternatively:
+
+      %put %mf_getquotedstr(these words are double quoted,quote=D)
+
+  for:
+> "these","words","are","double","quoted"
+
   @param in_str the unquoted, spaced delimited string to transform
   @param dlm= the delimeter to be applied to the output (default comma)
   @param indlm= the delimeter used for the input (default is space)
@@ -4754,6 +4787,97 @@ proc sql
  %mp_binarycopy(inloc="&inloc",outref=_webout)
 %end;
 
+%mend;/**
+  @file
+  @brief Runs arbitrary code for a specified amount of time
+  @details Executes a series of procs and data steps to enable performance
+  testing of arbitrary jobs.
+
+      %mp_testjob(
+         duration=60*5
+      )
+
+  @param [in] duration= the time in seconds which the job should run for. Actual
+  time may vary, as the check is done in between steps.  Default = 30 (seconds).
+
+  <h4> SAS Macros </h4>
+  @li mf_getuniquelibref.sas
+  @li mf_getuniquename.sas
+  @li mf_mkdir.sas
+
+  @version 9.4
+  @author Allan Bowe
+
+**/
+
+%macro mp_testjob(duration=30
+)/*/STORE SOURCE*/;
+%local lib dir ds1 ds2 ds3 start_tm i;
+
+%let start_tm=%sysfunc(datetime());
+%let duration=%sysevalf(&duration);
+
+/* create a temporary library in WORK */
+%let lib=%mf_getuniquelibref();
+%let dir=%mf_getuniquename();
+%mf_mkdir(%sysfunc(pathname(work))/&dir)
+libname &lib "%sysfunc(pathname(work))/&dir";
+
+/* loop through until time expires */
+%let ds1=%mf_getuniquename();
+%let ds2=%mf_getuniquename();
+%let ds3=%mf_getuniquename();
+%do i=0 %to 1;
+
+  /* create big dataset */
+  data &lib..&ds1(compress=no );
+    do x=1 to 1000000;
+      randnum0=ranuni(0)*3;
+      randnum1=ranuni(0)*2;
+      bigchar=repeat('A',300);
+      output;
+    end;
+  run;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  proc summary ;
+    class randnum0 randnum1;
+    output out=&lib..&ds2;
+  run;quit;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  /* add more data */
+  proc sql;
+  create table &lib..&ds3 as
+    select *, ranuni(0)*10 as randnum2
+  from &lib..&ds1
+  order by randnum1;
+  quit;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  proc sort data=&lib..&ds3;
+    by descending x;
+  run;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  /* wait 5 seconds */
+  data _null_;
+    call sleep(5,1);
+  run;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  %let i=0;
+
+%end;
+
+%gate:
+%put time is up!;
+proc datasets lib=&lib kill;
+run;
+quit;
+libname &lib clear;
+
+
 %mend;/**
   @file mp_testwritespeedlibrary.sas
   @brief Tests the write speed of a new table in a SAS library
@@ -12366,7 +12490,7 @@ run;
 %let fpath3=%sysfunc(pathname(&fname2));
 
 /* compile the lua JSON module */
-%ml_json() 
+%ml_json()
 /* read using LUA - this allows the code to be of any length */
 data _null_;
   file "&fpath3..lua";
@@ -12398,6 +12522,192 @@ filename &fname1 clear;
 filename &fname2 clear;
 %mend;
 /**
+  @file
+  @brief Extract the log from a completed SAS Viya Job
+  @details Extracts log from a Viya job and writes it out to a fileref
+
+  To query the job, you need the URI.  Sample code for achieving this
+  is provided below.
+
+  ## Example
+
+  First, compile the macros:
+
+      filename mc url "https://raw.githubusercontent.com/sasjs/core/main/all.sas";
+      %inc mc;
+
+  Next, create a job (in this case, a web service):
+
+      filename ft15f001 temp;
+      parmcards4;
+        data ;
+          rand=ranuni(0)*1000;
+          do x=1 to rand;
+            y=rand*4;
+            output;
+          end;
+        run;
+        proc sort data=&syslast
+          by descending y;
+        run;
+      ;;;;
+      %mv_createwebservice(path=/Public/temp,name=demo)
+
+  Execute it:
+
+      %mv_jobexecute(path=/Public/temp
+        ,name=demo
+        ,outds=work.info
+      )
+
+  Wait for it to finish, and grab the uri:
+
+      data _null_;
+        set work.info;
+        if method='GET' and rel='self';
+        call symputx('uri',uri);
+      run;
+
+  Finally, fetch the log:
+
+      %mv_getjoblog(uri=&uri,outref=mylog)
+
+  This macro is used by the mv_jobwaitfor macro, which is generally a more
+  convenient way to wait for the job to finish before fetching the log.
+
+
+  @param [in] access_token_var= The global macro variable to contain the access token
+  @param [in] grant_type= valid values:
+    @li password
+    @li authorization_code
+    @li detect - will check if access_token exists, if not will use sas_services if
+        a SASStudioV session else authorization_code.  Default option.
+    @li sas_services - will use oauth_bearer=sas_services.
+  @param [in] uri= The uri of the running job for which to fetch the status,
+    in the format `/jobExecution/jobs/$UUID/state` (unquoted).
+  @param [out] outref= The output fileref to which to APPEND the log (is always
+  appended).
+
+
+  @version VIYA V.03.04
+  @author Allan Bowe, source: https://github.com/sasjs/core
+
+  <h4> SAS Macros </h4>
+  @li mp_abort.sas
+  @li mf_getplatform.sas
+  @li mf_existfileref.sas
+  @li ml_json.sas
+
+**/
+
+%macro mv_getjoblog(uri=0,outref=0
+    ,contextName=SAS Job Execution compute context
+    ,access_token_var=ACCESS_TOKEN
+    ,grant_type=sas_services
+  );
+%local oauth_bearer;
+%if &grant_type=detect %then %do;
+  %if %symexist(&access_token_var) %then %let grant_type=authorization_code;
+  %else %let grant_type=sas_services;
+%end;
+%if &grant_type=sas_services %then %do;
+    %let oauth_bearer=oauth_bearer=sas_services;
+    %let &access_token_var=;
+%end;
+%put &sysmacroname: grant_type=&grant_type;
+%mp_abort(iftrue=(&grant_type ne authorization_code and &grant_type ne password
+    and &grant_type ne sas_services
+  )
+  ,mac=&sysmacroname
+  ,msg=%str(Invalid value for grant_type: &grant_type)
+)
+
+/* validation in datastep for better character safety */
+%local errmsg errflg;
+data _null_;
+  uri=symget('uri');
+  if length(uri)<12 then do;
+    call symputx('errflg',1);
+    call symputx('errmsg',"URI is invalid (too short) - '&uri'",'l');
+  end;
+  if scan(uri,-1)='state' or scan(uri,1) ne 'jobExecution' then do;
+    call symputx('errflg',1);
+    call symputx('errmsg',
+      "URI should be in format /jobExecution/jobs/$$$$UUID$$$$"
+      !!" but is actually like: &uri",'l');
+  end;
+run;
+
+%mp_abort(iftrue=(&errflg=1)
+  ,mac=&sysmacroname
+  ,msg=%str(&errmsg)
+)
+
+%mp_abort(iftrue=(&outref=0)
+  ,mac=&sysmacroname
+  ,msg=%str(Output fileref should be provided)
+)
+
+%if %mf_existfileref(&outref) ne 1 %then %do;
+  filename &outref temp;
+%end;
+
+options noquotelenmax;
+%local base_uri; /* location of rest apis */
+%let base_uri=%mf_getplatform(VIYARESTAPI);
+
+/* prepare request*/
+%local  fname1;
+%let fname1=%mf_getuniquefileref();
+proc http method='GET' out=&fname1 &oauth_bearer
+  url="&base_uri&joburi";
+  headers
+  %if &grant_type=authorization_code %then %do;
+          "Authorization"="Bearer &&&access_token_var"
+  %end;
+  ;
+run;
+%if &SYS_PROCHTTP_STATUS_CODE ne 200 and &SYS_PROCHTTP_STATUS_CODE ne 201 %then
+%do;
+  data _null_;infile &fname1;input;putlog _infile_;run;
+  %mp_abort(mac=&sysmacroname
+    ,msg=%str(&SYS_PROCHTTP_STATUS_CODE &SYS_PROCHTTP_STATUS_PHRASE)
+  )
+%end;
+%local  fname2 fname3 fpath1 fpath2 fpath3;
+%let fname2=%mf_getuniquefileref();
+%let fname3=%mf_getuniquefileref();
+%let fpath1=%sysfunc(pathname(&fname1));
+%let fpath2=%sysfunc(pathname(&fname2));
+%let fpath3=%sysfunc(pathname(&fname2));
+
+/* compile the lua JSON module */
+%ml_json()
+/* read using LUA - this allows the code to be of any length */
+data _null_;
+  file "&fpath3..lua";
+  put '
+    infile = io.open (sas.symget("fpath1"), "r")
+    outfile = io.open (sas.symget("fpath2"), "w")
+    io.input(infile)
+    local resp=json.decode(io.read())
+    local logloc=resp["logLocation"]
+    outfile:write(logloc)
+    io.close(infile)
+    io.close(outfile)
+   ';
+run;
+%inc "&fpath3..lua";
+/* get log path*/
+data _null_;
+  infile &fname2;
+  input;
+  call symputx('logloc',_infile_,'l');
+run;
+%put &=logloc;
+filename &fname1 clear;
+filename &fname2 clear;
+%mend;/**
   @file
   @brief Extract the status from a running SAS Viya job
   @details Extracts the status from a running job and appends it to an output
@@ -12878,7 +13188,7 @@ libname &libref1 clear;
     %let oauth_bearer=oauth_bearer=sas_services;
     %let &access_token_var=;
 %end;
-%put &sysmacroname: grant_type=&grant_type;
+
 %mp_abort(iftrue=(&grant_type ne authorization_code and &grant_type ne password
     and &grant_type ne sas_services
   )
@@ -12984,12 +13294,31 @@ libname &libref;
   The input table is formed as per below.  Each observation represents one job.
   Each variable is converted into a macro variable with the same name.
 
-      | variable| description   |
-      |---|---|---|
-      |FLOW_ID| Numeric value, provides sequential ordering capability|
-      |_CONTEXTNAME|Dictates which context should be used to run the job. If
-       blank, will default to `SAS Job Execution compute context`.|
-      |_PROGRAM|Provides the path to the job itself|
+  ## Input table (minimum variables needed)
+
+  @li FLOW_ID - Numeric value, provides sequential ordering capability
+  @li _CONTEXTNAME - Dictates which context should be used to run the job. If
+    blank, will default to `SAS Job Execution compute context`.
+  @li _PROGRAM - Provides the path to the job itself
+
+  Any additional variables provided in this table are converted into macro
+  variables and passed into the relevant job.
+
+  | FLOW_ID| _CONTEXTNAME   |_PROGRAM|
+  |---|---|---|
+  |0|SAS Job Execution compute context|/Public/jobs/somejob1|
+  |0|SAS Job Execution compute context|/Public/jobs/somejob2|
+
+  ## Output table (minimum variables produced)
+
+  @li _PROGRAM - the SAS Drive path of the job
+  @li URI - the URI of the executed job
+  @li STATE - the completed state of the job
+  @li TIMESTAMP - the datetime that the job completed
+  @li JOBPARAMS - the parameters that were passed to the job
+  @li FLOW_ID - the id of the flow in which the job was executed
+
+  ![https://i.imgur.com/nZE9PvT.png](https://i.imgur.com/nZE9PvT.png)
 
 
   ## Example
@@ -13010,6 +13339,7 @@ libname &libref;
           rand=ranuni(0)*&macrovar1;
           do x=1 to rand;
             y=rand*&macrovar2;
+            if y=100 then abort;
             output;
           end;
         run;
@@ -13022,17 +13352,17 @@ libname &libref;
       data work.inputjobs;
         _contextName='SAS Job Execution compute context';
         do flow_id=1 to 3;
-          do job=1 to 10;
+          do i=1 to 20;
             _program='/Public/temp/demo1';
-            macrovar1=10*job;
-            macrovar2=4*job;
+            macrovar1=10*i;
+            macrovar2=4*i;
             output;
+            i+1;
             _program='/Public/temp/demo2';
-            macrovar1=40*job;
-            macrovar2=44*job;
+            macrovar1=40*i;
+            macrovar2=44*i;
             output;
           end;
-          drop job;
         end;
       run;
 
@@ -13056,10 +13386,13 @@ libname &libref;
   @author Allan Bowe, source: https://github.com/sasjs/core
 
   <h4> SAS Macros </h4>
+  @li mf_nobs.sas
   @li mp_abort.sas
   @li mf_getplatform.sas
   @li mf_getuniquefileref.sas
   @li mf_existvarlist.sas
+  @li mv_jobwaitfor.sas
+  @li mv_jobexecute.sas
 
 **/
 
@@ -13077,7 +13410,7 @@ libname &libref;
     %let oauth_bearer=oauth_bearer=sas_services;
     %let &access_token_var=;
 %end;
-%put &sysmacroname: grant_type=&grant_type;
+
 %mp_abort(iftrue=(&grant_type ne authorization_code and &grant_type ne password
     and &grant_type ne sas_services
   )
@@ -13109,6 +13442,16 @@ select count(*) into: missings
   ,msg=%str(input dataset contains &missings missing values for FLOW_ID or _PROGRAM)
 )
 
+%if %mf_nobs(&inds)=0 %then %do;
+  %put No observations in &inds!  Leaving macro &sysmacroname;
+  %return;
+%end;
+
+/* ensure output table is available */
+data &outds;run;
+proc sql;
+drop table &outds;
+
 options noquotelenmax;
 %local base_uri; /* location of rest apis */
 %let base_uri=%mf_getplatform(VIYARESTAPI);
@@ -13163,13 +13506,15 @@ data;run;%let jdswaitfor=&syslast;
     data _null_;
       infile &jfref lrecl=32767;
       input;
-      call symputx("jparams&jid",substr(_infile_,3,length(_infile_)-4));
+      jparams='jparams'!!left(symget('jid'));
+      call symputx(jparams,substr(_infile_,3,length(_infile_)-4));
     run;
     %local joburi&jid;
     %let joburi&jid=0; /* used in next loop */
   %end;
-  %local concurrency;
+  %local concurrency completed;
   %let concurrency=0;
+  %let completed=0;
   proc sql; drop table &jdsrunning;
   %do jid=1 %to &jcnt;
     /**
@@ -13187,16 +13532,54 @@ data;run;%let jdswaitfor=&syslast;
           ,paramstring=%superq(jparams&jid)
           ,outds=&jdsapp
         )
-        proc append base=&jdsrunning
-          data=&jdsapp(where=(method='GET' and rel='state'));
+        data &jdsapp;
+          format jobparams $32767.;
+          set &jdsapp(where=(method='GET' and rel='state'));
+          jobparams=symget("jparams&jid");
+          call symputx("joburi&jid",uri,'l');
+        run;
+        proc append base=&jdsrunning data=&jdsapp;
         run;
         %let concurrency=%eval(&concurrency+1);
-        %let joburi&jid=1;
+      %end;
+      %else %if %sysfunc(exist(&outds))=1 %then %do;
+        /* check to see if the job has finished as was previously executed */
+        %local jobcheck;  %let jobcheck=0;
+        proc sql noprint;
+        select count(*) into: jobcheck
+          from &outds where uri="&&joburi&jid";
+        %if &jobcheck>0 %then %do;
+          %put &&job&jid in flow &fid with uri &&joburi&jid completed!;
+          %let job&jid=0;
+        %end;
       %end;
     %end;
     %if &jid=&jcnt %then %do;
       /* we are at the end of the loop - time to see which jobs have finished */
       %mv_jobwaitfor(ANY,inds=&jdsrunning,outds=&jdswaitfor)
+      %local done;
+      %let done=%mf_nobs(&jdswaitfor);
+      %if &done>0 %then %do;
+        %let completed=%eval(&completed+&done);
+        %let concurrency=%eval(&concurrency-&done);
+        data &jdsapp;
+          set &jdswaitfor;
+          flow_id=&&flow&fid;
+        run;
+        proc append base=&outds data=&jdsapp;
+        run;
+      %end;
+      proc sql;
+      delete from &jdsrunning
+        where uri in (select uri from &outds
+          where state in ('canceled','completed','failed')
+        );
+
+      /* loop again if jobs are left */
+      %if &completed < &jcnt %then %do;
+        %let jid=0;
+        %put looping flow &fid again - &completed of &jcnt jobs completed, &concurrency jobs running;
+      %end;
     %end;
   %end;
   /* back up and execute the next flow */
@@ -13307,7 +13690,7 @@ data;run;%let jdswaitfor=&syslast;
     %let oauth_bearer=oauth_bearer=sas_services;
     %let &access_token_var=;
 %end;
-%put &sysmacroname: grant_type=&grant_type;
+
 %mp_abort(iftrue=(&grant_type ne authorization_code and &grant_type ne password
     and &grant_type ne sas_services
   )
@@ -13338,9 +13721,11 @@ options noquotelenmax;
 %let base_uri=%mf_getplatform(VIYARESTAPI);
 
 data _null_;
+  length jobparams $32767;
   set &inds end=last;
   call symputx(cats('joburi',_n_),uri,'l');
   call symputx(cats('jobname',_n_),_program,'l');
+  call symputx(cats('jobparams',_n_),jobparams,'l');
   if last then call symputx('uricnt',_n_,'l');
 run;
 
@@ -13353,7 +13738,7 @@ run;
 %let fname0=%mf_getuniquefileref();
 
 data &outds;
-  format _program uri $128. state $32. timestamp datetime19.;
+  format _program uri $128. state $32. timestamp datetime19. jobparams $32767.;
   stop;
 run;
 
@@ -13387,7 +13772,8 @@ run;
         _program="&&jobname&i",
         uri="&&joburi&i",
         state="&status",
-        timestamp=datetime();
+        timestamp=datetime(),
+        jobparams=symget("jobparams&i");
       %let joburi&i=0; /* do not re-check */
     %end;
     %else %if &status=idle or &status=pending or &status=running %then %do;

base/mf_existfileref.sas

@@ -0,0 +1,27 @@
+/**
+  @file
+  @brief Checks whether a fileref exists
+  @details You can probably do without this macro as it is just a one liner.
+  Mainly it is here as a convenient way to remember the syntax!
+
+  @param fref the fileref to detect
+
+  @return output Returns 1 if found and 0 if not found.  Note - it is possible
+  that the fileref is found, but the file does not (yet) exist. If you need
+  to test for this, you may as well use the fileref function directly.
+
+  @version 8
+  @author [Allan Bowe](https://www.linkedin.com/in/allanbowe/)
+**/
+
+%macro mf_existfileref(fref
+)/*/STORE SOURCE*/;
+
+  %if %sysfunc(fileref(&fref))=0 %then %do;
+    1
+  %end;
+  %else %do;
+    0
+  %end;
+
+%mend;

base/mf_getquotedstr.sas

@@ -3,11 +3,18 @@
   @brief Adds custom quotes / delimiters to a  delimited string
   @details Can be used in open code, eg as follows:
 
-    %put %mf_getquotedstr(blah   blah  blah);
+      %put %mf_getquotedstr(blah   blah  blah);
 
   which returns:
 > 'blah','blah','blah'
 
+  Alternatively:
+
+      %put %mf_getquotedstr(these words are double quoted,quote=D)
+
+  for:
+> "these","words","are","double","quoted"
+
   @param in_str the unquoted, spaced delimited string to transform
   @param dlm= the delimeter to be applied to the output (default comma)
   @param indlm= the delimeter used for the input (default is space)

base/mp_testjob.sas

@@ -0,0 +1,92 @@
+/**
+  @file
+  @brief Runs arbitrary code for a specified amount of time
+  @details Executes a series of procs and data steps to enable performance
+  testing of arbitrary jobs.
+
+      %mp_testjob(
+         duration=60*5
+      )
+
+  @param [in] duration= the time in seconds which the job should run for. Actual
+  time may vary, as the check is done in between steps.  Default = 30 (seconds).
+
+  <h4> SAS Macros </h4>
+  @li mf_getuniquelibref.sas
+  @li mf_getuniquename.sas
+  @li mf_mkdir.sas
+
+  @version 9.4
+  @author Allan Bowe
+
+**/
+
+%macro mp_testjob(duration=30
+)/*/STORE SOURCE*/;
+%local lib dir ds1 ds2 ds3 start_tm i;
+
+%let start_tm=%sysfunc(datetime());
+%let duration=%sysevalf(&duration);
+
+/* create a temporary library in WORK */
+%let lib=%mf_getuniquelibref();
+%let dir=%mf_getuniquename();
+%mf_mkdir(%sysfunc(pathname(work))/&dir)
+libname &lib "%sysfunc(pathname(work))/&dir";
+
+/* loop through until time expires */
+%let ds1=%mf_getuniquename();
+%let ds2=%mf_getuniquename();
+%let ds3=%mf_getuniquename();
+%do i=0 %to 1;
+
+  /* create big dataset */
+  data &lib..&ds1(compress=no );
+    do x=1 to 1000000;
+      randnum0=ranuni(0)*3;
+      randnum1=ranuni(0)*2;
+      bigchar=repeat('A',300);
+      output;
+    end;
+  run;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  proc summary ;
+    class randnum0 randnum1;
+    output out=&lib..&ds2;
+  run;quit;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  /* add more data */
+  proc sql;
+  create table &lib..&ds3 as
+    select *, ranuni(0)*10 as randnum2
+  from &lib..&ds1
+  order by randnum1;
+  quit;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  proc sort data=&lib..&ds3;
+    by descending x;
+  run;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  /* wait 5 seconds */
+  data _null_;
+    call sleep(5,1);
+  run;
+  %if %sysevalf( (%sysfunc(datetime())-&start_tm)>&duration ) %then %goto gate;
+
+  %let i=0;
+
+%end;
+
+%gate:
+%put time is up!;
+proc datasets lib=&lib kill;
+run;
+quit;
+libname &lib clear;
+
+
+%mend;

viya/mv_jobexecute.sas

@@ -72,7 +72,7 @@
     %let oauth_bearer=oauth_bearer=sas_services;
     %let &access_token_var=;
 %end;
-%put &sysmacroname: grant_type=&grant_type;
+
 %mp_abort(iftrue=(&grant_type ne authorization_code and &grant_type ne password
     and &grant_type ne sas_services
   )

viya/mv_jobflow.sas

@@ -0,0 +1,302 @@
+/**
+  @file
+  @brief Execute a series of job flows
+  @details Very (very) simple flow manager.  Jobs execute in sequential waves,
+  all previous waves must finish successfully.
+
+  The input table is formed as per below.  Each observation represents one job.
+  Each variable is converted into a macro variable with the same name.
+
+  ## Input table (minimum variables needed)
+
+  @li FLOW_ID - Numeric value, provides sequential ordering capability
+  @li _CONTEXTNAME - Dictates which context should be used to run the job. If
+    blank, will default to `SAS Job Execution compute context`.
+  @li _PROGRAM - Provides the path to the job itself
+
+  Any additional variables provided in this table are converted into macro
+  variables and passed into the relevant job.
+
+  | FLOW_ID| _CONTEXTNAME   |_PROGRAM|
+  |---|---|---|
+  |0|SAS Job Execution compute context|/Public/jobs/somejob1|
+  |0|SAS Job Execution compute context|/Public/jobs/somejob2|
+
+  ## Output table (minimum variables produced)
+
+  @li _PROGRAM - the SAS Drive path of the job
+  @li URI - the URI of the executed job
+  @li STATE - the completed state of the job
+  @li TIMESTAMP - the datetime that the job completed
+  @li JOBPARAMS - the parameters that were passed to the job
+  @li FLOW_ID - the id of the flow in which the job was executed
+
+  ![https://i.imgur.com/nZE9PvT.png](https://i.imgur.com/nZE9PvT.png)
+
+
+  ## Example
+
+  First, compile the macros:
+
+      filename mc url
+      "https://raw.githubusercontent.com/sasjs/core/main/all.sas";
+      %inc mc;
+
+  Next, create some jobs (in this case, as web services):
+
+      filename ft15f001 temp;
+      parmcards4;
+        %put this is job: &_program;
+        %put this was run in flow &flow_id;
+        data ;
+          rand=ranuni(0)*&macrovar1;
+          do x=1 to rand;
+            y=rand*&macrovar2;
+            if y=100 then abort;
+            output;
+          end;
+        run;
+      ;;;;
+      %mv_createwebservice(path=/Public/temp,name=demo1)
+      %mv_createwebservice(path=/Public/temp,name=demo2)
+
+  Prepare an input table with 60 executions:
+
+      data work.inputjobs;
+        _contextName='SAS Job Execution compute context';
+        do flow_id=1 to 3;
+          do i=1 to 20;
+            _program='/Public/temp/demo1';
+            macrovar1=10*i;
+            macrovar2=4*i;
+            output;
+            i+1;
+            _program='/Public/temp/demo2';
+            macrovar1=40*i;
+            macrovar2=44*i;
+            output;
+          end;
+        end;
+      run;
+
+  Trigger the flow
+
+      %mv_jobflow(inds=work.inputjobs,outds=work.results,maxconcurrency=4)
+
+
+  @param [in] access_token_var= The global macro variable to contain the access token
+  @param [in] grant_type= valid values:
+      @li password
+      @li authorization_code
+      @li detect - will check if access_token exists, if not will use sas_services if
+        a SASStudioV session else authorization_code.  Default option.
+      @li sas_services - will use oauth_bearer=sas_services
+  @param [in] inds= The input dataset containing a list of jobs and parameters
+  @param [in] maxconcurrency= The max number of parallel jobs to run.  Default=8.
+  @param [out] outds= The output dataset containing the results
+
+  @version VIYA V.03.05
+  @author Allan Bowe, source: https://github.com/sasjs/core
+
+  <h4> SAS Macros </h4>
+  @li mf_nobs.sas
+  @li mp_abort.sas
+  @li mf_getplatform.sas
+  @li mf_getuniquefileref.sas
+  @li mf_existvarlist.sas
+  @li mv_jobwaitfor.sas
+  @li mv_jobexecute.sas
+
+**/
+
+%macro mv_jobflow(inds=0,outds=work.mv_jobflow
+    ,maxconcurrency=8
+    ,access_token_var=ACCESS_TOKEN
+    ,grant_type=sas_services
+  );
+%local oauth_bearer;
+%if &grant_type=detect %then %do;
+  %if %symexist(&access_token_var) %then %let grant_type=authorization_code;
+  %else %let grant_type=sas_services;
+%end;
+%if &grant_type=sas_services %then %do;
+    %let oauth_bearer=oauth_bearer=sas_services;
+    %let &access_token_var=;
+%end;
+
+%mp_abort(iftrue=(&grant_type ne authorization_code and &grant_type ne password
+    and &grant_type ne sas_services
+  )
+  ,mac=&sysmacroname
+  ,msg=%str(Invalid value for grant_type: &grant_type)
+)
+
+%mp_abort(iftrue=("&inds"="0")
+  ,mac=&sysmacroname
+  ,msg=%str(Input dataset was not provided)
+)
+%mp_abort(iftrue=(%mf_existVarList(&inds,_CONTEXTNAME FLOW_ID _PROGRAM)=0)
+  ,mac=&sysmacroname
+  ,msg=%str(The following columns must exist on input dataset &inds:
+    _CONTEXTNAME FLOW_ID _PROGRAM)
+)
+%mp_abort(iftrue=(&maxconcurrency<1)
+  ,mac=&sysmacroname
+  ,msg=%str(The maxconcurrency variable should be a positive integer)
+)
+
+%local missings;
+proc sql noprint;
+select count(*) into: missings
+  from &inds
+  where flow_id is null or _program is null;
+%mp_abort(iftrue=(&missings>0)
+  ,mac=&sysmacroname
+  ,msg=%str(input dataset contains &missings missing values for FLOW_ID or _PROGRAM)
+)
+
+%if %mf_nobs(&inds)=0 %then %do;
+  %put No observations in &inds!  Leaving macro &sysmacroname;
+  %return;
+%end;
+
+/* ensure output table is available */
+data &outds;run;
+proc sql;
+drop table &outds;
+
+options noquotelenmax;
+%local base_uri; /* location of rest apis */
+%let base_uri=%mf_getplatform(VIYARESTAPI);
+
+
+/* get flows */
+proc sort data=&inds;
+  by flow_id;
+run;
+data _null_;
+  set &inds (keep=flow_id) end=last;
+  by flow_id;
+  if last.flow_id then do;
+    cnt+1;
+    call symputx(cats('flow',cnt),flow_id,'l');
+  end;
+  if last then call symputx('flowcnt',cnt,'l');
+run;
+
+/* prepare temporary datasets and frefs */
+%local fid jid jds jjson jdsapp jdsrunning jdswaitfor jfref;
+data;run;%let jds=&syslast;
+data;run;%let jjson=&syslast;
+data;run;%let jdsapp=&syslast;
+data;run;%let jdsrunning=&syslast;
+data;run;%let jdswaitfor=&syslast;
+%let jfref=%mf_getuniquefileref();
+
+/* start loop */
+%do fid=1 %to &flowcnt;
+  %put preparing job attributes for flow &&flow&fid;
+  %local jds jcnt;
+  data &jds(drop=_contextName _program);
+    set &inds(where=(flow_id=&&flow&fid));
+    if _contextName='' then _contextName="SAS Job Execution compute context";
+    call symputx(cats('job',_n_),_program,'l');
+    call symputx(cats('context',_n_),_contextName,'l');
+    call symputx('jcnt',_n_,'l');
+  run;
+  %put exporting job variables in json format;
+  %do jid=1 %to &jcnt;
+    data &jjson;
+      set &jds;
+      if _n_=&jid then do;
+        output;
+        stop;
+      end;
+    run;
+    proc json out=&jfref;
+      export &jjson / nosastags fmtnumeric;
+    run;
+    data _null_;
+      infile &jfref lrecl=32767;
+      input;
+      jparams='jparams'!!left(symget('jid'));
+      call symputx(jparams,substr(_infile_,3,length(_infile_)-4));
+    run;
+    %local joburi&jid;
+    %let joburi&jid=0; /* used in next loop */
+  %end;
+  %local concurrency completed;
+  %let concurrency=0;
+  %let completed=0;
+  proc sql; drop table &jdsrunning;
+  %do jid=1 %to &jcnt;
+    /**
+      * now we can execute the jobs up to the maxconcurrency setting
+      */
+    %if "&&job&jid" ne "0" %then %do; /* this var is zero if job finished */
+      %if "&&joburi&jid"="0" and &concurrency<&maxconcurrency %then %do;
+        /* job has not been triggered and we have free slots */
+        %local jobname jobpath;
+        %let jobname=%scan(&&job&jid,-1,/);
+        %let jobpath=%substr(&&job&jid,1,%length(&&job&jid)-%length(&jobname)-1);
+        %put executing &jobpath/&jobname with paramstring &&jparams&jid;
+        %mv_jobexecute(path=&jobpath
+          ,name=&jobname
+          ,paramstring=%superq(jparams&jid)
+          ,outds=&jdsapp
+        )
+        data &jdsapp;
+          format jobparams $32767.;
+          set &jdsapp(where=(method='GET' and rel='state'));
+          jobparams=symget("jparams&jid");
+          call symputx("joburi&jid",uri,'l');
+        run;
+        proc append base=&jdsrunning data=&jdsapp;
+        run;
+        %let concurrency=%eval(&concurrency+1);
+      %end;
+      %else %if %sysfunc(exist(&outds))=1 %then %do;
+        /* check to see if the job has finished as was previously executed */
+        %local jobcheck;  %let jobcheck=0;
+        proc sql noprint;
+        select count(*) into: jobcheck
+          from &outds where uri="&&joburi&jid";
+        %if &jobcheck>0 %then %do;
+          %put &&job&jid in flow &fid with uri &&joburi&jid completed!;
+          %let job&jid=0;
+        %end;
+      %end;
+    %end;
+    %if &jid=&jcnt %then %do;
+      /* we are at the end of the loop - time to see which jobs have finished */
+      %mv_jobwaitfor(ANY,inds=&jdsrunning,outds=&jdswaitfor)
+      %local done;
+      %let done=%mf_nobs(&jdswaitfor);
+      %if &done>0 %then %do;
+        %let completed=%eval(&completed+&done);
+        %let concurrency=%eval(&concurrency-&done);
+        data &jdsapp;
+          set &jdswaitfor;
+          flow_id=&&flow&fid;
+        run;
+        proc append base=&outds data=&jdsapp;
+        run;
+      %end;
+      proc sql;
+      delete from &jdsrunning
+        where uri in (select uri from &outds
+          where state in ('canceled','completed','failed')
+        );
+
+      /* loop again if jobs are left */
+      %if &completed < &jcnt %then %do;
+        %let jid=0;
+        %put looping flow &fid again - &completed of &jcnt jobs completed, &concurrency jobs running;
+      %end;
+    %end;
+  %end;
+  /* back up and execute the next flow */
+%end;
+
+
+%mend;

viya/mv_jobwaitfor.sas

@@ -101,7 +101,7 @@
     %let oauth_bearer=oauth_bearer=sas_services;
     %let &access_token_var=;
 %end;
-%put &sysmacroname: grant_type=&grant_type;
+
 %mp_abort(iftrue=(&grant_type ne authorization_code and &grant_type ne password
     and &grant_type ne sas_services
   )
@@ -132,9 +132,11 @@ options noquotelenmax;
 %let base_uri=%mf_getplatform(VIYARESTAPI);
 
 data _null_;
+  length jobparams $32767;
   set &inds end=last;
   call symputx(cats('joburi',_n_),uri,'l');
   call symputx(cats('jobname',_n_),_program,'l');
+  call symputx(cats('jobparams',_n_),jobparams,'l');
   if last then call symputx('uricnt',_n_,'l');
 run;
 
@@ -147,7 +149,7 @@ run;
 %let fname0=%mf_getuniquefileref();
 
 data &outds;
-  format _program uri $128. state $32. timestamp datetime19.;
+  format _program uri $128. state $32. timestamp datetime19. jobparams $32767.;
   stop;
 run;
 
@@ -181,7 +183,8 @@ run;
         _program="&&jobname&i",
         uri="&&joburi&i",
         state="&status",
-        timestamp=datetime();
+        timestamp=datetime(),
+        jobparams=symget("jobparams&i");
       %let joburi&i=0; /* do not re-check */
     %end;
     %else %if &status=idle or &status=pending or &status=running %then %do;