Register the connector-service (webapi) plugin in the Flyte 2 executor

executor/pkg/plugin/resource_manager.go

@@ -0,0 +1,45 @@
+package plugin
+
+import (
+	"context"
+
+	pluginsCore "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/core"
+)
+
+// The webapi plugin machinery requires the host to supply a ResourceRegistrar at plugin setup and a
+// ResourceManager at task execution, for its allocation-token quota feature. FlytePropeller (v1)
+// always supplied them, defaulting to a no-op manager when no quota backend is configured. The v2
+// executor never reimplemented this, so a webapi plugin that declares ResourceQuotas dereferences a
+// nil registrar or manager. These no-op types restore the contract and grant every allocation. Swap
+// in a real ResourceManager to enforce quotas without touching any plugin.
+
+// noopResourceRegistrar accepts quota registrations without recording them.
+type noopResourceRegistrar struct{}
+
+var _ pluginsCore.ResourceRegistrar = noopResourceRegistrar{}
+
+func (noopResourceRegistrar) RegisterResourceQuota(_ context.Context, _ pluginsCore.ResourceNamespace, _ int) error {
+	return nil
+}
+
+// noopResourceManager grants every allocation request and ignores releases.
+type noopResourceManager struct{}
+
+var _ pluginsCore.ResourceManager = noopResourceManager{}
+
+func (noopResourceManager) GetID() string { return "executor-noop-resource-manager" }
+
+func (noopResourceManager) AllocateResource(_ context.Context, _ pluginsCore.ResourceNamespace, _ string, _ pluginsCore.ResourceConstraintsSpec) (pluginsCore.AllocationStatus, error) {
+	return pluginsCore.AllocationStatusGranted, nil
+}
+
+func (noopResourceManager) ReleaseResource(_ context.Context, _ pluginsCore.ResourceNamespace, _ string) error {
+	return nil
+}
+
+// NewNoopResourceRegistrar returns a ResourceRegistrar that accepts quota declarations without
+// enforcing them.
+func NewNoopResourceRegistrar() pluginsCore.ResourceRegistrar { return noopResourceRegistrar{} }
+
+// NewNoopResourceManager returns a ResourceManager that grants every allocation.
+func NewNoopResourceManager() pluginsCore.ResourceManager { return noopResourceManager{} }

executor/pkg/plugin/resource_manager_test.go

@@ -0,0 +1,30 @@
+package plugin
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	pluginsCore "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/core"
+)
+
+func TestNoopResourceRegistrar(t *testing.T) {
+	r := NewNoopResourceRegistrar()
+	assert.NotNil(t, r)
+	// Accepts a quota declaration (the connector ships ResourceQuotas{"default":1000}) without error.
+	assert.NoError(t, r.RegisterResourceQuota(context.Background(), "default", 1000))
+}
+
+func TestNoopResourceManager(t *testing.T) {
+	m := NewNoopResourceManager()
+	assert.NotNil(t, m)
+	assert.Equal(t, "executor-noop-resource-manager", m.GetID())
+
+	// Every allocation is granted, matching v1 propeller with no quota backend configured.
+	status, err := m.AllocateResource(context.Background(), "default", "token", pluginsCore.ResourceConstraintsSpec{})
+	assert.NoError(t, err)
+	assert.Equal(t, pluginsCore.AllocationStatusGranted, status)
+
+	assert.NoError(t, m.ReleaseResource(context.Background(), "default", "token"))
+}

executor/setup.go

@@ -29,6 +29,7 @@ import (
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/catalog"
 	cachecatalog "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/catalog/cache_service"
 	webhookConfig "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/secret/config"
+	connectorplugin "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/plugins/webapi/connector"
 	"github.com/flyteorg/flyte/v2/flytestdlib/app"
 	"github.com/flyteorg/flyte/v2/flytestdlib/otelutils"
 	"github.com/flyteorg/flyte/v2/flytestdlib/promutils"
@@ -66,6 +67,11 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 		utilruntime.Must(reg.AddToScheme(scheme))
 	}
 
+	// Register the connector (webapi) backend plugin so task types backed by an external connector
+	// service are routed to it. This must run before plugin.NewRegistry below, which snapshots the
+	// core plugins once.
+	connectorplugin.RegisterConnectorPlugin(&connectorplugin.ConnectorService{})
+
 	var tlsOpts []func(*tls.Config)
 	if !cfg.EnableHTTP2 {
 		tlsOpts = append(tlsOpts, func(c *tls.Config) {
@@ -129,7 +135,7 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 	}
 
 	setupCtx := plugin.NewSetupContext(
-		mgr, nil, nil, nil, nil,
+		mgr, nil, plugin.NewNoopResourceRegistrar(), nil, nil,
 		"TaskAction",
 		executorScope.NewSubScope("plugin"),
 	)
@@ -177,6 +183,10 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 	reconciler.CatalogClient = asyncCatalogClient
 	reconciler.Catalog = cacheClient
 	reconciler.Recorder = mgr.GetEventRecorder("taskaction-controller")
+	// Supply a ResourceManager for the webapi allocation-token path, used by connector-backed task
+	// types that declare ResourceQuotas. It grants every allocation by default, matching
+	// FlytePropeller with no quota backend. Swap in a real one to enforce quotas.
+	reconciler.ResourceManager = plugin.NewNoopResourceManager()
 	if cfg.MaxSystemFailures < 0 {
 		return fmt.Errorf("executor: maxSystemFailures must be non-negative, got %d", cfg.MaxSystemFailures)
 	}