Skip to content

Commit bb3368d

Browse files
02strich02strich
committed
Add Nexus-based compute providers for worker invocation and scaling
Introduce a Nexus RPC compute providers for invoke and worker-set compute providers. As the Nexus-based compute providers are called directly from the workflow, adding a new code path to the Workflow to do that - and fallback to activity when another compute provider is used. Also backfilling the so far missing field for the Nexus endpoint - which already existed in the frontend API. Add a cmd/nexus-subprocess-example binary demonstrating a subprocess-backed Nexus invoke service, along with Makefile targets and .gitignore entries for the generated files and new binary.
1 parent 4320b34 commit bb3368d

31 files changed

Lines changed: 2809 additions & 132 deletions

‎.github/workflows/ci.yaml‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,18 @@ jobs:
6363
with:
6464
go-version-file: "go.mod"
6565

66+
- name: Setup Node
67+
uses: actions/setup-node@v6
68+
with:
69+
node-version: "lts/*"
70+
71+
- name: Install code generation tools
72+
run: |
73+
sudo apt-get update
74+
sudo apt-get install -y protobuf-compiler
75+
go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.11
76+
echo "$(go env GOPATH)/bin" >> "$GITHUB_PATH"
77+
6678
- name: Run Unit Tests
6779
env:
6880
UNIT_TEST_FLAGS: -v -cover
@@ -79,6 +91,21 @@ jobs:
7991
with:
8092
go-version-file: "tests/go.mod"
8193

94+
- name: Setup Node
95+
uses: actions/setup-node@v6
96+
with:
97+
node-version: "lts/*"
98+
99+
- name: Install code generation tools
100+
run: |
101+
sudo apt-get update
102+
sudo apt-get install -y protobuf-compiler
103+
go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.11
104+
echo "$(go env GOPATH)/bin" >> "$GITHUB_PATH"
105+
106+
- name: Generate Code
107+
run: make generate
108+
82109
- name: Run Integration Tests
83110
working-directory: tests
84111
run: go test -tags=test_dep -timeout 300s ./integration/...
@@ -95,6 +122,21 @@ jobs:
95122
with:
96123
go-version-file: "tests/go.mod"
97124

125+
- name: Setup Node
126+
uses: actions/setup-node@v6
127+
with:
128+
node-version: "lts/*"
129+
130+
- name: Install code generation tools
131+
run: |
132+
sudo apt-get update
133+
sudo apt-get install -y protobuf-compiler
134+
go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.11
135+
echo "$(go env GOPATH)/bin" >> "$GITHUB_PATH"
136+
137+
- name: Generate Code
138+
run: make generate
139+
98140
- name: Update server to main
99141
working-directory: tests
100142
run: |

‎.github/workflows/golangci-lint.yml‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,22 @@ jobs:
1616
- uses: actions/setup-go@v6
1717
with:
1818
go-version: stable
19+
20+
- name: Setup Node
21+
uses: actions/setup-node@v6
22+
with:
23+
node-version: "lts/*"
24+
25+
- name: Install code generation tools
26+
run: |
27+
sudo apt-get update
28+
sudo apt-get install -y protobuf-compiler
29+
go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.11
30+
echo "$(go env GOPATH)/bin" >> "$GITHUB_PATH"
31+
32+
- name: Generate Code
33+
run: make generate
34+
1935
- name: golangci-lint
2036
uses: golangci/golangci-lint-action@v9
2137
with:

‎.gitignore‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,15 @@ profile.cov
2525
go.work
2626
go.work.sum
2727

28+
# Nexus RPC generated files
29+
/wci/workflow/compute_provider/nexus/nexus.pb.go
30+
/wci/workflow/compute_provider/nexus/nexus_invoke_nexusserver_gen.go
31+
/wci/workflow/compute_provider/nexus/nexus_invoke_nexusrpc_gen.go
32+
/wci/workflow/compute_provider/nexus/nexus_invoke.pb.go
33+
/wci/workflow/compute_provider/nexus/nexus_worker_set_nexusserver_gen.go
34+
/wci/workflow/compute_provider/nexus/nexus_worker_set_nexusrpc_gen.go
35+
/wci/workflow/compute_provider/nexus/nexus_worker_set.pb.go
36+
2837
# env file
2938
.env
3039

@@ -34,3 +43,4 @@ go.work.sum
3443

3544
# Executables produced by temporal repo
3645
/temporal-auto-scaled-workers
46+
/nexus-subprocess-example

‎Makefile‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
############################# Main targets #############################
2-
# Rebuild binaries (used by Dockerfile).
3-
bins: temporal-auto-scaled-workers
2+
# Rebuild binaries.
3+
bins: temporal-auto-scaled-workers nexus-subprocess-example
44

55
# Install all tools, run all possible checks and tests (long but comprehensive).
66
all: clean bins test
@@ -9,7 +9,7 @@ all: clean bins test
99
clean: clean-bins clean-test-output
1010
########################################################################
1111

12-
.PHONY: bins clean
12+
.PHONY: bins clean generate
1313

1414
##### Variables ######
1515

@@ -21,31 +21,43 @@ ALL_SRC += go.mod
2121
UNIT_TEST_DIRS ?= ./...
2222
LINT_DIRS ?= ./...
2323
MAIN_BRANCH ?= main
24+
NEXUS_EXAMPLE_ADDRESS ?= localhost:7233
25+
NEXUS_EXAMPLE_NAMESPACE ?= default
26+
NEXUS_EXAMPLE_TASK_QUEUE ?= nexus-subprocess-example
2427

2528
##### Binaries #####
2629
clean-bins:
2730
@printf $(COLOR) "Delete old binaries..."
28-
@rm -f temporal-auto-scaled-workers
31+
@rm -f temporal-auto-scaled-workers nexus-subprocess-example
2932

30-
temporal-auto-scaled-workers: $(ALL_SRC)
33+
temporal-auto-scaled-workers: generate $(ALL_SRC)
3134
@printf $(COLOR) "Build temporal-auto-scaled-workers with CGO_ENABLED=$(CGO_ENABLED) for $(GOOS)/$(GOARCH)..."
3235
CGO_ENABLED=$(CGO_ENABLED) go build $(BUILD_TAG_FLAG) -o temporal-auto-scaled-workers ./cmd/worker
3336

37+
nexus-subprocess-example: generate $(ALL_SRC)
38+
@printf $(COLOR) "Build nexus-subprocess-example with CGO_ENABLED=$(CGO_ENABLED) for $(GOOS)/$(GOARCH)..."
39+
CGO_ENABLED=$(CGO_ENABLED) go build $(BUILD_TAG_FLAG) -o nexus-subprocess-example ./cmd/nexus-subprocess-example
40+
41+
##### Code generation #####
42+
generate:
43+
@printf $(COLOR) "Run go generate..."
44+
@go generate ./...
45+
3446
##### Tests #####
3547
clean-test-output:
3648
@printf $(COLOR) "Delete test output..."
3749
@rm -rf $(TEST_OUTPUT_ROOT)
3850
@go clean -testcache
3951

40-
unit-test: clean-test-output
52+
unit-test: generate clean-test-output
4153
@printf $(COLOR) "Run unit tests..."
4254
@CGO_ENABLED=$(CGO_ENABLED) go test $(UNIT_TEST_FLAGS) $(UNIT_TEST_DIRS) $(COMPILED_TEST_ARGS) 2>&1 | tee -a test.log
4355
@! grep -q "^--- FAIL" test.log
4456

4557
test: unit-test
4658

4759
##### Linting / formatting #####
48-
lint:
60+
lint: generate
4961
@printf $(COLOR) "Run golangci-lint..."
5062
@golangci-lint run $(LINT_DIRS)
5163

@@ -54,7 +66,7 @@ lint-branch:
5466
@printf $(COLOR) "Run golangci-lint on changes since merge-base with $(MAIN_BRANCH)..."
5567
@golangci-lint run --new-from-rev=$$(git merge-base HEAD $(MAIN_BRANCH)) $(LINT_DIRS)
5668

57-
fmt:
69+
fmt: generate
5870
@printf $(COLOR) "Format with golangci-lint..."
5971
@golangci-lint fmt $(LINT_DIRS)
6072

@@ -74,3 +86,6 @@ start-sqlite: temporal-auto-scaled-workers
7486

7587
start-sqlite-file: temporal-auto-scaled-workers
7688
./temporal-auto-scaled-workers --config-file config/development-sqlite-file.yaml start
89+
90+
start-nexus-subprocess-example: nexus-subprocess-example
91+
./nexus-subprocess-example --address $(NEXUS_EXAMPLE_ADDRESS) --namespace $(NEXUS_EXAMPLE_NAMESPACE) --task-queue $(NEXUS_EXAMPLE_TASK_QUEUE)

‎README.md‎

Lines changed: 112 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,32 @@ Multiple **scaling groups** can be defined per WCI, each mapping a set of task q
2222
| GCP Cloud Run | `gcp-cloud-run` | Worker set |
2323
| Kubernetes | `k8s` | Worker set |
2424
| Subprocess | `subprocess` | Invoke (dev/test only) |
25+
| Nexus invoke | `nexus-invoke` | Invoke (external Nexus service) |
26+
| Nexus worker set | `nexus-worker-set` | Worker set (external Nexus service) |
2527

2628
**Invoke** providers are called once per scaling event to start a short-lived worker.
2729
**Worker-set** providers manage a persistent pool whose size is adjusted up or down.
2830

31+
### Nexus compute providers
32+
33+
Nexus compute providers let WCI call an implementation hosted outside the WCI worker. Unlike native providers, Nexus providers are invoked directly from workflow code through Temporal Nexus APIs, without scheduling the activity-based native provider path.
34+
35+
For Nexus providers, `compute.nexus_endpoint` names the Temporal Nexus endpoint to call and `compute.config` is forwarded to the Nexus service as the provider config payload.
36+
37+
Invoke-style Nexus providers use the `InvokeComputeProvider` service:
38+
39+
| Operation | Purpose |
40+
|---|---|
41+
| `validate_config` | Validate the provider config payload |
42+
| `invoke_worker` | Start one worker instance |
43+
44+
Worker-set Nexus providers use the `WorkerSetComputeProvider` service:
45+
46+
| Operation | Purpose |
47+
|---|---|
48+
| `validate_config` | Validate the provider config payload |
49+
| `update_worker_set_size` | Resize the managed worker set |
50+
2951
## Supported Scaling Algorithms
3052

3153
| Algorithm | Type string | Description |
@@ -90,6 +112,62 @@ A WCI spec is a map of named scaling groups:
90112

91113
A group with no `task_types` acts as a catch-all for any task type not claimed by another group. At most one catch-all group is allowed. The `scaling` block is optional; omitting it leaves the group with the default scaling configuration for the given compute provider.
92114

115+
### Nexus invoke spec example
116+
117+
This example routes activity scaling events to a Nexus endpoint named `subprocess-compute`. The endpoint must be configured in Temporal to route to the task queue served by the example server below. The JSON below shows the decoded config values for readability; in the Go API, `compute.config` is a `*commonpb.Payload`.
118+
119+
```json
120+
{
121+
"scaling_group_specs": {
122+
"activities": {
123+
"task_types": ["ACTIVITY"],
124+
"compute": {
125+
"provider_type": "nexus-invoke",
126+
"nexus_endpoint": "subprocess-compute",
127+
"config": {
128+
"command": "my-worker",
129+
"args": "--namespace,default,--task-queue,my-task-queue"
130+
}
131+
},
132+
"scaling": {
133+
"scaling_algorithm": "no-sync",
134+
"config": {
135+
"scale_up_backlog_threshold": "5"
136+
}
137+
}
138+
}
139+
}
140+
}
141+
```
142+
143+
When constructing the spec in Go, encode the provider config before assigning it. Note that `PreferProtoDataConverter` comes from `go.temporal.io/server/common/sdk`, not `go.temporal.io/sdk`:
144+
145+
```go
146+
import (
147+
enumspb "go.temporal.io/api/enums/v1"
148+
computeprovider "go.temporal.io/auto-scaled-workers/wci/workflow/compute_provider"
149+
"go.temporal.io/auto-scaled-workers/wci/workflow/iface"
150+
"go.temporal.io/server/common/sdk"
151+
)
152+
153+
computeConfig, err := sdk.PreferProtoDataConverter.ToPayload(computeprovider.ComputeProviderConfig{
154+
"command": "my-worker",
155+
"args": "--namespace,default,--task-queue,my-task-queue",
156+
})
157+
if err != nil {
158+
return err
159+
}
160+
161+
spec := iface.ScalingGroupSpec{
162+
TaskTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY},
163+
Compute: iface.ComputeProviderSpec{
164+
ProviderType: iface.ComputeProviderTypeNexusInvoke,
165+
NexusEndpoint: "subprocess-compute",
166+
Config: computeConfig,
167+
},
168+
}
169+
```
170+
93171
### `no-sync` algorithm config
94172

95173
| Key | Default | Description |
@@ -145,13 +223,46 @@ Enable per namespace via the `WorkerControllerEnabled` dynamic config setting.
145223
## Building
146224

147225
```bash
148-
# Build
226+
# Build the worker and example binaries
149227
make bins
150228

229+
# Build only the local Nexus subprocess example server
230+
make nexus-subprocess-example
231+
151232
# Run tests
152233
make test
153234
```
154235

236+
## Nexus Subprocess Example Server
237+
238+
`cmd/nexus-subprocess-example` is a local-development Nexus service implementation backed by the `subprocess` compute provider. It registers the generated `InvokeComputeProvider` Nexus service with a Temporal SDK worker and delegates `validate_config` and `invoke_worker` to the subprocess provider.
239+
240+
Build and run it with:
241+
242+
```bash
243+
make nexus-subprocess-example
244+
make start-nexus-subprocess-example
245+
```
246+
247+
The run target uses these overridable variables:
248+
249+
| Variable | Default |
250+
|---|---|
251+
| `NEXUS_EXAMPLE_ADDRESS` | `localhost:7233` |
252+
| `NEXUS_EXAMPLE_NAMESPACE` | `default` |
253+
| `NEXUS_EXAMPLE_TASK_QUEUE` | `nexus-subprocess-example` |
254+
255+
Example:
256+
257+
```bash
258+
make start-nexus-subprocess-example \
259+
NEXUS_EXAMPLE_ADDRESS=localhost:7233 \
260+
NEXUS_EXAMPLE_NAMESPACE=default \
261+
NEXUS_EXAMPLE_TASK_QUEUE=nexus-subprocess-example
262+
```
263+
264+
The example is a local-development helper and is not intended to ship to production. It requires GNU `timeout` and the configured subprocess `command` to be available on `PATH`. To call it from WCI, configure a Temporal Nexus endpoint whose name matches `compute.nexus_endpoint` and whose target task queue matches `NEXUS_EXAMPLE_TASK_QUEUE`.
265+
155266
## Run Together with a Local Temporal Server
156267

157268
1. Check out the [Temporal Server](https://github.com/temporalio/temporal) alongside this repository.

0 commit comments

Comments
 (0)