Browse Source

New DAG execution model, works! But no dedupe yet

rei/symlinks
Olivier 'reivilibre' 8 months ago
parent
commit
58af07bbb2
25 changed files with 1699 additions and 927 deletions
  1. +5
    -0
      .gitignore
  2. +6
    -0
      mypy.ini
  3. +3
    -1
      scone/__main__.py
  4. +32
    -18
      scone/default/recipes/apt.py
  5. +35
    -30
      scone/default/recipes/filesystem.py
  6. +22
    -17
      scone/default/recipes/fridge.py
  7. +36
    -8
      scone/default/recipes/linux.py
  8. +9
    -9
      scone/default/recipes/postgres.py
  9. +15
    -18
      scone/default/recipes/python.py
  10. +28
    -21
      scone/default/recipes/systemd.py
  11. +3
    -3
      scone/default/steps/basic_steps.py
  12. +1
    -2
      scone/default/steps/fridge_steps.py
  13. +0
    -201
      scone/head/__init__.py
  14. +41
    -41
      scone/head/cli/__init__.py
  15. +193
    -0
      scone/head/dag.py
  16. +103
    -228
      scone/head/dependency_tracking.py
  17. +57
    -0
      scone/head/dot_emitter.py
  18. +163
    -0
      scone/head/grammar/scoml.tx
  19. +209
    -0
      scone/head/head.py
  20. +247
    -83
      scone/head/kitchen.py
  21. +443
    -99
      scone/head/menu_reader.py
  22. +33
    -142
      scone/head/recipe.py
  23. +6
    -3
      scone/head/sshconn.py
  24. +7
    -2
      scone/head/variables.py
  25. +2
    -1
      setup.py

+ 5
- 0
.gitignore View File

@ -0,0 +1,5 @@
/.idea
__pycache__
/scone.egg-info
/dist

+ 6
- 0
mypy.ini View File

@ -21,3 +21,9 @@ ignore_missing_imports = True
[mypy-asyncpg]
ignore_missing_imports = True
[mypy-frozendict]
ignore_missing_imports = True
[mypy-textx]
ignore_missing_imports = True

+ 3
- 1
scone/__main__.py View File

@ -3,7 +3,9 @@
# import sys
# from typing import List
#
# from scone.head import Head, Recipe
# from scone.head.head import Head
# from scone.head.recipe import Recipe
# from scone.head.kitchen import Kitchen
# from scone.head.recipe import Preparation


+ 32
- 18
scone/default/recipes/apt.py View File

@ -1,9 +1,9 @@
from typing import Dict, List, Set, Tuple
from scone.default.utensils.basic_utensils import SimpleExec
from scone.head import Head, Recipe
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
from scone.head.recipe import Recipe, RecipeContext
from scone.head.utils import check_type
@ -17,8 +17,8 @@ class AptInstallInternal(Recipe):
# TODO(extension, low): expand this into apt-install-now if we need
# the flexibility
def __init__(self, host: str, slug: str, args: dict, head: "Head"):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.packages: Set[str] = set()
@ -66,23 +66,37 @@ class AptPackage(Recipe):
internal_installers: Dict[Tuple[Head, str], AptInstallInternal] = {}
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.packages: List[str] = check_type(args["packages"], list)
def prepare(self, preparation: Preparation, head: Head) -> None:
super().prepare(preparation, head)
pair = (head, self.get_host())
if pair not in AptPackage.internal_installers:
install_internal = AptInstallInternal(self.get_host(), "internal", {}, head)
AptPackage.internal_installers[pair] = install_internal
preparation.subrecipe(install_internal)
preparation.provides("apt-stage", "internal-install-packages")
internal_installer = AptPackage.internal_installers.get(pair)
assert internal_installer is not None
internal_installer.packages.update(self.packages)
for package in self.packages:
preparation.provides("apt-package", package)
async def cook(self, kitchen: Kitchen) -> None:
# can't be tracked
kitchen.get_dependency_tracker().ignore()
# this is a one-off task assuming everything works
kitchen.get_dependency_tracker()
if self.packages:
update = await kitchen.ut1areq(
SimpleExec(["apt-get", "-yq", "update"], "/"), SimpleExec.Result
)
if update.exit_code != 0:
raise RuntimeError(
f"apt update failed with err {update.exit_code}: {update.stderr!r}"
)
install_args = ["apt-get", "-yq", "install"]
install_args += list(self.packages)
install = await kitchen.ut1areq(
SimpleExec(install_args, "/"), SimpleExec.Result
)
if install.exit_code != 0:
raise RuntimeError(
f"apt install failed with err {install.exit_code}:"
f" {install.stderr!r}"
)

+ 35
- 30
scone/default/recipes/filesystem.py View File

@ -12,9 +12,9 @@ from scone.default.utensils.basic_utensils import (
Stat,
)
from scone.default.utensils.dynamic_dependencies import HasChangedInSousStore
from scone.head import Head, Recipe
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
from scone.head.recipe import Recipe, RecipeContext
from scone.head.utils import check_type, check_type_opt
@ -28,7 +28,7 @@ class DeclareFile(Recipe):
_NAME = "declare-file"
def prepare(self, preparation: Preparation, head: Head):
preparation.provides("file", self._args["path"])
preparation.provides("file", self.arguments["path"])
async def cook(self, kitchen: Kitchen):
# mark as tracked.
@ -45,7 +45,7 @@ class DeclareDirectory(Recipe):
_NAME = "declare-dir"
def prepare(self, preparation: Preparation, head: Head):
preparation.provides("directory", self._args["path"])
preparation.provides("directory", self.arguments["path"])
async def cook(self, kitchen: Kitchen):
# mark as tracked.
@ -59,8 +59,8 @@ class EnsureDirectory(Recipe):
_NAME = "directory"
def __init__(self, host: str, slug: str, args: dict, head: "Head"):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
parents = args.get("parents", 0)
assert isinstance(parents, int)
@ -74,7 +74,7 @@ class EnsureDirectory(Recipe):
self.parents = parents
self.mode = parse_mode(mode, directory=True)
self._make: List[str] = []
self.targ_user = args.get("owner", self.get_user(head))
self.targ_user = args.get("owner", recipe_context.user)
self.targ_group = args.get("group", self.targ_user)
def prepare(self, preparation: Preparation, head: "Head"):
@ -123,8 +123,8 @@ class ExtractTar(Recipe):
_NAME = "tar-extract"
def __init__(self, host: str, slug: str, args: dict, head: "Head"):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.tar = check_type(args.get("tar"), str)
self.dir = check_type(args.get("dir"), str)
@ -165,8 +165,8 @@ class RunScript(Recipe):
_NAME = "script-run"
def __init__(self, host: str, slug: str, args: dict, head: "Head"):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.working_dir = check_type(args.get("working_dir"), str)
@ -196,8 +196,8 @@ class CommandOnChange(Recipe):
_NAME = "command-on-change"
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.purpose = check_type(args.get("purpose"), str)
self.command = check_type(args.get("command"), list)
@ -232,8 +232,8 @@ class GitCheckout(Recipe):
# declare SAFE_TO_SKIP. Perhaps we want to stop that unless you opt out?
# But oh well for now.
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.repo_src = check_type(args.get("src"), str)
self.dest_dir = check_type(args.get("dest"), str)
@ -270,9 +270,7 @@ class GitCheckout(Recipe):
stat = await k.ut1a(Stat(self.dest_dir), Stat.Result)
if stat is None:
# doesn't exist; git init it
await exec_no_fails(
k, ["git", "init", self.dest_dir], "/"
)
await exec_no_fails(k, ["git", "init", self.dest_dir], "/")
stat = await k.ut1a(Stat(self.dest_dir), Stat.Result)
if stat is None:
@ -283,29 +281,30 @@ class GitCheckout(Recipe):
# add the remote, removing it first to ensure it's what we want
# don't care if removing fails
await k.ut1areq(SimpleExec(["git", "remote", "remove", "scone"], self.dest_dir), SimpleExec.Result)
await k.ut1areq(
SimpleExec(["git", "remote", "remove", "scone"], self.dest_dir),
SimpleExec.Result,
)
await exec_no_fails(
k, ["git", "remote", "add", "scone", self.repo_src], self.dest_dir
)
# fetch the latest from the remote
await exec_no_fails(
k, ["git", "fetch", "scone"], self.dest_dir
)
await exec_no_fails(k, ["git", "fetch", "scone"], self.dest_dir)
# figure out what ref we want to use
# TODO(performance): fetch only this ref?
ref = self.ref or f"scone/{self.branch}"
# switch to that ref
await exec_no_fails(
k, ["git", "switch", "--detach", ref], self.dest_dir
)
await exec_no_fails(k, ["git", "switch", "--detach", ref], self.dest_dir)
# if we use submodules
if self.submodules:
await exec_no_fails(
k, ["git", "submodule", "update", "--init", "--recursive"], self.dest_dir
k,
["git", "submodule", "update", "--init", "--recursive"],
self.dest_dir,
)
for expected in self.expect:
@ -313,10 +312,16 @@ class GitCheckout(Recipe):
# TODO(performance, low): parallelise these
stat = await k.ut1a(Stat(expected_path_str), Stat.Result)
if not stat:
raise RuntimeError(f"expected {expected_path_str} to exist but it did not")
raise RuntimeError(
f"expected {expected_path_str} to exist but it did not"
)
if stat.dir and not expected.endswith("/"):
raise RuntimeError(f"expected {expected_path_str} to exist as a file but it is a dir")
raise RuntimeError(
f"expected {expected_path_str} to exist as a file but it is a dir"
)
if not stat.dir and expected.endswith("/"):
raise RuntimeError(f"expected {expected_path_str} to exist as a dir but it is a file")
raise RuntimeError(
f"expected {expected_path_str} to exist as a dir but it is a file"
)

+ 22
- 17
scone/default/recipes/fridge.py View File

@ -8,11 +8,15 @@ from urllib.request import urlretrieve
from scone.common.misc import sha256_file
from scone.common.modeutils import DEFAULT_MODE_FILE, parse_mode
from scone.default.steps import fridge_steps
from scone.default.steps.fridge_steps import FridgeMetadata, load_and_transform, SUPERMARKET_RELATIVE
from scone.default.utensils.basic_utensils import WriteFile, Chown
from scone.head import Head
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation, Recipe
from scone.default.steps.fridge_steps import (
SUPERMARKET_RELATIVE,
FridgeMetadata,
load_and_transform,
)
from scone.default.utensils.basic_utensils import Chown, WriteFile
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
from scone.head.recipe import Recipe, RecipeContext
from scone.head.utils import check_type
@ -23,8 +27,9 @@ class FridgeCopy(Recipe):
_NAME = "fridge-copy"
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head: Head):
super().__init__(recipe_context, args, head)
fp = fridge_steps.search_in_fridge(head, args["src"])
if fp is None:
raise ValueError(f"Cannot find {args['src']} in the fridge.")
@ -44,7 +49,7 @@ class FridgeCopy(Recipe):
mode = args.get("mode", DEFAULT_MODE_FILE)
assert isinstance(mode, str) or isinstance(mode, int)
self.fridge_path: str = args["src"]
self.fridge_path: str = check_type(args["src"], str)
self.real_path: Path = fp
self.fridge_meta: FridgeMetadata = meta
self.mode = parse_mode(mode, directory=False)
@ -56,7 +61,7 @@ class FridgeCopy(Recipe):
async def cook(self, k: Kitchen) -> None:
data = await load_and_transform(
k, self.fridge_meta, self.real_path, self.get_host()
k, self.fridge_meta, self.real_path, self.recipe_context.sous
)
dest_str = str(self.destination)
chan = await k.start(WriteFile(dest_str, self.mode))
@ -69,9 +74,7 @@ class FridgeCopy(Recipe):
# hash_of_data = sha256_bytes(data)
# k.get_dependency_tracker().register_remote_file(dest_str, hash_of_data)
await k.get_dependency_tracker().register_fridge_file(
self.fridge_path, self.real_path
)
k.get_dependency_tracker().register_fridge_file(self.fridge_path)
class Supermarket(Recipe):
@ -84,8 +87,8 @@ class Supermarket(Recipe):
# dict of target path → future that will complete when it's downloaded
in_progress: Dict[str, Future] = dict()
def __init__(self, host: str, slug: str, args: dict, head: "Head"):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.url = args.get("url")
assert isinstance(self.url, str)
@ -101,7 +104,7 @@ class Supermarket(Recipe):
else:
self.destination = Path(args["dest"]).resolve()
self.owner = check_type(args.get("owner", self.get_user(head)), str)
self.owner = check_type(args.get("owner", self.recipe_context.user), str)
self.group = check_type(args.get("group", self.owner), str)
mode = args.get("mode", DEFAULT_MODE_FILE)
@ -115,7 +118,9 @@ class Supermarket(Recipe):
async def cook(self, kitchen: "Kitchen"):
# need to ensure we download only once, even in a race…
supermarket_path = Path(kitchen.head.directory, SUPERMARKET_RELATIVE, self.sha256)
supermarket_path = Path(
kitchen.head.directory, SUPERMARKET_RELATIVE, self.sha256
)
if self.sha256 in Supermarket.in_progress:
await Supermarket.in_progress[self.sha256]
@ -136,7 +141,7 @@ Downloaded by {self}
self.url,
str(supermarket_path),
self.sha256,
note
note,
),
)


+ 36
- 8
scone/default/recipes/linux.py View File

@ -4,9 +4,9 @@ from typing import Optional
from scone.default.steps import linux_steps
from scone.default.utensils.linux_utensils import GetPasswdEntry
from scone.head import Head, Recipe
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
from scone.head.recipe import Recipe, RecipeContext
from scone.head.utils import check_type, check_type_opt
logger = logging.getLogger(__name__)
@ -15,12 +15,10 @@ logger = logging.getLogger(__name__)
class LinuxUser(Recipe):
_NAME = "os-user"
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
if slug[0] == "@":
raise ValueError("os-user should be used like [os-user.username].")
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.user_name = slug
self.user_name = check_type(args.get("name"), str)
self.make_group = check_type(args.get("make_group", True), bool)
self.make_home = check_type(args.get("make_home", True), bool)
self.home: Optional[str] = check_type_opt(args.get("home"), str)
@ -62,3 +60,33 @@ class LinuxUser(Recipe):
self.make_group,
self.home,
)
class DeclareLinuxUser(Recipe):
_NAME = "declare-os-user"
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.user_name = check_type(args.get("name"), str)
def prepare(self, preparation: Preparation, head: "Head") -> None:
preparation.provides("os-user", self.user_name)
async def cook(self, kitchen: Kitchen) -> None:
kitchen.get_dependency_tracker()
class DeclareLinuxGroup(Recipe):
_NAME = "declare-os-group"
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.name = check_type(args.get("name"), str)
def prepare(self, preparation: Preparation, head: "Head") -> None:
preparation.provides("os-group", self.name)
async def cook(self, kitchen: Kitchen) -> None:
kitchen.get_dependency_tracker()

+ 9
- 9
scone/default/recipes/postgres.py View File

@ -1,17 +1,17 @@
from scone.default.utensils.db_utensils import PostgresTransaction
from scone.head import Head, Recipe
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
from scone.head.recipe import Recipe, RecipeContext
from scone.head.utils import check_type
class PostgresDatabase(Recipe):
_NAME = "pg-db"
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.database_name = slug
self.database_name = check_type(args.get("name"), str)
self.owner = check_type(args.get("owner"), str)
self.encoding = args.get("encoding", "utf8")
self.collate = args.get("collate", "en_GB.utf8")
@ -56,10 +56,10 @@ class PostgresDatabase(Recipe):
class PostgresUser(Recipe):
_NAME = "pg-user"
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.user_name = slug
self.user_name = check_type(args.get("name"), str)
self.password = check_type(args.get("password"), str)
def prepare(self, preparation: Preparation, head: Head) -> None:


+ 15
- 18
scone/default/recipes/python.py View File

@ -1,12 +1,11 @@
from pathlib import Path
from typing import Tuple, List
from typing import List, Tuple
from scone.default.recipes.apt import AptPackage
from scone.default.steps.basic_steps import exec_no_fails
from scone.default.steps.filesystem_steps import depend_remote_file
from scone.head import Head, Recipe
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
from scone.head.recipe import Recipe, RecipeContext
from scone.head.utils import check_type
@ -20,8 +19,8 @@ class PythonVenv(Recipe):
_NAME = "python-venv"
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.dir = check_type(args.get("dir"), str)
self.interpreter = check_type(args.get("interpreter"), str)
@ -39,31 +38,29 @@ class PythonVenv(Recipe):
def prepare(self, preparation: Preparation, head: Head):
super().prepare(preparation, head)
preparation.needs("dir", str(Path(self.dir).parent))
preparation.needs("directory", str(Path(self.dir).parent))
for name, flags in self.install:
if "-r" in flags:
preparation.needs("file", name)
elif "git" in flags or "dir" in flags:
preparation.needs("dir", name)
preparation.needs("directory", name)
final_script = str(Path(self.dir, "bin/python"))
preparation.provides("file", str(final_script))
if not self.no_apt_install:
preparation.subrecipe(
AptPackage(
self.get_host(), "@venv-apt", {"packages": ["python3-venv"]}, head
)
)
preparation.needs("apt-stage", "packages-installed")
# preparation.subrecipe(
# AptPackage(self.recipe_context, {"packages": ["python3-venv"]})
# )
# preparation.needs("apt-stage", "packages-installed")
preparation.needs("apt-package", "python3-venv")
async def cook(self, kitchen: Kitchen):
dt = kitchen.get_dependency_tracker()
await exec_no_fails(
kitchen, [self.interpreter, "-m", "venv", self.dir], "/"
)
await exec_no_fails(kitchen, [self.interpreter, "-m", "venv", self.dir], "/")
install_args = []
for name, flags in self.install:


+ 28
- 21
scone/default/recipes/systemd.py View File

@ -2,9 +2,9 @@ from typing import Dict
from scone.default.recipes.filesystem import CommandOnChange
from scone.default.utensils.basic_utensils import SimpleExec
from scone.head import Head, Recipe
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
from scone.head.recipe import Recipe, RecipeContext
from scone.head.utils import check_type, check_type_opt
@ -17,10 +17,11 @@ class SystemdUnit(Recipe):
daemon_reloaders: Dict[str, CommandOnChange] = {}
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.unit_name = slug if "." in slug else slug + ".service"
unit = check_type(args.get("unit"), str)
self.unit_name = unit if "." in unit else unit + ".service"
self.at = check_type(args.get("at"), str)
self.enabled = check_type_opt(args.get("enabled"), bool)
self.restart_on = check_type_opt(args.get("restart_on"), list)
@ -32,44 +33,49 @@ class SystemdUnit(Recipe):
if self.enabled is not None:
enable_recipe = SystemdEnabled(
self.get_host(),
self.unit_name,
{"enabled": self.enabled, "at": self.at, ".user": "root"},
head,
self.recipe_context,
{
"unit": self.unit_name,
"enabled": self.enabled,
"at": self.at,
".user": "root",
},
None,
)
preparation.subrecipe(enable_recipe)
preparation.needs("systemd-stage", "enabled")
daemon_reloader = SystemdUnit.daemon_reloaders.get(self.get_host(), None)
daemon_reloader = SystemdUnit.daemon_reloaders.get(
self.recipe_context.sous, None
)
if not daemon_reloader:
# TODO this should be replaced with a dedicated command which provides
# those units.
daemon_reloader = CommandOnChange(
self.get_host(),
"systemd-internal",
self.recipe_context,
{
"purpose": "systemd.daemon_reload",
"command": ["systemctl", "daemon-reload"],
"files": [],
".user": "root",
},
head,
None,
)
preparation.subrecipe(daemon_reloader)
file_list = getattr(daemon_reloader, "_args")["files"]
# file_list = getattr(daemon_reloader, "_args")["files"]
file_list = [] # TODO
file_list.append(self.at)
if self.restart_on:
service_reloader = CommandOnChange(
self.get_host(),
"systemd-internal",
self.recipe_context,
{
"purpose": "systemd.unit_reload",
"command": ["systemctl", "reload", self.unit_name],
"files": self.restart_on + [self.at],
".user": "root",
},
head,
None,
)
preparation.subrecipe(service_reloader)
@ -85,10 +91,11 @@ class SystemdEnabled(Recipe):
_NAME = "systemd-enabled"
def __init__(self, host: str, slug: str, args: dict, head: Head):
super().__init__(host, slug, args, head)
def __init__(self, recipe_context: RecipeContext, args: dict, head):
super().__init__(recipe_context, args, head)
self.unit_name = slug if "." in slug else slug + ".service"
unit = check_type(args.get("unit"), str)
self.unit_name = unit if "." in unit else unit + ".service"
self.at = check_type(args.get("at"), str)
self.enabled = check_type_opt(args.get("enabled"), bool)


+ 3
- 3
scone/default/steps/basic_steps.py View File

@ -2,9 +2,9 @@ from pathlib import PurePath
from typing import List, Optional, Union
from scone.default.utensils.basic_utensils import SimpleExec
from scone.head import Recipe
from scone.head.exceptions import CookingError
from scone.head.kitchen import Kitchen, current_recipe
from scone.head.recipe import Recipe
class ExecutionFailure(CookingError):
@ -47,8 +47,8 @@ async def exec_no_fails(
raise ExecutionFailure(
args,
working_dir,
recipe.get_host(),
recipe.get_user(kitchen.head),
recipe.recipe_context.sous,
recipe.recipe_context.user,
result,
)
else:


+ 1
- 2
scone/default/steps/fridge_steps.py View File

@ -4,10 +4,9 @@ from typing import List, Optional, Tuple, Union
from jinja2 import Template
from scone.head import Head
from scone.head.head import Head
from scone.head.kitchen import Kitchen
SUPERMARKET_RELATIVE = ".scone-cache/supermarket"


+ 0
- 201
scone/head/__init__.py View File

@ -1,201 +0,0 @@
import copy
import itertools
import logging
import re
import sys
from os import path
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, cast
import toml
from nacl.encoding import URLSafeBase64Encoder
from scone.common.loader import ClassLoader
from scone.common.misc import eprint
from scone.common.pools import Pools
from scone.head import menu_reader
from scone.head.menu_reader import HostMenu, Menu
from scone.head.recipe import Recipe, recipe_name_getter
from scone.head.secrets import SecretAccess
from scone.head.variables import Variables, merge_right_into_left_inplace
logger = logging.getLogger(__name__)
class Head:
def __init__(
self,
directory: str,
recipe_loader: ClassLoader[Recipe],
menu: Menu,
sous: Dict[str, dict],
groups: Dict[str, List[str]],
secret_access: Optional[SecretAccess],
pools: Pools,
):
self.directory = directory
self.recipe_loader = recipe_loader
self.menu = menu
self.souss = sous
self.groups = groups
self.secret_access = secret_access
self.variables: Dict[str, Variables] = dict()
self.pools = pools
@staticmethod
def open(directory: str):
with open(path.join(directory, "scone.head.toml")) as head_toml:
head_data = toml.load(head_toml)
secret_access: Optional[SecretAccess] = None
if "freezer" in head_data and "restaurant_id" in head_data["freezer"]:
secret_access = SecretAccess(head_data["freezer"]["restaurant_id"])
secret_access.get_existing()
if not secret_access.key:
eprint("Failed to load freezer secret.")
sys.exit(12)
recipe_module_roots = head_data.get("recipe_roots", ["scone.default.recipes"])
# load available recipes
recipe_loader: ClassLoader[Recipe] = ClassLoader(Recipe, recipe_name_getter)
for recipe_root in recipe_module_roots:
recipe_loader.add_package_root(recipe_root)
sous = head_data.get("sous", dict())
groups = head_data.get("group", dict())
groups["all"] = list(sous.keys())
# load the menu
menu = menu_reader.parse_toml_menu_descriptors(path.join(directory, "menu"))
pools = Pools()
head = Head(directory, recipe_loader, menu, sous, groups, secret_access, pools)
head._load_variables()
return head
def _preload_variables(self, who_for: str) -> Tuple[dict, dict]:
out_frozen: Dict[str, Any] = {}
out_chilled: Dict[str, Any] = {}
vardir = Path(self.directory, "vars", who_for)
logger.debug("preloading vars for %s in %s", who_for, str(vardir))
for file in vardir.glob("*.vf.toml"):
if not file.is_file():
continue
with file.open() as var_file:
logger.debug("Opened %s for frozen vars", file)
frozen_vars = cast(Dict[Any, Any], toml.load(var_file))
merge_right_into_left_inplace(out_frozen, frozen_vars)
for file in vardir.glob("*.v.toml"):
if not file.is_file():
continue
with file.open() as var_file:
logger.debug("Opened %s for vars", file)
chilled_vars = cast(Dict[Any, Any], toml.load(var_file))
merge_right_into_left_inplace(out_chilled, chilled_vars)
to_transform = [out_frozen]
while to_transform:
next_dict = to_transform.pop()
for k, v in next_dict.items():
if isinstance(v, str):
b64_secret = re.sub(r"\s", "", v)
if not self.secret_access:
raise RuntimeError("Secret access disabled; cannot thaw.")
next_dict[k] = self.secret_access.decrypt_bytes(
b64_secret.encode(), encoder=URLSafeBase64Encoder
).decode()
elif isinstance(v, dict):
to_transform.append(v)
else:
raise ValueError(f"Not permitted in frozen variables file: '{v}'.")
return out_chilled, out_frozen
def _load_variables(self):
preload: Dict[str, Tuple[dict, dict]] = dict()
for who_name in itertools.chain(self.souss, self.groups):
preload[who_name] = self._preload_variables(who_name)
for sous_name in self.souss:
order = ["all"]
order += [
group
for group, members in self.groups.items()
if sous_name in members and group != "all"
]
order.append(sous_name)
chilled: Dict[str, Any] = {}
frozen: Dict[str, Any] = {}
for who_name in order:
in_chilled, in_frozen = preload[who_name]
merge_right_into_left_inplace(chilled, in_chilled)
merge_right_into_left_inplace(frozen, in_frozen)
sous_vars = Variables()
sous_vars.load_plain(frozen)
sous_vars.load_vars_with_substitutions(chilled)
self.variables[sous_name] = sous_vars
def _construct_hostmenu_for(
self, hostmenu: HostMenu, host: str, recipe_list: List[Recipe], head: "Head"
) -> None:
for recipe_id, dishes in hostmenu.dishes.items():
recipe_cls = self.recipe_loader.get_class(recipe_id)
if not recipe_cls:
raise RuntimeError(f"Unable to find recipe class for '{recipe_id}'.")
for slug, args in dishes.items():
args = copy.deepcopy(args)
self.variables[host].substitute_inplace_in_dict(args)
recipe = recipe_cls.from_menu(host, slug, args, head)
recipe_list.append(recipe)
def construct_recipes(self):
recipes = {}
for sous in self.souss:
logger.debug("Constructing recipes for %s", sous)
sous_recipe_list: List[Recipe] = []
# construct recipes for it only
sous_hm = self.menu.hostmenus.get(sous)
if sous_hm is not None:
self._construct_hostmenu_for(sous_hm, sous, sous_recipe_list, self)
# construct recipes for it that are for groups it is in
for group, members in self.groups.items():
if sous in members:
group_hm = self.menu.hostmenus.get(group)
if group_hm is not None:
self._construct_hostmenu_for(
group_hm, sous, sous_recipe_list, self
)
recipes[sous] = sous_recipe_list
logger.info("Constructed %d recipes for %s.", len(sous_recipe_list), sous)
return recipes
def debug_info(self) -> str:
lines = []
lines.append("Head Configuration")
lines.append(" Sous List")
for name, sous in self.souss.items():
lines.append(f" - {name} = {sous}")
lines.append("")
lines.append(" Sous Groups")
for name, group in self.groups.items():
lines.append(f" - {name} = {group}")
lines.append("")
lines += [" " + line for line in str(self.recipe_loader).splitlines()]
lines.append("")
lines += [" " + line for line in str(self.menu).splitlines()]
lines.append("")
return "\n".join(lines)

+ 41
- 41
scone/head/cli/__init__.py View File

@ -8,10 +8,10 @@ from pathlib import Path
from scone.common.misc import eprint
from scone.common.pools import Pools
from scone.head import Head
from scone.head.dependency_tracking import DependencyCache, run_dep_checks
from scone.head.kitchen import Kitchen
from scone.head.recipe import Preparation, Recipe
from scone.head import dot_emitter
from scone.head.dependency_tracking import DependencyCache
from scone.head.head import Head
from scone.head.kitchen import Kitchen, Preparation
def cli() -> None:
@ -64,43 +64,38 @@ async def cli_async() -> int:
eprint(f"Selected the following souss: {', '.join(hosts)}")
recipes_by_sous = head.construct_recipes()
recipes_to_do = []
for sous in hosts:
recipes_to_do += recipes_by_sous.get(sous, [])
eprint(f"Preparing {len(recipes_to_do)} recipes…")
prepare = Preparation(recipes_to_do)
eprint("Preparing recipes…")
prepare = Preparation(head)
start_ts = time.monotonic()
order = prepare.prepare(head)
notifying_provides = prepare.notifying_provides
prepare.prepare_all()
del prepare
end_ts = time.monotonic()
eprint(f"Preparation completed in {end_ts - start_ts:.3f} s.")
eprint(f"{len(order)} courses planned.")
# eprint(f"{len(order)} courses planned.")
dot_emitter.emit_dot(head.dag, Path(cdir, "dag.0.dot"))
dep_cache = await DependencyCache.open(
os.path.join(head.directory, "depcache.sqlite3")
)
eprint("Checking dependency cache…")
start_ts = time.monotonic()
depchecks = await run_dep_checks(head, dep_cache, order)
end_ts = time.monotonic()
eprint(f"Checking finished in {end_ts - start_ts:.3f} s.") # TODO show counts
for epoch, items in enumerate(order):
print(f"----- Course {epoch} -----")
for item in items:
if isinstance(item, Recipe):
state = depchecks[item].label.name
print(f" > recipe ({state}) {item}")
elif isinstance(item, tuple):
kind, ident, extra = item
print(f" - we now have {kind} {ident} {dict(extra)}")
# eprint("Checking dependency cache…")
# start_ts = time.monotonic()
# depchecks = await run_dep_checks(head, dep_cache, order)
# end_ts = time.monotonic()
# eprint(f"Checking finished in {end_ts - start_ts:.3f} s.") # TODO show counts
#
# for epoch, items in enumerate(order):
# print(f"----- Course {epoch} -----")
#
# for item in items:
# if isinstance(item, Recipe):
# state = depchecks[item].label.name
# print(f" > recipe ({state}) {item}")
# elif isinstance(item, tuple):
# kind, ident, extra = item
# print(f" - we now have {kind} {ident} {dict(extra)}")
eprint("Ready to cook? [y/N]: ", end="")
if argp.yes:
@ -110,16 +105,21 @@ async def cli_async() -> int:
eprint("Stopping.")
return 101
kitchen = Kitchen(head, dep_cache, notifying_provides)
for epoch, epoch_items in enumerate(order):
print(f"Cooking Course {epoch} of {len(order)}")
await kitchen.run_epoch(
epoch_items, depchecks, concurrency_limit_per_host=8
)
for sous in hosts:
await dep_cache.sweep_old(sous)
kitchen = Kitchen(head, dep_cache)
# for epoch, epoch_items in enumerate(order):
# print(f"Cooking Course {epoch} of {len(order)}")
# await kitchen.run_epoch(
# epoch_items, depchecks, concurrency_limit_per_host=8
# )
#
# for sous in hosts: TODO this is not definitely safe
# await dep_cache.sweep_old(sous)
try:
await kitchen.cook_all()
finally:
dot_emitter.emit_dot(head.dag, Path(cdir, "dag.9.dot"))
return 0
finally:


+ 193
- 0
scone/head/dag.py View File

@ -0,0 +1,193 @@
from collections import defaultdict
from enum import Enum
from typing import Dict, Optional, Set, Union
import attr
from frozendict import frozendict
from scone.head.recipe import Recipe
class RecipeState(Enum):
# Just loaded from menu, or otherwise created
LOADED = 0
# Has been prepared — we know its dependencies for this run
PREPARED = 1
# This recipe needs to be cooked, but may be blocked by dependencies
PENDING = 2
# This recipe is not blocked by any further
COOKABLE = 3
# This recipe is being cooked
BEING_COOKED = 4
# This recipe has been cooked!
COOKED = 5
# This recipe has not been cooked because it didn't need to be.
SKIPPED = 10
@staticmethod
def is_completed(state):
return state in (RecipeState.COOKED, RecipeState.SKIPPED)
@attr.s(auto_attribs=True)
class RecipeMeta:
"""
State of the recipe.
"""
state: RecipeState = RecipeState.LOADED
"""
Uncompleted incoming edge count.
"""
incoming_uncompleted: int = 0
@attr.s(auto_attribs=True, frozen=True)
class Resource:
"""
Resource kind.
"""
kind: str
"""
Resource ID
"""
id: str
"""
Resource sous, or None if it's on the head
"""
sous: Optional[str]
"""
Optional dict of extra parameters needed to disambiguate the resource,
though should only be used where necessary and sensible to do so.
"""
# extra_params: Optional[frozendict[str, str]] = None
extra_params: Optional[frozendict] = None
def __str__(self) -> str:
extra_str = "" if not self.extra_params else f" {self.extra_params!r}"
sous_str = "" if not self.sous else f" on {self.sous}"
return f"{self.kind}({self.id}){extra_str}{sous_str}"
@attr.s(auto_attribs=True)
class ResourceMeta:
"""
Whether the resource is completed or not.
A resource becomes completed when all its incoming edges are completed,
or it has no incoming edges and is not a hard need.
"""
completed: bool = False
"""
Uncompleted incoming edge count.
"""
incoming_uncompleted: int = 0
"""
Whether the resource is considered a hard need.
A resource is a hard need when we cannot proceed without something
providing it.
"""
hard_need: bool = False
Vertex = Union["Recipe", Resource]
class RecipeDag:
def __init__(self):
self.vertices: Set[Vertex] = set()
# edges go from A -> B where B needs A to run.
self.edges: Dict[Vertex, Set[Vertex]] = defaultdict(set)
self.reverse_edges: Dict[Vertex, Set[Vertex]] = defaultdict(set)
self.recipe_meta: Dict[Recipe, RecipeMeta] = dict()
self.resource_meta: Dict[Resource, ResourceMeta] = dict()
self.resource_time: Dict[Resource, int] = dict()
def add(self, vertex: Vertex):
self.vertices.add(vertex)
if isinstance(vertex, Recipe):
self.recipe_meta[vertex] = RecipeMeta()
elif isinstance(vertex, Resource):
self.resource_meta[vertex] = ResourceMeta()
def needs(
self, needer: "Recipe", resource: Resource, soft_wants: bool = False
) -> None:
if needer not in self.vertices:
raise ValueError(f"Needer {needer} not in vertices!")
if resource not in self.vertices:
self.add(resource)
if needer in self.edges[resource]:
return
self.edges[resource].add(needer)
self.reverse_edges[needer].add(resource)
needer_meta = self.recipe_meta[needer]
resource_meta = self.resource_meta[resource]
if not soft_wants:
resource_meta.hard_need = True
if not resource_meta.completed:
needer_meta.incoming_uncompleted += 1
def provides(self, provider: "Recipe", resource: Resource) -> None:
if provider not in self.vertices:
raise ValueError(f"Provider {provider} not in vertices!")
if resource not in self.vertices:
self.add(resource)
if resource in self.edges[provider]:
return
self.edges[provider].add(resource)
self.reverse_edges[resource].add(provider)
provider_meta = self.recipe_meta[provider]
resource_meta = self.resource_meta[resource]
if not RecipeState.is_completed(provider_meta.state):
resource_meta.incoming_uncompleted += 1
resource_meta.completed = False
else:
if resource_meta.incoming_uncompleted == 0:
resource_meta.completed = True
def add_ordering(self, before: "Recipe", after: "Recipe") -> None:
if before not in self.vertices:
raise ValueError(f"Before {before} not in vertices!")
if after not in self.vertices:
raise ValueError(f"After {after} not in vertices!")
after_meta = self.recipe_meta[after]
before_meta = self.recipe_meta[before]
if after in self.edges[before]:
return
self.edges[before].add(after)
self.reverse_edges[after].add(before)
if not RecipeState.is_completed(before_meta.state):
after_meta.incoming_uncompleted += 1
# TODO if after_meta.state ==
# TODO else ...

+ 103
- 228
scone/head/dependency_tracking.py View File

@ -1,11 +1,8 @@
import asyncio
import json
import logging
import time
from asyncio import Queue
from enum import Enum
from hashlib import sha256
from typing import Any, Dict, List, NamedTuple, Optional, Tuple, Union
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
import aiosqlite
import attr
@ -13,10 +10,12 @@ import canonicaljson
import cattr
from aiosqlite import Connection
from scone.common.misc import sha256_file
from scone.common.pools import Pools
from scone.head import Head, Recipe, Variables, recipe_name_getter
from scone.head.recipe import DepEle
from scone.head.dag import Resource
from scone.head.recipe import recipe_name_getter
if TYPE_CHECKING:
from scone.head.dag import RecipeDag
from scone.head.recipe import Recipe
canonicaljson.set_json_library(json)
logger = logging.getLogger(__name__)
@ -24,9 +23,6 @@ logger = logging.getLogger(__name__)
# TODO(security, low): how to prevent passwords being recovered from the
# paramhashes in a dependency store?
# TODO(correctness, perf): recipes with @src@0 slugs should not be registered
# to a slug.
def _canonicalise_dict(input: Dict[str, Any]) -> Dict[str, Any]:
output: Dict[str, Any] = {}
@ -48,139 +44,104 @@ def hash_dict(value: dict) -> str:
).hexdigest()
def paramhash_recipe(recipe: Recipe) -> str:
args = getattr(recipe, "_args").copy()
del args[".source"]
return hash_dict(args)
def paramhash_recipe(recipe: "Recipe") -> str:
return hash_dict(
{
"args": recipe.arguments,
"sous": recipe.recipe_context.sous,
"user": recipe.recipe_context.user,
}
)
@attr.s(auto_attribs=True)
class DependencyBook:
var_names: List[str]
var_hash: str
fridge_hashes: Dict[str, str]
recipe_revisions: Dict[str, int]
dyn_sous_file_hashes: Dict[str, str]
async def can_skip_static(self, head: Head, recipe: Recipe) -> bool:
from scone.default.steps.fridge_steps import search_in_fridge
# start with variables
sous_vars = head.variables[recipe.get_host()]
var_comp = dict()
for var_name in self.var_names:
try:
var_comp[var_name] = sous_vars.get_dotted(var_name)
except KeyError:
return False
if hash_dict(var_comp) != self.var_hash:
return False
# now we have to check files in the fridge
for fridge_name, expected_hash in self.fridge_hashes.items():
real_pathstr = search_in_fridge(head, fridge_name)
if not real_pathstr:
# vanished locally; that counts as a change
return False
real_hash = await asyncio.get_running_loop().run_in_executor(
head.pools.threaded, sha256_file, real_pathstr
)
if real_hash != expected_hash:
return False
provided: Dict[Resource, int] = dict()
watching: Dict[Resource, int] = dict()
last_changed: int = 0
cache_data: Dict[str, Any] = dict()
ignored: bool = False
# TODO(performance, feature): track more in-depth details, perhaps as a
# per-resource cache thing, so that we can track the info needed to know
# if it changed...?
def _unstructure(self) -> dict:
return {
"provided": cattr.unstructure(tuple(self.provided.items())),
"watching": cattr.unstructure(tuple(self.watching.items())),
"last_changed": self.last_changed,
"cache_data": self.cache_data,
"ignored": self.ignored,
}
@staticmethod
def _structure(dictionary: dict) -> "DependencyBook":
provided = {cattr.structure(k, Resource): v for k, v in dictionary["provided"]}
watching = {cattr.structure(k, Resource): v for k, v in dictionary["watching"]}
return DependencyBook(
provided=provided,
watching=watching,
last_changed=dictionary["last_changed"],
cache_data=dictionary["cache_data"],
ignored=dictionary["ignored"],
)
return True
def has_dynamic(self) -> bool:
return len(self.dyn_sous_file_hashes) > 0
cattr.global_converter.register_unstructure_hook(
DependencyBook, DependencyBook._unstructure
)
cattr.global_converter.register_structure_hook(
DependencyBook, DependencyBook._structure
)
class DependencyTracker:
"""
Tracks the dependencies of a task and then inserts a row as needed.
"""
def __init__(self, pools: Pools):
self._vars: Dict[str, Any] = {}
self._fridge: Dict[str, str] = {}
self._recipe_revisions: Dict[str, int] = {}
self._dyn_sous_files: Dict[str, str] = {}
self._ignored = False
self._pools = pools
def ignore(self):
"""
Call when dependency tracking is not desired (or not advanced enough to
be useful.)
"""
self._ignored = True
async def register_fridge_file(self, fridge_path: str, real_path: str):
if fridge_path not in self._fridge:
f_hash = await asyncio.get_running_loop().run_in_executor(
self._pools.threaded, sha256_file, real_path
)
self._fridge[fridge_path] = f_hash
def __init__(self, book: DependencyBook, dag: "RecipeDag"):
self.book: DependencyBook = book
self._dag: RecipeDag = dag
self._time: int = int(time.time() * 1000)
def register_recipe(self, recipe: Recipe):
cls = recipe.__class__
rec_name = recipe_name_getter(cls)
if not rec_name:
return
self._recipe_revisions[rec_name] = getattr(cls, "_REVISION", None)
def watch(self, resource: Resource) -> None:
# XXX self.book.watching[resource] = self._dag.resource_time[resource]
self.book.watching[resource] = -42
def register_variable(self, variable: str, value: Union[dict, str, int]):
self._vars[variable] = value
def provide(self, resource: Resource, time: Optional[int] = None) -> None:
if time is None:
time = self._time
self._dag.resource_time[resource] = time
def register_remote_file(self, file: str, file_hash: str):
self._dyn_sous_files[file] = file_hash
def ignore(self) -> None:
self.book.ignored = True
def make_depbook(self) -> Optional[DependencyBook]:
if self._ignored:
return None
dep_book = DependencyBook(
list(self._vars.keys()),
hash_dict(self._vars),
self._fridge.copy(),
self._recipe_revisions,
self._dyn_sous_files,
)
return dep_book
def get_j2_compatible_dep_var_proxies(
self, variables: Variables
) -> Dict[str, "DependencyVarProxy"]:
result = {}
for key, vars in variables.toplevel().items():
result[key] = DependencyVarProxy(self, vars, key + ".")
return result
class DependencyVarProxy:
"""
Provides convenient access to variables that also properly tracks
dependencies.
"""
def __init__(
self, dependency_tracker: DependencyTracker, variables: dict, prefix: str = ""
):
self._dvp_dt: DependencyTracker = dependency_tracker
self._dvp_prefix = prefix
self._dvp_vars = variables
def __getattr__(self, key: str):
fully_qualified_varname = self._dvp_prefix + key
value = self._dvp_vars.get(key, ...)
if value is ...:
raise KeyError(f"Variable does not exist: {fully_qualified_varname}")
elif isinstance(value, dict):
return DependencyVarProxy(self._dvp_dt, value, key + ".")
else:
self._dvp_dt.register_variable(fully_qualified_varname, value)