Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions agentplane/runner_gateway/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,7 @@ go_library(
"//agentplane/runner_gateway/environments",
"//agentplane/runner_gateway/environments/lifecycle",
"//agentplane/runner_gateway/internal/gateway",
"//agentplane/runner_gateway/internal/sandboxinventory",
"//agentplane/runner_gateway/migrations",
"//agentplane/sandboxes/api",
"//agentplane/sandboxes/runtimeconfig",
Expand Down
1 change: 1 addition & 0 deletions agentplane/runner_gateway/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ conversation and from Runner admission.
`CreateEnvironment` names the Identity the Sandbox acts as and binds no PolicySets: a launch
creates the Identity and grants it, including the default sets, before calling it.
- `internal/gateway/`: Runner connections, authenticated proxy, Sandbox/Pod validation and connection ownership.
- `internal/sandboxinventory/`: informer-backed Sandbox reads and the uncached Identity reader that supply the Dispatcher contract.
- `dispatcher/`: durable admission, watch, delivery, guarded wake and scheduling ownership.
- `migrations/`: Goose migrations for the Dispatcher database.
- `client.py`: the Python Runner proxy client; `testing/` provides narrow consumer peers and
Expand Down
21 changes: 0 additions & 21 deletions agentplane/runner_gateway/dispatcher/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ go_library(
"delivery_wake.go",
"ownership.go",
"runner_client.go",
"sandbox_kubernetes.go",
"snapshots.go",
"store.go",
"terminal.go",
Expand Down Expand Up @@ -43,19 +42,15 @@ go_library(
"//agentplane/runner_gateway/environments/lifecycle",
"//agentplane/sandboxes/api",
"//agentplane/sandboxes/environmentruntimeclient",
"//agentplane/sandboxes/sandboxreader",
"@com_github_google_uuid//:uuid",
"@com_github_lib_pq//:pq",
"@io_k8s_api//coordination/v1:coordination",
"@io_k8s_apimachinery//pkg/api/errors",
"@io_k8s_apimachinery//pkg/api/meta",
"@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
"@io_k8s_apimachinery//pkg/types",
"@io_k8s_apimachinery//pkg/util/sets",
"@io_k8s_apimachinery//pkg/watch",
"@io_k8s_client_go//tools/cache",
"@io_k8s_client_go//util/workqueue",
"@io_k8s_sigs_controller_runtime//pkg/cache",
"@io_k8s_sigs_controller_runtime//pkg/client",
"@io_k8s_utils//ptr",
"@io_opentelemetry_go_otel//attribute",
Expand All @@ -71,22 +66,6 @@ go_library(
],
)

go_test(
name = "sandbox_kubernetes_test",
srcs = ["sandbox_kubernetes_test.go"],
embed = [":dispatcher"],
deps = [
"//agentplane/sandboxes/api",
"@io_k8s_apimachinery//pkg/api/errors",
"@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
"@io_k8s_apimachinery//pkg/runtime",
"@io_k8s_apimachinery//pkg/runtime/schema",
"@io_k8s_sigs_controller_runtime//pkg/client",
"@io_k8s_sigs_controller_runtime//pkg/client/fake",
"@io_k8s_sigs_controller_runtime//pkg/client/interceptor",
],
)

go_test(
name = "methods_test",
srcs = ["methods_test.go"],
Expand Down
32 changes: 32 additions & 0 deletions agentplane/runner_gateway/internal/sandboxinventory/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")

go_library(
name = "sandboxinventory",
srcs = ["sandbox_kubernetes.go"],
importpath = "github.com/agentydragon/ducktape/agentplane/runner_gateway/internal/sandboxinventory",
visibility = ["//agentplane/runner_gateway:__pkg__"],
deps = [
"//agentplane/proto:identity_go_proto",
"//agentplane/runner_gateway/dispatcher",
"//agentplane/sandboxes/api",
"//agentplane/sandboxes/sandboxreader",
"@io_k8s_apimachinery//pkg/api/errors",
"@io_k8s_apimachinery//pkg/api/meta",
"@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
"@io_k8s_apimachinery//pkg/types",
"@io_k8s_client_go//tools/cache",
"@io_k8s_sigs_controller_runtime//pkg/cache",
"@io_k8s_sigs_controller_runtime//pkg/client",
],
)

go_test(
name = "sandboxinventory_test",
srcs = ["sandbox_kubernetes_test.go"],
embed = [":sandboxinventory"],
deps = [
"//agentplane/sandboxes/api",
"@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
"@io_k8s_apimachinery//pkg/types",
],
)
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
package dispatcher
package sandboxinventory

import (
"context"
"errors"
"fmt"

identitypb "github.com/agentydragon/ducktape/agentplane/proto/identity"
"github.com/agentydragon/ducktape/agentplane/runner_gateway/dispatcher"
sandboxapi "github.com/agentydragon/ducktape/agentplane/sandboxes/api"
"github.com/agentydragon/ducktape/agentplane/sandboxes/sandboxreader"
apierrors "k8s.io/apimachinery/pkg/api/errors"
Expand Down Expand Up @@ -68,37 +69,37 @@ func (k *KubernetesSandboxes) WaitSynced(ctx context.Context) bool {
return k.inventory.WaitForCacheSync(ctx)
}

func (k *KubernetesSandboxes) GetSandbox(ctx context.Context, name string) (SandboxSnapshot, bool, error) {
func (k *KubernetesSandboxes) GetSandbox(ctx context.Context, name string) (dispatcher.SandboxSnapshot, bool, error) {
if !k.informer.HasSynced() {
return SandboxSnapshot{}, false, errors.New("Sandbox inventory is not synchronized")
return dispatcher.SandboxSnapshot{}, false, errors.New("Sandbox inventory is not synchronized")
}
var object sandboxapi.Sandbox
err := k.inventory.Get(ctx, types.NamespacedName{Namespace: k.namespace, Name: name}, &object)
if apierrors.IsNotFound(err) {
return SandboxSnapshot{}, false, nil
return dispatcher.SandboxSnapshot{}, false, nil
}
if err != nil {
return SandboxSnapshot{}, false, err
return dispatcher.SandboxSnapshot{}, false, err
}
view, err := sandboxSnapshot(&object)
return view, err == nil, err
}

// GetExecutionIdentity returns the Identity the exact live Sandbox UID names, read uncached so
// delivery never authorizes from a stale informer snapshot.
func (k *KubernetesSandboxes) GetExecutionIdentity(ctx context.Context, sandbox SandboxSnapshot) (*identitypb.Identity, bool, error) {
func (k *KubernetesSandboxes) GetExecutionIdentity(ctx context.Context, sandbox dispatcher.SandboxSnapshot) (*identitypb.Identity, bool, error) {
return k.reader.ExecutionIdentity(ctx, &sandboxreader.Identity{Namespace: sandbox.Namespace, Name: sandbox.Name, Uid: sandbox.UID})
}

func (k *KubernetesSandboxes) ListSandboxes(ctx context.Context) ([]SandboxSnapshot, error) {
func (k *KubernetesSandboxes) ListSandboxes(ctx context.Context) ([]dispatcher.SandboxSnapshot, error) {
if !k.informer.HasSynced() {
return nil, errors.New("Sandbox inventory is not synchronized")
}
var objects sandboxapi.SandboxList
if err := k.inventory.List(ctx, &objects, client.InNamespace(k.namespace)); err != nil {
return nil, err
}
result := make([]SandboxSnapshot, 0, len(objects.Items))
result := make([]dispatcher.SandboxSnapshot, 0, len(objects.Items))
for i := range objects.Items {
view, err := sandboxSnapshot(&objects.Items[i])
if err != nil {
Expand Down Expand Up @@ -126,14 +127,14 @@ func (k *KubernetesSandboxes) CurrentSandboxUIDs(ctx context.Context) (map[strin
return result, nil
}

func sandboxSnapshot(object *sandboxapi.Sandbox) (SandboxSnapshot, error) {
func sandboxSnapshot(object *sandboxapi.Sandbox) (dispatcher.SandboxSnapshot, error) {
if object.Name == "" || object.UID == "" {
return SandboxSnapshot{}, errors.New("Sandbox has no name or UID")
return dispatcher.SandboxSnapshot{}, errors.New("Sandbox has no name or UID")
}
if object.Namespace == "" {
return SandboxSnapshot{}, errors.New("Sandbox has no namespace")
return dispatcher.SandboxSnapshot{}, errors.New("Sandbox has no namespace")
}
view := SandboxSnapshot{
view := dispatcher.SandboxSnapshot{
Name: object.Name, Namespace: object.Namespace, UID: string(object.UID),
Deleting: object.DeletionTimestamp != nil,
WakeOnSend: object.Spec.WakeOnSend,
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
package dispatcher
package sandboxinventory

import (
"testing"
Expand Down
3 changes: 2 additions & 1 deletion agentplane/runner_gateway/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import (
"github.com/agentydragon/ducktape/agentplane/runner_gateway/environments"
"github.com/agentydragon/ducktape/agentplane/runner_gateway/environments/lifecycle"
"github.com/agentydragon/ducktape/agentplane/runner_gateway/internal/gateway"
"github.com/agentydragon/ducktape/agentplane/runner_gateway/internal/sandboxinventory"
"github.com/agentydragon/ducktape/agentplane/runner_gateway/migrations"
sandboxapi "github.com/agentydragon/ducktape/agentplane/sandboxes/api"
"github.com/agentydragon/ducktape/agentplane/watchhealth"
Expand Down Expand Up @@ -239,7 +240,7 @@ func serve(ctx context.Context, c config, rpcListener net.Listener) error {
if err != nil {
return err
}
catalog, err := dispatcher.NewKubernetesSandboxes(inventory, objects, c.Namespace)
catalog, err := sandboxinventory.NewKubernetesSandboxes(inventory, objects, c.Namespace)
if err != nil {
return err
}
Expand Down
Loading