feat(moonzero): consul-backed discovery driver over the agent HTTP API.

Signed-off-by: 林晨 (Leo Cheng) <chengkelfan@qq.com>

Author: Lfan-ke

consul.mbt

@@ -0,0 +1,238 @@
+// A consul client over consul's HTTP API: a `ConsulHttp` carries a request to a consul
+// agent and hands back the status and body, and `ConsulClient` wraps it with the agent
+// operations the discovery driver needs (register a service with a TTL check, pass the
+// check to keep it alive, deregister it, and read a service's healthy instances). The
+// transport is an interface, so the same client drives a real consul over an HTTP
+// socket (the native `discov` driver) or an in-process fake in a test; the request
+// shaping and JSON parsing here run on every backend.
+
+///|
+/// A consul API call that failed — a transport error, or a non-2xx agent response.
+pub suberror ConsulError {
+  ConsulError(String)
+}
+
+///|
+/// A consul agent's HTTP response: the status code and the raw body bytes.
+pub struct ConsulResponse {
+  status : Int
+  body : Bytes
+}
+
+///|
+/// A consul response.
+pub fn ConsulResponse::new(status : Int, body : Bytes) -> ConsulResponse {
+  { status, body }
+}
+
+///|
+/// A transport to a consul agent: it performs one HTTP request (`method` and `path`,
+/// with `body` for writes) and returns the response. Implementations own the medium —
+/// a real HTTP socket, or an in-memory fake.
+pub trait ConsulHttp {
+  fn request(Self, String, String, Bytes) -> ConsulResponse raise
+}
+
+///|
+/// A consul client over a `ConsulHttp`, exposing the agent operations discovery uses.
+pub struct ConsulClient {
+  http : &ConsulHttp
+}
+
+///|
+/// A client over `http`.
+pub fn ConsulClient::new(http : &ConsulHttp) -> ConsulClient {
+  { http, }
+}
+
+///|
+/// Perform a request and require a 2xx status, mapping a transport failure or a non-2xx
+/// response to `ConsulError`.
+fn ConsulClient::call(
+  self : ConsulClient,
+  verb : String,
+  path : String,
+  body : Bytes,
+) -> ConsulResponse raise ConsulError {
+  let resp = self.http.request(verb, path, body) catch {
+    e => raise ConsulError("consul transport error: " + e.to_string())
+  }
+  if resp.status < 200 || resp.status >= 300 {
+    raise ConsulError(
+      verb +
+      " " +
+      path +
+      " -> HTTP " +
+      resp.status.to_string() +
+      ": " +
+      @utf8.decode_lossy(resp.body[:]),
+    )
+  }
+  resp
+}
+
+///|
+/// The check id consul assigns an inline service TTL check: `service:<service-id>`.
+pub fn consul_check_id(service_id : String) -> String {
+  "service:" + service_id
+}
+
+///|
+/// `PUT /v1/agent/service/register`: register an instance under `name` at
+/// `address:port` with an id, held alive by a TTL check that consul deregisters
+/// `ttl*3` seconds after it stops passing. Pass the check with `check_pass` before each
+/// TTL lapses to stay healthy.
+pub fn ConsulClient::register_service(
+  self : ConsulClient,
+  id : String,
+  name : String,
+  address : String,
+  port : Int,
+  ttl_secs : Int,
+) -> Unit raise ConsulError {
+  let check : Map[String, Json] = Map([
+    ("TTL", (ttl_secs.to_string() + "s").to_json()),
+    (
+      "DeregisterCriticalServiceAfter",
+      ((ttl_secs * 3).to_string() + "s").to_json(),
+    ),
+  ])
+  let body : Map[String, Json] = Map([
+    ("ID", id.to_json()),
+    ("Name", name.to_json()),
+    ("Address", address.to_json()),
+    ("Port", port.to_json()),
+    ("Check", check.to_json()),
+  ])
+  let _ = self.call(
+    "PUT",
+    "/v1/agent/service/register",
+    @utf8.encode(body.to_json().stringify()),
+  )
+}
+
+///|
+/// `PUT /v1/agent/check/pass/service:<id>`: mark an instance's TTL check passing, the
+/// keep-alive that renews its lease.
+pub fn ConsulClient::check_pass(
+  self : ConsulClient,
+  service_id : String,
+) -> Unit raise ConsulError {
+  let _ = self.call(
+    "PUT",
+    "/v1/agent/check/pass/" + consul_check_id(service_id),
+    b"",
+  )
+}
+
+///|
+/// `PUT /v1/agent/service/deregister/<id>`: deregister one instance.
+pub fn ConsulClient::deregister_service(
+  self : ConsulClient,
+  id : String,
+) -> Unit raise ConsulError {
+  let _ = self.call("PUT", "/v1/agent/service/deregister/" + id, b"")
+}
+
+///|
+/// `GET /v1/health/service/<name>?passing=true`: the healthy instances of `name`, each
+/// as its `Service.Address:Service.Port` endpoint (falling back to `Node.Address` when
+/// the service advertises no address of its own, as consul's clients do).
+pub fn ConsulClient::health_service(
+  self : ConsulClient,
+  name : String,
+) -> Array[Endpoint] raise ConsulError {
+  let resp = self.call(
+    "GET",
+    "/v1/health/service/" + name + "?passing=true",
+    b"",
+  )
+  parse_health_response(resp.body)
+}
+
+///|
+/// `GET /v1/agent/services`: every service instance registered on this agent, as
+/// `(instance-id, service-name)` pairs — the list a by-name deregister filters to find
+/// the ids to drop.
+pub fn ConsulClient::agent_services(
+  self : ConsulClient,
+) -> Array[(String, String)] raise ConsulError {
+  let resp = self.call("GET", "/v1/agent/services", b"")
+  let text = @utf8.decode_lossy(resp.body[:])
+  let json = @json.parse(text) catch {
+    e => raise ConsulError("invalid consul JSON: " + e.to_string())
+  }
+  let obj = match json {
+    Object(o) => o
+    _ => raise ConsulError("consul agent/services is not a JSON object")
+  }
+  let out : Array[(String, String)] = []
+  for _id, svc in obj {
+    let id = consul_str_field(svc, "ID")
+    let name = consul_str_field(svc, "Service")
+    if id != "" {
+      out.push((id, name))
+    }
+  }
+  out
+}
+
+///|
+/// Parse a consul `health/service` JSON array into endpoints.
+fn parse_health_response(body : Bytes) -> Array[Endpoint] raise ConsulError {
+  let text = @utf8.decode_lossy(body[:])
+  let json = @json.parse(text) catch {
+    e => raise ConsulError("invalid consul JSON: " + e.to_string())
+  }
+  let entries = match json {
+    Array(a) => a
+    _ => raise ConsulError("consul health response is not a JSON array")
+  }
+  let out : Array[Endpoint] = []
+  for entry in entries {
+    let service = match json_field(entry, "Service") {
+      Some(s) => s
+      None => continue
+    }
+    let node_addr = match json_field(entry, "Node") {
+      Some(n) => consul_str_field(n, "Address")
+      None => ""
+    }
+    let svc_addr = consul_str_field(service, "Address")
+    let address = if svc_addr == "" { node_addr } else { svc_addr }
+    let port = consul_int_field(service, "Port")
+    if address != "" && port > 0 {
+      out.push(Endpoint::new(address, port))
+    }
+  }
+  out
+}
+
+///|
+/// The value of `key` in a JSON object, or `None` if `obj` is not an object or lacks
+/// the key.
+fn json_field(obj : Json, key : String) -> Json? {
+  if obj is Object(m) {
+    m.get(key)
+  } else {
+    None
+  }
+}
+
+///|
+/// The string value of `key` in a JSON object, or `""`.
+fn consul_str_field(obj : Json, key : String) -> String {
+  match json_field(obj, key) {
+    Some(String(s)) => s
+    _ => ""
+  }
+}
+
+///|
+/// The integer value of a JSON number `key`, truncated toward zero, or `0`.
+fn consul_int_field(obj : Json, key : String) -> Int {
+  match json_field(obj, key) {
+    Some(Number(n, ..)) => n.to_int()
+    _ => 0
+  }
+}

consul_discovery.mbt

@@ -0,0 +1,92 @@
+// The consul-backed discovery driver: the same register -> resolve -> deregister flow
+// as the etcd and redis drivers, expressed over consul's agent API. An instance
+// registers as a consul service with a TTL check (the lease), a keep-alive passes that
+// check, and a client resolves a service to its healthy instances. This is a third
+// `Resolve` for the balancer and load-balanced channel, chosen by swapping the driver.
+
+///|
+/// A consul-backed service registry / resolver over a `ConsulClient`. Each instance is
+/// a consul service named `service`, uniquely identified per agent by its address.
+pub struct ConsulDiscovery {
+  client : ConsulClient
+}
+
+///|
+/// A consul discovery over `client`.
+pub fn ConsulDiscovery::new(client : ConsulClient) -> ConsulDiscovery {
+  { client, }
+}
+
+///|
+/// The per-agent-unique instance id for `endpoint` of `service`:
+/// `<service>-<host>-<port>`.
+fn consul_instance_id(service : String, endpoint : Endpoint) -> String {
+  service + "-" + endpoint.host + "-" + endpoint.port.to_string()
+}
+
+///|
+/// Register `endpoint` for `service` with a `ttl`-second TTL check and immediately pass
+/// the check so the instance is healthy at once (a fresh TTL check starts critical).
+/// Returns the instance id; renew it with `keepalive` before the TTL lapses.
+pub fn ConsulDiscovery::register(
+  self : ConsulDiscovery,
+  service : String,
+  endpoint : Endpoint,
+  ttl? : Int = 10,
+) -> String raise {
+  let id = consul_instance_id(service, endpoint)
+  self.client.register_service(id, service, endpoint.host, endpoint.port, ttl)
+  self.client.check_pass(id)
+  id
+}
+
+///|
+/// Refresh an instance's lease by passing its TTL check.
+pub fn ConsulDiscovery::keepalive(
+  self : ConsulDiscovery,
+  service : String,
+  endpoint : Endpoint,
+) -> Unit raise {
+  self.client.check_pass(consul_instance_id(service, endpoint))
+}
+
+///|
+/// Resolve `service` to its healthy endpoints.
+pub fn ConsulDiscovery::resolve(
+  self : ConsulDiscovery,
+  service : String,
+) -> Array[Endpoint] raise {
+  self.client.health_service(service)
+}
+
+///|
+/// Deregister one instance of `service`.
+pub fn ConsulDiscovery::deregister_instance(
+  self : ConsulDiscovery,
+  service : String,
+  endpoint : Endpoint,
+) -> Unit raise {
+  self.client.deregister_service(consul_instance_id(service, endpoint))
+}
+
+///|
+/// Deregister every instance of `service` registered on this agent.
+pub fn ConsulDiscovery::deregister(
+  self : ConsulDiscovery,
+  service : String,
+) -> Unit raise {
+  for pair in self.client.agent_services() {
+    let (id, name) = pair
+    if name == service {
+      self.client.deregister_service(id)
+    }
+  }
+}
+
+///|
+/// This consul discovery as a `Resolve` interface value, so the balancer and the
+/// load-balanced channel run against consul unchanged. A resolve error surfaces as an
+/// empty endpoint set, matching the other drivers.
+pub fn ConsulDiscovery::resolver(self : ConsulDiscovery) -> Resolve {
+  service => self.resolve(service) catch { _ => [] }
+}