✅ Infra for Galvan testing

management/testing/integration-tests/pom.xml

@@ -68,6 +68,23 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-core</artifactId>
+      <version>1.3</version>
+    </dependency>
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <version>${logback.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+      <version>2.6.3</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

management/testing/integration-tests/src/test/java/org/terracotta/management/integration/tests/AbstractHATest.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright Terracotta, Inc.
+ *
+ * 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 org.terracotta.management.integration.tests;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.terracotta.testing.rules.BasicExternalCluster;
+
+import java.io.File;
+
+import static java.util.Collections.emptyList;
+
+/**
+ * @author Mathieu Carbou
+ */
+public abstract class AbstractHATest extends AbstractTest {
+
+  private final String offheapResource = "primary-server-resource";
+  private final String resourceConfig =
+      "<config xmlns:ohr='http://www.terracotta.org/config/offheap-resource'>"
+          + "<ohr:offheap-resources>"
+          + "<ohr:resource name=\"" + offheapResource + "\" unit=\"MB\">64</ohr:resource>"
+          + "</ohr:offheap-resources>" +
+          "</config>\n";
+
+  @Rule
+  public org.terracotta.testing.rules.Cluster voltron =
+      new BasicExternalCluster(new File("target/galvan"), 2, emptyList(), "", resourceConfig, "");
+
+  @Before
+  public void setUp() throws Exception {
+    voltron.getClusterControl().waitForActive();
+    voltron.getClusterControl().waitForRunningPassivesInStandby();
+    commonSetUp(voltron);
+    tmsAgentService.readMessages();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    commonTearDown();
+  }
+
+}

management/testing/integration-tests/src/test/java/org/terracotta/management/integration/tests/AbstractTest.java

@@ -0,0 +1,201 @@
+/*
+ * Copyright Terracotta, Inc.
+ *
+ * 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 org.terracotta.management.integration.tests;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import org.junit.Rule;
+import org.junit.rules.Timeout;
+import org.terracotta.connection.Connection;
+import org.terracotta.connection.ConnectionFactory;
+import org.terracotta.connection.ConnectionPropertyNames;
+import org.terracotta.management.entity.sample.Cache;
+import org.terracotta.management.entity.sample.client.CacheFactory;
+import org.terracotta.management.entity.tms.TmsAgentConfig;
+import org.terracotta.management.entity.tms.client.TmsAgentEntity;
+import org.terracotta.management.entity.tms.client.TmsAgentEntityFactory;
+import org.terracotta.management.entity.tms.client.TmsAgentService;
+import org.terracotta.management.model.capabilities.context.CapabilityContext;
+import org.terracotta.management.model.stats.ContextualStatistics;
+import org.terracotta.management.registry.collect.StatisticConfiguration;
+import org.terracotta.testing.rules.Cluster;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Mathieu Carbou
+ */
+public abstract class AbstractTest {
+
+  private final ObjectMapper mapper = new ObjectMapper();
+
+  private Connection managementConnection;
+  private Cluster cluster;
+
+  protected final List<CacheFactory> webappNodes = new ArrayList<>();
+  protected final Map<String, List<Cache>> caches = new HashMap<>();
+  protected TmsAgentService tmsAgentService;
+
+  @Rule
+  public Timeout timeout = Timeout.seconds(60);
+
+  protected final void commonSetUp(Cluster cluster) throws Exception {
+    this.cluster = cluster;
+
+    mapper.configure(SerializationFeature.INDENT_OUTPUT, true);
+    mapper.addMixIn(CapabilityContext.class, CapabilityContextMixin.class);
+
+    connectManagementClients(cluster.getConnectionURI());
+
+    addWebappNode(cluster.getConnectionURI().resolve("pet-clinic"));
+    addWebappNode(cluster.getConnectionURI().resolve("pet-clinic"));
+
+    getCaches("pets");
+    getCaches("clients");
+  }
+
+  protected final void commonTearDown() throws Exception {
+    closeNodes();
+    if (managementConnection != null) {
+      managementConnection.close();
+    }
+    if (cluster != null) {
+      cluster.getClusterControl().terminateAllServers();
+    }
+  }
+
+  protected JsonNode readJson(String file) {
+    try {
+      return mapper.readTree(new File(AbstractTest.class.getResource("/" + file).toURI()));
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  protected JsonNode toJson(Object o) {
+    try {
+      return mapper.readTree(mapper.writeValueAsString(o));
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  protected int size(int nodeIdx, String cacheName) {
+    return caches.get(cacheName).get(nodeIdx).size();
+  }
+
+  protected String get(int nodeIdx, String cacheName, String key) {
+    return caches.get(cacheName).get(nodeIdx).get(key);
+  }
+
+  protected void put(int nodeIdx, String cacheName, String key, String value) {
+    caches.get(cacheName).get(nodeIdx).put(key, value);
+  }
+
+  protected void remove(int nodeIdx, String cacheName, String key) {
+    caches.get(cacheName).get(nodeIdx).remove(key);
+  }
+
+  protected void closeNodes() {
+    webappNodes.forEach(cacheFactory -> {
+      try {
+        cacheFactory.getConnection().close();
+      } catch (IOException ignored) {
+      }
+    });
+  }
+
+  protected void getCaches(String name) {
+    caches.put(name, webappNodes.stream().map(cacheFactory -> cacheFactory.getCache(name)).collect(Collectors.toList()));
+  }
+
+  protected void addWebappNode(URI uri) throws Exception {
+    StatisticConfiguration statisticConfiguration = new StatisticConfiguration()
+        .setAverageWindowDuration(1, TimeUnit.MINUTES)
+        .setHistorySize(100)
+        .setHistoryInterval(1, TimeUnit.SECONDS)
+        .setTimeToDisable(5, TimeUnit.SECONDS);
+    CacheFactory cacheFactory = new CacheFactory(uri, statisticConfiguration);
+    cacheFactory.init();
+    webappNodes.add(cacheFactory);
+  }
+
+  public static abstract class CapabilityContextMixin {
+    @JsonIgnore
+    public abstract Collection<String> getRequiredAttributeNames();
+
+    @JsonIgnore
+    public abstract Collection<CapabilityContext.Attribute> getRequiredAttributes();
+  }
+
+  private void connectManagementClients(URI uri) throws Exception {
+    // connects to server
+    Properties properties = new Properties();
+    properties.setProperty(ConnectionPropertyNames.CONNECTION_NAME, getClass().getSimpleName());
+    properties.setProperty(ConnectionPropertyNames.CONNECTION_TIMEOUT, "5000");
+    this.managementConnection = ConnectionFactory.connect(uri, properties);
+
+    // create a tms entity
+    TmsAgentEntityFactory tmsAgentEntityFactory = new TmsAgentEntityFactory(managementConnection, getClass().getSimpleName());
+    TmsAgentEntity tmsAgentEntity = tmsAgentEntityFactory.retrieveOrCreate(new TmsAgentConfig()
+        .setMaximumUnreadMessages(1024 * 1024)
+        .setStatisticConfiguration(new StatisticConfiguration()
+            .setAverageWindowDuration(1, TimeUnit.MINUTES)
+            .setHistorySize(100)
+            .setHistoryInterval(1, TimeUnit.SECONDS)
+            .setTimeToDisable(5, TimeUnit.SECONDS)));
+    this.tmsAgentService = new TmsAgentService(tmsAgentEntity);
+    this.tmsAgentService.setOperationTimeout(10, TimeUnit.SECONDS);
+  }
+
+  protected void queryAllRemoteStatsUntil(Predicate<List<? extends ContextualStatistics>> test) throws Exception {
+    List<? extends ContextualStatistics> statistics;
+    do {
+      statistics = tmsAgentService.readMessages()
+          .stream()
+          .filter(message -> message.getType().equals("STATISTICS"))
+          .flatMap(message -> message.unwrap(ContextualStatistics.class).stream())
+          .collect(Collectors.toList());
+      // PLEASE KEEP THIS ! Really useful when troubleshooting stats!
+      /*if (!statistics.isEmpty()) {
+        System.out.println("received at " + System.currentTimeMillis() + ":");
+        statistics.stream()
+            .flatMap(o -> o.getStatistics().entrySet().stream())
+            .forEach(System.out::println);
+      }*/
+      Thread.sleep(500);
+    } while (!Thread.currentThread().isInterrupted() && (statistics.isEmpty() || !test.test(statistics)));
+    assertFalse(Thread.currentThread().isInterrupted());
+    assertTrue(test.test(statistics));
+  }
+
+}

management/testing/integration-tests/src/test/java/org/terracotta/management/integration/tests/FailoverTest.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright Terracotta, Inc.
+ *
+ * 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 org.terracotta.management.integration.tests;
+
+import org.junit.Ignore;
+import org.junit.Test;
+import org.terracotta.management.model.cluster.Cluster;
+import org.terracotta.management.model.cluster.Server;
+import org.terracotta.management.model.message.Message;
+import org.terracotta.management.model.notification.ContextualNotification;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * @author Mathieu Carbou
+ */
+@Ignore // TODO activate
+public class FailoverTest extends AbstractHATest {
+
+  @Test
+  public void failover_management() throws Exception {
+    Cluster cluster = tmsAgentService.readTopology();
+    Server active = cluster.serverStream().filter(Server::isActive).findFirst().get();
+    Server passive = cluster.serverStream().filter(server -> !server.isActive()).findFirst().get();
+    assertThat(active.getState(), equalTo(Server.State.ACTIVE));
+    assertThat(passive.getState(), equalTo(Server.State.PASSIVE));
+
+    // clear buffer
+    tmsAgentService.readMessages();
+
+    // kill active - passive should take the active role
+    voltron.getClusterControl().terminateActive();
+    voltron.getClusterControl().waitForActive();
+
+    cluster = tmsAgentService.readTopology();
+    Server newActive = cluster.serverStream().filter(Server::isActive).findFirst().get();
+    assertThat(newActive.getState(), equalTo(Server.State.ACTIVE));
+    assertThat(newActive.getServerName(), equalTo(passive.getServerName()));
+
+    // read messages
+    List<Message> messages = tmsAgentService.readMessages();
+    assertThat(messages.size(), equalTo(3));
+
+    List<ContextualNotification> notifs = messages.stream()
+        .filter(message -> message.getType().equals("NOTIFICATION"))
+        .flatMap(message -> message.unwrap(ContextualNotification.class).stream())
+        .filter(notif -> notif.getType().equals("SERVER_STATE_CHANGED"))
+        .collect(Collectors.toList());
+
+    assertThat(
+        notifs.stream().map(notif -> notif.getContext().get(Server.NAME_KEY)).collect(Collectors.toList()),
+        equalTo(Arrays.asList(newActive.getServerName(), newActive.getServerName())));
+
+    assertThat(
+        notifs.stream().map(notif -> notif.getAttributes().get("state")).collect(Collectors.toList()),
+        equalTo(Arrays.asList("ACTIVE", "ACTIVE")));
+
+    //TODO: complete with Galvan
+    //- test topology (like topology_includes_passives), client should have re-exposed their management metadata
+    //- check notifications: server states
+    //- check notification that might be there: CLIENT_RECONNECTED and SERVER_ENTITY_FAILOVER_COMPLETE
+  }
+
+}