From 648e2d8b188d20ec922c1dce63ea0bebbdc9e1c9 Mon Sep 17 00:00:00 2001 From: ascrutae Date: Wed, 16 Nov 2016 10:20:12 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=BC=82=E6=AD=A5=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E6=8E=A5=E5=8F=A3=EF=BC=8C=E4=BC=98=E5=8C=96=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E5=AD=98=E5=82=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../provider/AsyncTraceSearchService.java | 30 +++++++- .../listener/AsyncTraceSearchListener.java | 6 ++ .../src/main/proto/TraceSearchService.proto | 2 +- .../com/a/eye/skywalking/storage/Main.java | 3 +- .../skywalking/storage/config/Constants.java | 7 +- .../storage/data/SpanDataFinder.java | 11 +-- .../storage/data/file/DataFile.java | 37 +++++----- .../storage/data/file/DataFileNameDesc.java | 74 +++++++++++++++++++ .../storage/data/index/IndexDBConnector.java | 29 +++++--- .../storage/data/index/IndexMetaInfo.java | 11 +-- .../storage/listener/SearchListener.java | 6 +- .../eye/skywalking/storage/SearchClient.java | 46 ++++++++++++ 12 files changed, 213 insertions(+), 49 deletions(-) create mode 100644 skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileNameDesc.java create mode 100644 skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/storage/SearchClient.java diff --git a/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/provider/AsyncTraceSearchService.java b/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/provider/AsyncTraceSearchService.java index e6339f5f4..9caad6ed2 100644 --- a/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/provider/AsyncTraceSearchService.java +++ b/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/provider/AsyncTraceSearchService.java @@ -3,20 +3,44 @@ package com.a.eye.skywalking.network.grpc.provider; import com.a.eye.skywalking.network.grpc.AsyncTraceSearchServiceGrpc; import com.a.eye.skywalking.network.grpc.QueryTask; import com.a.eye.skywalking.network.grpc.SearchResult; +import com.a.eye.skywalking.network.grpc.Span; import com.a.eye.skywalking.network.listener.AsyncTraceSearchListener; import io.grpc.stub.StreamObserver; +import java.util.List; + /** * Created by xin on 2016/11/15. */ public class AsyncTraceSearchService extends AsyncTraceSearchServiceGrpc.AsyncTraceSearchServiceImplBase { - public AsyncTraceSearchService(AsyncTraceSearchListener asyncTraceSearchListener) { + private AsyncTraceSearchListener searchListener; + public AsyncTraceSearchService(AsyncTraceSearchListener searchListener) { + this.searchListener = searchListener; } @Override - public StreamObserver search(StreamObserver responseObserver) { - return super.search(responseObserver); + public StreamObserver search(final StreamObserver responseObserver) { + return new StreamObserver() { + private List spans; + private int taskId; + + @Override + public void onNext(QueryTask value) { + taskId = value.getTaskId(); + spans = searchListener.search(value.getTraceId()); + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + responseObserver.onNext(SearchResult.newBuilder().addAllSpans(spans).setTaskId(taskId).build()); + responseObserver.onCompleted(); + } + }; } } diff --git a/skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/AsyncTraceSearchListener.java b/skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/AsyncTraceSearchListener.java index 16e0a6404..d6fb79a5d 100644 --- a/skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/AsyncTraceSearchListener.java +++ b/skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/AsyncTraceSearchListener.java @@ -1,7 +1,13 @@ package com.a.eye.skywalking.network.listener; +import com.a.eye.skywalking.network.grpc.Span; +import com.a.eye.skywalking.network.grpc.TraceId; + +import java.util.List; + /** * Created by xin on 2016/11/15. */ public interface AsyncTraceSearchListener { + List search(TraceId traceId); } diff --git a/skywalking-network/src/main/proto/TraceSearchService.proto b/skywalking-network/src/main/proto/TraceSearchService.proto index 201d6b9ea..f7c4d53c4 100644 --- a/skywalking-network/src/main/proto/TraceSearchService.proto +++ b/skywalking-network/src/main/proto/TraceSearchService.proto @@ -21,6 +21,6 @@ message QueryTask { } message SearchResult { - string traceid = 1; + int32 taskId = 1; repeated Span spans = 2; } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/Main.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/Main.java index afe9e0783..1fb8acfac 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/Main.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/Main.java @@ -51,7 +51,7 @@ public class Main { IndexDataCapacityMonitor.start(); provider = ServiceProvider.newBuilder(Config.Server.PORT).addSpanStorageService(new StorageListener()) - .addTraceSearchService(new SearchListener()).build(); + .addAsyncTraceSearchService(new SearchListener()).build(); provider.start(); if (logger.isDebugEnable()) { @@ -63,6 +63,7 @@ public class Main { logger.info("SkyWalking storage server started."); Thread.currentThread().join(); } catch (Throwable e) { + e.printStackTrace(); logger.error("SkyWalking storage server start failure.", e); } finally { provider.stop(); diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java index 6866e98da..8a7e933b3 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java @@ -17,7 +17,8 @@ public class Constants { + " tid_s4 INT NOT NULL,\n" + " tid_s5 INT NOT NULL,\n" + " span_type INT NOT NULL, \n" - + " file_name VARCHAR(32) NOT NULL,\n" + + " file_name BIGINT NOT NULL,\n" + + " file_name_suffix INT NOT NULL,\n" + " offset BIGINT NOT NULL,\n" + " length INT NOT NULL\n" + ");\n"; @@ -25,14 +26,14 @@ public class Constants { + "(tid_s0,tid_s1,tid_s2,tid_s3,tid_s4,tid_s5);"; public static final String INSERT_INDEX = "INSERT INTO " +TABLE_NAME + "(tid_s0,tid_s1,tid_s2,tid_s3,tid_s4,tid_s5,span_type" - + ",file_name,offset,length) VALUES(?,?,?,?,?,?,?,?,?,?)"; + + ",file_name,file_name_suffix,offset,length) VALUES(?,?,?,?,?,?,?,?,?,?,?)"; public static final String QUERY_TABLES = "SELECT count(1) AS TABLE_COUNT FROM INFORMATION_SCHEMA.TABLES " + "WHERE TABLE_NAME= '" + TABLE_NAME.toUpperCase() + "';"; public static final String QUERY_INDEX_SIZE = "SELECT count(1) AS INDEX_SIZE FROM " + TABLE_NAME; - public static final String QUERY_TRACE_ID = "SELECT span_type, file_name, offset, length " + public static final String QUERY_TRACE_ID = "SELECT span_type, file_name,file_name_suffix, offset, length " + " FROM "+ TABLE_NAME+ " WHERE tid_s0 = ? AND tid_s1 = ? AND tid_s2 = ? AND tid_s3=? AND tid_s4=? AND" + " tid_s5 = ?"; diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataFinder.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataFinder.java index d780fea89..7fe81f0fc 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataFinder.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataFinder.java @@ -2,6 +2,7 @@ package com.a.eye.skywalking.storage.data; import com.a.eye.skywalking.logging.api.ILog; import com.a.eye.skywalking.logging.api.LogManager; +import com.a.eye.skywalking.network.grpc.TraceId; import com.a.eye.skywalking.storage.block.index.BlockIndexEngine; import com.a.eye.skywalking.storage.config.Config; import com.a.eye.skywalking.storage.config.Constants; @@ -21,12 +22,11 @@ import static com.a.eye.skywalking.storage.config.Constants.SQL.DEFAULT_USER; import static com.a.eye.skywalking.storage.util.PathResolver.getAbsolutePath; public class SpanDataFinder { - private static ILog logger = LogManager.getLogger(SpanDataFinder.class); private static IndexDataSourceCache datasourceCache = new IndexDataSourceCache(Config.Finder.CACHED_SIZE); private static ReentrantLock createDatasourceLock = new ReentrantLock(); - public static List find(String traceId) { - long blockIndex = BlockIndexEngine.newFinder().find(fetchStartTimeFromTraceId(traceId)); + public static List find(TraceId traceId) { + long blockIndex = BlockIndexEngine.newFinder().find(traceId.getSegments(1)); if (blockIndex == 0) { return new ArrayList(); } @@ -35,7 +35,8 @@ public class SpanDataFinder { IndexMetaCollection indexMetaCollection = null; try { indexDBConnector = fetchIndexDBConnector(blockIndex); - indexMetaCollection = indexDBConnector.queryByTraceId(spiltTraceId(traceId)); + indexMetaCollection = indexDBConnector.queryByTraceId(traceId.getSegmentsList().toArray(new Long[traceId + .getSegmentsCount()])); } finally { if (indexDBConnector != null) { indexDBConnector.close(); @@ -49,7 +50,7 @@ public class SpanDataFinder { Iterator> iterator = IndexMetaCollections.group(indexMetaCollection, new GroupKeyBuilder() { @Override public String buildKey(IndexMetaInfo metaInfo) { - return metaInfo.getFileName(); + return metaInfo.getFileName().fileName(); } }).iterator(); diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java index 327878477..ec22007a6 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java @@ -1,6 +1,5 @@ package com.a.eye.skywalking.storage.data.file; -import com.a.eye.datacarrier.common.AtomicRangeInteger; import com.a.eye.skywalking.health.report.HealthCollector; import com.a.eye.skywalking.health.report.HeathReading; import com.a.eye.skywalking.logging.api.ILog; @@ -16,9 +15,6 @@ import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; -import java.text.SimpleDateFormat; -import java.util.Date; -import java.util.concurrent.atomic.AtomicInteger; import static com.a.eye.skywalking.storage.util.PathResolver.getAbsolutePath; @@ -28,10 +24,10 @@ import static com.a.eye.skywalking.storage.util.PathResolver.getAbsolutePath; public class DataFile { private static ILog logger = LogManager.getLogger(DataFile.class); - private String fileName; + private DataFileNameDesc nameDesc; private long currentOffset; private DataFileOperator operator; - private static final AtomicRangeInteger DATA_FILE_NAME_SUFFIX = new AtomicRangeInteger(1000, 9999); + static { File dataFileDir = new File(getAbsolutePath(Config.DataFile.PATH)); @@ -41,21 +37,27 @@ public class DataFile { } public DataFile() { - this.fileName = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").format(new Date()) + "_" + DATA_FILE_NAME_SUFFIX - .getAndIncrement(); + this.nameDesc = new DataFileNameDesc(); this.currentOffset = 0; operator = new DataFileOperator(); createFile(); } - public DataFile(String fileName) { - this.fileName = fileName; + public DataFile(String fileName){ + this.nameDesc = new DataFileNameDesc(fileName); + this.currentOffset = 0; + operator = new DataFileOperator(); + createFile(); + } + + public DataFile(DataFileNameDesc nameDesc) { + this.nameDesc = nameDesc; operator = new DataFileOperator(); createFile(); } public DataFile(File file) { - this.fileName = file.getName(); + this.nameDesc = new DataFileNameDesc(file.getName()); this.currentOffset = file.length(); operator = new DataFileOperator(); createFile(); @@ -67,7 +69,7 @@ public class DataFile { try { dataFile.createNewFile(); if (logger.isDebugEnable()) { - logger.debug("Create an new data file[{}].", fileName); + logger.debug("Create an new data file[{}].", nameDesc.fileName()); } HealthCollector.getCurrentHeathReading("DataFile") .updateData(HeathReading.INFO, "Create an new data " + "file."); @@ -81,7 +83,7 @@ public class DataFile { public boolean overLimitLength() { boolean isOverLimitLength = currentOffset >= Config.DataFile.SIZE; if (isOverLimitLength) { - logger.info("Data File[{}] is over limit length.", fileName); + logger.info("Data File[{}] is over limit length.", nameDesc.fileName()); } return isOverLimitLength; } @@ -90,7 +92,7 @@ public class DataFile { byte[] bytes = data.toByteArray(); try { operator.getWriter().write(bytes); - IndexMetaInfo metaInfo = new IndexMetaInfo(data, fileName, currentOffset, bytes.length); + IndexMetaInfo metaInfo = new IndexMetaInfo(data, nameDesc, currentOffset, bytes.length); currentOffset += bytes.length; return metaInfo; } catch (IOException e) { @@ -118,7 +120,7 @@ public class DataFile { return data; } catch (IOException e) { throw new SpanDataReadFailedException( - "Failed to read dataFile[" + fileName + "], offset: " + offset + " " + "lenght: " + length, e); + "Failed to read dataFile[" + nameDesc.fileName() + "], offset: " + offset + " " + "lenght: " + length, e); } } @@ -169,12 +171,13 @@ public class DataFile { } } + private File getDataFile() { - return new File(getAbsolutePath(Config.DataFile.PATH), fileName); + return new File(getAbsolutePath(Config.DataFile.PATH), nameDesc.fileName()); } @Override public String toString() { - return "DataFile{" + "fileName='" + fileName + '\'' + '}'; + return "DataFile{" + "fileName='" + nameDesc.fileName() + '\'' + '}'; } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileNameDesc.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileNameDesc.java new file mode 100644 index 000000000..4a8cd6e07 --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileNameDesc.java @@ -0,0 +1,74 @@ +package com.a.eye.skywalking.storage.data.file; + +import com.a.eye.datacarrier.common.AtomicRangeInteger; + +import java.text.ParseException; +import java.text.SimpleDateFormat; + +/** + * Created by xin on 2016/11/16. + */ +public class DataFileNameDesc { + private static final AtomicRangeInteger DATA_FILE_NAME_SUFFIX = new AtomicRangeInteger(1000, 9999); + private long name; + private int suffix; + private String fileNameStr; + + public DataFileNameDesc() { + name = System.currentTimeMillis(); + suffix = DATA_FILE_NAME_SUFFIX.getAndIncrement(); + fileNameStr = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").format(name) + "_" + suffix; + } + + public DataFileNameDesc(long name, int suffix) { + this.name = name; + this.suffix = suffix; + fileNameStr = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").format(name) + "_" + suffix; + } + + public DataFileNameDesc(String fileName) { + int lastIndex = fileName.lastIndexOf('_'); + try { + this.name = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").parse(fileName.substring(0, lastIndex - 1)) + .getTime(); + } catch (ParseException e) { + } + this.suffix = Integer.parseInt(fileName.substring(lastIndex + 1)); + fileNameStr = fileName; + } + + + public String fileName() { + return fileNameStr; + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + + DataFileNameDesc that = (DataFileNameDesc) o; + + if (name != that.name) + return false; + return suffix == that.suffix; + + } + + @Override + public int hashCode() { + int result = (int) (name ^ (name >>> 32)); + result = 31 * result + suffix; + return result; + } + + public long getName() { + return name; + } + + public int getSuffix() { + return suffix; + } +} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java index 96f026ce4..802e6f436 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java @@ -5,6 +5,7 @@ import com.a.eye.skywalking.logging.api.LogManager; import com.a.eye.skywalking.storage.config.Config; import com.a.eye.skywalking.storage.config.Constants; import com.a.eye.skywalking.storage.data.exception.ConnectorInitializeFailedException; +import com.a.eye.skywalking.storage.data.file.DataFileNameDesc; import com.a.eye.skywalking.storage.data.spandata.AckSpanData; import com.a.eye.skywalking.storage.data.spandata.RequestSpanData; import com.a.eye.skywalking.storage.data.spandata.SpanData; @@ -109,9 +110,10 @@ public class IndexDBConnector { ps.setInt(5, metaInfo.getTraceId()[4].intValue()); ps.setInt(6, metaInfo.getTraceId()[5].intValue()); ps.setInt(7, metaInfo.getSpanType().getValue()); - ps.setString(8, metaInfo.getFileName()); - ps.setLong(9, metaInfo.getOffset()); - ps.setInt(10, metaInfo.getLength()); + ps.setLong(8, metaInfo.getFileName().getName()); + ps.setInt(9, metaInfo.getFileName().getSuffix()); + ps.setLong(10, metaInfo.getOffset()); + ps.setInt(11, metaInfo.getLength()); ps.addBatch(); if (++currentIndex > Constants.MAX_BATCH_SIZE) { ps.executeBatch(); @@ -144,15 +146,15 @@ public class IndexDBConnector { return indexSize; } - public IndexMetaCollection queryByTraceId(long[] traceId) { + public IndexMetaCollection queryByTraceId(Long[] traceId) { try { PreparedStatement ps = connection.prepareStatement(QUERY_TRACE_ID); - ps.setInt(1, (int) traceId[0]); - ps.setLong(2, (int) traceId[1]); - ps.setInt(3, (int) traceId[2]); - ps.setInt(4, (int) traceId[3]); - ps.setInt(5, (int) traceId[4]); - ps.setInt(6, (int) traceId[5]); + ps.setInt(1, traceId[0].intValue()); + ps.setLong(2, traceId[1]); + ps.setInt(3, traceId[2].intValue()); + ps.setInt(4, traceId[3].intValue()); + ps.setInt(5, traceId[4].intValue()); + ps.setInt(6, traceId[5].intValue()); ResultSet rs = ps.executeQuery(); IndexMetaCollection collection = new IndexMetaCollection(); @@ -166,8 +168,9 @@ public class IndexDBConnector { spanData = new RequestSpanData(); } - collection.add(new IndexMetaInfo(spanData, rs.getString("file_name"), rs.getLong("offset"), - rs.getInt("length"))); + collection.add(new IndexMetaInfo(spanData, + new DataFileNameDesc(rs.getLong("file_name"), rs.getInt("file_name_suffix")), + rs.getLong("offset"), rs.getInt("length"))); } return collection; } catch (SQLException e) { @@ -183,4 +186,6 @@ public class IndexDBConnector { logger.error("Failed to close index db connector", e); } } + + } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java index 71030b997..af4d6f675 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java @@ -1,5 +1,6 @@ package com.a.eye.skywalking.storage.data.index; +import com.a.eye.skywalking.storage.data.file.DataFileNameDesc; import com.a.eye.skywalking.storage.data.spandata.SpanData; import com.a.eye.skywalking.storage.data.spandata.SpanType; @@ -7,21 +8,21 @@ public class IndexMetaInfo { private SpanData spanData; - private String fileName; + private DataFileNameDesc nameDesc; private long offset; private int length; - public IndexMetaInfo(SpanData data, String fileName, long offset, int length) { + public IndexMetaInfo(SpanData data, DataFileNameDesc fileNameDesc, long offset, int length) { this.spanData = data; - this.fileName = fileName; + this.nameDesc = fileNameDesc; this.offset = offset; this.length = length; } - public String getFileName() { - return fileName; + public DataFileNameDesc getFileName() { + return nameDesc; } public long getOffset() { diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/listener/SearchListener.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/listener/SearchListener.java index bbb0ae63f..0bd8e8d27 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/listener/SearchListener.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/listener/SearchListener.java @@ -5,6 +5,8 @@ import com.a.eye.skywalking.health.report.HeathReading; import com.a.eye.skywalking.logging.api.ILog; import com.a.eye.skywalking.logging.api.LogManager; import com.a.eye.skywalking.network.grpc.Span; +import com.a.eye.skywalking.network.grpc.TraceId; +import com.a.eye.skywalking.network.listener.AsyncTraceSearchListener; import com.a.eye.skywalking.network.listener.TraceSearchListener; import com.a.eye.skywalking.storage.data.SpanDataFinder; import com.a.eye.skywalking.storage.data.spandata.SpanData; @@ -13,12 +15,12 @@ import com.a.eye.skywalking.storage.data.spandata.SpanDataHelper; import java.util.ArrayList; import java.util.List; -public class SearchListener implements TraceSearchListener { +public class SearchListener implements AsyncTraceSearchListener { private static ILog logger = LogManager.getLogger(SearchListener.class); @Override - public List search(String traceId) { + public List search(TraceId traceId) { try { List data = SpanDataFinder.find(traceId); SpanDataHelper helper = new SpanDataHelper(data); diff --git a/skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/storage/SearchClient.java b/skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/storage/SearchClient.java new file mode 100644 index 000000000..70d92127a --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/storage/SearchClient.java @@ -0,0 +1,46 @@ +package com.a.eye.skywalking.storage; + +import com.a.eye.skywalking.network.dependencies.io.grpc.ManagedChannel; +import com.a.eye.skywalking.network.dependencies.io.grpc.ManagedChannelBuilder; +import com.a.eye.skywalking.network.dependencies.io.grpc.stub.StreamObserver; +import com.a.eye.skywalking.network.grpc.*; + +import static com.a.eye.skywalking.network.grpc.AsyncTraceSearchServiceGrpc.newStub; + + +public class SearchClient { + private static ManagedChannel channel = + ManagedChannelBuilder.forAddress("127.0.0.1", 34000).usePlaintext(true).build(); + + private static AsyncTraceSearchServiceGrpc.AsyncTraceSearchServiceStub searchServiceStub = newStub(channel); + + + + public static void main(String[] args) throws InterruptedException { + StreamObserver serverStreamObserver = new StreamObserver() { + @Override + public void onNext(SearchResult searchResult) { + System.out.println(searchResult.getSpansCount()); + } + + @Override + public void onError(Throwable throwable) { + + } + + @Override + public void onCompleted() { + + } + }; + StreamObserver searchResult = searchServiceStub.search(serverStreamObserver); + + searchResult.onNext(QueryTask.newBuilder().setTraceId( + TraceId.newBuilder().addSegments(201611).addSegments(1478661327960L).addSegments(8504828) + .addSegments(2277).addSegments(53).addSegments(3).build()).setTaskId(1).build()); + searchResult.onCompleted(); + + Thread.sleep(10000); + + } +}