001/** 002* Licensed to the Apache Software Foundation (ASF) under one 003* or more contributor license agreements. See the NOTICE file 004* distributed with this work for additional information 005* regarding copyright ownership. The ASF licenses this file 006* to you under the Apache License, Version 2.0 (the 007* "License"); you may not use this file except in compliance 008* with the License. You may obtain a copy of the License at 009* 010* http://www.apache.org/licenses/LICENSE-2.0 011* 012* Unless required by applicable law or agreed to in writing, software 013* distributed under the License is distributed on an "AS IS" BASIS, 014* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015* See the License for the specific language governing permissions and 016* limitations under the License. 017*/ 018 019package org.apache.hadoop.yarn.logaggregation; 020 021import java.io.DataInput; 022import java.io.DataInputStream; 023import java.io.DataOutput; 024import java.io.DataOutputStream; 025import java.io.EOFException; 026import java.io.File; 027import java.io.FileInputStream; 028import java.io.IOException; 029import java.io.InputStreamReader; 030import java.io.OutputStream; 031import java.io.PrintStream; 032import java.io.Writer; 033import java.nio.charset.Charset; 034import java.security.PrivilegedExceptionAction; 035import java.util.ArrayList; 036import java.util.Arrays; 037import java.util.Collections; 038import java.util.EnumSet; 039import java.util.HashMap; 040import java.util.HashSet; 041import java.util.Iterator; 042import java.util.List; 043import java.util.Map; 044import java.util.Map.Entry; 045import java.util.Set; 046import java.util.regex.Pattern; 047 048import org.apache.commons.io.input.BoundedInputStream; 049import org.apache.commons.io.output.WriterOutputStream; 050import org.apache.commons.logging.Log; 051import org.apache.commons.logging.LogFactory; 052import org.apache.hadoop.classification.InterfaceAudience.Private; 053import org.apache.hadoop.classification.InterfaceAudience.Public; 054import org.apache.hadoop.classification.InterfaceStability.Evolving; 055import org.apache.hadoop.conf.Configuration; 056import org.apache.hadoop.fs.CreateFlag; 057import org.apache.hadoop.fs.FSDataInputStream; 058import org.apache.hadoop.fs.FSDataOutputStream; 059import org.apache.hadoop.fs.FileContext; 060import org.apache.hadoop.fs.Options; 061import org.apache.hadoop.fs.Path; 062import org.apache.hadoop.fs.permission.FsPermission; 063import org.apache.hadoop.io.IOUtils; 064import org.apache.hadoop.io.SecureIOUtils; 065import org.apache.hadoop.io.Writable; 066import org.apache.hadoop.io.file.tfile.TFile; 067import org.apache.hadoop.security.UserGroupInformation; 068import org.apache.hadoop.yarn.api.records.ApplicationAccessType; 069import org.apache.hadoop.yarn.api.records.ContainerId; 070import org.apache.hadoop.yarn.api.records.LogAggregationContext; 071import org.apache.hadoop.yarn.conf.YarnConfiguration; 072import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; 073import org.apache.hadoop.yarn.util.ConverterUtils; 074import org.apache.hadoop.yarn.util.Times; 075 076import com.google.common.annotations.VisibleForTesting; 077import com.google.common.base.Predicate; 078import com.google.common.collect.Iterables; 079import com.google.common.collect.Sets; 080 081@Public 082@Evolving 083public class AggregatedLogFormat { 084 085 private static final Log LOG = LogFactory.getLog(AggregatedLogFormat.class); 086 private static final LogKey APPLICATION_ACL_KEY = new LogKey("APPLICATION_ACL"); 087 private static final LogKey APPLICATION_OWNER_KEY = new LogKey("APPLICATION_OWNER"); 088 private static final LogKey VERSION_KEY = new LogKey("VERSION"); 089 private static final Map<String, LogKey> RESERVED_KEYS; 090 //Maybe write out the retention policy. 091 //Maybe write out a list of containerLogs skipped by the retention policy. 092 private static final int VERSION = 1; 093 094 /** 095 * Umask for the log file. 096 */ 097 private static final FsPermission APP_LOG_FILE_UMASK = FsPermission 098 .createImmutable((short) (0640 ^ 0777)); 099 100 101 static { 102 RESERVED_KEYS = new HashMap<String, AggregatedLogFormat.LogKey>(); 103 RESERVED_KEYS.put(APPLICATION_ACL_KEY.toString(), APPLICATION_ACL_KEY); 104 RESERVED_KEYS.put(APPLICATION_OWNER_KEY.toString(), APPLICATION_OWNER_KEY); 105 RESERVED_KEYS.put(VERSION_KEY.toString(), VERSION_KEY); 106 } 107 108 @Public 109 public static class LogKey implements Writable { 110 111 private String keyString; 112 113 public LogKey() { 114 115 } 116 117 public LogKey(ContainerId containerId) { 118 this.keyString = containerId.toString(); 119 } 120 121 public LogKey(String keyString) { 122 this.keyString = keyString; 123 } 124 125 @Override 126 public int hashCode() { 127 return keyString == null ? 0 : keyString.hashCode(); 128 } 129 130 @Override 131 public boolean equals(Object obj) { 132 if (obj instanceof LogKey) { 133 LogKey other = (LogKey) obj; 134 if (this.keyString == null) { 135 return other.keyString == null; 136 } 137 return this.keyString.equals(other.keyString); 138 } 139 return false; 140 } 141 142 @Private 143 @Override 144 public void write(DataOutput out) throws IOException { 145 out.writeUTF(this.keyString); 146 } 147 148 @Private 149 @Override 150 public void readFields(DataInput in) throws IOException { 151 this.keyString = in.readUTF(); 152 } 153 154 @Override 155 public String toString() { 156 return this.keyString; 157 } 158 } 159 160 @Private 161 public static class LogValue { 162 163 private final List<String> rootLogDirs; 164 private final ContainerId containerId; 165 private final String user; 166 private final LogAggregationContext logAggregationContext; 167 private Set<File> uploadedFiles = new HashSet<File>(); 168 private final Set<String> alreadyUploadedLogFiles; 169 private Set<String> allExistingFileMeta = new HashSet<String>(); 170 private final boolean appFinished; 171 // TODO Maybe add a version string here. Instead of changing the version of 172 // the entire k-v format 173 174 public LogValue(List<String> rootLogDirs, ContainerId containerId, 175 String user) { 176 this(rootLogDirs, containerId, user, null, new HashSet<String>(), true); 177 } 178 179 public LogValue(List<String> rootLogDirs, ContainerId containerId, 180 String user, LogAggregationContext logAggregationContext, 181 Set<String> alreadyUploadedLogFiles, boolean appFinished) { 182 this.rootLogDirs = new ArrayList<String>(rootLogDirs); 183 this.containerId = containerId; 184 this.user = user; 185 186 // Ensure logs are processed in lexical order 187 Collections.sort(this.rootLogDirs); 188 this.logAggregationContext = logAggregationContext; 189 this.alreadyUploadedLogFiles = alreadyUploadedLogFiles; 190 this.appFinished = appFinished; 191 } 192 193 private Set<File> getPendingLogFilesToUploadForThisContainer() { 194 Set<File> pendingUploadFiles = new HashSet<File>(); 195 for (String rootLogDir : this.rootLogDirs) { 196 File appLogDir = 197 new File(rootLogDir, 198 ConverterUtils.toString( 199 this.containerId.getApplicationAttemptId(). 200 getApplicationId()) 201 ); 202 File containerLogDir = 203 new File(appLogDir, ConverterUtils.toString(this.containerId)); 204 205 if (!containerLogDir.isDirectory()) { 206 continue; // ContainerDir may have been deleted by the user. 207 } 208 209 pendingUploadFiles 210 .addAll(getPendingLogFilesToUpload(containerLogDir)); 211 } 212 return pendingUploadFiles; 213 } 214 215 public void write(DataOutputStream out, Set<File> pendingUploadFiles) 216 throws IOException { 217 List<File> fileList = new ArrayList<File>(pendingUploadFiles); 218 Collections.sort(fileList); 219 220 for (File logFile : fileList) { 221 // We only aggregate top level files. 222 // Ignore anything inside sub-folders. 223 if (logFile.isDirectory()) { 224 LOG.warn(logFile.getAbsolutePath() + " is a directory. Ignore it."); 225 continue; 226 } 227 228 FileInputStream in = null; 229 try { 230 in = secureOpenFile(logFile); 231 } catch (IOException e) { 232 logErrorMessage(logFile, e); 233 IOUtils.cleanup(LOG, in); 234 continue; 235 } 236 237 final long fileLength = logFile.length(); 238 // Write the logFile Type 239 out.writeUTF(logFile.getName()); 240 241 // Write the log length as UTF so that it is printable 242 out.writeUTF(String.valueOf(fileLength)); 243 244 // Write the log itself 245 try { 246 byte[] buf = new byte[65535]; 247 int len = 0; 248 long bytesLeft = fileLength; 249 while ((len = in.read(buf)) != -1) { 250 //If buffer contents within fileLength, write 251 if (len < bytesLeft) { 252 out.write(buf, 0, len); 253 bytesLeft-=len; 254 } 255 //else only write contents within fileLength, then exit early 256 else { 257 out.write(buf, 0, (int)bytesLeft); 258 break; 259 } 260 } 261 long newLength = logFile.length(); 262 if(fileLength < newLength) { 263 LOG.warn("Aggregated logs truncated by approximately "+ 264 (newLength-fileLength) +" bytes."); 265 } 266 this.uploadedFiles.add(logFile); 267 } catch (IOException e) { 268 String message = logErrorMessage(logFile, e); 269 out.write(message.getBytes(Charset.forName("UTF-8"))); 270 } finally { 271 IOUtils.cleanup(LOG, in); 272 } 273 } 274 } 275 276 @VisibleForTesting 277 public FileInputStream secureOpenFile(File logFile) throws IOException { 278 return SecureIOUtils.openForRead(logFile, getUser(), null); 279 } 280 281 private static String logErrorMessage(File logFile, Exception e) { 282 String message = "Error aggregating log file. Log file : " 283 + logFile.getAbsolutePath() + ". " + e.getMessage(); 284 LOG.error(message, e); 285 return message; 286 } 287 288 // Added for testing purpose. 289 public String getUser() { 290 return user; 291 } 292 293 private Set<File> getPendingLogFilesToUpload(File containerLogDir) { 294 Set<File> candidates = 295 new HashSet<File>(Arrays.asList(containerLogDir.listFiles())); 296 for (File logFile : candidates) { 297 this.allExistingFileMeta.add(getLogFileMetaData(logFile)); 298 } 299 300 if (this.logAggregationContext != null && candidates.size() > 0) { 301 filterFiles( 302 this.appFinished ? this.logAggregationContext.getIncludePattern() 303 : this.logAggregationContext.getRolledLogsIncludePattern(), 304 candidates, false); 305 306 filterFiles( 307 this.appFinished ? this.logAggregationContext.getExcludePattern() 308 : this.logAggregationContext.getRolledLogsExcludePattern(), 309 candidates, true); 310 311 Iterable<File> mask = 312 Iterables.filter(candidates, new Predicate<File>() { 313 @Override 314 public boolean apply(File next) { 315 return !alreadyUploadedLogFiles 316 .contains(getLogFileMetaData(next)); 317 } 318 }); 319 candidates = Sets.newHashSet(mask); 320 } 321 return candidates; 322 } 323 324 private void filterFiles(String pattern, Set<File> candidates, 325 boolean exclusion) { 326 if (pattern != null && !pattern.isEmpty()) { 327 Pattern filterPattern = Pattern.compile(pattern); 328 for (Iterator<File> candidatesItr = candidates.iterator(); candidatesItr 329 .hasNext();) { 330 File candidate = candidatesItr.next(); 331 boolean match = filterPattern.matcher(candidate.getName()).find(); 332 if ((!match && !exclusion) || (match && exclusion)) { 333 candidatesItr.remove(); 334 } 335 } 336 } 337 } 338 339 public Set<Path> getCurrentUpLoadedFilesPath() { 340 Set<Path> path = new HashSet<Path>(); 341 for (File file : this.uploadedFiles) { 342 path.add(new Path(file.getAbsolutePath())); 343 } 344 return path; 345 } 346 347 public Set<String> getCurrentUpLoadedFileMeta() { 348 Set<String> info = new HashSet<String>(); 349 for (File file : this.uploadedFiles) { 350 info.add(getLogFileMetaData(file)); 351 } 352 return info; 353 } 354 355 public Set<String> getAllExistingFilesMeta() { 356 return this.allExistingFileMeta; 357 } 358 359 private String getLogFileMetaData(File file) { 360 return containerId.toString() + "_" + file.getName() + "_" 361 + file.lastModified(); 362 } 363 } 364 365 /** 366 * The writer that writes out the aggregated logs. 367 */ 368 @Private 369 public static class LogWriter { 370 371 private final FSDataOutputStream fsDataOStream; 372 private final TFile.Writer writer; 373 private FileContext fc; 374 375 public LogWriter(final Configuration conf, final Path remoteAppLogFile, 376 UserGroupInformation userUgi) throws IOException { 377 try { 378 this.fsDataOStream = 379 userUgi.doAs(new PrivilegedExceptionAction<FSDataOutputStream>() { 380 @Override 381 public FSDataOutputStream run() throws Exception { 382 fc = FileContext.getFileContext(remoteAppLogFile.toUri(), conf); 383 fc.setUMask(APP_LOG_FILE_UMASK); 384 return fc.create( 385 remoteAppLogFile, 386 EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE), 387 new Options.CreateOpts[] {}); 388 } 389 }); 390 } catch (InterruptedException e) { 391 throw new IOException(e); 392 } 393 394 // Keys are not sorted: null arg 395 // 256KB minBlockSize : Expected log size for each container too 396 this.writer = 397 new TFile.Writer(this.fsDataOStream, 256 * 1024, conf.get( 398 YarnConfiguration.NM_LOG_AGG_COMPRESSION_TYPE, 399 YarnConfiguration.DEFAULT_NM_LOG_AGG_COMPRESSION_TYPE), null, conf); 400 //Write the version string 401 writeVersion(); 402 } 403 404 @VisibleForTesting 405 public TFile.Writer getWriter() { 406 return this.writer; 407 } 408 409 private void writeVersion() throws IOException { 410 try (DataOutputStream out = this.writer.prepareAppendKey(-1)) { 411 VERSION_KEY.write(out); 412 } 413 try (DataOutputStream out = this.writer.prepareAppendValue(-1)) { 414 out.writeInt(VERSION); 415 } 416 } 417 418 public void writeApplicationOwner(String user) throws IOException { 419 try (DataOutputStream out = this.writer.prepareAppendKey(-1)) { 420 APPLICATION_OWNER_KEY.write(out); 421 } 422 try (DataOutputStream out = this.writer.prepareAppendValue(-1)) { 423 out.writeUTF(user); 424 } 425 } 426 427 public void writeApplicationACLs(Map<ApplicationAccessType, String> appAcls) 428 throws IOException { 429 try (DataOutputStream out = this.writer.prepareAppendKey(-1)) { 430 APPLICATION_ACL_KEY.write(out); 431 } 432 try (DataOutputStream out = this.writer.prepareAppendValue(-1)) { 433 for (Entry<ApplicationAccessType, String> entry : appAcls.entrySet()) { 434 out.writeUTF(entry.getKey().toString()); 435 out.writeUTF(entry.getValue()); 436 } 437 } 438 } 439 440 public void append(LogKey logKey, LogValue logValue) throws IOException { 441 Set<File> pendingUploadFiles = 442 logValue.getPendingLogFilesToUploadForThisContainer(); 443 if (pendingUploadFiles.size() == 0) { 444 return; 445 } 446 try (DataOutputStream out = this.writer.prepareAppendKey(-1)) { 447 logKey.write(out); 448 } 449 try (DataOutputStream out = this.writer.prepareAppendValue(-1)) { 450 logValue.write(out, pendingUploadFiles); 451 } 452 } 453 454 public void close() { 455 try { 456 this.writer.close(); 457 } catch (IOException e) { 458 LOG.warn("Exception closing writer", e); 459 } 460 IOUtils.closeStream(fsDataOStream); 461 } 462 } 463 464 @Public 465 @Evolving 466 public static class LogReader { 467 468 private final FSDataInputStream fsDataIStream; 469 private final TFile.Reader.Scanner scanner; 470 private final TFile.Reader reader; 471 472 public LogReader(Configuration conf, Path remoteAppLogFile) 473 throws IOException { 474 FileContext fileContext = 475 FileContext.getFileContext(remoteAppLogFile.toUri(), conf); 476 this.fsDataIStream = fileContext.open(remoteAppLogFile); 477 reader = 478 new TFile.Reader(this.fsDataIStream, fileContext.getFileStatus( 479 remoteAppLogFile).getLen(), conf); 480 this.scanner = reader.createScanner(); 481 } 482 483 private boolean atBeginning = true; 484 485 /** 486 * Returns the owner of the application. 487 * 488 * @return the application owner. 489 * @throws IOException 490 */ 491 public String getApplicationOwner() throws IOException { 492 TFile.Reader.Scanner ownerScanner = reader.createScanner(); 493 LogKey key = new LogKey(); 494 while (!ownerScanner.atEnd()) { 495 TFile.Reader.Scanner.Entry entry = ownerScanner.entry(); 496 key.readFields(entry.getKeyStream()); 497 if (key.toString().equals(APPLICATION_OWNER_KEY.toString())) { 498 DataInputStream valueStream = entry.getValueStream(); 499 return valueStream.readUTF(); 500 } 501 ownerScanner.advance(); 502 } 503 return null; 504 } 505 506 /** 507 * Returns ACLs for the application. An empty map is returned if no ACLs are 508 * found. 509 * 510 * @return a map of the Application ACLs. 511 * @throws IOException 512 */ 513 public Map<ApplicationAccessType, String> getApplicationAcls() 514 throws IOException { 515 // TODO Seek directly to the key once a comparator is specified. 516 TFile.Reader.Scanner aclScanner = reader.createScanner(); 517 LogKey key = new LogKey(); 518 Map<ApplicationAccessType, String> acls = 519 new HashMap<ApplicationAccessType, String>(); 520 while (!aclScanner.atEnd()) { 521 TFile.Reader.Scanner.Entry entry = aclScanner.entry(); 522 key.readFields(entry.getKeyStream()); 523 if (key.toString().equals(APPLICATION_ACL_KEY.toString())) { 524 DataInputStream valueStream = entry.getValueStream(); 525 while (true) { 526 String appAccessOp = null; 527 String aclString = null; 528 try { 529 appAccessOp = valueStream.readUTF(); 530 } catch (EOFException e) { 531 // Valid end of stream. 532 break; 533 } 534 try { 535 aclString = valueStream.readUTF(); 536 } catch (EOFException e) { 537 throw new YarnRuntimeException("Error reading ACLs", e); 538 } 539 acls.put(ApplicationAccessType.valueOf(appAccessOp), aclString); 540 } 541 542 } 543 aclScanner.advance(); 544 } 545 return acls; 546 } 547 548 /** 549 * Read the next key and return the value-stream. 550 * 551 * @param key 552 * @return the valueStream if there are more keys or null otherwise. 553 * @throws IOException 554 */ 555 public DataInputStream next(LogKey key) throws IOException { 556 if (!this.atBeginning) { 557 this.scanner.advance(); 558 } else { 559 this.atBeginning = false; 560 } 561 if (this.scanner.atEnd()) { 562 return null; 563 } 564 TFile.Reader.Scanner.Entry entry = this.scanner.entry(); 565 key.readFields(entry.getKeyStream()); 566 // Skip META keys 567 if (RESERVED_KEYS.containsKey(key.toString())) { 568 return next(key); 569 } 570 DataInputStream valueStream = entry.getValueStream(); 571 return valueStream; 572 } 573 574 /** 575 * Get a ContainerLogsReader to read the logs for 576 * the specified container. 577 * 578 * @param containerId 579 * @return object to read the container's logs or null if the 580 * logs could not be found 581 * @throws IOException 582 */ 583 @Private 584 public ContainerLogsReader getContainerLogsReader( 585 ContainerId containerId) throws IOException { 586 ContainerLogsReader logReader = null; 587 588 final LogKey containerKey = new LogKey(containerId); 589 LogKey key = new LogKey(); 590 DataInputStream valueStream = next(key); 591 while (valueStream != null && !key.equals(containerKey)) { 592 valueStream = next(key); 593 } 594 595 if (valueStream != null) { 596 logReader = new ContainerLogsReader(valueStream); 597 } 598 599 return logReader; 600 } 601 602 //TODO Change Log format and interfaces to be containerId specific. 603 // Avoid returning completeValueStreams. 604// public List<String> getTypesForContainer(DataInputStream valueStream){} 605// 606// /** 607// * @param valueStream 608// * The Log stream for the container. 609// * @param fileType 610// * the log type required. 611// * @return An InputStreamReader for the required log type or null if the 612// * type is not found. 613// * @throws IOException 614// */ 615// public InputStreamReader getLogStreamForType(DataInputStream valueStream, 616// String fileType) throws IOException { 617// valueStream.reset(); 618// try { 619// while (true) { 620// String ft = valueStream.readUTF(); 621// String fileLengthStr = valueStream.readUTF(); 622// long fileLength = Long.parseLong(fileLengthStr); 623// if (ft.equals(fileType)) { 624// BoundedInputStream bis = 625// new BoundedInputStream(valueStream, fileLength); 626// return new InputStreamReader(bis); 627// } else { 628// long totalSkipped = 0; 629// long currSkipped = 0; 630// while (currSkipped != -1 && totalSkipped < fileLength) { 631// currSkipped = valueStream.skip(fileLength - totalSkipped); 632// totalSkipped += currSkipped; 633// } 634// // TODO Verify skip behaviour. 635// if (currSkipped == -1) { 636// return null; 637// } 638// } 639// } 640// } catch (EOFException e) { 641// return null; 642// } 643// } 644 645 /** 646 * Writes all logs for a single container to the provided writer. 647 * @param valueStream 648 * @param writer 649 * @param logUploadedTime 650 * @throws IOException 651 */ 652 public static void readAcontainerLogs(DataInputStream valueStream, 653 Writer writer, long logUploadedTime) throws IOException { 654 OutputStream os = null; 655 PrintStream ps = null; 656 try { 657 os = new WriterOutputStream(writer, Charset.forName("UTF-8")); 658 ps = new PrintStream(os); 659 while (true) { 660 try { 661 readContainerLogs(valueStream, ps, logUploadedTime); 662 } catch (EOFException e) { 663 // EndOfFile 664 return; 665 } 666 } 667 } finally { 668 IOUtils.cleanup(LOG, ps); 669 IOUtils.cleanup(LOG, os); 670 } 671 } 672 673 /** 674 * Writes all logs for a single container to the provided writer. 675 * @param valueStream 676 * @param writer 677 * @throws IOException 678 */ 679 public static void readAcontainerLogs(DataInputStream valueStream, 680 Writer writer) throws IOException { 681 readAcontainerLogs(valueStream, writer, -1); 682 } 683 684 private static void readContainerLogs(DataInputStream valueStream, 685 PrintStream out, long logUploadedTime) throws IOException { 686 byte[] buf = new byte[65535]; 687 688 String fileType = valueStream.readUTF(); 689 String fileLengthStr = valueStream.readUTF(); 690 long fileLength = Long.parseLong(fileLengthStr); 691 out.print("LogType:"); 692 out.println(fileType); 693 if (logUploadedTime != -1) { 694 out.print("Log Upload Time:"); 695 out.println(Times.format(logUploadedTime)); 696 } 697 out.print("LogLength:"); 698 out.println(fileLengthStr); 699 out.println("Log Contents:"); 700 701 long curRead = 0; 702 long pendingRead = fileLength - curRead; 703 int toRead = 704 pendingRead > buf.length ? buf.length : (int) pendingRead; 705 int len = valueStream.read(buf, 0, toRead); 706 while (len != -1 && curRead < fileLength) { 707 out.write(buf, 0, len); 708 curRead += len; 709 710 pendingRead = fileLength - curRead; 711 toRead = 712 pendingRead > buf.length ? buf.length : (int) pendingRead; 713 len = valueStream.read(buf, 0, toRead); 714 } 715 out.println("End of LogType:" + fileType); 716 out.println(""); 717 } 718 719 /** 720 * Keep calling this till you get a {@link EOFException} for getting logs of 721 * all types for a single container. 722 * 723 * @param valueStream 724 * @param out 725 * @param logUploadedTime 726 * @throws IOException 727 */ 728 public static void readAContainerLogsForALogType( 729 DataInputStream valueStream, PrintStream out, long logUploadedTime) 730 throws IOException { 731 readContainerLogs(valueStream, out, logUploadedTime); 732 } 733 734 /** 735 * Keep calling this till you get a {@link EOFException} for getting logs of 736 * all types for a single container. 737 * 738 * @param valueStream 739 * @param out 740 * @throws IOException 741 */ 742 public static void readAContainerLogsForALogType( 743 DataInputStream valueStream, PrintStream out) 744 throws IOException { 745 readAContainerLogsForALogType(valueStream, out, -1); 746 } 747 748 public void close() { 749 IOUtils.cleanup(LOG, scanner, reader, fsDataIStream); 750 } 751 } 752 753 @Private 754 public static class ContainerLogsReader { 755 private DataInputStream valueStream; 756 private String currentLogType = null; 757 private long currentLogLength = 0; 758 private BoundedInputStream currentLogData = null; 759 private InputStreamReader currentLogISR; 760 761 public ContainerLogsReader(DataInputStream stream) { 762 valueStream = stream; 763 } 764 765 public String nextLog() throws IOException { 766 if (currentLogData != null && currentLogLength > 0) { 767 // seek to the end of the current log, relying on BoundedInputStream 768 // to prevent seeking past the end of the current log 769 do { 770 if (currentLogData.skip(currentLogLength) < 0) { 771 break; 772 } 773 } while (currentLogData.read() != -1); 774 } 775 776 currentLogType = null; 777 currentLogLength = 0; 778 currentLogData = null; 779 currentLogISR = null; 780 781 try { 782 String logType = valueStream.readUTF(); 783 String logLengthStr = valueStream.readUTF(); 784 currentLogLength = Long.parseLong(logLengthStr); 785 currentLogData = 786 new BoundedInputStream(valueStream, currentLogLength); 787 currentLogData.setPropagateClose(false); 788 currentLogISR = new InputStreamReader(currentLogData, 789 Charset.forName("UTF-8")); 790 currentLogType = logType; 791 } catch (EOFException e) { 792 } 793 794 return currentLogType; 795 } 796 797 public String getCurrentLogType() { 798 return currentLogType; 799 } 800 801 public long getCurrentLogLength() { 802 return currentLogLength; 803 } 804 805 public long skip(long n) throws IOException { 806 return currentLogData.skip(n); 807 } 808 809 public int read() throws IOException { 810 return currentLogData.read(); 811 } 812 813 public int read(byte[] buf, int off, int len) throws IOException { 814 return currentLogData.read(buf, off, len); 815 } 816 817 public int read(char[] buf, int off, int len) throws IOException { 818 return currentLogISR.read(buf, off, len); 819 } 820 } 821}