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.oozie.util;
020
021import java.io.BufferedReader;
022import java.io.IOException;
023import java.io.Writer;
024import java.text.ParseException;
025import java.text.SimpleDateFormat;
026import java.util.ArrayList;
027
028import org.apache.commons.lang.StringUtils;
029import org.apache.oozie.service.Services;
030import org.apache.oozie.service.XLogStreamingService;
031
032  /**
033 * Encapsulates the parsing and filtering of the log messages from a BufferedReader. It makes sure not to read in the entire log
034 * into memory at the same time; at most, it will have two messages (which can be multi-line in the case of exception stack traces).
035 * <p>
036 * To use this class: Calling {@link TimestampedMessageParser#increment()} will tell the parser to read the next message from the
037 * Reader. It will return true if there are more messages and false if not. Calling
038 * {@link TimestampedMessageParser#getLastMessage()} and {@link TimestampedMessageParser#getLastTimestamp()} will return the last
039 * message and timestamp, respectively, that were parsed when {@link TimestampedMessageParser#increment()} was called. Calling
040 * {@link TimestampedMessageParser#processRemaining(java.io.Writer)} will write the remaining log messages to the given Writer.
041 */
042public class TimestampedMessageParser {
043
044    protected BufferedReader reader;
045    private String nextLine = null;
046    private String lastTimestamp = null;
047    private XLogFilter filter;
048    private boolean empty = false;
049    private String lastMessage = null;
050    private boolean patternMatched = false;
051    public int count = 0;
052
053    /**
054     * Creates a TimestampedMessageParser with the given BufferedReader and filter.
055     *
056     * @param reader The BufferedReader to get the log messages from
057     * @param filter The filter
058     */
059    public TimestampedMessageParser(BufferedReader reader, XLogFilter filter) {
060        this.reader = reader;
061        this.filter = filter;
062        if (filter == null) {
063            filter = new XLogFilter();
064        }
065        filter.constructPattern();
066    }
067
068    /**
069     * Causes the next message and timestamp to be parsed from the BufferedReader.
070     *
071     * @return true if there are more messages left; false if not
072     * @throws IOException If there was a problem reading from the Reader
073     */
074    public boolean increment() throws IOException {
075        if (empty) {
076            return false;
077        }
078
079        StringBuilder message = new StringBuilder();
080
081        if (nextLine == null) {     // first time only
082            nextLine = parseNextLine();
083            if (nextLine == null) { // reader finished
084                empty = true;
085                return false;
086            }
087        }
088        lastTimestamp = parseTimestamp(nextLine);
089        String nextTimestamp = null;
090        while (nextTimestamp == null) {
091            message.append(nextLine).append("\n");
092            nextLine = parseNextLine();
093            if (nextLine != null) {
094                nextTimestamp = parseTimestamp(nextLine);   // exit loop if we have a timestamp, continue if not
095            }
096            else {                                          // reader finished
097                empty = true;
098                nextTimestamp = "";                         // exit loop
099            }
100        }
101
102        lastMessage = message.toString();
103        return true;
104    }
105
106    /**
107     * Returns the timestamp from the last message that was parsed.
108     *
109     * @return the timestamp from the last message that was parsed
110     */
111    public String getLastTimestamp() {
112        return lastTimestamp;
113    }
114
115    /**
116     * Returns the message that was last parsed.
117     *
118     * @return the message that was last parsed
119     */
120    public String getLastMessage() {
121        return lastMessage;
122    }
123
124    /**
125     * Closes the Reader.
126     *
127     * @throws IOException
128     */
129    public void closeReader() throws IOException {
130        reader.close();
131    }
132
133    /**
134     * Reads the next line from the Reader and checks if it matches the filter.  It can also handle multi-line messages (i.e.
135     * exception stack traces).  If it returns null, then there are no lines left in the Reader.
136     *
137     * @return The next line, or null
138     * @throws IOException
139     */
140    protected String parseNextLine() throws IOException {
141        String line;
142        while ((line = reader.readLine()) != null) {
143            ArrayList<String> logParts = filter.splitLogMessage(line);
144            if (logParts != null) {
145                patternMatched = filter.matches(logParts);
146            }
147            if (patternMatched) {
148                if (filter.getLogLimit() != -1) {
149                    if (logParts != null) {
150                        if (count >= filter.getLogLimit()) {
151                            return null;
152                        }
153                        count++;
154                    }
155                }
156                if (logParts != null) {
157                    if (filter.getEndDate() != null) {
158                        //Ignore the milli second part
159                        if (logParts.get(0).substring(0, 19).compareTo(filter.getFormattedEndDate()) > 0)
160                            return null;
161                    }
162                }
163                return line;
164            }
165        }
166        return line;
167    }
168
169    /**
170     * Parses the timestamp out of the passed in line.  If there isn't one, it returns null.
171     *
172     * @param line The line to check
173     * @return the timestamp of the line, or null
174     */
175    private String parseTimestamp(String line) {
176        String timestamp = null;
177        ArrayList<String> logParts = filter.splitLogMessage(line);
178        if (logParts != null) {
179            timestamp = logParts.get(0);
180        }
181        return timestamp;
182    }
183
184    /**
185     * Streams log messages to the passed in Writer. Flushes the log writing
186     * based on buffer len
187     *
188     * @param writer
189     * @param bufferLen maximum len of log buffer
190     * @param bytesWritten num bytes already written to writer
191     * @throws IOException
192     */
193    public void processRemaining(Writer writer, int bufferLen, int bytesWritten) throws IOException {
194        while (increment()) {
195            writer.write(lastMessage);
196            bytesWritten += lastMessage.length();
197            if (bytesWritten > bufferLen) {
198                writer.flush();
199                bytesWritten = 0;
200            }
201        }
202        writer.flush();
203    }
204
205    /**
206     * Streams log messages to the passed in Writer, with zero bytes already
207     * written
208     *
209     * @param writer
210     * @param bufferLen maximum len of log buffer
211     * @throws IOException
212     */
213    public void processRemaining(Writer writer, int bufferLen) throws IOException {
214        processRemaining(writer, bufferLen, 0);
215    }
216
217    /**
218     * Streams log messages to the passed in Writer, with default buffer len 4K
219     * and zero bytes already written
220     *
221     * @param writer
222     * @throws IOException
223     */
224    public void processRemaining(Writer writer) throws IOException {
225        processRemaining(writer, Services.get().get(XLogStreamingService.class).getBufferLen());
226    }
227
228}