[ML] Reinstate ML daily maintenance actions (#47103)

A refactoring in 6.6 meant that the ML daily
maintenance actions have not been run at all
since then. This change installs the local
master listener that schedules the ML daily
maintenance, and also defends against some
subtle race conditions that could occur in the
future if a node flipped very quickly between
master and non-master.

Fixes #47003

Author: droberts195

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlDailyMaintenanceService.java

@@ -79,12 +79,12 @@ private static TimeValue delayToNextTime(ClusterName clusterName) {
         return TimeValue.timeValueMillis(next.toInstant().toEpochMilli() - now.toInstant().toEpochMilli());
     }
 
-    public void start() {
+    public synchronized void start() {
         LOGGER.debug("Starting ML daily maintenance service");
         scheduleNext();
     }
 
-    public void stop() {
+    public synchronized void stop() {
         LOGGER.debug("Stopping ML daily maintenance service");
         if (cancellable != null && cancellable.isCancelled() == false) {
             cancellable.cancel();
@@ -100,7 +100,7 @@ public void close() {
         stop();
     }
 
-    private void scheduleNext() {
+    private synchronized void scheduleNext() {
         try {
             cancellable = threadPool.schedule(this::triggerTasks, schedulerProvider.get(), ThreadPool.Names.GENERIC);
         } catch (EsRejectedExecutionException e) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlInitializationService.java

@@ -39,6 +39,7 @@ class MlInitializationService implements LocalNodeMasterListener, ClusterStateLi
         this.clusterService = clusterService;
         this.client = client;
         clusterService.addListener(this);
+        clusterService.addLocalNodeMasterListener(this);
     }
 
     @Override
@@ -80,7 +81,7 @@ public String executorName() {
         return ThreadPool.Names.GENERIC;
     }
 
-    private void installDailyMaintenanceService() {
+    private synchronized void installDailyMaintenanceService() {
         if (mlDailyMaintenanceService == null) {
             mlDailyMaintenanceService = new MlDailyMaintenanceService(clusterService.getClusterName(), threadPool, client);
             mlDailyMaintenanceService.start();
@@ -93,7 +94,7 @@ public void beforeStop() {
         }
     }
 
-    private void uninstallDailyMaintenanceService() {
+    private synchronized void uninstallDailyMaintenanceService() {
         if (mlDailyMaintenanceService != null) {
             mlDailyMaintenanceService.stop();
             mlDailyMaintenanceService = null;
@@ -106,7 +107,7 @@ MlDailyMaintenanceService getDailyMaintenanceService() {
     }
 
     /** For testing */
-    void setDailyMaintenanceService(MlDailyMaintenanceService service) {
+    synchronized void setDailyMaintenanceService(MlDailyMaintenanceService service) {
         mlDailyMaintenanceService = service;
     }
 }