Merge branch 'master' into INLONG-11342

inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/metric/AbstractMetric.java

@@ -20,4 +20,5 @@
 public interface AbstractMetric {
 
     public void report();
+    public void stop();
 }

inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/metric/MetricsManager.java

@@ -94,5 +94,6 @@ public void addSendFailed(long count) {
     }
     public void shutdown() {
         timer.shutdown();
+        metric.stop();
     }
 }

inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/metric/prometheus/ProxyPrometheusMetric.java

@@ -80,4 +80,9 @@ private MetricFamilySamples.Sample createSample(MetricDimension key, double valu
     public void report() {
         LOGGER.info("Report proxy prometheus metric: {} ", metricItem.toString());
     }
+
+    @Override
+    public void stop() {
+        server.close();
+    }
 }

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/Application.java → inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/Application.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.audit;
+package org.apache.inlong.audit.store;
 
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.SpringBootApplication;

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/config/ConfigConstants.java

@@ -0,0 +1,31 @@
+/*
+ * 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.inlong.audit.store.config;
+
+/**
+ * Config constants
+ */
+public class ConfigConstants {
+
+    public static final String AUDIT_STORE_SERVER_NAME = "audit-store";
+    public static final String KEY_PROMETHEUS_PORT = "audit.store.prometheus.port";
+    public static final int DEFAULT_PROMETHEUS_PORT = 10083;
+    public static final String KEY_STORE_METRIC_CLASSNAME = "audit.store.metric.classname";
+    public static final String DEFAULT_STORE_METRIC_CLASSNAME =
+            "org.apache.inlong.audit.store.metric.prometheus.StorePrometheusMetric";
+}

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/config/JdbcConfig.java → inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/config/JdbcConfig.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.audit.config;
+package org.apache.inlong.audit.store.config;
 
 import lombok.Data;
 import org.springframework.beans.factory.annotation.Value;

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/config/MessageQueueConfig.java → inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/config/MessageQueueConfig.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.audit.config;
+package org.apache.inlong.audit.store.config;
 
 import lombok.Getter;
 import lombok.Setter;

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/config/StoreConfig.java → inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/config/StoreConfig.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.audit.config;
+package org.apache.inlong.audit.store.config;
 
 import lombok.Getter;
 import lombok.Setter;

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/db/entities/JdbcDataPo.java → inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/entities/JdbcDataPo.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.audit.db.entities;
+package org.apache.inlong.audit.store.entities;
 
 import lombok.Data;
 import org.apache.pulsar.client.api.Consumer;

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/metric/MetricDimension.java

@@ -0,0 +1,37 @@
+/*
+ * 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.inlong.audit.store.metric;
+
+public enum MetricDimension {
+
+    RECEIVE_COUNT_SUCCESS("receiveCountSuccess"),
+    RECEIVE_FAILED("receiveFailed"),
+    SEND_COUNT_SUCCESS("sendCountSuccess"),
+    SEND_COUNT_FAILED("sendCountFailed"),
+    SEND_DURATION("sendDuration");
+
+    private final String key;
+
+    MetricDimension(String key) {
+        this.key = key;
+    }
+
+    public String getKey() {
+        return key;
+    }
+}

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/metric/MetricItem.java

@@ -0,0 +1,40 @@
+/*
+ * 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.inlong.audit.store.metric;
+
+import lombok.Data;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+@Data
+public class MetricItem {
+
+    public static final String K_DIMENSION_KEY = "dimensionName";
+    private AtomicLong receiveCountSuccess = new AtomicLong(0);
+    private AtomicLong receiveFailed = new AtomicLong(0);
+    private AtomicLong sendCountSuccess = new AtomicLong(0);
+    private AtomicLong sendCountFailed = new AtomicLong(0);
+    private AtomicLong sendDuration = new AtomicLong(0);
+    public void resetAllMetrics() {
+        receiveCountSuccess.set(0);
+        receiveFailed.set(0);
+        sendCountSuccess.set(0);
+        sendCountFailed.set(0);
+        sendDuration.set(0);
+    }
+}

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/metric/MetricsManager.java

@@ -0,0 +1,95 @@
+/*
+ * 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.inlong.audit.store.metric;
+
+import org.apache.inlong.audit.file.ConfigManager;
+import org.apache.inlong.audit.metric.AbstractMetric;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.inlong.audit.store.config.ConfigConstants.DEFAULT_STORE_METRIC_CLASSNAME;
+import static org.apache.inlong.audit.store.config.ConfigConstants.KEY_STORE_METRIC_CLASSNAME;
+
+public class MetricsManager {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MetricsManager.class);
+
+    private static class Holder {
+
+        private static final MetricsManager INSTANCE = new MetricsManager();
+    }
+
+    private AbstractMetric metric;
+
+    public void init() {
+        try {
+            String metricClassName =
+                    ConfigManager.getInstance().getValue(KEY_STORE_METRIC_CLASSNAME, DEFAULT_STORE_METRIC_CLASSNAME);
+            LOGGER.info("Metric class name: {}", metricClassName);
+            Constructor<?> constructor = Class.forName(metricClassName)
+                    .getDeclaredConstructor(MetricItem.class);
+            constructor.setAccessible(true);
+            metric = (AbstractMetric) constructor.newInstance(metricItem);
+
+            timer.scheduleWithFixedDelay(() -> {
+                metric.report();
+                metricItem.resetAllMetrics();
+            }, 0, 1, TimeUnit.MINUTES);
+        } catch (ClassNotFoundException | NoSuchMethodException | InstantiationException | IllegalAccessException
+                | InvocationTargetException exception) {
+            LOGGER.error("Init metrics manager has exception: ", exception);
+        }
+    }
+
+    public static MetricsManager getInstance() {
+        return Holder.INSTANCE;
+    }
+
+    private final MetricItem metricItem = new MetricItem();
+    protected final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();
+
+    public void addReceiveSuccess(long count) {
+        metricItem.getReceiveCountSuccess().addAndGet(count);
+    }
+
+    public void addReceiveFailed(long pack) {
+        metricItem.getReceiveFailed().addAndGet(pack);
+    }
+
+    public void addSendSuccess(long count, long duration) {
+        metricItem.getSendCountSuccess().addAndGet(count);
+        metricItem.getSendDuration().addAndGet(duration);
+    }
+
+    public void addSendFailed(long count, long duration) {
+        metricItem.getSendCountFailed().addAndGet(count);
+        metricItem.getSendDuration().addAndGet(duration);
+    }
+
+    public void shutdown() {
+        timer.shutdown();
+        metric.stop();
+    }
+}

inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/metric/prometheus/StorePrometheusMetric.java

@@ -0,0 +1,91 @@
+/*
+ * 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.inlong.audit.store.metric.prometheus;
+
+import org.apache.inlong.audit.file.ConfigManager;
+import org.apache.inlong.audit.metric.AbstractMetric;
+import org.apache.inlong.audit.store.metric.MetricDimension;
+import org.apache.inlong.audit.store.metric.MetricItem;
+
+import io.prometheus.client.Collector;
+import io.prometheus.client.exporter.HTTPServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.inlong.audit.store.config.ConfigConstants.AUDIT_STORE_SERVER_NAME;
+import static org.apache.inlong.audit.store.config.ConfigConstants.DEFAULT_PROMETHEUS_PORT;
+import static org.apache.inlong.audit.store.config.ConfigConstants.KEY_PROMETHEUS_PORT;
+
+/**
+ * PrometheusMetric
+ */
+public class StorePrometheusMetric extends Collector implements AbstractMetric {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(StorePrometheusMetric.class);
+    private static final String HELP_DESCRIPTION = "help";
+
+    private final MetricItem metricItem;
+    private HTTPServer server;
+
+    public StorePrometheusMetric(MetricItem metricItem) {
+        this.metricItem = metricItem;
+        try {
+            server = new HTTPServer(ConfigManager.getInstance().getValue(KEY_PROMETHEUS_PORT, DEFAULT_PROMETHEUS_PORT));
+            this.register();
+        } catch (IOException e) {
+            LOGGER.error("Construct store prometheus metric has IOException", e);
+        }
+    }
+
+    @Override
+    public List<MetricFamilySamples> collect() {
+        List<MetricFamilySamples.Sample> samples = Arrays.asList(
+                createSample(MetricDimension.RECEIVE_COUNT_SUCCESS, metricItem.getReceiveCountSuccess().doubleValue()),
+                createSample(MetricDimension.RECEIVE_FAILED, metricItem.getReceiveFailed().doubleValue()),
+                createSample(MetricDimension.SEND_COUNT_SUCCESS, metricItem.getSendCountSuccess().doubleValue()),
+                createSample(MetricDimension.SEND_COUNT_FAILED, metricItem.getSendCountFailed().doubleValue()),
+                createSample(MetricDimension.SEND_DURATION, metricItem.getSendDuration().doubleValue()));
+
+        MetricFamilySamples metricFamilySamples =
+                new MetricFamilySamples(AUDIT_STORE_SERVER_NAME, Type.GAUGE, HELP_DESCRIPTION, samples);
+
+        return Collections.singletonList(metricFamilySamples);
+    }
+
+    private MetricFamilySamples.Sample createSample(MetricDimension key, double value) {
+        return new MetricFamilySamples.Sample(AUDIT_STORE_SERVER_NAME,
+                Collections.singletonList(MetricItem.K_DIMENSION_KEY),
+                Collections.singletonList(key.getKey()), value);
+    }
+
+    @Override
+    public void report() {
+        LOGGER.info("Report store prometheus metric: {} ", metricItem.toString());
+    }
+
+    @Override
+    public void stop() {
+        server.close();
+    }
+
+}