diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java index 8e3d137b66..340ce33e44 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java @@ -22,6 +22,7 @@ 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.register.annotation.InventoryTypeListener; import org.apache.skywalking.oap.server.core.remote.*; import org.apache.skywalking.oap.server.core.remote.annotation.*; import org.apache.skywalking.oap.server.core.remote.client.RemoteClientManager; @@ -92,6 +93,7 @@ public class CoreModuleProvider extends ModuleProvider { annotationScan.registerListener(storageAnnotationListener); annotationScan.registerListener(streamAnnotationListener); annotationScan.registerListener(new IndicatorTypeListener(getManager())); + annotationScan.registerListener(new InventoryTypeListener(getManager())); } @Override public void start() throws ModuleStartException { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java index 4c0ab0f493..3579ad8bf3 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java @@ -19,7 +19,7 @@ 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.analysis.generated.endpoint.EndpointDispatcher; import org.apache.skywalking.oap.server.core.source.Scope; import org.slf4j.*; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointLatencyAvgIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointLatencyAvgIndicator.java deleted file mode 100644 index 718d357c10..0000000000 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointLatencyAvgIndicator.java +++ /dev/null @@ -1,125 +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 java.util.*; -import lombok.*; -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 = "endpoint_latency_avg", builder = EndpointLatencyAvgIndicator.Builder.class) -public class EndpointLatencyAvgIndicator extends AvgIndicator { - - private static final String ID = "id"; - private static final String SERVICE_ID = "service_id"; - private static final String SERVICE_INSTANCE_ID = "service_instance_id"; - - @Setter @Getter @Column(columnName = ID) private int id; - @Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId; - @Setter @Getter @Column(columnName = SERVICE_INSTANCE_ID) private int serviceInstanceId; - - @Override public String id() { - return String.valueOf(id); - } - - @Override public int hashCode() { - int result = 17; - result = 31 * result + id; - result = 31 * result + (int)getTimeBucket(); - return result; - } - - @Override public boolean equals(Object obj) { - if (this == obj) - return true; - if (obj == null) - return false; - if (getClass() != obj.getClass()) - return false; - - EndpointLatencyAvgIndicator indicator = (EndpointLatencyAvgIndicator)obj; - if (id != indicator.id) - return false; - if (getTimeBucket() != indicator.getTimeBucket()) - return false; - - return true; - } - - @Override public RemoteData.Builder serialize() { - RemoteData.Builder remoteBuilder = RemoteData.newBuilder(); - remoteBuilder.setDataIntegers(0, getId()); - remoteBuilder.setDataIntegers(1, getServiceId()); - remoteBuilder.setDataIntegers(2, getServiceInstanceId()); - remoteBuilder.setDataIntegers(3, getCount()); - - remoteBuilder.setDataLongs(0, getTimeBucket()); - remoteBuilder.setDataLongs(1, getSummation()); - remoteBuilder.setDataLongs(2, getValue()); - - return remoteBuilder; - } - - @Override public void deserialize(RemoteData remoteData) { - setId(remoteData.getDataIntegers(0)); - setServiceId(remoteData.getDataIntegers(1)); - setServiceInstanceId(remoteData.getDataIntegers(2)); - setCount(remoteData.getDataIntegers(3)); - - setTimeBucket(remoteData.getDataLongs(0)); - setSummation(remoteData.getDataLongs(1)); - setValue(remoteData.getDataLongs(2)); - } - - public static class Builder implements StorageBuilder { - - @Override public EndpointLatencyAvgIndicator map2Data(Map 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 data2Map(EndpointLatencyAvgIndicator storageData) { - Map 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; - } - } -} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/EndpointInventoryCacheService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/EndpointInventoryCacheService.java index 05e43d8f99..5923b520a5 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/EndpointInventoryCacheService.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/EndpointInventoryCacheService.java @@ -42,9 +42,9 @@ public class EndpointInventoryCacheService implements Service { this.moduleManager = moduleManager; } - private final Cache idCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build(); + private final Cache idCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(100000).build(); - private final Cache sequenceCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build(); + private final Cache sequenceCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(100000).build(); public int get(int serviceId, String serviceName, int srcSpanType) { String id = serviceId + Const.ID_SPLIT + serviceName + Const.ID_SPLIT + srcSpanType; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/EndpointInventory.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/EndpointInventory.java index d54363c76a..efdc5bf5d3 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/EndpointInventory.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/EndpointInventory.java @@ -21,14 +21,17 @@ package org.apache.skywalking.oap.server.core.register; import java.util.*; import lombok.*; import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.register.annotation.InventoryType; 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.source.Scope; import org.apache.skywalking.oap.server.core.storage.StorageBuilder; import org.apache.skywalking.oap.server.core.storage.annotation.*; /** * @author peng-yongsheng */ +@InventoryType(scope = Scope.Endpoint) @StreamData @StorageEntity(name = "endpoint_inventory", builder = EndpointInventory.Builder.class) public class EndpointInventory extends RegisterSource { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryAnnotationUtils.java similarity index 53% rename from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointDispatcher.java rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryAnnotationUtils.java index 9bf8838aa5..6afdc87c80 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/endpoint/EndpointDispatcher.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryAnnotationUtils.java @@ -16,26 +16,22 @@ * */ -package org.apache.skywalking.oap.server.core.analysis.endpoint; +package org.apache.skywalking.oap.server.core.register.annotation; -import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher; -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.core.UnexpectedException; +import org.apache.skywalking.oap.server.core.source.Scope; /** * @author peng-yongsheng */ -public class EndpointDispatcher implements SourceDispatcher { +public class InventoryAnnotationUtils { - @Override public void dispatch(Endpoint source) { - avg(source); - } - - private void avg(Endpoint source) { - EndpointLatencyAvgIndicator indicator = new EndpointLatencyAvgIndicator(); - indicator.setId(source.getId()); - indicator.setTimeBucket(source.getTimeBucket()); - indicator.combine(source.getLatency(), 1); - IndicatorProcess.INSTANCE.in(indicator); + public static Scope getScope(Class aClass) { + if (aClass.isAnnotationPresent(InventoryType.class)) { + InventoryType annotation = (InventoryType)aClass.getAnnotation(InventoryType.class); + return annotation.scope(); + } else { + throw new UnexpectedException(""); + } } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryType.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryType.java new file mode 100644 index 0000000000..d0f9a7faf0 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryType.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.core.register.annotation; + +import java.lang.annotation.*; +import org.apache.skywalking.oap.server.core.source.Scope; + +/** + * @author peng-yongsheng + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +public @interface InventoryType { + Scope scope(); +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryTypeListener.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryTypeListener.java new file mode 100644 index 0000000000..a31aee52d2 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/annotation/InventoryTypeListener.java @@ -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.annotation; + +import java.lang.annotation.Annotation; +import org.apache.skywalking.oap.server.core.annotation.AnnotationListener; +import org.apache.skywalking.oap.server.core.register.worker.InventoryProcess; +import org.apache.skywalking.oap.server.library.module.ModuleManager; + +/** + * @author peng-yongsheng + */ +public class InventoryTypeListener implements AnnotationListener { + + private final ModuleManager moduleManager; + + public InventoryTypeListener(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + @Override public Class annotation() { + return InventoryType.class; + } + + @Override public void notify(Class aClass) { + InventoryProcess.INSTANCE.create(moduleManager, aClass); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryProcess.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryProcess.java new file mode 100644 index 0000000000..31a07770b5 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryProcess.java @@ -0,0 +1,68 @@ +/* + * 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.UnexpectedException; +import org.apache.skywalking.oap.server.core.register.RegisterSource; +import org.apache.skywalking.oap.server.core.register.annotation.InventoryAnnotationUtils; +import org.apache.skywalking.oap.server.core.source.Scope; +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 InventoryProcess { + INSTANCE; + + private Map, RegisterDistinctWorker> entryWorkers = new HashMap<>(); + + public void in(RegisterSource registerSource) { + entryWorkers.get(registerSource.getClass()).in(registerSource); + } + + public void create(ModuleManager moduleManager, Class inventoryClass) { + String modelName = StorageEntityAnnotationUtils.getModelName(inventoryClass); + Scope scope = InventoryAnnotationUtils.getScope(inventoryClass); + + Class builderClass = StorageEntityAnnotationUtils.getBuilder(inventoryClass); + + StorageDAO storageDAO = moduleManager.find(StorageModule.NAME).getService(StorageDAO.class); + IRegisterDAO registerDAO; + try { + registerDAO = storageDAO.newRegisterDao(builderClass.newInstance()); + } catch (InstantiationException | IllegalAccessException e) { + throw new UnexpectedException(""); + } + + RegisterPersistentWorker persistentWorker = new RegisterPersistentWorker(WorkerIdGenerator.INSTANCES.generate(), modelName, moduleManager, registerDAO, scope); + WorkerInstances.INSTANCES.put(persistentWorker.getWorkerId(), persistentWorker); + + RegisterRemoteWorker remoteWorker = new RegisterRemoteWorker(WorkerIdGenerator.INSTANCES.generate(), moduleManager, persistentWorker); + WorkerInstances.INSTANCES.put(remoteWorker.getWorkerId(), remoteWorker); + + RegisterDistinctWorker distinctWorker = new RegisterDistinctWorker(WorkerIdGenerator.INSTANCES.generate(), remoteWorker); + WorkerInstances.INSTANCES.put(distinctWorker.getWorkerId(), distinctWorker); + + entryWorkers.put(inventoryClass, distinctWorker); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java index b6cf4d643f..187cd78955 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java @@ -19,6 +19,7 @@ package org.apache.skywalking.oap.server.core.storage; import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; +import org.apache.skywalking.oap.server.core.register.RegisterSource; import org.apache.skywalking.oap.server.library.module.Service; /** @@ -27,4 +28,6 @@ import org.apache.skywalking.oap.server.library.module.Service; public interface StorageDAO extends Service { IIndicatorDAO newIndicatorDao(StorageBuilder storageBuilder); + + IRegisterDAO newRegisterDao(StorageBuilder storageBuilder); } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java index 641582f7e6..44c9df0bf0 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java @@ -19,6 +19,7 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base; import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; +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; @@ -34,4 +35,8 @@ public class StorageEsDAO extends EsDAO implements StorageDAO { @Override public IIndicatorDAO newIndicatorDao(StorageBuilder storageBuilder) { return new IndicatorEsDAO(getClient(), storageBuilder); } + + @Override public IRegisterDAO newRegisterDao(StorageBuilder storageBuilder) { + return new RegisterEsDAO(getClient(), storageBuilder); + } }