#659: Reduce logging in hyperdrive (#663)

* #659: Reduce logging in hyperdrive

Author: jozefbakus

src/main/resources/logback.xml

@@ -43,7 +43,7 @@
         </layout>
     </appender>
 
-    <logger name="za.co.absa.hyperdrive.trigger" level="debug" additivity="false">
+    <logger name="za.co.absa.hyperdrive.trigger" level="info" additivity="false">
         <appender-ref ref="FILE-ROLLING"/>
         <appender-ref ref="Console" />
     </logger>

src/main/scala/za/co/absa/hyperdrive/trigger/api/rest/controllers/NotificationRuleController.scala

@@ -15,7 +15,6 @@
 
 package za.co.absa.hyperdrive.trigger.api.rest.controllers
 
-import org.slf4j.LoggerFactory
 import org.springframework.web.bind.annotation._
 import za.co.absa.hyperdrive.trigger.api.rest.services.NotificationRuleService
 import za.co.absa.hyperdrive.trigger.models._
@@ -28,7 +27,6 @@ import scala.concurrent.ExecutionContext.Implicits.global
 
 @RestController
 class NotificationRuleController @Inject()(notificationRuleService: NotificationRuleService) {
-  private val logger = LoggerFactory.getLogger(this.getClass)
 
   @PutMapping(path = Array("/notificationRule"))
   def createNotificationRule(@RequestBody notificationRule: NotificationRule): CompletableFuture[NotificationRule] = {

src/main/scala/za/co/absa/hyperdrive/trigger/configuration/liquibase/LiquibaseConfiguration.scala

@@ -69,7 +69,7 @@ class LiquibaseConfiguration(properties: LiquibaseProperties, dbConfig: Database
     if (!unrunChangeSets.isEmpty) {
       liquibase.update(contexts)
     } else {
-      configLogger.debug("Database schema is up-to-date")
+      configLogger.info("Database schema is up-to-date")
     }
   }
 

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/JobScheduler.scala

@@ -63,7 +63,7 @@ class JobScheduler @Inject()(sensors: Sensors, executors: Executors, dagInstance
       runningScheduler =
         Future {
           while (isManagerRunningAtomic.get()) {
-            logger.debug("Running manager heart beat.")
+            logger.info("Running manager heart beat.")
             assignWorkflows(firstIteration)
             sendNotifications()
             firstIteration = false
@@ -72,7 +72,7 @@ class JobScheduler @Inject()(sensors: Sensors, executors: Executors, dagInstance
         }
       runningScheduler.onComplete {
         case Success(_) =>
-          logger.debug("Manager stopped.")
+          logger.info("Manager stopped.")
         case Failure(exception) =>
           logger.error(s"Manager stopped with exception.", exception)
       }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/cluster/SchedulerInstanceService.scala

@@ -48,7 +48,7 @@ class SchedulerInstanceServiceImpl @Inject()(schedulerInstanceRepository: Schedu
         Future{}
       }
       deactivatedCount <- schedulerInstanceRepository.deactivateLaggingInstances(instanceId, currentHeartbeat, lagThreshold)
-      _ = if (deactivatedCount != 0) logger.debug(s"Deactivated $deactivatedCount instances at current heartbeat $currentHeartbeat")
+      _ = if (deactivatedCount != 0) logger.info(s"Deactivated $deactivatedCount instances at current heartbeat $currentHeartbeat")
       allInstances <- schedulerInstanceRepository.getAllInstances()
     } yield allInstances
   }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/eventProcessor/EventProcessor.scala

@@ -32,7 +32,7 @@ class EventProcessor(eventRepository: EventRepository,
 
   def eventProcessor(triggeredBy: String)(events: Seq[Event], sensorId: Long)(implicit ec: ExecutionContext): Future[Boolean] = {
     val fut = processEvents(events, sensorId, triggeredBy)
-    logger.debug(s"Processing events. Sensor id: $sensorId. Events: ${events.map(_.id)}")
+    logger.info(s"Processing events. Sensor id: $sensorId. Events: ${events.map(_.id)}")
     fut
   }
 

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/Executors.scala

@@ -91,7 +91,7 @@ class Executors @Inject()(dagInstanceRepository: DagInstanceRepository, jobInsta
           }
         }
         fut.onComplete {
-          case Success(_) => logger.debug(s"Executing job. Job instance id = ${jobInstance}")
+          case Success(_) => logger.debug(s"Executing job. Job instance = ${jobInstance}")
           case Failure(exception) => {
             logger.error(s"Executing job failed. Job instance id = ${jobInstance}.", exception)
           }
@@ -101,6 +101,7 @@ class Executors @Inject()(dagInstanceRepository: DagInstanceRepository, jobInsta
   }
 
   private def updateJob(jobInstance: JobInstance): Future[Unit] = {
+    logger.info(s"Job updated. ID = ${jobInstance.id} STATUS = ${jobInstance.jobStatus} EXECUTOR_ID = ${jobInstance.executorJobId}")
     jobInstanceRepository.updateJob(jobInstance)
   }
 

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/shell/ShellExecutor.scala

@@ -40,7 +40,7 @@ object ShellExecutor {
     updateJob(jobInstance.copy(jobStatus = Running)).map { _ =>
       Try {
         jobParameters.scriptLocation.!(new ProcessLogger {
-          override def out(s: => String): Unit = logger.info(s)
+          override def out(s: => String): Unit = logger.debug(s)
           override def err(s: => String): Unit = logger.error(s)
           override def buffer[T](f: => T): T = f
         })

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/sensors/Sensors.scala

@@ -45,7 +45,7 @@ class Sensors @Inject()(eventProcessor: EventProcessor, sensorRepository: Sensor
   private val sensors: mutable.Map[Long, Sensor[ _<: SensorProperties]] = mutable.Map.empty[Long, Sensor[ _<: SensorProperties]]
 
   def processEvents(assignedWorkflowIds: Seq[Long], firstIteration: Boolean): Future[Unit] = {
-    logger.debug(s"Processing events. Sensors: ${sensors.keys}")
+    logger.info(s"Processing events. Sensors: ${sensors.keys}")
     removeReleasedSensors(assignedWorkflowIds)
     val fut = for {
       _ <- removeInactiveSensors()
@@ -57,7 +57,7 @@ class Sensors @Inject()(eventProcessor: EventProcessor, sensorRepository: Sensor
     }
 
     fut.onComplete {
-      case Success(_) => logger.debug("Processing events successful")
+      case Success(_) => logger.info("Processing events successful")
       case Failure(exception) => {
         logger.debug("Processing events failed.", exception)
       }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/sensors/kafka/KafkaSensor.scala

@@ -63,7 +63,7 @@ class KafkaSensor(
       }
     })
   } catch {
-    case e: Exception => logger.debug(s"$logMsgPrefix. Exception during subscribe.", e)
+    case e: Exception => logger.error(s"$logMsgPrefix. Exception during subscribe.", e)
   }
 
   override def poll(): Future[Unit] = {
@@ -108,7 +108,7 @@ class KafkaSensor(
     val payload = Try(
       Json.parse(record.value())
     ).getOrElse {
-      logger.debug(s"$logMsgPrefix. Invalid message.")
+      logger.error(s"$logMsgPrefix. Invalid message.")
       Json.parse(s"""{"errorMessage": "${record.value()}"}""")
     }
     Event(sourceEventId, sensorDefinition.id, payload)