diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java index 61fd9d1da..d82704000 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java @@ -52,27 +52,24 @@ public class RemoteSenderService implements Service { public void send(String nextWorkName, StreamData streamData, Selector selector) { RemoteClientManager clientManager = moduleManager.find(CoreModule.NAME).provider().getService(RemoteClientManager.class); + RemoteClient remoteClient = null; List clientList = clientManager.getRemoteClient(); if (clientList.size() == 0) { logger.warn("There is no available remote server for now, ignore the streaming data until the cluster metadata initialized."); return; } - - RemoteClient remoteClient; switch (selector) { case HashCode: remoteClient = hashCodeSelector.select(clientList, streamData); - remoteClient.push(nextWorkName, streamData); break; case Rolling: remoteClient = rollingSelector.select(clientList, streamData); - remoteClient.push(nextWorkName, streamData); break; case ForeverFirst: remoteClient = foreverFirstSelector.select(clientList, streamData); - remoteClient.push(nextWorkName, streamData); break; } + remoteClient.push(nextWorkName, streamData); } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java index b585150ae..852bdd3ab 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java @@ -18,9 +18,10 @@ package org.apache.skywalking.oap.server.core.remote.client; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Sets; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.LinkedList; import java.util.List; @@ -29,6 +30,10 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.Setter; import org.apache.skywalking.oap.server.core.cluster.ClusterModule; import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery; import org.apache.skywalking.oap.server.core.cluster.RemoteInstance; @@ -53,27 +58,24 @@ public class RemoteClientManager implements Service { private final ModuleDefineHolder moduleDefineHolder; private ClusterNodesQuery clusterNodesQuery; - private final List clientsA; - private final List clientsB; private volatile List usingClients; private GaugeMetrics gauge; private int remoteTimeout; /** * Initial the manager for all remote communication clients. + * * @param moduleDefineHolder for looking up other modules - * @param remoteTimeout for cluster internal communication, in second unit. + * @param remoteTimeout for cluster internal communication, in second unit. */ public RemoteClientManager(ModuleDefineHolder moduleDefineHolder, int remoteTimeout) { this.moduleDefineHolder = moduleDefineHolder; - this.clientsA = new LinkedList<>(); - this.clientsB = new LinkedList<>(); - this.usingClients = clientsA; + this.usingClients = ImmutableList.of(); this.remoteTimeout = remoteTimeout; } public void start() { - Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(this::refresh, 1, 5, TimeUnit.SECONDS); + Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(this::refresh, 1, 5, TimeUnit.SECONDS); } /** @@ -128,7 +130,7 @@ public class RemoteClientManager implements Service { private void printRemoteClientList() { if (logger.isDebugEnabled()) { StringBuilder addresses = new StringBuilder(); - getRemoteClient().forEach(client -> addresses.append(client.getAddress().toString()).append(",")); + this.usingClients.forEach(client -> addresses.append(client.getAddress().toString()).append(",")); logger.debug("Remote client list: {}", addresses); } } @@ -157,77 +159,58 @@ public class RemoteClientManager implements Service { return usingClients; } - private List getFreeClients() { - if (usingClients.equals(clientsA)) { - return clientsB; - } else { - return clientsA; - } - } - - private void switchCurrentClients() { - if (usingClients.equals(clientsA)) { - usingClients = clientsB; - } else { - usingClients = clientsA; - } - } - /** * Compare clients between exist clients and remote instance collection. Move the clients into new client collection * which are alive to avoid create a new channel. Shutdown the clients which could not find in cluster config. - * + *

* Create a gRPC client for remote instance except for self-instance. * * @param remoteInstances Remote instance collection by query cluster config. */ - private synchronized void reBuildRemoteClients(List remoteInstances) { - getFreeClients().clear(); + private void reBuildRemoteClients(List remoteInstances) { + final Map remoteClientCollection = this.usingClients.stream() + .collect(Collectors.toMap(RemoteClient::getAddress, client -> new RemoteClientAction(client, Action.Close))); - Map remoteClients = new HashMap<>(); - getRemoteClient().forEach(client -> remoteClients.put(client.getAddress(), client)); + final Map latestRemoteClients = remoteInstances.stream() + .collect(Collectors.toMap(RemoteInstance::getAddress, remote -> new RemoteClientAction(null, Action.Create))); - Map tempRemoteClients = new HashMap<>(); - getRemoteClient().forEach(client -> tempRemoteClients.put(client.getAddress(), Action.Close)); + final Set

unChangeAddresses = Sets.intersection(remoteClientCollection.keySet(), latestRemoteClients.keySet()); - remoteInstances.forEach(remoteInstance -> { - if (tempRemoteClients.containsKey(remoteInstance.getAddress())) { - tempRemoteClients.put(remoteInstance.getAddress(), Action.Leave); - } else { - tempRemoteClients.put(remoteInstance.getAddress(), Action.Create); - } - }); + unChangeAddresses.stream() + .filter(remoteClientCollection::containsKey) + .forEach(unChangeAddress -> remoteClientCollection.get(unChangeAddress).setAction(Action.Unchanged)); - tempRemoteClients.forEach((address, action) -> { - switch (action) { - case Leave: - if (remoteClients.containsKey(address)) { - getFreeClients().add(remoteClients.get(address)); - } + // make the latestRemoteClients including the new clients only + unChangeAddresses.forEach(latestRemoteClients::remove); + remoteClientCollection.putAll(latestRemoteClients); + + final List newRemoteClients = new LinkedList<>(); + remoteClientCollection.forEach((address, clientAction) -> { + switch (clientAction.getAction()) { + case Unchanged: + newRemoteClients.add(clientAction.getRemoteClient()); break; case Create: if (address.isSelf()) { RemoteClient client = new SelfRemoteClient(moduleDefineHolder, address); - getFreeClients().add(client); + newRemoteClients.add(client); } else { RemoteClient client = new GRPCRemoteClient(moduleDefineHolder, address, 1, 3000, remoteTimeout); client.connect(); - getFreeClients().add(client); + newRemoteClients.add(client); } break; } }); - Collections.sort(getFreeClients()); - switchCurrentClients(); + //for stable ordering for rolling selector + Collections.sort(newRemoteClients); + this.usingClients = ImmutableList.copyOf(newRemoteClients); - tempRemoteClients.forEach((address, action) -> { - if (Action.Close.equals(action) && remoteClients.containsKey(address)) { - remoteClients.get(address).close(); - } - }); - - getFreeClients().clear(); + remoteClientCollection.values() + .stream() + .filter(remoteClientAction -> remoteClientAction.getAction().equals(Action.Close)) + .forEach(remoteClientAction -> remoteClientAction.getRemoteClient().close()); } private boolean compare(List remoteInstances) { @@ -244,6 +227,15 @@ public class RemoteClientManager implements Service { } enum Action { - Close, Leave, Create + Close, Unchanged, Create + } + + @Getter + @AllArgsConstructor + static private class RemoteClientAction { + private RemoteClient remoteClient; + + @Setter + private Action action; } } diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java index c561273b1..8c825538d 100644 --- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java @@ -18,8 +18,7 @@ package org.apache.skywalking.oap.server.core.remote.client; -import java.util.ArrayList; -import java.util.List; +import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.cluster.ClusterModule; import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery; @@ -30,19 +29,31 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator; import org.apache.skywalking.oap.server.testing.module.ModuleDefineTesting; import org.apache.skywalking.oap.server.testing.module.ModuleManagerTesting; import org.junit.Assert; +import org.junit.Before; import org.junit.Test; +import org.mockito.internal.verification.AtLeast; -import static org.mockito.Mockito.any; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import static org.mockito.Mockito.*; /** * @author peng-yongsheng */ +@Slf4j public class RemoteClientManagerTestCase { - @Test - public void refresh() { + private RemoteClientManager clientManager; + private ClusterNodesQuery clusterNodesQuery; + + @Before + public void setup() { ModuleManagerTesting moduleManager = new ModuleManagerTesting(); ModuleDefineTesting clusterModuleDefine = new ModuleDefineTesting(); moduleManager.put(ClusterModule.NAME, clusterModuleDefine); @@ -50,7 +61,7 @@ public class RemoteClientManagerTestCase { ModuleDefineTesting coreModuleDefine = new ModuleDefineTesting(); moduleManager.put(CoreModule.NAME, coreModuleDefine); - ClusterNodesQuery clusterNodesQuery = mock(ClusterNodesQuery.class); + this.clusterNodesQuery = mock(ClusterNodesQuery.class); clusterModuleDefine.provider().registerServiceImplementation(ClusterNodesQuery.class, clusterNodesQuery); @@ -80,8 +91,11 @@ public class RemoteClientManagerTestCase { moduleManager.put(TelemetryModule.NAME, telemetryModuleDefine); telemetryModuleDefine.provider().registerServiceImplementation(MetricsCreator.class, metricsCreator); - RemoteClientManager clientManager = new RemoteClientManager(moduleManager, 10); + this.clientManager = spy(new RemoteClientManager(moduleManager, 10)); + } + @Test + public void refresh() { when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupOneInstances()); clientManager.refresh(); @@ -125,4 +139,99 @@ public class RemoteClientManagerTestCase { instances.add(new RemoteInstance(new Address("host4", 100, false))); return instances; } + + @Test + public void testConcurrenceGetRemoteClientAndRefresh() throws Exception { + this.refresh(); //guarantee has any client in clientManager + + CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> { + log.debug("begin concurrency test"); + }); + + final ExecutorService executorService = Executors.newFixedThreadPool(3); + final Future refreshFuture = executorService.submit(() -> { + try { + cyclicBarrier.await(); + this.refresh(); + } catch (InterruptedException | BrokenBarrierException e) { + e.printStackTrace(); + } + }); + + executorService.submit(() -> { + try { + int i = 0; + cyclicBarrier.await(); + while (!refreshFuture.isDone()) { + Assert.assertFalse(this.clientManager.getRemoteClient().isEmpty()); + log.debug("thread {} invoke {} times", Thread.currentThread().getName(), i++); + } + } catch (InterruptedException | BrokenBarrierException e) { + e.printStackTrace(); + } + }); + + try { + int i = 0; + cyclicBarrier.await(); + while (!refreshFuture.isDone()) { + Assert.assertFalse(this.clientManager.getRemoteClient().isEmpty()); + log.debug("thread {} invoke {} times", Thread.currentThread().getName(), i++); + } + } catch (InterruptedException | BrokenBarrierException e) { + e.printStackTrace(); + } + + verify(this.clientManager, new AtLeast(2)).getRemoteClient(); + } + + @Test + public void testGetRemoteClientAndNeverChange() { + when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupOneInstances()); + this.clientManager.refresh(); + final List gotGroupOneInstances = this.clientManager.getRemoteClient(); + + when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupTwoInstances()); + this.clientManager.refresh(); + final List gotGroupTwoInstances = this.clientManager.getRemoteClient(); + + Assert.assertEquals(gotGroupOneInstances.size(), groupOneInstances().size()); + Assert.assertEquals(gotGroupTwoInstances.size(), groupTwoInstances().size()); + Assert.assertNotEquals(gotGroupOneInstances.size(), gotGroupTwoInstances.size()); + } + + @Test + public void testCompare() { + when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupOneInstances()); + clientManager.refresh(); + + List groupOneRemoteClients = clientManager.getRemoteClient(); + + + when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupOneInstances()); + clientManager.refresh(); + + List newGroupOneRemoteClients = clientManager.getRemoteClient(); + + Assert.assertArrayEquals(groupOneRemoteClients.toArray(), newGroupOneRemoteClients.toArray()); + } + + @Test + public void testUnChangeRefresh() { + final List groupOneInstances = groupOneInstances(); + when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupOneInstances); + clientManager.refresh(); + + List groupOneRemoteClients = clientManager.getRemoteClient(); + + groupOneInstances.add(new RemoteInstance(new Address("host4", 100, false))); + when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupOneInstances); + clientManager.refresh(); + + List newGroupOneRemoteClients = clientManager.getRemoteClient(); + + + Assert.assertEquals(groupOneRemoteClients.get(0).getAddress(), newGroupOneRemoteClients.get(0).getAddress()); + Assert.assertEquals(newGroupOneRemoteClients.get(3).getAddress().getHost(), "host4"); + } }