001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.juneau.console.views; 018 019import static java.nio.charset.StandardCharsets.*; 020import static org.apache.juneau.commons.utils.Shorts.*; 021 022import java.io.*; 023import java.nio.*; 024import java.nio.channels.*; 025import java.nio.file.*; 026import java.util.*; 027import java.util.concurrent.atomic.*; 028import java.util.function.*; 029import java.util.logging.*; 030import java.util.regex.*; 031 032import org.apache.juneau.marshall.json.JsonSerializer; 033import org.apache.juneau.marshall.marshaller.Json; 034import org.apache.juneau.terminal.report.*; 035 036/** 037 * A {@link TerminalViewSource} over one growing JSON Lines file, one event per line, paged through a 038 * {@link FileLineIndex}. 039 * 040 * <p> 041 * <b>Path safety.</b> The application supplies the {@link Path} when it builds the source; nothing in a request ever 042 * becomes a path. An application that derives paths from ids must resolve them against a fixed root, 043 * {@code normalize()} the result, and require {@code result.startsWith(root)}. 044 * 045 * <p> 046 * <b>Numbering.</b> {@code seq} is the 1-based line number. A {@code seq} member in the file is ignored and 047 * overwritten, so the producer writes events without one. A line that is not valid JSON, is not a known event, or is 048 * too long is skipped: it consumes its line number, so a gap appears in {@code seq}. The first such line a source 049 * meets is logged once as a {@code WARNING}, with its line number; later ones are skipped without a log record, so a 050 * corrupt file polled every second does not flood the log. Blank lines are not counted as malformed. 051 * 052 * <p> 053 * <b>Tokens</b> are {@code <epoch>.<n>}: the epoch of the file generation and the last line number consumed. A stale 054 * epoch (the file was truncated, replaced or rewritten) or an {@code n} beyond the indexed lines throws 055 * {@link TerminalViewSource.UnknownTokenException UnknownTokenException}. This includes a token issued while the file did 056 * not exist yet, once the file has been created. 057 * 058 * <p> 059 * Only complete lines are read, so a partially written last line never appears. Keep <b>one instance per file</b>; 060 * a new instance re-indexes the file from byte 0. 061 * 062 * <h5 class='section'>Example:</h5> 063 * <p class='bjava'> 064 * FileTerminalViewSource <jv>src</jv> = FileTerminalViewSource.<jsm>create</jsm>(<jv>root</jv>.resolve(<js>"run-42.jsonl"</js>)) 065 * .terminal(() -> <jv>run</jv>.isFinished()) 066 * .build(); 067 * </p> 068 */ 069@SuppressWarnings({ 070 "java:S135", // Guard clauses keep the loop readable. 071 "java:S3776" // Validation and rendering chains many independent guards; splitting them adds no clarity. 072}) 073public final class FileTerminalViewSource implements TerminalViewSource { 074 075 /** Lines longer than this many bytes are skipped without being decoded. */ 076 private static final int MAX_LINE_BYTES = TerminalEvent.MAX_EVENT_CHARS * 4; 077 private static final Logger LOG = Logger.getLogger(FileTerminalViewSource.class.getName()); 078 079 private static final Pattern TOKEN = Pattern.compile("^([0-9a-z]{6})\\.(0|[1-9]\\d{0,17})$"); 080 081 /** 082 * Builder for {@link FileTerminalViewSource}. 083 */ 084 public static final class Builder { 085 private final Path file; 086 private BooleanSupplier terminal = () -> false; 087 private FileLineIndex index; 088 089 Builder(Path file) { 090 this.file = Objects.requireNonNull(file, "file"); 091 } 092 093 /** 094 * Sets the supplier that reports whether the producer will write no more, called once per request before the 095 * file is read. 096 * 097 * @param value The supplier. 098 * @return This object. 099 */ 100 public Builder terminal(BooleanSupplier value) { 101 terminal = Objects.requireNonNull(value, "terminal"); 102 return this; 103 } 104 105 /** 106 * Sets the index (default {@link FileLineIndex#of(Path)}), so one index can be shared across requests. 107 * 108 * @param value An index over the same file. 109 * @return This object. 110 */ 111 public Builder index(FileLineIndex value) { 112 index = value; 113 return this; 114 } 115 116 /** 117 * Builds the source. 118 * 119 * @return A new source. 120 */ 121 public FileTerminalViewSource build() { 122 if (index != null && ! index.file().equals(file)) 123 throw iaex("FileTerminalViewSource index is for a different file: '%s' vs '%s'.", index.file(), file); 124 return new FileTerminalViewSource(this); 125 } 126 } 127 128 private final Path file; 129 private final BooleanSupplier terminal; 130 private final FileLineIndex index; 131 private final AtomicBoolean warned = new AtomicBoolean(); 132 133 private FileTerminalViewSource(Builder b) { 134 file = b.file; 135 terminal = b.terminal; 136 index = b.index != null ? b.index : FileLineIndex.of(b.file); 137 } 138 139 /** 140 * Creates a builder. 141 * 142 * @param file The file, supplied by the application. 143 * @return A new builder. 144 */ 145 public static Builder create(Path file) { 146 return new Builder(file); 147 } 148 149 @Override /* TerminalViewSource */ 150 public TerminalViewPage page(String after, int max) { 151 // Read the flag first: if the producer writes its last events and flips the flag after this point, the page 152 // is simply not terminal yet and the next request sees both. 153 var done = terminal.getAsBoolean(); 154 var s = refresh(); 155 var n = after == null ? 0L : position(after, s); 156 if (n >= s.indexedLines()) 157 return TerminalViewPage.of(List.of(), token(s, n), false, done); 158 var start = seek(n + 1); 159 if (start < 0) 160 throw new UnknownTokenException(after); 161 var limit = Math.max(1, Math.min(max, TerminalViewPage.MAX_PAGE_EVENTS)); 162 var events = new ArrayList<TerminalEvent>(); 163 var last = n; 164 try (var r = new LineReader(file, start, s.indexedBytes())) { 165 var chars = 0L; 166 var lineNo = n; 167 while (events.size() < limit) { 168 var text = r.next(); 169 if (text == null) 170 break; 171 lineNo++; 172 var e = parse(text, lineNo); 173 if (e != null) { 174 var c = serializedLength(e); 175 if (! events.isEmpty() && chars + c > TerminalViewPage.MAX_PAGE_CHARS) 176 break; 177 chars += c; 178 events.add(e); 179 } 180 last = lineNo; 181 } 182 } catch (IOException e) { 183 throw new UncheckedIOException(e); 184 } 185 var more = last < s.indexedLines(); 186 return TerminalViewPage.of(events, token(s, last), more, done && ! more); 187 } 188 189 private TerminalEvent parse(String text, long lineNo) { 190 if (text.isEmpty()) 191 return null; 192 if (text.length() > TerminalEvent.MAX_EVENT_CHARS) 193 return skipped(lineNo, "longer than " + TerminalEvent.MAX_EVENT_CHARS + " characters"); 194 try { 195 return TerminalEvent.fromMap(Json.to(text, Map.class)).withSeq(lineNo); 196 } catch (Exception e) { 197 return skipped(lineNo, e.getMessage()); 198 } 199 } 200 201 /** Logs the first skipped line of this source, and returns <jk>null</jk>. */ 202 private TerminalEvent skipped(long lineNo, String why) { 203 if (warned.compareAndSet(false, true)) 204 LOG.warning(() -> "FileTerminalViewSource '" + file + "' skipped malformed line " + lineNo + " (" + why 205 + "); later malformed lines in this source are skipped without a warning."); 206 return null; 207 } 208 209 private static int serializedLength(TerminalEvent e) { 210 return JsonSerializer.DEFAULT.toString(e.toContractMap()).length(); 211 } 212 213 private FileLineIndex.Snapshot refresh() { 214 try { 215 return index.refresh(); 216 } catch (IOException e) { 217 throw new UncheckedIOException(e); 218 } 219 } 220 221 private long seek(long n) { 222 try { 223 return index.seek(n); 224 } catch (IOException e) { 225 throw new UncheckedIOException(e); 226 } 227 } 228 229 private long position(String token, FileLineIndex.Snapshot s) { 230 var m = TOKEN.matcher(token); 231 if (! m.matches() || ! index.acceptsEpoch(m.group(1))) 232 throw new UnknownTokenException(token); 233 var n = Long.parseLong(m.group(2)); 234 if (n > s.indexedLines()) 235 throw new UnknownTokenException(token); 236 return n; 237 } 238 239 private static String token(FileLineIndex.Snapshot s, long n) { 240 return s.epoch() + "." + n; 241 } 242 243 /** Sequential reader of the complete lines in [start, end); opens the file lazily. */ 244 @SuppressWarnings({ 245 "resource" // The channel is opened lazily by next() and closed by close(). 246 }) 247 private static final class LineReader implements Closeable { 248 private final Path file; 249 private final long end; 250 private final ByteBuffer buf = ByteBuffer.allocate(FileLineIndex.BLOCK_BYTES); 251 private FileChannel ch; 252 private long pos; 253 private boolean first; 254 private boolean eof; 255 256 LineReader(Path file, long start, long end) { 257 this.file = file; 258 this.end = end; 259 pos = start; 260 first = start == 0; 261 buf.limit(0); 262 } 263 264 /** Returns the next line's text, {@code ""} for an unusable line (overlong or blank), or <jk>null</jk> at the end. */ 265 String next() throws IOException { 266 if (eof || pos >= end) 267 return null; 268 if (ch == null) { 269 try { 270 ch = FileChannel.open(file, StandardOpenOption.READ); 271 } catch (NoSuchFileException e) { 272 eof = true; 273 return null; 274 } 275 } 276 var acc = new ByteArrayOutputStream(); 277 var overlong = false; 278 while (pos < end) { 279 if (! buf.hasRemaining()) { 280 buf.clear(); 281 if (end - pos < buf.capacity()) 282 buf.limit((int)(end - pos)); 283 var r = ch.read(buf, pos); 284 buf.flip(); 285 if (r <= 0) { 286 eof = true; 287 return null; 288 } 289 } 290 var b = buf.get(); 291 pos++; 292 if (b == '\n') 293 break; 294 if (acc.size() < MAX_LINE_BYTES) 295 acc.write(b); 296 else 297 overlong = true; 298 } 299 var wasFirst = first; 300 first = false; 301 if (overlong) 302 return ""; 303 var bytes = acc.toByteArray(); 304 var from = wasFirst && bytes.length >= 3 && (bytes[0] & 0xFF) == 0xEF && (bytes[1] & 0xFF) == 0xBB && (bytes[2] & 0xFF) == 0xBF ? 3 : 0; 305 var len = bytes.length > from && bytes[bytes.length - 1] == '\r' ? bytes.length - 1 : bytes.length; 306 return new String(bytes, from, len - from, UTF_8); 307 } 308 309 @Override /* Closeable */ 310 public void close() { 311 if (ch == null) 312 return; 313 try { 314 ch.close(); 315 } catch (IOException e) { 316 // Read-only channel; nothing to recover. 317 } finally { 318 ch = null; 319 } 320 } 321 } 322}