Ensure default watches are updated for rolling upgrades. (#57185)

For a rolling/mixed cluster upgrade (add new version to existing cluster
then shutdown old instances), the watches that ship by default
with monitoring may not get properly updated to the new version.
Monitoring watches can only get published if the internal state is
marked as dirty. If a node is not master, will also get marked as
clean (e.g. not dirty).

For a mixed cluster upgrade, it is possible for the new node to be
added, not as master, the internal state gets marked as clean so
that no more attempts can be made to publish the watches. This
happens on all new nodes. Once the old nodes are de-commissioned
one of the new version nodes in the cluster gets promoted to master.
However, that new master node (with out intervention like restarting
the node or removing/adding exporters) will never attempt to re-publish
since the internal state was already marked as clean.

This commit adds a cluster state listener to mark the resource dirty
when a node is promoted to master. This will allow the new resource
to be published without any intervention.

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExporter.java

@@ -21,6 +21,7 @@ import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.sniff.ElasticsearchNodesSniffer;
 import org.elasticsearch.client.sniff.Sniffer;
+import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
@@ -363,7 +364,7 @@ public class HttpExporter extends Exporter {
     private static final ConcurrentHashMap<String, SecureString> SECURE_AUTH_PASSWORDS = new ConcurrentHashMap<>();
     private final ThreadContext threadContext;
     private final DateFormatter dateTimeFormatter;
-
+    private final ClusterStateListener onLocalMasterListener;
     /**
      * Create an {@link HttpExporter}.
      *
@@ -424,6 +425,14 @@ public class HttpExporter extends Exporter {
 
         // mark resources as dirty after any node failure or license change
         listener.setResource(resource);
+
+        //for a mixed cluster upgrade, ensure that if master changes and this is the master, allow the resources to re-publish
+        onLocalMasterListener = clusterChangedEvent -> {
+            if (clusterChangedEvent.nodesDelta().masterNodeChanged() && clusterChangedEvent.localNodeMaster()) {
+                resource.markDirty();
+            }
+        };
+        config.clusterService().addListener(onLocalMasterListener);
     }
 
     /**
@@ -864,9 +873,11 @@ public class HttpExporter extends Exporter {
     @Override
     public void doClose() {
         try {
+            config.clusterService().removeListener(onLocalMasterListener);
             if (sniffer != null) {
                 sniffer.close();
             }
+
         } catch (Exception e) {
             logger.error("an error occurred while closing the internal client sniffer", e);
         } finally {