Get job id using query params (#1480)

vishalbollu · web-flow · commit 1cd918e82775 · 2020-10-26T21:48:28.000-04:00
diff --git a/cli/cluster/delete.go b/cli/cluster/delete.go
@@ -79,7 +79,7 @@ func StopJob(operatorConfig OperatorConfig, apiName string, jobID string) (schem
 		"jobID":   jobID,
 	}
 
-	httpRes, err := HTTPDelete(operatorConfig, path.Join("/batch", apiName, jobID), params)
+	httpRes, err := HTTPDelete(operatorConfig, path.Join("/batch", apiName), params)
 	if err != nil {
 		return schema.DeleteResponse{}, err
 	}
diff --git a/cli/cluster/get.go b/cli/cluster/get.go
@@ -52,8 +52,8 @@ func GetAPI(operatorConfig OperatorConfig, apiName string) ([]schema.APIResponse
 }
 
 func GetJob(operatorConfig OperatorConfig, apiName string, jobID string) (schema.JobResponse, error) {
-	endpoint := path.Join("/batch", apiName, jobID)
-	httpRes, err := HTTPGet(operatorConfig, endpoint)
+	endpoint := path.Join("/batch", apiName)
+	httpRes, err := HTTPGet(operatorConfig, endpoint, map[string]string{"jobID": jobID})
 	if err != nil {
 		return schema.JobResponse{}, err
 	}
diff --git a/cli/cluster/lib_http_client.go b/cli/cluster/lib_http_client.go
@@ -52,7 +52,7 @@ func (oc OperatorConfig) AuthHeader() string {
 }
 
 func HTTPGet(operatorConfig OperatorConfig, endpoint string, qParams ...map[string]string) ([]byte, error) {
-	req, err := operatorRequest(operatorConfig, "GET", endpoint, nil, qParams)
+	req, err := operatorRequest(operatorConfig, "GET", endpoint, nil, qParams...)
 	if err != nil {
 		return nil, err
 	}
@@ -69,7 +69,7 @@ func HTTPPostObjAsJSON(operatorConfig OperatorConfig, endpoint string, requestDa
 
 func HTTPPostJSON(operatorConfig OperatorConfig, endpoint string, jsonRequestData []byte, qParams ...map[string]string) ([]byte, error) {
 	payload := bytes.NewBuffer(jsonRequestData)
-	req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, payload, qParams)
+	req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, payload, qParams...)
 	if err != nil {
 		return nil, err
 	}
@@ -78,15 +78,15 @@ func HTTPPostJSON(operatorConfig OperatorConfig, endpoint string, jsonRequestDat
 }
 
 func HTTPPostNoBody(operatorConfig OperatorConfig, endpoint string, qParams ...map[string]string) ([]byte, error) {
-	req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, nil, qParams)
+	req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, nil, qParams...)
 	if err != nil {
 		return nil, err
 	}
 	return makeOperatorRequest(operatorConfig, req)
 }
 
 func HTTPDelete(operatorConfig OperatorConfig, endpoint string, qParams ...map[string]string) ([]byte, error) {
-	req, err := operatorRequest(operatorConfig, http.MethodDelete, endpoint, nil, qParams)
+	req, err := operatorRequest(operatorConfig, http.MethodDelete, endpoint, nil, qParams...)
 	if err != nil {
 		return nil, err
 	}
@@ -124,7 +124,7 @@ func HTTPUpload(operatorConfig OperatorConfig, endpoint string, input *HTTPUploa
 		return nil, errors.Wrap(err, _errStrCantMakeRequest)
 	}
 
-	req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, body, qParams)
+	req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, body, qParams...)
 	if err != nil {
 		return nil, err
 	}
@@ -159,7 +159,7 @@ func HTTPUploadZip(operatorConfig OperatorConfig, endpoint string, zipInput *arc
 	return HTTPUpload(operatorConfig, endpoint, uploadInput, qParams...)
 }
 
-func operatorRequest(operatorConfig OperatorConfig, method string, endpoint string, body io.Reader, qParams []map[string]string) (*http.Request, error) {
+func operatorRequest(operatorConfig OperatorConfig, method string, endpoint string, body io.Reader, qParams ...map[string]string) (*http.Request, error) {
 	req, err := http.NewRequest(method, operatorConfig.OperatorEndpoint+endpoint, body)
 	if err != nil {
 		return nil, errors.Wrap(err, _errStrCantMakeRequest)
diff --git a/cli/cluster/logs.go b/cli/cluster/logs.go
@@ -34,10 +34,18 @@ import (
 )
 
 func StreamLogs(operatorConfig OperatorConfig, apiName string) error {
+	return streamLogs(operatorConfig, "/logs/"+apiName)
+}
+
+func StreamJobLogs(operatorConfig OperatorConfig, apiName string, jobID string) error {
+	return streamLogs(operatorConfig, "/logs/"+apiName, map[string]string{"jobID": jobID})
+}
+
+func streamLogs(operatorConfig OperatorConfig, path string, qParams ...map[string]string) error {
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt)
 
-	req, err := operatorRequest(operatorConfig, "GET", "/logs/"+apiName, nil, nil)
+	req, err := operatorRequest(operatorConfig, "GET", path, nil, qParams...)
 	if err != nil {
 		return err
 	}
diff --git a/cli/cmd/logs.go b/cli/cmd/logs.go
@@ -18,7 +18,6 @@ package cmd
 
 import (
 	"fmt"
-	"path"
 
 	"github.com/cortexlabs/cortex/cli/cluster"
 	"github.com/cortexlabs/cortex/cli/local"
@@ -54,10 +53,17 @@ var _logsCmd = &cobra.Command{
 
 		apiName := args[0]
 		if env.Provider == types.AWSProviderType {
-			logPath := path.Join(args...)
-			err := cluster.StreamLogs(MustGetOperatorConfig(env.Name), logPath)
-			if err != nil {
-				exit.Error(err)
+			if len(args) == 1 {
+				err := cluster.StreamLogs(MustGetOperatorConfig(env.Name), apiName)
+				if err != nil {
+					exit.Error(err)
+				}
+			}
+			if len(args) == 2 {
+				err := cluster.StreamJobLogs(MustGetOperatorConfig(env.Name), apiName, args[1])
+				if err != nil {
+					exit.Error(err)
+				}
 			}
 		} else {
 			if len(args) == 2 {
diff --git a/docs/deployments/batch-api/endpoints.md b/docs/deployments/batch-api/endpoints.md
@@ -33,7 +33,7 @@ Submitting data in the request can be useful in the following scenarios:
 * you want to avoid using S3 as an intermediate storage layer
 
 ```yaml
-POST <batch_api_endpoint>/:
+POST <batch_api_endpoint>:
 {
     "workers": <int>,         # the number of workers to allocate for this job (required)
     "item_list": {
@@ -74,7 +74,7 @@ This submission pattern can be useful in the following scenarios:
 If a single S3 file contains a lot of samples/rows, try the next submission strategy.
 
 ```yaml
-POST <batch_api_endpoint>/:
+POST <batch_api_endpoint>:
 {
     "workers": <int>,            # the number of workers to allocate for this job (required)
     "file_path_lister": {
@@ -113,7 +113,7 @@ This submission pattern is useful in the following scenarios:
 * one or more S3 files contains a large number of samples and must be broken down into batches
 
 ```yaml
-POST <batch_api_endpoint>/:
+POST <batch_api_endpoint>:
 {
     "workers": <int>,            # the number of workers to allocate for this job (required)
     "delimited_files": {
@@ -146,7 +146,7 @@ You can get the status of a job by making a GET request to `<batch_api_endpoint>
 See [Job Status Codes](statuses.md) for a list of the possible job statuses and what they mean.
 
 ```yaml
-GET <batch_api_endpoint>/<job_id>:
+GET <batch_api_endpoint>?jobID=<jobID>:
 
 RESPONSE:
 {
@@ -188,7 +188,7 @@ Stop a job in progress. You can also use the Cortex CLI command
 You stop a running job by making a DELETE request to `<batch_api_endpoint>/<job_id>` (note that you can also delete a job with the Cortex CLI command `cortex delete <api_name> <job_id>`).
 
 ```yaml
-DELETE <batch_api_endpoint>/<job_id>:
+DELETE <batch_api_endpoint>?jobID=<jobID>:
 
 RESPONSE:
 {"message":"stopped job <job_id>"}
diff --git a/examples/batch/image-classifier/README.md b/examples/batch/image-classifier/README.md
@@ -234,7 +234,7 @@ endpoint: https://abcdefg.execute-api.us-west-2.amazonaws.com/image-classifier
 You can make a GET request to your `<BATCH_API_ENDPOINT>/JOB_ID` to get the status of your job.
 
 ```bash
-$ curl https://abcdefg.execute-api.us-west-2.amazonaws.com/69d6faf82e4660d3
+$ curl https://abcdefg.execute-api.us-west-2.amazonaws.com?jobID=69d6faf82e4660d3
 
 {
     "job_status":{
@@ -545,7 +545,7 @@ You can stop a running job by sending a DELETE request to `<BATCH_API_ENDPOINT>/
 
 ```bash
 $ export BATCH_API_ENDPOINT=<BATCH_API_ENDPOINT>  # e.g. export BATCH_API_ENDPOINT=https://abcdefg.execute-api.us-west-2.amazonaws.com/image-classifier
-$ curl -X DELETE $BATCH_API_ENDPOINT/69d96a01ea55da8c
+$ curl -X DELETE $BATCH_API_ENDPOINT?jobID=69d96a01ea55da8c
 
 stopped job 69d96a01ea55da8c
 ```
diff --git a/pkg/operator/endpoints/get_job.go b/pkg/operator/endpoints/get_job.go
@@ -32,7 +32,11 @@ import (
 func GetJob(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	apiName := vars["apiName"]
-	jobID := vars["jobID"]
+	jobID, err := getRequiredQueryParam("jobID", r)
+	if err != nil {
+		respondError(w, r, err)
+		return
+	}
 
 	deployedResource, err := resources.GetDeployedResourceByName(apiName)
 	if err != nil {
diff --git a/pkg/operator/endpoints/logs.go b/pkg/operator/endpoints/logs.go
@@ -28,6 +28,12 @@ import (
 
 func ReadLogs(w http.ResponseWriter, r *http.Request) {
 	apiName := mux.Vars(r)["apiName"]
+	jobID := getOptionalQParam("jobID", r)
+
+	if jobID != "" {
+		ReadJobLogs(w, r)
+		return
+	}
 
 	deployedResource, err := resources.GetDeployedResourceByName(apiName)
 	if err != nil {
diff --git a/pkg/operator/endpoints/logs_job.go b/pkg/operator/endpoints/logs_job.go
@@ -29,7 +29,11 @@ import (
 
 func ReadJobLogs(w http.ResponseWriter, r *http.Request) {
 	apiName := mux.Vars(r)["apiName"]
-	jobID := mux.Vars(r)["jobID"]
+	jobID, err := getRequiredQueryParam("jobID", r)
+	if err != nil {
+		respondError(w, r, err)
+		return
+	}
 
 	deployedResource, err := resources.GetDeployedResourceByName(apiName)
 	if err != nil {
diff --git a/pkg/operator/endpoints/stop_job.go b/pkg/operator/endpoints/stop_job.go
@@ -29,9 +29,13 @@ import (
 func StopJob(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	apiName := vars["apiName"]
-	jobID := vars["jobID"]
+	jobID, err := getRequiredQueryParam("jobID", r)
+	if err != nil {
+		respondError(w, r, err)
+		return
+	}
 
-	err := batchapi.StopJob(spec.JobKey{APIName: apiName, ID: jobID})
+	err = batchapi.StopJob(spec.JobKey{APIName: apiName, ID: jobID})
 	if err != nil {
 		respondError(w, r, err)
 		return
diff --git a/pkg/operator/main.go b/pkg/operator/main.go
@@ -71,9 +71,8 @@ func main() {
 	routerWithoutAuth.Use(endpoints.PanicMiddleware)
 	routerWithoutAuth.HandleFunc("/verifycortex", endpoints.VerifyCortex).Methods("GET")
 	routerWithoutAuth.HandleFunc("/batch/{apiName}", endpoints.SubmitJob).Methods("POST")
-	routerWithoutAuth.HandleFunc("/batch/{apiName}/{jobID}", endpoints.GetJob).Methods("GET")
-	routerWithoutAuth.HandleFunc("/batch/{apiName}/{jobID}", endpoints.StopJob).Methods("DELETE")
-	routerWithoutAuth.HandleFunc("/logs/{apiName}/{jobID}", endpoints.ReadJobLogs)
+	routerWithoutAuth.HandleFunc("/batch/{apiName}", endpoints.GetJob).Methods("GET")
+	routerWithoutAuth.HandleFunc("/batch/{apiName}", endpoints.StopJob).Methods("DELETE")
 
 	routerWithAuth := router.NewRoute().Subrouter()
 
diff --git a/pkg/operator/operator/gateway.go b/pkg/operator/operator/gateway.go
diff --git a/pkg/operator/resources/batchapi/api.go b/pkg/operator/resources/batchapi/api.go
diff --git a/pkg/operator/resources/batchapi/enqueue.go b/pkg/operator/resources/batchapi/enqueue.go
diff --git a/pkg/operator/resources/batchapi/manage_resources_cron.go b/pkg/operator/resources/batchapi/manage_resources_cron.go
diff --git a/pkg/operator/resources/realtimeapi/api.go b/pkg/operator/resources/realtimeapi/api.go
diff --git a/pkg/operator/resources/trafficsplitter/api.go b/pkg/operator/resources/trafficsplitter/api.go

Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,7 @@ func StopJob(operatorConfig OperatorConfig, apiName string, jobID string) (schem`
`79`	`79`	`"jobID": jobID,`
`80`	`80`	`}`
`81`	`81`
`82`		`- httpRes, err := HTTPDelete(operatorConfig, path.Join("/batch", apiName, jobID), params)`
	`82`	`+ httpRes, err := HTTPDelete(operatorConfig, path.Join("/batch", apiName), params)`
`83`	`83`	`if err != nil {`
`84`	`84`	`return schema.DeleteResponse{}, err`
`85`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -52,8 +52,8 @@ func GetAPI(operatorConfig OperatorConfig, apiName string) ([]schema.APIResponse`
`52`	`52`	`}`
`53`	`53`
`54`	`54`	`func GetJob(operatorConfig OperatorConfig, apiName string, jobID string) (schema.JobResponse, error) {`
`55`		`- endpoint := path.Join("/batch", apiName, jobID)`
`56`		`- httpRes, err := HTTPGet(operatorConfig, endpoint)`
	`55`	`+ endpoint := path.Join("/batch", apiName)`
	`56`	`+ httpRes, err := HTTPGet(operatorConfig, endpoint, map[string]string{"jobID": jobID})`
`57`	`57`	`if err != nil {`
`58`	`58`	`return schema.JobResponse{}, err`
`59`	`59`	`}`
Original file line number	Diff line number	Diff line change
`@@ -52,7 +52,7 @@ func (oc OperatorConfig) AuthHeader() string {`
`52`	`52`	`}`
`53`	`53`
`54`	`54`	`func HTTPGet(operatorConfig OperatorConfig, endpoint string, qParams ...map[string]string) ([]byte, error) {`
`55`		`- req, err := operatorRequest(operatorConfig, "GET", endpoint, nil, qParams)`
	`55`	`+ req, err := operatorRequest(operatorConfig, "GET", endpoint, nil, qParams...)`
`56`	`56`	`if err != nil {`
`57`	`57`	`return nil, err`
`58`	`58`	`}`
`@@ -69,7 +69,7 @@ func HTTPPostObjAsJSON(operatorConfig OperatorConfig, endpoint string, requestDa`
`69`	`69`
`70`	`70`	`func HTTPPostJSON(operatorConfig OperatorConfig, endpoint string, jsonRequestData []byte, qParams ...map[string]string) ([]byte, error) {`
`71`	`71`	`payload := bytes.NewBuffer(jsonRequestData)`
`72`		`- req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, payload, qParams)`
	`72`	`+ req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, payload, qParams...)`
`73`	`73`	`if err != nil {`
`74`	`74`	`return nil, err`
`75`	`75`	`}`
`@@ -78,15 +78,15 @@ func HTTPPostJSON(operatorConfig OperatorConfig, endpoint string, jsonRequestDat`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`func HTTPPostNoBody(operatorConfig OperatorConfig, endpoint string, qParams ...map[string]string) ([]byte, error) {`
`81`		`- req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, nil, qParams)`
	`81`	`+ req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, nil, qParams...)`
`82`	`82`	`if err != nil {`
`83`	`83`	`return nil, err`
`84`	`84`	`}`
`85`	`85`	`return makeOperatorRequest(operatorConfig, req)`
`86`	`86`	`}`
`87`	`87`
`88`	`88`	`func HTTPDelete(operatorConfig OperatorConfig, endpoint string, qParams ...map[string]string) ([]byte, error) {`
`89`		`- req, err := operatorRequest(operatorConfig, http.MethodDelete, endpoint, nil, qParams)`
	`89`	`+ req, err := operatorRequest(operatorConfig, http.MethodDelete, endpoint, nil, qParams...)`
`90`	`90`	`if err != nil {`
`91`	`91`	`return nil, err`
`92`	`92`	`}`
`@@ -124,7 +124,7 @@ func HTTPUpload(operatorConfig OperatorConfig, endpoint string, input *HTTPUploa`
`124`	`124`	`return nil, errors.Wrap(err, _errStrCantMakeRequest)`
`125`	`125`	`}`
`126`	`126`
`127`		`- req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, body, qParams)`
	`127`	`+ req, err := operatorRequest(operatorConfig, http.MethodPost, endpoint, body, qParams...)`
`128`	`128`	`if err != nil {`
`129`	`129`	`return nil, err`
`130`	`130`	`}`
`@@ -159,7 +159,7 @@ func HTTPUploadZip(operatorConfig OperatorConfig, endpoint string, zipInput *arc`
`159`	`159`	`return HTTPUpload(operatorConfig, endpoint, uploadInput, qParams...)`
`160`	`160`	`}`
`161`	`161`
`162`		`-func operatorRequest(operatorConfig OperatorConfig, method string, endpoint string, body io.Reader, qParams []map[string]string) (*http.Request, error) {`
	`162`	`+func operatorRequest(operatorConfig OperatorConfig, method string, endpoint string, body io.Reader, qParams ...map[string]string) (*http.Request, error) {`
`163`	`163`	`req, err := http.NewRequest(method, operatorConfig.OperatorEndpoint+endpoint, body)`
`164`	`164`	`if err != nil {`
`165`	`165`	`return nil, errors.Wrap(err, _errStrCantMakeRequest)`