Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jobs API Quickstarts #1064

Merged
merged 10 commits into from
Jul 25, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/env/global.env
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
DAPR_CLI_VERSION: 1.13.0
DAPR_RUNTIME_VERSION: 1.13.5
DAPR_CLI_VERSION: 1.14.0-rc.7
DAPR_RUNTIME_VERSION: 1.14.0-rc.5
rochabr marked this conversation as resolved.
Show resolved Hide resolved
DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v${DAPR_CLI_VERSION}/install/
DAPR_DEFAULT_IMAGE_REGISTRY: ghcr
MACOS_PYTHON_VERSION: 3.10
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/R2D2 \
-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
rochabr marked this conversation as resolved.
Show resolved Hide resolved
== 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
rochabr marked this conversation as resolved.
Show resolved Hide resolved
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)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

consider removing these. but for now if stable leave them. I think it confuses to have sleeps here in code, also in MMD readme.


// 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.
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
}
Loading
Loading