src/store/process-logs-panel/process-logs-panel-actions.ts

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 import { unionize, ofType, UnionOf } from "common/unionize";
   6 import { ProcessLogs } from './process-logs-panel';
   7 import { LogEventType } from 'models/log';
   8 import { RootState } from 'store/store';
   9 import { ServiceRepository } from 'services/services';
  10 import { Dispatch } from 'redux';
  11 import { LogFragment, LogService, logFileToLogType } from 'services/log-service/log-service';
  12 import { Process, getProcess } from 'store/processes/process';
  13 import { navigateTo } from 'store/navigation/navigation-action';
  14 import { snackbarActions, SnackbarKind } from 'store/snackbar/snackbar-actions';
  15 import { CollectionFile, CollectionFileType } from "models/collection-file";
  16 import { ContainerRequestResource } from "models/container-request";
  17
  18 const SNIPLINE = `================ ✀ ================ ✀ ========= Some log(s) were skipped ========= ✀ ================ ✀ ================`;
  19 const LOG_TIMESTAMP_PATTERN = /^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{9}Z/;
  20
  21 export const processLogsPanelActions = unionize({
  22     RESET_PROCESS_LOGS_PANEL: ofType<{}>(),
  23     INIT_PROCESS_LOGS_PANEL: ofType<{ filters: string[], logs: ProcessLogs }>(),
  24     SET_PROCESS_LOGS_PANEL_FILTER: ofType<string>(),
  25     ADD_PROCESS_LOGS_PANEL_ITEM: ofType<ProcessLogs>(),
  26 });
  27
  28 // Max size of logs to fetch in bytes
  29 const maxLogFetchSize: number = 128 * 1000;
  30
  31 type FileWithProgress = {
  32     file: CollectionFile;
  33     lastByte: number;
  34 }
  35
  36 export type ProcessLogsPanelAction = UnionOf<typeof processLogsPanelActions>;
  37
  38 export const setProcessLogsPanelFilter = (filter: string) =>
  39     processLogsPanelActions.SET_PROCESS_LOGS_PANEL_FILTER(filter);
  40
  41 export const initProcessLogsPanel = (processUuid: string) =>
  42     async (dispatch: Dispatch, getState: () => RootState, { logService }: ServiceRepository) => {
  43         try {
  44             dispatch(processLogsPanelActions.RESET_PROCESS_LOGS_PANEL());
  45             const process = getProcess(processUuid)(getState().resources);
  46             if (process?.containerRequest?.uuid) {
  47                 // Get log file size info
  48                 const logFiles = await loadContainerLogFileList(process.containerRequest, logService);
  49
  50                 // Populate lastbyte 0 for each file
  51                 const filesWithProgress = logFiles.map((file) => ({file, lastByte: 0}));
  52
  53                 // Fetch array of LogFragments
  54                 const logLines = await loadContainerLogFileContents(filesWithProgress, logService, process);
  55
  56                 // Populate initial state with filters
  57                 const initialState = createInitialLogPanelState(logFiles, logLines);
  58                 dispatch(processLogsPanelActions.INIT_PROCESS_LOGS_PANEL(initialState));
  59             }
  60         } catch(e) {
  61             // On error, populate empty state to allow polling to start
  62             const initialState = createInitialLogPanelState([], []);
  63             dispatch(processLogsPanelActions.INIT_PROCESS_LOGS_PANEL(initialState));
  64             dispatch(snackbarActions.OPEN_SNACKBAR({ message: 'Could not load process logs', hideDuration: 2000, kind: SnackbarKind.ERROR }));
  65         }
  66     };
  67
  68 export const pollProcessLogs = (processUuid: string) =>
  69     async (dispatch: Dispatch, getState: () => RootState, { logService }: ServiceRepository) => {
  70         try {
  71             // Get log panel state and process from store
  72             const currentState = getState().processLogsPanel;
  73             const process = getProcess(processUuid)(getState().resources);
  74
  75             // Check if container request is present and initial logs state loaded
  76             if (process?.containerRequest?.uuid && Object.keys(currentState.logs).length > 0) {
  77                 const logFiles = await loadContainerLogFileList(process.containerRequest, logService);
  78
  79                 // Determine byte to fetch from while filtering unchanged files
  80                 const filesToUpdateWithProgress = logFiles.reduce((acc, updatedFile) => {
  81                     // Fetch last byte or 0 for new log files
  82                     const currentStateLogLastByte = currentState.logs[logFileToLogType(updatedFile)]?.lastByte || 0;
  83
  84                     const isNew = !Object.keys(currentState.logs).find((currentStateLogName) => (updatedFile.name.startsWith(currentStateLogName)));
  85                     const isChanged = !isNew && currentStateLogLastByte < updatedFile.size;
  86
  87                     if (isNew || isChanged) {
  88                         return acc.concat({file: updatedFile, lastByte: currentStateLogLastByte});
  89                     } else {
  90                         return acc;
  91                     }
  92                 }, [] as FileWithProgress[]);
  93
  94                 // Perform range request(s) for each file
  95                 const logFragments = await loadContainerLogFileContents(filesToUpdateWithProgress, logService, process);
  96
  97                 if (logFragments.length) {
  98                     // Convert LogFragments to ProcessLogs with All/Main sorting & line-merging
  99                     const groupedLogs = groupLogs(logFiles, logFragments);
 100                     await dispatch(processLogsPanelActions.ADD_PROCESS_LOGS_PANEL_ITEM(groupedLogs));
 101                 }
 102             }
 103             return Promise.resolve();
 104         } catch (e) {
 105             // Remove log when polling error is handled in some way instead of being ignored
 106             console.error("Error occurred in pollProcessLogs:", e);
 107             return Promise.reject();
 108         }
 109     };
 110
 111 const loadContainerLogFileList = async (containerRequest: ContainerRequestResource, logService: LogService) => {
 112     const logCollectionContents = await logService.listLogFiles(containerRequest);
 113
 114     // Filter only root directory files matching log event types which have bytes
 115     return logCollectionContents.filter((file): file is CollectionFile => (
 116         file.type === CollectionFileType.FILE &&
 117         PROCESS_PANEL_LOG_EVENT_TYPES.indexOf(logFileToLogType(file)) > -1 &&
 118         file.size > 0
 119     ));
 120 };
 121
 122 /**
 123  * Loads the contents of each file from each file's lastByte simultaneously
 124  *   while respecting the maxLogFetchSize by requesting the start and end
 125  *   of the desired block and inserting a snipline.
 126  * @param logFilesWithProgress CollectionFiles with the last byte previously loaded
 127  * @param logService
 128  * @param process
 129  * @returns LogFragment[] containing a single LogFragment corresponding to each input file
 130  */
 131 const loadContainerLogFileContents = async (logFilesWithProgress: FileWithProgress[], logService: LogService, process: Process) => (
 132     (await Promise.allSettled(logFilesWithProgress.filter(({file}) => file.size > 0).map(({file, lastByte}) => {
 133         const requestSize = file.size - lastByte;
 134         if (requestSize > maxLogFetchSize) {
 135             const chunkSize = Math.floor(maxLogFetchSize / 2);
 136             const firstChunkEnd = lastByte+chunkSize-1;
 137             return Promise.all([
 138                 logService.getLogFileContents(process.containerRequest, file, lastByte, firstChunkEnd),
 139                 logService.getLogFileContents(process.containerRequest, file, file.size-chunkSize, file.size-1)
 140             ] as Promise<(LogFragment)>[]);
 141         } else {
 142             return Promise.all([logService.getLogFileContents(process.containerRequest, file, lastByte, file.size-1)]);
 143         }
 144     })).then((res) => {
 145         if (res.length && res.every(promiseResult => (promiseResult.status === 'rejected'))) {
 146             // Since allSettled does not pass promise rejection we throw an
 147             //   error if every request failed
 148             const error = res.find(
 149                 (promiseResult): promiseResult is PromiseRejectedResult => promiseResult.status === 'rejected'
 150               )?.reason;
 151             return Promise.reject(error);
 152         }
 153         return res.filter((promiseResult): promiseResult is PromiseFulfilledResult<LogFragment[]> => (
 154             // Filter out log files with rejected promises
 155             //   (Promise.all rejects on any failure)
 156             promiseResult.status === 'fulfilled' &&
 157             // Filter out files where any fragment is empty
 158             //   (prevent incorrect snipline generation or an un-resumable situation)
 159             !!promiseResult.value.every(logFragment => logFragment.contents.length)
 160         )).map(one => one.value)
 161     })).map((logResponseSet)=> {
 162         // For any multi fragment response set, modify the last line of non-final chunks to include a line break and snip line
 163         //   Don't add snip line as a separate line so that sorting won't reorder it
 164         for (let i = 1; i < logResponseSet.length; i++) {
 165             const fragment = logResponseSet[i-1];
 166             const lastLineIndex = fragment.contents.length-1;
 167             const lastLineContents = fragment.contents[lastLineIndex];
 168             const newLastLine = `${lastLineContents}\n${SNIPLINE}`;
 169
 170             logResponseSet[i-1].contents[lastLineIndex] = newLastLine;
 171         }
 172
 173         // Merge LogFragment Array (representing multiple log line arrays) into single LogLine[] / LogFragment
 174         return logResponseSet.reduce((acc, curr: LogFragment) => ({
 175             logType: curr.logType,
 176             contents: [...(acc.contents || []), ...curr.contents]
 177         }), {} as LogFragment);
 178     })
 179 );
 180
 181 const createInitialLogPanelState = (logFiles: CollectionFile[], logFragments: LogFragment[]): {filters: string[], logs: ProcessLogs} => {
 182     const logs = groupLogs(logFiles, logFragments);
 183     const filters = Object.keys(logs);
 184     return { filters, logs };
 185 }
 186
 187 /**
 188  * Converts LogFragments into ProcessLogs, grouping and sorting All/Main logs
 189  * @param logFiles
 190  * @param logFragments
 191  * @returns ProcessLogs for the store
 192  */
 193 const groupLogs = (logFiles: CollectionFile[], logFragments: LogFragment[]): ProcessLogs => {
 194     const sortableLogFragments = mergeMultilineLoglines(logFragments);
 195
 196     const allLogs = mergeSortLogFragments(sortableLogFragments);
 197     const mainLogs = mergeSortLogFragments(sortableLogFragments.filter((fragment) => (MAIN_EVENT_TYPES.includes(fragment.logType))));
 198
 199     const groupedLogs = logFragments.reduce((grouped, fragment) => ({
 200         ...grouped,
 201         [fragment.logType as string]: {lastByte: fetchLastByteNumber(logFiles, fragment.logType), contents: fragment.contents}
 202     }), {});
 203
 204     return {
 205         [MAIN_FILTER_TYPE]: {lastByte: undefined, contents: mainLogs},
 206         [ALL_FILTER_TYPE]: {lastByte: undefined, contents: allLogs},
 207         ...groupedLogs,
 208     }
 209 };
 210
 211 /**
 212  * Checks for non-timestamped log lines and merges them with the previous line, assumes they are multi-line logs
 213  *   If there is no previous line (first line has no timestamp), the line is deleted.
 214  *   Only used for combined logs that need sorting by timestamp after merging
 215  * @param logFragments
 216  * @returns Modified LogFragment[]
 217  */
 218 const mergeMultilineLoglines = (logFragments: LogFragment[]) => (
 219     logFragments.map((fragment) => {
 220         // Avoid altering the original fragment copy
 221         let fragmentCopy: LogFragment = {
 222             logType: fragment.logType,
 223             contents: [...fragment.contents],
 224         }
 225         // Merge any non-timestamped lines in sortable log types with previous line
 226         if (fragmentCopy.contents.length && !NON_SORTED_LOG_TYPES.includes(fragmentCopy.logType)) {
 227             for (let i = 0; i < fragmentCopy.contents.length; i++) {
 228                 const lineContents = fragmentCopy.contents[i];
 229                 if (!lineContents.match(LOG_TIMESTAMP_PATTERN)) {
 230                     // Partial line without timestamp detected
 231                     if (i > 0) {
 232                         // If not first line, copy line to previous line
 233                         const previousLineContents = fragmentCopy.contents[i-1];
 234                         const newPreviousLineContents = `${previousLineContents}\n${lineContents}`;
 235                         fragmentCopy.contents[i-1] = newPreviousLineContents;
 236                     }
 237                     // Delete the current line and prevent iterating
 238                     fragmentCopy.contents.splice(i, 1);
 239                     i--;
 240                 }
 241             }
 242         }
 243         return fragmentCopy;
 244     })
 245 );
 246
 247 /**
 248  * Merges log lines of different types and sorts types that contain timestamps (are sortable)
 249  * @param logFragments
 250  * @returns string[] of merged and sorted log lines
 251  */
 252 const mergeSortLogFragments = (logFragments: LogFragment[]): string[] => {
 253     const sortableLines = fragmentsToLines(logFragments
 254         .filter((fragment) => (!NON_SORTED_LOG_TYPES.includes(fragment.logType))));
 255
 256     const nonSortableLines = fragmentsToLines(logFragments
 257         .filter((fragment) => (NON_SORTED_LOG_TYPES.includes(fragment.logType)))
 258         .sort((a, b) => (a.logType.localeCompare(b.logType))));
 259
 260     return [...nonSortableLines, ...sortableLines.sort(sortLogLines)]
 261 };
 262
 263 const sortLogLines = (a: string, b: string) => {
 264     return a.localeCompare(b);
 265 };
 266
 267 const fragmentsToLines = (fragments: LogFragment[]): string[] => (
 268     fragments.reduce((acc, fragment: LogFragment) => (
 269         acc.concat(...fragment.contents)
 270     ), [] as string[])
 271 );
 272
 273 const fetchLastByteNumber = (logFiles: CollectionFile[], key: string) => {
 274     return logFiles.find((file) => (file.name.startsWith(key)))?.size
 275 };
 276
 277 export const navigateToLogCollection = (uuid: string) =>
 278     async (dispatch: Dispatch<any>, getState: () => RootState, services: ServiceRepository) => {
 279         try {
 280             await services.collectionService.get(uuid);
 281             dispatch<any>(navigateTo(uuid));
 282         } catch {
 283             dispatch(snackbarActions.OPEN_SNACKBAR({ message: 'Could not request collection', hideDuration: 2000, kind: SnackbarKind.ERROR }));
 284         }
 285     };
 286
 287 const ALL_FILTER_TYPE = 'All logs';
 288
 289 const MAIN_FILTER_TYPE = 'Main logs';
 290 const MAIN_EVENT_TYPES = [
 291     LogEventType.CRUNCH_RUN,
 292     LogEventType.STDERR,
 293     LogEventType.STDOUT,
 294 ];
 295
 296 const PROCESS_PANEL_LOG_EVENT_TYPES = [
 297     LogEventType.ARV_MOUNT,
 298     LogEventType.CRUNCH_RUN,
 299     LogEventType.CRUNCHSTAT,
 300     LogEventType.DISPATCH,
 301     LogEventType.HOSTSTAT,
 302     LogEventType.NODE_INFO,
 303     LogEventType.STDERR,
 304     LogEventType.STDOUT,
 305     LogEventType.CONTAINER,
 306     LogEventType.KEEPSTORE,
 307 ];
 308
 309 const NON_SORTED_LOG_TYPES = [
 310     LogEventType.NODE_INFO,
 311     LogEventType.CONTAINER,
 312 ];