ssh: say when a probe is copying the kernel to a host, in the picker and the chat
albedo_ssh.py probe announces a staging step on its own line before it copies the bundle; albedo_ssh.erl passes lines on as they arrive and keeps the last as the answer. /hosts/:host carries step while warming, and the session status's kernel carries it while a remote kernel boots, so both read "copying the kernel to chernobog…" instead of "connecting to". Signed-off-by: dawn <90008@klbr.net>
120 unmodified linesExpand all121122123124125125126127128129130131131132133134135136137123 unmodified linesExpand all26126226326426326526626726826927027127215 unmodified lines120 unmodified linesExpand alla kernel is stale when it runs another bundle than the bridge that reached it,speaks another protocol (both read from the hello), or booted with anothermodule set than its session now has (found at reattach). `kernel.stale`answers the reason and whether the swap was forced; the session status carries`kernel: {stale, reason?, link}`, and the tui shows `kernel older` beside the`kernel: {stale, reason?, link, step?}`, and the tui shows `kernel older` beside theheader counts. `link` is how the session reaches its kernel: `none` before itneeded one, `booting` while one opens (or a daemon start's attach is awaited),`attached`, `reattaching` while the bridge is down and the port owner retries(`albedo_python:linked/1`), and `lost` once it gave up. the tui fades a remotehost in the header while booting or reattaching, and colors it as an erroronce lost.once lost. `step` is `staging` while a remote kernel's boot waits for itshost's probe to copy the bundle over, so the status line can say `copyingthe kernel to chernobog…` instead of `connecting to chernobog…`.
the swap happens at the session's next idle moment: `session_namespace.ready`,which every turn, compaction and background call goes through, lets go of akernel that `runtime.upgradable` says nothing keeps, so the session asks the123 unmodified linesExpand all to a minute.
both answer `{host, state, detail}` with `state` one of `ready`, `warming`,`needs_auth` (with `control_path`), `unreachable`, `unsupported`, and `os`,`arch`, `home` when ready. `/health` lists `remote_hosts`. `GET /hosts``arch`, `home` when ready. a `warming` answer carries `step: "staging"` whilethe probe copies the bundle: `albedo_ssh.py probe` prints `{"step":"staging"}` on its own line first, and `albedo_ssh.erl` passes each line onas it arrives (`albedo_ssh:step/1`), keeping the last line as the answer. `/health` lists `remote_hosts`. `GET /hosts`(the picker's host completion) is in workspaces.md. the same probe alsoanswers the `gather` command the folder browser and project readers run.
### signing in15 unmodified lines7 unmodified lines8910111213141516121314151617181920218 unmodified lines303132333435363738171 unmodified linesExpand all21021121221321221421521621721821922033 unmodified lines25425525625725825926026126242 unmodified lines30530630730830931031131231331430631531631731831912 unmodified lines7 unmodified lines albedo_ssh.py probe <[user@]host> albedo_ssh.py commands <[user@]host> <home> albedo_ssh.py hosts
`probe` prints one JSON object: `{state, detail, os, arch, home, cpus,python}` and, once the home is known, the `commands` answer: `{argv, control,bundle, bridge, signal, remove, gather, auth_sock}`, the complete remotecommands the daemon runs there. state is ready, needs_auth, unreachable or unsupported. Aready probe has checked python >= 3.11 and staged the content-hashed bundle`probe` prints `{"step": "staging"}` on a line of its own before it copiesthe bundle over, so a waiting client can say so, then one JSON object:`{state, detail, os, arch, home, cpus, python}` and, once the home is known,the `commands` answer: `{argv, control, bundle, bridge, signal, remove,gather, auth_sock}`, the complete remote commands the daemon runs there.state is ready, needs_auth, unreachable or unsupported. A ready probe has checked python >= 3.11 and staged the content-hashed bundleunder `~/.albedo-remote/<digest>`. `commands` needs no network, for a hostknown only from a kernel's record while ssh is down. `hosts` lists the `Host`names of ~/.ssh/config and its includes, patterns left out."""8 unmodified linesimport subprocessimport sysimport tarfileimport tempfilefrom collections.abc import Callable
import albedo_bundle
CONNECT_TIMEOUT = 10 # seconds ssh may take to reach a host171 unmodified linesExpand all 'test -f "$HOME/{bundle}/albedo_bridge.py" && echo staged || echo missing')
def probe(target: str) -> dict[str, object]:def probe( target: str, step: Callable[[str], None] = lambda _: None) -> dict[str, object]: answer: dict[str, object] = { "host": target, "argv": [*base(), target], "control": f"{control_dir()}/%C",33 unmodified lines "state": "unsupported", "detail": f"needs python >= 3.11 (found {python})", } if staged != "staged": step("staging") try: staging = run( target, in_login_shell(stage_script(f"$HOME/{bundle}")),42 unmodified lines names += config_hosts(included, seen) return list(dict.fromkeys(names))
def announce(step: str) -> None: print(json.dumps({"step": step}), flush=True)
def main(argv: list[str]) -> int: if len(argv) == 3 and argv[1] == "probe": print(json.dumps(probe(argv[2]))) print(json.dumps(probe(argv[2], announce))) return 0 if len(argv) == 2 and argv[1] == "hosts": print(json.dumps(config_hosts())) return 012 unmodified lines41 unmodified lines424344454647484950515240 unmodified lines939495969798991001018 unmodified lines1101111121131141151161171188 unmodified lines12712812913013113213313413560 unmodified lines196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227291 unmodified linesExpand all41 unmodified lines gone) echo "ssh: connect to host gone port 22: Connection refused" >&2; exit 255 ;; locked) echo "locked: Permission denied (publickey)." >&2; exit 255 ;; slow-*) sleep 3; echo "ssh: connect to host $target port 22: Connection refused" >&2; exit 255 ;;esaccase "$*" in *.part.*) [ -e {slow} ] && sleep 3 ;;esacpath={bin}:/usr/bin:/bin[ "$target" = oldhost ] && path={old}:$pathexec env -i HOME={home} PATH="$path" SHELL={bin}/loginsh /bin/sh -c "$*""""40 unmodified lines remote_home: Path ssh_log: Path python_log: Path down: Path slow_stage: Path
def setUp(self): self.cells = []
8 unmodified lines self.remote_home.mkdir() self.ssh_log = root / "ssh.log" self.python_log = root / "local-python.log" self.down = root / "fakehost-down" self.slow_stage = root / "slow-stage" real = shutil.which("python3") or sys.executable local, remote, old = ( root / "local-bin", root / "remote-bin",8 unmodified lines bin=remote, old=old, home=self.remote_home, down=self.down, slow=self.slow_stage, ), ) write( local / "python3", LOCAL_PYTHON.format(log=self.python_log, python=real)60 unmodified lines # The model is told where it runs. system = json.dumps(self.provider.requests[0]["request"]) self.assertIn("execute on fakehost (", system)
def test_a_first_visit_says_it_is_copying_the_kernel(self): self.slow_stage.touch() # each copy of the bundle takes 3 s session = self.create(f"fakehost:{self.project}")["id"] seen = set()
def watch(): deadline = time.monotonic() + 15 while time.monotonic() < deadline: with self.app.api(f"/sessions/{session}/status") as response: kernel = json.load(response)["kernel"] with self.app.api("/hosts/fakehost") as response: host = json.load(response) seen.update( {("session", kernel.get("step")), ("host", host.get("step"))} ) if kernel["link"] == "attached": return time.sleep(0.2)
result = self.cell(session, "1", during=watch) self.assertEqual(result["status"], "ok", result) self.assertIn(("session", "staging"), seen) self.assertIn(("host", "staging"), seen)
def test_a_bridge_killed_mid_cell_reattaches_over_ssh(self): session = self.create(f"fakehost:{self.project}")["id"] first = self.cell(session, "import os\nos.getpid()") (bridge,) = processes(f"{self.remote_home}/.albedo-remote/", "albedo_bridge.py")291 unmodified linesExpand all291 unmodified linesExpand all29229329429529629629729829930030130230330430530682 unmodified lines291 unmodified linesExpand all}
// HostStatus is what the daemon knows of a remote host over ssh: State is// ready, warming, needs_auth, unreachable or unsupported; Detail says why// when it is not ready. OS, Arch and Home come with a ready one.// when it is not ready. OS, Arch and Home come with a ready one; Step is// staging while a warming probe copies albedo's kernel to the host.type HostStatus struct { Host string `json:"host"` State string `json:"state"` Detail string `json:"detail"` Step string `json:"step"` OS string `json:"os"` Arch string `json:"arch"` Home string `json:"home"` ControlPath string `json:"control_path"`82 unmodified lines41 unmodified lines424344454647484950515227 unmodified lines8081828384858687888589909192933 unmodified lines41 unmodified lines KernelStale bool `json:"kernelStale,omitempty"` // KernelLink is how the session reaches its kernel: none, booting, // attached, reattaching or lost; empty from a daemon that does not say. KernelLink string `json:"kernelLink,omitempty"` // KernelStep is staging while a booting kernel waits for albedo to be // copied to its remote host. KernelStep string `json:"kernelStep,omitempty"`}
// ContextWindow reads the model's context window from the session's last// prepared request. It is nil when neither a catalog nor the configuration27 unmodified lines } var kernel struct { Stale bool `json:"stale"` Link string `json:"link"` Step string `json:"step"` } if err = required(fields, "kernel", &kernel); err != nil { return err } result.KernelStale, result.KernelLink = kernel.Stale, kernel.Link result.KernelStale, result.KernelLink, result.KernelStep = kernel.Stale, kernel.Link, kernel.Step return nil }) if err != nil { return nil, err3 unmodified lines2098 unmodified linesExpand all2099210021012102210321032104210521062107411 unmodified linesExpand all2098 unmodified linesExpand all } return "" } if host := m.reaching(); host != "" { return "connecting to " + host + "…" return reachingText(host, m.Status.KernelStep) } // the port owner gave up and forgot the kernel, so nothing reattaches if host, _ := daemon.SplitLocation(m.Workspace); host != "" && m.Status.KernelLink == "lost" { return "kernel on " + cmp.Or(m.Host, host) + " lost · the next turn starts a fresh one"411 unmodified linesExpand all867 unmodified linesExpand all868869870871872873874875876177 unmodified linesExpand all867 unmodified linesExpand all delete(m.probes, msg.Host) } return nil case msg.Status.State == "warming": m.probes[msg.Host] = msg.Status // it may say what the probe is doing return pollHost(msg.Host) } m.probes[msg.Host] = msg.Status if msg.Status.Home != "" {177 unmodified linesExpand all302 unmodified linesExpand all30330430530630730830931031131231331431531631731831932032132232313 unmodified lines302 unmodified linesExpand all t.Fatal("the face kept ticking after the host settled") }}
func TestPickerSaysWhenItCopiesTheKernelOver(t *testing.T) { src := remoteFolders("warming", "") src.probe.Step = "staging" m := NewFolderBrowser(src, "/Users/dawn/proj/albedo", "") m.SetSize(140, 16) m = settle(t, m, m.Init()) m = typeQuery(t, m, "cher") m, _ = m.Update(hostStatusMsg{Host: "mayer@chernobog", Status: src.probe}) if frame := pickerFrame(m); !strings.Contains(frame, "copying the kernel to chernobog… ") { t.Fatalf("a probe staging the bundle does not say so:\n%s", frame) }}
func TestPickerOffersSignInForAHostThatNeedsAPerson(t *testing.T) { src := remoteFolders("needs_auth", "Permission denied (publickey)") m := NewFolderBrowser(src, "/Users/dawn/proj/albedo", "") m.SetSize(100, 16)13 unmodified lines183 unmodified linesExpand all184185186187188188189190191192193204 unmodified linesExpand all398399400401401402403404405406205 unmodified linesExpand all183 unmodified linesExpand all return "no known hosts match; type host: to browse one" case m.listed == "": return "no matches" case l == nil && remote && m.probe(m.canonical(host)).State == "warming": return "connecting to " + m.labelOf(host) + "… " + DefaultStyles.Agent.Render(connectingFace(m.labelOf(host), m.frame)) label := m.labelOf(host) return reachingText(label, m.probe(m.canonical(host)).Step) + " " + DefaultStyles.Agent.Render(connectingFace(label, m.frame)) case l == nil: return "loading…" case l.Err != nil: return l.Err.Error()204 unmodified linesExpand all label := m.labelOf(row.hostKey) var body []string switch p.State { case "", "warming": body = []string{DefaultStyles.Faint.Render("connecting to "+label+"… ") + DefaultStyles.Agent.Render(connectingFace(label, m.frame))} body = []string{DefaultStyles.Faint.Render(reachingText(label, p.Step)+" ") + DefaultStyles.Agent.Render(connectingFace(label, m.frame))} case "ready": if p.OS != "" { body = append(body, DefaultStyles.Muted.Render(strings.TrimSpace(p.OS+" "+p.Arch))) }205 unmodified linesExpand all76 unmodified lines7778798081828384858687888913 unmodified lines76 unmodified lines m.Status = daemon.AgentStatus{Idle: true, KernelLink: "booting"} if got := m.statusLine(); got != "connecting to chernobog…" { t.Fatalf("a booting remote kernel reads %q", got) } m.Status.KernelStep = "staging" if got := m.statusLine(); got != "copying the kernel to chernobog…" { t.Fatalf("a first boot on the host reads %q", got) } m.Status.KernelStep = "" // an idle session animates too, with the face the picker shows for the host m.ProgressFrame = 1 if view := ansi.Strip(m.View()); !m.animating() || !strings.Contains(view, "connecting to chernobog… "+connectingFace("chernobog", 1)) { t.Fatalf("connecting should animate (animating %v):\n%s", m.animating(), view)13 unmodified lines91 unmodified lines929394959697989910010110210310410510610710810941 unmodified lines91 unmodified lines set := pool[m.pick(seed, len(pool))] return set[tick%len(set)]}
// reachingText is what waiting on a host reads as, by its label: connecting,// or copying albedo's kernel over (step staging), which the first visit to a// host, and the first after albedo's python changed, waits for.func reachingText(label, step string) string { if step == "staging" { return "copying the kernel to " + label + "…" } return "connecting to " + label + "…"}
// connectingFace is the animation for reaching a host, by its label, at// tick: the same one for a host wherever it shows.func connectingFace(host string, tick int) string { return moodConnecting.frame(int64(fnv1a(fnvOffset64, host)), tick)41 unmodified lines1 unmodified line23456789101112200 unmodified linesExpand all213214215216215217218219220221222223224225226227218219220221222223224225228229230231232233234235236237238239240241242243244106 unmodified linesExpand all1 unmodified line
import albedo/daemon/events as viewimport albedo/daemon/session_stateimport albedo/harness/extensions/python/kernel as pythonimport albedo/harness/locationimport albedo/harness/runtimeimport albedo/harness/sshimport gleam/intimport gleam/jsonimport gleam/listimport gleam/option.{type Option, None, Some}200 unmodified linesExpand all missed -> ", " <> int.to_string(list.length(missed)) <> " not carried" }}
/// The kernel for the session status: `{stale, reason?, link}`./// The kernel for the session status: `{stale, reason?, link, step?}`./// `step` is `staging` while a kernel booting on a remote host waits for/// albedo's bundle to be copied there.pub fn kernel_json(state: session_state.State(message)) -> json.Json { let reason = option.then(state.kernel, runtime.stale) let link = link_name(state) let step = case link, location.parse(state.info.cwd) { "booting", Ok(location.Remote(..) as at) -> location.ssh_target(at) |> result.map(ssh.step) |> result.unwrap("") _, _ -> "" } json.object([ #("stale", json.bool(reason != None)), #("link", json.string(link_name(state))), ..case reason { Some(reason) -> [#("reason", json.string(stale_name(reason)))] None -> [] } ]) json.object( list.flatten([ [#("stale", json.bool(reason != None)), #("link", json.string(link))], case reason { Some(reason) -> [#("reason", json.string(stale_name(reason)))] None -> [] }, case step { "" -> [] step -> [#("step", json.string(step))] }, ]), )}
/// How the session reaches its kernel: `booting` while one opens (a boot, or/// waiting for a daemon start to attach the recorded one), `attached`,106 unmodified linesExpand all5 unmodified lines678910101112131412 unmodified lines27282930313233343536373839 unmodified lines787980817882838485863 unmodified lines909192939495969798991 unmodified line101102103104991051061071081091041101111121131141151161171181191207 unmodified lines1281291301311201321331341351361 unmodified line1381391401411301311321421431441451461471481491501511521531541551561571581591361371601611621631641651661671681691701711451721731741751761771511781791801811821831531841851861871881891901911921936 unmodified lines5 unmodified lines%% A ready answer is kept for a minute and a failure for five seconds; asking%% again while a probe runs joins it instead of starting another.-module(albedo_ssh).-behaviour(gen_server).-export([probe/2, peek/1, cached/1, forget/1, commands/2, exec/5, config_hosts/0]).-export([probe/2, peek/1, cached/1, step/1, forget/1, commands/2, exec/5, config_hosts/0]).-export([init/1, handle_call/3, handle_cast/2, handle_info/2]).
-define(READY_MS, 60000).-define(FAILED_MS, 5000).12 unmodified lines
%% A fresh cached answer, never starting a probe.cached(Target) -> gen_server:call(ensure(), {cached, Target}).
%% What a running probe is doing past connecting: <<"staging">> while it%% copies the bundle over, <<>> otherwise.step(Target) -> gen_server:call(ensure(), {step, Target}).
%% The Host names of ~/.ssh/config and its includes, as a JSON list.config_hosts() -> case helper([<<"hosts">>], 5000) of {ok, Json} -> {ok, Json};39 unmodified lines {ok, Pid} -> Pid; {error, {already_started, Pid}} -> Pid end.
init([]) -> {ok, #{hosts => #{}, running => #{}}}.init([]) -> {ok, #{hosts => #{}, running => #{}, steps => #{}}}.
handle_call({probe, Target}, From, S) -> case fresh(Target, S) of {ok, Json} -> {reply, {ok, Json}, S};3 unmodified lines case fresh(Target, S) of {ok, Json} -> {reply, {ok, Json}, S}; none -> {reply, {error, nil}, S} end;handle_call({step, Target}, _, S = #{steps := Steps}) -> {reply, maps:get(Target, Steps, <<>>), S};handle_call({peek, Target}, _, S) -> case fresh(Target, S) of {ok, Json} -> {reply, {ok, Json}, S}; none -> {reply, {error, nil}, start(Target, none, S)}1 unmodified line
handle_cast({forget, Target}, S = #{hosts := Hosts}) -> {noreply, S#{hosts => maps:remove(Target, Hosts)}}.
handle_info({probed, Target, Json}, S = #{hosts := Hosts, running := Running}) ->handle_info({probed, Target, Json}, S = #{hosts := Hosts, running := Running, steps := Steps}) -> Waiters = maps:get(Target, Running, []), [gen_server:reply(From, {ok, Json}) || From <- Waiters], Now = erlang:monotonic_time(millisecond), {noreply, S#{hosts => Hosts#{Target => {Json, Now + ttl(Json)}}, running => maps:remove(Target, Running)}}; running => maps:remove(Target, Running), steps => maps:remove(Target, Steps)}};handle_info({stepped, Target, Step}, S = #{running := Running, steps := Steps}) -> case maps:is_key(Target, Running) of true -> {noreply, S#{steps => Steps#{Target => Step}}}; false -> {noreply, S} end;handle_info(_, S) -> {noreply, S}.
fresh(Target, #{hosts := Hosts}) -> Now = erlang:monotonic_time(millisecond),7 unmodified lines case maps:find(Target, Running) of {ok, Earlier} -> S#{running => Running#{Target => Waiters ++ Earlier}}; error -> Self = self(), spawn(fun() -> Self ! {probed, Target, run(Target)} end), spawn(fun() -> Self ! {probed, Target, run(Target, Self)} end), S#{running => Running#{Target => Waiters}} end.
ttl(Json) ->1 unmodified line #{<<"state">> := <<"ready">>} -> ?READY_MS; _ -> ?FAILED_MS end.
run(Target) -> case helper([<<"probe">>, Target], ?HELPER_MS) of {ok, Json} -> Json;%% The probe's answer is its last line; a step line before it is passed on%% to the server as it arrives.run(Target, Server) -> Stepped = fun(Line) -> case catch json:decode(Line) of #{<<"step">> := Step} when is_binary(Step) -> Server ! {stepped, Target, Step}; _ -> ok end end, case helper([<<"probe">>, Target], ?HELPER_MS, Stepped) of {ok, Out} -> case [L || L <- binary:split(Out, <<"\n">>, [global]), L =/= <<>>] of [] -> failed("albedo_ssh.py answered nothing"); Lines -> lists:last(Lines) end; {error, Why} -> failed(Why) end.
%% One run of albedo_ssh.py: its stdout, or why there is none.helper(Args, Wait) ->%% One run of albedo_ssh.py: its stdout, or why there is none. Given an%% OnLine, each line goes to it as soon as it is complete.helper(Args, Wait) -> helper(Args, Wait, none).
helper(Args, Wait, OnLine) -> case albedo_python:local_paths() of {ok, {Python, Kernel}} -> Helper = filename:join(filename:dirname(Kernel), <<"albedo_ssh.py">>), try open_port({spawn_executable, binary_to_list(Python)}, [binary, exit_status, use_stdio, hide, {args, ["-u", binary_to_list(Helper) | [binary_to_list(A) || A <- Args]]}, {env, albedo_python:clean_environment()}]) of Port -> collect(Port, [], erlang:monotonic_time(millisecond) + Wait) Port -> collect(Port, [], erlang:monotonic_time(millisecond) + Wait, OnLine, <<>>) catch _:Reason -> {error, io_lib:format("~p", [Reason])} end; {error, {unavailable, Why}} -> {error, Why} end.
collect(Port, Acc, Deadline) ->collect(Port, Acc, Deadline) -> collect(Port, Acc, Deadline, none, <<>>).
%% Without OnLine nothing is split, so a large one-line answer (a gather)%% is never copied again per chunk.collect(Port, Acc, Deadline, OnLine, Partial) -> receive {Port, {data, Data}} -> collect(Port, [Data | Acc], Deadline); {Port, {data, Data}} when OnLine =:= none -> collect(Port, [Data | Acc], Deadline, none, Partial); {Port, {data, Data}} -> [Rest | Lines] = lists:reverse(binary:split(<<Partial/binary, Data/binary>>, <<"\n">>, [global])), lists:foreach(OnLine, lists:reverse(Lines)), collect(Port, [Data | Acc], Deadline, OnLine, Rest); {Port, {exit_status, 0}} -> {ok, iolist_to_binary(lists:reverse(Acc))}; {Port, {exit_status, Status}} -> {error, io_lib:format("albedo_ssh.py exited ~p", [Status])} after max(0, Deadline - erlang:monotonic_time(millisecond)) -> _ = try port_close(Port) catch _:_ -> ok end,6 unmodified lines