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 d22648e58..91d0b338e 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 @@ -20,6 +20,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.analysis.record.annotation.RecordTypeListener; import org.apache.skywalking.oap.server.core.annotation.AnnotationScan; import org.apache.skywalking.oap.server.core.cache.*; import org.apache.skywalking.oap.server.core.cluster.*; @@ -115,6 +116,7 @@ public class CoreModuleProvider extends ModuleProvider { annotationScan.registerListener(streamAnnotationListener); annotationScan.registerListener(new IndicatorTypeListener(getManager())); annotationScan.registerListener(new InventoryTypeListener(getManager())); + annotationScan.registerListener(new RecordTypeListener(getManager())); this.remoteClientManager = new RemoteClientManager(getManager()); this.registerServiceImplementation(RemoteClientManager.class, remoteClientManager); 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 a67650004..b216ff849 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 @@ -31,6 +31,7 @@ import org.apache.skywalking.oap.server.core.analysis.generated.serviceinstancej import org.apache.skywalking.oap.server.core.analysis.generated.serviceinstancerelation.ServiceInstanceRelationDispatcher; import org.apache.skywalking.oap.server.core.analysis.generated.servicerelation.ServiceRelationDispatcher; import org.apache.skywalking.oap.server.core.analysis.manual.endpointrelation.EndpointCallRelationDispatcher; +import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentDispatcher; import org.apache.skywalking.oap.server.core.analysis.manual.service.*; import org.apache.skywalking.oap.server.core.analysis.manual.servicerelation.ServiceCallRelationDispatcher; import org.apache.skywalking.oap.server.core.source.*; @@ -50,6 +51,8 @@ public class DispatcherManager { this.dispatcherMap.put(Scope.All, new SourceDispatcher[] {new AllDispatcher()}); + this.dispatcherMap.put(Scope.Segment, new SourceDispatcher[] {new SegmentDispatcher()}); + this.dispatcherMap.put(Scope.Service, new SourceDispatcher[] {new ServiceDispatcher()}); this.dispatcherMap.put(Scope.ServiceInstance, new SourceDispatcher[] {new ServiceInstanceDispatcher()}); this.dispatcherMap.put(Scope.Endpoint, new SourceDispatcher[] {new EndpointDispatcher()}); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCache.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCache.java index b5de44361..778de6bb2 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCache.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCache.java @@ -23,11 +23,11 @@ import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; /** * @author peng-yongsheng */ -public class MergeDataCache extends Window> implements DataCache { +public class MergeDataCache extends Window implements DataCache { - private MergeDataCollection lockedMergeDataCollection; + private SWCollection lockedMergeDataCollection; - @Override public MergeDataCollection collectionInstance() { + @Override public SWCollection collectionInstance() { return new MergeDataCollection<>(); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCollection.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCollection.java index bd53c39fa..5a703e5ad 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCollection.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergeDataCollection.java @@ -24,13 +24,14 @@ import org.apache.skywalking.oap.server.core.remote.data.StreamData; /** * @author peng-yongsheng */ -public class MergeDataCollection implements Collection> { - private Map data; +public class MergeDataCollection implements SWCollection { + + private Map collection; private volatile boolean writing; private volatile boolean reading; MergeDataCollection() { - this.data = new HashMap<>(); + this.collection = new HashMap<>(); this.writing = false; this.reading = false; } @@ -59,27 +60,27 @@ public class MergeDataCollection implements Coll return reading; } - boolean containsKey(STREAM_DATA key) { - return data.containsKey(key); + @Override public boolean containsKey(STREAM_DATA key) { + return collection.containsKey(key); } - void put(STREAM_DATA value) { - data.put(value, value); + @Override public void put(STREAM_DATA value) { + collection.put(value, value); } - public STREAM_DATA get(STREAM_DATA key) { - return data.get(key); + @Override public STREAM_DATA get(STREAM_DATA key) { + return collection.get(key); } @Override public int size() { - return data.size(); + return collection.size(); } @Override public void clear() { - data.clear(); + collection.clear(); } - public Map collection() { - return data; + @Override public Collection collection() { + return collection.values(); } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCache.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCache.java index ab40f7dea..238b127c4 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCache.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCache.java @@ -18,21 +18,21 @@ package org.apache.skywalking.oap.server.core.analysis.data; -import org.apache.skywalking.oap.server.core.remote.data.StreamData; +import org.apache.skywalking.oap.server.core.storage.StorageData; /** * @author peng-yongsheng */ -public class NonMergeDataCache extends Window> implements DataCache { +public class NonMergeDataCache extends Window implements DataCache { - private NonMergeDataCollection lockedMergeDataCollection; + private SWCollection lockedMergeDataCollection; - @Override public NonMergeDataCollection collectionInstance() { + @Override public SWCollection collectionInstance() { return new NonMergeDataCollection<>(); } - public void add(STREAM_DATA data) { - lockedMergeDataCollection.add(data); + public void add(STORAGE_DATA data) { + lockedMergeDataCollection.put(data); } @Override public void writing() { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCollection.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCollection.java index b0498c3e4..2a4ac3ae6 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCollection.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/NonMergeDataCollection.java @@ -19,19 +19,19 @@ package org.apache.skywalking.oap.server.core.analysis.data; import java.util.*; -import org.apache.skywalking.oap.server.core.remote.data.StreamData; +import org.apache.skywalking.oap.server.core.storage.StorageData; /** * @author peng-yongsheng */ -public class NonMergeDataCollection implements Collection> { +public class NonMergeDataCollection implements SWCollection { - private final List data; + private final List data; private volatile boolean writing; private volatile boolean reading; NonMergeDataCollection() { - this.data = new LinkedList<>(); + this.data = new ArrayList<>(); this.writing = false; this.reading = false; } @@ -60,10 +60,6 @@ public class NonMergeDataCollection implements C return reading; } - void add(STREAM_DATA value) { - data.add(value); - } - @Override public int size() { return data.size(); } @@ -72,7 +68,19 @@ public class NonMergeDataCollection implements C data.clear(); } - public List collection() { + @Override public boolean containsKey(STORAGE_DATA key) { + throw new UnsupportedOperationException("None merge data collection not support containsKey operation."); + } + + @Override public STORAGE_DATA get(STORAGE_DATA key) { + throw new UnsupportedOperationException("None merge data collection not support get operation."); + } + + @Override public void put(STORAGE_DATA value) { + data.add(value); + } + + @Override public Collection collection() { return data; } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Collection.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/SWCollection.java similarity index 85% rename from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Collection.java rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/SWCollection.java index f1d52a800..032563d86 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Collection.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/SWCollection.java @@ -18,10 +18,12 @@ package org.apache.skywalking.oap.server.core.analysis.data; +import java.util.Collection; + /** * @author peng-yongsheng */ -public interface Collection { +public interface SWCollection { void reading(); @@ -39,5 +41,11 @@ public interface Collection { void finishWriting(); - Data collection(); + Collection collection(); + + boolean containsKey(DATA key); + + DATA get(DATA key); + + void put(DATA value); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Window.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Window.java index 38570b3e9..6f9486dae 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Window.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/Window.java @@ -23,22 +23,22 @@ import java.util.concurrent.atomic.AtomicInteger; /** * @author peng-yongsheng */ -public abstract class Window { +public abstract class Window { private AtomicInteger windowSwitch = new AtomicInteger(0); - private WINDOW_COLLECTION pointer; + private SWCollection pointer; - private WINDOW_COLLECTION windowDataA; - private WINDOW_COLLECTION windowDataB; + private SWCollection windowDataA; + private SWCollection windowDataB; - protected Window() { + Window() { this.windowDataA = collectionInstance(); this.windowDataB = collectionInstance(); this.pointer = windowDataA; } - public abstract WINDOW_COLLECTION collectionInstance(); + public abstract SWCollection collectionInstance(); public boolean trySwitchPointer() { return windowSwitch.incrementAndGet() == 1 && !getLast().isReading(); @@ -57,7 +57,7 @@ public abstract class Window { getLast().reading(); } - protected WINDOW_COLLECTION getCurrentAndWriting() { + SWCollection getCurrentAndWriting() { if (pointer == windowDataA) { windowDataA.writing(); return windowDataA; @@ -67,7 +67,7 @@ public abstract class Window { } } - private WINDOW_COLLECTION getCurrent() { + private SWCollection getCurrent() { return pointer; } @@ -75,7 +75,7 @@ public abstract class Window { return getCurrent().size(); } - public WINDOW_COLLECTION getLast() { + public SWCollection getLast() { if (pointer == windowDataA) { return windowDataB; } else { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentDispatcher.java new file mode 100644 index 000000000..9a4c51f23 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentDispatcher.java @@ -0,0 +1,45 @@ +/* + * 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.manual.segment; + +import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher; +import org.apache.skywalking.oap.server.core.analysis.worker.RecordProcess; +import org.apache.skywalking.oap.server.core.source.Segment; + +/** + * @author peng-yongsheng + */ +public class SegmentDispatcher implements SourceDispatcher { + + @Override public void dispatch(Segment source) { + SegmentRecord segment = new SegmentRecord(); + segment.setSegmentId(source.getSegmentId()); + segment.setTraceId(source.getTraceId()); + segment.setServiceId(source.getServiceId()); + segment.setEndpointName(source.getEndpointName()); + segment.setStartTime(source.getStartTime()); + segment.setEndTime(source.getEndTime()); + segment.setLatency(source.getLatency()); + segment.setIsError(source.getIsError()); + segment.setDataBinary(source.getDataBinary()); + segment.setTimeBucket(source.getTimeBucket()); + + RecordProcess.INSTANCE.in(segment); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java new file mode 100644 index 000000000..ba5fc8821 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java @@ -0,0 +1,103 @@ +/* + * 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.manual.segment; + +import java.util.*; +import lombok.*; +import org.apache.skywalking.apm.util.StringUtil; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.analysis.record.Record; +import org.apache.skywalking.oap.server.core.analysis.record.annotation.RecordType; +import org.apache.skywalking.oap.server.core.storage.StorageBuilder; +import org.apache.skywalking.oap.server.core.storage.annotation.*; +import org.apache.skywalking.oap.server.library.util.CollectionUtils; + +/** + * @author peng-yongsheng + */ +@RecordType +@StorageEntity(name = SegmentRecord.INDEX_NAME, builder = SegmentRecord.Builder.class) +public class SegmentRecord extends Record { + + public static final String INDEX_NAME = "segment"; + public static final String SEGMENT_ID = "segment_id"; + public static final String TRACE_ID = "trace_id"; + public static final String SERVICE_ID = "service_id"; + public static final String ENDPOINT_NAME = "endpoint_name"; + public static final String START_TIME = "start_time"; + public static final String END_TIME = "end_time"; + public static final String LATENCY = "latency"; + public static final String IS_ERROR = "is_error"; + public static final String DATA_BINARY = "data_binary"; + + @Setter @Getter @Column(columnName = SEGMENT_ID) @IDColumn private String segmentId; + @Setter @Getter @Column(columnName = TRACE_ID) @IDColumn private String traceId; + @Setter @Getter @Column(columnName = SERVICE_ID) @IDColumn private int serviceId; + @Setter @Getter @Column(columnName = ENDPOINT_NAME) @IDColumn private String endpointName; + @Setter @Getter @Column(columnName = START_TIME) @IDColumn private long startTime; + @Setter @Getter @Column(columnName = END_TIME) @IDColumn private long endTime; + @Setter @Getter @Column(columnName = LATENCY) @IDColumn private int latency; + @Setter @Getter @Column(columnName = IS_ERROR) @IDColumn private int isError; + @Setter @Getter @Column(columnName = DATA_BINARY) @IDColumn private byte[] dataBinary; + + @Override public String id() { + return segmentId; + } + + public static class Builder implements StorageBuilder { + + @Override public Map data2Map(SegmentRecord storageData) { + Map map = new HashMap<>(); + map.put(SEGMENT_ID, storageData.getSegmentId()); + map.put(TRACE_ID, storageData.getTraceId()); + map.put(SERVICE_ID, storageData.getServiceId()); + map.put(ENDPOINT_NAME, storageData.getEndpointName()); + map.put(START_TIME, storageData.getStartTime()); + map.put(END_TIME, storageData.getEndTime()); + map.put(LATENCY, storageData.getLatency()); + map.put(IS_ERROR, storageData.getIsError()); + map.put(TIME_BUCKET, storageData.getTimeBucket()); + if (CollectionUtils.isEmpty(storageData.getDataBinary())) { + map.put(DATA_BINARY, Const.EMPTY_STRING); + } else { + map.put(DATA_BINARY, new String(Base64.getEncoder().encode(storageData.getDataBinary()))); + } + return map; + } + + @Override public SegmentRecord map2Data(Map dbMap) { + SegmentRecord record = new SegmentRecord(); + record.setSegmentId((String)dbMap.get(SEGMENT_ID)); + record.setTraceId((String)dbMap.get(TRACE_ID)); + record.setServiceId(((Number)dbMap.get(SERVICE_ID)).intValue()); + record.setEndpointName((String)dbMap.get(ENDPOINT_NAME)); + record.setStartTime(((Number)dbMap.get(START_TIME)).longValue()); + record.setEndTime(((Number)dbMap.get(END_TIME)).longValue()); + record.setLatency(((Number)dbMap.get(LATENCY)).intValue()); + record.setIsError(((Number)dbMap.get(IS_ERROR)).intValue()); + record.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue()); + if (StringUtil.isEmpty((String)dbMap.get(DATA_BINARY))) { + record.setDataBinary(new byte[] {}); + } else { + record.setDataBinary(Base64.getDecoder().decode((String)dbMap.get(DATA_BINARY))); + } + return record; + } + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/Record.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/Record.java new file mode 100644 index 000000000..0973a63d8 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/Record.java @@ -0,0 +1,33 @@ +/* + * 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.record; + +import lombok.*; +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 Record implements StorageData { + + public static final String TIME_BUCKET = "time_bucket"; + + @Getter @Setter @Column(columnName = TIME_BUCKET) private long timeBucket; +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/annotation/RecordType.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/annotation/RecordType.java new file mode 100644 index 000000000..880477cdc --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/annotation/RecordType.java @@ -0,0 +1,29 @@ +/* + * 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.record.annotation; + +import java.lang.annotation.*; + +/** + * @author peng-yongsheng + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +public @interface RecordType { +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/annotation/RecordTypeListener.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/annotation/RecordTypeListener.java new file mode 100644 index 000000000..6eb0d0728 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/record/annotation/RecordTypeListener.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.analysis.record.annotation; + +import java.lang.annotation.Annotation; +import org.apache.skywalking.oap.server.core.analysis.worker.RecordProcess; +import org.apache.skywalking.oap.server.core.annotation.AnnotationListener; +import org.apache.skywalking.oap.server.library.module.ModuleManager; + +/** + * @author peng-yongsheng + */ +public class RecordTypeListener implements AnnotationListener { + + private final ModuleManager moduleManager; + + public RecordTypeListener(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + @Override public Class annotation() { + return RecordType.class; + } + + @Override public void notify(Class aClass) { + RecordProcess.INSTANCE.create(moduleManager, aClass); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java index 8d0bf889c..15ff40b4f 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java @@ -71,7 +71,7 @@ public class IndicatorAggregateWorker extends AbstractWorker { } } - mergeDataCache.getLast().collection().forEach((Indicator key, Indicator data) -> { + mergeDataCache.getLast().collection().forEach(data -> { if (logger.isDebugEnabled()) { logger.debug(data.toString()); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java index bca9752df..6a614a087 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java @@ -19,9 +19,9 @@ package org.apache.skywalking.oap.server.core.analysis.worker; import java.util.*; -import org.apache.skywalking.oap.server.core.analysis.data.*; +import org.apache.skywalking.oap.server.core.analysis.data.MergeDataCache; 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.IIndicatorDAO; import org.apache.skywalking.oap.server.core.worker.AbstractWorker; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.slf4j.*; @@ -31,48 +31,28 @@ import static java.util.Objects.nonNull; /** * @author peng-yongsheng */ -public class IndicatorPersistentWorker extends AbstractWorker { +public class IndicatorPersistentWorker extends PersistenceWorker> { private static final Logger logger = LoggerFactory.getLogger(IndicatorPersistentWorker.class); private final String modelName; private final MergeDataCache mergeDataCache; - private final IBatchDAO batchDAO; private final IIndicatorDAO indicatorDAO; - private final int blockBatchPersistenceSize; private final AbstractWorker nextWorker; IndicatorPersistentWorker(int workerId, String modelName, int batchSize, ModuleManager moduleManager, IIndicatorDAO indicatorDAO, AbstractWorker nextWorker) { - super(workerId); + super(moduleManager, workerId, batchSize); this.modelName = modelName; - this.blockBatchPersistenceSize = batchSize; this.mergeDataCache = new MergeDataCache<>(); - this.batchDAO = moduleManager.find(StorageModule.NAME).getService(IBatchDAO.class); this.indicatorDAO = indicatorDAO; this.nextWorker = nextWorker; } - public final Window> getCache() { + @Override public MergeDataCache getCache() { return mergeDataCache; } - @Override public final void in(Indicator input) { - if (getCache().currentCollectionSize() >= blockBatchPersistenceSize) { - try { - if (getCache().trySwitchPointer()) { - getCache().switchPointer(); - - List collection = buildBatchCollection(); - batchDAO.batchPersistence(collection); - } - } finally { - getCache().trySwitchPointerFinally(); - } - } - cacheData(input); - } - public boolean flushAndSwitch() { boolean isSwitch; try { @@ -85,29 +65,9 @@ public class IndicatorPersistentWorker extends AbstractWorker { return isSwitch; } - public final List buildBatchCollection() { - List batchCollection = new LinkedList<>(); - try { - while (getCache().getLast().isWriting()) { - try { - Thread.sleep(10); - } catch (InterruptedException e) { - logger.warn("thread wake up"); - } - } - - if (getCache().getLast().collection() != null) { - batchCollection = prepareBatch(getCache().getLast()); - } - } finally { - getCache().finishReadingLast(); - } - return batchCollection; - } - - private List prepareBatch(MergeDataCollection collection) { + @Override public List prepareBatch(MergeDataCache cache) { List batchCollection = new LinkedList<>(); - collection.collection().forEach((id, data) -> { + cache.getLast().collection().forEach(data -> { Indicator dbData = null; try { dbData = indicatorDAO.get(modelName, data); @@ -131,7 +91,7 @@ public class IndicatorPersistentWorker extends AbstractWorker { return batchCollection; } - private void cacheData(Indicator input) { + @Override public void cacheData(Indicator input) { mergeDataCache.writing(); if (mergeDataCache.containsKey(input)) { Indicator indicator = mergeDataCache.get(input); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java new file mode 100644 index 000000000..d731c733a --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java @@ -0,0 +1,97 @@ +/* + * 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.analysis.data.Window; +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 abstract class PersistenceWorker> extends AbstractWorker { + + private static final Logger logger = LoggerFactory.getLogger(PersistenceWorker.class); + + private final int batchSize; + private final IBatchDAO batchDAO; + + PersistenceWorker(ModuleManager moduleManager, int workerId, int batchSize) { + super(workerId); + this.batchSize = batchSize; + this.batchDAO = moduleManager.find(StorageModule.NAME).getService(IBatchDAO.class); + } + + @Override public final void in(INPUT input) { + if (getCache().currentCollectionSize() >= batchSize) { + try { + if (getCache().trySwitchPointer()) { + getCache().switchPointer(); + + List collection = buildBatchCollection(); + batchDAO.batchPersistence(collection); + } + } finally { + getCache().trySwitchPointerFinally(); + } + } + cacheData(input); + } + + public abstract void cacheData(INPUT input); + + public abstract CACHE getCache(); + + public boolean flushAndSwitch() { + boolean isSwitch; + try { + if (isSwitch = getCache().trySwitchPointer()) { + getCache().switchPointer(); + } + } finally { + getCache().trySwitchPointerFinally(); + } + return isSwitch; + } + + public abstract List prepareBatch(CACHE cache); + + public final List buildBatchCollection() { + List batchCollection = new LinkedList<>(); + try { + while (getCache().getLast().isWriting()) { + try { + Thread.sleep(10); + } catch (InterruptedException e) { + logger.warn("thread wake up"); + } + } + + if (getCache().getLast().collection() != null) { + batchCollection = prepareBatch(getCache()); + } + } finally { + getCache().finishReadingLast(); + } + return batchCollection; + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.java new file mode 100644 index 000000000..ea8a47d2e --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordPersistentWorker.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.analysis.worker; + +import java.util.*; +import org.apache.skywalking.oap.server.core.analysis.data.NonMergeDataCache; +import org.apache.skywalking.oap.server.core.analysis.record.Record; +import org.apache.skywalking.oap.server.core.storage.IRecordDAO; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.slf4j.*; + +/** + * @author peng-yongsheng + */ +public class RecordPersistentWorker extends PersistenceWorker> { + + private static final Logger logger = LoggerFactory.getLogger(RecordPersistentWorker.class); + + private final String modelName; + private final NonMergeDataCache nonMergeDataCache; + private final IRecordDAO recordDAO; + + RecordPersistentWorker(int workerId, String modelName, int batchSize, ModuleManager moduleManager, + IRecordDAO recordDAO) { + super(moduleManager, workerId, batchSize); + this.modelName = modelName; + this.nonMergeDataCache = new NonMergeDataCache<>(); + this.recordDAO = recordDAO; + } + + @Override public NonMergeDataCache getCache() { + return nonMergeDataCache; + } + + @Override public List prepareBatch(NonMergeDataCache cache) { + List batchCollection = new LinkedList<>(); + cache.getLast().collection().forEach(record -> { + try { + batchCollection.add(recordDAO.prepareBatchInsert(modelName, record)); + } catch (Throwable t) { + logger.error(t.getMessage(), t); + } + }); + return batchCollection; + } + + @Override public void cacheData(Record input) { + nonMergeDataCache.writing(); + nonMergeDataCache.add(input); + nonMergeDataCache.finishWriting(); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordProcess.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordProcess.java new file mode 100644 index 000000000..838136d8c --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordProcess.java @@ -0,0 +1,62 @@ +/* + * 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 lombok.Getter; +import org.apache.skywalking.oap.server.core.UnexpectedException; +import org.apache.skywalking.oap.server.core.analysis.record.Record; +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 RecordProcess { + INSTANCE; + + private Map, RecordPersistentWorker> workers = new HashMap<>(); + + public void in(Record record) { + workers.get(record.getClass()).in(record); + } + + @Getter private List persistentWorkers = new ArrayList<>(); + + public void create(ModuleManager moduleManager, Class recordClass) { + String modelName = StorageEntityAnnotationUtils.getModelName(recordClass); + Class builderClass = StorageEntityAnnotationUtils.getBuilder(recordClass); + + StorageDAO storageDAO = moduleManager.find(StorageModule.NAME).getService(StorageDAO.class); + IRecordDAO recordDAO; + try { + recordDAO = storageDAO.newRecordDao(builderClass.newInstance()); + } catch (InstantiationException | IllegalAccessException e) { + throw new UnexpectedException(""); + } + + RecordPersistentWorker persistentWorker = new RecordPersistentWorker(WorkerIdGenerator.INSTANCES.generate(), modelName, + 1000, moduleManager, recordDAO); + WorkerInstances.INSTANCES.put(persistentWorker.getWorkerId(), persistentWorker); + persistentWorkers.add(persistentWorker); + workers.put(recordClass, persistentWorker); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java index 396ecabdc..e77d03601 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java @@ -24,5 +24,6 @@ package org.apache.skywalking.oap.server.core.source; public enum Scope { All, Service, ServiceInstance, Endpoint, ServiceRelation, ServiceInstanceRelation, EndpointRelation, NetworkAddress, ServiceInstanceJVMCPU, ServiceInstanceJVMMemory, ServiceInstanceJVMMemoryPool, ServiceInstanceJVMGC, - ServiceComponent, ServiceMapping + ServiceComponent, ServiceMapping, + Segment } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Segment.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Segment.java new file mode 100644 index 000000000..b66917176 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Segment.java @@ -0,0 +1,47 @@ +/* + * 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.source; + +import lombok.*; +import org.apache.skywalking.oap.server.core.source.annotation.SourceType; + +/** + * @author peng-yongsheng + */ +@SourceType +public class Segment extends Source { + + @Override public Scope scope() { + return Scope.Segment; + } + + @Override public String getEntityId() { + return segmentId; + } + + @Setter @Getter private String segmentId; + @Setter @Getter private String traceId; + @Setter @Getter private int serviceId; + @Setter @Getter private String endpointName; + @Setter @Getter private long startTime; + @Setter @Getter private long endTime; + @Setter @Getter private int latency; + @Setter @Getter private int isError; + @Setter @Getter private byte[] dataBinary; +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java new file mode 100644 index 000000000..d5e156f39 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java @@ -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; + +import java.io.IOException; +import org.apache.skywalking.oap.server.core.analysis.record.Record; + +/** + * @author peng-yongsheng + */ +public interface IRecordDAO extends DAO { + + INSERT prepareBatchInsert(String modelName, Record record) throws IOException; + + void deleteHistory(String modelName, Long timeBucketBefore); +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java index 7271f0420..4d8ebd705 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java @@ -21,7 +21,7 @@ package org.apache.skywalking.oap.server.core.storage; import java.util.*; import java.util.concurrent.*; import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; -import org.apache.skywalking.oap.server.core.analysis.worker.IndicatorProcess; +import org.apache.skywalking.oap.server.core.analysis.worker.*; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.slf4j.*; @@ -65,7 +65,11 @@ public enum PersistenceTimer { long startTime = System.currentTimeMillis(); try { List batchAllCollection = new LinkedList(); - IndicatorProcess.INSTANCE.getPersistentWorkers().forEach(worker -> { + List persistenceWorkers = new ArrayList<>(); + persistenceWorkers.addAll(IndicatorProcess.INSTANCE.getPersistentWorkers()); + persistenceWorkers.addAll(RecordProcess.INSTANCE.getPersistentWorkers()); + + persistenceWorkers.forEach(worker -> { if (logger.isDebugEnabled()) { logger.debug("extract {} worker data and save", worker.getClass().getName()); } 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 187cd7895..ddb348e38 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.analysis.record.Record; import org.apache.skywalking.oap.server.core.register.RegisterSource; import org.apache.skywalking.oap.server.library.module.Service; @@ -30,4 +31,6 @@ public interface StorageDAO extends Service { IIndicatorDAO newIndicatorDao(StorageBuilder storageBuilder); IRegisterDAO newRegisterDao(StorageBuilder storageBuilder); + + IRecordDAO newRecordDao(StorageBuilder storageBuilder); } diff --git a/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/CollectionUtils.java b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/CollectionUtils.java index 8faaffad5..c3b657c16 100644 --- a/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/CollectionUtils.java +++ b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/CollectionUtils.java @@ -52,4 +52,12 @@ public class CollectionUtils { public static boolean isNotEmpty(T[] array) { return array != null && array.length > 0; } + + public static boolean isEmpty(byte[] array) { + return array == null || array.length == 0; + } + + public static boolean isNotEmpty(byte[] array) { + return !isEmpty(array); + } } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java index 266e85df6..83065eadc 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java @@ -26,6 +26,7 @@ import org.apache.skywalking.oap.server.receiver.trace.module.TraceModule; import org.apache.skywalking.oap.server.receiver.trace.provider.handler.TraceSegmentServiceHandler; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.*; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.endpoint.MultiScopesSpanListener; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.segment.SegmentSpanListener; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.service.*; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardizationWorker; @@ -60,6 +61,7 @@ public class TraceModuleProvider extends ModuleProvider { listenerManager.add(new MultiScopesSpanListener.Factory()); listenerManager.add(new ServiceComponentSpanListener.Factory()); listenerManager.add(new ServiceMappingSpanListener.Factory()); + listenerManager.add(new SegmentSpanListener.Factory()); GRPCHandlerRegister grpcHandlerRegister = getManager().find(CoreModule.NAME).getService(GRPCHandlerRegister.class); try { diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java index 78c6a7842..d07c4a5a9 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java @@ -78,9 +78,7 @@ public class SegmentParse implements DataStreamReader.CallBack if (logger.isDebugEnabled()) { logger.debug("This segment id exchange success, id: {}", segmentCoreInfo.getSegmentId()); } - notifyListenerToBuild(); - buildSegment(segmentCoreInfo.getSegmentId(), segmentDecorator.toByteArray()); return true; } } catch (Throwable e) { @@ -111,6 +109,7 @@ public class SegmentParse implements DataStreamReader.CallBack segmentCoreInfo.setSegmentId(segmentIdBuilder.toString()); segmentCoreInfo.setApplicationId(segmentDecorator.getApplicationId()); segmentCoreInfo.setApplicationInstanceId(segmentDecorator.getApplicationInstanceId()); + segmentCoreInfo.setDataBinary(segmentDecorator.toByteArray()); for (int i = 0; i < segmentDecorator.getSpansCount(); i++) { SpanDecorator spanDecorator = segmentDecorator.getSpans(i); @@ -159,13 +158,6 @@ public class SegmentParse implements DataStreamReader.CallBack return true; } - private void buildSegment(String id, byte[] dataBinary) { -// Segment segment = new Segment(); -// segment.setId(id); -// segment.setDataBinary(dataBinary); -// segment.setTimeBucket(segmentCoreInfo.getMinuteTimeBucket()); - } - private void writeToBufferFile(String id, UpstreamSegment upstreamSegment) { if (logger.isDebugEnabled()) { logger.debug("push to segment buffer write worker, id: {}", id); @@ -215,7 +207,7 @@ public class SegmentParse implements DataStreamReader.CallBack private void notifyGlobalsListener(UniqueId uniqueId) { spanListeners.forEach(listener -> { - if (listener.containsPoint(SpanListener.Point.GlobalTraceIds)) { + if (listener.containsPoint(SpanListener.Point.TraceIds)) { ((GlobalTraceIdsListener)listener).parseGlobalTraceId(uniqueId, segmentCoreInfo); } }); diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/decorator/SegmentCoreInfo.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/decorator/SegmentCoreInfo.java index 78817ea3a..a0c0b7f8a 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/decorator/SegmentCoreInfo.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/decorator/SegmentCoreInfo.java @@ -18,80 +18,20 @@ package org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator; +import lombok.*; + /** * @author peng-yongsheng */ +@Getter +@Setter public class SegmentCoreInfo { private String segmentId; - private String traceId; private int applicationId; private int applicationInstanceId; private long startTime; private long endTime; private boolean isError; private long minuteTimeBucket; - - public String getSegmentId() { - return segmentId; - } - - public void setSegmentId(String segmentId) { - this.segmentId = segmentId; - } - - public int getApplicationId() { - return applicationId; - } - - public void setApplicationId(int applicationId) { - this.applicationId = applicationId; - } - - public int getApplicationInstanceId() { - return applicationInstanceId; - } - - public void setApplicationInstanceId(int applicationInstanceId) { - this.applicationInstanceId = applicationInstanceId; - } - - public long getStartTime() { - return startTime; - } - - public void setStartTime(long startTime) { - this.startTime = startTime; - } - - public long getEndTime() { - return endTime; - } - - public void setEndTime(long endTime) { - this.endTime = endTime; - } - - public boolean isError() { - return isError; - } - - public void setError(boolean error) { - isError = error; - } - - public long getMinuteTimeBucket() { - return minuteTimeBucket; - } - - public void setMinuteTimeBucket(long minuteTimeBucket) { - this.minuteTimeBucket = minuteTimeBucket; - } - - public String getTraceId() { - return traceId; - } - - public void setTraceId(String traceId) { - this.traceId = traceId; - } + private byte[] dataBinary; } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/SpanListener.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/SpanListener.java index ce7008125..08e3b808b 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/SpanListener.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/SpanListener.java @@ -27,6 +27,6 @@ public interface SpanListener { boolean containsPoint(Point point); enum Point { - Entry, Exit, Local, First, GlobalTraceIds + Entry, Exit, Local, First, TraceIds } } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/segment/SegmentSpanListener.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/segment/SegmentSpanListener.java new file mode 100644 index 000000000..d3dceb8d7 --- /dev/null +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/segment/SegmentSpanListener.java @@ -0,0 +1,106 @@ +/* + * 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.receiver.trace.provider.parser.listener.segment; + +import org.apache.skywalking.apm.network.language.agent.UniqueId; +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.cache.EndpointInventoryCache; +import org.apache.skywalking.oap.server.core.source.*; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.library.util.*; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.*; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.*; +import org.slf4j.*; + +/** + * @author peng-yongsheng + */ +public class SegmentSpanListener implements FirstSpanListener, EntrySpanListener, GlobalTraceIdsListener { + + private static final Logger logger = LoggerFactory.getLogger(SegmentSpanListener.class); + + private final SourceReceiver sourceReceiver; + private final Segment segment = new Segment(); + private final EndpointInventoryCache serviceNameCacheService; + private int entryEndpointId = 0; + private int firstEndpointId = 0; + + private SegmentSpanListener(ModuleManager moduleManager) { + this.sourceReceiver = moduleManager.find(CoreModule.NAME).getService(SourceReceiver.class); + this.serviceNameCacheService = moduleManager.find(CoreModule.NAME).getService(EndpointInventoryCache.class); + } + + @Override public boolean containsPoint(Point point) { + return Point.First.equals(point) || Point.Entry.equals(point) || Point.TraceIds.equals(point); + } + + @Override + public void parseFirst(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) { + long timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(segmentCoreInfo.getStartTime()); + + segment.setSegmentId(segmentCoreInfo.getSegmentId()); + segment.setSegmentId(segmentCoreInfo.getSegmentId()); + segment.setServiceId(segmentCoreInfo.getApplicationId()); + segment.setLatency((int)(segmentCoreInfo.getEndTime() - segmentCoreInfo.getStartTime())); + segment.setStartTime(segmentCoreInfo.getStartTime()); + segment.setEndTime(segmentCoreInfo.getEndTime()); + segment.setIsError(BooleanUtils.booleanToValue(segmentCoreInfo.isError())); + segment.setTimeBucket(timeBucket); + segment.setDataBinary(segmentCoreInfo.getDataBinary()); + + firstEndpointId = spanDecorator.getOperationNameId(); + } + + @Override public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) { + entryEndpointId = spanDecorator.getOperationNameId(); + } + + @Override public void parseGlobalTraceId(UniqueId uniqueId, SegmentCoreInfo segmentCoreInfo) { + StringBuilder traceIdBuilder = new StringBuilder(); + for (int i = 0; i < uniqueId.getIdPartsList().size(); i++) { + if (i == 0) { + traceIdBuilder.append(uniqueId.getIdPartsList().get(i)); + } else { + traceIdBuilder.append(".").append(uniqueId.getIdPartsList().get(i)); + } + } + segment.setTraceId(traceIdBuilder.toString()); + } + + @Override public void build() { + if (logger.isDebugEnabled()) { + logger.debug("segment duration listener build"); + } + + if (entryEndpointId == 0) { + segment.setEndpointName(serviceNameCacheService.get(firstEndpointId).getName()); + } else { + segment.setEndpointName(serviceNameCacheService.get(entryEndpointId).getName()); + } + + sourceReceiver.receive(segment); + } + + public static class Factory implements SpanListenerFactory { + + @Override public SpanListener create(ModuleManager moduleManager) { + return new SegmentSpanListener(moduleManager); + } + } +} diff --git a/oap-server/server-starter/src/main/resources/log4j2.xml b/oap-server/server-starter/src/main/resources/log4j2.xml index 2da2d1369..a6ce5a263 100644 --- a/oap-server/server-starter/src/main/resources/log4j2.xml +++ b/oap-server/server-starter/src/main/resources/log4j2.xml @@ -32,8 +32,8 @@ - - + + diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java index 37c3db840..9df21489d 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/ColumnTypeEsMapping.java @@ -37,6 +37,8 @@ public class ColumnTypeEsMapping implements DataTypeMapping { return "keyword"; } else if (IntKeyLongValueArray.class.equals(type)) { return "keyword"; + } else if (byte[].class.equals(type)) { + return "binary"; } else { throw new IllegalArgumentException("Unsupported data type: " + type.getName()); } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java new file mode 100644 index 000000000..bec4cabaf --- /dev/null +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java @@ -0,0 +1,60 @@ +/* + * 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.analysis.record.Record; +import org.apache.skywalking.oap.server.core.storage.*; +import org.apache.skywalking.oap.server.core.storage.type.StorageDataType; +import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; +import org.elasticsearch.action.index.IndexRequest; +import org.elasticsearch.common.xcontent.*; + +/** + * @author peng-yongsheng + */ +public class RecordEsDAO extends EsDAO implements IRecordDAO { + + private final StorageBuilder storageBuilder; + + public RecordEsDAO(ElasticSearchClient client, StorageBuilder storageBuilder) { + super(client); + this.storageBuilder = storageBuilder; + } + + @Override public IndexRequest prepareBatchInsert(String modelName, Record record) throws IOException { + Map objectMap = storageBuilder.data2Map(record); + + XContentBuilder builder = XContentFactory.jsonBuilder().startObject(); + for (String key : objectMap.keySet()) { + Object value = objectMap.get(key); + if (value instanceof StorageDataType) { + builder.field(key, ((StorageDataType)value).toStorageData()); + } else { + builder.field(key, value); + } + } + builder.endObject(); + return getClient().prepareInsert(modelName, record.id(), builder); + } + + @Override public void deleteHistory(String modelName, Long timeBucketBefore) { + } +} 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 44c9df0bf..3774d65df 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.analysis.record.Record; 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; @@ -39,4 +40,8 @@ public class StorageEsDAO extends EsDAO implements StorageDAO { @Override public IRegisterDAO newRegisterDao(StorageBuilder storageBuilder) { return new RegisterEsDAO(getClient(), storageBuilder); } + + @Override public IRecordDAO newRecordDao(StorageBuilder storageBuilder) { + return new RecordEsDAO(getClient(), storageBuilder); + } } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java index 95fb0949d..19e616ae7 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsInstaller.java @@ -19,11 +19,12 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base; import java.io.IOException; -import org.apache.skywalking.oap.server.core.storage.*; +import org.apache.skywalking.oap.server.core.storage.StorageException; import org.apache.skywalking.oap.server.core.storage.model.*; import org.apache.skywalking.oap.server.library.client.Client; import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.elasticsearch.common.Strings; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.xcontent.*; import org.slf4j.*; @@ -80,9 +81,9 @@ public class StorageEsInstaller extends ModelInstaller { Settings settings = createSettingBuilder(); try { mappingBuilder = createMappingBuilder(tableDefine); - logger.info("mapping builder str: {}", mappingBuilder.prettyPrint()); + logger.info("index {}'s mapping builder str: {}", tableDefine.getName(), Strings.toString(mappingBuilder.prettyPrint())); } catch (Exception e) { - logger.error("create {} index mapping builder error", tableDefine.getName()); + logger.error("create {} index mapping builder error, error message: {}", tableDefine.getName(), e.getMessage()); } boolean isAcknowledged;