<commit_before>/*
 * Copyright © 2016 - 2017 VMware, Inc. All Rights Reserved.
 *
 * Licensed 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 com.vrg.rapid;

import com.google.common.net.HostAndPort;
import com.vrg.rapid.pb.MembershipServiceGrpc;
import io.grpc.ClientInterceptor;
import io.grpc.MethodDescriptor;
import io.grpc.ServerInterceptor;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * Test public API
 */
public class ClusterTest {
    @SuppressWarnings("FieldCanBeLocal")
    @Nullable private static Logger grpcLogger = null;
    @Nullable private static Logger nettyLogger = null;
    private static final int RPC_TIMEOUT_SHORT_MS = 100;
    private final Map<HostAndPort, Cluster> instances = new ConcurrentHashMap<>();
    private final Map<HostAndPort, StaticFailureDetector> staticFds = new ConcurrentHashMap<>();
    private final Map<HostAndPort, List<ServerInterceptor>> serverInterceptors = new ConcurrentHashMap<>();
    private final Map<HostAndPort, List<ClientInterceptor>> clientInterceptors = new ConcurrentHashMap<>();
    private boolean useStaticFd = false;
    @Nullable private Random random = null;
    private long seed;
    private int basePort;
    @Nullable private AtomicInteger portCounter = null;

    @Rule
    public final TestWatcher testWatcher = new TestWatcher() {
        @Override
        protected void failed(final Throwable e, final Description description) {

            System.out.println("\u001B[31m     [FAILED] [Seed: " + seed + "] " + description.getMethodName()
                               + "\u001B[0m");
        }
    };

    @BeforeClass
    public static void beforeClass() {
        // gRPC INFO logs clutter the test output
        grpcLogger = Logger.getLogger("io.grpc");
        grpcLogger.setLevel(Level.WARNING);
        nettyLogger = Logger.getLogger("io.grpc.netty.NettyServerHandler");
        nettyLogger.setLevel(Level.OFF);
    }

    @Before
    public void beforeTest() {
        basePort =  1234;
        portCounter = new AtomicInteger(basePort);
        instances.clear();
        seed = ThreadLocalRandom.current().nextLong();
        random = new Random(seed);

        // Tests need to opt out of the in-process channel
        RpcServer.USE_IN_PROCESS_SERVER = true;
        RpcClient.USE_IN_PROCESS_CHANNEL = true;

        // Tests that depend on failure detection should set intervals by themselves
        MembershipService.FAILURE_DETECTOR_INITIAL_DELAY_IN_MS = 100000;
        MembershipService.FAILURE_DETECTOR_INTERVAL_IN_MS = 100000;
        RpcClient.Conf.RPC_JOIN_PHASE_2_TIMEOUT = RpcClient.Conf.RPC_TIMEOUT_MEDIUM_MS * 20;
        RpcClient.Conf.RPC_TIMEOUT_MS = RpcClient.Conf.RPC_TIMEOUT_MEDIUM_MS;
        RpcClient.Conf.RPC_PROBE_TIMEOUT = 5000;

        useStaticFd = false;
        staticFds.clear();
        serverInterceptors.clear();
        clientInterceptors.clear();
    }

    @After
    public void afterTest() throws InterruptedException {
        for (final Cluster cluster: instances.values()) {
            cluster.shutdown();
        }
    }

    /**
     * Test with a single node joining through a seed.
     */
    @Test
    public void singleNodeJoinsThroughSeed() throws IOException, InterruptedException, ExecutionException {
        RpcServer.USE_IN_PROCESS_SERVER = false;
        RpcClient.USE_IN_PROCESS_CHANNEL = false;

        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(1, seedHost);
        verifyCluster(1, seedHost);
        extendCluster(1, seedHost);
        verifyCluster(2, seedHost);
    }

    /**
     * Test with K nodes joining the network through a single seed.
     */
    @Test
    public void tenNodesJoinSequentially() throws IOException, InterruptedException {
        // Explicitly test netty
        RpcServer.USE_IN_PROCESS_SERVER = false;
        RpcClient.USE_IN_PROCESS_CHANNEL = false;

        final int numNodes = 10;
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(1, seedHost); // Only bootstrap a seed.
        verifyCluster(1, seedHost);
        for (int i = 0; i < numNodes; i++) {
            extendCluster(1, seedHost);
            waitAndVerifyAgreement( i + 2, 5, 1000, seedHost);
        }
    }

    /**
     * Identical to the previous test, but with more than K nodes joining in serial.
     */
    @Test
    public void twentyNodesJoinSequentially() throws IOException, InterruptedException {
        // Explicitly test netty
        RpcServer.USE_IN_PROCESS_SERVER = false;
        RpcClient.USE_IN_PROCESS_CHANNEL = false;

        final int numNodes = 20;
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(1, seedHost); // Only bootstrap a seed.
        verifyCluster(1, seedHost);

        for (int i = 0; i < numNodes; i++) {
            extendCluster(1, seedHost);
            waitAndVerifyAgreement( i + 2, 5, 1000, seedHost);
        }
    }

    /**
     * Identical to the previous test, but with more than K nodes joining in parallel.
     * <p>
     * The test starts with a single seed and all N - 1 subsequent nodes initiate their join protocol at the same
     * time. This tests a single seed's ability to bootstrap a large cluster in one step.
     */
    @Test(timeout=150000)
    public void fiveHundredNodesJoinInParallel() throws IOException, InterruptedException {
        final int numNodes = 500; // Includes the size of the cluster
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(numNodes, seedHost);
        verifyCluster(numNodes, seedHost);
    }


    /**
     * This test starts with a single seed, and a wave where 50 subsequent nodes initiate their join protocol
     * concurrently. Following this, a subsequent wave begins where 100 nodes then start together.
     */
    @Test(timeout=150000)
    public void hundredNodesJoinFiftyNodeCluster() throws IOException, InterruptedException {
        RpcServer.USE_IN_PROCESS_SERVER = true;
        RpcClient.USE_IN_PROCESS_CHANNEL = true;

        final int numNodesPhase1 = 50;
        final int numNodesPhase2 = 100;
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(numNodesPhase1, seedHost);
        waitAndVerifyAgreement(numNodesPhase1, 10, 100, seedHost);
        extendCluster(numNodesPhase2, seedHost);
        waitAndVerifyAgreement(numNodesPhase1 + numNodesPhase2, 10, 1000, seedHost);
    }

    /**
     * This test starts with a 4 node cluster. We then fail a single node to see if the monitoring mechanism
     * identifies the failing node and arrives at a decision to remove it.
     */
    @Test
    public void oneFailureOutOfFourNodes() throws IOException, InterruptedException {<commit_msg>Reduce 500 node join test to 300<commit_after>/*
 * Copyright © 2016 - 2017 VMware, Inc. All Rights Reserved.
 *
 * Licensed 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 com.vrg.rapid;

import com.google.common.net.HostAndPort;
import com.vrg.rapid.pb.MembershipServiceGrpc;
import io.grpc.ClientInterceptor;
import io.grpc.MethodDescriptor;
import io.grpc.ServerInterceptor;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * Test public API
 */
public class ClusterTest {
    @SuppressWarnings("FieldCanBeLocal")
    @Nullable private static Logger grpcLogger = null;
    @Nullable private static Logger nettyLogger = null;
    private static final int RPC_TIMEOUT_SHORT_MS = 100;
    private final Map<HostAndPort, Cluster> instances = new ConcurrentHashMap<>();
    private final Map<HostAndPort, StaticFailureDetector> staticFds = new ConcurrentHashMap<>();
    private final Map<HostAndPort, List<ServerInterceptor>> serverInterceptors = new ConcurrentHashMap<>();
    private final Map<HostAndPort, List<ClientInterceptor>> clientInterceptors = new ConcurrentHashMap<>();
    private boolean useStaticFd = false;
    @Nullable private Random random = null;
    private long seed;
    private int basePort;
    @Nullable private AtomicInteger portCounter = null;

    @Rule
    public final TestWatcher testWatcher = new TestWatcher() {
        @Override
        protected void failed(final Throwable e, final Description description) {

            System.out.println("\u001B[31m     [FAILED] [Seed: " + seed + "] " + description.getMethodName()
                               + "\u001B[0m");
        }
    };

    @BeforeClass
    public static void beforeClass() {
        // gRPC INFO logs clutter the test output
        grpcLogger = Logger.getLogger("io.grpc");
        grpcLogger.setLevel(Level.WARNING);
        nettyLogger = Logger.getLogger("io.grpc.netty.NettyServerHandler");
        nettyLogger.setLevel(Level.OFF);
    }

    @Before
    public void beforeTest() {
        basePort =  1234;
        portCounter = new AtomicInteger(basePort);
        instances.clear();
        seed = ThreadLocalRandom.current().nextLong();
        random = new Random(seed);

        // Tests need to opt out of the in-process channel
        RpcServer.USE_IN_PROCESS_SERVER = true;
        RpcClient.USE_IN_PROCESS_CHANNEL = true;

        // Tests that depend on failure detection should set intervals by themselves
        MembershipService.FAILURE_DETECTOR_INITIAL_DELAY_IN_MS = 100000;
        MembershipService.FAILURE_DETECTOR_INTERVAL_IN_MS = 100000;
        RpcClient.Conf.RPC_JOIN_PHASE_2_TIMEOUT = RpcClient.Conf.RPC_TIMEOUT_MEDIUM_MS * 20;
        RpcClient.Conf.RPC_TIMEOUT_MS = RpcClient.Conf.RPC_TIMEOUT_MEDIUM_MS;
        RpcClient.Conf.RPC_PROBE_TIMEOUT = 5000;

        useStaticFd = false;
        staticFds.clear();
        serverInterceptors.clear();
        clientInterceptors.clear();
    }

    @After
    public void afterTest() throws InterruptedException {
        for (final Cluster cluster: instances.values()) {
            cluster.shutdown();
        }
    }

    /**
     * Test with a single node joining through a seed.
     */
    @Test
    public void singleNodeJoinsThroughSeed() throws IOException, InterruptedException, ExecutionException {
        RpcServer.USE_IN_PROCESS_SERVER = false;
        RpcClient.USE_IN_PROCESS_CHANNEL = false;

        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(1, seedHost);
        verifyCluster(1, seedHost);
        extendCluster(1, seedHost);
        verifyCluster(2, seedHost);
    }

    /**
     * Test with K nodes joining the network through a single seed.
     */
    @Test
    public void tenNodesJoinSequentially() throws IOException, InterruptedException {
        // Explicitly test netty
        RpcServer.USE_IN_PROCESS_SERVER = false;
        RpcClient.USE_IN_PROCESS_CHANNEL = false;

        final int numNodes = 10;
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(1, seedHost); // Only bootstrap a seed.
        verifyCluster(1, seedHost);
        for (int i = 0; i < numNodes; i++) {
            extendCluster(1, seedHost);
            waitAndVerifyAgreement( i + 2, 5, 1000, seedHost);
        }
    }

    /**
     * Identical to the previous test, but with more than K nodes joining in serial.
     */
    @Test
    public void twentyNodesJoinSequentially() throws IOException, InterruptedException {
        // Explicitly test netty
        RpcServer.USE_IN_PROCESS_SERVER = false;
        RpcClient.USE_IN_PROCESS_CHANNEL = false;

        final int numNodes = 20;
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(1, seedHost); // Only bootstrap a seed.
        verifyCluster(1, seedHost);

        for (int i = 0; i < numNodes; i++) {
            extendCluster(1, seedHost);
            waitAndVerifyAgreement( i + 2, 5, 1000, seedHost);
        }
    }

    /**
     * Identical to the previous test, but with more than K nodes joining in parallel.
     * <p>
     * The test starts with a single seed and all N - 1 subsequent nodes initiate their join protocol at the same
     * time. This tests a single seed's ability to bootstrap a large cluster in one step.
     */
    @Test(timeout=150000)
    public void threeHundredNodesJoinInParallel() throws IOException, InterruptedException {
        final int numNodes = 300; // Includes the size of the cluster
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(numNodes, seedHost);
        verifyCluster(numNodes, seedHost);
    }


    /**
     * This test starts with a single seed, and a wave where 50 subsequent nodes initiate their join protocol
     * concurrently. Following this, a subsequent wave begins where 100 nodes then start together.
     */
    @Test(timeout=150000)
    public void hundredNodesJoinFiftyNodeCluster() throws IOException, InterruptedException {
        RpcServer.USE_IN_PROCESS_SERVER = true;
        RpcClient.USE_IN_PROCESS_CHANNEL = true;

        final int numNodesPhase1 = 50;
        final int numNodesPhase2 = 100;
        final HostAndPort seedHost = HostAndPort.fromParts("127.0.0.1", basePort);
        createCluster(numNodesPhase1, seedHost);
        waitAndVerifyAgreement(numNodesPhase1, 10, 100, seedHost);
        extendCluster(numNodesPhase2, seedHost);
        waitAndVerifyAgreement(numNodesPhase1 + numNodesPhase2, 10, 1000, seedHost);
    }

    /**
     * This test starts with a 4 node cluster. We then fail a single node to see if the monitoring mechanism
     * identifies the failing node and arrives at a decision to remove it.
     */
    @Test
    public void oneFailureOutOfFourNodes() throws IOException, InterruptedException {
{'fraction_non_alphanumeric': 0.08609510086455331, 'fraction_numerical': 0.018371757925072046, 'mean_word_length': 3.2010088272383355, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 10, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '249007', 'n_tokens_mistral': 4866, 'n_tokens_neox': 4625, 'n_words': 1360}