amadan

carlos / platform Public

activator: deduplicated child env, the Control seam, the per-role log listener (plan 3, task 3)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 33d6062486ae7722d2ebeb592988631ac470ff09 parent 10df49b
8 files changed, +266 −14
  • internal/activator/activator_test.go +19 −0
  • internal/activator/backend_exec.go +32 −9
  • internal/activator/control.go +68 −0
  • internal/activator/loglistener.go +77 −0
  • internal/activator/loglistener_test.go +65 −0
  • internal/edge/purge.go +1 −1
  • internal/edge/purge_listen_unix.go +2 −2
  • internal/edge/purge_listen_windows.go +2 −2

CI passed — run details

diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 8127d94f1..0a9075823 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -3995,3 +3995,22 @@ func TestChildProcStopClosesTheControlSocketWhenTheChildIsAlreadyGone(t *testing
t.Fatalf("control socket after Stop on an exited child: stat err = %v, want it unlinked", err)
}
}
+
+func TestChildEnvKeepsOneEntryPerKeyLastWins(t *testing.T) {
+ t.Setenv("PATH", "/usr/bin")
+ env := childEnv(map[string]string{"PATH": "/tenant/bin", "A": "1"})
+ seen := map[string]string{}
+ for _, kv := range env {
+ k, v, _ := strings.Cut(kv, "=")
+ if _, dup := seen[k]; dup {
+ t.Fatalf("duplicate key %s in %v", k, env)
+ }
+ seen[k] = v
+ }
+ if seen["PATH"] != "/tenant/bin" {
+ t.Errorf("PATH = %q, want the tenant's value to win over the inherited one", seen["PATH"])
+ }
+ if seen["A"] != "1" {
+ t.Errorf("A = %q", seen["A"])
+ }
+}
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index e6c46dc00..06c7a4573 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -14,6 +14,7 @@ import (
"os"
"os/exec"
"path/filepath"
+ "sort"
"strings"
"sync"
"syscall"
@@ -789,21 +790,43 @@ func (b *ExecBackend) instanceEnv(rt registry.Route) (map[string]string, error)
return mergeEnv(cfgEnv, ownEnv), nil
}
-// childEnv approximates a systemd unit's environment from inside the agent:
-// inherit the process env but strip every CARLOS_* var first, so the agent's
-// own config (hibernation, registry, tokens) can never leak into a child;
-// then layer the child's EnvironmentFile on top. A nil fileEnv (no env file,
-// like the unit's "-" prefix) yields just the cleaned inherit.
+// childEnv is the child's complete environment: the agent's own, minus
+// every CARLOS_* variable, with the instance's file values layered on top.
+// One entry per key, last value wins: os/exec deduplicates cmd.Env that
+// way, but the unit path hands this list to syscall.Exec, which passes
+// duplicates through, and a Go child then keeps the FIRST occurrence
+// (design §6.2, codex R1-16). Composing it deduplicated here keeps the
+// two spawn paths identical.
func childEnv(fileEnv map[string]string) []string {
- env := make([]string, 0, 32)
+ order := make([]string, 0, 32)
+ values := make(map[string]string, 32)
+ set := func(k, v string) {
+ if _, seen := values[k]; !seen {
+ order = append(order, k)
+ }
+ values[k] = v
+ }
for _, kv := range os.Environ() {
if strings.HasPrefix(kv, "CARLOS_") {
continue
}
- env = append(env, kv)
+ k, v, ok := strings.Cut(kv, "=")
+ if !ok {
+ continue
+ }
+ set(k, v)
+ }
+ keys := make([]string, 0, len(fileEnv))
+ for k := range fileEnv {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, k := range keys {
+ set(k, fileEnv[k])
}
- for k, v := range fileEnv {
- env = append(env, k+"="+v)
+ env := make([]string, 0, len(order))
+ for _, k := range order {
+ env = append(env, k+"="+values[k])
}
return env
}
diff --git a/internal/activator/control.go b/internal/activator/control.go
new file mode 100644
index 000000000..3fe7bf79c
--- /dev/null
+++ b/internal/activator/control.go
@@ -0,0 +1,68 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/instancectl"
+)
+
+// Control is what the exec backend needs from the root instance-control
+// channel (design §5.3). instancectl.Client is the production
+// implementation; tests use a fake.
+type Control interface {
+ Start(ctx context.Context, host string, role instancectl.Role, memoryMax int64) (string, error)
+ Seed(ctx context.Context, host string, memoryMax int64) (instancectl.SeedResponse, error)
+ Stop(ctx context.Context, host string, role instancectl.Role) error
+ Reset(ctx context.Context, host string, role instancectl.Role) error
+ Sweep(ctx context.Context) (instancectl.SweepResponse, error)
+ Ready(ctx context.Context) error
+}
+
+var _ Control = (*instancectl.Client)(nil)
+
+// UnitState answers `systemctl show -p ActiveState,InvocationID <unit>`,
+// which needs no privilege. An unloaded unit answers inactive with an
+// empty invocation.
+type UnitState func(ctx context.Context, unit string) (active, invocation string, err error)
+
+func systemctlUnitState(ctx context.Context, unit string) (string, string, error) {
+ out, err := exec.CommandContext(ctx, "systemctl", "show", "-p", "ActiveState,InvocationID", unit).Output()
+ if err != nil {
+ return "", "", err
+ }
+ var active, inv string
+ for _, line := range strings.Split(string(out), "\n") {
+ if k, v, ok := strings.Cut(line, "="); ok {
+ switch k {
+ case "ActiveState":
+ active = strings.TrimSpace(v)
+ case "InvocationID":
+ inv = strings.TrimSpace(v)
+ }
+ }
+ }
+ return active, inv, nil
+}
+
+// CgroupEvents reads the unit's cgroup.events under the host slice. cgroup
+// v2 files are world-readable, so the agent reads them as carlos, as the
+// sampler does. An absent directory is the unit's cgroup already removed.
+type CgroupEvents func(slice, unit string) (exists, populated bool, err error)
+
+func cgroupEventsUnder(root string) CgroupEvents {
+ return func(slice, unit string) (bool, bool, error) {
+ b, err := os.ReadFile(filepath.Join(root, "carlos.slice", "carlos-instances.slice", slice, unit, "cgroup.events"))
+ if errors.Is(err, os.ErrNotExist) {
+ return false, false, nil
+ }
+ if err != nil {
+ return true, true, err
+ }
+ return true, !strings.Contains(string(b), "populated 0"), nil
+ }
+}
diff --git a/internal/activator/loglistener.go b/internal/activator/loglistener.go
new file mode 100644
index 000000000..81c1eaf40
--- /dev/null
+++ b/internal/activator/loglistener.go
@@ -0,0 +1,77 @@
+package activator
+
+import (
+ "bufio"
+ "io"
+ "net"
+ "os"
+ "sync"
+ "sync/atomic"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/edge"
+)
+
+// logListener is the agent's end of a unit's stdout (design §6.4): the
+// unit's StandardOutput=file: connects to this AF_UNIX socket once, at
+// exec, and every line is teed to the agent's own stdout (journald) and to
+// the app log sink, exactly as teeStd does for an exec child. One
+// connection per invocation is accepted; a second is closed unread. EOF
+// ends log collection and nothing else: it is never a sign the process is
+// gone (codex R1-4).
+type logListener struct {
+ ln net.Listener
+ emit func(string)
+ accepted atomic.Bool
+ closed atomic.Bool
+ wg sync.WaitGroup
+}
+
+func listenLogs(path string, emit func(string)) (*logListener, error) {
+ ln, err := edge.SecureUnixListen(path)
+ if err != nil {
+ return nil, err
+ }
+ l := &logListener{ln: ln, emit: emit}
+ l.wg.Add(1)
+ go l.accept()
+ return l, nil
+}
+
+func (l *logListener) accept() {
+ defer l.wg.Done()
+ for {
+ c, err := l.ln.Accept()
+ if err != nil {
+ return
+ }
+ if l.accepted.Swap(true) {
+ c.Close()
+ continue
+ }
+ go l.tee(c)
+ }
+}
+
+func (l *logListener) tee(c net.Conn) {
+ defer c.Close()
+ var sink io.Writer = os.Stdout
+ if l.emit != nil {
+ sink = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, l.emit))
+ }
+ r := bufio.NewReaderSize(c, execLogLineMax)
+ _, _ = io.Copy(sink, r)
+}
+
+// Exited always reports false: a log listener knows nothing about process
+// state, and this method exists so a caller cannot mistake it for one.
+func (l *logListener) Exited() bool { return false }
+
+func (l *logListener) Close() error {
+ if l.closed.Swap(true) {
+ return nil
+ }
+ err := l.ln.Close()
+ l.wg.Wait()
+ return err
+}
diff --git a/internal/activator/loglistener_test.go b/internal/activator/loglistener_test.go
new file mode 100644
index 000000000..f1c25e73e
--- /dev/null
+++ b/internal/activator/loglistener_test.go
@@ -0,0 +1,65 @@
+package activator
+
+import (
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+func TestLogListenerTeesOneConnectionAndRefusesASecond(t *testing.T) {
+ dir, err := os.MkdirTemp("", "ll")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ path := filepath.Join(dir, "h.run.sock")
+ var mu sync.Mutex
+ var lines []string
+ l, err := listenLogs(path, func(s string) { mu.Lock(); lines = append(lines, s); mu.Unlock() })
+ if err != nil {
+ t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
+ }
+ defer l.Close()
+ c, err := net.Dial("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := c.Write([]byte("hello\nworld\n")); err != nil {
+ t.Fatal(err)
+ }
+ c.Close()
+ deadline := time.Now().Add(2 * time.Second)
+ for {
+ mu.Lock()
+ n := len(lines)
+ mu.Unlock()
+ if n == 2 || time.Now().After(deadline) {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ mu.Lock()
+ got := strings.Join(lines, "|")
+ mu.Unlock()
+ if got != "hello|world" {
+ t.Fatalf("lines = %q", got)
+ }
+ // A second connection during the same invocation is closed unread.
+ c2, err := net.Dial("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c2.SetReadDeadline(time.Now().Add(time.Second))
+ buf := make([]byte, 1)
+ if _, err := c2.Read(buf); err == nil {
+ t.Fatal("second connection was not closed")
+ }
+ c2.Close()
+ if l.Exited() {
+ t.Error("EOF on the log connection must not mark the process exited")
+ }
+}
diff --git a/internal/edge/purge.go b/internal/edge/purge.go
index a36b3a568..50b4558c2 100644
--- a/internal/edge/purge.go
+++ b/internal/edge/purge.go
@@ -106,7 +106,7 @@ type purgeRequest struct {
// ceiling alone, and their once-timers go through the console rather than
// the socket, until that gets its own design.
func ListenControl(socketPath, host string, cache *Cache, mount func(*http.ServeMux)) (*ControlListener, error) {
- ln, err := secureUnixListen(socketPath)
+ ln, err := SecureUnixListen(socketPath)
if err != nil {
return nil, err
}
diff --git a/internal/edge/purge_listen_unix.go b/internal/edge/purge_listen_unix.go
index 5f37bb1f7..d4a8cdb3b 100644
--- a/internal/edge/purge_listen_unix.go
+++ b/internal/edge/purge_listen_unix.go
@@ -17,7 +17,7 @@ import (
// before being chmod'd to 0600.
var listenMu sync.Mutex
-// secureUnixListen binds socketPath as a unix socket reachable only by its
+// SecureUnixListen binds socketPath as a unix socket reachable only by its
// owner, removing any stale file left behind by a crashed previous instance.
//
// The owner-only guarantee is the whole point of the function and the reason
@@ -25,7 +25,7 @@ var listenMu sync.Mutex
// that anyone on the box can connect to hands out exactly the wake-storm
// lever this feature exists to remove. Making that guarantee unrepresentable
// on an OS without umask is safer than making it a no-op there.
-func secureUnixListen(socketPath string) (net.Listener, error) {
+func SecureUnixListen(socketPath string) (net.Listener, error) {
if err := os.Remove(socketPath); err != nil && !errors.Is(err, os.ErrNotExist) {
return nil, err
}
diff --git a/internal/edge/purge_listen_windows.go b/internal/edge/purge_listen_windows.go
index c7946c6b1..d5e62b0fe 100644
--- a/internal/edge/purge_listen_windows.go
+++ b/internal/edge/purge_listen_windows.go
@@ -7,7 +7,7 @@ import (
"net"
)
-// secureUnixListen has no Windows implementation, and that is deliberate.
+// SecureUnixListen has no Windows implementation, and that is deliberate.
//
// Windows is a CLIENT target for carlos: the CLI is the same binary as the
// box agent, so the whole edge package must COMPILE there, but no part of it
@@ -19,6 +19,6 @@ import (
// internal/activator/backend_exec.go, which treats a failure here as "no
// control listener": the instance runs without its control verbs and
// CARLOS_CONTROL_SOCKET is stripped from its environment.
-func secureUnixListen(socketPath string) (net.Listener, error) {
+func SecureUnixListen(socketPath string) (net.Listener, error) {
return nil, errors.New("control listeners need a unix socket with owner-only permissions; not supported on windows")
}