Sandbox-manager support filtering sandbox-related custom resources

cmd/sandbox-manager/main.go

@@ -34,6 +34,8 @@ func main() {
 	var e2bMaxTimeout int
 	var sysNs string
 	var peerSelector string
+	var sandboxNamespace string
+	var sandboxLabelSelector string
 	var maxClaimWorkers int
 	var maxCreateQPS int
 	var extProcMaxConcurrency int
@@ -54,6 +56,8 @@ func main() {
 	pflag.IntVar(&e2bMaxTimeout, "e2b-max-timeout", models.DefaultMaxTimeout, "E2B maximum timeout in seconds")
 	pflag.StringVar(&sysNs, "system-namespace", utils.DefaultSandboxDeployNamespace, "The namespace where the sandbox manager is running (required)")
 	pflag.StringVar(&peerSelector, "peer-selector", "", "Peer selector for sandbox manager (required)")
+	pflag.StringVar(&sandboxNamespace, "sandbox-namespace", "", "Namespace to filter sandbox-related custom resources (Sandbox, SandboxSet, Checkpoint, SandboxTemplate). Defaults to all.")
+	pflag.StringVar(&sandboxLabelSelector, "sandbox-label-selector", "", "Label selector to filter sandbox-related custom resources (Sandbox, SandboxSet, Checkpoint, SandboxTemplate). Defaults to all.")
 	pflag.IntVar(&maxClaimWorkers, "max-claim-workers", consts.DefaultClaimWorkers, "Maximum number of claim workers (0 uses default)")
 	pflag.IntVar(&maxCreateQPS, "max-create-qps", consts.DefaultCreateQPS, "Maximum QPS for sandbox creation (0 uses default)")
 	pflag.IntVar(&extProcMaxConcurrency, "ext-proc-max-concurrency", consts.DefaultExtProcConcurrency, "Maximum concurrency for external processor (0 uses default)")
@@ -129,7 +133,7 @@ func main() {
 		klog.Fatalf("Failed to initialize Kubernetes client: %v", err)
 	}
 
-	sandboxController := e2b.NewController(domain, e2bAdminKey, sysNs, e2bMaxTimeout, maxClaimWorkers, maxCreateQPS, uint32(extProcMaxConcurrency),
+	sandboxController := e2b.NewController(domain, e2bAdminKey, sysNs, sandboxNamespace, sandboxLabelSelector, e2bMaxTimeout, maxClaimWorkers, maxCreateQPS, uint32(extProcMaxConcurrency),
 		port, e2bEnableAuth, clientSet)
 	if err := sandboxController.Init(); err != nil {
 		klog.Fatalf("Failed to initialize sandbox controller: %v", err)

pkg/sandbox-manager/config/manager.go

@@ -7,6 +7,8 @@ import (
 
 type SandboxManagerOptions struct {
 	SystemNamespace       string
+	SandboxNamespace      string
+	SandboxLabelSelector  string
 	MaxClaimWorkers       int
 	MaxCreateQPS          int
 	ExtProcMaxConcurrency uint32

pkg/sandbox-manager/infra/sandboxcr/cache.go

@@ -9,6 +9,7 @@ import (
 
 	"golang.org/x/sync/singleflight"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
@@ -45,7 +46,16 @@ type Cache struct {
 
 func NewCache(client *clients.ClientSet, opts config.SandboxManagerOptions) (*Cache, error) {
 	// Create informer factory for custom Sandbox resources
-	informerFactory := informers.NewSharedInformerFactory(client.SandboxClient, time.Minute*10)
+	informerOptions := []informers.SharedInformerOption{}
+	if opts.SandboxNamespace != "" {
+		informerOptions = append(informerOptions, informers.WithNamespace(opts.SandboxNamespace))
+	}
+	if opts.SandboxLabelSelector != "" {
+		informerOptions = append(informerOptions, informers.WithTweakListOptions(func(lo *metav1.ListOptions) {
+			lo.LabelSelector = opts.SandboxLabelSelector
+		}))
+	}
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(client.SandboxClient, time.Minute*10, informerOptions...)
 	sandboxInformer := informerFactory.Api().V1alpha1().Sandboxes().Informer()
 	sandboxSetInformer := informerFactory.Api().V1alpha1().SandboxSets().Informer()
 	checkpointInformer := informerFactory.Api().V1alpha1().Checkpoints().Informer()
@@ -87,11 +97,15 @@ func NewCache(client *clients.ClientSet, opts config.SandboxManagerOptions) (*Ca
 }
 
 func NewTestCache(t *testing.T) (*Cache, *clients.ClientSet, error) {
-	t.Helper()
-	clientSet := clients.NewFakeClientSet(t)
-	c, err := NewCache(clientSet, config.SandboxManagerOptions{
+	return NewTestCacheWithOptions(t, config.SandboxManagerOptions{
 		SystemNamespace: utils.DefaultSandboxDeployNamespace,
 	})
+}
+
+func NewTestCacheWithOptions(t *testing.T, opts config.SandboxManagerOptions) (*Cache, *clients.ClientSet, error) {
+	t.Helper()
+	clientSet := clients.NewFakeClientSet(t)
+	c, err := NewCache(clientSet, opts)
 	if err != nil {
 		return nil, nil, err
 	}

pkg/sandbox-manager/infra/sandboxcr/cache_test.go

@@ -7,6 +7,7 @@ import (
 
 	agentsv1alpha1 "github.com/openkruise/agents/api/v1alpha1"
 	"github.com/openkruise/agents/pkg/sandbox-manager/clients"
+	"github.com/openkruise/agents/pkg/sandbox-manager/config"
 	constantUtils "github.com/openkruise/agents/pkg/utils"
 	sandboxManagerUtils "github.com/openkruise/agents/pkg/utils/sandbox-manager"
 	utils "github.com/openkruise/agents/pkg/utils/sandbox-manager"
@@ -439,3 +440,43 @@ func TestCache_GetSecret_FromSync(t *testing.T) {
 	assert.Equal(t, testSecret.Data, result.Data)
 	assert.Equal(t, testSecret.Type, result.Type)
 }
+
+func TestCache_InformerWithFilter_GetSandboxSet(t *testing.T) {
+	sandboxManagerUtils.InitLogOutput()
+	cache, clientSet, err := NewTestCacheWithOptions(t, config.SandboxManagerOptions{
+		SandboxNamespace:     "ns",
+		SandboxLabelSelector: "user=bar",
+	})
+	require.NoError(t, err)
+	defer cache.Stop()
+	sandboxClient := clientSet.SandboxClient
+
+	templateName := "template-1"
+	// SandboxSet with wrong namespace
+	sbs1 := &agentsv1alpha1.SandboxSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      templateName,
+			Namespace: "ns-1",
+		},
+	}
+	_, err = sandboxClient.ApiV1alpha1().SandboxSets(sbs1.Namespace).Create(t.Context(), sbs1, metav1.CreateOptions{})
+	require.NoError(t, err)
+	// Wait for informer sync
+	time.Sleep(300 * time.Millisecond)
+	_, err = cache.GetSandboxSet(templateName)
+	require.Error(t, err)
+
+	// SandboxSet in namespace
+	sbs3 := &agentsv1alpha1.SandboxSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      templateName,
+			Namespace: "ns",
+		},
+	}
+	_, err = sandboxClient.ApiV1alpha1().SandboxSets(sbs3.Namespace).Create(t.Context(), sbs3, metav1.CreateOptions{})
+	require.NoError(t, err)
+	// Wait for informer sync
+	time.Sleep(300 * time.Millisecond)
+	_, err = cache.GetSandboxSet(templateName)
+	require.NoError(t, err)
+}

pkg/servers/e2b/core.go

@@ -34,6 +34,8 @@ type Controller struct {
 	maxClaimWorkers       int
 	maxCreateQPS          int
 	extProcMaxConcurrency uint32
+	sandboxLabelSelector  string
+	sandboxNamespace      string
 
 	// fields
 	mux             *http.ServeMux
@@ -49,7 +51,7 @@ type Controller struct {
 }
 
 // NewController creates a new E2B Controller
-func NewController(domain, adminKey string, sysNs string, maxTimeout, maxClaimWorkers, maxCreateQPS int, extProcMaxConcurrency uint32,
+func NewController(domain, adminKey string, sysNs, sandboxNamespace, sandboxLabelSelector string, maxTimeout, maxClaimWorkers, maxCreateQPS int, extProcMaxConcurrency uint32,
 	port int, enableAuth bool, clientSet *clients.ClientSet) *Controller {
 	sc := &Controller{
 		mux:                   http.NewServeMux(),
@@ -59,6 +61,8 @@ func NewController(domain, adminKey string, sysNs string, maxTimeout, maxClaimWo
 		port:                  port,
 		maxTimeout:            maxTimeout,
 		systemNamespace:       sysNs, // the namespace where the sandbox manager is running
+		sandboxNamespace:      sandboxNamespace,
+		sandboxLabelSelector:  sandboxLabelSelector,
 		maxClaimWorkers:       maxClaimWorkers,
 		maxCreateQPS:          maxCreateQPS,
 		extProcMaxConcurrency: extProcMaxConcurrency,
@@ -88,6 +92,8 @@ func (sc *Controller) Init() error {
 	adapter := adapters.DefaultAdapterFactory(sc.port)
 	sandboxManager, err := sandbox_manager.NewSandboxManager(sc.client, adapter, config.SandboxManagerOptions{
 		SystemNamespace:       sc.systemNamespace,
+		SandboxNamespace:      sc.sandboxNamespace,
+		SandboxLabelSelector:  sc.sandboxLabelSelector,
 		MaxClaimWorkers:       sc.maxClaimWorkers,
 		ExtProcMaxConcurrency: sc.extProcMaxConcurrency,
 		MaxCreateQPS:          sc.maxCreateQPS,

pkg/servers/e2b/core_test.go

@@ -85,7 +85,7 @@ func Setup(t *testing.T) (*Controller, *clients.ClientSet, func()) {
 	_, err := clientSet.CoreV1().Secrets(namespace).Create(t.Context(), secret, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
-	controller := NewController("example.com", InitKey, namespace, models.DefaultMaxTimeout, 10,
+	controller := NewController("example.com", InitKey, namespace, "", "", models.DefaultMaxTimeout, 10,
 		0, 0, TestServerPort, true, clientSet)
 	assert.NoError(t, controller.Init())
 	_, err = controller.Run(namespace, "component=sandbox-manager")