Fixes #15675: Leak in Cache of Node Compliance and NodeInfo and perfs improvement

webapp/sources/rudder/pom.xml

@@ -59,7 +59,14 @@ along with Rudder.  If not, see <http://www.gnu.org/licenses/>.
     <!-- fill-in rule templates -->
   </modules>
 
-  <dependencies/>
+  <dependencies>
+	  <dependency>
+	  <groupId>com.github.jbellis</groupId>
+	  <artifactId>jamm</artifactId>
+	  <version>0.3.3</version>
+	  <scope>provided</scope>
+  </dependency>
+  </dependencies>
 
   <repositories>
     <repository>

webapp/sources/rudder/rudder-core/src/main/scala/com/normation/rudder/domain/reports/ComplianceLevel.scala

@@ -144,7 +144,7 @@ object ComplianceLevel {
   private def pc_for(i:Int, total:Int) : Double = if(total == 0) 0 else (i * 100 / BigDecimal(total)).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble
 
 
-  def compute(reports: Iterable[ReportType]): ComplianceLevel = {
+  def compute2(reports: Iterable[ReportType]): ComplianceLevel = {
     import ReportType._
     if(reports.isEmpty) { ComplianceLevel(notApplicable = 1)}
     else reports.foldLeft(ComplianceLevel()) { case (compliance, report) =>
@@ -167,10 +167,122 @@ object ComplianceLevel {
     }
   }
 
- def sum(compliances: Iterable[ComplianceLevel]): ComplianceLevel = {
+  def compute(reports: Iterable[ReportType]): ComplianceLevel = {
+    import ReportType._
+    if(reports.isEmpty) {
+      ComplianceLevel(notApplicable = 1)
+    } else {
+      var notApplicable = 0
+      var success = 0
+      var repaired = 0
+      var error = 0
+      var unexpected = 0
+      var missing = 0
+      var noAnswer = 0
+      var pending = 0
+      var reportsDisabled = 0
+      var compliant = 0
+      var auditNotApplicable = 0
+      var nonCompliant = 0
+      var auditError = 0
+      var badPolicyMode = 0
+
+      reports.foreach { report =>
+        report match {
+          case EnforceNotApplicable => notApplicable += 1
+          case EnforceSuccess => success += 1
+          case EnforceRepaired => repaired += 1
+          case EnforceError => error += 1
+          case Unexpected => unexpected += 1
+          case Missing => missing += 1
+          case NoAnswer => noAnswer += 1
+          case Pending => pending += 1
+          case Disabled => reportsDisabled += 1
+          case AuditCompliant => compliant += 1
+          case AuditNotApplicable => auditNotApplicable += 1
+          case AuditNonCompliant => nonCompliant += 1
+          case AuditError => auditError += 1
+          case BadPolicyMode => badPolicyMode += 1
+        }
+      }
+      ComplianceLevel(
+        pending = pending
+      , success = success
+      , repaired = repaired
+      , error =error
+      , unexpected =unexpected
+      , missing = missing
+      , noAnswer=noAnswer
+      , notApplicable=notApplicable
+      , reportsDisabled=reportsDisabled
+      , compliant  = compliant
+      , auditNotApplicable= auditNotApplicable
+      , nonCompliant  =nonCompliant
+      , auditError = auditError
+      , badPolicyMode =badPolicyMode
+      )
+    }
+  }
+
+ def sum2(compliances: Iterable[ComplianceLevel]): ComplianceLevel = {
    if(compliances.isEmpty) ComplianceLevel()
    else compliances.reduce( _ + _)
  }
+
+  def sum(compliances: Iterable[ComplianceLevel]): ComplianceLevel = {
+    if (compliances.isEmpty) {
+      ComplianceLevel()
+    } else {
+      var pending: Int = 0
+      var success: Int = 0
+      var repaired: Int = 0
+      var error: Int = 0
+      var unexpected: Int = 0
+      var missing: Int = 0
+      var noAnswer: Int = 0
+      var notApplicable: Int = 0
+      var reportsDisabled: Int = 0
+      var compliant: Int = 0
+      var auditNotApplicable: Int = 0
+      var nonCompliant: Int = 0
+      var auditError: Int = 0
+      var badPolicyMode: Int = 0
+
+
+      compliances.foreach { compliance =>
+        pending += compliance.pending
+        success += compliance.success
+        repaired += compliance.repaired
+        error += compliance.error
+        unexpected += compliance.unexpected
+        missing += compliance.missing
+        noAnswer += compliance.noAnswer
+        notApplicable += compliance.notApplicable
+        reportsDisabled += compliance.reportsDisabled
+        compliant += compliance.compliant
+        auditNotApplicable += compliance.auditNotApplicable
+        nonCompliant += compliance.nonCompliant
+        auditError += compliance.auditError
+        badPolicyMode += compliance.badPolicyMode
+      }
+      ComplianceLevel(
+        pending = pending
+        , success = success
+        , repaired = repaired
+        , error = error
+        , unexpected = unexpected
+        , missing = missing
+        , noAnswer = noAnswer
+        , notApplicable = notApplicable
+        , reportsDisabled = reportsDisabled
+        , compliant = compliant
+        , auditNotApplicable = auditNotApplicable
+        , nonCompliant = nonCompliant
+        , auditError = auditError
+        , badPolicyMode = badPolicyMode
+      )
+    }
+  }
 }
 
 

webapp/sources/rudder/rudder-core/src/main/scala/com/normation/rudder/domain/reports/StatusReports.scala

@@ -116,7 +116,7 @@ final class NodeStatusReport private (
 
 object NodeStatusReport {
   def apply(nodeId: NodeId, runInfo:  RunAndConfigInfo, statusInfo: RunComplianceInfo, overrides : List[OverridenPolicy], reports: Iterable[RuleNodeStatusReport]) = {
-    new NodeStatusReport(nodeId, runInfo, statusInfo, overrides, AggregatedStatusReport(reports.toSet.filter( _.nodeId == nodeId)))
+    new NodeStatusReport(nodeId, runInfo, statusInfo, overrides, AggregatedStatusReport(reports.filter( _.nodeId == nodeId).toSet))
   }
 
   // To use when you are sure that all reports are indeed for the designated node.
@@ -200,11 +200,11 @@ final case class RuleNodeStatusReport(
   |  ${directives.values.toSeq.sortBy( _.directiveId.value ).map { x => s"${x}" }.mkString("\n  ")}]
   |""".stripMargin('|')
 
-
+/*
   def getValues(predicate: ComponentValueStatusReport => Boolean): Seq[(DirectiveId, String, ComponentValueStatusReport)] = {
       directives.values.flatMap( _.getValues(predicate)).toSeq
   }
-
+*/
   def withFilteredElements(
       directive: DirectiveStatusReport => Boolean
     , component: ComponentStatusReport => Boolean
@@ -226,7 +226,7 @@ object RuleNodeStatusReport {
       val newDirectives = DirectiveStatusReport.merge(reports.flatMap( _.directives.values))
 
       //the merge of two reports expire when the first one expire
-      val expire = new DateTime( reports.map( _.expirationDate.getMillis).min )
+      val expire = new DateTime(reports.minBy(_.expirationDate.getMillis).expirationDate.getMillis)
       (id, RuleNodeStatusReport(id._1, id._2, id._3, id._4, newDirectives, expire))
     }.toMap
   }
@@ -239,7 +239,7 @@ final case class DirectiveStatusReport(
 ) extends StatusReport {
   override lazy val compliance = ComplianceLevel.sum(components.map(_._2.compliance) )
   def getValues(predicate: ComponentValueStatusReport => Boolean): Seq[(DirectiveId, String, ComponentValueStatusReport)] = {
-      components.values.flatMap( _.getValues(predicate) ).toSeq.map { case(s,v) => (directiveId,s,v) }
+      components.values.view.flatMap( _.getValues(predicate) ).map { case(s,v) => (directiveId,s,v) }.force.toSeq
   }
 
   override def toString() = s"""[${directiveId.value} =>
@@ -288,7 +288,7 @@ final case class ComponentStatusReport(
    * Get all values matching the predicate
    */
   def getValues(predicate: ComponentValueStatusReport => Boolean): Seq[(String, ComponentValueStatusReport)] = {
-      componentValues.values.filter(predicate(_)).toSeq.map(x => (componentName, x))
+      componentValues.values.view.filter(predicate(_)).map(x => (componentName, x)).force.toSeq
   }
 
   /*
@@ -307,7 +307,7 @@ object ComponentStatusReport extends Loggable {
     components.groupBy( _.componentName).map { case (cptName, reports) =>
       val newValues = ComponentValueStatusReport.merge(reports.flatMap( _.componentValues.values))
       (cptName, ComponentStatusReport(cptName, newValues))
-    }.toMap
+    }
   }
 }
 
@@ -349,7 +349,7 @@ object ComponentValueStatusReport extends Loggable {
       )
 
     }
-    pairs.toMap
+    pairs
   }
 }
 

webapp/sources/rudder/rudder-core/src/main/scala/com/normation/rudder/reports/execution/ReportsExecutionRepository.scala

@@ -117,7 +117,7 @@ class CachedReportsExecutionRepository(
     cache = Map()
   }
 
-  override def getNodesLastRun(nodeIds: Set[NodeId]): Box[Map[NodeId, Option[AgentRunWithNodeConfig]]] = this.synchronized {
+  override def getNodesLastRun(nodeIds: Set[NodeId]): Box[Map[NodeId, Option[AgentRunWithNodeConfig]]] = scala.concurrent.blocking { this.synchronized {
     val n1 = System.currentTimeMillis
     (for {
       runs <- readBackend.getNodesLastRun(nodeIds.diff(cache.keySet))
@@ -127,7 +127,7 @@ class CachedReportsExecutionRepository(
       cache = cache ++ runs
       cache.filterKeys { x => nodeIds.contains(x) }
     }) ?~! s"Error when trying to update the cache of Agent Runs informations"
-  }
+  } }
 
   override def updateExecutions(executions : Seq[AgentRun]) : Seq[Box[AgentRun]] = this.synchronized {
     logger.trace(s"Update runs for nodes [${executions.map( _.agentRunId.nodeId.value ).mkString(", ")}]")

webapp/sources/rudder/rudder-core/src/main/scala/com/normation/rudder/reports/execution/ReportsExecutionRepositoryImpl.scala

@@ -142,9 +142,7 @@ class RoReportsExecutionRepositoryImpl (
               (run.agentRunId.nodeId, AgentRunWithNodeConfig(run.agentRunId, config, run.isCompleted, run.insertionId))
             }).toMap
             ids.map(id => (id, runsMap.get(id))).toMap
-
           }
-
           query.transact(xa).attempt.unsafeRunSync.box
       }
     }.map(_.flatten.toMap)

webapp/sources/rudder/rudder-core/src/main/scala/com/normation/rudder/services/policies/DataStructures.scala

@@ -356,8 +356,8 @@ final case class Policy(
   // == map .values (keep order) ==> Iterator[List[Variable]]
   // == .toList (keep order)     ==> List[List[Variable]]
   // == flatten (keep order)     ==> List[Variable]
-  val expandedVars    = Policy.mergeVars(policyVars.map( _.expandedVars.values).toList.flatten)
-  val originalVars    = Policy.mergeVars(policyVars.map( _.originalVars.values).toList.flatten)
+  def expandedVars    = Policy.mergeVars(policyVars.map( _.expandedVars.values).toList.flatten)
+  //val originalVars    = Policy.mergeVars(policyVars.map( _.originalVars.values).toList.flatten)
   val trackerVariable = policyVars.head.trackerVariable.spec.cloneSetMultivalued.toVariable(policyVars.map(_.trackerVariable.values).toList.flatten)
 }
 
@@ -382,9 +382,9 @@ final object Policy {
     path.replace(TAG_OF_RUDDER_MULTI_POLICY, p.id.getRudderUniqueId)
   }
 
-  def withParams(p:Policy) : String  = {
-    s"${p.technique.id.name.value}(${p.expandedVars.values.map(_.values.headOption.getOrElse(""))})"
-  }
+  //def withParams(p:Policy) : String  = {
+  //  s"${p.technique.id.name.value}(${p.expandedVars.values.map(_.values.headOption.getOrElse(""))})"
+  //}
 
   /*
    * merge an ordered seq of variables.