Log configuration with logback with file-system watcher over the configuration file

LogbackLoggerConfigurator.scala

package org.myproject.scalaonly

import java.nio.charset.Charset
import java.nio.file.StandardWatchEventKinds.{ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY}
import java.nio.file.WatchEvent.Kind
import java.nio.file._
import java.util.concurrent.locks.ReentrantReadWriteLock
import java.util.logging.{Level, LogManager}

import ch.qos.logback.classic.joran.JoranConfigurator
import ch.qos.logback.classic.{AsyncAppender, LoggerContext}
import ch.qos.logback.core.AsyncAppenderBase
import ch.qos.logback.core.util.StatusPrinter
import com.typesafe.scalalogging.StrictLogging
import org.myproject.scalaonly.FileSystemWatcher.FileSystemWatcherRunnable
import org.slf4j.bridge.SLF4JBridgeHandler
import org.slf4j.impl.StaticLoggerBinder

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.util.control.NonFatal

// dependencies:
// group: 'com.typesafe.scala-logging', name: "scala-logging_2.12", version: '3.9.2'
// group: 'org.slf4j', name: 'jul-to-slf4j', version: '1.7.7'
// group: 'ch.qos.logback', name: 'logback-classic', version: '1.1.7'

object LogbackLoggerConfigurator extends ReadersWriterLockHelper {

  /* --- Constants --- */

  private val ASYNC_APPENDER_PREFIX = "async-"

  private val LOG_FILE_PROPERTY = "logger.file"

  private val FILE_EVENTS = Set[Kind[_]](StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_MODIFY)

  private val FILE_SYSTEM: FileSystem = FileSystems.getDefault

  private val CONTEXT: LoggerContext = StaticLoggerBinder.getSingleton.getLoggerFactory.asInstanceOf[LoggerContext]

  /* --- Data Members --- */

  @volatile
  private var watcher: FileSystemWatcher = _

  @volatile
  private var version: Int = _

  /* --- Methods --- */

  /* --- Public Methods --- */

  /**
   * Configures logback for the first time and starts a file-watcher on the directory where the configuration resides.
   *
   * The configuration file is set via the system property [[LOG_FILE_PROPERTY]].
   *
   * @param logsDirectoryPath The directory for the log files (not the directory of the configuration file).
   * @throws IllegalStateException if logback was already configured and wasn't shutdown yet.
   */
  @throws[IllegalStateException]
  def configure(logsDirectoryPath: String): Unit = {

    write {
      if (watcher != null) {
        throw new IllegalStateException("Log already configured")
      }

      val configurationFile = resolveConfigurationFile()

      SLF4JBridgeHandler.removeHandlersForRootLogger();
      SLF4JBridgeHandler.install()

      version = 0
      configureLog(configurationFile, logsDirectoryPath)
      watchLogConfiguration(configurationFile, logsDirectoryPath)

      addShutdownHook()
    }
  }

  def loggingVersion: Int = {
    read {
      version
    }
  }

  /* --- Private Methods --- */

  private def addShutdownHook(): Unit = {

    Runtime.getRuntime.addShutdownHook(new Thread(new Runnable {
      override def run(): Unit = {
        watcher.stop()
        org.slf4j.bridge.SLF4JBridgeHandler.uninstall()
        CONTEXT.stop()
      }
    }))
  }

  private def resolveConfigurationFile(): Path = {

    val path = sys.props.get(LOG_FILE_PROPERTY).map(FILE_SYSTEM.getPath(_))

    path.getOrElse(
      throw new Exception(s"Could not detect a logback configuration file (use -D${LOG_FILE_PROPERTY}=/path/to/file)"))
  }

  private def configureLog(configurationFile: Path, logsDirectoryPath: String): Unit = {

    write {
      val root = LogManager.getLogManager.getLogger("")
      root.setLevel(Level.FINEST)
      root.info(s"Configuring logback with file: $configurationFile")

      val configurator = new JoranConfigurator
      configurator.setContext(CONTEXT)
      CONTEXT.reset()

      CONTEXT.putProperty("path.logs", logsDirectoryPath)

      configurator.doConfigure(configurationFile.toFile)

      for (logger <- CONTEXT.getLoggerList.asScala) {
        for (blockingAppended <- logger.iteratorForAppenders().asScala
             if !blockingAppended.isInstanceOf[AsyncAppenderBase[_]]) {
          val asyncAppender = new AsyncAppender
          asyncAppender.setContext(CONTEXT)
          asyncAppender.setName(s"$ASYNC_APPENDER_PREFIX${blockingAppended.getName}")
          asyncAppender.setDiscardingThreshold(0)
          asyncAppender.addAppender(blockingAppended)
          asyncAppender.setIncludeCallerData(true)
          asyncAppender.start()

          logger.detachAppender(blockingAppended)
          logger.addAppender(asyncAppender)
        }
      }

      version += 1

      root.info(s"Logback configured with file: $configurationFile")
      root.info(s"Logs directory: $logsDirectoryPath")
      root.info(s"Logging version: $version")

      StatusPrinter.print(CONTEXT)
    }
  }

  private def watchLogConfiguration(configurationFile: Path, logsDirectoryPath: String): Unit = {

    def readConfigurationContent(): String = {
      Files.readAllLines(configurationFile, Charset.forName("UTF-8")).asScala.mkString("\n")
    }

    var lastConent = readConfigurationContent()

    watcher = new FileSystemWatcher(
      new FileSystemListener {

        @throws[Throwable]
        override def onEvent(event: WatchEvent[_]): Unit = {

          if (FILE_EVENTS.contains(event.kind())) {

            event.context() match {
              case path: Path if path == configurationFile.getFileName =>
                val currentContent = readConfigurationContent()
                if (currentContent != lastConent) {
                  configureLog(configurationFile, logsDirectoryPath)
                  lastConent = currentContent
                }
              case _ => // ignore
            }
          }
        }

        override def onExit(): Unit = {}

      },
      FILE_SYSTEM
    )

    watcher.watch(configurationFile.getParent.toString)
  }
}

/**
 * Watches directories in the given file system in a new thread and invokes the given listener for each change in those
 * directories.
 *
 * @param listener   The given listener.
 * @param fileSystem The given file system.
 */
class FileSystemWatcher(listener: FileSystemListener, fileSystem: FileSystem) {

  /* --- Data Members --- */

  /** A runnable that watches the directory and notifies the listener for every event. */
  private var runnable: Option[FileSystemWatcherRunnable] = None

  /** The thread that runs the runnable. */
  private var thread: Option[Thread] = None

  /** The path of the currently watched directory. */
  private val currentlyWatchedDirectories = mutable.Set.empty[String]

  /* --- Constructors --- */

  def this(listener: FileSystemListener) = {

    this(listener, FileSystems.getDefault)
  }

  /* --- Methods --- */

  /* --- Public Methods --- */

  /**
   * Starts watching the directories of the given paths.
   *
   * @param directories The given paths.
   */
  def watch(directories: String*): Unit = {

    if (isWatching) {
      throw new IllegalStateException(
        s"Can't watch new directories ($directories) since there are other currently " +
          s"being watched: $currentlyWatchedDirectories")
    }

    val watchService = registerToDirectory(directories: _*)

    runnable = Some(new FileSystemWatcherRunnable(watchService, listener, directories.toSet))
    thread = runnable.flatMap { case r => Some(new Thread(r, "filesystem-watcher")) }

    thread.get.start()

    currentlyWatchedDirectories ++= directories
  }

  /**
   * Stops watching the current directory.
   */
  def stop(): Unit = {

    runnable.foreach(_.stop())
    runnable = None

    thread.foreach(_.interrupt())
    thread = None

    currentlyWatchedDirectories.clear()
  }

  /**
   * @return Whether it is currently watching a directory or not.
   */
  def isWatching: Boolean = thread.nonEmpty

  /* --- Private Methods --- */

  /**
   * Creates a watch-service that'll register to the given directories, and returns it.
   *
   * @param directories The given directories.
   * @return The new watch-service.
   */
  private def registerToDirectory(directories: String*): WatchService = {

    def toPath(directory: String): Path = {

      val path = fileSystem.getPath(directory)

      if (path == null || !Files.exists(path)) {
        throw new RuntimeException(s"Can't find path: $directory")
      }
      if (!Files.isDirectory(path)) {
        throw new RuntimeException(s"Path is not a directory: $directory")
      }

      path
    }

    val paths = directories.map(toPath)
    val watchService = fileSystem.newWatchService()
    paths.foreach(_.register(watchService, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY))

    watchService
  }
}

object FileSystemWatcher {

  /**
   * A runnable which listens to the given directories using the given watch-service, and notifies the given listener
   * for every change in those directories.
   *
   * @param watchService The given watch-service.
   * @param listener     The given listener.
   * @param directories  The given directories.
   */
  private class FileSystemWatcherRunnable(watchService: WatchService,
                                          listener: FileSystemListener,
                                          directories: Set[String])
    extends Runnable
      with StrictLogging {

    /* --- Data Members --- */

    /** Whether the runnable should keep running or not. */
    @volatile
    private var shouldRun: Boolean = true

    /* --- Methods --- */

    /* --- Public Methods --- */

    /**
     * Awaits for the next event until it is either (signaled to be) stopped or until its thread is interrupted;
     * for each event it notifies the listener.
     */
    override def run(): Unit = {

      logger.info(s"Starting the file system watcher on directories: $directories")

      val executionThread = Thread.currentThread()

      try {
        while (shouldRun && !executionThread.isInterrupted) {
          waitAndNotifyListener()
        }
      } catch {
        case NonFatal(e) =>
          logger.error(s"An exception occurred due to an operation performed while watching directories: $directories",
            e)
        case _: InterruptedException => executionThread.interrupt()
      }

      logger.info(s"Stopping the file system watcher on directories: $directories")

      watchService.close()
      listener.onExit()
    }

    /**
     * Signals the runnable to stop.
     */
    def stop(): Unit = {

      shouldRun = false
    }

    /* --- Private Methods --- */

    /**
     * Awaits for the next file system event and notifies the listener when that happens.
     *
     * @throws InterruptedException if the current thread was interrupted while watching the directories.
     */
    @throws[InterruptedException]
    private def waitAndNotifyListener(): Unit = {

      watchService.take() match {
        case key: WatchKey =>
          for (event <- key.pollEvents().asScala) {
            key.reset()
            listener.onEvent(event)
          }
        case null =>
          shouldRun = false
          logger.error(s"Something went wrong with the watch-service on directories: $directories")
      }
    }
  }

}

/**
 * Performs an operation for each change in the directories that are being watched by a [[FileSystemWatcher]].
 */
trait FileSystemListener {

  /* --- Methods --- */

  /* --- Public Methods --- */

  /**
   * Performs an operation for each change event in the watched directory.
   *
   * @param event The change event.
   * @throws Throwable if thrown then the watcher will be stopped.
   */
  @throws[Throwable]
  def onEvent(event: WatchEvent[_]): Unit

  /**
   * Performs a cleanup operation when the watcher stops.
   */
  def onExit(): Unit
}


/**
 * Provides helper methods to use a readers-writer lock.
 */
trait ReadersWriterLockHelper {

  /* --- Data Members --- */

  private val lock = new ReentrantReadWriteLock()

  /* --- Methods --- */

  /* --- Protected Methods --- */

  protected def read[A](f: => A): A = {

    try {
      lock.readLock().lock()
      f
    } finally {
      lock.readLock().unlock()
    }
  }

  protected def write[A](f: => A): A = {

    try {
      lock.writeLock().lock()
      f
    } finally {
      lock.writeLock().unlock()
    }
  }
}