Skip to content

Commit 10fb149

Browse files
authored
Support ECS launcher (#178)
* Fix pool issues * Add initial ECS support * Support configuring pool idle timeout * Support configuring idle timeout * Store tokens in admin database * Support managing secrets * Fix workspace/scope matching * Fix flaky tests * Tidy/fix definition of scopes * Rename field for consistency * Validate merged pool * Support configuring ECS pool with role * Have workers/pools host all modules by default * Tidy duration field naming/conventions * Clean up after tests * Improve API error messages * Capture logs from task * Fix idle timeout units conversion * Send initial stop signal to pool workers through connection * Format * Fix test
1 parent 4404f67 commit 10fb149

108 files changed

Lines changed: 8213 additions & 1556 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎adapters/python/CHANGELOG.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
## 0.13.0
22

3-
No changes.
3+
Changes:
4+
5+
- The manifest and submission payloads name durations with their unit: `delay` and `timeout` are `delay_ms` and `timeout_ms`, matching `max_age_ms` and `backoff_min_ms`/`backoff_max_ms`, which already did. Decorator arguments are unchanged — `delay`, `timeout` and `cache` still take seconds or a `timedelta`.
46

57
## 0.12.0
68

‎adapters/python/coflux/__main__.py‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,11 @@ def main() -> int:
2626
)
2727
discover_parser.add_argument(
2828
"modules",
29-
nargs="+",
30-
help="Python modules to scan for targets",
29+
nargs="*",
30+
help=(
31+
"Python modules to scan for targets; packages are scanned recursively. "
32+
"With none, every module in the working directory is scanned."
33+
),
3134
)
3235

3336
# execute command

‎adapters/python/coflux/context.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -241,11 +241,11 @@ def submit_execution(
241241
cache: dict[str, Any] | None = None,
242242
defer: dict[str, Any] | None = None,
243243
memo: bool | list[int] | None = None,
244-
delay: float | None = None,
244+
delay_ms: int | None = None,
245245
retries: dict[str, Any] | None = None,
246246
recurrent: bool = False,
247247
requires: dict[str, list[str]] | None = None,
248-
timeout: int = 0,
248+
timeout_ms: int = 0,
249249
streams: dict[str, Any] | None = None,
250250
concurrency: dict[str, Any] | None = None,
251251
) -> dict[str, Any]:
@@ -267,11 +267,11 @@ def submit_execution(
267267
cache=cache,
268268
defer=defer,
269269
memo=memo,
270-
delay=delay,
270+
delay_ms=delay_ms,
271271
retries=retries,
272272
recurrent=recurrent,
273273
requires=requires,
274-
timeout=timeout,
274+
timeout_ms=timeout_ms,
275275
streams=streams,
276276
concurrency=concurrency,
277277
)

‎adapters/python/coflux/discovery.py‎

Lines changed: 35 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import dataclasses
66
import importlib
77
import json
8+
import os
89
import pkgutil
910
import sys
1011
import traceback
@@ -20,14 +21,41 @@
2021
serialize_streams,
2122
)
2223

24+
# Top-level names that are never where targets live, and that tend to have
25+
# import-time side effects or dependencies the worker doesn't have.
26+
_SKIPPED_TOP_LEVEL = frozenset({"setup", "conftest", "tests", "test"})
27+
28+
29+
def _working_directory_modules() -> list[str]:
30+
"""The top-level modules and packages in the working directory.
31+
32+
This is what a worker hosts when it's given no modules: the directory
33+
it was started in, which ``python -m`` puts on ``sys.path``. Private
34+
(``_``-prefixed) names and a few conventional non-targets are left out,
35+
and packages are only found by their ``__init__.py``, as elsewhere.
36+
"""
37+
cwd = os.getcwd()
38+
if cwd not in sys.path and "" not in sys.path:
39+
sys.path.insert(0, cwd)
40+
41+
return sorted(
42+
info.name
43+
for info in pkgutil.iter_modules([cwd])
44+
if not info.name.startswith("_") and info.name not in _SKIPPED_TOP_LEVEL
45+
)
46+
2347

2448
def _expand_modules(module_names: list[str]) -> list[str]:
2549
"""Expand package names into their constituent submodules.
2650
2751
If a name refers to a Python package (has __path__), it is recursively
2852
walked using pkgutil.walk_packages. Plain module names are passed through
2953
unchanged. Submodules whose final component starts with '_' are skipped.
54+
No names at all means every module in the working directory.
3055
"""
56+
if not module_names:
57+
module_names = _working_directory_modules()
58+
3159
result: list[str] = []
3260
seen: set[str] = set()
3361

@@ -75,11 +103,13 @@ def discover_targets(
75103
"""Discover all targets in the specified modules.
76104
77105
If a module name refers to a package, all submodules are scanned
78-
recursively (private submodules starting with '_' are skipped).
106+
recursively (private submodules starting with '_' are skipped). An
107+
empty list scans every module in the working directory.
79108
80109
Args:
81110
modules: List of module or package names to scan
82-
(e.g., ["myapp.workflows", "myapp.tasks"] or just ["myapp"])
111+
(e.g., ["myapp.workflows", "myapp.tasks"] or just ["myapp"]),
112+
or [] for the working directory
83113
84114
Returns:
85115
A ``(targets, errors)`` pair — target definitions suitable for JSON
@@ -150,7 +180,7 @@ def _build_target_definition(target: Any, module_name: str) -> dict[str, Any]:
150180
result["defer"] = serialize_defer(definition.defer, definition.parameters)
151181

152182
if definition.delay:
153-
result["delay"] = _to_ms(definition.delay)
183+
result["delay_ms"] = _to_ms(definition.delay)
154184

155185
if definition.memo:
156186
result["memo"] = definition.memo
@@ -159,7 +189,7 @@ def _build_target_definition(target: Any, module_name: str) -> dict[str, Any]:
159189
result["requires"] = definition.requires
160190

161191
if definition.timeout:
162-
result["timeout"] = _to_ms(definition.timeout)
192+
result["timeout_ms"] = _to_ms(definition.timeout)
163193

164194
if definition.recurrent:
165195
result["recurrent"] = True
@@ -192,7 +222,7 @@ def run_discovery(modules: list[str]) -> int:
192222
file would mean restarting by hand — has it.
193223
194224
Args:
195-
modules: List of module names to scan.
225+
modules: List of module names to scan, or [] for the working directory.
196226
197227
Returns:
198228
Exit code (0 for success, 1 for error).

‎adapters/python/coflux/protocol.py‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -193,11 +193,11 @@ def request_submit_execution(
193193
cache: dict[str, Any] | None = None,
194194
defer: dict[str, Any] | None = None,
195195
memo: bool | list[int] | None = None,
196-
delay: float | None = None,
196+
delay_ms: int | None = None,
197197
retries: dict[str, Any] | None = None,
198198
recurrent: bool = False,
199199
requires: dict[str, list[str]] | None = None,
200-
timeout: int = 0,
200+
timeout_ms: int = 0,
201201
streams: dict[str, Any] | None = None,
202202
concurrency: dict[str, Any] | None = None,
203203
) -> int:
@@ -220,16 +220,16 @@ def request_submit_execution(
220220
params["defer"] = defer
221221
if memo is not None:
222222
params["memo"] = memo
223-
if delay is not None:
224-
params["delay"] = delay
223+
if delay_ms is not None:
224+
params["delay_ms"] = delay_ms
225225
if retries is not None:
226226
params["retries"] = retries
227227
if recurrent:
228228
params["recurrent"] = recurrent
229229
if requires:
230230
params["requires"] = requires
231-
if timeout:
232-
params["timeout"] = timeout
231+
if timeout_ms:
232+
params["timeout_ms"] = timeout_ms
233233
if streams is not None:
234234
params["streams"] = streams
235235
if concurrency is not None:

‎adapters/python/coflux/target.py‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -714,11 +714,13 @@ def submit(self, *args: P.args, **kwargs: P.kwargs) -> Execution[T]:
714714
cache=cache_dict,
715715
defer=defer_dict,
716716
memo=memo_val,
717-
delay=_to_ms(self._definition.delay) if self._definition.delay else None,
717+
delay_ms=_to_ms(self._definition.delay) if self._definition.delay else None,
718718
retries=retries_dict,
719719
recurrent=self._definition.recurrent,
720720
requires=self._definition.requires,
721-
timeout=_to_ms(self._definition.timeout) if self._definition.timeout else 0,
721+
timeout_ms=_to_ms(self._definition.timeout)
722+
if self._definition.timeout
723+
else 0,
722724
streams=streams_dict,
723725
concurrency=concurrency_dict,
724726
)
Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
"""What discovery scans when it's given a package, or nothing at all.
2+
3+
A package name covers everything under it. No names means the working
4+
directory: its top-level modules and packages, less the private ones and
5+
the conventional non-targets that tend to blow up on import.
6+
"""
7+
8+
from __future__ import annotations
9+
10+
import importlib
11+
import sys
12+
import textwrap
13+
14+
import pytest
15+
16+
from coflux.discovery import discover_targets
17+
18+
_TASK = textwrap.dedent(
19+
"""
20+
import coflux as cf
21+
22+
@cf.task()
23+
def {name}():
24+
return 1
25+
"""
26+
)
27+
28+
29+
@pytest.fixture
30+
def project(tmp_path, monkeypatch):
31+
"""A throwaway project as the working directory, cleaned out of
32+
``sys.modules`` afterwards so its names don't leak between tests."""
33+
monkeypatch.chdir(tmp_path)
34+
monkeypatch.syspath_prepend(str(tmp_path))
35+
importlib.invalidate_caches()
36+
before = set(sys.modules)
37+
38+
def write(path, source):
39+
file = tmp_path / path
40+
file.parent.mkdir(parents=True, exist_ok=True)
41+
file.write_text(source)
42+
43+
yield write
44+
45+
for name in set(sys.modules) - before:
46+
sys.modules.pop(name, None)
47+
48+
49+
def _names(targets):
50+
return sorted((t["module"], t["name"]) for t in targets)
51+
52+
53+
def test_a_package_covers_its_submodules(project):
54+
project("dpkg/__init__.py", "")
55+
project("dpkg/flows.py", _TASK.format(name="flow"))
56+
project("dpkg/deep/__init__.py", "")
57+
project("dpkg/deep/jobs.py", _TASK.format(name="job"))
58+
project("dpkg/_private.py", _TASK.format(name="hidden"))
59+
60+
targets, errors = discover_targets(["dpkg"])
61+
62+
assert errors == []
63+
assert _names(targets) == [("dpkg.deep.jobs", "job"), ("dpkg.flows", "flow")]
64+
65+
66+
def test_no_modules_scans_the_working_directory(project):
67+
project("dwd_app/__init__.py", "")
68+
project("dwd_app/flows.py", _TASK.format(name="flow"))
69+
project("dwd_scratch.py", _TASK.format(name="scratch"))
70+
project("_dwd_hidden.py", _TASK.format(name="hidden"))
71+
# Skipped by name, so importing them is never attempted
72+
project("conftest.py", "raise RuntimeError('conftest imported')")
73+
project("setup.py", "raise RuntimeError('setup imported')")
74+
project("tests/__init__.py", "raise RuntimeError('tests imported')")
75+
# Not a package without __init__.py, and not a module at all
76+
project("dwd_data/flows.py", _TASK.format(name="orphan"))
77+
project("notes.txt", "")
78+
79+
targets, errors = discover_targets([])
80+
81+
assert errors == []
82+
assert _names(targets) == [("dwd_app.flows", "flow"), ("dwd_scratch", "scratch")]
83+
84+
85+
def test_an_empty_working_directory_finds_nothing(project):
86+
project("notes.txt", "")
87+
88+
assert discover_targets([]) == ([], [])
89+
90+
91+
def test_a_broken_module_in_the_working_directory_is_reported(project):
92+
project("dwd_ok.py", _TASK.format(name="ok"))
93+
project("dwd_broken.py", "import does_not_exist")
94+
95+
targets, errors = discover_targets([])
96+
97+
assert _names(targets) == [("dwd_ok", "ok")]
98+
assert [e.module for e in errors] == ["dwd_broken"]

‎cli/CHANGELOG.md‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,16 @@
11
## 0.13.0
22

3-
No changes.
3+
Enhancements:
4+
5+
- Adds `--type ecs` support for `pools create` and `pools update`, including the `roleArn` and `roleExternalId` fields for a role to assume before calling ECS.
6+
- Adds the `idleTimeout` pool field, for how long a pool keeps an idle worker before stopping it (`--set idleTimeout=5m`, exported as `idle_timeout = "5m"`).
7+
- Adds `secrets set`, `secrets list` and `secrets delete`. Pools refer to secrets by name (`tokenSecret`, `credentialsSecret`, `envSecrets`) instead of holding credentials, so `pools export` no longer needs `--include-secrets`. Each secret is set for one or more workspace patterns, given as a required `--workspaces`, in the same language `tokens create --workspaces` uses.
8+
- The worker exits when the server asks it to (a `stop` command over its connection, sent to a pool worker that has been idle past its pool's timeout), draining in-flight executions first as it does on SIGTERM.
9+
10+
Changes:
11+
12+
- Durations are written as durations. `submit --delay` takes `30s` or `5m` rather than a bare number of seconds, and no longer accepts one; `logs.flush_interval` and `metrics.flush_interval` in `coflux.toml` are duration strings (`"500ms"`) rather than floats, and a bare number is now an error rather than being read as nanoseconds. Both need editing by hand.
13+
- Follows the server's rename of duration fields: the manifest and submission payloads carry `delayMs`, `timeoutMs`, `maxAgeMs`, `backoffMinMs` and `backoffMaxMs`, and the worker protocol carries `max_age_ms`, `backoff_min_ms` and `backoff_max_ms`. A CLI of this version needs a server of this version, as before.
414

515
## 0.12.0
616

‎cli/cmd/coflux/main.go‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,10 @@ import (
44
"fmt"
55
"log/slog"
66
"os"
7+
"reflect"
78
"runtime"
89
"strings"
10+
"time"
911

1012
"github.com/bitroot/coflux/cli/internal/config"
1113
"github.com/bitroot/coflux/cli/internal/version"
@@ -49,9 +51,9 @@ func init() {
4951
viper.SetDefault("worker.concurrency", min(runtime.NumCPU()+4, 32))
5052
viper.SetDefault("blobs.threshold", 100)
5153
viper.SetDefault("logs.batch_size", 100)
52-
viper.SetDefault("logs.flush_interval", 0.5)
54+
viper.SetDefault("logs.flush_interval", "500ms")
5355
viper.SetDefault("metrics.batch_size", 100)
54-
viper.SetDefault("metrics.flush_interval", 0.2)
56+
viper.SetDefault("metrics.flush_interval", "200ms")
5557
viper.SetDefault("log_level", "info")
5658

5759
// Global flags
@@ -105,7 +107,8 @@ func init() {
105107
queueCmd.GroupID = "management"
106108
inputsCmd.GroupID = "management"
107109
catalogCmd.GroupID = "management"
108-
rootCmd.AddCommand(workspacesCmd, manifestsCmd, poolsCmd, tokensCmd, assetsCmd, blobsCmd, logsCmd, sessionsCmd, queueCmd, inputsCmd, catalogCmd)
110+
secretsCmd.GroupID = "management"
111+
rootCmd.AddCommand(workspacesCmd, manifestsCmd, poolsCmd, tokensCmd, secretsCmd, assetsCmd, blobsCmd, logsCmd, sessionsCmd, queueCmd, inputsCmd, catalogCmd)
109112
}
110113

111114
func initConfig(cmd *cobra.Command, args []string) error {
@@ -135,11 +138,38 @@ func initConfig(cmd *cobra.Command, args []string) error {
135138
return nil
136139
}
137140

141+
var durationType = reflect.TypeOf(time.Duration(0))
142+
143+
// stringToDurationHook decodes duration settings from duration strings
144+
// ("500ms", "2s"). A bare number is rejected rather than truncated to
145+
// nanoseconds: before 0.13 these were written as a float of seconds, and
146+
// silently turning `flush_interval = 0.5` into zero would be worse than
147+
// failing.
148+
func stringToDurationHook(from reflect.Type, to reflect.Type, data any) (any, error) {
149+
if to != durationType || from == durationType {
150+
return data, nil
151+
}
152+
if from.Kind() != reflect.String {
153+
return nil, fmt.Errorf("expected a duration string (e.g. \"500ms\"), got %v", data)
154+
}
155+
value, err := time.ParseDuration(data.(string))
156+
if err != nil {
157+
return nil, fmt.Errorf("invalid duration %q (expected e.g. \"500ms\")", data)
158+
}
159+
return value, nil
160+
}
161+
138162
// loadConfig unmarshals viper config into a Config struct
139163
func loadConfig() (*config.Config, error) {
140164
cfg := &config.Config{}
141165
if err := viper.Unmarshal(cfg, func(dc *mapstructure.DecoderConfig) {
142166
dc.ErrorUnused = true
167+
// Replaces viper's default StringToTimeDurationHookFunc with a
168+
// stricter one; the slice hook is viper's default, kept as-is.
169+
dc.DecodeHook = mapstructure.ComposeDecodeHookFunc(
170+
stringToDurationHook,
171+
mapstructure.StringToSliceHookFunc(","),
172+
)
143173
}); err != nil {
144174
return nil, fmt.Errorf("failed to unmarshal config: %w", err)
145175
}

0 commit comments

Comments
 (0)