Fix per-run log fetching

Author: Vishnuujain

openmetadata-service/src/main/java/org/openmetadata/service/clients/pipeline/MeteredPipelineServiceClient.java

@@ -136,6 +136,13 @@ public Map<String, String> getLastIngestionLogs(
         GET_LOGS, () -> this.decoratedClient.getLastIngestionLogs(ingestionPipeline, after));
   }
 
+  @Override
+  public Map<String, String> getIngestionLogs(
+      IngestionPipeline ingestionPipeline, String after, String runId) {
+    return executeWithMetering(
+        GET_LOGS, () -> this.decoratedClient.getIngestionLogs(ingestionPipeline, after, runId));
+  }
+
   public PipelineServiceClientResponse killIngestion(IngestionPipeline ingestionPipeline) {
     return this.respondWithMetering(
         KILL, () -> this.decoratedClient.killIngestion(ingestionPipeline));

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java

@@ -718,6 +718,41 @@ public PipelineStatus getPipelineStatus(String ingestionPipelineFQN, String runI
         PipelineStatus.class);
   }
 
+  /**
+   * Upsert only the time-series record for a specific run without overwriting the pipeline-level
+   * current status. Use this when stopping a specific run while other runs may still be active.
+   * Inserts a new record if none exists for the runId, otherwise updates the existing one.
+   */
+  public void updatePipelineStatusByRunId(String fqn, PipelineStatus pipelineStatus) {
+    IngestionPipeline ingestionPipeline = findByName(fqn, Include.NON_DELETED);
+    String pipelineFqn = ingestionPipeline.getFullyQualifiedName();
+    String json = JsonUtils.pojoToJson(pipelineStatus);
+    PipelineStatus storedPipelineStatus =
+        JsonUtils.readValue(
+            daoCollection
+                .entityExtensionTimeSeriesDao()
+                .getLatestExtensionByKey(
+                    RUN_ID_EXTENSION_KEY,
+                    pipelineStatus.getRunId(),
+                    pipelineFqn,
+                    PIPELINE_STATUS_EXTENSION),
+            PipelineStatus.class);
+    if (storedPipelineStatus != null) {
+      daoCollection
+          .entityExtensionTimeSeriesDao()
+          .updateExtensionByKey(
+              RUN_ID_EXTENSION_KEY,
+              pipelineStatus.getRunId(),
+              pipelineFqn,
+              PIPELINE_STATUS_EXTENSION,
+              json);
+    } else {
+      daoCollection
+          .entityExtensionTimeSeriesDao()
+          .insert(pipelineFqn, PIPELINE_STATUS_EXTENSION, PIPELINE_STATUS_JSON_SCHEMA, json);
+    }
+  }
+
   @Transaction
   public IngestionPipeline deletePipelineStatusByRunId(UUID ingestionPipelineId, UUID runId) {
     IngestionPipeline ingestionPipeline = find(ingestionPipelineId, Include.NON_DELETED);

openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppResource.java

@@ -535,7 +535,14 @@ public Response getLastLogs(
               schema = @Schema(type = "string"))
           @QueryParam("after")
           @DefaultValue("")
-          String after) {
+          String after,
+      @Parameter(
+              description =
+                  "Pipeline run ID (Argo workflow UID) to fetch logs for a specific run. "
+                      + "If not provided, returns logs for the latest run.",
+              schema = @Schema(type = "string"))
+          @QueryParam("runId")
+          String runId) {
     App installation = repository.getByName(uriInfo, name, repository.getFields("id,pipelines"));
     if (installation.getAppType().equals(AppType.Internal)) {
       AppRunRecord latestRun = repository.getLatestAppRunsOptional(installation).orElse(null);
@@ -552,7 +559,7 @@ public Response getLastLogs(
             ingestionPipelineRepository.get(
                 uriInfo, pipelineRef.getId(), ingestionPipelineRepository.getFields(FIELD_OWNERS));
         return Response.ok(
-                pipelineServiceClient.getLastIngestionLogs(ingestionPipeline, after),
+                pipelineServiceClient.getIngestionLogs(ingestionPipeline, after, runId),
                 MediaType.APPLICATION_JSON_TYPE)
             .build();
       }
@@ -1338,8 +1345,11 @@ private void markPipelineStatusAsStopped(
       if (!PipelineStatusUtils.isTerminalState(status.getPipelineState())) {
         status.setPipelineState(PipelineStatusType.STOPPED);
         status.setEndDate(System.currentTimeMillis());
-        ingestionPipelineRepository.addPipelineStatus(
-            uriInfo, ingestionPipeline.getFullyQualifiedName(), status);
+        // Use updatePipelineStatusByRunId instead of addPipelineStatus to avoid overwriting
+        // the pipeline-level current status. When stopping a specific run, other runs may still
+        // be active and their status should not be affected.
+        ingestionPipelineRepository.updatePipelineStatusByRunId(
+            ingestionPipeline.getFullyQualifiedName(), status);
       }
     } catch (Exception e) {
       LOG.error(

openmetadata-spec/src/main/java/org/openmetadata/sdk/PipelineServiceClientInterface.java

@@ -127,6 +127,13 @@ default PipelineServiceClientResponse runPipeline(
   /* Get the all last run logs of a deployed pipeline */
   Map<String, String> getLastIngestionLogs(IngestionPipeline ingestionPipeline, String after);
 
+  /* Get logs for a specific pipeline run identified by runId (e.g. Argo workflow UID).
+   * When runId is null or blank, falls back to getLastIngestionLogs (latest run). */
+  default Map<String, String> getIngestionLogs(
+      IngestionPipeline ingestionPipeline, String after, String runId) {
+    return getLastIngestionLogs(ingestionPipeline, after);
+  }
+
   /* Get the all last run logs of a deployed pipeline */
   PipelineServiceClientResponse killIngestion(IngestionPipeline ingestionPipeline);
 

openmetadata-ui/src/main/resources/ui/src/components/Settings/Applications/AppRunsHistory/AppRunsHistory.component.tsx

@@ -138,16 +138,21 @@ const AppRunsHistory = forwardRef(
     }, [appData, appRunsHistoryData, isExternalApp]);
 
     const handleRowExpandable = useCallback(
-      (key?: string) => {
+      (key?: string, record?: AppRunRecordWithId) => {
         if (key) {
           if (isExternalApp && appData) {
-            return navigate(
-              getLogsViewerPath(
-                GlobalSettingOptions.APPLICATIONS,
-                appData.name ?? '',
-                appData.name ?? ''
-              )
+            const basePath = getLogsViewerPath(
+              GlobalSettingOptions.APPLICATIONS,
+              appData.name ?? '',
+              appData.name ?? ''
             );
+            const rawRunId = record?.properties?.pipelineRunId;
+            const runId = typeof rawRunId === 'string' ? rawRunId : undefined;
+            const path = runId
+              ? `${basePath}?runId=${encodeURIComponent(runId)}`
+              : basePath;
+
+            return navigate(path);
           }
           if (expandedRowKeys.includes(key)) {
             setExpandedRowKeys((prev) => prev.filter((item) => item !== key));
@@ -190,7 +195,7 @@ const AppRunsHistory = forwardRef(
               disabled={showLogAction(record)}
               size="small"
               type="link"
-              onClick={() => handleRowExpandable(record.id)}>
+              onClick={() => handleRowExpandable(record.id, record)}>
               {t('label.log-plural')}
             </Button>
             <Button

openmetadata-ui/src/main/resources/ui/src/pages/LogsViewerPage/LogsViewerPage.tsx

@@ -36,6 +36,7 @@ import {
   useState,
 } from 'react';
 import { useTranslation } from 'react-i18next';
+import { useSearchParams } from 'react-router-dom';
 import { ReactComponent as TimeDateIcon } from '../../assets/svg/time-date.svg';
 import { CopyToClipboardButton } from '../../components/common/CopyToClipboardButton/CopyToClipboardButton';
 import ErrorPlaceHolder from '../../components/common/ErrorWithPlaceholder/ErrorPlaceHolder';
@@ -84,6 +85,8 @@ import { LogViewerParams } from './LogsViewerPage.interfaces';
 const LogsViewerPage = () => {
   const { logEntityType } = useRequiredParams<LogViewerParams>();
   const { fqn: ingestionName } = useFqn();
+  const [searchParams] = useSearchParams();
+  const runId = searchParams.get('runId') ?? undefined;
 
   const { t } = useTranslation();
   const theme = useTheme();
@@ -116,7 +119,7 @@ const LogsViewerPage = () => {
           endTs: currentTime,
         });
 
-        const logs = await getLatestApplicationRuns(ingestionName);
+        const logs = await getLatestApplicationRuns(ingestionName, runId);
         setAppRuns(data);
         setLogs(logs.data_insight_task || logs.application_task);
 
@@ -314,7 +317,7 @@ const LogsViewerPage = () => {
       }.log`;
 
       if (isApplicationType) {
-        const logs = await downloadAppLogs(ingestionName);
+        const logs = await downloadAppLogs(ingestionName, runId);
         fileName = `${ingestionName}.log`;
         const element = document.createElement('a');
         const file = new Blob([logs || ''], { type: 'text/plain' });
@@ -344,6 +347,7 @@ const LogsViewerPage = () => {
     ingestionName,
     isApplicationType,
     reset,
+    runId,
     updateProgress,
   ]);
 
@@ -520,7 +524,7 @@ const LogsViewerPage = () => {
     } else {
       fetchIngestionDetailsByName();
     }
-  }, []);
+  }, [runId]);
 
   return (
     <PageLayoutV1 pageTitle={t('label.log-viewer')}>

openmetadata-ui/src/main/resources/ui/src/rest/applicationAPI.ts

@@ -102,9 +102,15 @@ export const getExternalApplicationRuns = async (
   return response.data;
 };
 
-export const getLatestApplicationRuns = async (appName: string) => {
+export const getLatestApplicationRuns = async (
+  appName: string,
+  runId?: string
+) => {
   const response = await APIClient.get<DataInsightLatestRun>(
-    `${BASE_URL}/name/${getEncodedFqn(appName)}/logs`
+    `${BASE_URL}/name/${getEncodedFqn(appName)}/logs`,
+    {
+      params: runId ? { runId } : undefined,
+    }
   );
 
   return response.data;
@@ -165,10 +171,15 @@ export const stopApp = async (name: string, runId?: string) => {
   );
 };
 
-export const getApplicationLogs = (appName: string, after?: string) => {
+export const getApplicationLogs = (
+  appName: string,
+  after?: string,
+  runId?: string
+) => {
   return APIClient.get(`${BASE_URL}/name/${appName}/logs`, {
     params: {
       after,
+      ...(runId && { runId }),
     },
   });
 };

openmetadata-ui/src/main/resources/ui/src/utils/IngestionLogs/LogsUtils.ts

@@ -62,23 +62,24 @@ export const getLogsFromResponse = (
 export const fetchLogsRecursively = async (
   ingestionId: string,
   pipelineType: string,
-  after?: string
+  after?: string,
+  runId?: string
 ) => {
   let logs = '';
 
   const {
     data: { total, after: afterCursor, ...rest },
   } =
     pipelineType === PipelineType.Application
-      ? await getApplicationLogs(ingestionId, after)
+      ? await getApplicationLogs(ingestionId, after, runId)
       : await getIngestionPipelineLogById(ingestionId, after);
   logs = logs.concat(getLogsFromResponse(rest, pipelineType));
   if (afterCursor && total) {
     const progress = round((Number(afterCursor) * 100) / Number(total));
     useDownloadProgressStore.getState().updateProgress(progress);
 
     logs = logs.concat(
-      await fetchLogsRecursively(ingestionId, pipelineType, afterCursor)
+      await fetchLogsRecursively(ingestionId, pipelineType, afterCursor, runId)
     );
   }
 
@@ -101,13 +102,18 @@ export const downloadIngestionLog = async (ingestionId?: string) => {
   }
 };
 
-export const downloadAppLogs = async (appName?: string) => {
+export const downloadAppLogs = async (appName?: string, runId?: string) => {
   if (!appName) {
     return '';
   }
 
   try {
-    return await fetchLogsRecursively(appName, PipelineType.Application);
+    return await fetchLogsRecursively(
+      appName,
+      PipelineType.Application,
+      undefined,
+      runId
+    );
   } catch (err) {
     showErrorToast(err as AxiosError);