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