kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Change the model installation into the reactive module (#4761)

This commit is contained in:
吴晟 Wu Sheng 2020-05-08 21:21:03 +08:00 committed by GitHub
parent 1d5a0b0f1a
commit 7edcd5aa33
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
34 changed files with 284 additions and 356 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/OALRuntime.java
View File

@ -64,6 +64,7 @@ import org.apache.skywalking.oap.server.core.analysis.Stream;
import org.apache.skywalking.oap.server.core.analysis.StreamAnnotationListener;
import org.apache.skywalking.oap.server.core.oal.rt.OALCompileException;
import org.apache.skywalking.oap.server.core.oal.rt.OALEngine;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.library.util.ResourceUtils;
@ -165,7 +166,13 @@ public class OALRuntime implements OALEngine {
@Override
public void notifyAllListeners() throws ModuleStartException {
metricsClasses.forEach(streamAnnotationListener::notify);
for (Class metricsClass : metricsClasses) {
try {
streamAnnotationListener.notify(metricsClass);
} catch (StorageException e) {
throw new ModuleStartException(e.getMessage(), e);
}
}
for (Class dispatcherClass : dispatcherClasses) {
try {
dispatcherDetectorListener.addIfAsSourceDispatcher(dispatcherClass);

3
oap-server/oal-rt/src/test/java/org/apache/skywalking/oal/rt/parser/DeepAnalysisTest.java
View File

@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.List;
import org.apache.skywalking.oap.server.core.annotation.AnnotationScan;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@ -29,7 +30,7 @@ import org.junit.Test;
public class DeepAnalysisTest {
@BeforeClass
public static void init() throws IOException {
public static void init() throws IOException, StorageException {
AnnotationScan scopeScan = new AnnotationScan();
scopeScan.registerListener(new DefaultScopeDefine.Listener());
scopeScan.scan();

3
oap-server/oal-rt/src/test/java/org/apache/skywalking/oal/rt/parser/ScriptParserTest.java
View File

@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.List;
import org.apache.skywalking.oap.server.core.annotation.AnnotationScan;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@ -29,7 +30,7 @@ import org.junit.Test;
public class ScriptParserTest {
@BeforeClass
public static void init() throws IOException {
public static void init() throws IOException, StorageException {
MetricsHolder.init();
AnnotationScan scopeScan = new AnnotationScan();

8
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java
View File

@ -45,8 +45,8 @@ import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
import org.apache.skywalking.oap.server.core.server.JettyHandlerRegister;
import org.apache.skywalking.oap.server.core.source.SourceReceiver;
import org.apache.skywalking.oap.server.core.storage.model.IModelManager;
import org.apache.skywalking.oap.server.core.storage.model.IModelOverride;
import org.apache.skywalking.oap.server.core.storage.model.INewModel;
import org.apache.skywalking.oap.server.core.storage.model.ModelManipulator;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceGetter;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
@ -110,9 +110,9 @@ public class CoreModule extends ModuleDefine {
}
private void addInsideService(List<Class> classes) {
classes.add(INewModel.class);
classes.add(ModelCreator.class);
classes.add(IModelManager.class);
classes.add(IModelOverride.class);
classes.add(ModelManipulator.class);
classes.add(RemoteClientManager.class);
classes.add(RemoteSenderService.class);
}

18
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
View File

@ -68,9 +68,10 @@ import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.source.SourceReceiver;
import org.apache.skywalking.oap.server.core.source.SourceReceiverImpl;
import org.apache.skywalking.oap.server.core.storage.PersistenceTimer;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.model.IModelManager;
import org.apache.skywalking.oap.server.core.storage.model.IModelOverride;
import org.apache.skywalking.oap.server.core.storage.model.INewModel;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.ModelManipulator;
import org.apache.skywalking.oap.server.core.storage.model.StorageModels;
import org.apache.skywalking.oap.server.core.storage.ttl.DataTTLKeeperTimer;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceGetter;
@ -155,15 +156,14 @@ public class CoreModuleProvider extends ModuleProvider {
throw new ModuleStartException(e.getMessage(), e);
}
MeterSystem meterSystem = MeterSystem.meterSystem(getManager());
this.registerServiceImplementation(MeterSystem.class, meterSystem);
this.registerServiceImplementation(MeterSystem.class, new MeterSystem(getManager()));
AnnotationScan oalDisable = new AnnotationScan();
oalDisable.registerListener(DisableRegister.INSTANCE);
oalDisable.registerListener(new DisableRegister.SingleDisableScanListener());
try {
oalDisable.scan();
} catch (IOException e) {
} catch (IOException | StorageException e) {
throw new ModuleStartException(e.getMessage(), e);
}
@ -210,9 +210,9 @@ public class CoreModuleProvider extends ModuleProvider {
this.registerServiceImplementation(IWorkerInstanceSetter.class, instancesService);
this.registerServiceImplementation(RemoteSenderService.class, new RemoteSenderService(getManager()));
this.registerServiceImplementation(INewModel.class, storageModels);
this.registerServiceImplementation(ModelCreator.class, storageModels);
this.registerServiceImplementation(IModelManager.class, storageModels);
this.registerServiceImplementation(IModelOverride.class, storageModels);
this.registerServiceImplementation(ModelManipulator.class, storageModels);
this.registerServiceImplementation(
NetworkAddressAliasCache.class, new NetworkAddressAliasCache(moduleConfig));
@ -256,8 +256,6 @@ public class CoreModuleProvider extends ModuleProvider {
@Override
public void start() throws ModuleStartException {
MeterSystem.closeMeterCreationChannel();
grpcServer.addHandler(new RemoteServiceHandler(getManager()));
grpcServer.addHandler(new HealthCheckServiceHandler());
remoteClientManager.start();
@ -267,7 +265,7 @@ public class CoreModuleProvider extends ModuleProvider {
annotationScan.scan();
oalEngine.notifyAllListeners();
} catch (IOException | IllegalAccessException | InstantiationException e) {
} catch (IOException | IllegalAccessException | InstantiationException | StorageException e) {
throw new ModuleStartException(e.getMessage(), e);
}

3
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java
View File

@ -25,6 +25,7 @@ import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamingProces
import org.apache.skywalking.oap.server.core.analysis.worker.RecordStreamProcessor;
import org.apache.skywalking.oap.server.core.analysis.worker.TopNStreamProcessor;
import org.apache.skywalking.oap.server.core.annotation.AnnotationListener;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
/**
@ -45,7 +46,7 @@ public class StreamAnnotationListener implements AnnotationListener {
@SuppressWarnings("unchecked")
@Override
public void notify(Class aClass) {
public void notify(Class aClass) throws StorageException {
if (aClass.isAnnotationPresent(Stream.class)) {
Stream stream = (Stream) aClass.getAnnotation(Stream.class);

3
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamProcessor.java
View File

@ -18,11 +18,12 @@
package org.apache.skywalking.oap.server.core.analysis;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
public interface StreamProcessor<STREAM> {
void in(STREAM stream);
void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends STREAM> streamClass);
void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends STREAM> streamClass) throws StorageException;
}

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

@ -23,9 +23,7 @@ import com.google.common.reflect.ClassPath;
import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javassist.CannotCompileException;
import javassist.ClassPool;
@ -34,7 +32,6 @@ import javassist.CtConstructor;
import javassist.CtNewConstructor;
import javassist.CtNewMethod;
import javassist.NotFoundException;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@ -45,6 +42,7 @@ import org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableV
import org.apache.skywalking.oap.server.core.analysis.meter.function.MeterFunction;
import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
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.apache.skywalking.oap.server.library.module.Service;
@ -57,29 +55,18 @@ import org.apache.skywalking.oap.server.library.module.Service;
@Slf4j
public class MeterSystem implements Service {
private static final String METER_CLASS_PACKAGE = "org.apache.skywalking.oap.server.core.analysis.meter.dynamic.";
private static ModuleManager MANAGER;
private static ClassPool CLASS_POOL;
private static List<NewMeter> TO_BE_CREATED_METERS = new ArrayList<>();
private static Map<String, Class<? extends MeterFunction>> FUNCTION_REGISTER = new HashMap<>();
private ModuleManager manager;
private ClassPool classPool;
private Map<String, Class<? extends MeterFunction>> functionRegister = new HashMap<>();
/**
* Host the dynamic meter prototype classes. These classes could be create dynamically through {@link
* Object#clone()} in the runtime;
*/
private static Map<String, MeterDefinition> METER_PROTOTYPES = new HashMap<>();
private static MeterSystem METER_SYSTEM;
private static boolean METER_CREATABLE = true;
private Map<String, MeterDefinition> meterPrototypes = new HashMap<>();
private MeterSystem() {
}
public synchronized static MeterSystem meterSystem(final ModuleManager manager) {
if (METER_SYSTEM != null) {
return METER_SYSTEM;
}
MANAGER = manager;
CLASS_POOL = ClassPool.getDefault();
public MeterSystem(final ModuleManager manager) {
this.manager = manager;
classPool = ClassPool.getDefault();
ClassPath classpath = null;
try {
@ -97,14 +84,12 @@ public class MeterSystem implements Service {
throw new IllegalArgumentException(
"Function " + functionClass.getCanonicalName() + " doesn't implement AcceptableValue.");
}
FUNCTION_REGISTER.put(
functionRegister.put(
metricsFunction.functionName(),
(Class<? extends MeterFunction>) functionClass
);
}
}
METER_SYSTEM = new MeterSystem();
return METER_SYSTEM;
}
/**
@ -121,126 +106,100 @@ public class MeterSystem implements Service {
String functionName,
ScopeType type,
Class<T> dataType) throws IllegalArgumentException {
if (!METER_CREATABLE) {
throw new IllegalStateException("Can't create new metrics anymore");
/**
* Create a new meter class dynamically.
*/
final Class<? extends MeterFunction> meterFunction = functionRegister.get(functionName);
if (meterFunction == null) {
throw new IllegalArgumentException("Function " + functionName + " can't be found.");
}
final NewMeter newMeter = new NewMeter(metricsName, functionName, type, dataType);
if (TO_BE_CREATED_METERS.contains(newMeter)) {
return false;
boolean foundDataType = false;
String acceptance = null;
for (final Type genericInterface : meterFunction.getGenericInterfaces()) {
if (genericInterface instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) genericInterface;
if (parameterizedType.getRawType().getTypeName().equals(AcceptableValue.class.getName())) {
Type[] arguments = parameterizedType.getActualTypeArguments();
if (arguments[0].equals(dataType)) {
foundDataType = true;
} else {
acceptance = arguments[0].getTypeName();
}
}
if (foundDataType) {
break;
}
}
}
if (!foundDataType) {
throw new IllegalArgumentException("Function " + functionName
+ " requires <" + acceptance + "> in AcceptableValue"
+ " but using " + dataType.getName() + " in the creation");
}
TO_BE_CREATED_METERS.add(newMeter);
final CtClass parentClass;
try {
parentClass = classPool.get(meterFunction.getCanonicalName());
if (!Metrics.class.isAssignableFrom(meterFunction)) {
throw new IllegalArgumentException(
"Function " + functionName + " doesn't inherit from Metrics.");
}
} catch (NotFoundException e) {
throw new IllegalArgumentException("Function " + functionName + " can't be found by javaassist.");
}
final String className = formatName(metricsName);
CtClass metricsClass = classPool.makeClass(METER_CLASS_PACKAGE + className, parentClass);
/**
* Create empty construct
*/
try {
CtConstructor defaultConstructor = CtNewConstructor.make(
"public " + className + "() {}", metricsClass);
metricsClass.addConstructor(defaultConstructor);
} catch (CannotCompileException e) {
log.error("Can't add empty constructor in " + className + ".", e);
throw new UnexpectedException(e.getMessage(), e);
}
/**
* Generate `AcceptableValue<T> createNew()` method.
*/
try {
metricsClass.addMethod(CtNewMethod.make(
""
+ "public org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue createNew() {"
+ " return new " + METER_CLASS_PACKAGE + className + "();"
+ " }"
, metricsClass));
} catch (CannotCompileException e) {
log.error("Can't generate createNew method for " + className + ".", e);
throw new UnexpectedException(e.getMessage(), e);
}
Class targetClass;
try {
targetClass = metricsClass.toClass(MeterSystem.class.getClassLoader(), null);
AcceptableValue prototype = (AcceptableValue) targetClass.newInstance();
meterPrototypes.put(metricsName, new MeterDefinition(type, prototype, dataType));
log.debug("Generate metrics class, " + metricsClass.getName());
MetricsStreamProcessor.getInstance().create(
manager,
new StreamDefinition(
metricsName, type.getScopeId(), prototype.builder(), MetricsStreamProcessor.class),
targetClass
);
} catch (CannotCompileException | IllegalAccessException | InstantiationException | StorageException e) {
log.error("Can't compile/load/init " + className + ".", e);
throw new UnexpectedException(e.getMessage(), e);
}
return true;
}
/**
* Close the {@link #create(String, String, ScopeType, Class)} channel, and build the model and streaming
* definitions.
*/
public static void closeMeterCreationChannel() {
METER_CREATABLE = false;
TO_BE_CREATED_METERS.forEach(newMeter -> {
String metricsName = newMeter.metricsName;
String functionName = newMeter.functionName;
ScopeType type = newMeter.type;
Class<?> dataType = newMeter.dataType;
/**
* Create a new meter class dynamically.
*/
final Class<? extends MeterFunction> meterFunction = FUNCTION_REGISTER.get(functionName);
if (meterFunction == null) {
throw new IllegalArgumentException("Function " + functionName + " can't be found.");
}
boolean foundDataType = false;
String acceptance = null;
for (final Type genericInterface : meterFunction.getGenericInterfaces()) {
if (genericInterface instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) genericInterface;
if (parameterizedType.getRawType().getTypeName().equals(AcceptableValue.class.getName())) {
Type[] arguments = parameterizedType.getActualTypeArguments();
if (arguments[0].equals(dataType)) {
foundDataType = true;
} else {
acceptance = arguments[0].getTypeName();
}
}
if (foundDataType) {
break;
}
}
}
if (!foundDataType) {
throw new IllegalArgumentException("Function " + functionName
+ " requires <" + acceptance + "> in AcceptableValue"
+ " but using " + dataType.getName() + " in the creation");
}
final CtClass parentClass;
try {
parentClass = CLASS_POOL.get(meterFunction.getCanonicalName());
if (!Metrics.class.isAssignableFrom(meterFunction)) {
throw new IllegalArgumentException(
"Function " + functionName + " doesn't inherit from Metrics.");
}
} catch (NotFoundException e) {
throw new IllegalArgumentException("Function " + functionName + " can't be found by javaassist.");
}
final String className = formatName(metricsName);
CtClass metricsClass = CLASS_POOL.makeClass(METER_CLASS_PACKAGE + className, parentClass);
/**
* Create empty construct
*/
try {
CtConstructor defaultConstructor = CtNewConstructor.make(
"public " + className + "() {}", metricsClass);
metricsClass.addConstructor(defaultConstructor);
} catch (CannotCompileException e) {
log.error("Can't add empty constructor in " + className + ".", e);
throw new UnexpectedException(e.getMessage(), e);
}
/**
* Generate `AcceptableValue<T> createNew()` method.
*/
try {
metricsClass.addMethod(CtNewMethod.make(
""
+ "public org.apache.skywalking.oap.server.core.analysis.meter.function.AcceptableValue createNew() {"
+ " return new " + METER_CLASS_PACKAGE + className + "();"
+ " }"
, metricsClass));
} catch (CannotCompileException e) {
log.error("Can't generate createNew method for " + className + ".", e);
throw new UnexpectedException(e.getMessage(), e);
}
Class targetClass;
try {
targetClass = metricsClass.toClass(MeterSystem.class.getClassLoader(), null);
AcceptableValue prototype = (AcceptableValue) targetClass.newInstance();
METER_PROTOTYPES.put(metricsName, new MeterDefinition(type, prototype, dataType));
log.debug("Generate metrics class, " + metricsClass.getName());
MetricsStreamProcessor.getInstance().create(
MANAGER,
new StreamDefinition(
metricsName, type.getScopeId(), prototype.builder(), MetricsStreamProcessor.class),
targetClass
);
} catch (CannotCompileException | IllegalAccessException | InstantiationException e) {
log.error("Can't compile/load/init " + className + ".", e);
throw new UnexpectedException(e.getMessage(), e);
}
});
}
/**
* Create an {@link AcceptableValue} instance for streaming calculation. AcceptableValue instance is stateful,
* shouldn't do {@link AcceptableValue#accept(MeterEntity, Object)} once it is pushed into {@link
@ -253,7 +212,7 @@ public class MeterSystem implements Service {
*/
public <T> AcceptableValue<T> buildMetrics(String metricsName,
Class<T> dataType) {
MeterDefinition meterDefinition = METER_PROTOTYPES.get(metricsName);
MeterDefinition meterDefinition = meterPrototypes.get(metricsName);
if (meterDefinition == null) {
throw new IllegalArgumentException("Uncreated metrics " + metricsName);
}
@ -284,15 +243,6 @@ public class MeterSystem implements Service {
return metricsName.toLowerCase();
}
@RequiredArgsConstructor
@EqualsAndHashCode
public static class NewMeter {
private final String metricsName;
private final String functionName;
private final ScopeType type;
private final Class<?> dataType;
}
@RequiredArgsConstructor
@Getter
private static class MeterDefinition {

9
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
View File

@ -36,9 +36,10 @@ import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.config.DownSamplingConfigService;
import org.apache.skywalking.oap.server.core.storage.IMetricsDAO;
import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
import org.apache.skywalking.oap.server.core.storage.model.INewModel;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
@ -92,14 +93,14 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
* @param stream definition of the metrics class.
* @param metricsClass data type of the streaming calculation.
*/
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends Metrics> metricsClass) {
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends Metrics> metricsClass) throws StorageException {
this.create(moduleDefineHolder, StreamDefinition.from(stream), metricsClass);
}
@SuppressWarnings("unchecked")
public void create(ModuleDefineHolder moduleDefineHolder,
StreamDefinition stream,
Class<? extends Metrics> metricsClass) {
Class<? extends Metrics> metricsClass) throws StorageException {
if (DisableRegister.INSTANCE.include(stream.getName())) {
return;
}
@ -112,7 +113,7 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
throw new UnexpectedException("Create " + stream.getBuilder().getSimpleName() + " metrics DAO failure.", e);
}
INewModel modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(INewModel.class);
ModelCreator modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(ModelCreator.class);
DownSamplingConfigService configService = moduleDefineHolder.find(CoreModule.NAME)
.provider()
.getService(DownSamplingConfigService.class);

7
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamingProcessor.java
View File

@ -29,9 +29,10 @@ import org.apache.skywalking.oap.server.core.analysis.StreamProcessor;
import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
import org.apache.skywalking.oap.server.core.storage.INoneStreamDAO;
import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
import org.apache.skywalking.oap.server.core.storage.model.INewModel;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
@ -58,7 +59,7 @@ public class NoneStreamingProcessor implements StreamProcessor<NoneStream> {
}
@Override
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends NoneStream> streamClass) {
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends NoneStream> streamClass) throws StorageException {
if (DisableRegister.INSTANCE.include(stream.name())) {
return;
}
@ -72,7 +73,7 @@ public class NoneStreamingProcessor implements StreamProcessor<NoneStream> {
.getSimpleName() + " none stream record DAO failure.", e);
}
INewModel modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(INewModel.class);
ModelCreator modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(ModelCreator.class);
Model model = modelSetter.add(streamClass, stream.scopeId(), new Storage(stream.name(), DownSampling.Second), true);
final NoneStreamPersistentWorker persistentWorker = new NoneStreamPersistentWorker(moduleDefineHolder, model, noneStream);

7
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java
View File

@ -29,9 +29,10 @@ import org.apache.skywalking.oap.server.core.analysis.StreamProcessor;
import org.apache.skywalking.oap.server.core.analysis.record.Record;
import org.apache.skywalking.oap.server.core.storage.IRecordDAO;
import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
import org.apache.skywalking.oap.server.core.storage.model.INewModel;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
@ -53,7 +54,7 @@ public class RecordStreamProcessor implements StreamProcessor<Record> {
}
@SuppressWarnings("unchecked")
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends Record> recordClass) {
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends Record> recordClass) throws StorageException {
if (DisableRegister.INSTANCE.include(stream.name())) {
return;
}
@ -66,7 +67,7 @@ public class RecordStreamProcessor implements StreamProcessor<Record> {
throw new UnexpectedException("Create " + stream.builder().getSimpleName() + " record DAO failure.", e);
}
INewModel modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(INewModel.class);
ModelCreator modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(ModelCreator.class);
Model model = modelSetter.add(
recordClass, stream.scopeId(), new Storage(stream.name(), DownSampling.Second), true);
RecordPersistentWorker persistentWorker = new RecordPersistentWorker(moduleDefineHolder, model, recordDAO);

7
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNStreamProcessor.java
View File

@ -34,9 +34,10 @@ import org.apache.skywalking.oap.server.core.analysis.record.Record;
import org.apache.skywalking.oap.server.core.analysis.topn.TopN;
import org.apache.skywalking.oap.server.core.storage.IRecordDAO;
import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
import org.apache.skywalking.oap.server.core.storage.model.INewModel;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
@ -63,7 +64,7 @@ public class TopNStreamProcessor implements StreamProcessor<TopN> {
}
@SuppressWarnings("unchecked")
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends TopN> topNClass) {
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends TopN> topNClass) throws StorageException {
if (DisableRegister.INSTANCE.include(stream.name())) {
return;
}
@ -77,7 +78,7 @@ public class TopNStreamProcessor implements StreamProcessor<TopN> {
.getSimpleName() + " top n record DAO failure.", e);
}
INewModel modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(INewModel.class);
ModelCreator modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(ModelCreator.class);
Model model = modelSetter.add(
topNClass, stream.scopeId(), new Storage(stream.name(), DownSampling.Second), true);

3
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/annotation/AnnotationListener.java
View File

@ -19,10 +19,11 @@
package org.apache.skywalking.oap.server.core.annotation;
import java.lang.annotation.Annotation;
import org.apache.skywalking.oap.server.core.storage.StorageException;
public interface AnnotationListener {
Class<? extends Annotation> annotation();
void notify(Class aClass);
void notify(Class aClass) throws StorageException;
}

13
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/annotation/AnnotationScan.java
View File

@ -25,6 +25,7 @@ import java.lang.annotation.Annotation;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import org.apache.skywalking.oap.server.core.storage.StorageException;
/**
* Scan the annotation, and notify the listener(s)
@ -49,7 +50,7 @@ public class AnnotationScan {
/**
* Begin to scan classes.
*/
public void scan() throws IOException {
public void scan() throws IOException, StorageException {
ClassPath classpath = ClassPath.from(this.getClass().getClassLoader());
ImmutableSet<ClassPath.ClassInfo> classes = classpath.getTopLevelClassesRecursive("org.apache.skywalking");
for (ClassPath.ClassInfo classInfo : classes) {
@ -62,7 +63,9 @@ public class AnnotationScan {
}
}
listeners.forEach(AnnotationListenerCache::complete);
for (AnnotationListenerCache listener : listeners) {
listener.complete();
}
}
private class AnnotationListenerCache {
@ -82,9 +85,11 @@ public class AnnotationScan {
matchedClass.add(aClass);
}
private void complete() {
private void complete() throws StorageException {
matchedClass.sort(Comparator.comparing(Class::getName));
matchedClass.forEach(aClass -> listener.notify(aClass));
for (Class<?> aClass : matchedClass) {
listener.notify(aClass);
}
}
}
}

4
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/Column.java
View File

@ -24,7 +24,7 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import lombok.Getter;
import org.apache.skywalking.oap.server.core.query.sql.Function;
import org.apache.skywalking.oap.server.core.storage.model.IModelOverride;
import org.apache.skywalking.oap.server.core.storage.model.ModelManipulator;
/**
* Data column of all persistent entity.
@ -34,7 +34,7 @@ import org.apache.skywalking.oap.server.core.storage.model.IModelOverride;
public @interface Column {
/**
* column name in the storage. Most of the storage will keep the name consistently. But in same cases, this name
* could be a keyword, then, the implementation will use {@link IModelOverride} to replace the column name.
* could be a keyword, then, the implementation will use {@link ModelManipulator} to replace the column name.
*/
String columnName();

11
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/INewModel.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelCreator.java
View File

@ -18,17 +18,24 @@
package org.apache.skywalking.oap.server.core.storage.model;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
import org.apache.skywalking.oap.server.library.module.Service;
/**
* INewModel implementation supports creating a new module.
*/
public interface INewModel extends Service {
public interface ModelCreator extends Service {
/**
* Add a new model
*
* @return the created new model
*/
Model add(Class<?> aClass, int scopeId, Storage storage, boolean record);
Model add(Class<?> aClass, int scopeId, Storage storage, boolean record) throws StorageException;
void addModelListener(CreatingListener listener) throws StorageException;
interface CreatingListener {
void whenCreating(Model model) throws StorageException;
}
}

61
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java
View File

@ -18,7 +18,7 @@
package org.apache.skywalking.oap.server.core.storage.model;
import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.RunningMode;
@ -29,63 +29,52 @@ import org.apache.skywalking.oap.server.library.module.ModuleManager;
/**
* The core module installation controller.
*/
@RequiredArgsConstructor
@Slf4j
public abstract class ModelInstaller {
public abstract class ModelInstaller implements ModelCreator.CreatingListener {
protected final Client client;
private final ModuleManager moduleManager;
public ModelInstaller(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
/**
* Entrance of the storage entity installation work.
*/
public final void install(Client client) throws StorageException {
IModelManager modelGetter = moduleManager.find(CoreModule.NAME).provider().getService(IModelManager.class);
List<Model> models = modelGetter.allModels();
public void whenCreating(Model model) throws StorageException {
if (RunningMode.isNoInitMode()) {
for (Model model : models) {
while (!isExists(client, model)) {
try {
log.info(
"table: {} does not exist. OAP is running in 'no-init' mode, waiting... retry 3s later.",
model
.getName()
);
Thread.sleep(3000L);
} catch (InterruptedException e) {
log.error(e.getMessage());
}
while (!isExists(model)) {
try {
log.info(
"table: {} does not exist. OAP is running in 'no-init' mode, waiting... retry 3s later.",
model
.getName()
);
Thread.sleep(3000L);
} catch (InterruptedException e) {
log.error(e.getMessage());
}
}
} else {
for (Model model : models) {
if (!isExists(client, model)) {
log.info("table: {} does not exist", model.getName());
createTable(client, model);
}
if (!isExists(model)) {
log.info("table: {} does not exist", model.getName());
createTable(model);
}
}
}
/**
* Installer implementation could use this API to request a column name replacement. This method delegates for
* {@link IModelOverride}.
* {@link ModelManipulator}.
*/
protected final void overrideColumnName(String columnName, String newName) {
IModelOverride modelOverride = moduleManager.find(CoreModule.NAME).provider().getService(IModelOverride.class);
ModelManipulator modelOverride = moduleManager.find(CoreModule.NAME)
.provider()
.getService(ModelManipulator.class);
modelOverride.overrideColumnName(columnName, newName);
}
/**
* Check whether the storage entity exists. Need to implement based on the real storage.
*/
protected abstract boolean isExists(Client client, Model model) throws StorageException;
protected abstract boolean isExists(Model model) throws StorageException;
/**
* Create the storage entity. All creations should be after the {@link #isExists(Client, Model)} check.
* Create the storage entity. All creations should be after the {@link #isExists(Model)} check.
*/
protected abstract void createTable(Client client, Model model) throws StorageException;
protected abstract void createTable(Model model) throws StorageException;
}

2
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/IModelOverride.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelManipulator.java
View File

@ -23,6 +23,6 @@ import org.apache.skywalking.oap.server.library.module.Service;
/**
* Override service provides ways to rename the existing column or table name.
*/
public interface IModelOverride extends Service {
public interface ModelManipulator extends Service {
void overrideColumnName(String columnName, String newName);
}

43
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
View File

@ -20,11 +20,12 @@ package org.apache.skywalking.oap.server.core.storage.model;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.core.storage.annotation.MultipleQueryUnifiedIndex;
import org.apache.skywalking.oap.server.core.storage.annotation.QueryUnifiedIndex;
@ -35,15 +36,19 @@ import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetad
* StorageModels manages all models detected by the core.
*/
@Slf4j
public class StorageModels implements IModelManager, INewModel, IModelOverride {
public class StorageModels implements IModelManager, ModelCreator, ModelManipulator {
private final List<Model> models;
private final HashMap<String, String> columnNameOverrideRule;
private final List<CreatingListener> listeners;
public StorageModels() {
this.models = new LinkedList<>();
this.models = new ArrayList<>();
this.columnNameOverrideRule = new HashMap<>();
this.listeners = new ArrayList<>();
}
@Override
public Model add(Class<?> aClass, int scopeId, Storage storage, boolean record) {
public Model add(Class<?> aClass, int scopeId, Storage storage, boolean record) throws StorageException {
// Check this scope id is valid.
DefaultScopeDefine.nameOf(scopeId);
@ -55,11 +60,30 @@ public class StorageModels implements IModelManager, INewModel, IModelOverride {
storage.getModelName(), modelColumns, extraQueryIndices, scopeId,
storage.getDownsampling(), record
);
this.followColumnNameRules(model);
models.add(model);
for (final CreatingListener listener : listeners) {
listener.whenCreating(model);
}
return model;
}
/**
* CreatingListener listener could react when {@link #add(Class, int, Storage, boolean)} model happens. Also, the
* added models are being notified in this add operation.
*/
@Override
public void addModelListener(final CreatingListener listener) throws StorageException {
listeners.add(listener);
for (Model model : models) {
listener.whenCreating(model);
}
}
/**
* Read model column metadata based on the class level definition.
*/
private void retrieval(Class<?> clazz,
String modelName,
List<ModelColumn> modelColumns,
@ -108,9 +132,14 @@ public class StorageModels implements IModelManager, INewModel, IModelOverride {
@Override
public void overrideColumnName(String columnName, String newName) {
models.forEach(model -> {
model.getColumns().forEach(column -> column.getColumnName().overrideName(columnName, newName));
model.getExtraQueryIndices().forEach(extraQueryIndex -> extraQueryIndex.overrideName(columnName, newName));
columnNameOverrideRule.put(columnName, newName);
models.forEach(this::followColumnNameRules);
}
private void followColumnNameRules(Model model) {
columnNameOverrideRule.forEach((oldName, newName) -> {
model.getColumns().forEach(column -> column.getColumnName().overrideName(oldName, newName));
model.getExtraQueryIndices().forEach(extraQueryIndex -> extraQueryIndex.overrideName(oldName, newName));
});
}

66
oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/StorageInstallerTestCase.java
View File

@ -1,66 +0,0 @@
/*
* 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.storage;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.CoreModuleProvider;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller;
import org.apache.skywalking.oap.server.library.client.Client;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
import org.junit.Test;
import org.mockito.Mockito;
import org.powermock.reflect.Whitebox;
public class StorageInstallerTestCase {
@Test
public void testInstall() throws StorageException, ServiceNotProvidedException {
CoreModuleProvider moduleProvider = Mockito.mock(CoreModuleProvider.class);
CoreModule moduleDefine = Mockito.spy(CoreModule.class);
ModuleManager moduleManager = Mockito.mock(ModuleManager.class);
Whitebox.setInternalState(moduleDefine, "loadedProvider", moduleProvider);
Mockito.when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleDefine);
// streamDataMapping.generate();
// TestStorageInstaller installer = new TestStorageInstaller(moduleManager);
// installer.install(null);
}
class TestStorageInstaller extends ModelInstaller {
public TestStorageInstaller(ModuleManager moduleManager) {
super(moduleManager);
}
@Override
protected boolean isExists(Client client, Model tableDefine) throws StorageException {
return false;
}
@Override
protected void createTable(Client client, Model tableDefine) throws StorageException {
}
}
}

3
oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/StorageModelsTest.java
View File

@ -26,6 +26,7 @@ import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProces
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
import org.apache.skywalking.oap.server.core.storage.StorageData;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.core.storage.annotation.QueryUnifiedIndex;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
@ -47,7 +48,7 @@ public class StorageModelsTest {
}
@Test
public void testStorageModels() {
public void testStorageModels() throws StorageException {
StorageModels models = new StorageModels();
models.add(TestModel.class, -1,
new Storage("StorageModelsTest", DownSampling.Hour),

19
oap-server/server-fetcher-plugin/prometheus-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/fetcher/prometheus/provider/PrometheusFetcherProvider.java
View File

@ -59,17 +59,6 @@ public class PrometheusFetcherProvider extends ModuleProvider {
@Override
public void prepare() throws ServiceNotProvidedException, ModuleStartException {
if (config.isActive()) {
// TODO. This is only a demo about creating new metrics
// We should create it based on metrics configuration.
final MeterSystem meterSystem = MeterSystem.meterSystem(getManager());
meterSystem.create("test_long_metrics", "avg", ScopeType.SERVICE, Long.class);
meterSystem.create("test_histogram_metrics", "histogram", ScopeType.SERVICE, BucketedValues.class);
meterSystem.create(
"test_percentile_metrics", "percentile", ScopeType.SERVICE,
PercentileFunction.PercentileArgument.class
);
}
}
@Override
@ -82,6 +71,14 @@ public class PrometheusFetcherProvider extends ModuleProvider {
if (config.isActive()) {
// TODO. This is only a demo about fetching the data and push into the calculation stream.
final MeterSystem service = getManager().find(CoreModule.NAME).provider().getService(MeterSystem.class);
service.create("test_long_metrics", "avg", ScopeType.SERVICE, Long.class);
service.create("test_histogram_metrics", "histogram", ScopeType.SERVICE, BucketedValues.class);
service.create(
"test_percentile_metrics", "percentile", ScopeType.SERVICE,
PercentileFunction.PercentileArgument.class
);
Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable() {
@Override
public void run() {

6
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java
View File

@ -35,6 +35,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskLogQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileThreadSnapshotQueryDAO;
@ -178,10 +179,9 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
public void start() throws ModuleStartException {
try {
elasticSearchClient.connect();
StorageEsInstaller installer = new StorageEsInstaller(elasticSearchClient, getManager(), config);
StorageEsInstaller installer = new StorageEsInstaller(getManager(), config);
installer.install(elasticSearchClient);
getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
} catch (StorageException | IOException | KeyStoreException | NoSuchAlgorithmException | KeyManagementException | CertificateException e) {
throw new ModuleStartException(e.getMessage(), e);
}

8
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java
View File

@ -41,14 +41,14 @@ public class StorageEsInstaller extends ModelInstaller {
private final StorageModuleElasticsearchConfig config;
protected final ColumnTypeEsMapping columnTypeEsMapping;
public StorageEsInstaller(ModuleManager moduleManager, final StorageModuleElasticsearchConfig config) {
super(moduleManager);
public StorageEsInstaller(Client client, ModuleManager moduleManager, final StorageModuleElasticsearchConfig config) {
super(client, moduleManager);
this.columnTypeEsMapping = new ColumnTypeEsMapping();
this.config = config;
}
@Override
protected boolean isExists(Client client, Model model) throws StorageException {
protected boolean isExists(Model model) throws StorageException {
ElasticSearchClient esClient = (ElasticSearchClient) client;
try {
String timeSeriesIndexName = TimeSeriesUtils.latestWriteIndexName(model);
@ -59,7 +59,7 @@ public class StorageEsInstaller extends ModelInstaller {
}
@Override
protected void createTable(Client client, Model model) throws StorageException {
protected void createTable(Model model) throws StorageException {
ElasticSearchClient esClient = (ElasticSearchClient) client;
Map<String, Object> settings = createSetting(model.isRecord());

5
oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java
View File

@ -33,6 +33,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskLogQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileThreadSnapshotQueryDAO;
@ -184,8 +185,8 @@ public class StorageModuleElasticsearch7Provider extends ModuleProvider {
try {
elasticSearch7Client.connect();
StorageEs7Installer installer = new StorageEs7Installer(getManager(), config);
installer.install(elasticSearch7Client);
StorageEs7Installer installer = new StorageEs7Installer(elasticSearch7Client, getManager(), config);
getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
} catch (StorageException | IOException | KeyStoreException | NoSuchAlgorithmException | KeyManagementException | CertificateException e) {
throw new ModuleStartException(e.getMessage(), e);
}

19
oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/base/StorageEs7Installer.java
View File

@ -17,22 +17,21 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.base;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.library.client.Client;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.StorageEsInstaller;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.StorageModuleElasticsearch7Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
@Slf4j
public class StorageEs7Installer extends StorageEsInstaller {
private static final Logger logger = LoggerFactory.getLogger(StorageEs7Installer.class);
public StorageEs7Installer(final ModuleManager moduleManager, final StorageModuleElasticsearch7Config config) {
super(moduleManager, config);
public StorageEs7Installer(final Client client,
final ModuleManager moduleManager,
final StorageModuleElasticsearch7Config config) {
super(client, moduleManager, config);
}
@SuppressWarnings("unchecked")
@ -41,7 +40,7 @@ public class StorageEs7Installer extends StorageEsInstaller {
Map<String, Object> type = (Map<String, Object>) mapping.remove(ElasticSearchClient.TYPE);
mapping.put("properties", type.get("properties"));
logger.debug("elasticsearch index template setting: {}", mapping.toString());
log.debug("elasticsearch index template setting: {}", mapping.toString());
return mapping;
}

5
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageProvider.java
View File

@ -26,6 +26,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskLogQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileThreadSnapshotQueryDAO;
@ -114,9 +115,9 @@ public class InfluxStorageProvider extends ModuleProvider {
public void start() throws ServiceNotProvidedException, ModuleStartException {
client.connect();
InfluxTableInstaller installer = new InfluxTableInstaller(getManager());
InfluxTableInstaller installer = new InfluxTableInstaller(client, getManager());
try {
installer.install(client);
getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
} catch (StorageException e) {
throw new ModuleStartException(e.getMessage(), e);
}

8
oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxTableInstaller.java
View File

@ -26,18 +26,18 @@ import org.apache.skywalking.oap.server.library.module.ModuleManager;
public class InfluxTableInstaller extends ModelInstaller {
public InfluxTableInstaller(ModuleManager moduleManager) {
super(moduleManager);
public InfluxTableInstaller(Client client, ModuleManager moduleManager) {
super(client, moduleManager);
}
@Override
protected boolean isExists(final Client client, final Model model) throws StorageException {
protected boolean isExists(final Model model) throws StorageException {
TableMetaInfo.addModel(model);
return true;
}
@Override
protected void createTable(final Client client, final Model model) throws StorageException {
protected void createTable(final Model model) throws StorageException {
// Automatically create table
}
}

5
oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java
View File

@ -26,6 +26,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskLogQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileThreadSnapshotQueryDAO;
@ -132,8 +133,8 @@ public class H2StorageProvider extends ModuleProvider {
try {
h2Client.connect();
H2TableInstaller installer = new H2TableInstaller(getManager());
installer.install(h2Client);
H2TableInstaller installer = new H2TableInstaller(h2Client, getManager());
getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
} catch (StorageException e) {
throw new ModuleStartException(e.getMessage(), e);
}

8
oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java
View File

@ -44,18 +44,18 @@ import org.apache.skywalking.oap.server.storage.plugin.jdbc.TableMetaInfo;
public class H2TableInstaller extends ModelInstaller {
public static final String ID_COLUMN = "id";
public H2TableInstaller(ModuleManager moduleManager) {
super(moduleManager);
public H2TableInstaller(Client client, ModuleManager moduleManager) {
super(client, moduleManager);
}
@Override
protected boolean isExists(Client client, Model model) throws StorageException {
protected boolean isExists(Model model) throws StorageException {
TableMetaInfo.addModel(model);
return false;
}
@Override
protected void createTable(Client client, Model model) throws StorageException {
protected void createTable(Model model) throws StorageException {
JDBCHikariCPClient jdbcHikariCPClient = (JDBCHikariCPClient) client;
try (Connection connection = jdbcHikariCPClient.getConnection()) {
SQLBuilder tableCreateSQL = new SQLBuilder("CREATE TABLE IF NOT EXISTS " + model.getName() + " (");

5
oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java
View File

@ -25,6 +25,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageDAO;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskLogQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileTaskQueryDAO;
import org.apache.skywalking.oap.server.core.storage.profile.IProfileThreadSnapshotQueryDAO;
@ -122,8 +123,8 @@ public class MySQLStorageProvider extends ModuleProvider {
try {
mysqlClient.connect();
MySQLTableInstaller installer = new MySQLTableInstaller(getManager());
installer.install(mysqlClient);
MySQLTableInstaller installer = new MySQLTableInstaller(mysqlClient, getManager());
getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
} catch (StorageException e) {
throw new ModuleStartException(e.getMessage(), e);
}

6
oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java
View File

@ -40,8 +40,8 @@ import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TableInstal
*/
@Slf4j
public class MySQLTableInstaller extends H2TableInstaller {
public MySQLTableInstaller(ModuleManager moduleManager) {
super(moduleManager);
public MySQLTableInstaller(Client client, ModuleManager moduleManager) {
super(client, moduleManager);
/*
* Override column because the default column names in core have syntax conflict with MySQL.
*/
@ -50,7 +50,7 @@ public class MySQLTableInstaller extends H2TableInstaller {
}
@Override
protected boolean isExists(Client client, Model model) throws StorageException {
protected boolean isExists(Model model) throws StorageException {
TableMetaInfo.addModel(model);
JDBCHikariCPClient h2Client = (JDBCHikariCPClient) client;
try (Connection conn = h2Client.getConnection()) {

16
oap-server/server-tools/profile-exporter/tool-profile-snapshot-server-mock/src/main/java/org/apache/skywalking/oap/server/tool/profile/core/MockCoreModuleProvider.java
View File

@ -49,9 +49,10 @@ import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
import org.apache.skywalking.oap.server.core.server.JettyHandlerRegister;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.source.SourceReceiver;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.core.storage.model.IModelManager;
import org.apache.skywalking.oap.server.core.storage.model.IModelOverride;
import org.apache.skywalking.oap.server.core.storage.model.INewModel;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.ModelManipulator;
import org.apache.skywalking.oap.server.core.storage.model.StorageModels;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceGetter;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
@ -108,8 +109,7 @@ public class MockCoreModuleProvider extends CoreModuleProvider {
throw new ModuleStartException(e.getMessage(), e);
}
MeterSystem meterSystem = MeterSystem.meterSystem(getManager());
this.registerServiceImplementation(MeterSystem.class, meterSystem);
this.registerServiceImplementation(MeterSystem.class, new MeterSystem(getManager()));
CoreModuleConfig moduleConfig = new CoreModuleConfig();
this.registerServiceImplementation(ConfigService.class, new ConfigService(moduleConfig));
@ -129,9 +129,9 @@ public class MockCoreModuleProvider extends CoreModuleProvider {
this.registerServiceImplementation(IWorkerInstanceSetter.class, instancesService);
this.registerServiceImplementation(RemoteSenderService.class, new RemoteSenderService(getManager()));
this.registerServiceImplementation(INewModel.class, storageModels);
this.registerServiceImplementation(ModelCreator.class, storageModels);
this.registerServiceImplementation(IModelManager.class, storageModels);
this.registerServiceImplementation(IModelOverride.class, storageModels);
this.registerServiceImplementation(ModelManipulator.class, storageModels);
this.registerServiceImplementation(
NetworkAddressAliasCache.class, new NetworkAddressAliasCache(moduleConfig));
@ -160,11 +160,9 @@ public class MockCoreModuleProvider extends CoreModuleProvider {
@Override
public void start() throws ModuleStartException {
MeterSystem.closeMeterCreationChannel();
try {
annotationScan.scan();
} catch (IOException e) {
} catch (IOException | StorageException e) {
throw new ModuleStartException(e.getMessage(), e);
}
}

3
oap-server/server-tools/profile-exporter/tool-profile-snapshot-server-mock/src/main/java/org/apache/skywalking/oap/server/tool/profile/core/mock/MockStreamAnnotationListener.java
View File

@ -25,6 +25,7 @@ import org.apache.skywalking.oap.server.core.analysis.StreamAnnotationListener;
import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamingProcessor;
import org.apache.skywalking.oap.server.core.analysis.worker.RecordStreamProcessor;
import org.apache.skywalking.oap.server.core.annotation.AnnotationListener;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
/**
@ -44,7 +45,7 @@ public class MockStreamAnnotationListener implements AnnotationListener {
}
@Override
public void notify(Class aClass) {
public void notify(Class aClass) throws StorageException {
if (aClass.isAnnotationPresent(Stream.class)) {
Stream stream = (Stream) aClass.getAnnotation(Stream.class);