[ML] Fix Job Audit Messages filter. (#30490) (#30514)

The way job audit messages were fetched didn't retrieve the expected results if there were deleted jobs with messages still present for these jobs.
This fix allows to specify a list of job IDs to filter the audit messages on. For the jobs list UI, the currently existing job IDs will be passed on to ignore messages from deleted jobs.

x-pack/plugins/ml/server/models/job_audit_messages/job_audit_messages.js

@@ -103,34 +103,50 @@ export function jobAuditMessagesProvider(callWithRequest) {
   }
 
   // search highest, most recent audit messages for all jobs for the last 24hrs.
-  async function getAuditMessagesSummary(jobAggregationSize = 10) {
+  async function getAuditMessagesSummary(jobIds) {
     // TODO This is the current default value of the cluster setting `search.max_buckets`.
     // This should possibly consider the real settings in a future update.
     const maxBuckets = 10000;
+    let levelsPerJobAggSize = maxBuckets;
 
     try {
+      const query = {
+        bool: {
+          filter: [
+            {
+              range: {
+                timestamp: {
+                  gte: 'now-1d'
+                }
+              }
+            }
+          ]
+        }
+      };
+
+      // If the jobIds arg is supplied, add a query filter
+      // to only include those jobIds in the aggregations.
+      if (Array.isArray(jobIds) && jobIds.length > 0) {
+        query.bool.filter.push({
+          terms: {
+            job_id: jobIds
+          }
+        });
+        levelsPerJobAggSize = jobIds.length;
+      }
+
       const resp = await callWithRequest('search', {
         index: ML_NOTIFICATION_INDEX_PATTERN,
         ignore_unavailable: true,
         rest_total_hits_as_int: true,
         size: 0,
         body: {
-          query: {
-            bool: {
-              filter: {
-                range: {
-                  timestamp: {
-                    gte: 'now-1d'
-                  }
-                }
-              }
-            }
-          },
+          query,
           aggs: {
             levelsPerJob: {
               terms: {
                 field: 'job_id',
-                size: (Math.min(maxBuckets, jobAggregationSize) || 1), // don't allow a value of 0
+                size: levelsPerJobAggSize
               },
               aggs: {
                 levels: {

x-pack/plugins/ml/server/models/job_service/jobs.js

@@ -90,7 +90,8 @@ export function jobsProvider(callWithRequest) {
 
   async function jobsSummary(jobIds = []) {
     const fullJobsList = await createFullJobsList();
-    const auditMessages = await getAuditMessagesSummary(fullJobsList.length);
+    const fullJobsIds = fullJobsList.map(job => job.job_id);
+    const auditMessages = await getAuditMessagesSummary(fullJobsIds);
     const auditMessagesByJob = auditMessages.reduce((p, c) => {
       p[c.job_id] = c;
       return p;
@@ -127,7 +128,7 @@ export function jobsProvider(callWithRequest) {
         tempJob.fullJob = job;
       }
       const auditMessage = auditMessagesByJob[tempJob.id];
-      if (auditMessage !== undefined) {
+      if (auditMessage !== undefined && job.create_time <= auditMessage.msgTime) {
         tempJob.auditMessage = {
           level: auditMessage.highestLevel,
           text: auditMessage.highestLevelText