[INLONG-1872]The Broker of TubeMQ supports metrics with JMX

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/AbsMetricItem.java

@@ -0,0 +1,77 @@
+/**
+ * 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.tubemq.corebase.metric;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.inlong.tubemq.corebase.utils.Tuple2;
+
+public abstract class AbsMetricItem {
+
+    protected final MetricType metricType;
+    protected final MetricValueType valueType;
+    protected final String name;
+    protected final AtomicLong value = new AtomicLong(0);
+
+    public AbsMetricItem(MetricType metricType, String name) {
+        this(metricType, MetricValueType.NORMAL, name, 0);
+    }
+
+    public AbsMetricItem(MetricType metricType, MetricValueType valueType,
+                         String name, long initialValue) {
+        this.metricType = metricType;
+        this.valueType = valueType;
+        this.name = name;
+        this.value.set(initialValue);
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public long getValue() {
+        return value.get();
+    }
+
+    public boolean isCounterMetric() {
+        return metricType == MetricType.COUNTER;
+    }
+
+    public MetricType getMetricType() {
+        return metricType;
+    }
+
+    public MetricValueType getMetricValueType() {
+        return valueType;
+    }
+
+    public Tuple2<String, Long> getNameValue() {
+        return new Tuple2<>(name, value.get());
+    }
+
+    public long incrementAndGet() {
+        return value.incrementAndGet();
+    }
+
+    public long decrementAndGet() {
+        return value.decrementAndGet();
+    }
+
+    public abstract long getAndSet();
+
+    public abstract boolean update(long newValue);
+}

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/CountMetricItem.java

@@ -0,0 +1,36 @@
+/**
+ * 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.tubemq.corebase.metric;
+
+public class CountMetricItem extends AbsMetricItem {
+
+    public CountMetricItem(String name) {
+        super(MetricType.COUNTER, name);
+    }
+
+    @Override
+    public long getAndSet() {
+        return value.getAndSet(0);
+    }
+
+    @Override
+    public boolean update(long newValue) {
+        value.set(newValue);
+        return true;
+    }
+}

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/GaugeMaxMetricItem.java

@@ -0,0 +1,39 @@
+/**
+ * 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.tubemq.corebase.metric;
+
+public class GaugeMaxMetricItem extends AbsMetricItem {
+
+    public GaugeMaxMetricItem(String name) {
+        super(MetricType.GAUGE, MetricValueType.MAX, name, 0);
+    }
+
+    @Override
+    public long getAndSet() {
+        return value.getAndSet(0);
+    }
+
+    @Override
+    public boolean update(long newValue) {
+        long curValue = value.get();
+        if (newValue > curValue) {
+            return value.compareAndSet(curValue, newValue);
+        }
+        return false;
+    }
+}

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/GaugeMinMetricItem.java

@@ -0,0 +1,39 @@
+/**
+ * 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.tubemq.corebase.metric;
+
+public class GaugeMinMetricItem extends AbsMetricItem {
+
+    public GaugeMinMetricItem(String name) {
+        super(MetricType.GAUGE, MetricValueType.MIN, name, Long.MAX_VALUE);
+    }
+
+    @Override
+    public long getAndSet() {
+        return value.getAndSet(Long.MAX_VALUE);
+    }
+
+    @Override
+    public boolean update(long newValue) {
+        long curValue = value.get();
+        if (newValue < curValue) {
+            return value.compareAndSet(curValue, newValue);
+        }
+        return false;
+    }
+}

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/GaugeNormMetricItem.java

@@ -0,0 +1,36 @@
+/**
+ * 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.tubemq.corebase.metric;
+
+public class GaugeNormMetricItem extends AbsMetricItem {
+
+    public GaugeNormMetricItem(String name) {
+        super(MetricType.GAUGE, MetricValueType.MIN, name, 0);
+    }
+
+    @Override
+    public long getAndSet() {
+        return value.get();
+    }
+
+    @Override
+    public boolean update(long newValue) {
+        value.set(newValue);
+        return true;
+    }
+}

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/MetricType.java

@@ -0,0 +1,49 @@
+/**
+ * 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.tubemq.corebase.metric;
+
+public enum MetricType {
+    UNKNOWN(0, "Unknown"),
+    COUNTER(1, "Counter"),
+    GAUGE(2, "Gauge");
+
+    MetricType(int id, String name) {
+        this.id = id;
+        this.name = name;
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public static MetricType valueOf(int value) {
+        for (MetricType metricType : MetricType.values()) {
+            if (metricType.getId() == value) {
+                return metricType;
+            }
+        }
+        return UNKNOWN;
+    }
+
+    private final int id;
+    private final String name;
+}

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/MetricValue.java

@@ -0,0 +1,45 @@
+/*
+ * 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.tubemq.corebase.metric;
+
+import java.beans.ConstructorProperties;
+
+public class MetricValue {
+    private final String type;
+    private final String name;
+    private final long value;
+
+    @ConstructorProperties({"type", "name", "value"})
+    public MetricValue(String type, String name, long value) {
+        this.name = name;
+        this.type = type;
+        this.value = value;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public long getValue() {
+        return value;
+    }
+}

inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/metric/MetricValueType.java

@@ -0,0 +1,49 @@
+/**
+ * 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.tubemq.corebase.metric;
+
+public enum MetricValueType {
+    NORMAL(0, "Normal"),
+    MIN(1, "Min"),
+    MAX(2, "Max");
+
+    MetricValueType(int id, String name) {
+        this.id = id;
+        this.name = name;
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public static MetricValueType valueOf(int value) {
+        for (MetricValueType valueType : MetricValueType.values()) {
+            if (valueType.getId() == value) {
+                return valueType;
+            }
+        }
+        return NORMAL;
+    }
+
+    private final int id;
+    private final String name;
+}