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}