diff --git a/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/server/AsyncTraceSearchServer.java b/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/server/AsyncTraceSearchServer.java index 8561ff3ac..df53b3398 100644 --- a/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/server/AsyncTraceSearchServer.java +++ b/skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/server/AsyncTraceSearchServer.java @@ -24,18 +24,15 @@ public class AsyncTraceSearchServer extends AsyncTraceSearchServiceGrpc.AsyncTra 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) { SearchResult.Builder builder = SearchResult.newBuilder(); - builder = builder.setTaskId(taskId); responseObserver.onNext(builder.build()); responseObserver.onCompleted(); } @@ -46,7 +43,6 @@ public class AsyncTraceSearchServer extends AsyncTraceSearchServiceGrpc.AsyncTra if(spans != null) { builder = builder.addAllSpans(spans); } - builder = builder.setTaskId(taskId); responseObserver.onNext(builder.build()); responseObserver.onCompleted(); } diff --git a/skywalking-network/src/main/proto/TraceSearchService.proto b/skywalking-network/src/main/proto/TraceSearchService.proto index 4636aa9d4..82c4af372 100644 --- a/skywalking-network/src/main/proto/TraceSearchService.proto +++ b/skywalking-network/src/main/proto/TraceSearchService.proto @@ -16,11 +16,9 @@ service TraceSearchService { } message QueryTask { - int32 taskId = 1; TraceId traceId = 2; } message SearchResult { - int32 taskId = 1; repeated Span spans = 2; } diff --git a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/Main.java b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/Main.java index 0cfacf11f..7d3f0917a 100644 --- a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/Main.java +++ b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/Main.java @@ -8,6 +8,7 @@ import com.a.eye.skywalking.network.Server; import com.a.eye.skywalking.routing.config.Config; import com.a.eye.skywalking.routing.config.ConfigInitializer; import com.a.eye.skywalking.routing.listener.SpanStorageListenerImpl; +import com.a.eye.skywalking.routing.listener.TraceSearchListenerImpl; import com.a.eye.skywalking.routing.router.RoutingService; import com.a.eye.skywalking.routing.storage.listener.NotifyListenerImpl; @@ -25,12 +26,12 @@ public class Main { LogManager.setLogResolver(new Log4j2Resolver()); new NotifyListenerImpl(Config.StorageNode.SUBSCRIBE_PATH, RoutingService.getRouter()); - Server.newBuilder(Config.Routing.PORT).addSpanStorageService(new SpanStorageListenerImpl()).build().start(); + Server.newBuilder(Config.Routing.PORT).addSpanStorageService(new SpanStorageListenerImpl()).addTraceSearchService(new TraceSearchListenerImpl()).build().start(); logger.info("Skywalking routing service was started."); Thread.currentThread().join(); } catch (Exception e) { logger.error("Failed to start routing service.", e); - }finally { + } finally { RoutingService.stop(); } } diff --git a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/client/StorageClientCachePool.java b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/client/StorageClientCachePool.java new file mode 100644 index 000000000..358e377d3 --- /dev/null +++ b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/client/StorageClientCachePool.java @@ -0,0 +1,69 @@ +package com.a.eye.skywalking.routing.client; + +import com.a.eye.skywalking.network.Client; +import com.a.eye.skywalking.network.grpc.client.SpanStorageClient; +import com.a.eye.skywalking.network.grpc.client.TraceSearchClient; +import com.a.eye.skywalking.network.listener.client.StorageClientListener; + +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; + +/** + * Created by wusheng on 2016/12/3. + */ +public class StorageClientCachePool { + public static StorageClientCachePool INSTANCE = new StorageClientCachePool(); + + private Map pool = new ConcurrentHashMap<>(); + private Map spanStorageClientPool = new ConcurrentHashMap<>(); + private Map traceSearchClientPool = new ConcurrentHashMap<>(); + private ReentrantLock lock = new ReentrantLock(); + + private StorageClientCachePool() { + } + + public Collection getClients() { + return traceSearchClientPool.values(); + } + + public SpanStorageClient getSpanStorageClient(String connectionURL, StorageClientListener listener) { + SpanStorageClient spanStorageClient = spanStorageClientPool.get(connectionURL); + if (spanStorageClient != null) { + return spanStorageClient; + } + + lock.lock(); + try { + spanStorageClient = spanStorageClientPool.get(connectionURL); + if (spanStorageClient == null) { + String[] urlSegment = connectionURL.split(":"); + if (urlSegment.length != 2) { + throw new IllegalArgumentException(); + } + Client client = new Client(urlSegment[0], Integer.valueOf(urlSegment[1])); + pool.put(connectionURL, client); + spanStorageClientPool.put(connectionURL, client.newSpanStorageClient(listener)); + traceSearchClientPool.put(connectionURL, client.newTraceSearchClient()); + } + return spanStorageClient; + } finally { + lock.unlock(); + } + } + + public void shutdown(String connectionURL) { + lock.lock(); + try { + if (pool.containsKey(connectionURL)) { + Client client = pool.remove(connectionURL); + client.shutdown(); + spanStorageClientPool.remove(connectionURL); + traceSearchClientPool.remove(connectionURL); + } + } finally { + lock.unlock(); + } + } +} diff --git a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/config/Config.java b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/config/Config.java index 9b0807752..f0bb1d6b4 100644 --- a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/config/Config.java +++ b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/config/Config.java @@ -5,17 +5,26 @@ public class Config { public static int PORT = 23000; } + + public static class Search { + public static long CHECK_CYCLE = 100L; + public static long TIMEOUT = 3 * 1000L; + } + + public static class RegistryCenter { - public static String TYPE = "zookeeper"; + public static String TYPE = "zookeeper"; public static String CONNECT_URL = "127.0.0.1:2181"; public static String AUTH_SCHEMA = ""; - public static String AUTH_INFO = ""; + public static String AUTH_INFO = ""; } + public static class StorageNode { public static String SUBSCRIBE_PATH = "/skywalking/storage_list"; } + public static class Disruptor { public static int BUFFER_SIZE = 1024 * 128 * 4; diff --git a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/disruptor/AbstractRouteSpanEventHandler.java b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/disruptor/AbstractRouteSpanEventHandler.java index 9bbbfb569..9f94848fb 100644 --- a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/disruptor/AbstractRouteSpanEventHandler.java +++ b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/disruptor/AbstractRouteSpanEventHandler.java @@ -6,6 +6,7 @@ import com.a.eye.skywalking.network.Client; import com.a.eye.skywalking.network.grpc.RequestSpan; import com.a.eye.skywalking.network.grpc.client.SpanStorageClient; import com.a.eye.skywalking.network.listener.client.StorageClientListener; +import com.a.eye.skywalking.routing.client.StorageClientCachePool; import com.a.eye.skywalking.routing.config.Config; import com.lmax.disruptor.EventHandler; @@ -16,22 +17,15 @@ import java.util.List; * Created by xin on 2016/11/29. */ public abstract class AbstractRouteSpanEventHandler implements EventHandler { - protected Client client; + protected SpanStorageClient client; protected final int bufferSize; protected boolean stop; private volatile boolean previousSendFinish = true; + private StorageClientListener listener; public AbstractRouteSpanEventHandler(String connectionURL) { bufferSize = Config.Disruptor.FLUSH_SIZE; - String[] urlSegment = connectionURL.split(":"); - if (urlSegment.length != 2) { - throw new IllegalArgumentException(); - } - client = new Client(urlSegment[0], Integer.valueOf(urlSegment[1])); - } - - protected SpanStorageClient getStorageClient() { - SpanStorageClient spanStorageClient = client.newSpanStorageClient(new StorageClientListener() { + listener = new StorageClientListener() { @Override public void onError(Throwable throwable) { previousSendFinish = true; @@ -43,9 +37,13 @@ public abstract class AbstractRouteSpanEventHandler implements EventHandler search(TraceId traceId) { + Collection clients = StorageClientCachePool.INSTANCE.getClients(); + CountDownLatch countDownLatch = new CountDownLatch(clients.size()); + ConcurrentHashMap results = new ConcurrentHashMap<>(); + for (TraceSearchClient client : clients) { + client.search(QueryTask.newBuilder().setTraceId(traceId).build(), new SearchClientListener(){ + + @Override + public void onError(Throwable throwable) { + + } + + @Override + public void onReturn(SearchResult searchResult) { + results.put(searchResult, true); + countDownLatch.countDown(); + } + + @Override + public void onFinished() { + + } + }); + } + + long waitTime = 0; + while(countDownLatch.getCount() != 0){ + try { + Thread.sleep(Config.Search.CHECK_CYCLE); + } catch (InterruptedException e) { + } + waitTime += Config.Search.CHECK_CYCLE; + + if(waitTime > Config.Search.TIMEOUT){ + break; + } + } + + List spans = new ArrayList<>(30); + for (SearchResult searchResult : results.keySet()) { + List result = searchResult.getSpansList(); + if(result != null && result.size() > 0) { + spans.addAll(result); + } + } + + return spans; + } +} diff --git a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/router/Router.java b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/router/Router.java index dde30e3d6..6a55edb49 100644 --- a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/router/Router.java +++ b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/router/Router.java @@ -5,6 +5,7 @@ import com.a.eye.skywalking.logging.api.LogManager; import com.a.eye.skywalking.network.grpc.AckSpan; import com.a.eye.skywalking.network.grpc.RequestSpan; import com.a.eye.skywalking.registry.api.RegistryNode; +import com.a.eye.skywalking.routing.client.StorageClientCachePool; import com.a.eye.skywalking.routing.disruptor.NoopSpanDisruptor; import com.a.eye.skywalking.routing.disruptor.SpanDisruptor; import com.a.eye.skywalking.routing.storage.listener.NodeChangesListener; @@ -52,19 +53,16 @@ public class Router implements NodeChangesListener { } } - Collections.sort(newDisruptors, new Comparator() { - @Override - public int compare(SpanDisruptor o1, SpanDisruptor o2) { - long o1Key = Long.parseLong(o1.getConnectionURL().replace(".", "").replace(":", "")); - long o2Key = Long.parseLong(o2.getConnectionURL().replace(".", "").replace(":", "")); + Collections.sort(newDisruptors, (o1, o2) -> { + long o1Key = Long.parseLong(o1.getConnectionURL().replace(".", "").replace(":", "")); + long o2Key = Long.parseLong(o2.getConnectionURL().replace(".", "").replace(":", "")); - if (o1Key == o2Key) { - return 0; - } else if (o1Key > o2Key) { - return 1; - } else { - return -1; - } + if (o1Key == o2Key) { + return 0; + } else if (o1Key > o2Key) { + return 1; + } else { + return -1; } }); @@ -74,6 +72,7 @@ public class Router implements NodeChangesListener { // 而后stop for (SpanDisruptor removedDisruptor : removedDisruptors) { removedDisruptor.shutdown(); + StorageClientCachePool.INSTANCE.shutdown(removedDisruptor.getConnectionURL()); } } diff --git a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/storage/listener/NotifyListenerImpl.java b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/storage/listener/NotifyListenerImpl.java index d2696d976..111408602 100644 --- a/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/storage/listener/NotifyListenerImpl.java +++ b/skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/storage/listener/NotifyListenerImpl.java @@ -18,7 +18,6 @@ import static com.a.eye.skywalking.routing.storage.listener.NotifyListenerImpl.C public class NotifyListenerImpl implements NotifyListener { private NodeChangesListener listener; - private List childrenConnectionURLOfPreviousChanged = new ArrayList<>(); private ReentrantLock lock = new ReentrantLock(); public NotifyListenerImpl(String subscribePath, NodeChangesListener listener) {