Use host-port wait strategy of the Reaper (#223)

cristianrgreco · web-flow · commit daf4c321e969 · 2021-07-13T14:13:01.000+01:00
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -71,7 +71,8 @@
     "rimraf": "^3.0.2",
     "ts-jest": "^27.0.1",
     "ts-node": "^10.0.0",
-    "typescript": "^4.3.2"
+    "typescript": "^4.3.2",
+    "wait-for-expect": "^3.0.2"
   },
   "husky": {
     "hooks": {
diff --git a/src/generic-container.test.ts b/src/generic-container.test.ts
@@ -1,12 +1,9 @@
 import fetch from "node-fetch";
 import path from "path";
-import { createServer, Server } from "http";
 import { GenericContainer } from "./generic-container";
 import { AlwaysPullPolicy } from "./pull-policy";
 import { Wait } from "./wait";
 import { RandomUuid } from "./uuid";
-import { TestContainers } from "./test-containers";
-import { RandomPortClient } from "./port-client";
 import { getContainerById, getEvents, getRunningContainerNames } from "./test-helper";
 import { Network } from "./network";
 
diff --git a/src/modules/kafka/kafka-container.ts b/src/modules/kafka/kafka-container.ts
@@ -4,19 +4,19 @@ import { DockerClient } from "../../docker-client";
 import { Network, StartedNetwork } from "../../network";
 import { Host } from "../../docker-client-instance";
 import { Port } from "../../port";
-import { PortClient, RandomPortClient } from "../../port-client";
 import { RandomUuid, Uuid } from "../../uuid";
 import { StartedTestContainer, StoppedTestContainer } from "../..";
 import { StopOptions } from "../../test-container";
 import { log } from "../../logger";
 import { AbstractStartedContainer } from "../abstract-started-container";
+import { PortGenerator, RandomUniquePortGenerator } from "../../port-generator";
 
 export const KAFKA_IMAGE = "confluentinc/cp-kafka:5.5.4";
 export const ZK_IMAGE = "confluentinc/cp-zookeeper:5.5.4";
 
 export class KafkaContainer extends GenericContainer {
   private readonly uuid: Uuid = new RandomUuid();
-  private readonly portClient: PortClient = new RandomPortClient();
+  private readonly portGenerator: PortGenerator = new RandomUniquePortGenerator();
 
   private isZooKeeperProvided = false;
   private zooKeeperHost?: Host;
@@ -61,7 +61,7 @@ export class KafkaContainer extends GenericContainer {
       this.withEnv("KAFKA_ZOOKEEPER_CONNECT", `${this.zooKeeperHost}:${this.zooKeeperPort}`);
     } else {
       const zooKeeperHost = this.uuid.nextUuid();
-      const zooKeeperPort = await this.portClient.getPort();
+      const zooKeeperPort = await this.portGenerator.generatePort();
 
       this.withEnv("KAFKA_ZOOKEEPER_CONNECT", `${zooKeeperHost}:${zooKeeperPort}`);
 
diff --git a/src/port-binder.test.ts b/src/port-binder.test.ts
@@ -1,20 +1,10 @@
 import { PortBinder } from "./port-binder";
-import { FixedPortClient } from "./port-client";
+import { FixedPortGenerator } from "./port-generator";
 
 describe("PortBinder", () => {
   it("should bind each port to the host", async () => {
-    const portClient = new FixedPortClient([1000, 2000]);
-    const portBinder = new PortBinder(portClient);
-
-    const boundPorts = await portBinder.bind([1, 2]);
-
-    expect(boundPorts.getBinding(1)).toBe(1000);
-    expect(boundPorts.getBinding(2)).toBe(2000);
-  });
-
-  it("should not produce the same port more than once", async () => {
-    const portClient = new FixedPortClient([1000, 1000, 2000]);
-    const portBinder = new PortBinder(portClient);
+    const portGenerator = new FixedPortGenerator([1000, 2000]);
+    const portBinder = new PortBinder(portGenerator);
 
     const boundPorts = await portBinder.bind([1, 2]);
 
diff --git a/src/port-binder.ts b/src/port-binder.ts
@@ -1,23 +1,15 @@
 import { BoundPorts } from "./bound-ports";
 import { Port } from "./port";
-import { PortClient, RandomPortClient } from "./port-client";
+import { PortGenerator, RandomUniquePortGenerator } from "./port-generator";
 
 export class PortBinder {
-  private readonly ports: Set<number> = new Set();
-
-  constructor(private readonly portClient: PortClient = new RandomPortClient()) {}
+  constructor(private readonly portGenerator: PortGenerator = new RandomUniquePortGenerator()) {}
 
   public async bind(ports: Port[]): Promise<BoundPorts> {
     const boundPorts = new BoundPorts();
 
     for (const port of ports) {
-      let allocatedPort: Port;
-      do {
-        allocatedPort = await this.portClient.getPort();
-      } while (this.ports.has(allocatedPort));
-
-      this.ports.add(allocatedPort);
-      boundPorts.setBinding(port, allocatedPort);
+      boundPorts.setBinding(port, await this.portGenerator.generatePort());
     }
 
     return boundPorts;
diff --git a/src/port-client.test.ts b/src/port-client.test.ts
diff --git a/src/port-client.ts b/src/port-client.ts
diff --git a/src/port-forwarder.test.ts b/src/port-forwarder.test.ts
@@ -1,7 +1,7 @@
 import { createServer, Server } from "http";
 import { GenericContainer } from "./generic-container";
 import { TestContainers } from "./test-containers";
-import { RandomPortClient } from "./port-client";
+import { RandomUniquePortGenerator } from "./port-generator";
 import { Network } from "./network";
 
 describe("PortForwarder", () => {
@@ -11,13 +11,14 @@ describe("PortForwarder", () => {
   let server: Server;
 
   beforeEach(async () => {
-    randomPort = await new RandomPortClient().getPort();
-    server = await new Promise((resolve) => {
-      const server = createServer((req, res) => {
+    randomPort = await new RandomUniquePortGenerator().generatePort();
+
+    await new Promise<void>((resolve) => {
+      server = createServer((req, res) => {
         res.writeHead(200);
         res.end("hello world");
       });
-      server.listen(randomPort, () => resolve(server));
+      server.listen(randomPort, resolve);
     });
   });
 
diff --git a/src/port-generator.test.ts b/src/port-generator.test.ts
@@ -0,0 +1,11 @@
+import { FixedPortGenerator, RandomUniquePortGenerator } from "./port-generator";
+
+describe("PortGenerator", () => {
+  describe("RandomUniquePortGenerator", () => {
+    it("should generate a random and unique port across all instances", async () => {
+      const fixedPortGenerator = new FixedPortGenerator([1000, 1000, 1001]);
+      expect(await new RandomUniquePortGenerator(fixedPortGenerator).generatePort()).toBe(1000);
+      expect(await new RandomUniquePortGenerator(fixedPortGenerator).generatePort()).toBe(1001);
+    });
+  });
+});
diff --git a/src/port-generator.ts b/src/port-generator.ts
@@ -0,0 +1,44 @@
+import { Port } from "./port";
+import getRandomPort from "get-port";
+
+export interface PortGenerator {
+  generatePort(): Promise<Port>;
+}
+
+class RandomPortGenerator {
+  public generatePort(): Promise<Port> {
+    return getRandomPort({ port: this.randomBetweenInclusive(10000, 65535) });
+  }
+
+  private randomBetweenInclusive(min: number, max: number) {
+    return Math.floor(Math.random() * (max - min + 1) + min);
+  }
+}
+
+export class RandomUniquePortGenerator implements PortGenerator {
+  private static readonly assignedPorts = new Set();
+
+  constructor(private readonly portGenerator: PortGenerator = new RandomPortGenerator()) {}
+
+  public async generatePort(): Promise<Port> {
+    let port: Port;
+
+    do {
+      port = await this.portGenerator.generatePort();
+    } while (RandomUniquePortGenerator.assignedPorts.has(port));
+
+    RandomUniquePortGenerator.assignedPorts.add(port);
+
+    return port;
+  }
+}
+
+export class FixedPortGenerator implements PortGenerator {
+  private portIndex = 0;
+
+  constructor(private readonly ports: Port[]) {}
+
+  public generatePort(): Promise<Port> {
+    return Promise.resolve(this.ports[this.portIndex++]);
+  }
+}
diff --git a/src/reaper.test.ts b/src/reaper.test.ts
@@ -1,42 +1,57 @@
 import { GenericContainer } from "./generic-container";
-import { ReaperInstance } from "./reaper";
-import { getImagesRepoTags, getRunningContainerIds, getRunningNetworkIds } from "./test-helper";
+import {
+  getImagesRepoTags,
+  getReaperContainerId,
+  getRunningContainerIds,
+  getRunningNetworkIds,
+  stopReaper,
+} from "./test-helper";
 import { Network } from "./network";
 import path from "path";
 import { RandomUuid } from "./uuid";
+import waitForExpect from "wait-for-expect";
 
 describe("Reaper", () => {
   jest.setTimeout(60_000);
 
   it("should remove containers", async () => {
     const container = await new GenericContainer("cristianrgreco/testcontainer:1.1.12").start();
 
-    await ReaperInstance.stopInstance();
+    const reaperContainerId = await getReaperContainerId();
+    await stopReaper();
 
     expect(await getRunningContainerIds()).toContain(container.getId());
-    await new Promise((resolve) => setTimeout(resolve, 15_000));
-    expect(await getRunningContainerIds()).not.toContain(container.getId());
+    await waitForExpect(async () => {
+      expect(await getRunningContainerIds()).not.toContain(container.getId());
+      expect(await getRunningContainerIds()).not.toContain(reaperContainerId);
+    }, 20_000);
   });
 
   it("should remove networks", async () => {
     const network = await new Network().start();
 
-    await ReaperInstance.stopInstance();
+    const reaperContainerId = await getReaperContainerId();
+    await stopReaper();
 
     expect(await getRunningNetworkIds()).toContain(network.getId());
-    await new Promise((resolve) => setTimeout(resolve, 15_000));
-    expect(await getRunningNetworkIds()).not.toContain(network.getId());
+    await waitForExpect(async () => {
+      expect(await getRunningNetworkIds()).not.toContain(network.getId());
+      expect(await getRunningContainerIds()).not.toContain(reaperContainerId);
+    }, 20_000);
   });
 
   it("should remove images", async () => {
     const imageId = `${new RandomUuid().nextUuid()}:${new RandomUuid().nextUuid()}`;
     const context = path.resolve(__dirname, "..", "fixtures", "docker", "docker");
     await GenericContainer.fromDockerfile(context).build(imageId);
 
-    await ReaperInstance.stopInstance();
+    const reaperContainerId = await getReaperContainerId();
+    await stopReaper();
 
     expect(await getImagesRepoTags()).toContain(imageId);
-    await new Promise((resolve) => setTimeout(resolve, 15_000));
-    expect(await getImagesRepoTags()).not.toContain(imageId);
+    await waitForExpect(async () => {
+      expect(await getImagesRepoTags()).not.toContain(imageId);
+      expect(await getRunningContainerIds()).not.toContain(reaperContainerId);
+    }, 20_000);
   });
 });
diff --git a/src/reaper.ts b/src/reaper.ts
@@ -2,12 +2,12 @@ import { Socket } from "net";
 import { log } from "./logger";
 import { GenericContainer } from "./generic-container";
 import { StartedTestContainer } from "./test-container";
-import { Wait } from "./wait";
 import { Id } from "./container";
 import { DockerClient } from "./docker-client";
 
 export interface Reaper {
   addProject(projectName: string): void;
+  getContainerId(): string;
   stop(): void;
 }
 
@@ -22,6 +22,10 @@ class RealReaper implements Reaper {
     this.socket.write(`label=com.docker.compose.project=${projectName}\r\n`);
   }
 
+  public getContainerId(): string {
+    return this.container.getId();
+  }
+
   public stop(): void {
     this.socket.end();
   }
@@ -35,6 +39,10 @@ class DisabledReaper implements Reaper {
   public stop(): void {
     // noop
   }
+
+  public getContainerId(): string {
+    return "";
+  }
 }
 
 export class ReaperInstance {
@@ -86,7 +94,6 @@ export class ReaperInstance {
     const container = await new GenericContainer(this.getImage())
       .withName(`testcontainers-ryuk-${sessionId}`)
       .withExposedPorts(8080)
-      .withWaitStrategy(Wait.forLogMessage("Started!"))
       .withBindMount(dockerSocket, "/var/run/docker.sock")
       .withDaemonMode()
       .withPrivilegedMode()
@@ -101,8 +108,8 @@ export class ReaperInstance {
     socket.unref();
 
     socket
-      .on('timeout', () => log.error(`Reaper ${container.getId()} socket timed out`))
-      .on('error', err => log.error(`Reaper ${container.getId()} socket error: ${err}`))
+      .on("timeout", () => log.error(`Reaper ${container.getId()} socket timed out`))
+      .on("error", (err) => log.error(`Reaper ${container.getId()} socket error: ${err}`))
       .on("close", (hadError) => {
         if (hadError) {
           log.error(`Connection to Reaper ${container.getId()} closed with error`);
diff --git a/src/test-helper.ts b/src/test-helper.ts
@@ -1,5 +1,7 @@
 import Dockerode from "dockerode";
 import { Readable } from "stream";
+import { ReaperInstance } from "./reaper";
+import { DockerClientInstance } from "./docker-client-instance";
 
 export const dockerode = new Dockerode();
 
@@ -38,3 +40,11 @@ export const getVolumeNames = async (): Promise<string[]> => {
   const { Volumes: volumes } = await dockerode.listVolumes();
   return volumes.map((volume) => volume.Name);
 };
+
+export const getReaperContainerId = async (): Promise<string> => {
+  return (await ReaperInstance.getInstance(await DockerClientInstance.getInstance())).getContainerId();
+};
+
+export const stopReaper = async (): Promise<void> => {
+  return ReaperInstance.stopInstance();
+};
