Register start up. (#1550)
This commit is contained in:
彭勇升 pengys
GitHub
parent
946376e2e5
commit
a18c451ebc
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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.*;
|
||||
|
||||
|
|
|
|||
|
|
@ -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<EndpointLatencyAvgIndicator> {
|
||||
|
||||
@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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -42,9 +42,9 @@ public class EndpointInventoryCacheService implements Service {
|
|||
this.moduleManager = moduleManager;
|
||||
}
|
||||
|
||||
private final Cache<String, Integer> idCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build();
|
||||
private final Cache<String, Integer> idCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(100000).build();
|
||||
|
||||
private final Cache<Integer, EndpointInventory> sequenceCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build();
|
||||
private final Cache<Integer, EndpointInventory> 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;
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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<Endpoint> {
|
||||
public class InventoryAnnotationUtils {
|
||||
|
||||
@Override public void dispatch(Endpoint source) {
|
||||
avg(source);
|
||||
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("");
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
}
|
||||
|
|
@ -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<? extends Annotation> annotation() {
|
||||
return InventoryType.class;
|
||||
}
|
||||
|
||||
@Override public void notify(Class aClass) {
|
||||
InventoryProcess.INSTANCE.create(moduleManager, aClass);
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Class<? extends RegisterSource>, RegisterDistinctWorker> entryWorkers = new HashMap<>();
|
||||
|
||||
public void in(RegisterSource registerSource) {
|
||||
entryWorkers.get(registerSource.getClass()).in(registerSource);
|
||||
}
|
||||
|
||||
public void create(ModuleManager moduleManager, Class<? extends RegisterSource> inventoryClass) {
|
||||
String modelName = StorageEntityAnnotationUtils.getModelName(inventoryClass);
|
||||
Scope scope = InventoryAnnotationUtils.getScope(inventoryClass);
|
||||
|
||||
Class<? extends StorageBuilder> 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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Indicator> storageBuilder);
|
||||
|
||||
IRegisterDAO newRegisterDao(StorageBuilder<RegisterSource> storageBuilder);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Indicator> storageBuilder) {
|
||||
return new IndicatorEsDAO(getClient(), storageBuilder);
|
||||
}
|
||||
|
||||
@Override public IRegisterDAO newRegisterDao(StorageBuilder<RegisterSource> storageBuilder) {
|
||||
return new RegisterEsDAO(getClient(), storageBuilder);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue