@Test public void testReadLogMessageAndPosition() throws Exception { String path = FilenameUtils.concat(getTempPath(), "thrift2.log"); String tempPath = getTempPath(); SingerLogConfig singerLogConfig = new SingerLogConfig("test", tempPath, "thrift2.log", null, null, null); singerLogConfig.setFilenameMatchMode(FileNameMatchMode.PREFIX); SingerLog singerLog = new SingerLog(singerLogConfig); SingerSettings.setSingerConfig(new SingerConfig()); SingerSettings.getOrCreateFileSystemMonitor("").start(); SingerSettings.getOrCreateFileSystemMonitor("").registerPath(new File(tempPath).toPath()); LogStream logStream = new LogStream(singerLog, "thrift2.log"); LogStreamManager.addLogStream(logStream); SimpleThriftLogger<LogMessage> logger = new SimpleThriftLogger<>(path); LogPosition startPosition = null; int numOfLogRotations = 4; int numOfLogMessagesPerLogFile = 300; List<LogMessageAndPosition> messagesWritten = Lists .newArrayListWithExpectedSize(numOfLogRotations * numOfLogMessagesPerLogFile); try { writeThriftLogMessages(logger, 150, 500, 50); startPosition = new LogPosition(logger.getLogFile(), logger.getByteOffset()); for (int i = 0; i < numOfLogRotations - 1; ++i) { rotateWithDelay(logger, 1000); messagesWritten.addAll(writeThriftLogMessages(logger, numOfLogMessagesPerLogFile, 500, 50)); } writeThriftLogMessages(logger, 1, 500, 16001); messagesWritten.addAll(writeThriftLogMessages(logger, 10, 500, 500)); rotateWithDelay(logger, 1000); messagesWritten.addAll(writeThriftLogMessages(logger, numOfLogMessagesPerLogFile, 500, 50)); writeThriftLogMessages(logger, 1, 500, 160001); writeThriftLogMessages(logger, 10, 500, 500); } finally { logger.close(); } System.err.println("Waiting for file system events to be noticed by FSM"); while(logStream.isEmpty()) { Thread.sleep(1000); System.out.print("."); } DefaultLogStreamReader reader = new DefaultLogStreamReader(logStream, new ThriftLogFileReaderFactory(new ThriftReaderConfig(16000, 16000))); List<LogMessageAndPosition> messagesRead = Lists.newArrayListWithExpectedSize( numOfLogRotations * numOfLogMessagesPerLogFile); try { reader.seek(startPosition); LogMessageAndPosition message = reader.readLogMessageAndPosition(); while (message != null) { messagesRead.add(message); message = reader.readLogMessageAndPosition(); } } finally { reader.close(); } System.out.println( "# messages read = " + messagesRead.size() + " # messages written = " + messagesWritten .size()); assertThat(String.format("Wrote %d messages, read %d.", messagesWritten.size(), messagesRead.size()), messagesRead, is(messagesWritten)); }
@Override public LogMessageAndPosition readLogMessageAndPosition() throws LogStreamReaderException { if (closed) { throw new LogStreamReaderException("Reader closed."); } if (fileReader == null) { return null; } try { LogMessageAndPosition messageAndPosition = null; LogFile currentLogFile = fileReader.getLogFile(); long startByteOffset = fileReader.getByteOffset(); try { messageAndPosition = fileReader.readLogMessageAndPosition(); long bytesRead = fileReader.getByteOffset() - startByteOffset; LOG.trace("Read {} bytes from {}", bytesRead, fileReader.getLogFile()); } catch (Exception e) { LOG.error("Caught exception while reading from log file: {}", currentLogFile, e); } if (messageAndPosition == null) { fileReader.close(); fileReader = null; LogFileAndPath nextLogFileAndPath = logStream.getNext(currentLogFile); if (nextLogFileAndPath == null) { LOG.info("No more message in the current log file and it is the end of log stream: {}", logStream); return null; } fileReader = getNextLogFileReader(nextLogFileAndPath); if (fileReader == null) { Stats.incr("singer.processor.files.missing"); return null; } LOG.info("Continue to read from next log file {} in log stream {}", fileReader.getLogFile(), logStream.getLogStreamDescriptor()); long currentFileLength = new File(logStream.getLogFilePath(currentLogFile)).length(); long bytesSkipped = currentFileLength - startByteOffset; if (bytesSkipped > 0) { LOG.error("Skipped {} bytes in log file: {} in log stream: {}", bytesSkipped, currentLogFile, logStream); OpenTsdbMetricConverter.incr(SingerMetrics.SKIPPED_BYTES, (int) bytesSkipped, "log=" + logStream.getSingerLog().getSingerLogConfig().getName(), "host=" + SingerUtils.getHostname()); } return readLogMessageAndPosition(); } else { return messageAndPosition; } } catch (Exception e) { LOG.error("Caught exception while reading log message from log stream: {}", logStream, e); try { if (fileReader != null) { fileReader.close(); } fileReader = null; LOG.info("Closed fileReader when failed to seek on DefaultLogStreamReader"); } catch (IOException ex) { throw new LogStreamReaderException("Can not close fileReader.", ex); } throw new LogStreamReaderException( "Can not read a log message from current position in log stream: " + logStream .getLogStreamDescriptor(), e); } }
DefaultLogStreamReader implements LogStreamReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogStreamReaderException { if (closed) { throw new LogStreamReaderException("Reader closed."); } if (fileReader == null) { return null; } try { LogMessageAndPosition messageAndPosition = null; LogFile currentLogFile = fileReader.getLogFile(); long startByteOffset = fileReader.getByteOffset(); try { messageAndPosition = fileReader.readLogMessageAndPosition(); long bytesRead = fileReader.getByteOffset() - startByteOffset; LOG.trace("Read {} bytes from {}", bytesRead, fileReader.getLogFile()); } catch (Exception e) { LOG.error("Caught exception while reading from log file: {}", currentLogFile, e); } if (messageAndPosition == null) { fileReader.close(); fileReader = null; LogFileAndPath nextLogFileAndPath = logStream.getNext(currentLogFile); if (nextLogFileAndPath == null) { LOG.info("No more message in the current log file and it is the end of log stream: {}", logStream); return null; } fileReader = getNextLogFileReader(nextLogFileAndPath); if (fileReader == null) { Stats.incr("singer.processor.files.missing"); return null; } LOG.info("Continue to read from next log file {} in log stream {}", fileReader.getLogFile(), logStream.getLogStreamDescriptor()); long currentFileLength = new File(logStream.getLogFilePath(currentLogFile)).length(); long bytesSkipped = currentFileLength - startByteOffset; if (bytesSkipped > 0) { LOG.error("Skipped {} bytes in log file: {} in log stream: {}", bytesSkipped, currentLogFile, logStream); OpenTsdbMetricConverter.incr(SingerMetrics.SKIPPED_BYTES, (int) bytesSkipped, "log=" + logStream.getSingerLog().getSingerLogConfig().getName(), "host=" + SingerUtils.getHostname()); } return readLogMessageAndPosition(); } else { return messageAndPosition; } } catch (Exception e) { LOG.error("Caught exception while reading log message from log stream: {}", logStream, e); try { if (fileReader != null) { fileReader.close(); } fileReader = null; LOG.info("Closed fileReader when failed to seek on DefaultLogStreamReader"); } catch (IOException ex) { throw new LogStreamReaderException("Can not close fileReader.", ex); } throw new LogStreamReaderException( "Can not read a log message from current position in log stream: " + logStream .getLogStreamDescriptor(), e); } } }
DefaultLogStreamReader implements LogStreamReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogStreamReaderException { if (closed) { throw new LogStreamReaderException("Reader closed."); } if (fileReader == null) { return null; } try { LogMessageAndPosition messageAndPosition = null; LogFile currentLogFile = fileReader.getLogFile(); long startByteOffset = fileReader.getByteOffset(); try { messageAndPosition = fileReader.readLogMessageAndPosition(); long bytesRead = fileReader.getByteOffset() - startByteOffset; LOG.trace("Read {} bytes from {}", bytesRead, fileReader.getLogFile()); } catch (Exception e) { LOG.error("Caught exception while reading from log file: {}", currentLogFile, e); } if (messageAndPosition == null) { fileReader.close(); fileReader = null; LogFileAndPath nextLogFileAndPath = logStream.getNext(currentLogFile); if (nextLogFileAndPath == null) { LOG.info("No more message in the current log file and it is the end of log stream: {}", logStream); return null; } fileReader = getNextLogFileReader(nextLogFileAndPath); if (fileReader == null) { Stats.incr("singer.processor.files.missing"); return null; } LOG.info("Continue to read from next log file {} in log stream {}", fileReader.getLogFile(), logStream.getLogStreamDescriptor()); long currentFileLength = new File(logStream.getLogFilePath(currentLogFile)).length(); long bytesSkipped = currentFileLength - startByteOffset; if (bytesSkipped > 0) { LOG.error("Skipped {} bytes in log file: {} in log stream: {}", bytesSkipped, currentLogFile, logStream); OpenTsdbMetricConverter.incr(SingerMetrics.SKIPPED_BYTES, (int) bytesSkipped, "log=" + logStream.getSingerLog().getSingerLogConfig().getName(), "host=" + SingerUtils.getHostname()); } return readLogMessageAndPosition(); } else { return messageAndPosition; } } catch (Exception e) { LOG.error("Caught exception while reading log message from log stream: {}", logStream, e); try { if (fileReader != null) { fileReader.close(); } fileReader = null; LOG.info("Closed fileReader when failed to seek on DefaultLogStreamReader"); } catch (IOException ex) { throw new LogStreamReaderException("Can not close fileReader.", ex); } throw new LogStreamReaderException( "Can not read a log message from current position in log stream: " + logStream .getLogStreamDescriptor(), e); } } DefaultLogStreamReader(LogStream logStream, LogFileReaderFactory fileReaderFactory); }
DefaultLogStreamReader implements LogStreamReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogStreamReaderException { if (closed) { throw new LogStreamReaderException("Reader closed."); } if (fileReader == null) { return null; } try { LogMessageAndPosition messageAndPosition = null; LogFile currentLogFile = fileReader.getLogFile(); long startByteOffset = fileReader.getByteOffset(); try { messageAndPosition = fileReader.readLogMessageAndPosition(); long bytesRead = fileReader.getByteOffset() - startByteOffset; LOG.trace("Read {} bytes from {}", bytesRead, fileReader.getLogFile()); } catch (Exception e) { LOG.error("Caught exception while reading from log file: {}", currentLogFile, e); } if (messageAndPosition == null) { fileReader.close(); fileReader = null; LogFileAndPath nextLogFileAndPath = logStream.getNext(currentLogFile); if (nextLogFileAndPath == null) { LOG.info("No more message in the current log file and it is the end of log stream: {}", logStream); return null; } fileReader = getNextLogFileReader(nextLogFileAndPath); if (fileReader == null) { Stats.incr("singer.processor.files.missing"); return null; } LOG.info("Continue to read from next log file {} in log stream {}", fileReader.getLogFile(), logStream.getLogStreamDescriptor()); long currentFileLength = new File(logStream.getLogFilePath(currentLogFile)).length(); long bytesSkipped = currentFileLength - startByteOffset; if (bytesSkipped > 0) { LOG.error("Skipped {} bytes in log file: {} in log stream: {}", bytesSkipped, currentLogFile, logStream); OpenTsdbMetricConverter.incr(SingerMetrics.SKIPPED_BYTES, (int) bytesSkipped, "log=" + logStream.getSingerLog().getSingerLogConfig().getName(), "host=" + SingerUtils.getHostname()); } return readLogMessageAndPosition(); } else { return messageAndPosition; } } catch (Exception e) { LOG.error("Caught exception while reading log message from log stream: {}", logStream, e); try { if (fileReader != null) { fileReader.close(); } fileReader = null; LOG.info("Closed fileReader when failed to seek on DefaultLogStreamReader"); } catch (IOException ex) { throw new LogStreamReaderException("Can not close fileReader.", ex); } throw new LogStreamReaderException( "Can not read a log message from current position in log stream: " + logStream .getLogStreamDescriptor(), e); } } DefaultLogStreamReader(LogStream logStream, LogFileReaderFactory fileReaderFactory); @Override LogStream getLogStream(); @Override LogPosition getPos(); @Override void seek(LogPosition position); @Override LogMessageAndPosition readLogMessageAndPosition(); @Override void close(); }
DefaultLogStreamReader implements LogStreamReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogStreamReaderException { if (closed) { throw new LogStreamReaderException("Reader closed."); } if (fileReader == null) { return null; } try { LogMessageAndPosition messageAndPosition = null; LogFile currentLogFile = fileReader.getLogFile(); long startByteOffset = fileReader.getByteOffset(); try { messageAndPosition = fileReader.readLogMessageAndPosition(); long bytesRead = fileReader.getByteOffset() - startByteOffset; LOG.trace("Read {} bytes from {}", bytesRead, fileReader.getLogFile()); } catch (Exception e) { LOG.error("Caught exception while reading from log file: {}", currentLogFile, e); } if (messageAndPosition == null) { fileReader.close(); fileReader = null; LogFileAndPath nextLogFileAndPath = logStream.getNext(currentLogFile); if (nextLogFileAndPath == null) { LOG.info("No more message in the current log file and it is the end of log stream: {}", logStream); return null; } fileReader = getNextLogFileReader(nextLogFileAndPath); if (fileReader == null) { Stats.incr("singer.processor.files.missing"); return null; } LOG.info("Continue to read from next log file {} in log stream {}", fileReader.getLogFile(), logStream.getLogStreamDescriptor()); long currentFileLength = new File(logStream.getLogFilePath(currentLogFile)).length(); long bytesSkipped = currentFileLength - startByteOffset; if (bytesSkipped > 0) { LOG.error("Skipped {} bytes in log file: {} in log stream: {}", bytesSkipped, currentLogFile, logStream); OpenTsdbMetricConverter.incr(SingerMetrics.SKIPPED_BYTES, (int) bytesSkipped, "log=" + logStream.getSingerLog().getSingerLogConfig().getName(), "host=" + SingerUtils.getHostname()); } return readLogMessageAndPosition(); } else { return messageAndPosition; } } catch (Exception e) { LOG.error("Caught exception while reading log message from log stream: {}", logStream, e); try { if (fileReader != null) { fileReader.close(); } fileReader = null; LOG.info("Closed fileReader when failed to seek on DefaultLogStreamReader"); } catch (IOException ex) { throw new LogStreamReaderException("Can not close fileReader.", ex); } throw new LogStreamReaderException( "Can not read a log message from current position in log stream: " + logStream .getLogStreamDescriptor(), e); } } DefaultLogStreamReader(LogStream logStream, LogFileReaderFactory fileReaderFactory); @Override LogStream getLogStream(); @Override LogPosition getPos(); @Override void seek(LogPosition position); @Override LogMessageAndPosition readLogMessageAndPosition(); @Override void close();  }