Haproxy.ts

import { encodeText } from "/binary";
import { Reporter } from "/tool/console_reporter";
import { Event } from "/events";
import { Duration, DurationMeasurer, Percent } from "/numbers";
import { createResourceFromCallback, Resource } from "/resources";
import { processResource } from "/tool/resources";

class EventLoopLagTracker {
	constructor(delay: Duration) {
		this.#delay = delay;
		this.#lag = Duration.ms(0);
		this.#measurer = new DurationMeasurer();
	}

	#lag: Duration;
	#delay: Duration;
	#measurer: DurationMeasurer;

	public getLag(): Duration {
		return this.#lag;
	}

	public track(): Resource {
		let timeout: undefined | number;

		const queue = () => {
			this.#measurer.reset();
			timeout = setTimeout(() => {
				this.#lag = this.#measurer.since().subtract(this.#delay);
				queue();
			}, this.#delay.ms());
		};

		queue();

		return createResourceFromCallback("EventLoopLagTracker", () => {
			clearTimeout(timeout);
		});
	}
}

type CheckCallback = () => boolean;

type StatusLabel = "up" | "down" | "drain" | "ready" | "maint";

export class HAProxyAgent {
	constructor(reporter: Reporter) {
		this.#loopTracker = new EventLoopLagTracker(Duration.ms(20));
		this.#callbacks = [];
		this.#status = "up";
		this.#reporter = reporter;
		this.#ackEvent = new Event("HAProxyAgent.ackEvent");

		processResource.add(this.#loopTracker.track());
	}

	#ackEvent: Event<string, void>;
	#reporter: Reporter;
	#loopTracker: EventLoopLagTracker;
	#status: StatusLabel;
	#callbacks: CheckCallback[];

	public addCheck(callback: CheckCallback) {
		this.#callbacks.push(callback);
	}

	public getWeight(): Percent {
		const lag = this.#loopTracker.getLag();

		if (lag.seconds() >= 1) {
			return Percent.hundred(0);
		}

		return Percent.hundred(100 - lag.ms() / 10);
	}

	// down       - Marks the server as down.
	// up         - Marks the server as up.
	// maint      - Puts the server into maintenance mode.
	// ready      - Takes the server out of maintenance mode.
	// 50%        - Changes the server's weight to half its current value.
	// maxconn:30 - Changes the server's maxconn value to 30.
	public async setStatus(label: StatusLabel): Promise<void> {
		this.#reporter.logTags({ status: label });
		this.#status = label;
		await this.#ackEvent.wait();
	}

	public getStatusLabel(): StatusLabel {
		const label = this.#status;

		if (label === "up") {
			for (const callback of this.#callbacks) {
				if (!callback()) {
					return "down";
				}
			}
		}

		return label;
	}

	public getStatus(): string {
		const label = this.getStatusLabel();

		if (label === "up") {
			return `up ${Math.ceil(this.getWeight().hundred())}%`;
		}

		return label;
	}

	public async listen(port: number, hostname: string) {
		const listener = Deno.listen({ port, hostname });

		for await (const conn of listener) {
			const status = this.getStatus();
			this.#ackEvent.send(status);
			conn.write(encodeText(`${status}\n`));
			conn.close();
		}
	}
}