Repository: apache/streampipes
Pull request #2283 (2023-12-06 13:28:30)

extend monitoring by adding metrics for adapters
<!--
  ~ 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.
-->
While we already expose monitoring metrics for pipelines during the pipeline health check, adapter metrics are missing out so far.
This PR introduces the capability to define metrics for adapters (see `AdapterMetrics`) and expose them via Prometheus during the adapter health check.
For now, this supports just one metric (`adapter_events_published_total`) which tracks the total amount of events published for every adapter.
This can be used, for example, to monitor if all adapters are running.
Here is a screenshot of the metric for two adapters in Prometheus:

![image](https://github.com/apache/streampipes/assets/50115603/229220b6-c61d-4d0f-af35-f729d888fe47)




### 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


bossenti: @tenthe please review this commit https://github.com/apache/streampipes/pull/2283/commits/61bf68f8e8b52c4d431eda99190f99d0e4958686 carefully

Patch:
diff --git a/installer/cli/environments/lite-monitoring b/installer/cli/environments/lite-monitoring
index d99f12c3e4..ffc232a38e 100644
--- a/installer/cli/environments/lite-monitoring
+++ b/installer/cli/environments/lite-monitoring
@@ -16,7 +16,6 @@
 [environment:lite-monitoring]
 ui
 backend
-consul
 couchdb
 kafka
 zookeeper
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/HttpConstants.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/HttpConstants.java
index 63dece4bb0..9866f10e82 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/HttpConstants.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/HttpConstants.java
@@ -26,5 +26,6 @@ public class HttpConstants {
   public static final String APPLICATION_JSON_TYPE = "application/json";
   public static final String X_API_USER = "X-API-USER";
   public static final String X_API_KEY = "X-API-KEY";
+  public static final String BASIC = "Basic ";
 
 }
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/StreamPipesCollectorRegistry.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/StreamPipesCollectorRegistry.java
index 1795ea03d7..242344b1f2 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/StreamPipesCollectorRegistry.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/StreamPipesCollectorRegistry.java
@@ -43,6 +43,14 @@ public class StreamPipesCollectorRegistry {
       .register(collectorRegistry);
   }
 
+  public static Gauge registerGauge(String name, String help, String... labelNames) {
+    return Gauge.build()
+                .name(name)
+                .help(help)
+                .labelNames(labelNames)
+                .register(collectorRegistry);
+  }
+
   public static Counter registerCounter(String name, String help) {
     return Counter.build()
       .name(name)
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetrics.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetrics.java
new file mode 100644
index 0000000000..174531fb6e
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetrics.java
@@ -0,0 +1,126 @@
+/*
+ * 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.
+ *
+ */
+package org.apache.streampipes.commons.prometheus.adapter;
+
+import org.apache.streampipes.commons.prometheus.StreamPipesCollectorRegistry;
+
+import io.prometheus.client.Gauge;
+
+import java.util.HashMap;
+import java.util.NoSuchElementException;
+
+/**
+ * AdapterMetrics class manages monitoring metrics related to StreamPipes adapters.
+ * It allows register and remove adapter instances as well as updating the values of metrics.
+ * Currently, the following metrics are supported:
+ * <ul>
+ *   <li><strong>adapter_events_published_total:</strong> The total number of events published per adapter</li>
+ * </ul>
+ * <p>Usage Example:
+ * <pre>{@code
+ * AdapterMetrics adapterMetrics = new AdapterMetrics();
+ * adapterMetrics.register("adapter-1", "Sample Adapter");
+ * adapterMetrics.updateTotalEventsPublished("adapter-1", "Sample Adapter", 100);
+ * adapterMetrics.remove("adapter-1", "Sample Adapter");
+ * }</pre>
+ */
+public class AdapterMetrics {
+
+  // Map to keep book about the registered adapters and to serve as mapping between adapter id and adapter name
+  // Both are used as dimensions for the metric
+  private final HashMap<String, String> registeredAdapters;
+
+  private static final String ELEMENT_NOT_FOUND_TEMPLATE = "No entry for adapter '%s' found. Please register it first.";
+
+  private final Gauge totalAdapterEventsPublishedMetric;
+
+  public AdapterMetrics() {
+    this.totalAdapterEventsPublishedMetric = StreamPipesCollectorRegistry.registerGauge(
+        "adapter_events_published_total",
+        "Total amount of events published per adapter",
+        "adapterId", "adapterName"
+    );
+    this.registeredAdapters = new HashMap<>();
+  }
+
+  /**
+   * Registers a new adapter with the given adapterId and adapterName.
+   *
+   * @param adapterId   The unique identifier for the adapter
+   * @param adapterName The name of the adapter
+   */
+  public void register(String adapterId, String adapterName) {
+    if (!contains(adapterId)) {
+      this.registeredAdapters.put(adapterId, adapterName);
+    }
+  }
+
+  /**
+   * Checks if an adapter with the specified adapterId is registered.
+   *
+   * @param adapterId The unique identifier for the adapter
+   * @return True if the adapter is registered, false otherwise
+   */
+  public boolean contains(String adapterId) {
+    return this.registeredAdapters.containsKey(adapterId);
+  }
+
+  /**
+   * Updates the total number of events published for a specific adapter.
+   *
+   * @param adapterId            The unique identifier for the adapter
+   * @param adapterName          The name of the adapter
+   * @param totalEventsPublished The total number of events published by the adapter
+   * @throws NoSuchElementException If the adapter is not registered yet
+   */
+  public void updateTotalEventsPublished(String adapterId, String adapterName, long totalEventsPublished) {
+    if (contains(adapterId)) {
+      // Order of labels needs to be preserved and match the one in AdapterMetrics#register
+      totalAdapterEventsPublishedMetric.labels(adapterId, adapterName)
+                                       .set(totalEventsPublished);
+    } else {
+      throw new NoSuchElementException(ELEMENT_NOT_FOUND_TEMPLATE.formatted(adapterId));
+    }
+  }
+
+  /**
+   * Removes the specified adapter from the registry.
+   * This results in no more metrics being published for this adapter.
+   *
+   * @param adapterId   The unique identifier for the adapter
+   * @param adapterName The name of the adapter
+   * @throws NoSuchElementException If the adapter is not found in the registry
+   */
+  public void remove(String adapterId, String adapterName) {
+    if (contains(adapterId)) {
+      this.totalAdapterEventsPublishedMetric.remove(adapterId, adapterName);
+      this.registeredAdapters.remove(adapterId);
+    } else {
+      throw new NoSuchElementException(ELEMENT_NOT_FOUND_TEMPLATE.formatted(adapterId));
+    }
+  }
+
+  /**
+   * Returns the number of registered adapters in the registry.
+   *
+   * @return The number of registered adapters
+   */
+  public int size() {
+    return registeredAdapters.size();
+  }
+}
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetricsManager.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetricsManager.java
new file mode 100644
index 0000000000..8e211ac7ee
--- /dev/null
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetricsManager.java
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.streampipes.commons.prometheus.adapter;
+
+/**
+ * The `AdapterMetricsManager` enum provides a single access point for retrieving the instance of
+ * the `AdapterMetrics` class and ensures that only one instance of the metrics manager exists throughout the
+ * application's lifecycle.
+ */
+public enum AdapterMetricsManager {
+
+  INSTANCE;
+
+  private final AdapterMetrics adapterMetrics;
+
+  AdapterMetricsManager() {
+    this.adapterMetrics = new AdapterMetrics();
+  }
+
+  public static AdapterMetricsManager getInstance() {
+    return INSTANCE;
+  }
+
+  public AdapterMetrics getAdapterMetrics() {
+    return adapterMetrics;
+  }
+}
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 fabf477af5..8653ea0803 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
@@ -19,9 +19,12 @@
 package org.apache.streampipes.connect.management.health;
 
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetrics;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.AdapterMasterManagement;
 import org.apache.streampipes.connect.management.management.WorkerRestClient;
 import org.apache.streampipes.connect.management.util.WorkerPaths;
+import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsLogProvider;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.storage.api.IAdapterStorage;
 
@@ -57,12 +60,18 @@ public class AdapterHealthCheck implements Runnable {
    * In this method it is checked which adapters are currently running.
    * Then it calls all workers to validate if the adapter instance is
    * still running as expected. If the adapter is not running anymore a new worker instance is invoked.
+   * In addition, it publishes monitoring metrics for all running adapters (in line with
+   * {@link org.apache.streampipes.manager.health.PipelineHealthCheck}).
    */
   public void checkAndRestoreAdapters() {
-    // Get all adapters
+    // Get all running adapters
     Map<String, AdapterDescription> allRunningInstancesAdapterDescriptions =
         this.getAllRunningInstancesAdapterDescriptions();
 
+    if (!allRunningInstancesAdapterDescriptions.isEmpty()) {
+      updateMonitoringMetrics(allRunningInstancesAdapterDescriptions);
+    }
+
     // Get all worker containers that run adapters
     Map<String, List<AdapterDescription>> groupByWorker =
         this.getAllWorkersWithAdapters(allRunningInstancesAdapterDescriptions);
@@ -75,6 +84,37 @@ public class AdapterHealthCheck implements Runnable {
     this.recoverAdapters(allAdaptersToRecover);
   }
 
+  /**
+   * Updates the monitoring metrics based on the descriptions of running adapters.
+   *
+   * @param runningAdapterDescriptions A map containing the descriptions of running adapters, where the key is the
+   *                                   adapter's element ID and the value is the corresponding adapter description.
+   */
+  protected void updateMonitoringMetrics(Map<String, AdapterDescription> runningAdapterDescriptions) {
+
+    var adapterMetrics = AdapterMetricsManager.getInstance()
+                                              .getAdapterMetrics();
+    runningAdapterDescriptions.values()
+                              .forEach(
+                                  adapterDescription -> updateTotalEventsPublished(
+                                      adapterMetrics,
+                                      adapterDescription.getElementId(),
+                                      adapterDescription.getName()
+                                  ));
+    LOG.info("Monitoring {} adapter instances", adapterMetrics.size());
+  }
+
+  private void updateTotalEventsPublished(AdapterMetrics adapterMetrics, String adapterId, String adapterName) {
+    adapterMetrics.updateTotalEventsPublished(
+        adapterId,
+        adapterName,
+        ExtensionsLogProvider.INSTANCE.getMetricInfosForResource(adapterId)
+                                      .getMessagesOut()
+                                      .getCounter()
+    );
+  }
+
+
   public Map<String, AdapterDescription> getAllRunningInstancesAdapterDescriptions() {
     Map<String, AdapterDescription> result = new HashMap<>();
     List<AdapterDescription> allRunningInstancesAdapterDescription = this.adapterStorage.getAllAdapters();
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
index c8d16e4d33..8df3721dec 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.connect.management.management;
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
 import org.apache.streampipes.commons.exceptions.SepaParseException;
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetrics;
 import org.apache.streampipes.connect.management.util.GroundingUtils;
 import org.apache.streampipes.connect.management.util.WorkerPaths;
 import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsLogProvider;
@@ -30,9 +31,8 @@ import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.util.ElementIdGenerator;
 import org.apache.streampipes.resource.management.AdapterResourceManager;
 import org.apache.streampipes.resource.management.DataStreamResourceManager;
-import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.storage.api.IAdapterStorage;
-import org.apache.streampipes.storage.couchdb.impl.AdapterInstanceStorageImpl;
+import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,20 +48,18 @@ public class AdapterMasterManagement {
   private static final Logger LOG = LoggerFactory.getLogger(AdapterMasterManagement.class);
 
   private final IAdapterStorage adapterInstanceStorage;
+  private final AdapterMetrics adapterMetrics;
   private final AdapterResourceManager adapterResourceManager;
 
   private final DataStreamResourceManager dataStreamResourceManager;
 
-  public AdapterMasterManagement() {
-    this.adapterInstanceStorage = getAdapterInstanceStorage();
-    this.adapterResourceManager = new SpResourceManager().manageAdapters();
-    this.dataStreamResourceManager = new SpResourceManager().manageDataStreams();
-  }
-
   public AdapterMasterManagement(IAdapterStorage adapterStorage,
                                  AdapterResourceManager adapterResourceManager,
-                                 DataStreamResourceManager dataStreamResourceManager) {
+                                 DataStreamResourceManager dataStreamResourceManager,
+                                 AdapterMetrics adapterMetrics
+  ) {
     this.adapterInstanceStorage = adapterStorage;
+    this.adapterMetrics = adapterMetrics;
     this.adapterResourceManager = adapterResourceManager;
     this.dataStreamResourceManager = dataStreamResourceManager;
   }
@@ -149,6 +147,10 @@ public class AdapterMasterManagement {
 
     WorkerRestClient.stopStreamAdapter(ad.getSelectedEndpointUrl(), ad);
     ExtensionsLogProvider.INSTANCE.reset(elementId);
+
+    // remove the adapter from the metrics manager so that
+    // no metrics for this adapter are exposed anymore
+    adapterMetrics.remove(ad.getElementId(), ad.getName());
   }
 
   public void startStreamAdapter(String elementId) throws AdapterException {
@@ -166,6 +168,9 @@ public class AdapterMasterManagement {
       // Invoke adapter instance
       WorkerRestClient.invokeStreamAdapter(baseUrl, elementId);
 
+      // register the adapter at the metrics manager so that the AdapterHealthCheck can send metrics
+      adapterMetrics.register(ad.getElementId(), ad.getName());
+
       LOG.info("Started adapter " + elementId + " on: " + baseUrl);
     } catch (NoServiceEndpointsAvailableException | URISyntaxException e) {
       throw new AdapterException("Could not start adapter due to unavailable service endpoint", e);
@@ -184,6 +189,6 @@ public class AdapterMasterManagement {
   }
 
   private IAdapterStorage getAdapterInstanceStorage() {
-    return new AdapterInstanceStorageImpl();
+    return StorageDispatcher.INSTANCE.getNoSqlStore().getAdapterInstanceStorage();
   }
 }
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java
index 5378c9acc0..e058c40cbf 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java
@@ -19,16 +19,18 @@
 package org.apache.streampipes.connect.management.management;
 
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetrics;
 import org.apache.streampipes.connect.management.health.AdapterHealthCheck;
 import org.apache.streampipes.connect.management.health.AdapterOperationLock;
 import org.apache.streampipes.manager.assets.AssetManager;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+import org.apache.streampipes.resource.management.AdapterResourceManager;
+import org.apache.streampipes.resource.management.DataStreamResourceManager;
 import org.apache.streampipes.resource.management.PermissionResourceManager;
 import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.storage.api.IAdapterStorage;
 import org.apache.streampipes.storage.couchdb.CouchDbStorageManager;
-import org.apache.streampipes.storage.management.StorageDispatcher;
 import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
 
 import org.slf4j.Logger;
@@ -47,11 +49,20 @@ public class WorkerAdministrationManagement {
 
   private final AdapterHealthCheck adapterHealthCheck;
 
-  public WorkerAdministrationManagement() {
+  public WorkerAdministrationManagement(
+      IAdapterStorage adapterStorage,
+      AdapterMetrics adapterMetrics,
+      AdapterResourceManager adapterResourceManager,
+      DataStreamResourceManager dataStreamResourceManager
+  ) {
     this.adapterHealthCheck = new AdapterHealthCheck(
-        StorageDispatcher.INSTANCE.getNoSqlStore()
-                                  .getAdapterInstanceStorage(),
-        new AdapterMasterManagement()
+        adapterStorage,
+        new AdapterMasterManagement(
+            adapterStorage,
+            adapterResourceManager,
+            dataStreamResourceManager,
+            adapterMetrics
+        )
     );
     this.adapterDescriptionStorage = CouchDbStorageManager.INSTANCE.getAdapterDescriptionStorage();
   }
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java
index ff60661596..31f3fbe720 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java
@@ -19,6 +19,7 @@
 package org.apache.streampipes.rest;
 
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.AdapterMasterManagement;
 import org.apache.streampipes.dataexplorer.DataExplorerQueryManagement;
 import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement;
@@ -30,6 +31,7 @@ import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.file.FileMetadata;
 import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.resource.management.UserResourceManager;
 import org.apache.streampipes.storage.api.IDashboardStorage;
 import org.apache.streampipes.storage.api.IDashboardWidgetStorage;
@@ -72,7 +74,13 @@ public class ResetManagement {
     });
 
     // Stop and delete all adapters
-    AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement();
+    AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement(
+        StorageDispatcher.INSTANCE.getNoSqlStore()
+                                  .getAdapterInstanceStorage(),
+        new SpResourceManager().manageAdapters(),
+        new SpResourceManager().manageDataStreams(),
+        AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+    );
 
     try {
       List<AdapterDescription> allAdapters = adapterMasterManagement.getAllAdapterInstances();
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
index 4980bd993f..b34d50a73e 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
@@ -19,6 +19,7 @@
 package org.apache.streampipes.rest.impl.connect;
 
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.AdapterMasterManagement;
 import org.apache.streampipes.connect.management.management.AdapterUpdateManagement;
 import org.apache.streampipes.manager.pipeline.PipelineManager;
@@ -28,6 +29,7 @@ import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.message.Notifications;
 import org.apache.streampipes.model.monitoring.SpLogMessage;
 import org.apache.streampipes.resource.management.PermissionResourceManager;
+import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.rest.security.AuthConstants;
 import org.apache.streampipes.rest.shared.annotation.JacksonSerialized;
 import org.apache.streampipes.storage.api.IPipelineStorage;
@@ -61,7 +63,13 @@ public class AdapterResource extends AbstractAdapterResource<AdapterMasterManage
   private static final Logger LOG = LoggerFactory.getLogger(AdapterResource.class);
 
   public AdapterResource() {
-    super(AdapterMasterManagement::new);
+    super(() -> new AdapterMasterManagement(
+        StorageDispatcher.INSTANCE.getNoSqlStore()
+                                  .getAdapterInstanceStorage(),
+        new SpResourceManager().manageAdapters(),
+        new SpResourceManager().manageDataStreams(),
+        AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+    ));
   }
 
   @POST
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java
index c8e5386b34..208ed85e28 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java
@@ -21,13 +21,16 @@ package org.apache.streampipes.rest.impl.connect;
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
 import org.apache.streampipes.commons.exceptions.SpConfigurationException;
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.WorkerAdministrationManagement;
 import org.apache.streampipes.connect.management.management.WorkerRestClient;
 import org.apache.streampipes.connect.management.management.WorkerUrlProvider;
 import org.apache.streampipes.model.monitoring.SpLogMessage;
 import org.apache.streampipes.model.runtime.RuntimeOptionsRequest;
 import org.apache.streampipes.model.runtime.RuntimeOptionsResponse;
+import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.rest.shared.annotation.JacksonSerialized;
+import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -49,7 +52,13 @@ public class RuntimeResolvableResource extends AbstractAdapterResource<WorkerAdm
   private final WorkerUrlProvider workerUrlProvider;
 
   public RuntimeResolvableResource() {
-    super(WorkerAdministrationManagement::new);
+    super(() -> new WorkerAdministrationManagement(
+        StorageDispatcher.INSTANCE.getNoSqlStore()
+                                  .getAdapterInstanceStorage(),
+        AdapterMetricsManager.INSTANCE.getAdapterMetrics(),
+        new SpResourceManager().manageAdapters(),
+        new SpResourceManager().manageDataStreams()
+    ));
     this.workerUrlProvider = new WorkerUrlProvider();
   }
 
diff --git a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/security/UnauthorizedRequestEntryPoint.java b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/security/UnauthorizedRequestEntryPoint.java
index 144a990dc4..94304a6d5b 100644
--- a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/security/UnauthorizedRequestEntryPoint.java
+++ b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/security/UnauthorizedRequestEntryPoint.java
@@ -35,7 +35,7 @@ public class UnauthorizedRequestEntryPoint implements AuthenticationEntryPoint {
   @Override
   public void commence(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse,
                        AuthenticationException e) throws IOException {
-    LOG.error("Unauthorized request to {}", httpServletRequest.getPathInfo());
+    LOG.error("Unauthorized request to {}", httpServletRequest.getServletPath());
 
     httpServletResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED, e.getLocalizedMessage());
   }
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
index 2442816746..76ba815dcc 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
@@ -18,11 +18,13 @@
 
 package org.apache.streampipes.service.core;
 
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.WorkerAdministrationManagement;
 import org.apache.streampipes.manager.operations.Operations;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
 import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.storage.api.IPipelineStorage;
 import org.apache.streampipes.storage.couchdb.CouchDbStorageManager;
 import org.apache.streampipes.storage.management.StorageDispatcher;
@@ -52,7 +54,13 @@ public class PostStartupTask implements Runnable {
   public PostStartupTask(IPipelineStorage pipelineStorage) {
     this.pipelineStorage = pipelineStorage;
     this.executorService = Executors.newSingleThreadScheduledExecutor();
-    this.workerAdministrationManagement = new WorkerAdministrationManagement();
+    this.workerAdministrationManagement = new WorkerAdministrationManagement(
+        StorageDispatcher.INSTANCE.getNoSqlStore()
+                                  .getAdapterInstanceStorage(),
+        AdapterMetricsManager.INSTANCE.getAdapterMetrics(),
+        new SpResourceManager().manageAdapters(),
+        new SpResourceManager().manageDataStreams()
+    );
   }
 
   @Override
@@ -63,11 +71,13 @@ public class PostStartupTask implements Runnable {
   }
 
   private void performAdapterAssetUpdate() {
-    var installedAppIds = CouchDbStorageManager.INSTANCE.getExtensionsServiceStorage().getAll()
-        .stream()
-        .flatMap(config -> config.getTags().stream())
-        .filter(tag -> tag.getPrefix() == SpServiceTagPrefix.ADAPTER)
-        .toList();
+    var installedAppIds = CouchDbStorageManager.INSTANCE.getExtensionsServiceStorage()
+                                                        .getAll()
+                                                        .stream()
+                                                        .flatMap(config -> config.getTags()
+                                                                                 .stream())
+                                                        .filter(tag -> tag.getPrefix() == SpServiceTagPrefix.ADAPTER)
+                                                        .toList();
     workerAdministrationManagement.performAdapterMigrations(installedAppIds);
   }
 
@@ -118,16 +128,20 @@ public class PostStartupTask implements Runnable {
       storeFailedRestartAttempt(pipeline);
       int failedAttemptCount = failedPipelines.get(pipeline.getPipelineId());
       if (failedAttemptCount <= MAX_PIPELINE_START_RETRIES) {
-        LOG.error("Pipeline {} could not be restarted - I'll try again in {} seconds ({}/{} failed attempts)",
+        LOG.error(
+            "Pipeline {} could not be restarted - I'll try again in {} seconds ({}/{} failed attempts)",
             pipeline.getName(),
             WAIT_TIME_AFTER_FAILURE_IN_SECONDS,
             failedAttemptCount,
-            MAX_PIPELINE_START_RETRIES);
+            MAX_PIPELINE_START_RETRIES
+        );
 
         schedulePipelineStart(pipeline, restartOnReboot);
       } else {
-        LOG.error("Pipeline {} could not be restarted - are all pipeline element containers running?",
-            status.getPipelineName());
+        LOG.error(
+            "Pipeline {} could not be restarted - are all pipeline element containers running?",
+            status.getPipelineName()
+        );
       }
     }
   }
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
index d8c9781bd7..f08e9939e7 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
@@ -18,6 +18,7 @@
 package org.apache.streampipes.service.core;
 
 import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.health.AdapterHealthCheck;
 import org.apache.streampipes.connect.management.management.AdapterMasterManagement;
 import org.apache.streampipes.manager.health.CoreInitialInstallationProgress;
@@ -37,6 +38,7 @@ import org.apache.streampipes.messaging.pulsar.SpPulsarProtocolFactory;
 import org.apache.streampipes.model.configuration.SpCoreConfigurationStatus;
 import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.rest.security.SpPermissionEvaluator;
 import org.apache.streampipes.service.base.BaseNetworkingConfig;
 import org.apache.streampipes.service.base.StreamPipesServiceBase;
@@ -142,7 +144,13 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase {
             new PipelineHealthCheck(),
             new AdapterHealthCheck(
                 StorageDispatcher.INSTANCE.getNoSqlStore().getAdapterInstanceStorage(),
-                new AdapterMasterManagement()
+                new AdapterMasterManagement(
+                    StorageDispatcher.INSTANCE.getNoSqlStore()
+                                              .getAdapterInstanceStorage(),
+                    new SpResourceManager().manageAdapters(),
+                    new SpResourceManager().manageDataStreams(),
+                    AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+                )
             ))
     );
 
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java
index e389f4afbf..ac4c833594 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java
@@ -24,15 +24,18 @@ import org.apache.streampipes.model.client.user.ServiceAccount;
 import org.apache.streampipes.model.client.user.UserAccount;
 import org.apache.streampipes.storage.api.IUserStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
+import org.apache.streampipes.user.management.encryption.SecretEncryptionManager;
 import org.apache.streampipes.user.management.jwt.JwtTokenProvider;
 import org.apache.streampipes.user.management.model.PrincipalUserDetails;
 import org.apache.streampipes.user.management.model.ServiceAccountDetails;
 import org.apache.streampipes.user.management.model.UserAccountDetails;
 import org.apache.streampipes.user.management.service.TokenService;
+import org.apache.streampipes.user.management.util.PasswordUtil;
 import org.apache.streampipes.user.management.util.TokenUtil;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpHeaders;
 import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
 import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.security.web.authentication.WebAuthenticationDetailsSource;
@@ -45,12 +48,20 @@ import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
 
 import java.io.IOException;
+import java.security.NoSuchAlgorithmException;
+import java.security.spec.InvalidKeySpecException;
+import java.util.Base64;
+import java.util.List;
 
 public class TokenAuthenticationFilter extends OncePerRequestFilter {
 
   private final JwtTokenProvider tokenProvider;
   private final IUserStorage userStorage;
 
+  private final List<String> supportedBasicAuthPaths = List.of(
+      "/actuator/prometheus"
+  );
+
   private static final Logger logger = LoggerFactory.getLogger(TokenAuthenticationFilter.class);
 
   public TokenAuthenticationFilter() {
@@ -68,7 +79,7 @@ public class TokenAuthenticationFilter extends OncePerRequestFilter {
       if (StringUtils.hasText(jwt) && tokenProvider.validateJwtToken(jwt)) {
         String username = tokenProvider.getUserIdFromToken(jwt);
         applySuccessfulAuth(request, username);
-      } else {
+      } else if (isApiKeyAuth(request)) {
         String apiKey = getApiKeyFromRequest(request);
         String apiUser = getApiUserFromRequest(request);
         if (StringUtils.hasText(apiKey) && StringUtils.hasText(apiUser)) {
@@ -78,6 +89,22 @@ public class TokenAuthenticationFilter extends OncePerRequestFilter {
             applySuccessfulAuth(request, apiUser);
           }
         }
+      } else {
+        var authorizationHeader = request.getHeader(HttpHeaders.AUTHORIZATION);
+        if (authorizationHeader != null && authorizationHeader.startsWith(HttpConstants.BASIC)) {
+          if (supportedBasicAuthPaths.contains(request.getServletPath())) {
+            String base64Credentials = authorizationHeader.substring(HttpConstants.BASIC.length()).trim();
+            String credentials = new String(Base64.getDecoder().decode(base64Credentials));
+
+            String[] splitCredentials = credentials.split(":");
+            String username = splitCredentials[0];
+            String passphrase = splitCredentials[1];
+            var principal = StorageDispatcher.INSTANCE.getNoSqlStore().getUserStorageAPI().getUser(username);
+            if (principal != null && checkCredentials(principal, passphrase)) {
+              applySuccessfulAuth(request, username);
+            }
+          }
+        }
       }
     } catch (Exception ex) {
       logger.error("Could not set user authentication in security context", ex);
@@ -86,6 +113,21 @@ public class TokenAuthenticationFilter extends OncePerRequestFilter {
     filterChain.doFilter(request, response);
   }
 
+  private boolean checkCredentials(Principal principal, String passphrase)
+      throws NoSuchAlgorithmException, InvalidKeySpecException {
+    if (principal instanceof UserAccount) {
+      return PasswordUtil.validatePassword(passphrase, ((UserAccount) principal).getPassword());
+    } else if (principal instanceof ServiceAccount) {
+      return passphrase.equals(SecretEncryptionManager.decrypt(((ServiceAccount) principal).getClientSecret()));
+    } else {
+      throw new IllegalArgumentException("Unknown user instance");
+    }
+  }
+
+  private boolean isApiKeyAuth(HttpServletRequest request) {
+    return request.getHeader(HttpConstants.X_API_USER) != null && request.getHeader(HttpConstants.X_API_KEY) != null;
+  }
+
   private void applySuccessfulAuth(HttpServletRequest request,
                                    String username) {
     Principal user = userStorage.getUser(username);


Test patch:
diff --git a/streampipes-commons/src/test/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetricsTest.java b/streampipes-commons/src/test/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetricsTest.java
new file mode 100644
index 0000000000..24370f10c5
--- /dev/null
+++ b/streampipes-commons/src/test/java/org/apache/streampipes/commons/prometheus/adapter/AdapterMetricsTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.streampipes.commons.prometheus.adapter;
+
+import org.junit.Test;
+
+import java.util.NoSuchElementException;
+
+import static junit.framework.TestCase.assertEquals;
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertThrows;
+
+public class AdapterMetricsTest {
+
+  private static final String ADAPTER_ID = "adapterId";
+  private static final String ADAPTER_NAME = "adapterName";
+  private static final String OTHER_ADAPTER_ID = "otherAdapterId";
+
+  @Test
+  public void register() {
+
+    var metrics = AdapterMetricsManager.INSTANCE.getAdapterMetrics();
+
+    assertEquals(0, metrics.size());
+
+    metrics.register(ADAPTER_ID, ADAPTER_NAME);
+
+    assertEquals(1, metrics.size());
+    assertTrue(metrics.contains(ADAPTER_ID));
+  }
+
+  @Test
+  public void removeNoSuchElement() {
+    var metrics = AdapterMetricsManager.INSTANCE.getAdapterMetrics();
+
+    assertThrows(NoSuchElementException.class, () -> metrics.remove(OTHER_ADAPTER_ID, ADAPTER_NAME));
+  }
+
+  @Test
+  public void updateTotalEventsPublishedNoSuchElement() {
+    var metrics = AdapterMetricsManager.INSTANCE.getAdapterMetrics();
+
+    assertThrows(
+        NoSuchElementException.class,
+        () -> metrics.updateTotalEventsPublished(OTHER_ADAPTER_ID, ADAPTER_NAME, 0)
+    );
+
+  }
+}
diff --git a/streampipes-commons/src/test/java/TestResources.java b/streampipes-commons/src/test/java/org/apache/streampipes/commons/resources/TestResources.java
similarity index 90%
rename from streampipes-commons/src/test/java/TestResources.java
rename to streampipes-commons/src/test/java/org/apache/streampipes/commons/resources/TestResources.java
index ec7174e336..7db525a3a7 100644
--- a/streampipes-commons/src/test/java/TestResources.java
+++ b/streampipes-commons/src/test/java/org/apache/streampipes/commons/resources/TestResources.java
@@ -16,7 +16,7 @@
  *
  */
 
-import org.apache.streampipes.commons.resources.Resources;
+package org.apache.streampipes.commons.resources;
 
 import org.junit.Test;
 
@@ -31,6 +31,6 @@ public class TestResources {
   public void testResourceLoader() throws IOException {
     String filename = "test.txt";
     String content = Resources.asString(filename, StandardCharsets.UTF_8);
-    assertEquals("HelloWorld", content.replaceAll("\\s+",""));
+    assertEquals("HelloWorld", content.replaceAll("\\s+", ""));
   }
 }
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 12f34c4c5c..8eedbd725f 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
@@ -18,9 +18,12 @@
 
 package org.apache.streampipes.connect.management.health;
 
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.AdapterMasterManagement;
 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;
@@ -45,7 +48,16 @@ public class AdapterHealthCheckTest {
   public void getAllRunningInstancesAdapterDescriptionsEmpty() {
     when(adapterInstanceStorageMock.getAllAdapters()).thenReturn(List.of());
 
-    var healthCheck = new AdapterHealthCheck(adapterInstanceStorageMock, new AdapterMasterManagement());
+    var healthCheck = new AdapterHealthCheck(
+        adapterInstanceStorageMock,
+        new AdapterMasterManagement(
+            StorageDispatcher.INSTANCE.getNoSqlStore()
+                                      .getAdapterInstanceStorage(),
+            new SpResourceManager().manageAdapters(),
+            new SpResourceManager().manageDataStreams(),
+            AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+        )
+    );
     var result = healthCheck.getAllRunningInstancesAdapterDescriptions();
 
     assertTrue(result.isEmpty());
@@ -67,7 +79,16 @@ public class AdapterHealthCheckTest {
 
     when(adapterInstanceStorageMock.getAllAdapters()).thenReturn(List.of(stoppedAdapter, runningAdapter));
 
-    var healthCheck = new AdapterHealthCheck(adapterInstanceStorageMock, new AdapterMasterManagement());
+    var healthCheck = new AdapterHealthCheck(
+        adapterInstanceStorageMock,
+        new AdapterMasterManagement(
+            StorageDispatcher.INSTANCE.getNoSqlStore()
+                                      .getAdapterInstanceStorage(),
+            new SpResourceManager().manageAdapters(),
+            new SpResourceManager().manageDataStreams(),
+            AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+        )
+    );
     var result = healthCheck.getAllRunningInstancesAdapterDescriptions();
 
     assertEquals(1, result.size());
diff --git a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java
index bafde99a37..30b4c07ac1 100644
--- a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java
+++ b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java
@@ -19,6 +19,7 @@
 package org.apache.streampipes.connect.management.management;
 
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.resource.management.AdapterResourceManager;
 import org.apache.streampipes.storage.couchdb.impl.AdapterInstanceStorageImpl;
@@ -40,7 +41,12 @@ public class AdapterMasterManagementTest {
     when(adapterStorage.getAllAdapters()).thenReturn(null);
 
     AdapterMasterManagement adapterMasterManagement =
-        new AdapterMasterManagement(adapterStorage, resourceManager, null);
+        new AdapterMasterManagement(
+            adapterStorage,
+            resourceManager,
+            null,
+            AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+        );
 
     adapterMasterManagement.getAdapter("id2");
   }
@@ -53,7 +59,12 @@ public class AdapterMasterManagementTest {
     when(adapterStorage.getAllAdapters()).thenReturn(adapterDescriptions);
 
     AdapterMasterManagement adapterMasterManagement =
-        new AdapterMasterManagement(adapterStorage, resourceManager, null);
+        new AdapterMasterManagement(
+            adapterStorage,
+            resourceManager,
+            null,
+            AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+        );
 
     adapterMasterManagement.getAdapter("id2");
   }
@@ -66,7 +77,12 @@ public class AdapterMasterManagementTest {
     when(adapterStorage.getAllAdapters()).thenReturn(adapterDescriptions);
 
     AdapterMasterManagement adapterMasterManagement =
-        new AdapterMasterManagement(adapterStorage, resourceManager, null);
+        new AdapterMasterManagement(
+            adapterStorage,
+            resourceManager,
+            null,
+            AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+        );
 
     List<AdapterDescription> result = adapterMasterManagement.getAllAdapterInstances();
 
@@ -80,7 +96,12 @@ public class AdapterMasterManagementTest {
     when(adapterStorage.getAllAdapters()).thenReturn(null);
 
     AdapterMasterManagement adapterMasterManagement =
-        new AdapterMasterManagement(adapterStorage, resourceManager, null);
+        new AdapterMasterManagement(
+            adapterStorage,
+            resourceManager,
+            null,
+            AdapterMetricsManager.INSTANCE.getAdapterMetrics()
+        );
 
     adapterMasterManagement.getAllAdapterInstances();