test: Implemented IT for `HaproxyDataPlaneApiShardingPublischerStrategy`

* Enabled the HAProxy Data Plane API in the configuration, that is used
  for the tests that start an instance of HAProxy inside a container.

diff --git a/src/test/java/de/juplo/kafka/chat/backend/implementation/haproxy/HaproxyDataPlaneApiShardingPublisherStrategyIT.java b/src/test/java/de/juplo/kafka/chat/backend/implementation/haproxy/HaproxyDataPlaneApiShardingPublisherStrategyIT.java
new file mode 100644 (file)
index 0000000..9235a06
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/chat/backend/implementation/haproxy/HaproxyDataPlaneApiShardingPublisherStrategyIT.java
@@ -0,0 +1,105 @@
+package de.juplo.kafka.chat.backend.implementation.haproxy;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+
+import static de.juplo.kafka.chat.backend.implementation.haproxy.HaproxyDataPlaneApiShardingPublisherStrategy.MAP_ENTRY_PATH;
+import static de.juplo.kafka.chat.backend.implementation.haproxy.HaproxyDataPlaneApiShardingPublisherStrategy.MAP_PARAM;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+@Testcontainers
+@Slf4j
+public class HaproxyDataPlaneApiShardingPublisherStrategyIT
+{
+  @Test
+  void test()
+  {
+    Mono<String> result = shardingPublisherStrategy.publishOwnership(SHARD);
+
+    assertThat(result.block(Duration.ofSeconds(5)))
+        .isEqualTo(INSTANCE_ID);
+    assertThat(getMapEntryValueForKey(SHARD).block(Duration.ofSeconds(5)))
+        .isEqualTo(INSTANCE_ID);
+  }
+
+
+  private Mono<String> getMapEntryValueForKey(int key)
+  {
+    return webClient
+        .get()
+        .uri(uriBuilder -> uriBuilder
+            .path(MAP_ENTRY_PATH)
+            .queryParam(MAP_PARAM, "#" + shardingPublisherStrategy.getMapId())
+            .build(SHARD))
+        .accept(MediaType.APPLICATION_JSON)
+        .exchangeToMono(response ->
+        {
+          if (response.statusCode().equals(HttpStatus.OK))
+          {
+            return response.bodyToMono(MapEntryTo.class);
+          }
+          else
+          {
+            return response.createError();
+          }
+        })
+        .map(entry -> entry.value());
+  }
+
+
+  WebClient webClient;
+  HaproxyDataPlaneApiShardingPublisherStrategy shardingPublisherStrategy;
+
+
+  @BeforeEach
+  void setUpWebClient()
+  {
+    webClient = WebClient
+        .builder()
+        .baseUrl("http://localhost:" + HAPROXY.getMappedPort(5555) + "/v2/")
+        .defaultHeaders(httpHeaders -> httpHeaders.setBasicAuth("juplo", "juplo"))
+        .build();
+
+    shardingPublisherStrategy = new HaproxyDataPlaneApiShardingPublisherStrategy(
+        webClient,
+        MAP_PATH,
+        INSTANCE_ID);
+  }
+
+
+  static final String MAP_PATH = "/usr/local/etc/haproxy/sharding.map";
+  static final String INSTANCE_ID = "foo";
+  static final int SHARD = 6;
+
+  @Container
+  static final GenericContainer HAPROXY =
+      new GenericContainer(DockerImageName.parse("haproxytech/haproxy-debian:2.8"))
+          .withNetwork(Network.newNetwork())
+          .withNetworkAliases("haproxy")
+          .withCopyFileToContainer(
+              MountableFile.forClasspathResource("haproxy.cfg"),
+              "/usr/local/etc/haproxy/haproxy.cfg")
+          .withCopyFileToContainer(
+              MountableFile.forClasspathResource("dataplaneapi.yml"),
+              "/usr/local/etc/haproxy/dataplaneapi.yml")
+          .withCopyFileToContainer(
+              MountableFile.forClasspathResource("sharding.map"),
+              "/usr/local/etc/haproxy/sharding.map")
+          .withExposedPorts(8400, 8401, 8404, 5555)
+          .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("HAPROXY"));
+}

diff --git a/src/test/resources/dataplaneapi.yml b/src/test/resources/dataplaneapi.yml
new file mode 100644 (file)
index 0000000..2dcae5d
--- /dev/null
+++ b/src/test/resources/dataplaneapi.yml
@@ -0,0 +1,26 @@
+config_version: 2
+name: 87327b8eb539
+mode: single
+status: ""
+dataplaneapi:
+  host: 0.0.0.0
+  port: 5555
+  advertised:
+    api_address: ""
+    api_port: 0
+  scheme:
+  - http
+  transaction:
+    transaction_dir: /tmp/haproxy
+  user:
+  - name: juplo
+    insecure: true
+    password: juplo
+haproxy:
+  config_file: /usr/local/etc/haproxy/haproxy.cfg
+  haproxy_bin: /usr/sbin/haproxy
+  reload:
+    reload_delay: 5
+    reload_cmd: kill -SIGHUP 1
+    restart_cmd: kill -SIGINT 1
+    reload_strategy: custom

diff --git a/src/test/resources/haproxy.cfg b/src/test/resources/haproxy.cfg
index 230b8f9..72828a9 100644 (file)
--- a/src/test/resources/haproxy.cfg
+++ b/src/test/resources/haproxy.cfg
@@ -9,6 +9,7 @@ defaults
   default-server init-addr last,libc,none
 
 global
+  master-worker
   stats socket ipv4@:8401 level admin
   stats socket /var/run/haproxy.sock mode 666 level admin
   stats timeout 2m
@@ -37,3 +38,7 @@ backend backend_2
 
 backend backend_3
   server b3 backend-3:8080 check
+
+program api
+  command /usr/bin/dataplaneapi -f /usr/local/etc/haproxy/dataplaneapi.yml
+  no option start-on-reload