[CDAP-21242] Add optimized Pipeline Summary API endpoint

cdap-app-fabric/src/main/java/io/cdap/cdap/app/store/Store.java

@@ -29,6 +29,7 @@
 import io.cdap.cdap.common.ConflictException;
 import io.cdap.cdap.common.NotFoundException;
 import io.cdap.cdap.common.ProgramNotFoundException;
+import io.cdap.cdap.internal.app.store.AppSummary;
 import io.cdap.cdap.internal.app.store.ApplicationMeta;
 import io.cdap.cdap.internal.app.store.RunRecordDetail;
 import io.cdap.cdap.internal.app.store.WorkflowTable;
@@ -351,7 +352,7 @@
    * @return a {@link Map} from the {@link ProgramId} to the list of run records; there will be no entry for programs
    *     that do not exist.
    */
   Map<ProgramId, Collection<RunRecordDetail>> getActiveRuns(Collection<ProgramReference> programRefs);
 
   /**
    * Fetches the active (i.e STARTING or RUNNING or SUSPENDED) run records for the
@@ -495,6 +496,17 @@
   boolean scanApplications(ScanApplicationsRequest request, int txBatchSize,
                            BiConsumer<ApplicationId, ApplicationMeta> consumer);
 
+  /**
+   * Scans for application summaries according to the parameters passed in request.
+   *
+   * @param request     parameters defining filters and sorting
+   * @param txBatchSize maximum number of applications to scan in one transaction
+   * @param consumer    a {@link Consumer} to consume each application summary being scanned
+   * @return if limit was reached (true) or all items were scanned before reaching the limit (false)
+   */
+  boolean scanApplicationSummaries(ScanApplicationsRequest request, int txBatchSize,
+      Consumer<AppSummary> consumer);
+
   /**
    * Returns a Map of {@link ApplicationMeta} for the given set of {@link ApplicationId}.
    *

cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/ProgramLifecycleHttpHandler.java

@@ -16,7 +16,10 @@
 
 package io.cdap.cdap.gateway.handlers;
 
+import com.google.common.base.Splitter;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.reflect.TypeToken;
@@ -26,6 +29,8 @@
 import io.cdap.cdap.api.app.ApplicationSpecification;
 import io.cdap.cdap.api.service.ServiceUnavailableException;
 import io.cdap.cdap.app.mapreduce.MRJobInfoFetcher;
+import io.cdap.cdap.app.store.ApplicationFilter;
+import io.cdap.cdap.app.store.ScanApplicationsRequest;
 import io.cdap.cdap.app.store.Store;
 import io.cdap.cdap.common.BadRequestException;
 import io.cdap.cdap.common.ConflictException;
@@ -60,22 +65,29 @@
 import io.cdap.cdap.proto.RunRecord;
 import io.cdap.cdap.proto.id.ApplicationId;
 import io.cdap.cdap.proto.id.ApplicationReference;
+import io.cdap.cdap.proto.id.EntityId;
+import io.cdap.cdap.proto.id.NamespaceId;
 import io.cdap.cdap.proto.id.ProgramId;
 import io.cdap.cdap.proto.id.ProgramReference;
 import io.cdap.cdap.proto.id.ProgramRunId;
 import io.cdap.cdap.security.spi.authorization.UnauthorizedException;
+import io.cdap.cdap.spi.data.SortOrder;
 import io.cdap.http.HttpResponder;
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import javax.ws.rs.DefaultValue;
@@ -333,7 +345,7 @@
    *     (inclusive)
    * @param endTimeSeconds upper bound in millis of the stoppage time for programs (exclusive)
    *
    * Deprecated: Only allowing for the latest app version (active app).
    */
   @Deprecated
   @PUT
@@ -538,7 +550,7 @@
 
   /**
    * Get runtime args of a program with app version.
    *
    * Deprecated : runtime args are versionless.
    */
   @Deprecated
@@ -558,7 +570,7 @@
 
   /**
    * Save runtime args of program with app version.
    *
    * Deprecated : runtime args are versionless.
    */
   @Deprecated
@@ -594,7 +606,7 @@
     responder.sendStatus(HttpResponseStatus.OK);
   }
 
   @GET
   @Path("/apps/{app-name}/{program-type}/{program-name}")
   public void programSpecification(HttpRequest request, HttpResponder responder,
       @PathParam("namespace-id") String namespaceId, @PathParam("app-name") String appName,
@@ -608,7 +620,7 @@
   /*
   * Deprecated : program info fetch is only allowed for the latest version.
   * */
   @Deprecated
   @GET
   @Path("/apps/{app-name}/versions/{app-version}/{program-type}/{program-name}")
   public void programSpecificationVersioned(HttpRequest request, HttpResponder responder,
@@ -1011,7 +1023,7 @@
       @PathParam("namespace-id") String namespaceId) throws Exception {
     responder.sendJson(HttpResponseStatus.OK,
         ProgramHandlerUtil.toJson(
             lifecycleService.list(NamespaceHelper.validateNamespace(namespaceQueryAdmin,namespaceId),
                 ProgramType.MAPREDUCE)));
   }
 
@@ -1024,7 +1036,7 @@
       @PathParam("namespace-id") String namespaceId) throws Exception {
     responder.sendJson(HttpResponseStatus.OK,
         ProgramHandlerUtil.toJson(
             lifecycleService.list(NamespaceHelper.validateNamespace(namespaceQueryAdmin,namespaceId),
                 ProgramType.SPARK)));
   }
 
@@ -1037,7 +1049,7 @@
       @PathParam("namespace-id") String namespaceId) throws Exception {
     responder.sendJson(HttpResponseStatus.OK,
         ProgramHandlerUtil.toJson(
             lifecycleService.list(NamespaceHelper.validateNamespace(namespaceQueryAdmin,namespaceId),
                 ProgramType.WORKFLOW)));
   }
 
@@ -1050,20 +1062,122 @@
       @PathParam("namespace-id") String namespaceId) throws Exception {
     responder.sendJson(HttpResponseStatus.OK,
         ProgramHandlerUtil.toJson(
             lifecycleService.list(NamespaceHelper.validateNamespace(namespaceQueryAdmin,namespaceId),
                 ProgramType.SERVICE)));
   }
 
   @GET
   @Path("/workers")
   public void getAllWorkers(HttpRequest request, HttpResponder responder,
       @PathParam("namespace-id") String namespaceId) throws Exception {
     responder.sendJson(HttpResponseStatus.OK,
         ProgramHandlerUtil.toJson(
             lifecycleService.list(NamespaceHelper.validateNamespace(namespaceQueryAdmin,namespaceId),
                 ProgramType.WORKER)));
   }
 
+  /**
+   * Returns a summary of pipelines in the namespace, including metadata, run status, and run
+   * counts.
+   *
+   * <p>This endpoint is optimized to avoid full deserialization of application specifications.
+   *
+   * @param request          the HTTP request
+   * @param responder        the HTTP responder used to send the JSON response
+   * @param namespaceId      the ID of the namespace to scan
+   * @param artifactName     optional comma-separated list of artifact names to filter by
+   * @param artifactVersion  optional artifact version to filter by
+   * @param pageToken        optional page token for pagination
+   * @param pageSize         optional page size, defaults to 25
+   * @param orderBy          optional sort order
+   * @param nameFilter       optional filter for application name
+   * @param nameFilterType   optional type of name filter (e.g., EQUALS, CONTAINS)
+   * @param latestOnly       optional flag to return only the latest version, defaults to true
+   * @param sortCreationTime optional flag to sort by creation time
+   * @throws Exception if any error occurs during execution
+   */
+  @GET
+  @Path("/apps/summary")
+  public void getPipelineSummaries(HttpRequest request, HttpResponder responder,
+      @PathParam("namespace-id") String namespaceId,
+      @QueryParam("artifactName") String artifactName,
+      @QueryParam("artifactVersion") String artifactVersion,
+      @QueryParam("pageToken") String pageToken,
+      @QueryParam("pageSize") @DefaultValue("25") Integer pageSize,
+      @QueryParam("orderBy") SortOrder orderBy,
+      @QueryParam("nameFilter") String nameFilter,
+      @QueryParam("nameFilterType") NameFilterType nameFilterType,
+      @QueryParam("latestOnly") @DefaultValue("true") Boolean latestOnly,
+      @QueryParam("sortCreationTime") Boolean sortCreationTime) throws Exception {
+
+    ScanApplicationsRequest scanRequest = getSummaryScanRequest(namespaceId, artifactName,
+        artifactVersion,
+        pageToken, pageSize, orderBy, nameFilter, nameFilterType, latestOnly, sortCreationTime);
+
+    responder.sendJson(HttpResponseStatus.OK,
+        ProgramHandlerUtil.toJson(lifecycleService.getAppSummaries(scanRequest)));
+  }
+
+  private ScanApplicationsRequest getSummaryScanRequest(String namespaceId, String artifactName,
+      String artifactVersion, String pageToken,
+      Integer pageSize, SortOrder orderBy, String nameFilter,
+      NameFilterType nameFilterType, Boolean latestOnly, Boolean sortCreationTime) {
+    Set<String> names = new HashSet<>();
+    if (!Strings.isNullOrEmpty(artifactName)) {
+      for (String name : Splitter.on(',').omitEmptyStrings().trimResults()
+          .split(Objects.requireNonNull(artifactName))) {
+        names.add(name);
+      }
+    }
+
+    ScanApplicationsRequest.Builder builder = ScanApplicationsRequest.builder();
+    builder.setNamespaceId(new NamespaceId(namespaceId));
+    if (pageSize != null) {
+      builder.setLimit(pageSize);
+    }
+    if (latestOnly != null) {
+      builder.setLatestOnly(latestOnly);
+    }
+    if (orderBy != null) {
+      builder.setSortOrder(orderBy);
+    }
+    if (sortCreationTime != null) {
+      builder.setSortCreationTime(sortCreationTime);
+    }
+    if (!Strings.isNullOrEmpty(pageToken)) {
+      builder.setScanFrom(ApplicationId.fromIdParts(Iterables.concat(
+          Collections.singleton(namespaceId),
+          Arrays.asList(EntityId.IDSTRING_PART_SEPARATOR_PATTERN.split(pageToken.trim())))));
+    }
+
+    List<ApplicationFilter> filters = new ArrayList<>();
+    if (!names.isEmpty()) {
+      filters.add(new ApplicationFilter.ArtifactNamesInFilter(names));
+    }
+    if (!Strings.isNullOrEmpty(artifactVersion)) {
+      filters.add(new ApplicationFilter.ArtifactVersionFilter(artifactVersion.trim()));
+    }
+    if (!Strings.isNullOrEmpty(nameFilter)) {
+      String cleanNameFilter = nameFilter.trim();
+      NameFilterType type = nameFilterType != null ? nameFilterType : NameFilterType.CONTAINS;
+
+      switch (type) {
+        case EQUALS:
+          builder.setApplicationReference(new ApplicationReference(namespaceId, cleanNameFilter));
+          break;
+        case EQUALS_IGNORE_CASE:
+          filters.add(new ApplicationFilter.ApplicationIdEqualsFilter(cleanNameFilter));
+          break;
+        case CONTAINS:
+        default:
+          filters.add(new ApplicationFilter.ApplicationIdContainsFilter(cleanNameFilter));
+          break;
+      }
+    }
+    builder.addFilters(filters);
+    return builder.build();
+  }
+
   /**
    * Return the availability (i.e. discoverable registration) status of a service.
    */
@@ -1081,7 +1195,7 @@
 
   /**
    * Return the availability (i.e. discoverable registration) status of a service.
    *
    * Deprecated : Only allow checking availability of the service corresponding to the latest app version.
    */
   @Deprecated
@@ -1135,7 +1249,7 @@
     return EnumSet.of(ProgramType.SERVICE, ProgramType.WORKER).contains(programType);
   }
 
   /**
    * Used to filter out RunRecords initiated by a tethered instance
    */
   private boolean isTetheredRunRecord(RunRecord runRecord) {

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/services/ProgramLifecycleService.java

@@ -69,6 +69,10 @@
 import io.cdap.cdap.proto.ProgramType;
 import io.cdap.cdap.proto.RunCountResult;
 import io.cdap.cdap.proto.RunRecord;
+import io.cdap.cdap.proto.AppSummaryRecord;
+import io.cdap.cdap.proto.AppSummaryResponse;
+import io.cdap.cdap.api.artifact.ArtifactSummary;
+import io.cdap.cdap.internal.app.store.AppSummary;
 import io.cdap.cdap.proto.id.ApplicationId;
 import io.cdap.cdap.proto.id.ApplicationReference;
 import io.cdap.cdap.proto.id.EntityId;
@@ -100,6 +104,7 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -144,6 +149,7 @@
   private final int maxConcurrentLaunching;
   private final int defaultStopTimeoutSecs;
   private final int batchSize;
+  private final int summaryBatchSize;
 
   private final boolean userProgramLaunchDisabled;
 
@@ -170,13 +176,135 @@
     this.userProgramLaunchDisabled = cConf.getBoolean(
         Constants.AppFabric.USER_PROGRAM_LAUNCH_DISABLED, false);
     this.batchSize = cConf.getInt(Constants.AppFabric.STREAMING_BATCH_SIZE);
+    this.summaryBatchSize = cConf.getInt(Constants.AppFabric.SUMMARY_STREAMING_BATCH_SIZE);
     this.profileService = profileService;
     this.preferencesService = preferencesService;
     this.provisionerNotifier = provisionerNotifier;
     this.provisioningService = provisioningService;
     this.flowControlService = flowControlService;
   }
 
+  /**
+   * Returns a summary of applications in the namespace, including metadata, run status, and run
+   * counts. This method is optimized to avoid full deserialization of application specifications.
+   *
+   * @param request the scan request
+   * @return the response with pipelines and next page token
+   */
+  public AppSummaryResponse getAppSummaries(final ScanApplicationsRequest request)
+      throws Exception {
+    List<AppSummary> appSummaries = new ArrayList<>();
+    store.scanApplicationSummaries(request, summaryBatchSize, appSummaries::add);
+    if (appSummaries.isEmpty()) {
+      return new AppSummaryResponse(Collections.emptyList(), null);
+    }
+
+    List<AppSummary> visibleAppSummaries = filterVisibleSummaries(appSummaries);
+    List<ProgramReference> visibleProgramReferences = visibleAppSummaries.stream()
+        .map(AppSummary::getPrimaryProgram)
+        .filter(Objects::nonNull)
+        .map(ProgramId::getProgramReference)
+        .collect(Collectors.toList());
+
+    Map<ProgramReference, Long> runCounts = fetchRunCounts(visibleProgramReferences);
+    Map<ProgramReference, RunRecord> latestRuns = fetchLatestRuns(visibleProgramReferences);
+    List<AppSummaryRecord> pipelines = createResponseRecords(visibleAppSummaries, runCounts,
+        latestRuns);
+
+    String nextPageToken = generateNextPageToken(appSummaries);
+    return new AppSummaryResponse(pipelines, nextPageToken);
+  }
+
+  private List<AppSummary> filterVisibleSummaries(List<AppSummary> appSummaries) {
+    Set<ProgramId> allProgramIds = appSummaries.stream()
+        .map(AppSummary::getPrimaryProgram)
+        .filter(Objects::nonNull)
+        .collect(Collectors.toSet());
+    Set<? extends EntityId> visiblePrograms = allProgramIds.isEmpty()
+        ? Collections.emptySet()
+        : accessEnforcer.isVisible(allProgramIds, authenticationContext.getPrincipal());
+
+    return appSummaries.stream()
+        .filter(summary -> {
+          ProgramId programId = summary.getPrimaryProgram();
+          return programId == null || visiblePrograms.contains(programId);
+        })
+        .collect(Collectors.toList());
+  }
+
+  private Map<ProgramReference, Long> fetchRunCounts(List<ProgramReference> programRefs)
+      throws Exception {
+    Map<ProgramReference, Long> runCounts = new HashMap<>();
+    if (programRefs.isEmpty()) {
+      return runCounts;
+    }
+
+    for (RunCountResult countResult : getProgramTotalRunCounts(programRefs)) {
+      if (countResult.getException() == null) {
+        runCounts.put(countResult.getProgramReference(), countResult.getCount());
+      }
+    }
+    return runCounts;
+  }
+
+  private Map<ProgramReference, RunRecord> fetchLatestRuns(List<ProgramReference> programRefs)
+      throws Exception {
+    Map<ProgramReference, RunRecord> latestRuns = new HashMap<>();
+    if (programRefs.isEmpty()) {
+      return latestRuns;
+    }
+
+    List<ProgramHistory> histories = getRunRecords(
+        programRefs, ProgramRunStatus.ALL, 0, Long.MAX_VALUE, 1);
+    for (ProgramHistory history : histories) {
+      if (history.getException() == null && !history.getRuns().isEmpty()) {
+        latestRuns.put(history.getProgramId().getProgramReference(), history.getRuns().get(0));
+      }
+    }
+    return latestRuns;
+  }
+
+  private List<AppSummaryRecord> createResponseRecords(
+      List<AppSummary> visibleAppSummaries,
+      Map<ProgramReference, Long> runCounts,
+      Map<ProgramReference, RunRecord> latestRuns) {
+    return visibleAppSummaries.stream().map(summary -> {
+      ProgramId programId = summary.getPrimaryProgram();
+      ArtifactSummary artifactSummary = ArtifactSummary.from(summary.getArtifactId());
+      if (programId == null) {
+        return new AppSummaryRecord(
+            summary.getAppId().getApplication(),
+            summary.getAppId().getVersion(),
+            summary.getDescription(),
+            artifactSummary,
+            0L,
+            null
+        );
+      }
+
+      ProgramReference programRef = programId.getProgramReference();
+      return new AppSummaryRecord(
+          summary.getAppId().getApplication(),
+          summary.getAppId().getVersion(),
+          summary.getDescription(),
+          artifactSummary,
+          runCounts.getOrDefault(programRef, 0L),
+          latestRuns.get(programRef)
+      );
+    }).collect(Collectors.toList());
+  }
+
+  private String generateNextPageToken(List<AppSummary> appSummaries) {
+    ProgramId lastProgramId = appSummaries.get(appSummaries.size() - 1).getPrimaryProgram();
+    if (lastProgramId == null) {
+      return null;
+    }
+
+    return lastProgramId.getApplication()
+        + EntityId.IDSTRING_PART_SEPARATOR
+        + lastProgramId.getVersion();
+  }
+
   /**
    * Returns the status of the latest version program.
    *
@@ -739,9 +867,9 @@
   /**
    * Starts a Program with the specified argument overrides, skipping cluster lifecycle steps in the
    * run.
    *
    * NOTE: {@Link ProgramRuntimeService#run} needs be called to start the program run.
    *
    * NOTE: This method should only be called from preview runner.
    *
    * @param programId the {@link ProgramId} to start/stop
@@ -1227,7 +1355,7 @@
     return applicationId.program(programReference.getType(), programReference.getProgram());
   }
 
   public void checkCapability(ProgramDescriptor programDescriptor) throws Exception {
     // Check for capability at application class level.
     Set<ApplicationClass> applicationClasses = artifactRepository
         .getArtifact(Id.Artifact.fromEntityId(programDescriptor.getArtifactId())).getMeta()
@@ -1255,7 +1383,7 @@
   private void addAppCdapVersion(ProgramId programId, Map<String, String> systemArgs) {
     ApplicationSpecification appSpec = store.getApplication(programId.getParent());
     if (appSpec != null) {
       String appCDAPVersion = appSpec.getAppCDAPVersion();
       if (appCDAPVersion != null) {
         systemArgs.put(Constants.APP_CDAP_VERSION, appCDAPVersion);
       }
@@ -1345,7 +1473,7 @@
    * @param runRecords run records for the program
    * @return the program status
    */
   @VisibleForTesting
   static ProgramStatus getProgramStatus(Collection<RunRecordDetail> runRecords) {
     boolean hasStarting = false;
     for (RunRecordDetail runRecord : runRecords) {