Skip to content

Commit

Permalink
jobs api
Browse files Browse the repository at this point in the history
Signed-off-by: Fernando Rocha <fernando@diagrid.io>
  • Loading branch information
rochabr committed Jul 24, 2024
1 parent 407cebd commit c9e0515
Show file tree
Hide file tree
Showing 19 changed files with 1,048 additions and 0 deletions.
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ Pick a building block API (for example, PubSub, state management, etc) and rapid
| [Cryptography](./cryptography) | Perform cryptographic operations without exposing keys to your application |
| [Resiliency](./resiliency) | Define and apply fault-tolerant policies (retries/back-offs, timeouts and circuit breakers) to your Dapr API requests |
| [Workflow](./workflows) | Dapr Workflow enables you to create long running, fault-tolerant, stateful applications |
| [Jobs](./jobs) | Dapr Jobs enable you to manage and schedule tasks |

### Tutorials

Expand Down
152 changes: 152 additions & 0 deletions jobs/go/http/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,152 @@
# Dapr Jobs

In this quickstart, you'll schedule, get, and delete a job using Dapr's Job API. This API is responsible for scheduling and running jobs at a specific time or interval.

Visit [this](https://v1-14.docs.dapr.io/developing-applications/building-blocks/jobs/) link for more information about Dapr and the Jobs API.

> **Note:** This example leverages HTTP `requests` only. If you are looking for the example using the Dapr Client SDK (recommended) [click here](../sdk/).
This quickstart includes two apps:

- `job-scheduler.go`, responsible for scheduling, retrieving and deleting jobs.
- `job-service.go`, responsible for handling the triggered jobs.

## Run the app with the template file

This section shows how to run both applications at once using [multi-app run template files](https://docs.dapr.io/developing-applications/local-development/multi-app-dapr-run/multi-app-overview/) with `dapr run -f .`. This enables to you test the interactions between multiple applications and will `schedule`, `run`, `get`, and `delete` jobs within a single process.

Open a new terminal window and run the multi app run template:

<!-- STEP
name: Run multi app run template
expected_stdout_lines:
- '== APP - job-service == Received job request...'
- '== APP - job-service == Executing maintenance job: Oil Change'
- '== APP - job-scheduler == Job Scheduled: C-3PO'
- '== APP - job-service == Received job request...'
- '== APP - job-service == Executing maintenance job: Limb Calibration'
expected_stderr_lines:
output_match_mode: substring
match_order: none
background: true
sleep: 60
timeout_seconds: 120
-->

```bash
dapr run -f .
```

The terminal console output should look similar to this, where:

- The `R2-D2` job is being scheduled.
- The `R2-D2` job is being executed after 2 seconds.
- The `C-3PO` job is being scheduled.
- The `C-3PO` job is being retrieved.

```text
== APP - job-scheduler == Job Scheduled: R2-D2
== APP - job-service == Received job request...
== APP - job-service == Starting droid: R2-D2
== APP - job-service == Executing maintenance job: Oil Change
== APP - job-scheduler == Job Scheduled: C-3PO
== APP - job-scheduler == Job details: {"name":"C-3PO", "dueTime":"30s", "data":{"@type":"ttype.googleapis.com/google.protobuf.StringValue", "expression":"C-3PO:Limb Calibration"}}
```

After 30 seconds, the terminal output should present the `C-3PO` job being processed:

```text
== APP - job-service == Received job request...
== APP - job-service == Starting droid: C-3PO
== APP - job-service == Executing maintenance job: Limb Calibration
```

2. Stop and clean up application processes

```bash
dapr stop -f .
```

<!-- END_STEP -->

## Run the Jobs APIs individually

### Schedule Jobs

1. Open a terminal and run the `job-service` app:

```bash
dapr run --app-id job-service --app-port 6200 --dapr-http-port 6280 -- go run .
```

2. On a new terminal window, schedule the `R2-D2` Job using the Jobs API.

```bash
curl -X POST \
http://localhost:6280/v1.0-alpha1/jobs/jobforjabba \
-H "Content-Type: application/json" \
-d '{
"data": {
"@type": "type.googleapis.com/google.protobuf.StringValue",
"value": "R2-D2:Oil Change"
},
"dueTime": "2s"
}'
```

Back at the `job-service` app terminal window, the output should be:

```text
== APP - job-app == Received job request...
== APP - job-app == Starting droid: R2-D2
== APP - job-app == Executing maintenance job: Oil Change
```

3. On the same terminal window, schedule the `C-3PO` Job using the Jobs API.

```bash
curl -X POST \
http://localhost:6280/v1.0-alpha1/jobs/c-3po \
-H "Content-Type: application/json" \
-d '{
"data": {
"@type": "type.googleapis.com/google.protobuf.StringValue",
"value": "C-3PO:Limb Calibration"
},
"dueTime": "30s"
}'
```

### Get a scheduled job

1. On the same terminal window, run the command below to get the recently scheduled `C-3PO` job.

```bash
curl -X GET http://localhost:6280/v1.0-alpha1/jobs/c-3po -H "Content-Type: application/json"
```

You should see the following:

```text
{"name":"C-3PO", "dueTime":"30s", "data":{"@type":"type.googleapis.com/google.protobuf.StringValue", "expression":"C-3PO:Limb Calibration"}}
```

### Delete a scheduled job

1. On the same terminal window, run the command below to deleted the recently scheduled `C-3PO` job.

```bash
curl -X DELETE http://localhost:6280/v1.0-alpha1/jobs/c-3po -H "Content-Type: application/json"
```

2. Run the command below to attempt to retrieve the deleted job:

```bash
curl -X GET http://localhost:6280/v1.0-alpha1/jobs/c-3po -H "Content-Type: application/json"
```

Back at the `job-service` app terminal window, the output should be:

```text
ERRO[0249] Error getting job c-3po due to: rpc error: code = Unknown desc = job not found: app||default||job-service||c-3po instance=diagrid.local scope=dapr.api type=log ver=1.14.0-rc.2
```
12 changes: 12 additions & 0 deletions jobs/go/http/dapr.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
version: 1
apps:
- appDirPath: ./job-service/
appID: job-service
appPort: 6200
daprHTTPPort: 6280
command: ["go", "run", "."]
- appDirPath: ./job-scheduler/
appID: job-scheduler
appPort: 6300
daprHTTPPort: 6380
command: ["go", "run", "."]
3 changes: 3 additions & 0 deletions jobs/go/http/job-scheduler/go.mod
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
module job-scheduler

go 1.21
Empty file.
113 changes: 113 additions & 0 deletions jobs/go/http/job-scheduler/job-scheduler.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
package main

import (
"fmt"
"io"
"log"
"net/http"
"os"
"strings"
"time"
)

var c3poJobBody = `{
"data": {
"@type": "type.googleapis.com/google.protobuf.StringValue",
"value": "C-3PO:Limb Calibration"
},
"dueTime": "30s"
}`

var r2d2JobBody = `{
"data": {
"@type": "type.googleapis.com/google.protobuf.StringValue",
"value": "R2-D2:Oil Change"
},
"dueTime": "2s"
}`

func main() {
//Sleep for 5 seconds to wait for job-service to start
time.Sleep(5 * time.Second)

daprHost := os.Getenv("DAPR_HOST")
if daprHost == "" {
daprHost = "http://localhost"
}

schedulerDaprHttpPort := "6280"

client := http.Client{
Timeout: 15 * time.Second,
}

// Schedule a job using the Dapr Jobs API with short dueTime
jobName := "R2-D2"
reqURL := daprHost + ":" + schedulerDaprHttpPort + "/v1.0-alpha1/jobs/" + jobName

req, err := http.NewRequest("POST", reqURL, strings.NewReader(r2d2JobBody))
if err != nil {
log.Fatal(err.Error())
}

req.Header.Set("Content-Type", "application/json")

// Schedule a job using the Dapr Jobs API
res, err := client.Do(req)
if err != nil {
log.Fatal(err)
}

if res.StatusCode != http.StatusNoContent {
log.Fatalf("failed to register job event handler. status code: %v", res.StatusCode)
}

defer res.Body.Close()

fmt.Println("Job Scheduled:", jobName)

time.Sleep(5 * time.Second)

// Schedule a job using the Dapr Jobs API with long dueTime
jobName = "C-3PO"

reqURL = daprHost + ":" + schedulerDaprHttpPort + "/v1.0-alpha1/jobs/" + jobName

req, err = http.NewRequest("POST", reqURL, strings.NewReader(c3poJobBody))
if err != nil {
log.Fatal(err.Error())
}

req.Header.Set("Content-Type", "application/json")

// Schedule a job using the Dapr Jobs API
res, err = client.Do(req)
if err != nil {
log.Fatal(err)
}
defer res.Body.Close()

fmt.Println("Job Scheduled:", jobName)

time.Sleep(5 * time.Second)

// Gets a job using the Dapr Jobs API
jobName = "C-3PO"
reqURL = daprHost + ":" + schedulerDaprHttpPort + "/v1.0-alpha1/jobs/" + jobName

res, err = http.Get(reqURL)
if err != nil {
log.Fatal(err.Error())
}
defer res.Body.Close()

resBody, err := io.ReadAll(res.Body)
if err != nil {
log.Fatal(err.Error())

}

fmt.Println("Job details:", string(resBody))

time.Sleep(5 * time.Second)
}
3 changes: 3 additions & 0 deletions jobs/go/http/job-service/go.mod
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
module job-service

go 1.21
Empty file added jobs/go/http/job-service/go.sum
Empty file.
92 changes: 92 additions & 0 deletions jobs/go/http/job-service/job-service.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
/*
Copyright 2021 The Dapr Authors
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package main

import (
"encoding/base64"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"strings"
)

type Job struct {
TypeURL string `json:"type_url"`
Value string `json:"value"`
}

type DroidJob struct {
Droid string `json:"droid"`
Task string `json:"task"`
}

func main() {
appPort := os.Getenv("APP_PORT")
if appPort == "" {
appPort = "6200"
}

// Setup job handler
http.HandleFunc("/job/", handleJob)

fmt.Printf("Server started on port %v\n", appPort)
err := http.ListenAndServe(":"+appPort, nil)
if err != nil {
log.Fatal(err)
}

}

func handleJob(w http.ResponseWriter, r *http.Request) {
fmt.Println("Received job request...")
rawBody, err := io.ReadAll(r.Body)
if err != nil {
http.Error(w, fmt.Sprintf("error reading request body: %v", err), http.StatusBadRequest)
return
}

var jobData Job
if err := json.Unmarshal(rawBody, &jobData); err != nil {
http.Error(w, fmt.Sprintf("error decoding JSON: %v", err), http.StatusBadRequest)
return
}

// Decoding job data
decodedValue, err := base64.RawStdEncoding.DecodeString(jobData.Value)
if err != nil {
fmt.Printf("Error decoding base64: %v", err)
http.Error(w, fmt.Sprintf("error decoding base64: %v", err), http.StatusBadRequest)
return
}

// Creating Droid Job from decoded value
droidJob := setDroidJob(string(decodedValue))

fmt.Println("Starting droid:", droidJob.Droid)
fmt.Println("Executing maintenance job:", droidJob.Task)

w.WriteHeader(http.StatusOK)
}

func setDroidJob(decodedValue string) DroidJob {
// Removing new lines from decoded value - Workaround for base64 encoding issue
droidStr := strings.ReplaceAll(decodedValue, "\n", "")
droidArray := strings.Split(droidStr, ":")

droidJob := DroidJob{Droid: droidArray[0], Task: droidArray[1]}
return droidJob
}
2 changes: 2 additions & 0 deletions jobs/go/http/makefile
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
include ../../../docker.mk
include ../../../validate.mk
Loading

0 comments on commit c9e0515

Please sign in to comment.