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}