Package com.github.isuhorukov.log.watcher

Class PostgreSqlJson

java.lang.Object
com.github.isuhorukov.log.watcher.PostgreSqlJson
All Implemented Interfaces:
Closeable, AutoCloseable, Callable<Integer>
public class PostgreSqlJson extends Object implements Callable<Integer>, Closeable
The PostgreSqlJson class is a command-line tool for reading PostgreSQL DBMS logs in JSON format and sending them to an OpenTelemetry collector. It can optionally use a LogEnricher to add additional data to the logs.

  • Field Details

    • position

      final Map<String,Long> position
      Store current log position for each log file.

    • logEnricher

      LogEnricher logEnricher
      LogEnricher instance to use in log processing.

    • watchDir

      String watchDir
      Path to PostgreSQL log directory in JSON format.

    • saveInterval

      long saveInterval
      Interval of saving (in second) of the current read position in the log files. The value must be within a range from 1 till 1000 second.

    • posgreSqlHost

      String posgreSqlHost
      The host name of the PostgreSQL server.

    • posgreSqlPort

      int posgreSqlPort
      The port number the PostgreSQL server is listening on.

    • posgreSqlDatabase

      String posgreSqlDatabase
      The database name.

    • posgreSqlUserName

      String posgreSqlUserName
      The database user on whose behalf the connection is being made.

    • posgreSqlPassword

      String posgreSqlPassword
      Password for PG connection.

    • maximumQueryCacheSize

      int maximumQueryCacheSize
      Database query cache size.

    • currentLogPositionFile

      String currentLogPositionFile
      Path to file to save current processed position in log files. Required write capability for this program.

  • Constructor Details

    • PostgreSqlJson

      public PostgreSqlJson()

  • Method Details

    • main

      public static void main(String[] args)
      Main method to execute the application. Initializes the PostgreSqlJson class, executes the command line arguments, and exits the system if required.
      Parameters:
      args - Command line arguments passed to the application.

    • call

      public Integer call() throws Exception
      Calls the method responsible for watching PostgreSQL logs.
      Specified by:
      call in interface Callable<Integer>
      Returns:
      system exit code
      Throws:
      Exception - if an error occurs while watching the logs

    • watchPostgreSqlLogs

      public int watchPostgreSqlLogs() throws IOException, InterruptedException
      Monitors the PostgreSQL log directory for changes and processes the logs.

      This method sets up a WatchService to continuously monitor the specified directory for new or modified log files in JSON format. When such a file is detected, it is processed to extract relevant log information.

      The method also makes use of a log enricher (if configured) to enhance the log data with additional information.

      Returns:
      system exit code
      Throws:
      IOException - if an I/O error occurs initializing the watcher or processing the logs.
      InterruptedException - if the watch service is interrupted while waiting for events.

    • initLogEnricher

      void initLogEnricher()
      Initializes the log enricher for PostgreSQL logs.

      This method creates an instance of LogEnricherPostgreSql with the specified PostgreSQL host, port, database, username, password, and query cache size.

      If the PostgreSQL host is not set or is empty, the method does not initialize the log enricher, logs an error and instantiate EnrichmentOff instead of LogEnricherPostgreSql.

    • registerWatchEvent

      protected void registerWatchEvent(Path dirToWatch, WatchService watchService) throws IOException
      Registers the specified dir with the given watch service to listen for file creation and modification events.
      Parameters:
      dirToWatch - the directory path to be watched
      watchService - the watch service to which the directory should be registered
      Throws:
      IOException - if an I/O error occurs

    • getWatchService

      protected WatchService getWatchService() throws IOException
      Returns a new WatchService instance that can be used to watch objects for changes in a file tree.
      Returns:
      a new WatchService instance
      Throws:
      IOException - if an I/O error occurs

    • initialLogImport

      protected void initialLogImport(File sourceDirectory) throws IOException
      Performs the initial import of PostgreSQL JSON log files from the specified directory.

      This method reads all JSON log files in the given directory, sorts them by name, and processes each file by calling the readJsonLog(File) method.

      Parameters:
      sourceDirectory - the directory containing the PostgreSQL JSON log files to be imported.
      Throws:
      IOException - if an I/O error occurs while reading the log files.

    • parseLogLine

      void parseLogLine(String line, String logName)
      Parses a single line of PostgreSQL JSON log and logs it according to its severity level.

      This method reads a log line in JSON format, determines the severity level of the log message, and logs the message using the appropriate logging level. If the corresponding logging level (TRACE, DEBUG, INFO, WARN, ERROR) is not enabled, the method will return immediately without further processing the log line.

      The method uses the error_severity field from the JSON log to determine the logging level and creates a logging event with additional key-value pairs extracted from the JSON log. It also handles various log message formats, adjusting or adding specific log attributes such as duration, plan, parse, and bind if they are present in the message.

      Parameters:
      line - the log line in JSON format to be parsed.
      logName - the name of the log file from which the line was read.

    • positionFileTasks

      protected Thread positionFileTasks() throws IOException
      Manages the tasks related to handling positional information for PostgreSQL log files.

      This method processes and updates the positional information of log files to ensure that logs are correctly read and parsed from the last known position. It handles the initialization, periodic updates, and finalization of positions for different log files. The method ensures that log parsing can resume correctly after restarts or interruptions by accurately maintaining and updating the position information.

      The method typically involves the following tasks:

      • Initializing the position map for new log files.
      • Updating the position as logs are read and processed.
      • Saving the current position to persistent storage.

      This method is crucial for log processing scenarios where it's important to not miss any log entries and to avoid reprocessing already handled logs.

      Returns:
      return shutdown hook thread
      Throws:
      IOException - if an I/O error occurs while managing the log file positions.

    • saveLogFilesPosition

      protected void saveLogFilesPosition()
      Saves the current read positions of PostgreSQL log files.

      This method writes the current positions of log files to a specified file, ensuring accurate resume points after restarts or interruptions. It ensures atomicity and consistency to prevent data loss.

    • readJsonLog

      void readJsonLog(File jsonLog) throws IOException
      Reads a JSON log file and processes each line. This method reads through the specified JSON log file starting from the last saved position, parses each line, and processes it according to the log's severity level. It updates the read position after processing each log entry to ensure accurate resumption.
      Parameters:
      jsonLog - the JSON log file to read
      Throws:
      IOException - if an error occurs while reading the log file

    • close

      public void close() throws IOException
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable
      Throws:
      IOException