Meter system enhancement (#6398)

CHANGES.md

@@ -20,6 +20,7 @@ Release Notes.
 * Update the `timestamp` field type for `LogQuery`.
 * Support Zabbix protocol to receive agent metrics.
 * Update the Apdex metric combine calculator.
+* Enhance `MeterSystem` to allow creating metrics with same `metricName` / `function` / `scope`.
 
 #### UI
 * Update selector scroller to show in all pages.

oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/Analyzer.java

@@ -18,7 +18,6 @@
 
 package org.apache.skywalking.oap.meter.analyzer;
 
-import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import io.vavr.Tuple;
 import io.vavr.Tuple2;
@@ -208,12 +207,12 @@ public class Analyzer {
                 metricType = MetricType.labeled;
             }
         }
-        Preconditions.checkState(createMetric(ctx.getScopeType(), metricType.literal, ctx.getDownsampling()));
+        createMetric(ctx.getScopeType(), metricType.literal, ctx.getDownsampling());
     }
 
-    private boolean createMetric(final ScopeType scopeType, final String dataType, final DownsamplingType downsamplingType) {
+    private void createMetric(final ScopeType scopeType, final String dataType, final DownsamplingType downsamplingType) {
         String functionName = String.format(FUNCTION_NAME_TEMP, downsamplingType.toString().toLowerCase(), Strings.capitalize(dataType));
-        return meterSystem.create(metricName, functionName, scopeType);
+        meterSystem.create(metricName, functionName, scopeType);
     }
 
     private void send(final AcceptableValue<?> v, final long time) {

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/MeterSystem.java

@@ -99,11 +99,10 @@ public class MeterSystem implements Service {
      *
      * @param metricsName  The name used as the storage eneity and in the query stage.
      * @param functionName The function provided through {@link MeterFunction}.
-     * @return true if created, false if it exists.
      * @throws IllegalArgumentException if the parameter can't match the expectation.
      * @throws UnexpectedException      if binary code manipulation fails or stream core failure.
      */
-    public synchronized <T> boolean create(String metricsName,
+    public synchronized <T> void create(String metricsName,
         String functionName,
         ScopeType type) throws IllegalArgumentException {
         final Class<? extends MeterFunction> meterFunction = functionRegister.get(functionName);
@@ -123,7 +122,7 @@ public class MeterSystem implements Service {
             }
         }
         try {
-            return create(metricsName, functionName, type, Class.forName(Objects.requireNonNull(acceptance).getTypeName()));
+            create(metricsName, functionName, type, Class.forName(Objects.requireNonNull(acceptance).getTypeName()));
         } catch (ClassNotFoundException e) {
             throw new IllegalArgumentException(e);
         }
@@ -135,11 +134,10 @@ public class MeterSystem implements Service {
      *
      * @param metricsName  The name used as the storage eneity and in the query stage.
      * @param functionName The function provided through {@link MeterFunction}.
-     * @return true if created, false if it exists.
      * @throws IllegalArgumentException if the parameter can't match the expectation.
      * @throws UnexpectedException      if binary code manipulation fails or stream core failure.
      */
-    public synchronized <T> boolean create(String metricsName,
+    public synchronized <T> void create(String metricsName,
                                            String functionName,
                                            ScopeType type,
                                            Class<T> dataType) throws IllegalArgumentException {
@@ -187,6 +185,20 @@ public class MeterSystem implements Service {
             throw new IllegalArgumentException("Function " + functionName + " can't be found by javaassist.");
         }
         final String className = formatName(metricsName);
+
+        /**
+         * Check whether the metrics class is already defined or not
+         */
+        try {
+            CtClass existingMetric = classPool.get(METER_CLASS_PACKAGE + className);
+            if (existingMetric.getSuperclass() != parentClass || type != meterPrototypes.get(metricsName).getScopeType()) {
+                throw new IllegalArgumentException(metricsName + " has been defined, but calculate function or/are scope type is/are different.");
+            }
+            log.info("Metric {} is already defined, so skip the metric creation.", metricsName);
+            return ;
+        } catch (NotFoundException e) {
+        }
+
         CtClass metricsClass = classPool.makeClass(METER_CLASS_PACKAGE + className, parentClass);
 
         /**
@@ -234,7 +246,6 @@ public class MeterSystem implements Service {
             log.error("Can't compile/load/init " + className + ".", e);
             throw new UnexpectedException(e.getMessage(), e);
         }
-        return true;
     }
 
     /**

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/meter/MeterSystemTest.java

+/*
+ * 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.skywalking.oap.server.core.analysis.meter;
+
+import java.util.Map;
+import org.apache.skywalking.oap.server.core.analysis.StreamDefinition;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.storage.StorageException;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.powermock.reflect.Whitebox;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.spy;
+
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class MeterSystemTest {
+
+    @Mock
+    private ModuleManager moduleManager;
+    private MeterSystem meterSystem;
+
+    @Before
+    public void setup() throws StorageException {
+        meterSystem = spy(new MeterSystem(moduleManager));
+        Whitebox.setInternalState(MetricsStreamProcessor.class, "PROCESSOR",
+                                  Mockito.spy(MetricsStreamProcessor.getInstance()));
+        doNothing().when(MetricsStreamProcessor.getInstance()).create(any(), (StreamDefinition) any(), any());
+    }
+
+    @Test
+    public void testCreate() {
+        // validate with same name, function and scope types
+        meterSystem.create("test_meter", "avg", ScopeType.SERVICE);
+        validateMeterDefinition("test_meter", Long.class, ScopeType.SERVICE);
+        meterSystem.create("test_meter", "avg", ScopeType.SERVICE);
+        validateMeterDefinition("test_meter", Long.class, ScopeType.SERVICE);
+
+        // validate with same name, difference scope type
+        try {
+            meterSystem.create("test_meter", "avg", ScopeType.SERVICE_INSTANCE);
+            throw new IllegalStateException();
+        } catch (IllegalArgumentException e) {
+            // If wrong arguments is means right
+        }
+
+        // validate with same name, difference function
+        try {
+            meterSystem.create("test_meter", "avgLabeled", ScopeType.SERVICE);
+            throw new IllegalStateException();
+        } catch (IllegalArgumentException e) {
+            // If wrong arguments is means right
+        }
+    }
+
+    private void validateMeterDefinition(String meterName, Class<?> dataType, ScopeType type) {
+        Map<String, ?> meterPrototypes = Whitebox.getInternalState(meterSystem, "meterPrototypes");
+        Object meterDefinition = meterPrototypes.get(meterName);
+        Class<?> realDataType = Whitebox.getInternalState(meterDefinition, "dataType");
+        ScopeType realScopeTypes = Whitebox.getInternalState(meterDefinition, "scopeType");
+
+        Assert.assertEquals(dataType, realDataType);
+        Assert.assertEquals(type, realScopeTypes);
+    }
+
+}