Custom healthcheck with filters

Enable filtering with custom health checks based on node attributes:
```
opensearch.optimizedHealthcheck.filters: {
  attribute_key: "attribute_value",
}
```

Also, fixes issue that expects the response to array when it was a
dictionary.

Issue:
opensearch-project#2214
opensearch-project#2203

Signed-off-by: Kawika Avilla <kavilla414@gmail.com>

config/opensearch_dashboards.yml

@@ -40,7 +40,10 @@
 # This node attribute should assign all nodes of the same cluster an integer value that increments with each new cluster that is spun up
 # e.g. in opensearch.yml file you would set the value to a setting using node.attr.cluster_id: 
 # Should only be enabled if there is a corresponding node attribute created in your OpenSearch config that matches the value here
-#opensearch.optimizedHealthcheckId: "cluster_id"
+#opensearch.optimizedHealthcheck.id: "cluster_id"
+#opensearch.optimizedHealthcheck.filters: {
+#  attribute_key: "attribute_value",
+#}
 
 # If your OpenSearch is protected with basic authentication, these settings provide
 # the username and password that the OpenSearch Dashboards server uses to perform maintenance on the OpenSearch Dashboards

src/core/server/opensearch/opensearch_config.ts

@@ -84,7 +84,10 @@ export const configSchema = schema.object({
   requestTimeout: schema.duration({ defaultValue: '30s' }),
   pingTimeout: schema.duration({ defaultValue: schema.siblingRef('requestTimeout') }),
   logQueries: schema.boolean({ defaultValue: false }),
-  optimizedHealthcheckId: schema.maybe(schema.string()),
+  optimizedHealthcheck: schema.object({
+    id: schema.maybe(schema.string()),
+    filters: schema.maybe(schema.recordOf(schema.string(), schema.string(), { defaultValue: {} })),
+  }),
   ssl: schema.object(
     {
       verificationMode: schema.oneOf(
@@ -148,7 +151,8 @@ const deprecations: ConfigDeprecationProvider = ({ renameFromRoot }) => [
   renameFromRoot('elasticsearch.requestTimeout', 'opensearch.requestTimeout'),
   renameFromRoot('elasticsearch.pingTimeout', 'opensearch.pingTimeout'),
   renameFromRoot('elasticsearch.logQueries', 'opensearch.logQueries'),
-  renameFromRoot('elasticsearch.optimizedHealthcheckId', 'opensearch.optimizedHealthcheckId'),
+  renameFromRoot('elasticsearch.optimizedHealthcheckId', 'opensearch.optimizedHealthcheck.id'),
+  renameFromRoot('opensearch.optimizedHealthcheckId', 'opensearch.optimizedHealthcheck.id'),
   renameFromRoot('elasticsearch.ssl', 'opensearch.ssl'),
   renameFromRoot('elasticsearch.apiVersion', 'opensearch.apiVersion'),
   renameFromRoot('elasticsearch.healthCheck', 'opensearch.healthCheck'),
@@ -216,7 +220,7 @@ export class OpenSearchConfig {
    * Specifies whether Dashboards should only query the local OpenSearch node when
    * all nodes in the cluster have the same node attribute value
    */
-  public readonly optimizedHealthcheckId?: string;
+  public readonly optimizedHealthcheck?: OpenSearchConfigType['optimizedHealthcheck'];
 
   /**
    * Hosts that the client will connect to. If sniffing is enabled, this list will
@@ -297,7 +301,7 @@ export class OpenSearchConfig {
     this.ignoreVersionMismatch = rawConfig.ignoreVersionMismatch;
     this.apiVersion = rawConfig.apiVersion;
     this.logQueries = rawConfig.logQueries;
-    this.optimizedHealthcheckId = rawConfig.optimizedHealthcheckId;
+    this.optimizedHealthcheck = rawConfig.optimizedHealthcheck;
     this.hosts = Array.isArray(rawConfig.hosts) ? rawConfig.hosts : [rawConfig.hosts];
     this.requestHeadersWhitelist = Array.isArray(rawConfig.requestHeadersWhitelist)
       ? rawConfig.requestHeadersWhitelist

src/core/server/opensearch/opensearch_service.ts

@@ -95,7 +95,7 @@ export class OpenSearchService
 
     const opensearchNodesCompatibility$ = pollOpenSearchNodesVersion({
       internalClient: this.client.asInternalUser,
-      optimizedHealthcheckId: config.optimizedHealthcheckId,
+      optimizedHealthcheck: config.optimizedHealthcheck,
       log: this.log,
       ignoreVersionMismatch: config.ignoreVersionMismatch,
       opensearchVersionCheckInterval: config.healthCheckDelay.asMilliseconds(),

src/core/server/opensearch/version_check/ensure_opensearch_version.test.ts

@@ -180,7 +180,7 @@ describe('pollOpenSearchNodesVersion', () => {
 
     pollOpenSearchNodesVersion({
       internalClient,
-      optimizedHealthcheckId: 'cluster_id',
+      optimizedHealthcheck: { id: 'cluster_id' },
       opensearchVersionCheckInterval: 1,
       ignoreVersionMismatch: false,
       opensearchDashboardsVersion: OPENSEARCH_DASHBOARDS_VERSION,
@@ -204,7 +204,7 @@ describe('pollOpenSearchNodesVersion', () => {
 
     pollOpenSearchNodesVersion({
       internalClient,
-      optimizedHealthcheckId: 'cluster_id',
+      optimizedHealthcheck: { id: 'cluster_id' },
       opensearchVersionCheckInterval: 1,
       ignoreVersionMismatch: false,
       opensearchDashboardsVersion: OPENSEARCH_DASHBOARDS_VERSION,
@@ -233,7 +233,7 @@ describe('pollOpenSearchNodesVersion', () => {
 
     pollOpenSearchNodesVersion({
       internalClient,
-      optimizedHealthcheckId: 'cluster_id',
+      optimizedHealthcheck: { id: 'cluster_id' },
       opensearchVersionCheckInterval: 1,
       ignoreVersionMismatch: false,
       opensearchDashboardsVersion: OPENSEARCH_DASHBOARDS_VERSION,
@@ -269,7 +269,7 @@ describe('pollOpenSearchNodesVersion', () => {
 
       const opensearchNodesCompatibility$ = pollOpenSearchNodesVersion({
         internalClient,
-        optimizedHealthcheckId: 'cluster_id',
+        optimizedHealthcheck: { id: 'cluster_id' },
         opensearchVersionCheckInterval: 100,
         ignoreVersionMismatch: false,
         opensearchDashboardsVersion: OPENSEARCH_DASHBOARDS_VERSION,
@@ -309,7 +309,7 @@ describe('pollOpenSearchNodesVersion', () => {
 
       const opensearchNodesCompatibility$ = pollOpenSearchNodesVersion({
         internalClient,
-        optimizedHealthcheckId: 'cluster_id',
+        optimizedHealthcheck: { id: 'cluster_id' },
         opensearchVersionCheckInterval: 10,
         ignoreVersionMismatch: false,
         opensearchDashboardsVersion: OPENSEARCH_DASHBOARDS_VERSION,

src/core/server/opensearch/version_check/ensure_opensearch_version.ts

@@ -58,26 +58,54 @@ import type { OpenSearchClient } from '../client';
  */
 export const getNodeId = async (
   internalClient: OpenSearchClient,
-  healthcheckAttributeName: string
+  healthcheck?: OptimizedHealthcheck
 ): Promise<string | null> => {
   try {
-    const state = await internalClient.cluster.state({
+    let path = `nodes.*.attributes.${healthcheck?.id}`;
+    const filters = healthcheck?.filters;
+    if (filters) {
+      Object.keys(filters).forEach((key) => {
+        path += `,nodes.*.attributes.${key}`;
+      });
+    }
+
+    const state = (await internalClient.cluster.state({
       metric: 'nodes',
-      filter_path: [`nodes.*.attributes.${healthcheckAttributeName}`],
-    });
+      filter_path: [path],
+    })) as ApiResponse;
     /* Aggregate different cluster_ids from the OpenSearch nodes
      * if all the nodes have the same cluster_id, retrieve nodes.info from _local node only
-     * Using _cluster/state/nodes to retrieve the cluster_id of each node from master node which is considered to be a lightweight operation
+     * Using _cluster/state/nodes to retrieve the cluster_id of each node from cluster manager node which is considered to be a lightweight operation
      * else if the nodes have different cluster_ids then fan out the request to all nodes
      * else there are no nodes in the cluster
      */
-    const sharedClusterId =
-      state.body.nodes.length > 0
-        ? get(state.body.nodes[0], `attributes.${healthcheckAttributeName}`, null)
-        : null;
+    const nodes = state.body.nodes;
+    let nodeIds = Object.keys(nodes);
+    if (nodeIds.length === 0) {
+      return null;
+    }
+
+    if (filters) {
+      nodeIds.forEach((id) => {
+        Object.keys(filters).forEach((key) => {
+          const attributeValue = get(nodes[id], `attributes.${key}`, null);
+          if (attributeValue === filters[key]) {
+            delete nodes[id];
+          }
+        });
+      });
+
+      nodeIds = Object.keys(nodes);
+      if (nodeIds.length === 0) {
+        return null;
+      }
+    }
+
+    const sharedClusterId = get(nodes[nodeIds[0]], `attributes.${healthcheck?.id}`, null);
+
     return sharedClusterId === null ||
-      state.body.nodes.find(
-        (node: any) => sharedClusterId !== get(node, `attributes.${healthcheckAttributeName}`, null)
+      nodes.find(
+        (node: any) => sharedClusterId !== get(node, `attributes.${healthcheck?.id}`, null)
       )
       ? null
       : '_local';
@@ -88,7 +116,7 @@ export const getNodeId = async (
 
 export interface PollOpenSearchNodesVersionOptions {
   internalClient: OpenSearchClient;
-  optimizedHealthcheckId?: string;
+  optimizedHealthcheck?: OptimizedHealthcheck;
   log: Logger;
   opensearchDashboardsVersion: string;
   ignoreVersionMismatch: boolean;
@@ -118,6 +146,13 @@ export interface NodesVersionCompatibility {
   opensearchDashboardsVersion: string;
 }
 
+export interface OptimizedHealthcheck {
+  id?: string;
+  filters?: {
+    [key: string]: string;
+  };
+}
+
 function getHumanizedNodeName(node: NodeInfo) {
   const publishAddress = node?.http?.publish_address + ' ' || '';
   return 'v' + node.version + ' @ ' + publishAddress + '(' + node.ip + ')';
@@ -201,7 +236,7 @@ function compareNodes(prev: NodesVersionCompatibility, curr: NodesVersionCompati
 
 export const pollOpenSearchNodesVersion = ({
   internalClient,
-  optimizedHealthcheckId,
+  optimizedHealthcheck,
   log,
   opensearchDashboardsVersion,
   ignoreVersionMismatch,
@@ -216,8 +251,8 @@ export const pollOpenSearchNodesVersion = ({
        * For better dashboards resilience, the behaviour is changed to only query the local node when all the nodes have the same cluster_id
        * Using _cluster/state/nodes to retrieve the cluster_id of each node from the master node
        */
-      if (optimizedHealthcheckId) {
-        return from(getNodeId(internalClient, optimizedHealthcheckId)).pipe(
+      if (optimizedHealthcheck) {
+        return from(getNodeId(internalClient, optimizedHealthcheck)).pipe(
           mergeMap((nodeId: any) =>
             from(
               internalClient.nodes.info<NodesInfo>({