[ML] Allocate jobs based on JobParams rather than cluster state config

[davidkyle]

OpenJobPersistentTasksExecutor allocates jobs to nodes based on the number of open jobs and their memory usage in the method getAssignment. getAssignment is called by PersistentTasksClusterService (master node) and must not do any heavy or blocking work such as a search, so getAssignment must be called with all the information required to make the allocation decision. Previously this worked because all the jobs and their estimated memory usage was stored in the cluster state and readily available but this is no longer the case with the jobs stored in the index.

I considered 2 solutions

1. Add the estimated memory usage for all open jobs to OpenJobAction.JobParams and use those in OpenJobPersistentTasksExecutor.getAssignment
   The JobParams are created by the action and passed as the persistent task start params, some time later the persistent task will call getAssignment at which point the memory usage could be out of date.

2. Add the estimated memory usage to the persistent task parameters or MlMetadata and update it as the jobs change. This way the most up to date information will be available for the allocation decision.

2 requires frequent cluster state updates as the job's established memory changes avoiding these updates is one of the goals of this project so option 1 was chosen. TransportOpenJobAction now collects open jobs memory usage by node before starting the persistent task and passes that information in the JobParams

Update

The job assignment and balancing problem is too large a change for this PR. I simplified the code to make the decision based on open job count. The job balancing/assignment problem will be solved elsewhere possibly outside of this feature branch.

Persisting JobParams

OpenJobAction.JobParams implements ToXContentObject but does not have a lenient parser yet (#33950). During a rolling upgrade from 6.last to 7 the extra fields in JobParams can be tolerated, in 6.x they cannot (failing to parse the clusterstate takes down the node) so those fields are streamed but not persisted in JobParams.toXContent.

Re-allocation Problem

If the job is re-allocated for some reason (e.g. node failure) the open job information stored in the persistent task parameters can be very out of date. This problem isn't addressed here that is for a follow up PR.

Progress!

With this change it is now possible to open a job and POST some data to it and get results. Closing the job and using the datafeed are other matters.

There is also a small refactoring moving the ml persistent task names from TransportOpenJobAction and TransportStartDatafeedAction to MlTasks.

[elasticmachine]

Pinging @elastic/ml-core

[droberts195]

If the job is re-allocated for some reason (e.g. node failure) the open job information stored in the persistent task parameters can be very out of date. This problem isn't addressed here that is for a follow up PR.

Even though this is being left for another PR, is there a plan for how it will be done? I think this PR should not be merged until there is a plan for that case, because otherwise the way it's done may significantly change the way this change in this PR is done.

There is also another similar case to consider. What happens if a user has a system with no running jobs, then starts 5 jobs in quick succession, such that they are all in the master queue before any of the persistent tasks are allocated. I suspect that in that case the node assigned memory map will be empty for all 5 jobs, and they'll all get assigned to the same node. This scenario is not as unlikely as it might seem, as the UI can start many jobs at the same time if you use the pre-canned Nginx module jobs for example.

A possible option in both cases is that if the allocation detects that another persistent task has been allocated in between creation of the node assigned memory map and the current allocation decision then it fails to allocate the job and instead kicks off a new action that repopulates the node assigned memory map. There may be a better solution though.

[davidkyle]

Even though this is being left for another PR, is there a plan for how it will be done?

At the moment no. It is an unforeseen problem I welcome any suggestions.

The current count of open jobs and allocating jobs is available from the persistent tasks so we could fall back to using the basic mechanism of the count of open jobs to determine assignment but at the moment there is no way of specifying the memory usage map is stale/invalid.

[davidkyle]

I think the current approach is a bust due to the problems outlined above. It is possible to know if a task is re-assigning during a call to OpenJobPersistentTasksExecutor.getAssignment as the task will exist in the persistent tasks meta data based on that I suggest the following:

1. Add a map of job id to established memory usage to the task params
2. Add the jobs established model memory and max model memory (analysis limits) to each task's params
3. Max model memory cannot be changed without closing the job so the value in task param is always current
4. On initial assignment base the allocation decision on memory values stored in each task's parameters. This handles the case where a bunch of jobs open at the same time and the tasks weren't allocated when the job memory map was created. If the memory map contains the job then use that value as the established memory in preference to the one in the task's parameters as it is likely more recent.
5. On re-location of the job assume the established model memory values and the memory map are out of date and use the worst case value of max model memory

[davidkyle]

I pushed another commit removing the logic that makes the allocation decision based on job memory. The job balancing problem needs work which is beyond the scope of this PR

[droberts195]

OK cool, happy to merge this to the feature branch with the plan of fixing #34084 before the feature branch is merged into 6.x.