feat(ingestion): add Executors tab with create/delete executor pools

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/GmsGraphQLEngine.java

@@ -128,6 +128,9 @@
 import com.linkedin.datahub.graphql.resolvers.ingest.execution.IngestionSourceExecutionRequestsResolver;
 import com.linkedin.datahub.graphql.resolvers.ingest.execution.ListExecutionRequestsResolver;
 import com.linkedin.datahub.graphql.resolvers.ingest.execution.RollbackIngestionResolver;
+import com.linkedin.datahub.graphql.resolvers.ingest.executor.CreateExecutorPoolResolver;
+import com.linkedin.datahub.graphql.resolvers.ingest.executor.DeleteExecutorPoolsResolver;
+import com.linkedin.datahub.graphql.resolvers.ingest.executor.ListExecutorPoolsResolver;
 import com.linkedin.datahub.graphql.resolvers.ingest.secret.CreateSecretResolver;
 import com.linkedin.datahub.graphql.resolvers.ingest.secret.DeleteSecretResolver;
 import com.linkedin.datahub.graphql.resolvers.ingest.secret.GetSecretValuesResolver;
@@ -1126,6 +1129,7 @@ private void configureQueryResolvers(final RuntimeWiring.Builder builder) {
                 .dataFetcher(
                     "listExecutionRequests", new ListExecutionRequestsResolver(this.entityClient))
                 .dataFetcher("executionRequest", getResolver(executionRequestType))
+                .dataFetcher("listExecutorPools", new ListExecutorPoolsResolver())
                 .dataFetcher("getSchemaBlame", new GetSchemaBlameResolver(this.timelineService))
                 .dataFetcher("getTimeline", new GetTimelineResolver(this.timelineService))
                 .dataFetcher(
@@ -1314,6 +1318,8 @@ private void configureMutationResolvers(final RuntimeWiring.Builder builder) {
               .dataFetcher("deleteSecret", new DeleteSecretResolver(this.entityClient))
               .dataFetcher(
                   "updateSecret", new UpdateSecretResolver(this.entityClient, this.secretService))
+              .dataFetcher("createExecutorPool", new CreateExecutorPoolResolver())
+              .dataFetcher("deleteExecutorPools", new DeleteExecutorPoolsResolver())
               .dataFetcher(
                   "createAccessToken",
                   new CreateAccessTokenResolver(this.statefulTokenService, this.entityClient))

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/ingest/executor/CreateExecutorPoolResolver.java

@@ -0,0 +1,45 @@
+package com.linkedin.datahub.graphql.resolvers.ingest.executor;
+
+import static com.linkedin.datahub.graphql.resolvers.ResolverUtils.bindArgument;
+
+import com.linkedin.datahub.graphql.QueryContext;
+import com.linkedin.datahub.graphql.concurrency.GraphQLConcurrencyUtils;
+import com.linkedin.datahub.graphql.exception.AuthorizationException;
+import com.linkedin.datahub.graphql.generated.CreateExecutorPoolInput;
+import com.linkedin.datahub.graphql.resolvers.ingest.IngestionAuthUtils;
+import graphql.schema.DataFetcher;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Creates a remote executor pool. Requires the MANAGE ingestion privilege.
+ *
+ * <p>In OSS this uses an in-memory store ({@link ExecutorPoolStore}). Managed DataHub may
+ * override or replace this resolver to perform actual creation against an external executor service.
+ */
+public class CreateExecutorPoolResolver implements DataFetcher<CompletableFuture<String>> {
+
+  @Override
+  public CompletableFuture<String> get(final DataFetchingEnvironment environment) throws Exception {
+
+    final QueryContext context = environment.getContext();
+
+    if (IngestionAuthUtils.canManageIngestion(context)) {
+      return GraphQLConcurrencyUtils.supplyAsync(
+          () -> {
+            final CreateExecutorPoolInput input =
+                bindArgument(environment.getArgument("input"), CreateExecutorPoolInput.class);
+            if (input == null) {
+              return "";
+            }
+            String id = input.getId() != null ? input.getId() : "";
+            String name = input.getName();
+            return ExecutorPoolStore.create(id, name);
+          },
+          this.getClass().getSimpleName(),
+          "get");
+    }
+    throw new AuthorizationException(
+        "Unauthorized to perform this action. Please contact your DataHub administrator.");
+  }
+}

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/ingest/executor/DeleteExecutorPoolsResolver.java

@@ -0,0 +1,38 @@
+package com.linkedin.datahub.graphql.resolvers.ingest.executor;
+
+import com.linkedin.datahub.graphql.QueryContext;
+import com.linkedin.datahub.graphql.concurrency.GraphQLConcurrencyUtils;
+import com.linkedin.datahub.graphql.exception.AuthorizationException;
+import com.linkedin.datahub.graphql.resolvers.ingest.IngestionAuthUtils;
+import graphql.schema.DataFetcher;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Deletes one or more remote executor pools by id. Requires the MANAGE ingestion privilege.
+ *
+ * <p>In OSS this removes pools from the in-memory store ({@link ExecutorPoolStore}) and returns the
+ * deleted ids. Managed DataHub may override or replace this resolver to perform actual deletion
+ * against an external executor service.
+ */
+public class DeleteExecutorPoolsResolver implements DataFetcher<CompletableFuture<List<String>>> {
+
+  @Override
+  public CompletableFuture<List<String>> get(final DataFetchingEnvironment environment)
+      throws Exception {
+
+    final QueryContext context = environment.getContext();
+
+    if (IngestionAuthUtils.canManageIngestion(context)) {
+      @SuppressWarnings("unchecked")
+      final List<String> poolIds = environment.getArgument("poolIds");
+      return GraphQLConcurrencyUtils.supplyAsync(
+          () -> ExecutorPoolStore.delete(poolIds != null ? poolIds : List.of()),
+          this.getClass().getSimpleName(),
+          "get");
+    }
+    throw new AuthorizationException(
+        "Unauthorized to perform this action. Please contact your DataHub administrator.");
+  }
+}

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/ingest/executor/ExecutorPoolStore.java

@@ -0,0 +1,53 @@
+package com.linkedin.datahub.graphql.resolvers.ingest.executor;
+
+import com.linkedin.datahub.graphql.generated.ExecutorPool;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+/**
+ * In-memory store for executor pools. Used by OSS resolvers so create/list/delete work within a
+ * single JVM. Does not persist across restarts. Managed DataHub may use external storage instead.
+ */
+final class ExecutorPoolStore {
+
+  private static final Map<String, ExecutorPool> POOLS = new ConcurrentHashMap<>();
+
+  static List<ExecutorPool> list(int start, int count) {
+    List<ExecutorPool> all = new ArrayList<>(POOLS.values());
+    int total = all.size();
+    int from = Math.min(start, total);
+    int to = Math.min(start + count, total);
+    return new ArrayList<>(all.subList(from, to));
+  }
+
+  static int totalCount() {
+    return POOLS.size();
+  }
+
+  static String create(String id, String name) {
+    if (id == null || id.isBlank()) {
+      return "";
+    }
+    ExecutorPool pool = new ExecutorPool();
+    pool.setId(id.trim());
+    pool.setName(name != null && !name.isBlank() ? name.trim() : null);
+    POOLS.put(id.trim(), pool);
+    return id.trim();
+  }
+
+  static List<String> delete(List<String> poolIds) {
+    if (poolIds == null) {
+      return List.of();
+    }
+    return poolIds.stream()
+        .filter(Objects::nonNull)
+        .filter(id -> POOLS.remove(id) != null)
+        .collect(Collectors.toList());
+  }
+
+  private ExecutorPoolStore() {}
+}

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/ingest/executor/ListExecutorPoolsResolver.java

@@ -0,0 +1,66 @@
+package com.linkedin.datahub.graphql.resolvers.ingest.executor;
+
+import static com.linkedin.datahub.graphql.resolvers.ResolverUtils.bindArgument;
+
+import com.linkedin.datahub.graphql.QueryContext;
+import com.linkedin.datahub.graphql.concurrency.GraphQLConcurrencyUtils;
+import com.linkedin.datahub.graphql.exception.AuthorizationException;
+import com.linkedin.datahub.graphql.generated.ExecutorPool;
+import com.linkedin.datahub.graphql.generated.ListExecutorPoolsInput;
+import com.linkedin.datahub.graphql.generated.ListExecutorPoolsResult;
+import com.linkedin.datahub.graphql.resolvers.ingest.IngestionAuthUtils;
+import graphql.schema.DataFetcher;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Lists remote executor pools. Requires the MANAGE ingestion privilege.
+ *
+ * <p>In OSS this returns pools from an in-memory store (see ExecutorPoolStore). Managed DataHub
+ * may override or replace this resolver to return pools from an external executor service.
+ */
+public class ListExecutorPoolsResolver implements DataFetcher<CompletableFuture<ListExecutorPoolsResult>> {
+
+  private static final Integer DEFAULT_START = 0;
+  private static final Integer DEFAULT_COUNT = 20;
+
+  @Override
+  public CompletableFuture<ListExecutorPoolsResult> get(final DataFetchingEnvironment environment)
+      throws Exception {
+
+    final QueryContext context = environment.getContext();
+
+    if (IngestionAuthUtils.canManageIngestion(context)) {
+      return GraphQLConcurrencyUtils.supplyAsync(
+          () -> {
+            final ListExecutorPoolsInput input =
+                bindArgument(environment.getArgument("input"), ListExecutorPoolsInput.class);
+            if (input == null) {
+              final ListExecutorPoolsResult empty = new ListExecutorPoolsResult();
+              empty.setStart(DEFAULT_START);
+              empty.setCount(DEFAULT_COUNT);
+              empty.setTotal(0);
+              empty.setPools(List.of());
+              return empty;
+            }
+            final int start = input.getStart() == null ? DEFAULT_START : input.getStart();
+            final int count = input.getCount() == null ? DEFAULT_COUNT : input.getCount();
+
+            int total = ExecutorPoolStore.totalCount();
+            List<ExecutorPool> pools = ExecutorPoolStore.list(start, count);
+
+            final ListExecutorPoolsResult result = new ListExecutorPoolsResult();
+            result.setStart(start);
+            result.setCount(count);
+            result.setTotal(total);
+            result.setPools(pools);
+            return result;
+          },
+          this.getClass().getSimpleName(),
+          "get");
+    }
+    throw new AuthorizationException(
+        "Unauthorized to perform this action. Please contact your DataHub administrator.");
+  }
+}

datahub-graphql-core/src/main/resources/ingestion.graphql

@@ -37,6 +37,12 @@ extend type Query {
   urn: The primary key associated with the execution request.
   """
   executionRequest(urn: String!): ExecutionRequest
+
+  """
+  List all remote executor pools. Requires the MANAGE ingestion privilege.
+  In OSS this returns an empty list; managed DataHub may implement full support.
+  """
+  listExecutorPools(input: ListExecutorPoolsInput!): ListExecutorPoolsResult
 }
 
 extend type Mutation {
@@ -98,6 +104,18 @@ extend type Mutation {
   Rollback a specific ingestion execution run based on its runId
   """
   rollbackIngestion(input: RollbackIngestionInput!): String
+
+  """
+  Create a remote executor pool. Requires the MANAGE ingestion privilege.
+  In OSS this is a stub; managed DataHub may implement full support.
+  """
+  createExecutorPool(input: CreateExecutorPoolInput!): String
+
+  """
+  Delete one or more remote executor pools by id. Requires the MANAGE ingestion privilege.
+  In OSS this is a no-op; managed DataHub may implement full support.
+  """
+  deleteExecutorPools(poolIds: [String!]!): [String!]
 }
 
 """
@@ -348,6 +366,81 @@ type ListSecretsResult {
   secrets: [Secret!]!
 }
 
+"""
+A remote executor pool. In OSS this is a stub; managed DataHub may provide full pool metadata.
+"""
+type ExecutorPool {
+  """
+  Unique id of the executor pool
+  """
+  id: String!
+
+  """
+  Optional display name for the pool
+  """
+  name: String
+}
+
+"""
+Input for creating a remote executor pool
+"""
+input CreateExecutorPoolInput {
+  """
+  Unique id for the pool
+  """
+  id: String!
+
+  """
+  Optional display name
+  """
+  name: String
+}
+
+"""
+Input for listing remote executor pools
+"""
+input ListExecutorPoolsInput {
+  """
+  The starting offset of the result set
+  """
+  start: Int
+
+  """
+  The number of results to be returned
+  """
+  count: Int
+
+  """
+  An optional search query
+  """
+  query: String
+}
+
+"""
+Result of listing remote executor pools
+"""
+type ListExecutorPoolsResult {
+  """
+  The starting offset of the result set
+  """
+  start: Int!
+
+  """
+  The number of results returned
+  """
+  count: Int!
+
+  """
+  The total number of pools
+  """
+  total: Int!
+
+  """
+  The executor pools
+  """
+  pools: [ExecutorPool!]!
+}
+
 """
 A schedule associated with an Ingestion Source
 """

datahub-web-react/src/app/ingestV2/ManageIngestionPage.tsx

@@ -8,6 +8,7 @@ import { Tab } from '@components/components/Tabs/Tabs';
 import analytics, { EventType } from '@app/analytics';
 import { useUserContext } from '@app/context/useUserContext';
 import { ExecutionsTab } from '@app/ingestV2/executions/ExecutionsTab';
+import { ExecutorPoolsTab } from '@app/ingestV2/executor/ExecutorPoolsTab';
 import { useIngestionOnboardingRedesignV1 } from '@app/ingestV2/hooks/useIngestionOnboardingRedesignV1';
 import { SecretsList } from '@app/ingestV2/secret/SecretsList';
 import { IngestionSourceList } from '@app/ingestV2/source/IngestionSourceList';
@@ -79,6 +80,7 @@ export const ManageIngestionPage = () => {
     const canManageIngestion = platformPrivileges?.manageIngestion;
     const showIngestionTab = isIngestionEnabled && canManageIngestion;
     const showSecretsTab = isIngestionEnabled && platformPrivileges?.manageSecrets;
+    const showExecutorsTab = isIngestionEnabled && canManageIngestion;
     const showIngestionOnboardingRedesignV1 = useIngestionOnboardingRedesignV1();
 
     // undefined == not loaded, null == no permissions
@@ -177,6 +179,11 @@ export const ManageIngestionPage = () => {
             key: TabType.RunHistory as string,
             name: 'Run History',
         },
+        showExecutorsTab && {
+            component: <ExecutorPoolsTab />,
+            key: TabType.RemoteExecutors as string,
+            name: TabType.RemoteExecutors as string,
+        },
         showSecretsTab && {
             component: (
                 <SecretsList showCreateModal={showCreateSecretModal} setShowCreateModal={setShowCreateSecretModal} />