// 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`:
/// `--`.
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 { _ => [] }
}