Add Openmetric reporting for MI dataservices

components/data/data-services/org.wso2.micro.integrator.dataservices.core/src/main/java/org/wso2/micro/integrator/dataservices/core/DBInOutMessageReceiver.java

@@ -38,6 +38,8 @@
 public class DBInOutMessageReceiver extends RawXMLINOutMessageReceiver {
 
  private static final Log log = LogFactory.getLog(DBInOutMessageReceiver.class);
+
+ private static final String DATA_SERVICE_LATENCY_TIMER = "DATA_SERVICE_LATENCY_TIMER";
 
  /**
  * Invokes the business logic invocation on the service implementation class
@@ -94,6 +96,10 @@ public void invokeBusinessLogic(MessageContext msgContext,
  msgContext.setProperty(Constants.FAULT_NAME, DBConstants.DS_FAULT_NAME);
  throw DBUtils.createAxisFault(e);
  } finally {
+ if (msgContext.getProperty(DATA_SERVICE_LATENCY_TIMER) != null) {
+ newMsgContext.setProperty(DATA_SERVICE_LATENCY_TIMER,
+ msgContext.getProperty(DATA_SERVICE_LATENCY_TIMER));
+ }
  if (log.isDebugEnabled()) {
  String response;
  if (msgContext.getProperty(Constants.FAULT_NAME) != null &&

components/mediation/data-publishers/org.wso2.micro.integrator.observability/src/main/java/org/wso2/micro/integrator/observability/metric/handler/DSMetricHandler.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2023, WSO2 LLC. (http://www.wso2.org) All Rights Reserved.
+ *
+ * WSO2 LLC. 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.wso2.micro.integrator.observability.metric.handler;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.handlers.AbstractHandler;
+import org.wso2.micro.integrator.core.util.MicroIntegratorBaseUtils;
+import org.wso2.micro.integrator.observability.util.MetricConstants;
+import org.wso2.micro.integrator.observability.util.MetricUtils;
+
+public class DSMetricHandler extends AbstractHandler {
+
+ private MetricReporter metricReporter;
+ public DSMetricHandler() {
+ metricReporter = MetricUtils.getMetricReporter();
+ }
+
+ @Override
+ public InvocationResponse invoke(MessageContext messageContext) throws AxisFault {
+ if (MicroIntegratorBaseUtils.isDataService(messageContext)) {
+ String dataServiceName = messageContext.getAxisService().getName();
+ if (messageContext.isProcessingFault()) {
+ metricReporter.incrementCount(MetricConstants.DATA_SERVICE_REQUEST_COUNT_ERROR_TOTAL,
+ new String[]{dataServiceName, MetricConstants.DATA_SERVICE});
+ metricReporter.observeTime(messageContext.getProperty(MetricConstants.DATA_SERVICE_LATENCY_TIMER));
+ } else if (MetricConstants.MESSAGE_DIRECTION_IN.equalsIgnoreCase(
+ messageContext.getAxisMessage().getDirection())) {
+ metricReporter.incrementCount(MetricConstants.DATA_SERVICE_REQUEST_COUNT_TOTAL,
+ new String[]{dataServiceName, MetricConstants.DATA_SERVICE});
+ messageContext.setProperty(MetricConstants.DATA_SERVICE_LATENCY_TIMER,
+ metricReporter.getTimer(MetricConstants.DATA_SERVICE_LATENCY_SECONDS,
+ new String[]{dataServiceName,
+ MetricConstants.DATA_SERVICE}));
+ } else if (MetricConstants.MESSAGE_DIRECTION_OUT.equalsIgnoreCase(
+ messageContext.getAxisMessage().getDirection())) {
+ metricReporter.observeTime(messageContext.getProperty(MetricConstants.DATA_SERVICE_LATENCY_TIMER));
+ }
+ }
+ return InvocationResponse.CONTINUE;
+ }
+}

components/mediation/data-publishers/org.wso2.micro.integrator.observability/src/main/java/org/wso2/micro/integrator/observability/metric/handler/MetricHandler.java

@@ -29,13 +29,10 @@
 import org.apache.synapse.rest.RESTConstants;
 import org.apache.synapse.rest.RESTUtils;
 import org.apache.synapse.transport.nhttp.NhttpConstants;
-import org.wso2.config.mapper.ConfigParser;
 import org.wso2.micro.integrator.core.internal.MicroIntegratorBaseConstants;
 import org.wso2.micro.integrator.core.services.CarbonServerConfigurationService;
-import org.wso2.micro.integrator.observability.metric.handler.prometheus.reporter.PrometheusReporter;
 import org.wso2.micro.integrator.observability.util.MetricConstants;
-
-import java.util.Map;
+import org.wso2.micro.integrator.observability.util.MetricUtils;
 
 /**
  * Class for extracting metric information by wrapping the implementation and
@@ -44,7 +41,6 @@
 public class MetricHandler extends AbstractExtendedSynapseHandler {
 
  private static Log log = LogFactory.getLog(MetricHandler.class);
- private static final String METRIC_REPORTER = "metric_reporter";
  private static final String DELIMITER = "/";
  private static final String EMPTY = "";
 
@@ -61,7 +57,7 @@ public class MetricHandler extends AbstractExtendedSynapseHandler {
 
  @Override
  public boolean handleServerInit() {
- metricReporterInstance = this.getMetricReporter();
+ metricReporterInstance = MetricUtils.getMetricReporter();
  CarbonServerConfigurationService serverConfig = CarbonServerConfigurationService.getInstance();
  String miVersion = serverConfig.getServerVersion();
  String updateLevel = System.getProperty(MetricConstants.UPDATE_LEVEL);
@@ -71,47 +67,6 @@ public boolean handleServerInit() {
  return true;
  }
 
- /**
- * Load the MetricReporter class from the deployment.toml file if a user has defined a MetricReporter.
- * Use default PrometheusReporter if the user hasn't defined a MetricReporter or an error occurs
- * during custom MetricReporter class invocation.
- */
- private MetricReporter getMetricReporter() {
- Map<String, Object> configs = ConfigParser.getParsedConfigs();
- Object metricReporterClass = configs.get(MetricConstants.METRIC_HANDLER + "." + METRIC_REPORTER);
- Class loadedMetricClass;
- MetricReporter reporterInstance;
-
- if (metricReporterClass != null) {
- try {
- loadedMetricClass = Class.forName(metricReporterClass.toString());
- reporterInstance = (MetricReporter) loadedMetricClass.newInstance();
- if (log.isDebugEnabled()) {
- log.debug("The class " + metricReporterClass + " loaded successfully");
- }
- } catch (IllegalAccessException | ClassNotFoundException | InstantiationException e) {
- log.error("Error in loading the class " + metricReporterClass.toString() +
- " .Hence loading the default PrometheusReporter class ", e);
- reporterInstance = loadDefaultPrometheusReporter();
- }
- } else {
- reporterInstance = loadDefaultPrometheusReporter();
- }
- return reporterInstance;
- }
-
- /**
- * Load the PrometheusReporter class by default.
- */
- private MetricReporter loadDefaultPrometheusReporter() {
- MetricReporter reporterInstance = new PrometheusReporter();
- if (log.isDebugEnabled()) {
- log.debug("The class org.wso2.micro.integrator.obsrvability.handler.metrics.publisher.prometheus." +
- "reporter.PrometheusReporter was loaded successfully");
- }
- return reporterInstance;
- }
-
  @Override
  public boolean handleRequestInFlow(MessageContext synCtx) {
  synCtx.setProperty(RESTConstants.IS_PROMETHEUS_ENGAGED, null);

components/mediation/data-publishers/org.wso2.micro.integrator.observability/src/main/java/org/wso2/micro/integrator/observability/metric/handler/prometheus/reporter/PrometheusReporter.java

@@ -41,13 +41,16 @@ public class PrometheusReporter implements MetricReporter {
  private Counter TOTAL_REQUESTS_RECEIVED_PROXY_SERVICE;
  private Counter TOTAL_REQUESTS_RECEIVED_API;
  private Counter TOTAL_REQUESTS_RECEIVED_INBOUND_ENDPOINT;
+ private Counter TOTAL_REQUESTS_RECEIVED_DATA_SERVICE;
  private Counter ERROR_REQUESTS_RECEIVED_PROXY_SERVICE;
  private Counter ERROR_REQUESTS_RECEIVED_API;
  private Counter ERROR_REQUESTS_RECEIVED_INBOUND_ENDPOINT;
+ private Counter ERROR_REQUESTS_RECEIVED_DATA_SERVICE;
 
  private Histogram PROXY_LATENCY_HISTOGRAM;
  private Histogram API_LATENCY_HISTOGRAM;
  private Histogram INBOUND_ENDPOINT_LATENCY_HISTOGRAM;
+ private Histogram DATA_SERVICE_LATENCY_HISTOGRAM;
 
  private Gauge SERVER_UP;
  private Gauge SERVICE_UP;
@@ -58,6 +61,7 @@ public class PrometheusReporter implements MetricReporter {
  private double[] proxyLatencyBuckets;
  private double[] apiLatencyBuckets;
  private double[] inboundEndpointLatencyBuckets;
+ private double[] dataServiceLatencyBuckets;
 
  private Map<String, Object> metricMap = new HashMap();
 
@@ -70,6 +74,7 @@ public void initMetrics() {
  this.initializeProxyMetrics();
  this.initializeApiMetrics();
  this.initializeInboundEndpointMetrics();
+ this.initializeDataServiceMetrics();
  }
 
  @Override
@@ -79,6 +84,7 @@ public void createMetrics(String serviceType, String type, String metricName, St
  proxyLatencyBuckets = new double[]{0.19, 0.20, 0.25, 0.30, 0.35, 0.40, 0.50, 0.60, 1, 5};
  apiLatencyBuckets = new double[]{0.19, 0.20, 0.25, 0.30, 0.35, 0.40, 0.50, 0.60, 1, 5};
  inboundEndpointLatencyBuckets = new double[]{0.19, 0.20, 0.25, 0.30, 0.35, 0.40, 0.50, 0.60, 1, 5};
+ dataServiceLatencyBuckets = new double[]{0.19, 0.20, 0.25, 0.30, 0.35, 0.40, 0.50, 0.60, 1, 5};
 
  Map<String, Object> configs = ConfigParser.getParsedConfigs();
  createBuckets(configs);
@@ -132,6 +138,21 @@ public void createMetrics(String serviceType, String type, String metricName, St
  .register();
  metricMap.put(metricName, INBOUND_ENDPOINT_LATENCY_HISTOGRAM);
  }
+ } else if (serviceType.equalsIgnoreCase(SERVICE.DATA_SERVICE.name())) {
+ if (type.equals(MetricConstants.COUNTER)) {
+ TOTAL_REQUESTS_RECEIVED_DATA_SERVICE = Counter.build
+ (MetricConstants.DATA_SERVICE_REQUEST_COUNT_TOTAL, metricHelp).
+ labelNames(labels).register();
+ metricMap.put(metricName, TOTAL_REQUESTS_RECEIVED_DATA_SERVICE);
+ } else if (type.equals(MetricConstants.HISTOGRAM)) {
+ DATA_SERVICE_LATENCY_HISTOGRAM = Histogram.build()
+ .name(MetricConstants.DATA_SERVICE_LATENCY_SECONDS)
+ .help(metricHelp)
+ .labelNames(labels)
+ .buckets(dataServiceLatencyBuckets)
+ .register();
+ metricMap.put(metricName, DATA_SERVICE_LATENCY_HISTOGRAM);
+ }
  } else if (serviceType.equals(MetricConstants.SERVER)) {
  SERVER_UP = Gauge.build(MetricConstants.SERVER_UP, "Server status").
  labelNames(labels).register();
@@ -168,6 +189,12 @@ public void initErrorMetrics(String serviceType, String type, String metricName,
  register();
  metricMap.put(metricName, ERROR_REQUESTS_RECEIVED_INBOUND_ENDPOINT);
 
+ } else if (serviceType.equals(SERVICE.DATA_SERVICE.name())) {
+ ERROR_REQUESTS_RECEIVED_DATA_SERVICE = Counter.
+ build(MetricConstants.DATA_SERVICE_REQUEST_COUNT_ERROR_TOTAL, metricHelp).labelNames(labels).
+ register();
+ metricMap.put(metricName, ERROR_REQUESTS_RECEIVED_DATA_SERVICE);
+
  }
  }
 
@@ -242,7 +269,8 @@ public void serviceDown(String serviceName, String serviceType) {
  enum SERVICE {
  PROXY,
  API,
- INBOUND_ENDPOINT
+ INBOUND_ENDPOINT,
+ DATA_SERVICE
  }
 
  /**
@@ -259,6 +287,8 @@ private void createBuckets(Map<String, Object> configs) {
  MetricConstants.API_LATENCY_BUCKETS);
  Object inboundEndpointConfigBuckets = configs.get(MetricConstants.METRIC_HANDLER + "." +
  MetricConstants.INBOUND_ENDPOINT_LATENCY_BUCKETS);
+ Object dataServiceConfigBuckets = configs.get(MetricConstants.METRIC_HANDLER + "." +
+ MetricConstants.DATA_SERVICE_LATENCY_BUCKETS);
 
  if (null != proxyConfigBuckets) {
  List<Object> list = Arrays.asList(proxyConfigBuckets);
@@ -284,6 +314,14 @@ private void createBuckets(Map<String, Object> configs) {
  inboundEndpointLatencyBuckets[i] = (double) bucketList.get(i);
  }
  }
+ if (null != dataServiceConfigBuckets) {
+ List<Object> list = Arrays.asList(dataServiceConfigBuckets);
+ int size = list.size();
+ List<Object> bucketList = (ArrayList) list.get(0);
+ for (int i = 0; i < size; i++) {
+ dataServiceLatencyBuckets[i] = (double) bucketList.get(i);
+ }
+ }
  }
 
  /**
@@ -333,6 +371,22 @@ public void initializeInboundEndpointMetrics() {
  initializeInboundEndpointErrorMetrics();
  }
 
+ /**
+ * Create data services related metrics.
+ */
+ public void initializeDataServiceMetrics() {
+ String[] labels = {MetricConstants.SERVICE_NAME, MetricConstants.SERVICE_TYPE};
+
+ createMetrics("DATA_SERVICE", MetricConstants.COUNTER,
+ MetricConstants.DATA_SERVICE_REQUEST_COUNT_TOTAL,
+ "Total number of requests to a data service.", labels);
+ createMetrics("DATA_SERVICE", MetricConstants.HISTOGRAM,
+ MetricConstants.DATA_SERVICE_LATENCY_SECONDS,
+ "Latency of requests to a data service.", labels);
+
+ initializeDataServiceErrorMetrics();
+ }
+
  /**
  * Create the metrics related to failed proxy services.
  */
@@ -362,6 +416,16 @@ public void initializeInboundEndpointErrorMetrics() {
  new String[]{MetricConstants.SERVICE_NAME, MetricConstants.SERVICE_TYPE});
  }
 
+ /**
+ * Create the metrics related to failed dataservices.
+ */
+ public void initializeDataServiceErrorMetrics() {
+ initErrorMetrics("DATA_SERVICE", MetricConstants.COUNTER,
+ MetricConstants.DATA_SERVICE_REQUEST_COUNT_ERROR_TOTAL, "Total number of error" +
+ " requests when receiving the message by an inbound endpoint.",
+ new String[]{MetricConstants.SERVICE_NAME, MetricConstants.SERVICE_TYPE});
+ }
+
  /**
  * Create the metrics related to server startup.
  */

components/mediation/data-publishers/org.wso2.micro.integrator.observability/src/main/java/org/wso2/micro/integrator/observability/util/MetricConstants.java

@@ -24,6 +24,7 @@ public class MetricConstants {
 
  // Constants for Synapse artifacts
  public static final String INBOUND_ENDPOINT = "inbound-endpoint";
+ public static final String DATA_SERVICE = "data-service";
 
  public static final String HTTP_PORT = "http.nio.port";
  public static final String JAVA_VERSION = "java.vm.specification.version";
@@ -52,6 +53,13 @@ public class MetricConstants {
  public static final String INBOUND_ENDPOINT_LATENCY_SECONDS =
  "wso2_integration_inbound_endpoint_latency_seconds";
 
+ public static final String DATA_SERVICE_REQUEST_COUNT_TOTAL =
+ "wso2_integration_data_service_request_count_total";
+ public static final String DATA_SERVICE_REQUEST_COUNT_ERROR_TOTAL =
+ "wso2_integration_data_service_request_count_error_total";
+ public static final String DATA_SERVICE_LATENCY_SECONDS =
+ "wso2_integration_data_service_latency_seconds";
+
  public static final String SERVER_UP = "wso2_integration_server_up";
  public static final String SERVICE_UP = "wso2_integration_service_up";
  public static final String SERVER_VERSION = "wso2_integration_server_version";
@@ -60,10 +68,12 @@ public class MetricConstants {
  public static final String PROXY_LATENCY_BUCKETS = "proxy_latency_buckets";
  public static final String API_LATENCY_BUCKETS = "api_latency_buckets";
  public static final String INBOUND_ENDPOINT_LATENCY_BUCKETS = "inbound_endpoint_latency_buckets";
+ public static final String DATA_SERVICE_LATENCY_BUCKETS = "data_service_latency_buckets";
 
  public static final String PROXY_LATENCY_TIMER = "PROXY_LATENCY_TIMER";
  public static final String API_LATENCY_TIMER = "API_LATENCY_TIMER";
  public static final String INBOUND_ENDPOINT_LATENCY_TIMER = "INBOUND_ENDPOINT_LATENCY_TIMER";
+ public static final String DATA_SERVICE_LATENCY_TIMER = "DATA_SERVICE_LATENCY_TIMER";
 
  public static final String SERVER = "Server";
  public static final String SERVICE = "Service";
@@ -84,4 +94,8 @@ public class MetricConstants {
  public static final String VERSION_LABEL = "version";
  public static final String UPDATE_LEVEL_LABEL = "update_level";
 
+ public static final String MESSAGE_DIRECTION_IN = "in";
+
+ public static final String MESSAGE_DIRECTION_OUT = "out";
+
 }