1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
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";
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/;
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>(),
28 // Max size of logs to fetch in bytes
29 const maxLogFetchSize: number = 128 * 1000;
31 type FileWithProgress = {
36 export type ProcessLogsPanelAction = UnionOf<typeof processLogsPanelActions>;
38 export const setProcessLogsPanelFilter = (filter: string) =>
39 processLogsPanelActions.SET_PROCESS_LOGS_PANEL_FILTER(filter);
41 export const initProcessLogsPanel = (processUuid: string) =>
42 async (dispatch: Dispatch, getState: () => RootState, { logService }: ServiceRepository) => {
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);
50 // Populate lastbyte 0 for each file
51 const filesWithProgress = logFiles.map((file) => ({file, lastByte: 0}));
53 // Fetch array of LogFragments
54 const logLines = await loadContainerLogFileContents(filesWithProgress, logService, process);
56 // Populate initial state with filters
57 const initialState = createInitialLogPanelState(logFiles, logLines);
58 dispatch(processLogsPanelActions.INIT_PROCESS_LOGS_PANEL(initialState));
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 }));
68 export const pollProcessLogs = (processUuid: string) =>
69 async (dispatch: Dispatch, getState: () => RootState, { logService }: ServiceRepository) => {
71 // Get log panel state and process from store
72 const currentState = getState().processLogsPanel;
73 const process = getProcess(processUuid)(getState().resources);
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);
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;
84 const isNew = !Object.keys(currentState.logs).find((currentStateLogName) => (updatedFile.name.startsWith(currentStateLogName)));
85 const isChanged = !isNew && currentStateLogLastByte < updatedFile.size;
87 if (isNew || isChanged) {
88 return acc.concat({file: updatedFile, lastByte: currentStateLogLastByte});
92 }, [] as FileWithProgress[]);
94 // Perform range request(s) for each file
95 const logFragments = await loadContainerLogFileContents(filesToUpdateWithProgress, logService, process);
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));
103 return Promise.resolve();
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();
111 const loadContainerLogFileList = async (containerRequest: ContainerRequestResource, logService: LogService) => {
112 const logCollectionContents = await logService.listLogFiles(containerRequest);
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 &&
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
129 * @returns LogFragment[] containing a single LogFragment corresponding to each input file
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;
138 logService.getLogFileContents(process.containerRequest, file, lastByte, firstChunkEnd),
139 logService.getLogFileContents(process.containerRequest, file, file.size-chunkSize, file.size-1)
140 ] as Promise<(LogFragment)>[]);
142 return Promise.all([logService.getLogFileContents(process.containerRequest, file, lastByte, file.size-1)]);
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'
151 return Promise.reject(error);
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}`;
170 logResponseSet[i-1].contents[lastLineIndex] = newLastLine;
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);
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 };
188 * Converts LogFragments into ProcessLogs, grouping and sorting All/Main logs
190 * @param logFragments
191 * @returns ProcessLogs for the store
193 const groupLogs = (logFiles: CollectionFile[], logFragments: LogFragment[]): ProcessLogs => {
194 const sortableLogFragments = mergeMultilineLoglines(logFragments);
196 const allLogs = mergeSortLogFragments(sortableLogFragments);
197 const mainLogs = mergeSortLogFragments(sortableLogFragments.filter((fragment) => (MAIN_EVENT_TYPES.includes(fragment.logType))));
199 const groupedLogs = logFragments.reduce((grouped, fragment) => ({
201 [fragment.logType as string]: {lastByte: fetchLastByteNumber(logFiles, fragment.logType), contents: fragment.contents}
205 [MAIN_FILTER_TYPE]: {lastByte: undefined, contents: mainLogs},
206 [ALL_FILTER_TYPE]: {lastByte: undefined, contents: allLogs},
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[]
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],
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
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;
237 // Delete the current line and prevent iterating
238 fragmentCopy.contents.splice(i, 1);
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
252 const mergeSortLogFragments = (logFragments: LogFragment[]): string[] => {
253 const sortableLines = fragmentsToLines(logFragments
254 .filter((fragment) => (!NON_SORTED_LOG_TYPES.includes(fragment.logType))));
256 const nonSortableLines = fragmentsToLines(logFragments
257 .filter((fragment) => (NON_SORTED_LOG_TYPES.includes(fragment.logType)))
258 .sort((a, b) => (a.logType.localeCompare(b.logType))));
260 return [...nonSortableLines, ...sortableLines.sort(sortLogLines)]
263 const sortLogLines = (a: string, b: string) => {
264 return a.localeCompare(b);
267 const fragmentsToLines = (fragments: LogFragment[]): string[] => (
268 fragments.reduce((acc, fragment: LogFragment) => (
269 acc.concat(...fragment.contents)
273 const fetchLastByteNumber = (logFiles: CollectionFile[], key: string) => {
274 return logFiles.find((file) => (file.name.startsWith(key)))?.size
277 export const navigateToLogCollection = (uuid: string) =>
278 async (dispatch: Dispatch<any>, getState: () => RootState, services: ServiceRepository) => {
280 await services.collectionService.get(uuid);
281 dispatch<any>(navigateTo(uuid));
283 dispatch(snackbarActions.OPEN_SNACKBAR({ message: 'Could not request collection', hideDuration: 2000, kind: SnackbarKind.ERROR }));
287 const ALL_FILTER_TYPE = 'All logs';
289 const MAIN_FILTER_TYPE = 'Main logs';
290 const MAIN_EVENT_TYPES = [
291 LogEventType.CRUNCH_RUN,
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,
305 LogEventType.CONTAINER,
306 LogEventType.KEEPSTORE,
309 const NON_SORTED_LOG_TYPES = [
310 LogEventType.NODE_INFO,
311 LogEventType.CONTAINER,