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

Refactor register and analysis modules. (#1539) 

* Refactor register and analysis modules.

* Fixed the startup error.
This commit is contained in:
彭勇升 pengys 2018-08-14 14:39:03 +08:00 committed by GitHub
parent 81516a7d56
commit a5ad06ce46
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
51 changed files with 1151 additions and 483 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointLatencyAvgAggregateWorker.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/Const.java
View File

@ -16,23 +16,24 @@
*
*/
package org.apache.skywalking.oap.server.core.analysis.endpoint;
import org.apache.skywalking.oap.server.core.analysis.worker.AbstractAggregatorWorker;
import org.apache.skywalking.oap.server.core.worker.annotation.Worker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
package org.apache.skywalking.oap.server.core;
/**
* @author peng-yongsheng
*/
@Worker
public class EndpointLatencyAvgAggregateWorker extends AbstractAggregatorWorker<EndpointLatencyAvgIndicator> {
public EndpointLatencyAvgAggregateWorker(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public Class nextWorkerClass() {
return EndpointLatencyAvgRemoteWorker.class;
}
public class Const {
public static final int NONE = 0;
public static final String ID_SPLIT = "_";
public static final int NONE_APPLICATION_ID = 1;
public static final int NONE_INSTANCE_ID = 1;
public static final int NONE_SERVICE_ID = 1;
public static final String NONE_SERVICE_NAME = "None";
public static final String USER_CODE = "User";
public static final String SEGMENT_SPAN_SPLIT = "S";
public static final String UNKNOWN = "Unknown";
public static final String EXCEPTION = "Exception";
public static final String EMPTY_STRING = "";
public static final String FILE_SUFFIX = "sw";
public static final int SPAN_TYPE_VIRTUAL = 9;
public static final String DOMAIN_OPERATION_NAME = "{domain}";
}

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

@ -25,7 +25,6 @@ import org.apache.skywalking.oap.server.core.remote.client.RemoteClientManager;
import org.apache.skywalking.oap.server.core.server.*;
import org.apache.skywalking.oap.server.core.source.SourceReceiver;
import org.apache.skywalking.oap.server.core.storage.model.IModelGetter;
import org.apache.skywalking.oap.server.core.worker.annotation.WorkerAnnotationContainer;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
/**
@ -56,7 +55,6 @@ public class CoreModule extends ModuleDefine {
private void addInsideService(List<Class> classes) {
classes.add(IModelGetter.class);
classes.add(StreamDataClassGetter.class);
classes.add(WorkerAnnotationContainer.class);
classes.add(RemoteClientManager.class);
classes.add(RemoteSenderService.class);
}

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

@ -19,6 +19,7 @@
package org.apache.skywalking.oap.server.core;
import java.io.IOException;
import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorTypeListener;
import org.apache.skywalking.oap.server.core.annotation.AnnotationScan;
import org.apache.skywalking.oap.server.core.cluster.*;
import org.apache.skywalking.oap.server.core.remote.*;
@ -28,7 +29,6 @@ import org.apache.skywalking.oap.server.core.server.*;
import org.apache.skywalking.oap.server.core.source.*;
import org.apache.skywalking.oap.server.core.storage.annotation.StorageAnnotationListener;
import org.apache.skywalking.oap.server.core.storage.model.IModelGetter;
import org.apache.skywalking.oap.server.core.worker.annotation.*;
import org.apache.skywalking.oap.server.library.module.*;
import org.apache.skywalking.oap.server.library.server.ServerException;
import org.apache.skywalking.oap.server.library.server.grpc.GRPCServer;
@ -48,9 +48,7 @@ public class CoreModuleProvider extends ModuleProvider {
private final AnnotationScan annotationScan;
private final StorageAnnotationListener storageAnnotationListener;
private final StreamAnnotationListener streamAnnotationListener;
private final WorkerAnnotationListener workerAnnotationListener;
private final StreamDataAnnotationContainer streamDataAnnotationContainer;
private final WorkerAnnotationContainer workerAnnotationContainer;
public CoreModuleProvider() {
super();
@ -58,9 +56,7 @@ public class CoreModuleProvider extends ModuleProvider {
this.annotationScan = new AnnotationScan();
this.storageAnnotationListener = new StorageAnnotationListener();
this.streamAnnotationListener = new StreamAnnotationListener();
this.workerAnnotationListener = new WorkerAnnotationListener();
this.streamDataAnnotationContainer = new StreamDataAnnotationContainer();
this.workerAnnotationContainer = new WorkerAnnotationContainer();
}
@Override public String name() {
@ -85,10 +81,9 @@ public class CoreModuleProvider extends ModuleProvider {
this.registerServiceImplementation(GRPCHandlerRegister.class, new GRPCHandlerRegisterImpl(grpcServer));
this.registerServiceImplementation(JettyHandlerRegister.class, new JettyHandlerRegisterImpl(jettyServer));
this.registerServiceImplementation(SourceReceiver.class, new SourceReceiverImpl(getManager()));
this.registerServiceImplementation(SourceReceiver.class, new SourceReceiverImpl());
this.registerServiceImplementation(StreamDataClassGetter.class, streamDataAnnotationContainer);
this.registerServiceImplementation(WorkerAnnotationContainer.class, workerAnnotationContainer);
this.registerServiceImplementation(RemoteClientManager.class, new RemoteClientManager(getManager()));
this.registerServiceImplementation(RemoteSenderService.class, new RemoteSenderService(getManager()));
@ -96,7 +91,7 @@ public class CoreModuleProvider extends ModuleProvider {
annotationScan.registerListener(storageAnnotationListener);
annotationScan.registerListener(streamAnnotationListener);
annotationScan.registerListener(workerAnnotationListener);
annotationScan.registerListener(new IndicatorTypeListener(getManager()));
}
@Override public void start() throws ModuleStartException {
@ -105,9 +100,8 @@ public class CoreModuleProvider extends ModuleProvider {
try {
annotationScan.scan(() -> {
streamDataAnnotationContainer.generate(streamAnnotationListener.getStreamClasses());
workerAnnotationContainer.load(getManager(), workerAnnotationListener.getWorkerClasses());
});
} catch (WorkerDefineLoadException | IOException e) {
} catch (IOException e) {
throw new ModuleStartException(e.getMessage(), e);
}
}

5
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java
View File

@ -21,7 +21,6 @@ package org.apache.skywalking.oap.server.core.analysis;
import java.util.*;
import org.apache.skywalking.oap.server.core.analysis.endpoint.EndpointDispatcher;
import org.apache.skywalking.oap.server.core.source.Scope;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.slf4j.*;
/**
@ -33,9 +32,9 @@ public class DispatcherManager {
private Map<Scope, SourceDispatcher> dispatcherMap;
public DispatcherManager(ModuleManager moduleManager) {
public DispatcherManager() {
this.dispatcherMap = new HashMap<>();
this.dispatcherMap.put(Scope.Endpoint, new EndpointDispatcher(moduleManager));
this.dispatcherMap.put(Scope.Endpoint, new EndpointDispatcher());
}
public SourceDispatcher getDispatcher(Scope scope) {

18
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointDispatcher.java
View File

@ -18,38 +18,24 @@
package org.apache.skywalking.oap.server.core.analysis.endpoint;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
import org.apache.skywalking.oap.server.core.worker.annotation.WorkerAnnotationContainer;
import org.apache.skywalking.oap.server.core.analysis.worker.IndicatorProcess;
import org.apache.skywalking.oap.server.core.source.Endpoint;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
/**
* @author peng-yongsheng
*/
public class EndpointDispatcher implements SourceDispatcher<Endpoint> {
private final ModuleManager moduleManager;
private EndpointLatencyAvgAggregateWorker avgAggregator;
public EndpointDispatcher(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
@Override public void dispatch(Endpoint source) {
avg(source);
}
private void avg(Endpoint source) {
if (avgAggregator == null) {
WorkerAnnotationContainer workerMapper = moduleManager.find(CoreModule.NAME).getService(WorkerAnnotationContainer.class);
avgAggregator = (EndpointLatencyAvgAggregateWorker)workerMapper.findInstanceByClass(EndpointLatencyAvgAggregateWorker.class);
}
EndpointLatencyAvgIndicator indicator = new EndpointLatencyAvgIndicator();
indicator.setId(source.getId());
indicator.setTimeBucket(source.getTimeBucket());
indicator.combine(source.getLatency(), 1);
avgAggregator.in(indicator);
IndicatorProcess.INSTANCE.in(indicator);
}
}

58
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointLatencyAvgIndicator.java
View File

@ -20,20 +20,21 @@ package org.apache.skywalking.oap.server.core.analysis.endpoint;
import java.util.*;
import lombok.*;
import org.apache.skywalking.oap.server.core.analysis.indicator.*;
import org.apache.skywalking.oap.server.core.analysis.indicator.AvgIndicator;
import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorType;
import org.apache.skywalking.oap.server.core.remote.annotation.StreamData;
import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
import org.apache.skywalking.oap.server.core.storage.annotation.*;
/**
* @author peng-yongsheng
*/
@IndicatorType
@StreamData
@StorageEntity(name = EndpointLatencyAvgIndicator.NAME)
@StorageEntity(name = "endpoint_latency_avg", builder = EndpointLatencyAvgIndicator.Builder.class)
public class EndpointLatencyAvgIndicator extends AvgIndicator {
public static final String NAME = "endpoint_latency_avg";
private static final String ID = "id";
private static final String SERVICE_ID = "service_id";
private static final String SERVICE_INSTANCE_ID = "service_instance_id";
@ -42,10 +43,6 @@ public class EndpointLatencyAvgIndicator extends AvgIndicator {
@Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId;
@Setter @Getter @Column(columnName = SERVICE_INSTANCE_ID) private int serviceInstanceId;
@Override public String name() {
return NAME;
}
@Override public String id() {
return String.valueOf(id);
}
@ -99,27 +96,30 @@ public class EndpointLatencyAvgIndicator extends AvgIndicator {
setValue(remoteData.getDataLongs(2));
}
@Override public Map<String, Object> toMap() {
Map<String, Object> map = new HashMap<>();
map.put(ID, id);
map.put(SERVICE_ID, serviceId);
map.put(SERVICE_INSTANCE_ID, serviceInstanceId);
map.put(COUNT, getCount());
map.put(SUMMATION, getSummation());
map.put(VALUE, getValue());
map.put(TIME_BUCKET, getTimeBucket());
return map;
}
public static class Builder implements StorageBuilder<EndpointLatencyAvgIndicator> {
@Override public Indicator newOne(Map<String, Object> dbMap) {
EndpointLatencyAvgIndicator indicator = new EndpointLatencyAvgIndicator();
indicator.setId((Integer)dbMap.get(ID));
indicator.setServiceId((Integer)dbMap.get(SERVICE_ID));
indicator.setServiceInstanceId((Integer)dbMap.get(SERVICE_INSTANCE_ID));
indicator.setCount((Integer)dbMap.get(COUNT));
indicator.setSummation((Long)dbMap.get(SUMMATION));
indicator.setValue((Long)dbMap.get(VALUE));
indicator.setTimeBucket((Long)dbMap.get(TIME_BUCKET));
return indicator;
@Override public EndpointLatencyAvgIndicator map2Data(Map<String, Object> dbMap) {
EndpointLatencyAvgIndicator indicator = new EndpointLatencyAvgIndicator();
indicator.setId((Integer)dbMap.get(ID));
indicator.setServiceId((Integer)dbMap.get(SERVICE_ID));
indicator.setServiceInstanceId((Integer)dbMap.get(SERVICE_INSTANCE_ID));
indicator.setCount((Integer)dbMap.get(COUNT));
indicator.setSummation((Long)dbMap.get(SUMMATION));
indicator.setValue((Long)dbMap.get(VALUE));
indicator.setTimeBucket((Long)dbMap.get(TIME_BUCKET));
return indicator;
}
@Override public Map<String, Object> data2Map(EndpointLatencyAvgIndicator storageData) {
Map<String, Object> map = new HashMap<>();
map.put(ID, storageData.getId());
map.put(SERVICE_ID, storageData.getServiceId());
map.put(SERVICE_INSTANCE_ID, storageData.getServiceInstanceId());
map.put(COUNT, storageData.getCount());
map.put(SUMMATION, storageData.getSummation());
map.put(VALUE, storageData.getValue());
map.put(TIME_BUCKET, storageData.getTimeBucket());
return map;
}
}
}

43
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointLatencyAvgRemoteWorker.java
View File

@ -1,43 +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.analysis.endpoint;
import org.apache.skywalking.oap.server.core.analysis.worker.AbstractRemoteWorker;
import org.apache.skywalking.oap.server.core.remote.selector.Selector;
import org.apache.skywalking.oap.server.core.worker.annotation.Worker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
/**
* @author peng-yongsheng
*/
@Worker
public class EndpointLatencyAvgRemoteWorker extends AbstractRemoteWorker<EndpointLatencyAvgIndicator> {
public EndpointLatencyAvgRemoteWorker(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public Selector selector() {
return Selector.HashCode;
}
@Override public Class nextWorkerClass() {
return EndpointLatencyAvgPersistentWorker.class;
}
}

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

@ -20,13 +20,12 @@ package org.apache.skywalking.oap.server.core.analysis.indicator;
import lombok.*;
import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.*;
import org.apache.skywalking.oap.server.core.remote.selector.Selector;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
/**
* @author peng-yongsheng
*/
@IndicatorType(selector = Selector.HashCode, needMerge = true)
@IndicatorOperator
public abstract class AvgIndicator extends Indicator {
protected static final String SUMMATION = "summation";

10
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/Indicator.java
View File

@ -18,15 +18,15 @@
package org.apache.skywalking.oap.server.core.analysis.indicator;
import java.util.Map;
import lombok.*;
import org.apache.skywalking.oap.server.core.remote.data.StreamData;
import org.apache.skywalking.oap.server.core.storage.StorageData;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
/**
* @author peng-yongsheng
*/
public abstract class Indicator extends StreamData {
public abstract class Indicator extends StreamData implements StorageData {
protected static final String TIME_BUCKET = "time_bucket";
@ -35,10 +35,4 @@ public abstract class Indicator extends StreamData {
public abstract String id();
public abstract void combine(Indicator indicator);
public abstract String name();
public abstract Map<String, Object> toMap();
public abstract Indicator newOne(Map<String, Object> dbMap);
}

4
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/annotation/Worker.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/annotation/IndicatorOperator.java
View File

@ -16,7 +16,7 @@
*
*/
package org.apache.skywalking.oap.server.core.worker.annotation;
package org.apache.skywalking.oap.server.core.analysis.indicator.annotation;
import java.lang.annotation.*;
@ -25,5 +25,5 @@ import java.lang.annotation.*;
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface Worker {
public @interface IndicatorOperator {
}

6
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/annotation/IndicatorType.java
View File

@ -19,15 +19,11 @@
package org.apache.skywalking.oap.server.core.analysis.indicator.annotation;
import java.lang.annotation.*;
import org.apache.skywalking.oap.server.core.remote.selector.Selector;
/**
* @author peng-yongsheng
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.SOURCE)
@Retention(RetentionPolicy.RUNTIME)
public @interface IndicatorType {
Selector selector();
boolean needMerge();
}

23
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/annotation/WorkerAnnotationListener.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/annotation/IndicatorTypeListener.java
View File

@ -16,34 +16,29 @@
*
*/
package org.apache.skywalking.oap.server.core.worker.annotation;
package org.apache.skywalking.oap.server.core.analysis.indicator.annotation;
import java.lang.annotation.Annotation;
import java.util.*;
import lombok.Getter;
import org.apache.skywalking.oap.server.core.analysis.worker.IndicatorProcess;
import org.apache.skywalking.oap.server.core.annotation.AnnotationListener;
import org.slf4j.*;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
/**
* @author peng-yongsheng
*/
public class WorkerAnnotationListener implements AnnotationListener {
public class IndicatorTypeListener implements AnnotationListener {
private static final Logger logger = LoggerFactory.getLogger(WorkerAnnotationListener.class);
private final ModuleManager moduleManager;
@Getter private final List<Class> workerClasses;
public WorkerAnnotationListener() {
this.workerClasses = new LinkedList<>();
public IndicatorTypeListener(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
@Override public Class<? extends Annotation> annotation() {
return Worker.class;
return IndicatorType.class;
}
@Override public void notify(Class aClass) {
logger.info("The owner class of worker annotation, class name: {}", aClass.getName());
workerClasses.add(aClass);
IndicatorProcess.INSTANCE.create(moduleManager, aClass);
}
}

71
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/AbstractAggregatorWorker.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java
View File

@ -21,44 +21,41 @@ package org.apache.skywalking.oap.server.core.analysis.worker;
import java.util.*;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.analysis.data.*;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.worker.annotation.WorkerAnnotationContainer;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public abstract class AbstractAggregatorWorker<INPUT extends Indicator> extends AbstractWorker<INPUT> {
public class IndicatorAggregateWorker extends AbstractWorker<Indicator> {
private static final Logger logger = LoggerFactory.getLogger(AbstractAggregatorWorker.class);
private static final Logger logger = LoggerFactory.getLogger(IndicatorAggregateWorker.class);
private AbstractWorker worker;
private final ModuleManager moduleManager;
private final DataCarrier<INPUT> dataCarrier;
private final MergeDataCache<INPUT> mergeDataCache;
private AbstractWorker<Indicator> nextWorker;
private final DataCarrier<Indicator> dataCarrier;
private final MergeDataCache<Indicator> mergeDataCache;
private int messageNum;
public AbstractAggregatorWorker(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
IndicatorAggregateWorker(int workerId, AbstractWorker<Indicator> nextWorker) {
super(workerId);
this.nextWorker = nextWorker;
this.mergeDataCache = new MergeDataCache<>();
this.dataCarrier = new DataCarrier<>(1, 10000);
this.dataCarrier.consume(new AggregatorConsumer(this), 1);
}
@Override public final void in(INPUT input) {
input.setEndOfBatchContext(new EndOfBatchContext(false));
dataCarrier.produce(input);
@Override public final void in(Indicator indicator) {
indicator.setEndOfBatchContext(new EndOfBatchContext(false));
dataCarrier.produce(indicator);
}
private void onWork(INPUT message) {
private void onWork(Indicator indicator) {
messageNum++;
aggregate(message);
aggregate(indicator);
if (messageNum >= 1000 || message.getEndOfBatchContext().isEndOfBatch()) {
if (messageNum >= 1000 || indicator.getEndOfBatchContext().isEndOfBatch()) {
sendToNext();
messageNum = 0;
}
@ -74,41 +71,31 @@ public abstract class AbstractAggregatorWorker<INPUT extends Indicator> extends
}
}
mergeDataCache.getLast().collection().forEach((INPUT key, INPUT data) -> {
mergeDataCache.getLast().collection().forEach((Indicator key, Indicator data) -> {
if (logger.isDebugEnabled()) {
logger.debug(data.toString());
}
onNext(data);
nextWorker.in(data);
});
mergeDataCache.finishReadingLast();
}
private void onNext(INPUT data) {
if (worker == null) {
WorkerAnnotationContainer workerMapper = moduleManager.find(CoreModule.NAME).getService(WorkerAnnotationContainer.class);
worker = workerMapper.findInstanceByClass(nextWorkerClass());
}
worker.in(data);
}
public abstract Class nextWorkerClass();
private void aggregate(INPUT message) {
private void aggregate(Indicator indicator) {
mergeDataCache.writing();
if (mergeDataCache.containsKey(message)) {
mergeDataCache.get(message).combine(message);
if (mergeDataCache.containsKey(indicator)) {
mergeDataCache.get(indicator).combine(indicator);
} else {
mergeDataCache.put(message);
mergeDataCache.put(indicator);
}
mergeDataCache.finishWriting();
}
private class AggregatorConsumer implements IConsumer<INPUT> {
private class AggregatorConsumer implements IConsumer<Indicator> {
private final AbstractAggregatorWorker<INPUT> aggregator;
private final IndicatorAggregateWorker aggregator;
private AggregatorConsumer(AbstractAggregatorWorker<INPUT> aggregator) {
private AggregatorConsumer(IndicatorAggregateWorker aggregator) {
this.aggregator = aggregator;
}
@ -116,21 +103,21 @@ public abstract class AbstractAggregatorWorker<INPUT extends Indicator> extends
}
@Override public void consume(List<INPUT> data) {
Iterator<INPUT> inputIterator = data.iterator();
@Override public void consume(List<Indicator> data) {
Iterator<Indicator> inputIterator = data.iterator();
int i = 0;
while (inputIterator.hasNext()) {
INPUT input = inputIterator.next();
Indicator indicator = inputIterator.next();
i++;
if (i == data.size()) {
input.getEndOfBatchContext().setEndOfBatch(true);
indicator.getEndOfBatchContext().setEndOfBatch(true);
}
aggregator.onWork(input);
aggregator.onWork(indicator);
}
}
@Override public void onError(List<INPUT> data, Throwable t) {
@Override public void onError(List<Indicator> data, Throwable t) {
logger.error(t.getMessage(), t);
}

57
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/AbstractPersistentWorker.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java
View File

@ -31,26 +31,31 @@ import static java.util.Objects.nonNull;
/**
* @author peng-yongsheng
*/
public abstract class AbstractPersistentWorker<INPUT extends Indicator> extends AbstractWorker<INPUT> {
public class IndicatorPersistentWorker extends AbstractWorker<Indicator> {
private static final Logger logger = LoggerFactory.getLogger(AbstractPersistentWorker.class);
private static final Logger logger = LoggerFactory.getLogger(IndicatorPersistentWorker.class);
private final MergeDataCache<INPUT> mergeDataCache;
private final String modelName;
private final MergeDataCache<Indicator> mergeDataCache;
private final IBatchDAO batchDAO;
private final IPersistenceDAO<?, ?, INPUT> persistenceDAO;
private final int blockBatchPersistenceSize = 1000;
private final IIndicatorDAO indicatorDAO;
private final int blockBatchPersistenceSize;
public AbstractPersistentWorker(ModuleManager moduleManager) {
IndicatorPersistentWorker(int workerId, String modelName, int batchSize, ModuleManager moduleManager,
IIndicatorDAO indicatorDAO) {
super(workerId);
this.modelName = modelName;
this.blockBatchPersistenceSize = batchSize;
this.mergeDataCache = new MergeDataCache<>();
this.batchDAO = moduleManager.find(StorageModule.NAME).getService(IBatchDAO.class);
this.persistenceDAO = moduleManager.find(StorageModule.NAME).getService(IPersistenceDAO.class);
this.indicatorDAO = indicatorDAO;
}
public final Window<MergeDataCollection<INPUT>> getCache() {
public final Window<MergeDataCollection<Indicator>> getCache() {
return mergeDataCache;
}
@Override public final void in(INPUT input) {
@Override public final void in(Indicator input) {
if (getCache().currentCollectionSize() >= blockBatchPersistenceSize) {
try {
if (getCache().trySwitchPointer()) {
@ -86,33 +91,25 @@ public abstract class AbstractPersistentWorker<INPUT extends Indicator> extends
return batchCollection;
}
private List<Object> prepareBatch(MergeDataCollection<INPUT> collection) {
private List<Object> prepareBatch(MergeDataCollection<Indicator> collection) {
List<Object> batchCollection = new LinkedList<>();
collection.collection().forEach((id, data) -> {
if (needMergeDBData()) {
INPUT dbData = null;
Indicator dbData = null;
try {
dbData = indicatorDAO.get(modelName, data);
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
if (nonNull(dbData)) {
dbData.combine(data);
try {
dbData = persistenceDAO.get(data);
batchCollection.add(indicatorDAO.prepareBatchUpdate(modelName, dbData));
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
if (nonNull(dbData)) {
dbData.combine(data);
try {
batchCollection.add(persistenceDAO.prepareBatchUpdate(dbData));
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
} else {
try {
batchCollection.add(persistenceDAO.prepareBatchInsert(data));
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
}
} else {
try {
batchCollection.add(persistenceDAO.prepareBatchInsert(data));
batchCollection.add(indicatorDAO.prepareBatchInsert(modelName, data));
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
@ -122,7 +119,7 @@ public abstract class AbstractPersistentWorker<INPUT extends Indicator> extends
return batchCollection;
}
private void cacheData(INPUT input) {
private void cacheData(Indicator input) {
mergeDataCache.writing();
if (mergeDataCache.containsKey(input)) {
mergeDataCache.get(input).combine(input);
@ -132,6 +129,4 @@ public abstract class AbstractPersistentWorker<INPUT extends Indicator> extends
mergeDataCache.finishWriting();
}
protected abstract boolean needMergeDBData();
}

64
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorProcess.java Normal file
View File

@ -0,0 +1,64 @@
/*
* 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.worker;
import java.util.*;
import org.apache.skywalking.oap.server.core.UnexpectedException;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.storage.*;
import org.apache.skywalking.oap.server.core.storage.annotation.StorageEntityAnnotationUtils;
import org.apache.skywalking.oap.server.core.worker.*;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
/**
* @author peng-yongsheng
*/
public enum IndicatorProcess {
INSTANCE;
private Map<Class<? extends Indicator>, IndicatorAggregateWorker> entryWorkers = new HashMap<>();
public void in(Indicator indicator) {
entryWorkers.get(indicator.getClass()).in(indicator);
}
public void create(ModuleManager moduleManager, Class<? extends Indicator> indicatorClass) {
String modelName = StorageEntityAnnotationUtils.getModelName(indicatorClass);
Class<? extends StorageBuilder> builderClass = StorageEntityAnnotationUtils.getBuilder(indicatorClass);
StorageDAO storageDAO = moduleManager.find(StorageModule.NAME).getService(StorageDAO.class);
IIndicatorDAO indicatorDAO;
try {
indicatorDAO = storageDAO.newIndicatorDao(builderClass.newInstance());
} catch (InstantiationException | IllegalAccessException e) {
throw new UnexpectedException("");
}
IndicatorPersistentWorker persistentWorker = new IndicatorPersistentWorker(WorkerIdGenerator.INSTANCES.generate(), modelName, 1000, moduleManager, indicatorDAO);
WorkerInstances.INSTANCES.put(persistentWorker.getWorkerId(), persistentWorker);
IndicatorRemoteWorker remoteWorker = new IndicatorRemoteWorker(WorkerIdGenerator.INSTANCES.generate(), moduleManager, persistentWorker);
WorkerInstances.INSTANCES.put(remoteWorker.getWorkerId(), remoteWorker);
IndicatorAggregateWorker aggregateWorker = new IndicatorAggregateWorker(WorkerIdGenerator.INSTANCES.generate(), remoteWorker);
WorkerInstances.INSTANCES.put(aggregateWorker.getWorkerId(), aggregateWorker);
entryWorkers.put(indicatorClass, aggregateWorker);
}
}

32
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/AbstractRemoteWorker.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorRemoteWorker.java
View File

@ -20,7 +20,6 @@ package org.apache.skywalking.oap.server.core.analysis.worker;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.worker.annotation.WorkerAnnotationContainer;
import org.apache.skywalking.oap.server.core.remote.RemoteSenderService;
import org.apache.skywalking.oap.server.core.remote.selector.Selector;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
@ -30,35 +29,24 @@ import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public abstract class AbstractRemoteWorker<INPUT extends Indicator> extends AbstractWorker<INPUT> {
public class IndicatorRemoteWorker extends AbstractWorker<Indicator> {
private static final Logger logger = LoggerFactory.getLogger(AbstractRemoteWorker.class);
private static final Logger logger = LoggerFactory.getLogger(IndicatorRemoteWorker.class);
private final ModuleManager moduleManager;
private RemoteSenderService remoteSender;
private WorkerAnnotationContainer workerMapper;
private final AbstractWorker<Indicator> nextWorker;
private final RemoteSenderService remoteSender;
public AbstractRemoteWorker(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
IndicatorRemoteWorker(int workerId, ModuleManager moduleManager, AbstractWorker<Indicator> nextWorker) {
super(workerId);
this.remoteSender = moduleManager.find(CoreModule.NAME).getService(RemoteSenderService.class);
this.nextWorker = nextWorker;
}
@Override public final void in(INPUT input) {
if (remoteSender == null) {
remoteSender = moduleManager.find(CoreModule.NAME).getService(RemoteSenderService.class);
}
if (workerMapper == null) {
workerMapper = moduleManager.find(CoreModule.NAME).getService(WorkerAnnotationContainer.class);
}
@Override public final void in(Indicator indicator) {
try {
int nextWorkerId = workerMapper.findIdByClass(nextWorkerClass());
remoteSender.send(nextWorkerId, input, selector());
remoteSender.send(nextWorker.getWorkerId(), indicator, Selector.HashCode);
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
}
public abstract Class nextWorkerClass();
public abstract Selector selector();
}

95
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/EndpointCacheService.java vendored Normal file
View File

@ -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.skywalking.oap.server.core.cache;
import com.google.common.cache.*;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.register.endpoint.Endpoint;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.cache.IEndpointCacheDAO;
import org.apache.skywalking.oap.server.library.module.*;
import org.slf4j.*;
import static java.util.Objects.*;
/**
* @author peng-yongsheng
*/
public class EndpointCacheService implements Service {
private static final Logger logger = LoggerFactory.getLogger(EndpointCacheService.class);
private final ModuleManager moduleManager;
private IEndpointCacheDAO cacheDAO;
public EndpointCacheService(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
private final Cache<String, Integer> idCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build();
private final Cache<Integer, Endpoint> sequenceCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build();
public int get(int serviceId, String serviceName, int srcSpanType) {
String id = serviceId + Const.ID_SPLIT + serviceName + Const.ID_SPLIT + srcSpanType;
int endpointId = 0;
try {
endpointId = idCache.get(id, () -> getCacheDAO().get(id));
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
if (serviceId == 0) {
endpointId = getCacheDAO().get(id);
if (endpointId != 0) {
idCache.put(id, endpointId);
}
}
return endpointId;
}
public Endpoint get(int endpointId) {
Endpoint endpoint = null;
try {
endpoint = sequenceCache.get(endpointId, () -> getCacheDAO().get(endpointId));
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
if (isNull(endpoint)) {
endpoint = getCacheDAO().get(endpointId);
if (nonNull(endpoint)) {
sequenceCache.put(endpointId, endpoint);
} else {
logger.warn("Endpoint id {} is not in cache and persistent storage.", endpointId);
}
}
return endpoint;
}
private IEndpointCacheDAO getCacheDAO() {
if (isNull(cacheDAO)) {
cacheDAO = moduleManager.find(StorageModule.NAME).getService(IEndpointCacheDAO.class);
}
return cacheDAO;
}
}

44
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/RegisterSource.java Normal file
View File

@ -0,0 +1,44 @@
/*
* 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.register;
import lombok.*;
import org.apache.skywalking.oap.server.core.remote.data.StreamData;
import org.apache.skywalking.oap.server.core.storage.StorageData;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
/**
* @author peng-yongsheng
*/
public abstract class RegisterSource extends StreamData implements StorageData {
public static final String SEQUENCE = "sequence";
protected static final String REGISTER_TIME = "register_time";
protected static final String HEARTBEAT_TIME = "heartbeat_time";
@Getter @Setter @Column(columnName = SEQUENCE) private int sequence;
@Getter @Setter @Column(columnName = REGISTER_TIME) private long registerTime;
@Getter @Setter @Column(columnName = HEARTBEAT_TIME) private long heartbeatTime;
public final void combine(RegisterSource registerSource) {
if (heartbeatTime < registerSource.getHeartbeatTime()) {
heartbeatTime = registerSource.getHeartbeatTime();
}
}
}

124
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/endpoint/Endpoint.java Normal file
View File

@ -0,0 +1,124 @@
/*
* 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.register.endpoint;
import java.util.*;
import lombok.*;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
import org.apache.skywalking.oap.server.core.remote.annotation.StreamData;
import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
import org.apache.skywalking.oap.server.core.storage.annotation.*;
/**
* @author peng-yongsheng
*/
@StreamData
@StorageEntity(name = "endpoint", builder = Endpoint.Builder.class)
public class Endpoint extends RegisterSource {
private static final String SERVICE_ID = "service_id";
private static final String NAME = "name";
private static final String SRC_SPAN_TYPE = "src_span_type";
@Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId;
@Setter @Getter @Column(columnName = NAME, matchQuery = true) private String name;
@Setter @Getter @Column(columnName = SRC_SPAN_TYPE) private int srcSpanType;
@Override public String id() {
return String.valueOf(serviceId) + Const.ID_SPLIT + name + Const.ID_SPLIT + String.valueOf(srcSpanType);
}
@Override public int hashCode() {
int result = 17;
result = 31 * result + serviceId;
result = 31 * result + name.hashCode();
result = 31 * result + srcSpanType;
return result;
}
@Override public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
Endpoint source = (Endpoint)obj;
if (serviceId != source.getServiceId())
return false;
if (name.equals(source.getName()))
return false;
if (srcSpanType != source.getSrcSpanType())
return false;
return true;
}
@Override public RemoteData.Builder serialize() {
RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
remoteBuilder.setDataIntegers(0, getSequence());
remoteBuilder.setDataIntegers(1, serviceId);
remoteBuilder.setDataIntegers(2, srcSpanType);
remoteBuilder.setDataLongs(0, getRegisterTime());
remoteBuilder.setDataLongs(1, getHeartbeatTime());
remoteBuilder.setDataStrings(0, name);
return remoteBuilder;
}
@Override public void deserialize(RemoteData remoteData) {
setSequence(remoteData.getDataIntegers(0));
setServiceId(remoteData.getDataIntegers(1));
setSrcSpanType(remoteData.getDataIntegers(2));
setRegisterTime(remoteData.getDataLongs(0));
setHeartbeatTime(remoteData.getDataLongs(1));
setName(remoteData.getDataStrings(1));
}
public static class Builder implements StorageBuilder<Endpoint> {
@Override public Endpoint map2Data(Map<String, Object> dbMap) {
Endpoint endpoint = new Endpoint();
endpoint.setSequence((Integer)dbMap.get(SEQUENCE));
endpoint.setServiceId((Integer)dbMap.get(SERVICE_ID));
endpoint.setName((String)dbMap.get(NAME));
endpoint.setSrcSpanType((Integer)dbMap.get(SRC_SPAN_TYPE));
endpoint.setRegisterTime((Long)dbMap.get(REGISTER_TIME));
endpoint.setHeartbeatTime((Long)dbMap.get(HEARTBEAT_TIME));
return endpoint;
}
@Override public Map<String, Object> data2Map(Endpoint storageData) {
Map<String, Object> map = new HashMap<>();
map.put(SEQUENCE, storageData.getSequence());
map.put(SERVICE_ID, storageData.getServiceId());
map.put(NAME, storageData.getName());
map.put(SRC_SPAN_TYPE, storageData.getSrcSpanType());
map.put(REGISTER_TIME, storageData.getRegisterTime());
map.put(HEARTBEAT_TIME, storageData.getHeartbeatTime());
return map;
}
}
}

99
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterDistinctWorker.java Normal file
View File

@ -0,0 +1,99 @@
/*
* 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.register.worker;
import java.util.*;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
import org.apache.skywalking.oap.server.core.analysis.data.EndOfBatchContext;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public class RegisterDistinctWorker extends AbstractWorker<RegisterSource> {
private static final Logger logger = LoggerFactory.getLogger(RegisterDistinctWorker.class);
private final AbstractWorker<RegisterSource> nextWorker;
private final DataCarrier<RegisterSource> dataCarrier;
private final Map<RegisterSource, RegisterSource> sources;
private int messageNum;
public RegisterDistinctWorker(int workerId, AbstractWorker<RegisterSource> nextWorker) {
super(workerId);
this.nextWorker = nextWorker;
this.sources = new HashMap<>();
this.dataCarrier = new DataCarrier<>(1, 10000);
this.dataCarrier.consume(new AggregatorConsumer(this), 1);
}
@Override public final void in(RegisterSource source) {
source.setEndOfBatchContext(new EndOfBatchContext(false));
dataCarrier.produce(source);
}
private void onWork(RegisterSource source) {
messageNum++;
if (!sources.containsKey(source)) {
sources.get(source).combine(source);
}
if (messageNum >= 1000 || source.getEndOfBatchContext().isEndOfBatch()) {
sources.values().forEach(nextWorker::in);
messageNum = 0;
}
}
private class AggregatorConsumer implements IConsumer<RegisterSource> {
private final RegisterDistinctWorker aggregator;
private AggregatorConsumer(RegisterDistinctWorker aggregator) {
this.aggregator = aggregator;
}
@Override public void init() {
}
@Override public void consume(List<RegisterSource> sources) {
Iterator<RegisterSource> sourceIterator = sources.iterator();
int i = 0;
while (sourceIterator.hasNext()) {
RegisterSource source = sourceIterator.next();
i++;
if (i == sources.size()) {
source.getEndOfBatchContext().setEndOfBatch(true);
}
aggregator.onWork(source);
}
}
@Override public void onError(List<RegisterSource> sources, Throwable t) {
logger.error(t.getMessage(), t);
}
@Override public void onExit() {
}
}
}

81
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java Normal file
View File

@ -0,0 +1,81 @@
/*
* 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.register.worker;
import java.util.*;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
import org.apache.skywalking.oap.server.core.source.Scope;
import org.apache.skywalking.oap.server.core.storage.*;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public class RegisterPersistentWorker extends AbstractWorker<RegisterSource> {
private static final Logger logger = LoggerFactory.getLogger(RegisterPersistentWorker.class);
private final Scope scope;
private final String modelName;
private final Map<RegisterSource, RegisterSource> sources;
private final IRegisterLockDAO registerLockDAO;
private final IRegisterDAO registerDAO;
public RegisterPersistentWorker(int workerId, String modelName, ModuleManager moduleManager,
IRegisterDAO registerDAO, Scope scope) {
super(workerId);
this.modelName = modelName;
this.sources = new HashMap<>();
this.registerDAO = registerDAO;
this.registerLockDAO = moduleManager.find(StorageModule.NAME).getService(IRegisterLockDAO.class);
this.scope = scope;
}
@Override public final void in(RegisterSource registerSource) {
if (!sources.containsKey(registerSource)) {
sources.put(registerSource, registerSource);
}
if (registerSource.getEndOfBatchContext().isEndOfBatch()) {
if (registerLockDAO.tryLock(scope)) {
try {
sources.values().forEach(source -> {
try {
RegisterSource newSource = registerDAO.get(modelName, registerSource.id());
if (Objects.nonNull(newSource)) {
newSource.combine(newSource);
int sequence = registerDAO.max(modelName);
newSource.setSequence(sequence);
registerDAO.forceInsert(modelName, newSource);
} else {
registerDAO.forceUpdate(modelName, newSource);
}
} catch (Throwable t) {
logger.error(t.getMessage());
}
});
} finally {
registerLockDAO.releaseLock(scope);
}
}
}
}
}

52
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterRemoteWorker.java Normal file
View File

@ -0,0 +1,52 @@
/*
* 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.register.worker;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
import org.apache.skywalking.oap.server.core.remote.RemoteSenderService;
import org.apache.skywalking.oap.server.core.remote.selector.Selector;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public class RegisterRemoteWorker extends AbstractWorker<RegisterSource> {
private static final Logger logger = LoggerFactory.getLogger(RegisterRemoteWorker.class);
private final AbstractWorker<RegisterSource> nextWorker;
private final RemoteSenderService remoteSender;
RegisterRemoteWorker(int workerId, ModuleManager moduleManager, AbstractWorker<RegisterSource> nextWorker) {
super(workerId);
this.remoteSender = moduleManager.find(CoreModule.NAME).getService(RemoteSenderService.class);
this.nextWorker = nextWorker;
}
@Override public final void in(RegisterSource indicator) {
try {
remoteSender.send(nextWorker.getWorkerId(), indicator, Selector.ForeverFirst);
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
}
}

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

@ -24,7 +24,7 @@ import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataClassGetter;
import org.apache.skywalking.oap.server.core.remote.data.StreamData;
import org.apache.skywalking.oap.server.core.remote.grpc.proto.*;
import org.apache.skywalking.oap.server.core.worker.annotation.WorkerClassGetter;
import org.apache.skywalking.oap.server.core.worker.WorkerInstances;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler;
import org.slf4j.*;
@ -38,7 +38,6 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
private final ModuleManager moduleManager;
private StreamDataClassGetter streamDataClassGetter;
private WorkerClassGetter workerClassGetter;
public RemoteServiceHandler(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
@ -48,9 +47,6 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
if (Objects.isNull(streamDataClassGetter)) {
streamDataClassGetter = moduleManager.find(CoreModule.NAME).getService(StreamDataClassGetter.class);
}
if (Objects.isNull(streamDataClassGetter)) {
workerClassGetter = moduleManager.find(CoreModule.NAME).getService(WorkerClassGetter.class);
}
return new StreamObserver<RemoteMessage>() {
@Override public void onNext(RemoteMessage message) {
@ -62,7 +58,7 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
try {
StreamData streamData = streamDataClass.newInstance();
streamData.deserialize(remoteData);
workerClassGetter.getClassById(nextWorkerId).newInstance().in(streamData);
WorkerInstances.INSTANCES.get(nextWorkerId).in(streamData);
} catch (InstantiationException | IllegalAccessException e) {
logger.warn(e.getMessage());
}

4
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java
View File

@ -20,8 +20,8 @@ package org.apache.skywalking.oap.server.core.remote.client;
import java.util.*;
import java.util.concurrent.*;
import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataAnnotationContainer;
import org.apache.skywalking.oap.server.core.cluster.*;
import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataAnnotationContainer;
import org.apache.skywalking.oap.server.library.module.*;
import org.slf4j.*;
@ -96,7 +96,7 @@ public class RemoteClientManager implements Service {
client = currentClientsMap.get(address);
} else {
if (remoteInstance.isSelf()) {
client = new SelfRemoteClient(moduleManager, remoteInstance.getHost(), remoteInstance.getPort());
client = new SelfRemoteClient(remoteInstance.getHost(), remoteInstance.getPort());
} else {
client = new GRPCRemoteClient(indicatorMapper, remoteInstance, 1, 3000);
}

11
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java
View File

@ -18,22 +18,18 @@
package org.apache.skywalking.oap.server.core.remote.client;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.remote.data.StreamData;
import org.apache.skywalking.oap.server.core.worker.annotation.WorkerAnnotationContainer;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.core.worker.WorkerInstances;
/**
* @author peng-yongsheng
*/
public class SelfRemoteClient implements RemoteClient {
private final ModuleManager moduleManager;
private final String host;
private final int port;
public SelfRemoteClient(ModuleManager moduleManager, String host, int port) {
this.moduleManager = moduleManager;
public SelfRemoteClient(String host, int port) {
this.host = host;
this.port = port;
}
@ -47,7 +43,6 @@ public class SelfRemoteClient implements RemoteClient {
}
@Override public void push(int nextWorkerId, StreamData streamData) {
WorkerAnnotationContainer workerMapper = moduleManager.find(CoreModule.NAME).getService(WorkerAnnotationContainer.class);
workerMapper.findInstanceById(nextWorkerId).in(streamData);
WorkerInstances.INSTANCES.get(nextWorkerId).in(streamData);
}
}

5
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/SourceReceiverImpl.java
View File

@ -19,7 +19,6 @@
package org.apache.skywalking.oap.server.core.source;
import org.apache.skywalking.oap.server.core.analysis.DispatcherManager;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
/**
* @author peng-yongsheng
@ -28,8 +27,8 @@ public class SourceReceiverImpl implements SourceReceiver {
private final DispatcherManager dispatcherManager;
public SourceReceiverImpl(ModuleManager moduleManager) {
this.dispatcherManager = new DispatcherManager(moduleManager);
public SourceReceiverImpl() {
this.dispatcherManager = new DispatcherManager();
}
@Override public void receive(Source source) {

10
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IPersistenceDAO.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IIndicatorDAO.java
View File

@ -24,13 +24,13 @@ import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
/**
* @author peng-yongsheng
*/
public interface IPersistenceDAO<INSERT, UPDATE, INPUT extends Indicator> extends DAO {
public interface IIndicatorDAO<INSERT, UPDATE> extends DAO {
INPUT get(INPUT input) throws IOException;
Indicator get(String modelName, Indicator indicator) throws IOException;
INSERT prepareBatchInsert(INPUT input) throws IOException;
INSERT prepareBatchInsert(String modelName, Indicator indicator) throws IOException;
UPDATE prepareBatchUpdate(INPUT input) throws IOException;
UPDATE prepareBatchUpdate(String modelName, Indicator indicator) throws IOException;
void deleteHistory(Long timeBucketBefore);
void deleteHistory(String modelName, Long timeBucketBefore);
}

36
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRegisterDAO.java Normal file
View File

@ -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.skywalking.oap.server.core.storage;
import java.io.IOException;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
/**
* @author peng-yongsheng
*/
public interface IRegisterDAO extends DAO {
int max(String modelName) throws IOException;
RegisterSource get(String modelName, String id) throws IOException;
void forceInsert(String modelName, RegisterSource source) throws IOException;
void forceUpdate(String modelName, RegisterSource source) throws IOException;
}

31
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageBuilder.java Normal file
View File

@ -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.skywalking.oap.server.core.storage;
import java.util.Map;
/**
* @author peng-yongsheng
*/
public interface StorageBuilder<T extends StorageData> {
T map2Data(Map<String, Object> dbMap);
Map<String, Object> data2Map(T storageData);
}

9
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/annotation/WorkerClassGetter.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java
View File

@ -16,14 +16,15 @@
*
*/
package org.apache.skywalking.oap.server.core.worker.annotation;
package org.apache.skywalking.oap.server.core.storage;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.library.module.Service;
/**
* @author peng-yongsheng
*/
public interface WorkerClassGetter extends Service {
Class<AbstractWorker> getClassById(int workerId);
public interface StorageDAO extends Service {
IIndicatorDAO newIndicatorDao(StorageBuilder<Indicator> storageBuilder);
}

26
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageData.java Normal file
View File

@ -0,0 +1,26 @@
/*
* 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;
/**
* @author peng-yongsheng
*/
public interface StorageData {
String id();
}

2
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java
View File

@ -32,6 +32,6 @@ public class StorageModule extends ModuleDefine {
}
@Override public Class[] services() {
return new Class[] {IBatchDAO.class, IPersistenceDAO.class, IRegisterLockDAO.class};
return new Class[] {IBatchDAO.class, StorageDAO.class, IRegisterLockDAO.class};
}
}

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

@ -23,8 +23,12 @@ import java.lang.annotation.*;
/**
* @author peng-yongsheng
*/
@Target(ElementType.FIELD)
@Target({ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
public @interface Column {
String columnName();
boolean matchQuery() default false;
boolean termQuery() default true;
}

26
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/Query.java Normal file
View File

@ -0,0 +1,26 @@
/*
* 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.annotation;
/**
* @author peng-yongsheng
*/
public enum Query {
Term, Match
}

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

@ -49,8 +49,8 @@ public class StorageAnnotationListener implements AnnotationListener, IModelGett
List<ModelColumn> modelColumns = new LinkedList<>();
retrieval(aClass, modelColumns);
StorageEntity annotation = (StorageEntity)aClass.getAnnotation(StorageEntity.class);
models.add(new Model(annotation.name(), modelColumns));
String modelName = StorageEntityAnnotationUtils.getModelName(aClass);
models.add(new Model(modelName, modelColumns));
}
private void retrieval(Class clazz, List<ModelColumn> modelColumns) {

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

@ -19,6 +19,7 @@
package org.apache.skywalking.oap.server.core.storage.annotation;
import java.lang.annotation.*;
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
/**
* @author peng-yongsheng
@ -27,4 +28,6 @@ import java.lang.annotation.*;
@Retention(RetentionPolicy.RUNTIME)
public @interface StorageEntity {
String name();
Class<? extends StorageBuilder> builder();
}

46
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/StorageEntityAnnotationUtils.java Normal file
View File

@ -0,0 +1,46 @@
/*
* 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.annotation;
import org.apache.skywalking.oap.server.core.UnexpectedException;
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
/**
* @author peng-yongsheng
*/
public class StorageEntityAnnotationUtils {
public static String getModelName(Class aClass) {
if (aClass.isAnnotationPresent(StorageEntity.class)) {
StorageEntity annotation = (StorageEntity)aClass.getAnnotation(StorageEntity.class);
return annotation.name();
} else {
throw new UnexpectedException("");
}
}
public static Class<? extends StorageBuilder> getBuilder(Class aClass) {
if (aClass.isAnnotationPresent(StorageEntity.class)) {
StorageEntity annotation = (StorageEntity)aClass.getAnnotation(StorageEntity.class);
return annotation.builder();
} else {
throw new UnexpectedException("");
}
}
}

32
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/cache/IEndpointCacheDAO.java vendored Normal file
View File

@ -0,0 +1,32 @@
/*
* 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.cache;
import org.apache.skywalking.oap.server.core.register.endpoint.Endpoint;
import org.apache.skywalking.oap.server.core.storage.DAO;
/**
* @author peng-yongsheng
*/
public interface IEndpointCacheDAO extends DAO {
int get(String id);
Endpoint get(int sequence);
}

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

@ -18,10 +18,18 @@
package org.apache.skywalking.oap.server.core.worker;
import lombok.Getter;
/**
* @author peng-yongsheng
*/
public abstract class AbstractWorker<INPUT> {
@Getter private final int workerId;
public AbstractWorker(int workerId) {
this.workerId = workerId;
}
public abstract void in(INPUT input);
}

11
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/annotation/WorkerDefineLoadException.java → oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/WorkerIdGenerator.java
View File

@ -16,14 +16,17 @@
*
*/
package org.apache.skywalking.oap.server.core.worker.annotation;
package org.apache.skywalking.oap.server.core.worker;
/**
* @author peng-yongsheng
*/
public class WorkerDefineLoadException extends RuntimeException {
public enum WorkerIdGenerator {
INSTANCES;
public WorkerDefineLoadException(String message, Throwable cause) {
super(message, cause);
private int workerId = 0;
public synchronized int generate() {
return workerId++;
}
}

38
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/WorkerInstances.java Normal file
View File

@ -0,0 +1,38 @@
/*
* 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.worker;
import java.util.*;
/**
* @author peng-yongsheng
*/
public enum WorkerInstances {
INSTANCES;
private Map<Integer, AbstractWorker> instances = new HashMap<>();
public void put(int workerId, AbstractWorker instance) {
instances.put(workerId, instance);
}
public AbstractWorker get(int workerId) {
return instances.get(workerId);
}
}

84
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/annotation/WorkerAnnotationContainer.java
View File

@ -1,84 +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.worker.annotation;
import java.lang.reflect.Constructor;
import java.util.*;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public class WorkerAnnotationContainer implements WorkerClassGetter {
private static final Logger logger = LoggerFactory.getLogger(WorkerAnnotationContainer.class);
private int id = 0;
private final Map<Class<AbstractWorker>, Integer> classKeyMapping;
private final Map<Integer, Class<AbstractWorker>> idKeyMapping;
private final Map<Class<AbstractWorker>, AbstractWorker> classKeyInstanceMapping;
private final Map<Integer, AbstractWorker> idKeyInstanceMapping;
public WorkerAnnotationContainer() {
this.classKeyMapping = new HashMap<>();
this.idKeyMapping = new HashMap<>();
this.classKeyInstanceMapping = new HashMap<>();
this.idKeyInstanceMapping = new HashMap<>();
}
@SuppressWarnings(value = "unchecked")
public void load(ModuleManager moduleManager, List<Class> workerClasses) throws WorkerDefineLoadException {
if (Objects.isNull(workerClasses)) {
return;
}
try {
for (Class workerClass : workerClasses) {
id++;
classKeyMapping.put(workerClass, id);
idKeyMapping.put(id, workerClass);
Constructor<AbstractWorker> constructor = workerClass.getDeclaredConstructor(ModuleManager.class);
AbstractWorker worker = constructor.newInstance(moduleManager);
classKeyInstanceMapping.put(workerClass, worker);
idKeyInstanceMapping.put(id, worker);
}
} catch (Throwable t) {
throw new WorkerDefineLoadException(t.getMessage(), t);
}
}
@Override public Class<AbstractWorker> getClassById(int workerId) {
return idKeyMapping.get(id);
}
public int findIdByClass(Class workerClass) {
return classKeyMapping.get(workerClass);
}
public AbstractWorker findInstanceByClass(Class workerClass) {
return classKeyInstanceMapping.get(workerClass);
}
public AbstractWorker findInstanceById(int id) {
return idKeyInstanceMapping.get(id);
}
}

15
oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/indicator/define/TestAvgIndicator.java
View File

@ -18,9 +18,8 @@
package org.apache.skywalking.oap.server.core.analysis.indicator.define;
import java.util.Map;
import lombok.*;
import org.apache.skywalking.oap.server.core.analysis.indicator.*;
import org.apache.skywalking.oap.server.core.analysis.indicator.AvgIndicator;
import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
/**
@ -34,22 +33,10 @@ public class TestAvgIndicator extends AvgIndicator {
return null;
}
@Override public String name() {
return null;
}
@Override public void deserialize(RemoteData remoteData) {
}
@Override public String id() {
return null;
}
@Override public Map<String, Object> toMap() {
return null;
}
@Override public Indicator newOne(Map<String, Object> dbMap) {
return null;
}
}

19
oap-server/server-core/src/test/resources/META-INF/defines/indicator.def
View File

@ -1,19 +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.
#
#
org.apache.skywalking.oap.server.core.analysis.indicator.define.TestAvgIndicator

17
oap-server/server-core/src/test/resources/META-INF/defines/worker.def
View File

@ -1,17 +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.
#
#

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

@ -63,7 +63,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
elasticSearchClient = new ElasticSearchClient(config.getClusterNodes(), nameSpace);
this.registerServiceImplementation(IBatchDAO.class, new BatchProcessEsDAO(elasticSearchClient, config.getBulkActions(), config.getBulkSize(), config.getFlushInterval(), config.getConcurrentRequests()));
this.registerServiceImplementation(IPersistenceDAO.class, new PersistenceEsDAO(elasticSearchClient, nameSpace));
this.registerServiceImplementation(StorageDAO.class, new StorageEsDAO(elasticSearchClient));
this.registerServiceImplementation(IRegisterLockDAO.class, new RegisterLockDAOImpl(elasticSearchClient, 1000));
}

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

@ -18,54 +18,15 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
import java.io.IOException;
import org.apache.skywalking.oap.server.core.storage.AbstractDAO;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.max.Max;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public abstract class EsDAO extends AbstractDAO<ElasticSearchClient> {
private static final Logger logger = LoggerFactory.getLogger(EsDAO.class);
public EsDAO(ElasticSearchClient client) {
super(client);
}
protected final int getMaxId(String indexName, String columnName) {
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.aggregation(AggregationBuilders.max("agg").field(columnName));
searchSourceBuilder.size(0);
return getResponse(indexName, searchSourceBuilder);
}
protected final int getMinId(String indexName, String columnName) {
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.aggregation(AggregationBuilders.min("agg").field(columnName));
searchSourceBuilder.size(0);
return getResponse(indexName, searchSourceBuilder);
}
private int getResponse(String indexName, SearchSourceBuilder searchSourceBuilder) {
try {
SearchResponse searchResponse = getClient().search(indexName, searchSourceBuilder);
Max agg = searchResponse.getAggregations().get("agg");
int id = (int)agg.getValue();
if (id == Integer.MAX_VALUE || id == Integer.MIN_VALUE) {
return 0;
} else {
return id;
}
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
return 0;
}
}

35
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/PersistenceEsDAO.java → oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/IndicatorEsDAO.java
View File

@ -21,8 +21,7 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
import java.io.IOException;
import java.util.Map;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.storage.IPersistenceDAO;
import org.apache.skywalking.oap.server.library.client.NameSpace;
import org.apache.skywalking.oap.server.core.storage.*;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
@ -32,48 +31,46 @@ import org.elasticsearch.common.xcontent.*;
/**
* @author peng-yongsheng
*/
public class PersistenceEsDAO implements IPersistenceDAO<IndexRequest, UpdateRequest, Indicator> {
public class IndicatorEsDAO extends EsDAO implements IIndicatorDAO<IndexRequest, UpdateRequest> {
private final ElasticSearchClient client;
private final NameSpace nameSpace;
private final StorageBuilder<Indicator> storageBuilder;
public PersistenceEsDAO(ElasticSearchClient client, NameSpace nameSpace) {
this.client = client;
this.nameSpace = nameSpace;
public IndicatorEsDAO(ElasticSearchClient client, StorageBuilder<Indicator> storageBuilder) {
super(client);
this.storageBuilder = storageBuilder;
}
@Override public Indicator get(Indicator input) throws IOException {
GetResponse response = client.get(nameSpace.getNameSpace() + "_" + input.name(), input.id());
@Override public Indicator get(String modelName, Indicator indicator) throws IOException {
GetResponse response = getClient().get(modelName, indicator.id());
if (response.isExists()) {
return input.newOne(response.getSource());
return storageBuilder.map2Data(response.getSource());
} else {
return null;
}
}
@Override public IndexRequest prepareBatchInsert(Indicator input) throws IOException {
Map<String, Object> objectMap = input.toMap();
@Override public IndexRequest prepareBatchInsert(String modelName, Indicator indicator) throws IOException {
Map<String, Object> objectMap = storageBuilder.data2Map(indicator);
XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
for (String key : objectMap.keySet()) {
builder.field(key, objectMap.get(key));
}
builder.endObject();
return client.prepareInsert(nameSpace.getNameSpace() + "_" + input.name(), input.id(), builder);
return getClient().prepareInsert(modelName, indicator.id(), builder);
}
@Override public UpdateRequest prepareBatchUpdate(Indicator input) throws IOException {
Map<String, Object> objectMap = input.toMap();
@Override public UpdateRequest prepareBatchUpdate(String modelName, Indicator indicator) throws IOException {
Map<String, Object> objectMap = storageBuilder.data2Map(indicator);
XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
for (String key : objectMap.keySet()) {
builder.field(key, objectMap.get(key));
}
builder.endObject();
return client.prepareUpdate(nameSpace.getNameSpace() + "_" + input.name(), input.id(), builder);
return getClient().prepareUpdate(modelName, indicator.id(), builder);
}
@Override public void deleteHistory(Long timeBucketBefore) {
@Override public void deleteHistory(String modelName, Long timeBucketBefore) {
}
}

99
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RegisterEsDAO.java Normal file
View File

@ -0,0 +1,99 @@
/*
* 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.storage.plugin.elasticsearch.base;
import java.io.IOException;
import java.util.Map;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
import org.apache.skywalking.oap.server.core.storage.*;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.xcontent.*;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.max.Max;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public class RegisterEsDAO extends EsDAO implements IRegisterDAO {
private static final Logger logger = LoggerFactory.getLogger(RegisterEsDAO.class);
private final StorageBuilder<RegisterSource> storageBuilder;
public RegisterEsDAO(ElasticSearchClient client, StorageBuilder<RegisterSource> storageBuilder) {
super(client);
this.storageBuilder = storageBuilder;
}
@Override public RegisterSource get(String modelName, String id) throws IOException {
GetResponse response = getClient().get(modelName, id);
if (response.isExists()) {
return storageBuilder.map2Data(response.getSource());
} else {
return null;
}
}
@Override public void forceInsert(String modelName, RegisterSource source) throws IOException {
Map<String, Object> objectMap = storageBuilder.data2Map(source);
XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
for (String key : objectMap.keySet()) {
builder.field(key, objectMap.get(key));
}
builder.endObject();
getClient().forceInsert(modelName, source.id(), builder);
}
@Override public void forceUpdate(String modelName, RegisterSource source) throws IOException {
Map<String, Object> objectMap = storageBuilder.data2Map(source);
XContentBuilder builder = XContentFactory.jsonBuilder().startObject();
for (String key : objectMap.keySet()) {
builder.field(key, objectMap.get(key));
}
builder.endObject();
getClient().forceUpdate(modelName, source.id(), builder);
}
@Override public int max(String modelName) throws IOException {
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.aggregation(AggregationBuilders.max(RegisterSource.SEQUENCE).field(RegisterSource.SEQUENCE));
searchSourceBuilder.size(0);
return getResponse(modelName, searchSourceBuilder);
}
private int getResponse(String modelName, SearchSourceBuilder searchSourceBuilder) throws IOException {
SearchResponse searchResponse = getClient().search(modelName, searchSourceBuilder);
Max agg = searchResponse.getAggregations().get(RegisterSource.SEQUENCE);
int id = (int)agg.getValue();
if (id == Integer.MAX_VALUE || id == Integer.MIN_VALUE) {
return 0;
} else {
return id;
}
}
}

19
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointLatencyAvgPersistentWorker.java → oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java
View File

@ -16,23 +16,22 @@
*
*/
package org.apache.skywalking.oap.server.core.analysis.endpoint;
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
import org.apache.skywalking.oap.server.core.analysis.worker.AbstractPersistentWorker;
import org.apache.skywalking.oap.server.core.worker.annotation.Worker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.storage.*;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
/**
* @author peng-yongsheng
*/
@Worker
public class EndpointLatencyAvgPersistentWorker extends AbstractPersistentWorker<EndpointLatencyAvgIndicator> {
public class StorageEsDAO extends EsDAO implements StorageDAO {
public EndpointLatencyAvgPersistentWorker(ModuleManager moduleManager) {
super(moduleManager);
public StorageEsDAO(ElasticSearchClient client) {
super(client);
}
@Override protected boolean needMergeDBData() {
return true;
@Override public IIndicatorDAO newIndicatorDao(StorageBuilder<Indicator> storageBuilder) {
return new IndicatorEsDAO(getClient(), storageBuilder);
}
}

53
oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/EndpointCacheEsDAO.java vendored Normal file
View File

@ -0,0 +1,53 @@
/*
* 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.storage.plugin.elasticsearch.cache;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
import org.apache.skywalking.oap.server.core.register.endpoint.Endpoint;
import org.apache.skywalking.oap.server.core.storage.cache.IEndpointCacheDAO;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
import org.elasticsearch.action.get.GetResponse;
/**
* @author peng-yongsheng
*/
public class EndpointCacheEsDAO extends EsDAO implements IEndpointCacheDAO {
public EndpointCacheEsDAO(ElasticSearchClient client) {
super(client);
}
@Override public int get(String id) {
try {
GetResponse response = getClient().get("", id);
if (response.isExists()) {
return response.getField(RegisterSource.SEQUENCE).getValue();
} else {
return 0;
}
} catch (Throwable e) {
return 0;
}
}
@Override public Endpoint get(int sequence) {
return null;
}
}