Fix Prometheus issues

Author: liurui-1

README.md

@@ -34,12 +34,13 @@ We will provide more receivers in the future and welcome contributions. Right no
 
 | Receiver name | Type | Status | JDK required |
 |---------------|------|--------|--------------|
-| [Dameng Database](ojr-dameng-db/README.md)       | metrics | beta  | Java 8+ |
-| [Oceanbase Database](ojr-oceanbase-db/README.md) | metrics | alpha | Java 8+ |
-| [Informix Database](ojr-informix-db/README.md)   | metrics | alpha | Java 8+ |
-| [Linux Host](ojr-linux-host/README.md)           | metrics | alpha | Java 8+ |
+| [Dameng Database](ojr-dameng-db/README.md)       | metrics | beta  | Java 8+  |
+| [Oceanbase Database](ojr-oceanbase-db/README.md) | metrics | alpha | Java 8+  |
+| [Informix Database](ojr-informix-db/README.md)   | metrics | alpha | Java 8+  |
+| [Linux Host](ojr-linux-host/README.md)           | metrics | alpha | Java 8+  |
 | [SNMP Host](ojr-snmp-host/README.md)             | metrics | alpha | Java 11+ |
-| [IBM MQ Appliance](ojr-mq-appliance/README.md)   | metrics | beta  | Java 8+ |
+| [IBM MQ Appliance](ojr-mq-appliance/README.md)   | metrics | beta  | Java 8+  |
+| [IBM MQ](ojr-ibmmq/README.md)                    | metrics | beta  | Java 8+  |
 
 
 ## Common Parameters for Receivers

ojr-dameng-db/libs/ojr-core-0.1.0.jar

@@ -17,3 +17,21 @@ instances:
     #otel.backend.url: http://localhost:4318
     otel.transport: prometheus
     prometheus.port: 16543
+
+  - queueManager: SATURN
+    isLocal: false
+    #user: liurui
+    #password: xxxxxxxx
+    #port: 1414
+    queuesMonitored: Q*
+    #customEventQueues:
+    #keystore:
+    #keystorePassword:
+    #cipherSuite: TLS_RSA_WITH_AES_256_CBC_SHA256
+
+    ## Data collector properties:
+    poll.interval: 25
+    callback.interval: 30
+    #otel.backend.url: http://localhost:4318
+    otel.transport: prometheus
+    prometheus.port: 16543

ojr-dameng-db/libs/ojr-rdb-0.1.0.jar

@@ -87,7 +87,7 @@ public void collectData() {
         logger.info("Start to collect metrics");
         try {
             mqClient.connect();
-            for(DataQuerier querier : queriers){
+            for (DataQuerier querier : queriers) {
                 querier.collectData();
             }
         } catch (Exception e) {

ojr-ibmmq/config/config.yaml

@@ -26,7 +26,7 @@ instances:
     db.username: root@sys
     db.password: MTIzNDU2
     db.connection.url: jdbc:mysql://localhost:2881/oceanbase?autoReconnect=true&useSSL=false
-    db.name: oceanbase
+    db.name: oceanbase1
     db.entity.type: tenant
     db.tenant.name: test
 

ojr-ibmmq/libs/ojr-core-0.1.0.jar

@@ -28,6 +28,8 @@ dependencies {
     api("io.opentelemetry.semconv:opentelemetry-semconv-incubating:1.28.0-alpha")
     api("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.18.2")
 
+    implementation('io.prometheus:prometheus-metrics-exporter-httpserver:1.3.4')
+
     testImplementation platform('org.junit:junit-bom:5.11.3')
     testImplementation 'org.junit.jupiter:junit-jupiter'
     testImplementation 'org.mockito:mockito-core:4.11.0'

ojr-ibmmq/src/main/java/com/ojr/ibmmq/MQDc.java

@@ -1,5 +1,6 @@
 package com.ojr.core;
 
+import com.ojr.core.metric.OjrPrometheusHttpServer;
 import com.ojr.core.metric.RawMetric;
 import com.ojr.core.resources.ContainerResource;
 import io.opentelemetry.api.OpenTelemetry;
@@ -17,17 +18,16 @@
 import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder;
 import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
 import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder;
-import io.opentelemetry.exporter.prometheus.PrometheusHttpServer;
-import io.opentelemetry.exporter.prometheus.PrometheusHttpServerBuilder;
+import io.opentelemetry.exporter.prometheus.PrometheusMetricReader;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.OpenTelemetrySdkBuilder;
+import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.logs.SdkLoggerProvider;
 import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
 import io.opentelemetry.sdk.logs.export.LogRecordExporter;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
-import io.opentelemetry.sdk.metrics.export.MetricReader;
 import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
@@ -69,7 +69,7 @@ public abstract class AbstractDc<Cfg extends BasicDcConfig> implements IDc<Cfg>
     private String transport = DcUtil.DEFAULT_OTEL_TRANSPORT;
 
     private int prometheusPort = DcUtil.DEFAULT_PROMETHEUS_PORT;
-    private String prometheusHost = null;
+    private String prometheusHost = DcUtil.DEFAULT_PROMETHEUS_HOST;
 
     private String serviceName = DcUtil.DEFAULT_OTEL_SERVICE_NAME;
     private String serviceInstanceId = null;
@@ -420,14 +420,10 @@ public LogRecordExporter createOtlpHttpLogRecordExporter(Map<String, String> hea
         return builder.build();
     }
 
-    private static MetricReader prometheusMetricReader = null;
+    //private static MetricReader prometheusMetricReader = null;
 
-    private static Predicate<String> dftResAttrsFilterForPrometheus = (String key) -> {
-        if (DcUtil.OJR_PLUGIN.equals(key) || HostIncubatingAttributes.HOST_NAME.getKey().equals(key)) {
-            return true;
-        }
-        return false;
-    };
+    private static final Predicate<String> dftResAttrsFilterForPrometheus =
+            (String key) -> DcUtil.OJR_PLUGIN.equals(key) || HostIncubatingAttributes.HOST_NAME.getKey().equals(key);
 
     /**
      * Returns the filter for resource attributes used when exporting metrics to Prometheus.
@@ -440,15 +436,8 @@ public Predicate<String> updateResAttrsFilterForPrometheus(Predicate<String> res
         return resAttrsFilter;
     }
 
-    /**
-     * Creates and configures a Prometheus metric reader.
-     * <p>
-     * This method sets up a Prometheus HTTP server with a specified port.
-     * Optionally, if a host is provided and not empty, it sets the host as well.
-     *
-     * @return A configured MetricReader instance for Prometheus metrics.
-     */
-    public synchronized MetricReader createPrometheusMetricReader() {
+    /*
+    private synchronized MetricReader createPrometheusMetricReader0() {
         if (prometheusMetricReader != null) {
             return prometheusMetricReader;
         }
@@ -461,6 +450,44 @@ public synchronized MetricReader createPrometheusMetricReader() {
 
         return prometheusMetricReader = builder.build();
     }
+    */
+
+    private static OjrPrometheusHttpServer prometheusHttpServer = null;
+
+
+    public synchronized OjrPrometheusHttpServer createPrometheusHttpServerIfNotExist() {
+        if (prometheusHttpServer != null) {
+            return prometheusHttpServer; // Return early if the server is already created
+        }
+        prometheusHttpServer = new OjrPrometheusHttpServer(prometheusHost, prometheusPort, null, MemoryMode.REUSABLE_DATA);
+        return prometheusHttpServer;
+    }
+
+    /**
+     * Creates and configures a Prometheus metric reader.
+     * <p>
+     * This method sets up a Prometheus HTTP server with a specified port.
+     * Optionally, if a host is provided and not empty, it sets the host as well.
+     *
+     * @return A configured PrometheusMetricReader instance for Prometheus metrics.
+     */
+    public PrometheusMetricReader createPrometheusMetricReader() {
+        return new PrometheusMetricReader(false, updateResAttrsFilterForPrometheus(dftResAttrsFilterForPrometheus));
+    }
+
+
+    /**
+     * Initializes Prometheus for the OpenTelemetry Meter Provider.
+     *
+     * @param builder The SdkMeterProviderBuilder instance to configure Prometheus.
+     * @return The updated SdkMeterProviderBuilder instance.
+     */
+    public SdkMeterProviderBuilder initPrometheus(SdkMeterProviderBuilder builder) {
+        PrometheusMetricReader reader = createPrometheusMetricReader();
+        builder.registerMetricReader(reader);
+        createPrometheusHttpServerIfNotExist().registerReader(reader).start();
+        return builder;
+    }
 
     /**
      * Creates a default SDK meter provider with the specified configuration.
@@ -476,17 +503,17 @@ public SdkMeterProvider getDefaultSdkMeterProvider(Resource resource) {
         if (transport.contains(DcUtil.GRPC)) {
             SdkMeterProviderBuilder builder = SdkMeterProvider.builder().setResource(resource).registerMetricReader(PeriodicMetricReader.builder(createOtlpGrpcMetricExporter(headers, cert)).setInterval(Duration.ofSeconds(callbackInterval)).build());
             if (transport.contains(DcUtil.PROMETHEUS)) {
-                builder.registerMetricReader(createPrometheusMetricReader());
+                initPrometheus(builder);
             }
             return builder.build();
         } else if (transport.contains(DcUtil.HTTP)) {
             SdkMeterProviderBuilder builder = SdkMeterProvider.builder().setResource(resource).registerMetricReader(PeriodicMetricReader.builder(createOtlpHttpMetricExporter(headers, cert)).setInterval(Duration.ofSeconds(callbackInterval)).build());
             if (transport.contains(DcUtil.PROMETHEUS)) {
-                builder.registerMetricReader(createPrometheusMetricReader());
+                initPrometheus(builder);
             }
             return builder.build();
         } else if (transport.contains(DcUtil.PROMETHEUS)) {
-            return SdkMeterProvider.builder().setResource(resource).registerMetricReader(createPrometheusMetricReader()).build();
+            return initPrometheus(SdkMeterProvider.builder().setResource(resource)).build();
         } else {
             return SdkMeterProvider.builder().build();
         }

ojr-informix-db/libs/ojr-core-0.1.0.jar

@@ -44,6 +44,7 @@ public class DcUtil {
     public static final String PROMETHEUS_PORT = "prometheus.port";
     public static final int DEFAULT_PROMETHEUS_PORT = 16543; // Default Prometheus port
     public static final String PROMETHEUS_HOST = "prometheus.host";
+    public static final String DEFAULT_PROMETHEUS_HOST = "0.0.0.0";
 
     public final static String OTEL_SERVICE_NAME = "otel.service.name"; // Service name for OpenTelemetry
     public final static String DEFAULT_OTEL_SERVICE_NAME = "OJR"; // Default service name

ojr-informix-db/libs/ojr-rdb-0.1.0.jar

@@ -0,0 +1,177 @@
+package com.ojr.core.metric;
+
+import io.opentelemetry.exporter.prometheus.PrometheusMetricReader;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.common.export.MemoryMode;
+import io.opentelemetry.sdk.internal.DaemonThreadFactory;
+import io.prometheus.metrics.exporter.httpserver.HTTPServer;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Prometheus HTTP server implementation.
+ * This class manages the Prometheus HTTP server, including starting, stopping, and shutting down.
+ */
+public class OjrPrometheusHttpServer {
+    // HTTP server instance to handle incoming requests
+    private HTTPServer httpServer;
+
+    private final PrometheusMergedReader mReader = new PrometheusMergedReader();
+
+    // Prometheus registry to manage and expose metrics
+    private final PrometheusRegistry prometheusRegistry = new PrometheusRegistry();
+
+    // Host address where the HTTP server will listen
+    private final String host;
+
+    // Port number where the HTTP server will listen
+    private final int port;
+
+    // Memory mode configuration (e.g., REUSABLE_DATA)
+    private final MemoryMode memoryMode;
+
+    // Executor service for handling HTTP server tasks
+    private final ExecutorService executor;
+
+    /**
+     * Constructor to initialize the Prometheus HTTP server.
+     *
+     * @param host       Host address for the HTTP server
+     * @param port       Port number for the HTTP server
+     * @param executor   Optional executor service for handling tasks
+     * @param memoryMode Memory mode configuration
+     */
+    public OjrPrometheusHttpServer(
+            String host,
+            int port,
+            @Nullable ExecutorService executor,
+            MemoryMode memoryMode) {
+
+        this.host = host;
+        this.port = port;
+        this.memoryMode = memoryMode;
+        this.executor = executor;
+    }
+
+    /**
+     * Registers a PrometheusMetricReader
+     *
+     * @param reader PrometheusMetricReader instance to register
+     * @return The current OjrPrometheusHttpServer instance
+     */
+    public OjrPrometheusHttpServer registerReader(PrometheusMetricReader reader) {
+        mReader.registerReader(reader);
+        return this;
+    }
+
+    private static boolean isStarted = false;
+
+    /**
+     * Starts the Prometheus HTTP server.
+     */
+    public synchronized void start() {
+        if (isStarted)
+            return;
+
+        prometheusRegistry.register(mReader);
+
+        ExecutorService executor1 = executor;
+
+        // If memory mode is REUSABLE_DATA, create a dedicated thread pool
+        if (memoryMode == MemoryMode.REUSABLE_DATA) {
+            executor1 =
+                    new ThreadPoolExecutor(
+                            1,
+                            1,
+                            0L,
+                            TimeUnit.MILLISECONDS,
+                            new LinkedBlockingQueue<>(),
+                            new DaemonThreadFactory("prometheus-http-server"));
+        }
+        try {
+            // Build and start the HTTP server with the given configuration
+            httpServer =
+                    HTTPServer.builder()
+                            .hostname(host)
+                            .port(port)
+                            .executorService(executor1)
+                            .registry(prometheusRegistry)
+                            .buildAndStart();
+        } catch (IOException e) {
+            throw new UncheckedIOException("Could not create Prometheus HTTP server", e);
+        }
+
+        isStarted = true;
+    }
+
+    public PrometheusMergedReader getmReader() {
+        return mReader;
+    }
+
+    /**
+     * Returns the current memory mode configuration.
+     *
+     * @return MemoryMode enum value
+     */
+    public MemoryMode getMemoryMode() {
+        return memoryMode;
+    }
+
+    /**
+     * Shuts down the Prometheus HTTP server gracefully.
+     *
+     * @return CompletableResultCode indicating the shutdown result
+     */
+    public CompletableResultCode shutdown() {
+        CompletableResultCode rc = new CompletableResultCode();
+        Runnable shutdownFunction =
+                () -> {
+                    try {
+                        // Unregister each reader and stop the HTTP server
+                        prometheusRegistry.unregister(mReader);
+                        httpServer.stop();
+                    } catch (Throwable t) {
+                        rc.fail();
+                    }
+                };
+        Thread shutdownThread = new Thread(shutdownFunction, "Shutdown-OjrPrometheusHttpServer");
+        shutdownThread.setDaemon(true);
+        shutdownThread.start();
+        isStarted = false;
+        return rc;
+    }
+
+    /**
+     * Closes the Prometheus HTTP server by initiating a shutdown and waiting for completion.
+     */
+    public void close() {
+        shutdown().join(10, TimeUnit.SECONDS);
+    }
+
+    /**
+     * Returns a string representation of the server's address.
+     *
+     * @return String representation of the server's address
+     */
+    @Override
+    public String toString() {
+        return "OjrPrometheusHttpServer{address=" + getAddress() + "}";
+    }
+
+    /**
+     * Gets the InetSocketAddress of the HTTP server.
+     *
+     * @return InetSocketAddress object
+     */
+    InetSocketAddress getAddress() {
+        return new InetSocketAddress(host, httpServer.getPort());
+    }
+}