Service discovery¶

Service discovery is two problems wearing one name: publishing where a service instance lives, and finding one to talk to. etcd-recipes splits them into separate types and gives you a façade over both:

All four extend EtcdConnector, so they share the lifecycle, exception, and connection-state surface described in Core concepts.

Every key this recipe writes lives under <servicePath>/names, one key per instance at <servicePath>/names/<serviceName>/<id>. That layout is why the read-side types accept a names path rather than the service path — more on that below.

ServiceInstance¶

ServiceInstance is what gets written to etcd: a @Serializable Kotlin data class whose JSON is the wire format.

@Serializable
data class ServiceInstance(
  val name: String,
  var jsonPayload: String,
  var address: String = "",
  var port: Int = -1,
  var sslPort: Int = -1,
  var registrationTimeUTC: Long = Instant.now().toEpochMilli(),
  var serviceType: ServiceType = ServiceType.DYNAMIC,
  var uri: String = "",
  var enabled: Boolean = true,
) {
  val id: String = randomId(TOKEN_LENGTH)
}

The fields it does not understand live in jsonPayload, an opaque String the library never parses. id is random and assigned at construction: it is the last segment of the instance's etcd key, and the id that queryForInstance(name, id) takes.

val instance =
  serviceInstance("worker", """{"weight":5}""") {
    apply {
      address = "10.0.0.7"
      port = 8080
      sslPort = 8443
      uri = "https://10.0.0.7:8443"
      serviceType = ServiceType.DYNAMIC
      enabled = true
    }
  }

// id is random and assigned at construction. It is the last segment of the
// instance's etcd key, and what queryForInstance(name, id) takes.
logger.info { "${instance.name}/${instance.id} isDynamic=${instance.serviceType.isDynamic}" }

// toJson()/toObject() are the wire format the recipes store and read.
val parsed: ServiceInstance = ServiceInstance.toObject(instance.toJson())
logger.info { "Round-tripped ${parsed.name} on port ${parsed.port}" }

// The builder's setters return void, so there is no fluent chain to follow —
// set the fields, then build().
ServiceInstance.Companion.ServiceInstanceBuilder builder =
  ServiceInstance.newBuilder("worker", "{\"weight\":5}");
builder.setAddress("10.0.0.7");
builder.setPort(8080);
builder.setSslPort(8443);
builder.setUri("https://10.0.0.7:8443");
builder.setServiceType(ServiceType.DYNAMIC);
builder.setEnabled(true);

ServiceInstance instance = builder.build();

// id is random and assigned at construction. It is the last segment of the
// instance's etcd key, and what queryForInstance(name, id) takes.
System.out.println(instance.getName() + "/" + instance.getId());
System.out.println("isDynamic: " + instance.getServiceType().isDynamic());

// toJson()/toObject() are the wire format the recipes store and read.
ServiceInstance parsed = ServiceInstance.toObject(instance.toJson());
System.out.println("Round-tripped " + parsed.getName() + " on port " + parsed.getPort());

toJson() and the @JvmStatic toObject(json) are the round trip. newBuilder(name, jsonPayload) is the Java-facing constructor path; Kotlin has the serviceInstance(name, jsonPayload) { } DSL, whose lambda runs against a ServiceInstanceBuilder.

Typed payloads¶

ServiceInstance.payload(codec), setPayload(value, codec), and the serviceInstance(name, payload, codec) { } builder let you treat jsonPayload as a typed value without changing what lands in etcd:

// A UTF-8 text codec is required: jsonPayload is a String, so a binary codec
// cannot round-trip through it.
val codec = jsonCodec<WorkerMeta>()

val instance =
  serviceInstance("worker", WorkerMeta("us-east", 5), codec) {
    apply { port = 8080 }
  }

// Decode it back; the ServiceInstance wire format is unchanged by typing.
val meta: WorkerMeta = instance.payload(codec)
logger.info { "Registered in ${meta.region} with weight ${meta.weight}" }

// Re-encode a new value into jsonPayload in place, then updateService().
instance.setPayload(WorkerMeta("us-east", 9), codec)

// jsonPayload is a String, so the codec must be a UTF-8 text codec. jsonCodec()
// is `inline reified` and therefore Kotlin-only; from Java use StringCodec, or
// the Jackson module's JSON codec.
ServiceInstance instance =
  TypedServiceInstances.serviceInstance("worker", "{\"weight\":5}", StringCodec.INSTANCE);

String payload = TypedServiceInstances.payload(instance, StringCodec.INSTANCE);
System.out.println("Payload: " + payload);

// Re-encodes into jsonPayload in place; follow it with updateService().
TypedServiceInstances.setPayload(instance, "{\"weight\":9}", StringCodec.INSTANCE);

The wire format of toJson() is byte-for-byte unchanged by typing, so adopting a typed payload does not break clients still reading jsonPayload as a raw string. In Java, note that jsonCodec() is inline reified and therefore Kotlin-only; the extensions themselves are reachable as statics on TypedServiceInstances. See the Java guide.

ServiceType¶

enum class ServiceType { DYNAMIC, STATIC, PERMANENT, DYNAMIC_SEQUENTIAL }

serviceType.isDynamic is true for DYNAMIC and DYNAMIC_SEQUENTIAL. The enum is metadata you publish for your own consumers: the recipes do not branch on it. Every registration is lease-bound regardless of what the field says.

ServiceDiscovery¶

The façade. It owns a ServiceRegistry internally, hands out caches and providers, and closes all of them when it closes.

ServiceDiscovery(client, "/services/orders").use { discovery ->
  val instance =
    serviceInstance("worker", """{"weight":5}""") {
      apply {
        address = "10.0.0.7"
        port = 8080
      }
    }

  // Writes /services/orders/names/worker/<id>, bound to a self-healing lease.
  discovery.registerService(instance)

  // Mutate the instance, then push the new JSON to the same key.
  instance.port = 8081
  discovery.updateService(instance)

  // Removes the key now rather than leaving it to expire at the TTL.
  discovery.unregisterService(instance)
}

try (ServiceDiscovery discovery = new ServiceDiscovery(client, "/services/orders")) {
  ServiceInstance instance = ServiceInstance.newBuilder("worker", "{\"weight\":5}").build();
  instance.setAddress("10.0.0.7");
  instance.setPort(8080);

  // Writes /services/orders/names/worker/<id>, bound to a self-healing lease.
  discovery.registerService(instance);

  // Mutate the instance, then push the new JSON to the same key.
  instance.setPort(8081);
  discovery.updateService(instance);

  // Removes the key now rather than leaving it to expire at the TTL.
  discovery.unregisterService(instance);
}

registerService, updateService, and unregisterService all throw EtcdRecipeException: registration is a compare-and-swap that can lose, an update to something never registered is a caller bug, and both are worth failing loudly rather than silently no-oping.

The direct queries are one-shot reads — no cache, no watch, a range GET per call:

ServiceDiscovery(client, "/services/orders").use { discovery ->
  // One-shot reads: every call is a range GET, not a cached lookup.
  val names: List<String> = discovery.queryForNames()
  val instances: List<ServiceInstance> = discovery.queryForInstances("worker")

  logger.info { "Services: $names, worker instances: ${instances.size}" }

  // Throws EtcdRecipeException if that exact instance key is gone.
  instances.firstOrNull()?.let { first ->
    val one = discovery.queryForInstance("worker", first.id)
    logger.info { "Found ${one.name}/${one.id}" }
  }
}

try (ServiceDiscovery discovery = new ServiceDiscovery(client, "/services/orders")) {
  // One-shot reads: every call is a range GET, not a cached lookup.
  List<String> names = discovery.queryForNames();
  List<ServiceInstance> instances = discovery.queryForInstances("worker");

  System.out.println("Services: " + names + ", worker instances: " + instances.size());

  // Throws EtcdRecipeException if that exact instance key is gone.
  if (!instances.isEmpty()) {
    ServiceInstance one = discovery.queryForInstance("worker", instances.get(0).getId());
    System.out.println("Found " + one.getName() + "/" + one.getId());
  }
}

withServiceDiscovery, withServiceCache, and withServiceProvider are the scoped Kotlin forms:

// Closes the façade — and every cache and provider it handed out — on exit.
withServiceDiscovery(client, "/services/orders") {
  val instance = serviceInstance("worker", """{"weight":5}""")
  registerService(instance)
  logger.info { "Names: ${queryForNames()}" }
}

ServiceRegistry¶

The write side on its own. Depend on it when a process only publishes itself and never looks anything up — the read side brings watches and background threads a publisher has no use for.

// Write side only. Registration leases self-heal: if one expires, the healer
// grants a new lease and re-runs the registration CAS.
ServiceRegistry(client, "/services/orders", leaseTtlSecs = 5L).use { registry ->
  registry.addLeaseListener { event ->
    when (event) {
      is LeaseEvent.Expired -> logger.warn { "Registration lease expired; healing" }
      is LeaseEvent.Restored -> logger.info { "Re-registered under lease ${event.newLeaseId}" }
      is LeaseEvent.Failed -> logger.error { "Healing abandoned; this instance is unregistered" }
      is LeaseEvent.Suspended -> logger.warn { "Keep-alive stream is retrying" }
    }
  }

  val instance = serviceInstance("worker", """{"weight":5}""")
  registry.registerService(instance)
}

// Write side only. Registration leases self-heal: if one expires, the healer
// grants a new lease and re-runs the registration CAS.
try (ServiceRegistry registry = new ServiceRegistry(client, "/services/orders", 5L)) {
  registry.addLeaseListener(event ->
    System.out.println("Lease event: " + event));

  ServiceInstance instance = ServiceInstance.newBuilder("worker", "{\"weight\":5}").build();
  registry.registerService(instance);
}

Registration leases self-heal¶

This is the opposite choice from the one locks make, and the reason is worth stating plainly.

A lock's lease is deliberately never healed: if it expires, etcd has already given the lock to somebody else, and reclaiming it would corrupt mutual exclusion. A registration has no such rival. Nobody else is trying to be your instance id, so when the lease expires the only correct behaviour is to put the key back:

• on expiry the healer grants a new lease and re-runs the registration CAS,
• LeaseEvent.Expired then LeaseEvent.Restored arrive at every addLeaseListener,
• if healing is abandoned you get LeaseEvent.Failed, and the instance is genuinely gone from etcd until you re-register it.

The window between expiry and heal is real: for a few seconds your instance is absent from every consumer's cache and will not be selected. That is correct — a partitioned instance probably should not be selected — but it means registration is not a fire-and-forget call you can stop watching. See Leases and loss.

ServiceCache¶

The read side, backed by a watch. start() takes a consistent snapshot, then anchors the watch at that exact revision so there is no overlap and no gap. After that, instances is an in-memory read.

ServiceDiscovery(client, "/services/orders").use { discovery ->
  discovery.withServiceCache("worker") {
    // Register listeners BEFORE start(): the snapshot taken by start() does not
    // replay through them, and instances registered after it do.
    addListenerForChanges { eventType, isAdd, serviceName, serviceInstance ->
      logger.info { "$eventType isAdd=$isAdd $serviceName -> $serviceInstance" }
    }

    // One-shot: a second start() throws.
    start()

    // In-memory read of the watch-maintained set; no round trip.
    logger.info { "${instances.size} instances of worker" }
  }
}

// withServiceCache is a Kotlin inline extension; Java closes the cache itself.
// Note the path is <servicePath>/names — what ServiceDiscovery hands its
// caches internally.
try (ServiceCache cache = new ServiceCache(client, "/services/orders/names", "worker")) {
  // Register listeners BEFORE start(): the snapshot taken by start() does not
  // replay through them, and instances registered after it do.
  cache.addListenerForChanges((eventType, isAdd, serviceName, serviceInstance) ->
    System.out.println(eventType + " isAdd=" + isAdd + " " + serviceName + " -> " + serviceInstance));

  cache.addRecoveryListener(event ->
    System.out.println("Watch recovery: " + event));

  // One-shot: a second start() throws.
  cache.start();

  // In-memory read of the watch-maintained set; no round trip.
  System.out.println(cache.getInstances().size() + " instances of worker");
}

start() is one-shot: calling it twice throws EtcdRecipeRuntimeException. Register listeners first — the initial snapshot does not replay through addListenerForChanges, so a listener added afterwards only sees changes from that point on. Read instances for the current set; use the listener for the deltas.

addRecoveryListener reports the watch's own health: resubscribes after a stream death, resyncs after compaction (the cache reconciles itself against a fresh snapshot), and the Failed event that means recovery was abandoned and the instance list is now frozen and lying to you.

// Read-only side, no registry: note the path is <servicePath>/names, which is
// what ServiceDiscovery hands its caches internally.
ServiceCache(client, "/services/orders/names", "worker").use { cache ->
  cache.addRecoveryListener { event ->
    when (event) {
      is WatchRecoveryEvent.Resynced -> logger.info { "Watch resynced after compaction" }
      is WatchRecoveryEvent.Failed -> logger.error { "Watch abandoned; instances are stale" }
      else -> logger.info { "Watch recovery: $event" }
    }
  }
  cache.start()
  logger.info { "${cache.instances.size} instances" }
}

ServiceProvider¶

A client-side load balancer: it takes the cache's instance list, applies a strategy, and hands you one instance to call.

ServiceDiscovery(client, "/services/orders").use { discovery ->
  discovery.serviceProvider("worker").use { provider ->
    // start() opens an owned, watch-backed ServiceCache, so every later read is
    // in-memory. It is one-shot: a second start() throws.
    provider.start()

    // RandomStrategy by default. Throws EtcdRecipeException when nothing is
    // available — no instances registered, or all of them ejected.
    val instance = provider.getInstance()
    logger.info { "Selected ${instance.address}:${instance.port}" }

    logger.info { "${provider.getAllInstances().size} instances registered" }
  }
}

try (ServiceDiscovery discovery = new ServiceDiscovery(client, "/services/orders");
     ServiceProvider provider = discovery.serviceProvider("worker")) {
  // start() opens an owned, watch-backed ServiceCache, so every later read is
  // in-memory. Skip it and each read is a direct range GET instead.
  provider.start();

  // RandomStrategy by default. Throws EtcdRecipeException when nothing is
  // available — no instances registered, or all of them ejected.
  ServiceInstance instance = provider.getInstance();
  System.out.println("Selected " + instance.getAddress() + ":" + instance.getPort());

  System.out.println(provider.getAllInstances().size() + " instances registered");
}

getInstance() throws EtcdRecipeException when nothing is available — whether because nothing is registered or because everything has been ejected. It does not return null: "no instance to call" is an error your request path has to handle, not a value to ?.-away.

Two read modes¶

start() is optional, and what it changes is where reads come from:

ServiceDiscovery(client, "/services/orders").use { discovery ->
  discovery.serviceProvider("worker").use { provider ->
    // No start(): every read is a fresh range GET against etcd. Fine for a
    // one-off lookup, wrong for a hot path.
    val instance = provider.getInstance()
    logger.info { "Selected ${instance.address}:${instance.port}" }
  }
}

Use the direct mode for a one-off lookup in a script or a startup path. Use the cache-backed mode for anything that selects an instance per request — a GET per outbound call turns etcd into a hard dependency of your request path, which is exactly what a discovery cache exists to prevent. start() is one-shot, and close() on an un-started provider is a clean no-op.

Ejecting failing instances¶

noteError(instance) is how a provider learns that an instance it handed you did not work. After errorThreshold errors (default 3) the instance drops out of selection for downPeriod (default 30 seconds), then rejoins automatically:

ServiceDiscovery(client, "/services/orders").use { discovery ->
  discovery
    .serviceProvider(
      "worker",
      strategy = RoundRobinStrategy(),
      errorThreshold = 3,
      downPeriod = 30.seconds,
    ).use { provider ->
      provider.start()

      val instance = provider.getInstance()
      if (!sendRequest(instance)) {
        // Pass back the SAME instance, unmodified: ejection is keyed on the
        // instance's value, so a mutated copy silently records nothing useful.
        // After errorThreshold errors it drops out of selection for downPeriod,
        // then rejoins automatically.
        provider.noteError(instance)
      }
    }
}

try (ServiceDiscovery discovery = new ServiceDiscovery(client, "/services/orders");
     // A fresh RoundRobinStrategy for THIS provider: it carries a cursor, and
     // sharing one across providers interleaves their rotations.
     ServiceProvider provider = discovery.serviceProvider("worker", new RoundRobinStrategy(), 3)) {
  provider.start();

  ServiceInstance instance = provider.getInstance();
  if (!sendRequest(instance)) {
    // Pass back the SAME instance, unmodified: ejection is keyed on the
    // instance's value, so a mutated copy silently records nothing useful.
    // After errorThreshold errors it drops out of selection for downPeriod,
    // then rejoins automatically.
    provider.noteError(instance);
  }
}

Strategies¶

fun interface ProviderStrategy {
  fun select(instances: List<ServiceInstance>): ServiceInstance?
}

Three ship with the library:

ServiceDiscovery(client, "/services/orders").use { discovery ->
  // A fresh RoundRobinStrategy for THIS provider: it carries a cursor, and
  // sharing one across providers interleaves their rotations.
  discovery.withServiceProvider("worker", RoundRobinStrategy()) {
    start()
    repeat(6) { logger.info { "-> ${getInstance().port}" } }
  }
}

ServiceDiscovery(client, "/services/orders").use { discovery ->
  // Session affinity: keeps returning the same instance while it is still
  // registered, then delegates a fresh pick. Also stateful — one per provider.
  discovery.withServiceProvider("worker", StickyStrategy(RandomStrategy)) {
    start()
    logger.info { "Pinned to ${getInstance().port}" }
  }
}

The Java form is the ejection example above: discovery.serviceProvider("worker", new RoundRobinStrategy(), 3). withServiceProvider is an inline Kotlin extension, so Java constructs the provider and closes it itself — see the Java guide.

ProviderStrategy is a fun interface, so a custom policy is a lambda. Return null when the list is empty; the provider turns that into the EtcdRecipeException:

// ProviderStrategy is a fun interface: return null when the list is empty.
val leastLoaded = ProviderStrategy { instances -> instances.minByOrNull { it.port } }

ServiceDiscovery(client, "/services/orders").use { discovery ->
  discovery.withServiceProvider("worker", leastLoaded) {
    start()
    logger.info { "Selected ${getInstance().port}" }
  }
}

Strategies see only the available instances — the live set minus anything currently ejected — so a strategy never has to know about noteError at all.

A provider built directly rather than through the façade takes the same names path caveat as ServiceCache:

// Read side only, no ServiceDiscovery façade. Note the path is <servicePath>/names.
ServiceProvider(client, "/services/orders/names", "worker", RoundRobinStrategy()).use { provider ->
  provider.start()
  logger.info { "Selected ${provider.getInstance().port}" }
}

Coroutines¶

ServiceDiscovery's blocking calls have suspending twins — awaitRegisterService, awaitUpdateService, awaitUnregisterService, awaitQueryForNames, awaitQueryForInstances, awaitQueryForInstance — as does ServiceCache.awaitStart(). See Coroutines.

The cache is also a natural Flow source: ServiceCache.eventsAsFlow() and recoveryEventsAsFlow(), plus ServiceRegistry.leaseEventsAsFlow() for the healing lifecycle. Collection registers a listener and cancellation removes it — it never starts or closes the recipe. See Flows.

Observability¶

With the Micrometer module wired in, ServiceCache reports etcd.cache.sync (a timer over snapshot loads) and etcd.cache.size, and its watch recoveries land on the etcd.watch.recovery counter. A registry's healing shows up as connection-state transitions. See Observability.