Repository: apache/streampipes
Pull request #2335 (2023-12-14 12:26:40)

fix: orchestrate adapter health check such that monitoring of adapters works as expected


<!--
  ~ Licensed to the Apache Software Foundation (ASF) under one or more
  ~ contributor license agreements.  See the NOTICE file distributed with
  ~ this work for additional information regarding copyright ownership.
  ~ The ASF licenses this file to You under the Apache License, Version 2.0
  ~ (the "License"); you may not use this file except in compliance with
  ~ the License.  You may obtain a copy of the License at
  ~
  ~    http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->
#2328 introduced a bug where monitoring metrics were not exposed anymore for running adapters due to a side effect of `getAdaptersToRecover()`.
This PR introduces a fix so that metrics are exposed again for all adapters running and makes metric emitting more robust. However, parts of the introduced changes are only to be seen as a workaround.

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no
PR introduces (a) deprecation(s): no


Patch:
diff --git a/pom.xml b/pom.xml
index 77d61e7c8b..3234211234 100644
--- a/pom.xml
+++ b/pom.xml
@@ -141,7 +141,7 @@
         <formatter-maven-plugin.version>2.23.0</formatter-maven-plugin.version>
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
-        <maven-checkstyle-plugin.version>3.2.1</maven-checkstyle-plugin.version>
+        <maven-checkstyle-plugin.version>3.3.1</maven-checkstyle-plugin.version>
         <maven.dependency.check.plugin.version>8.4.0</maven.dependency.check.plugin.version>
         <maven-shade-plugin.version>3.4.1</maven-shade-plugin.version>
         <maven.version>3.8.7</maven.version>
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java
index f0882a1fc6..1d8843a62c 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java
@@ -31,7 +31,6 @@ import org.apache.streampipes.storage.api.IAdapterStorage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -66,28 +65,32 @@ public class AdapterHealthCheck implements Runnable {
    * {@link org.apache.streampipes.manager.health.PipelineHealthCheck}).
    */
   public void checkAndRestoreAdapters() {
-    // Get all running adapters
-    Map<String, AdapterDescription> allRunningInstancesAdapterDescriptions =
-        this.getAllRunningInstancesAdapterDescriptions();
+    // Get all adapters that are supposed to run according to the backend storage
+    Map<String, AdapterDescription> adapterInstancesSupposedToRun =
+        this.getAllAdaptersSupposedToRun();
 
-    // Get all worker containers that run adapters
+    // group all adapter instances supposed to run by their worker service URL
     Map<String, List<AdapterDescription>> groupByWorker =
-        this.getAllWorkersWithAdapters(allRunningInstancesAdapterDescriptions);
+        this.getAllWorkersWithAdapters(adapterInstancesSupposedToRun);
 
     // Get adapters that are not running anymore
     Map<String, AdapterDescription> allAdaptersToRecover =
-        this.getAdaptersToRecover(groupByWorker, allRunningInstancesAdapterDescriptions);
+        this.getAdaptersToRecover(groupByWorker, adapterInstancesSupposedToRun);
 
     try {
-      if (!allRunningInstancesAdapterDescriptions.isEmpty()) {
+      if (!adapterInstancesSupposedToRun.isEmpty()) {
         // Filter adapters so that only healthy and running adapters are updated in the metrics endpoint
-        updateMonitoringMetrics(
-            allRunningInstancesAdapterDescriptions
-                .entrySet()
-                .stream()
-                .filter((entry -> !allAdaptersToRecover.containsKey(entry.getKey())))
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
-        );
+        var adaptersToMonitor = adapterInstancesSupposedToRun
+            .entrySet()
+            .stream()
+            .filter((entry -> !allAdaptersToRecover.containsKey(entry.getKey())))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+        if (!adaptersToMonitor.isEmpty()) {
+          updateMonitoringMetrics(adaptersToMonitor);
+        } else {
+          LOG.info("No running adapter instances to monitor.");
+        }
       }
     } catch (NoSuchElementException e) {
       LOG.error("Could not update adapter metrics due to an invalid state. ({})", e.getMessage());
@@ -118,6 +121,15 @@ public class AdapterHealthCheck implements Runnable {
   }
 
   private void updateTotalEventsPublished(AdapterMetrics adapterMetrics, String adapterId, String adapterName) {
+
+    // Check if the adapter is already registered; if not, register it first.
+    // This step is crucial, especially when the StreamPipes Core service is restarted,
+    // and there are existing running adapters that need proper registration.
+    // Note: Proper registration is usually handled during the initial start of the adapter.
+    if (!adapterMetrics.contains(adapterId)) {
+      adapterMetrics.register(adapterId, adapterName);
+    }
+
     adapterMetrics.updateTotalEventsPublished(
         adapterId,
         adapterName,
@@ -128,7 +140,18 @@ public class AdapterHealthCheck implements Runnable {
   }
 
 
-  public Map<String, AdapterDescription> getAllRunningInstancesAdapterDescriptions() {
+  /**
+   * Retrieves a map of all adapter instances that are supposed to be running according to the backend storage.
+   * <p>
+   * This method queries the adapter storage to obtain information about all adapters
+   * and filters the running instances. The resulting map is keyed by the element ID
+   * of each running adapter, and the corresponding values are the respective
+   * {@link AdapterDescription} objects.
+   *
+   * @return A map containing all adapter instances supposed to be running according to the backend storage.
+   * The keys are element IDs, and the values are the corresponding adapter descriptions.
+   */
+  public Map<String, AdapterDescription> getAllAdaptersSupposedToRun() {
     Map<String, AdapterDescription> result = new HashMap<>();
     List<AdapterDescription> allRunningInstancesAdapterDescription = this.adapterStorage.getAllAdapters();
     allRunningInstancesAdapterDescription
@@ -144,41 +167,69 @@ public class AdapterHealthCheck implements Runnable {
   }
 
   public Map<String, List<AdapterDescription>> getAllWorkersWithAdapters(
-      Map<String, AdapterDescription> allRunningInstancesAdapterDescription) {
+      Map<String, AdapterDescription> adapterInstancesSupposedToRun
+  ) {
 
     Map<String, List<AdapterDescription>> groupByWorker = new HashMap<>();
-    allRunningInstancesAdapterDescription.values().forEach(ad -> {
-      String selectedEndpointUrl = ad.getSelectedEndpointUrl();
-      if (selectedEndpointUrl != null) {
-        if (groupByWorker.containsKey(selectedEndpointUrl)) {
-          groupByWorker.get(selectedEndpointUrl).add(ad);
-        } else {
-          List<AdapterDescription> tmp = new ArrayList<>();
-          tmp.add(ad);
-          groupByWorker.put(selectedEndpointUrl, tmp);
-        }
-      }
-    });
+    adapterInstancesSupposedToRun.values()
+                                 .forEach(ad -> {
+                                   String selectedEndpointUrl = ad.getSelectedEndpointUrl();
+                                   if (selectedEndpointUrl != null) {
+                                     if (groupByWorker.containsKey(selectedEndpointUrl)) {
+                                       groupByWorker.get(selectedEndpointUrl)
+                                                    .add(ad);
+                                     } else {
+                                       groupByWorker.put(selectedEndpointUrl, List.of(ad));
+                                     }
+                                   }
+                                 });
 
     return groupByWorker;
   }
 
+  /**
+   * Retrieves a map of adapters to recover by comparing the provided groupings of adapter instances
+   * with the instances supposed to run according to the storage.
+   * For every adapter instance it is verified that it actually runs on a worker node.
+   * If this is not the case, it is added to the output of adapters to recover.
+   *
+   * @param adapterInstancesGroupedByWorker A map grouping adapter instances by worker.
+   * @param adapterInstancesSupposedToRun   The map containing all adapter instances supposed to be running.
+   * @return A new map containing adapter instances to recover, filtered based on running instances.
+   */
   public Map<String, AdapterDescription> getAdaptersToRecover(
-      Map<String, List<AdapterDescription>> groupByWorker,
-      Map<String, AdapterDescription> allRunningInstancesAdapterDescription) {
-    groupByWorker.keySet().forEach(adapterEndpointUrl -> {
-      try {
-        List<AdapterDescription> allRunningInstancesOfOneWorker =
-            WorkerRestClient.getAllRunningAdapterInstanceDescriptions(
-                adapterEndpointUrl + WorkerPaths.getRunningAdaptersPath());
-        allRunningInstancesOfOneWorker.forEach(adapterDescription ->
-            allRunningInstancesAdapterDescription.remove(adapterDescription.getElementId()));
-      } catch (AdapterException e) {
-        LOG.info("Could not recover adapter at endpoint {} due to {}", adapterEndpointUrl, e.getMessage());
-      }
-    });
+      Map<String, List<AdapterDescription>> adapterInstancesGroupedByWorker,
+      Map<String, AdapterDescription> adapterInstancesSupposedToRun
+  ) {
 
-    return allRunningInstancesAdapterDescription;
+    // NOTE: This line is added to prevent modifying the existing map of instances supposed to run
+    // It looks like the parameter `adapterInstancesSupposedToRun` is not required at all,
+    // but this should be checked more carefully.
+    Map<String, AdapterDescription> adaptersToRecover = new HashMap<>(adapterInstancesSupposedToRun);
+
+    adapterInstancesGroupedByWorker.keySet()
+                                   .forEach(adapterEndpointUrl -> {
+                                     try {
+                                       List<AdapterDescription> allRunningInstancesOfOneWorker =
+                                           WorkerRestClient.getAllRunningAdapterInstanceDescriptions(
+                                               adapterEndpointUrl + WorkerPaths.getRunningAdaptersPath());
+
+                                       // only keep adapters where there is no running adapter instance
+                                       // therefore, all others are removed
+                                       allRunningInstancesOfOneWorker.forEach(
+                                           adapterDescription ->
+                                               adaptersToRecover.remove(
+                                                   adapterDescription.getElementId()));
+                                     } catch (AdapterException e) {
+                                       LOG.info(
+                                           "Could not recover adapter at endpoint {} due to {}",
+                                           adapterEndpointUrl,
+                                           e.getMessage()
+                                       );
+                                     }
+                                   });
+
+    return adaptersToRecover;
   }
 
 
diff --git a/streampipes-maven-plugin/pom.xml b/streampipes-maven-plugin/pom.xml
index 7c0da262c4..775f815b51 100644
--- a/streampipes-maven-plugin/pom.xml
+++ b/streampipes-maven-plugin/pom.xml
@@ -39,7 +39,7 @@
         <maven-clean-plugin.version>3.3.2</maven-clean-plugin.version>
         <maven-deploy-plugin.version>3.1.0</maven-deploy-plugin.version>
         <maven-install-plugin.version>2.5.2</maven-install-plugin.version>
-        <maven-invoker-plugin.version>3.4.0</maven-invoker-plugin.version>
+        <maven-invoker-plugin.version>3.6.0</maven-invoker-plugin.version>
         <maven-jar-plugin.version>3.0.2</maven-jar-plugin.version>
         <maven-plugin-plugin.version>3.10.2</maven-plugin-plugin.version>
         <maven-resources-plugin.version>3.3.0</maven-resources-plugin.version>
diff --git a/ui/cypress/support/utils/UserUtils.ts b/ui/cypress/support/utils/UserUtils.ts
index fb505aa81a..9db82d6410 100644
--- a/ui/cypress/support/utils/UserUtils.ts
+++ b/ui/cypress/support/utils/UserUtils.ts
@@ -66,5 +66,6 @@ export class UserUtils {
         this.goToUserConfiguration();
 
         cy.dataCy('user-delete-btn-' + user.name).click();
+        cy.dataCy('confirm-delete').click();
     }
 }
diff --git a/ui/package-lock.json b/ui/package-lock.json
index 805ae58688..01e92f4b24 100644
--- a/ui/package-lock.json
+++ b/ui/package-lock.json
@@ -86,7 +86,7 @@
         "@types/jqueryui": "^1.12.16",
         "@types/jshint": "^2.12.1",
         "@types/leaflet": "^1.9.0",
-        "@types/node": "^20.9.4",
+        "@types/node": "^20.10.4",
         "@types/rx": "^4.1.2",
         "@types/showdown": "1.9.4",
         "@typescript-eslint/eslint-plugin": "5.57.1",
@@ -7425,9 +7425,9 @@
       "dev": true
     },
     "node_modules/@types/node": {
-      "version": "20.9.4",
-      "resolved": "https://registry.npmjs.org/@types/node/-/node-20.9.4.tgz",
-      "integrity": "sha512-wmyg8HUhcn6ACjsn8oKYjkN/zUzQeNtMy44weTJSM6p4MMzEOuKbA3OjJ267uPCOW7Xex9dyrNTful8XTQYoDA==",
+      "version": "20.10.4",
+      "resolved": "https://registry.npmjs.org/@types/node/-/node-20.10.4.tgz",
+      "integrity": "sha512-D08YG6rr8X90YB56tSIuBaddy/UXAA9RKJoFvrsnogAum/0pmjkgi4+2nx96A330FmioegBWmEYQ+syqCFaveg==",
       "dev": true,
       "dependencies": {
         "undici-types": "~5.26.4"
@@ -33837,9 +33837,9 @@
       "dev": true
     },
     "@types/node": {
-      "version": "20.9.4",
-      "resolved": "https://registry.npmjs.org/@types/node/-/node-20.9.4.tgz",
-      "integrity": "sha512-wmyg8HUhcn6ACjsn8oKYjkN/zUzQeNtMy44weTJSM6p4MMzEOuKbA3OjJ267uPCOW7Xex9dyrNTful8XTQYoDA==",
+      "version": "20.10.4",
+      "resolved": "https://registry.npmjs.org/@types/node/-/node-20.10.4.tgz",
+      "integrity": "sha512-D08YG6rr8X90YB56tSIuBaddy/UXAA9RKJoFvrsnogAum/0pmjkgi4+2nx96A330FmioegBWmEYQ+syqCFaveg==",
       "dev": true,
       "requires": {
         "undici-types": "~5.26.4"
diff --git a/ui/package.json b/ui/package.json
index 184abe9bb4..37c8e2c516 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -108,7 +108,7 @@
     "@types/jqueryui": "^1.12.16",
     "@types/jshint": "^2.12.1",
     "@types/leaflet": "^1.9.0",
-    "@types/node": "^20.9.4",
+    "@types/node": "^20.10.4",
     "@types/rx": "^4.1.2",
     "@types/showdown": "1.9.4",
     "@typescript-eslint/eslint-plugin": "5.57.1",
diff --git a/ui/src/app/configuration/security-configuration/abstract-security-principal-config.ts b/ui/src/app/configuration/security-configuration/abstract-security-principal-config.ts
index 998f9e8e1f..b3f639b301 100644
--- a/ui/src/app/configuration/security-configuration/abstract-security-principal-config.ts
+++ b/ui/src/app/configuration/security-configuration/abstract-security-principal-config.ts
@@ -27,8 +27,13 @@ import {
     UserService,
 } from '@streampipes/platform-services';
 import { Observable } from 'rxjs';
-import { DialogService, PanelType } from '@streampipes/shared-ui';
+import {
+    ConfirmDialogComponent,
+    DialogService,
+    PanelType,
+} from '@streampipes/shared-ui';
 import { EditUserDialogComponent } from './edit-user-dialog/edit-user-dialog.component';
+import { MatDialog } from '@angular/material/dialog';
 
 @Directive()
 export abstract class AbstractSecurityPrincipalConfig<
@@ -47,6 +52,7 @@ export abstract class AbstractSecurityPrincipalConfig<
         protected userService: UserService,
         protected userAdminService: UserAdminService,
         protected dialogService: DialogService,
+        private dialog: MatDialog,
     ) {}
 
     ngOnInit(): void {
@@ -86,8 +92,24 @@ export abstract class AbstractSecurityPrincipalConfig<
     }
 
     deleteUser(account: UserAccount | ServiceAccount) {
-        this.userService.deleteUser(account.principalId).subscribe(() => {
-            this.load();
+        const dialogRef = this.dialog.open(ConfirmDialogComponent, {
+            width: '500px',
+            data: {
+                title: 'Are you sure you want to delete this account?',
+                subtitle: 'This action cannot be reversed!',
+                cancelTitle: 'Cancel',
+                okTitle: 'Delete User',
+                confirmAndCancel: true,
+            },
+        });
+        dialogRef.afterClosed().subscribe(result => {
+            if (result) {
+                this.userService
+                    .deleteUser(account.principalId)
+                    .subscribe(() => {
+                        this.load();
+                    });
+            }
         });
     }
 
diff --git a/ui/src/app/configuration/security-configuration/user-group-configuration/user-group-configuration.component.ts b/ui/src/app/configuration/security-configuration/user-group-configuration/user-group-configuration.component.ts
index d4f23cf61e..ef32d235d1 100644
--- a/ui/src/app/configuration/security-configuration/user-group-configuration/user-group-configuration.component.ts
+++ b/ui/src/app/configuration/security-configuration/user-group-configuration/user-group-configuration.component.ts
@@ -21,8 +21,13 @@ import { Group, UserGroupService } from '@streampipes/platform-services';
 import { MatPaginator } from '@angular/material/paginator';
 import { MatSort } from '@angular/material/sort';
 import { MatTableDataSource } from '@angular/material/table';
-import { DialogService, PanelType } from '@streampipes/shared-ui';
+import {
+    ConfirmDialogComponent,
+    DialogService,
+    PanelType,
+} from '@streampipes/shared-ui';
 import { EditGroupDialogComponent } from '../edit-group-dialog/edit-group-dialog.component';
+import { MatDialog } from '@angular/material/dialog';
 
 @Component({
     selector: 'sp-security-user-group-config',
@@ -41,6 +46,7 @@ export class SecurityUserGroupConfigComponent implements OnInit {
     constructor(
         private userGroupService: UserGroupService,
         private dialogService: DialogService,
+        private dialog: MatDialog,
     ) {}
 
     ngOnInit(): void {
@@ -60,8 +66,22 @@ export class SecurityUserGroupConfigComponent implements OnInit {
     }
 
     deleteGroup(group: Group) {
-        this.userGroupService.deleteGroup(group).subscribe(response => {
-            this.loadAllGroups();
+        const dialogRef = this.dialog.open(ConfirmDialogComponent, {
+            width: '500px',
+            data: {
+                title: 'Are you sure you want to delete this group?',
+                subtitle: 'This action cannot be reversed!',
+                cancelTitle: 'Cancel',
+                okTitle: 'Delete Group',
+                confirmAndCancel: true,
+            },
+        });
+        dialogRef.afterClosed().subscribe(result => {
+            if (result) {
+                this.userGroupService.deleteGroup(group).subscribe(response => {
+                    this.loadAllGroups();
+                });
+            }
         });
     }
 


Test patch:
diff --git a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java
index 8eedbd725f..e4f3c337cf 100644
--- a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java
+++ b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java
@@ -23,7 +23,6 @@ import org.apache.streampipes.connect.management.management.AdapterMasterManagem
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.storage.api.IAdapterStorage;
-import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -51,14 +50,13 @@ public class AdapterHealthCheckTest {
     var healthCheck = new AdapterHealthCheck(
         adapterInstanceStorageMock,
         new AdapterMasterManagement(
-            StorageDispatcher.INSTANCE.getNoSqlStore()
-                                      .getAdapterInstanceStorage(),
+            adapterInstanceStorageMock,
             new SpResourceManager().manageAdapters(),
             new SpResourceManager().manageDataStreams(),
             AdapterMetricsManager.INSTANCE.getAdapterMetrics()
         )
     );
-    var result = healthCheck.getAllRunningInstancesAdapterDescriptions();
+    var result = healthCheck.getAllAdaptersSupposedToRun();
 
     assertTrue(result.isEmpty());
   }
@@ -82,14 +80,13 @@ public class AdapterHealthCheckTest {
     var healthCheck = new AdapterHealthCheck(
         adapterInstanceStorageMock,
         new AdapterMasterManagement(
-            StorageDispatcher.INSTANCE.getNoSqlStore()
-                                      .getAdapterInstanceStorage(),
+            adapterInstanceStorageMock,
             new SpResourceManager().manageAdapters(),
             new SpResourceManager().manageDataStreams(),
             AdapterMetricsManager.INSTANCE.getAdapterMetrics()
         )
     );
-    var result = healthCheck.getAllRunningInstancesAdapterDescriptions();
+    var result = healthCheck.getAllAdaptersSupposedToRun();
 
     assertEquals(1, result.size());
     assertTrue(result.containsKey(nameRunningAdapter));