pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rdhaba...@apache.org
Subject [incubator-pulsar] branch master updated: Add worker specific system and jvm metrics (#2352)
Date Fri, 10 Aug 2018 07:21:57 GMT
This is an automated email from the ASF dual-hosted git repository.

rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 10c3511  Add worker specific system and jvm metrics (#2352)
10c3511 is described below

commit 10c351134a36457d3734a356b798b9f18b8d8593
Author: Rajan Dhabalia <rdhabalia@apache.org>
AuthorDate: Fri Aug 10 00:21:55 2018 -0700

    Add worker specific system and jvm metrics (#2352)
---
 .../apache/pulsar/broker/admin/v2/WorkerStats.java | 11 ++++++-
 .../broker/loadbalance/impl/LoadManagerShared.java |  4 +--
 .../pulsar/broker/stats/MetricsGenerator.java      |  4 +--
 .../prometheus/PrometheusMetricsGenerator.java     |  4 +--
 .../stats/BookieClientsStatsGeneratorTest.java     |  2 +-
 .../apache/pulsar/client/admin/WorkerStats.java    |  9 ++++++
 .../client/admin/internal/WorkerStatsImpl.java     | 18 ++++++++++-
 .../pulsar/admin/cli/CmdFunctionWorkerStats.java   | 24 +++++++++++---
 .../apache/pulsar/common/stats}/JvmMetrics.java    | 32 +++++++++++++------
 .../org/apache/pulsar/common/stats/Metrics.java    |  5 +++
 .../pulsar/functions/worker/MetricsGenerator.java  | 37 ++++++++++++++--------
 .../pulsar/functions/worker/WorkerService.java     |  3 ++
 .../functions/worker/rest/api/FunctionsImpl.java   | 21 +++++++++++-
 .../functions/worker/rest/api/v2/WorkerStats.java  | 17 +++++++---
 14 files changed, 150 insertions(+), 41 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/WorkerStats.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/WorkerStats.java
index 962c483..5a0e4b7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/WorkerStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/WorkerStats.java
@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.admin.v2;
 
 import java.io.IOException;
+import java.util.Collection;
 
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
@@ -41,7 +42,15 @@ public class WorkerStats extends FunctionApiResource {
     @ApiOperation(value = "Get metrics for all functions owned by worker", notes = "Requested
should be executed by Monitoring agent on each worker to fetch the metrics", response = Metrics.class)
     @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
             @ApiResponse(code = 503, message = "Worker service is not running") })
-    public Response getMetrics() throws IOException {
+    public Response getStats() throws IOException {
         return functions.getFunctionsMetrcis(clientAppId());
     }
+    
+    @GET
+    @Path("/metrics")
+    @ApiOperation(value = "Gets the metrics for Monitoring", notes = "Request should be executed
by Monitoring agent on each worker to fetch the worker-metrics", response = org.apache.pulsar.common.stats.Metrics.class,
responseContainer = "List")
+    @ApiResponses(value = { @ApiResponse(code = 401, message = "Don't have admin permission")
})
+    public Collection<org.apache.pulsar.common.stats.Metrics> getMetrics() throws Exception
{
+        return functions.getWorkerMetrcis(clientAppId());
+    }
 }
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java
index 80cb31c..6a5575c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java
@@ -41,7 +41,7 @@ import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.loadbalance.BrokerHostUsage;
 import org.apache.pulsar.broker.loadbalance.LoadData;
-import org.apache.pulsar.broker.stats.metrics.JvmMetrics;
+import static org.apache.pulsar.common.stats.JvmMetrics.getJvmDirectMemoryUsed;
 import org.apache.pulsar.common.naming.NamespaceBundle;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.ServiceUnitId;
@@ -230,7 +230,7 @@ public class LoadManagerShared {
         systemResourceUsage.memory.limit = (double) maxHeapMemoryInBytes / MIBI;
 
         // Collect JVM direct memory
-        systemResourceUsage.directMemory.usage = (double) (JvmMetrics.getJvmDirectMemoryUsed()
/ MIBI);
+        systemResourceUsage.directMemory.usage = (double) (getJvmDirectMemoryUsed() / MIBI);
         systemResourceUsage.directMemory.limit = (double) (PlatformDependent.maxDirectMemory()
/ MIBI);
 
         return systemResourceUsage;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/MetricsGenerator.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/MetricsGenerator.java
index 9472a7d..3aaa955 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/MetricsGenerator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/MetricsGenerator.java
@@ -24,7 +24,7 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.broker.stats.metrics.JvmMetrics;
+import org.apache.pulsar.common.stats.JvmMetrics;
 import org.apache.pulsar.broker.stats.metrics.ManagedLedgerCacheMetrics;
 import org.apache.pulsar.broker.stats.metrics.ManagedLedgerMetrics;
 import org.apache.pulsar.common.stats.Metrics;
@@ -38,7 +38,7 @@ public class MetricsGenerator {
 
     public MetricsGenerator(PulsarService pulsar) {
         this.pulsar = pulsar;
-        this.jvmMetrics = new JvmMetrics(pulsar);
+        this.jvmMetrics = new JvmMetrics(pulsar.getExecutor(), "brk");
     }
 
     public Collection<Metrics> generate() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
index 8f0d30d..5c42b51 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
@@ -23,7 +23,7 @@ import java.io.OutputStream;
 import java.util.Enumeration;
 
 import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.broker.stats.metrics.JvmMetrics;
+import static org.apache.pulsar.common.stats.JvmMetrics.getJvmDirectMemoryUsed;
 import org.apache.pulsar.common.util.SimpleTextOutputStream;
 
 import io.netty.buffer.ByteBuf;
@@ -51,7 +51,7 @@ public class PrometheusMetricsGenerator {
         Gauge.build("jvm_memory_direct_bytes_used", "-").create().setChild(new Child() {
             @Override
             public double get() {
-                return JvmMetrics.getJvmDirectMemoryUsed();
+                return getJvmDirectMemoryUsed();
             }
         }).register(CollectorRegistry.defaultRegistry);
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/BookieClientsStatsGeneratorTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/BookieClientsStatsGeneratorTest.java
index 251ed57..332bc07 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/BookieClientsStatsGeneratorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/BookieClientsStatsGeneratorTest.java
@@ -25,7 +25,7 @@ import java.util.Map;
 import org.apache.bookkeeper.mledger.proto.PendingBookieOpsStats;
 import org.apache.pulsar.broker.service.BrokerTestBase;
 import org.apache.pulsar.broker.stats.BookieClientStatsGenerator;
-import org.apache.pulsar.broker.stats.metrics.JvmMetrics;
+import org.apache.pulsar.common.stats.JvmMetrics;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/WorkerStats.java
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/WorkerStats.java
index 069e58f..4fc242f 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/WorkerStats.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/WorkerStats.java
@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.client.admin;
 
+import java.util.Collection;
+
 import org.apache.pulsar.functions.proto.InstanceCommunication.Metrics;
 
 /**
@@ -32,4 +34,11 @@ public interface WorkerStats {
      * @throws PulsarAdminException 
      */
     Metrics getFunctionsStats() throws PulsarAdminException;
+    
+    /**
+     * Get worker metrics.
+     * @return
+     * @throws PulsarAdminException
+     */
+    Collection<org.apache.pulsar.common.stats.Metrics> getMetrics() throws PulsarAdminException;
 }
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerStatsImpl.java
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerStatsImpl.java
index 5b6762c..f492d31 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerStatsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/WorkerStatsImpl.java
@@ -18,15 +18,20 @@
  */
 package org.apache.pulsar.client.admin.internal;
 
+import static org.apache.pulsar.client.admin.internal.FunctionsImpl.mergeJson;
+
+import java.util.Collection;
+import java.util.List;
+
 import javax.ws.rs.ClientErrorException;
 import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.GenericType;
 import javax.ws.rs.core.Response;
 
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.WorkerStats;
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.functions.proto.InstanceCommunication.Metrics;
-import static org.apache.pulsar.client.admin.internal.FunctionsImpl.mergeJson;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -55,4 +60,15 @@ public class WorkerStatsImpl extends BaseResource implements WorkerStats
{
            throw getApiException(e);
        }
    }
+
+    @Override
+    public Collection<org.apache.pulsar.common.stats.Metrics> getMetrics() throws PulsarAdminException
{
+        try {
+            return request(workerStats.path("metrics"))
+                    .get(new GenericType<List<org.apache.pulsar.common.stats.Metrics>>()
{
+                    });
+        } catch (Exception e) {
+            throw getApiException(e);
+        }
+    }
 }
\ No newline at end of file
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctionWorkerStats.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctionWorkerStats.java
index 640d7d7..6bd22ae 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctionWorkerStats.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctionWorkerStats.java
@@ -24,6 +24,7 @@ import org.apache.pulsar.functions.utils.Utils;
 
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
+import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.JsonParser;
 
@@ -34,8 +35,6 @@ import lombok.extern.slf4j.Slf4j;
 @Parameters(commandDescription = "Operations to collect function-worker statistics")
 public class CmdFunctionWorkerStats extends CmdBase {
 
-    private final FunctionsStats functionsStats;
-
     /**
      * Base command
      */
@@ -70,10 +69,27 @@ public class CmdFunctionWorkerStats extends CmdBase {
         }
     }
 
+    @Parameters(commandDescription = "dump metrics for Monitoring")
+    class CmdMonitoringMetrics extends BaseCommand {
+
+        @Parameter(names = { "-i", "--indent" }, description = "Indent JSON output", required
= false)
+        boolean indent = false;
+
+        @Override
+        void runCmd() throws Exception {
+            String json = new Gson().toJson(admin.workerStats().getMetrics());
+            GsonBuilder gsonBuilder = new GsonBuilder();
+            if (indent) {
+                gsonBuilder.setPrettyPrinting();
+            }
+            System.out.println(gsonBuilder.create().toJson(new JsonParser().parse(json)));
+        }
+    }
+
     public CmdFunctionWorkerStats(PulsarAdmin admin) throws PulsarClientException {
         super("functions", admin);
-        functionsStats = new FunctionsStats();
-        jcommander.addCommand("functions", functionsStats);
+        jcommander.addCommand("functions", new FunctionsStats());
+        jcommander.addCommand("monitoring-metrics", new CmdMonitoringMetrics());
     }
 
 }
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/metrics/JvmMetrics.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/JvmMetrics.java
similarity index 87%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/metrics/JvmMetrics.java
rename to pulsar-common/src/main/java/org/apache/pulsar/common/stats/JvmMetrics.java
index d6870a5..6353dfa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/metrics/JvmMetrics.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/JvmMetrics.java
@@ -16,18 +16,20 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.broker.stats.metrics;
+package org.apache.pulsar.common.stats;
 
 import java.lang.management.ManagementFactory;
 import java.lang.reflect.Field;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import javax.management.MBeanServer;
 import javax.management.MalformedObjectNameException;
 import javax.management.ObjectName;
 
-import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.common.stats.Metrics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,7 +43,7 @@ import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.util.internal.PlatformDependent;
 import java.util.concurrent.atomic.AtomicLong;
 
-public class JvmMetrics extends AbstractMetrics {
+public class JvmMetrics {
 
     private volatile long accumulatedYoungGcCount = 0;
     private volatile long currentYoungGcCount = 0;
@@ -55,6 +57,8 @@ public class JvmMetrics extends AbstractMetrics {
     
     private static final Logger log = LoggerFactory.getLogger(JvmMetrics.class);
     private static Field directMemoryUsage = null;
+    
+    private final String componentName;
     static {
         try {
             directMemoryUsage = PlatformDependent.class.getDeclaredField("DIRECT_MEMORY_COUNTER");
@@ -64,13 +68,14 @@ public class JvmMetrics extends AbstractMetrics {
         }
     }
 
-    public JvmMetrics(PulsarService pulsar) {
-        super(pulsar);
-        pulsar.getExecutor().scheduleAtFixedRate(this::updateGcStats, 0, 1, TimeUnit.MINUTES);
+    public JvmMetrics(ScheduledExecutorService executor, String componentName) {
+        if (executor != null) {
+            executor.scheduleAtFixedRate(this::updateGcStats, 0, 1, TimeUnit.MINUTES);
+        }
+        this.componentName = componentName;
     }
 
     @SuppressWarnings("restriction")
-    @Override
     public List<Metrics> generate() {
 
         Metrics m = createMetrics();
@@ -105,8 +110,8 @@ public class JvmMetrics extends AbstractMetrics {
             }
         }
 
-        m.put("brk_default_pool_allocated", totalAllocated);
-        m.put("brk_default_pool_used", totalUsed);
+        m.put(this.componentName + "_default_pool_allocated", totalAllocated);
+        m.put(this.componentName + "_default_pool_used", totalUsed);
 
         return Lists.newArrayList(m);
     }
@@ -173,5 +178,14 @@ public class JvmMetrics extends AbstractMetrics {
 
         return parentThreadGroup.activeCount();
     }
+    
+    private Metrics createMetrics() {
+        return createMetrics(Collections.singletonMap("metric", "jvm_metrics"));
+    }
+
+    private Metrics createMetrics(Map<String, String> dimensionMap) {
+        // create with current version
+        return Metrics.create(dimensionMap);
+    }
 
 }
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Metrics.java b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Metrics.java
index 0281da7..045acc1 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Metrics.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/Metrics.java
@@ -43,6 +43,11 @@ public class Metrics {
     @JsonInclude(content=Include.NON_EMPTY)
     final Map<String, String> dimensions;
 
+    public Metrics() {
+        metrics = Maps.newTreeMap();
+        dimensions = Maps.newHashMap();
+    }
+    
     // hide constructor
     protected Metrics(Map<String, String> unmodifiableDimensionMap) {
         this.metrics = Maps.newTreeMap();
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/WorkerStats.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MetricsGenerator.java
similarity index 53%
copy from pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/WorkerStats.java
copy to pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MetricsGenerator.java
index 069e58f..59e71d4 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/WorkerStats.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/MetricsGenerator.java
@@ -16,20 +16,29 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.client.admin;
+package org.apache.pulsar.functions.worker;
 
-import org.apache.pulsar.functions.proto.InstanceCommunication.Metrics;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.apache.pulsar.common.stats.JvmMetrics;
+import org.apache.pulsar.common.stats.Metrics;
+
+public class MetricsGenerator {
+
+    private final JvmMetrics jvmMetrics;
+
+    public MetricsGenerator(ScheduledExecutorService executor) {
+        this.jvmMetrics = new JvmMetrics(executor, "fun");
+    }
+
+    public List<Metrics> generate() {
+        List<Metrics> metricsCollection = new ArrayList<Metrics>();
+        metricsCollection.addAll(jvmMetrics.generate());
+        // add more metrics here..
+
+        return metricsCollection;
+    }
 
-/**
- * Admin interface for worker stats management.
- */
-public interface WorkerStats {
-    
-    
-    /**
-     * Get all functions stats on a worker
-     * @return
-     * @throws PulsarAdminException 
-     */
-    Metrics getFunctionsStats() throws PulsarAdminException;
 }
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
index 6af9c8f..e4cf2cc 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java
@@ -41,6 +41,7 @@ import org.apache.pulsar.client.api.ClientBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
+import org.apache.pulsar.common.stats.JvmMetrics;
 
 /**
  * A service component contains everything to run a worker except rest server.
@@ -63,11 +64,13 @@ public class WorkerService {
     private AuthenticationService authenticationService;
     private ConnectorsManager connectorsManager;
     private PulsarAdmin admin;
+    private final MetricsGenerator metricsGenerator;
 
     public WorkerService(WorkerConfig workerConfig) {
         this.workerConfig = workerConfig;
         this.statsUpdater = Executors
                 .newSingleThreadScheduledExecutor(new DefaultThreadFactory("worker-stats-updater"));
+        this.metricsGenerator = new MetricsGenerator(this.statsUpdater);
     }
 
     public void start(URI dlogUri) throws InterruptedException {
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index c35dd52..b71e6cd 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -60,6 +60,7 @@ import static org.apache.commons.lang3.StringUtils.isBlank;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.apache.commons.lang3.StringUtils.isEmpty;
 import static org.apache.commons.lang3.StringUtils.join;
+
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
@@ -976,7 +977,25 @@ public class FunctionsImpl {
     public boolean isSuperUser(String clientRole) {
         return clientRole != null && worker().getWorkerConfig().getSuperUserRoles().contains(clientRole);
     }
-    
+
+    public List<org.apache.pulsar.common.stats.Metrics> getWorkerMetrcis(String clientRole)
throws IOException {
+        if (worker().getWorkerConfig().isAuthorizationEnabled() && !isSuperUser(clientRole))
{
+            log.error("Client [{}] is not admin and authorized to get function-stats", clientRole);
+            throw new WebApplicationException(Response.status(Status.UNAUTHORIZED).type(MediaType.APPLICATION_JSON)
+                    .entity(new ErrorData(clientRole + " is not authorize to get metrics")).build());
+        }
+        return getWorkerMetrcis();
+    }
+
+    private List<org.apache.pulsar.common.stats.Metrics> getWorkerMetrcis() {
+        if (!isWorkerServiceAvailable()) {
+            throw new WebApplicationException(
+                    Response.status(Status.SERVICE_UNAVAILABLE).type(MediaType.APPLICATION_JSON)
+                            .entity(new ErrorData("Function worker service is not avaialable")).build());
+        }
+        return worker().getMetricsGenerator().generate();
+    }
+
     public Response getFunctionsMetrcis(String clientRole) throws IOException {
         if (worker().getWorkerConfig().isAuthorizationEnabled() && !isSuperUser(clientRole))
{
             log.error("Client [{}] is not admin and authorized to get function-stats", clientRole);
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerStats.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerStats.java
index 6e4ae55..146bb21 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerStats.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerStats.java
@@ -19,6 +19,7 @@
 package org.apache.pulsar.functions.worker.rest.api.v2;
 
 import java.io.IOException;
+import java.util.Collection;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.GET;
@@ -27,6 +28,7 @@ import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 
+import org.apache.pulsar.functions.proto.InstanceCommunication.Metrics;
 import org.apache.pulsar.functions.worker.rest.FunctionApiResource;
 
 import io.swagger.annotations.Api;
@@ -34,7 +36,6 @@ import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiResponse;
 import io.swagger.annotations.ApiResponses;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.functions.proto.InstanceCommunication.Metrics;
 
 @Slf4j
 @Path("/worker-stats")
@@ -45,10 +46,18 @@ public class WorkerStats extends FunctionApiResource {
 
     @GET
     @Path("/functions")
-    @ApiOperation(value = "Get metrics for all functions owned by worker", notes = "Requested
should be executed by Monitoring agent on each worker to fetch the metrics", response = Metrics.class)
-    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+    @ApiOperation(value = "Get stats for all functions owned by worker", notes = "Request
should be executed by Monitoring agent on each worker to fetch the function-metrics", response
= Metrics.class)
+    @ApiResponses(value = { @ApiResponse(code = 401, message = "Don't have admin permission"),
             @ApiResponse(code = 503, message = "Worker service is not running") })
-    public Response getMetrics() throws IOException {
+    public Response getStats() throws IOException {
         return functions.getFunctionsMetrcis(clientAppId());
     }
+    
+    @GET
+    @Path("/metrics")
+    @ApiOperation(value = "Gets the metrics for Monitoring", notes = "Request should be executed
by Monitoring agent on each worker to fetch the worker-metrics", response = org.apache.pulsar.common.stats.Metrics.class,
responseContainer = "List")
+    @ApiResponses(value = { @ApiResponse(code = 401, message = "Don't have admin permission")
})
+    public Collection<org.apache.pulsar.common.stats.Metrics> getMetrics() throws Exception
{
+        return functions.getWorkerMetrcis(clientAppId());
+    }
 }


Mime
View raw message