4
\$\begingroup\$

I need a distributed lock system for a service I'm writing. (There can be multiple instances of this service running at a time, and I need to make sure that only one instance does a particular cron job.)

According to this page, S3 has strong consistency, so I thought I'd make an attempt at implementing a distributed lock using S3 (the cron job does a bunch of processing of S3 objects, so it seemed a convenient option).

I realize distributed locks are complicated to implement. I've tested a version of this code out, and it seems to work okay. I've since added a timeout/heartbeat. Now, I want someone to rip it apart and explain why it shouldn't be used in production because of X, Y, or Z failure in the code.

By way of a quick outline of the algorithm:

  • We attempt to get a lock on the resources by putting an object at a particular place in s3 (see s3InstanceLockObjectKey).
  • We then use the leader election pattern to determine who, and in particular, if the current instance, is the leader.
    • We make a request to get all of the lock objects in S3 and ignore any that are prior to the timeout period (e.g. any lock objects older than 10 minutes are considered stale)
    • We sort the lock objects first by last modified date and then lexicographically, and choose the first one as the leader. So, we should get whoever put the lock object into S3 first, and if several did at the same time, the lexicographical sorting makes the final determination. — If the current instance is the leader, we setup a scheduled task to maintain leadership by continually updating the instance's lock object in S3 so that it doesn't become stale. Finally, we return true from the requestLock method, otherwise false.
  • The releaseLock cancels the scheduled task and deletes any lock objects in S3.
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.Delete;
import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.S3Object;

import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

public class DistributedS3LockService {

  public static class Builder {
    private S3Client s3Client;
    private String bucketName;
    private String s3LockObjectPrefix = "__distributedS3Lock";
    private long timeoutPeriod = 5;
    private TimeUnit timeoutUnit = TimeUnit.MINUTES;

    public Builder s3Client(S3Client s3Client) {
      this.s3Client = s3Client;
      return this;
    }

    public Builder bucketName(String bucketName) {
      this.bucketName = bucketName;
      return this;
    }

    public Builder s3LockObjectPrefix(String s3LockObjectPrefix) {
      this.s3LockObjectPrefix = s3LockObjectPrefix;
      return this;
    }

    public Builder timeoutPeriod(long timeoutPeriod) {
      this.timeoutPeriod = timeoutPeriod;
      return this;
    }

    public Builder timeoutUnit(TimeUnit timeoutUnit) {
      this.timeoutUnit = timeoutUnit;
      return this;
    }

    public DistributedS3LockService build() {
      if (s3Client == null) {
        throw new IllegalArgumentException("s3Client is required");
      }
      if (bucketName == null) {
        throw new IllegalArgumentException("bucketName is required");
      }
      return new DistributedS3LockService(
          s3Client, bucketName, s3LockObjectPrefix, timeoutPeriod, timeoutUnit);
    }
  }

  private static final String INSTANCE_ID = UUID
      .randomUUID()
      .toString()
      .replaceAll("-", "");

  private final S3Client s3Client;
  private final String bucketName;
  private final String s3LockObjectPrefix;
  private final String s3InstanceLockObjectKey;
  private final long timeoutPeriod;
  private final TimeUnit timeoutUnit;
  private ScheduledExecutorService scheduledExecutorService;

  public DistributedS3LockService(
      S3Client s3Client,
      String bucketName,
      String s3LockObjectPrefix,
      long timeoutPeriod,
      TimeUnit timeoutUnit) {
    this.s3Client = s3Client;
    this.bucketName = bucketName;
    this.s3LockObjectPrefix = s3LockObjectPrefix;
    this.s3InstanceLockObjectKey = "%s_%s".formatted(s3LockObjectPrefix, INSTANCE_ID);
    this.timeoutPeriod = timeoutPeriod;
    this.timeoutUnit = timeoutUnit;
  }

  public boolean requestLock() {
    putLockObject();
    if (isLeader()) {
      scheduleLockMaintenance();
      return true;
    }
    return false;
  }

  public boolean releaseLock() {
    scheduledExecutorService.shutdown();
    try {
      return scheduledExecutorService.awaitTermination(60, TimeUnit.SECONDS);
    }
    catch (InterruptedException e) {
      return false;
    }
    finally {
      deleteByPrefix(s3InstanceLockObjectKey);
    }
  }

  private void scheduleLockMaintenance() {
    if (scheduledExecutorService == null) {
      scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    }
    scheduledExecutorService.scheduleAtFixedRate(this::putLockObject, 0, timeoutPeriod, timeoutUnit);
  }

  private boolean isLeader() {
    return s3InstanceLockObjectKey.equals(
      getObjectsByPrefix(s3LockObjectPrefix)
          .filter(o -> o.lastModified().isAfter(Instant.now().minus(5, ChronoUnit.MINUTES)))
          .min(Comparator.comparing(S3Object::lastModified).thenComparing(S3Object::key))
          .map(S3Object::key)
          .orElse(null));
  }

  private Stream<S3Object> getObjectsByPrefix(String prefix) {
    return s3Client.listObjectsV2Paginator(ListObjectsV2Request.builder()
            .bucket(bucketName)
            .prefix(prefix)
            .build())
        .stream()
        .flatMap(response -> response.contents().stream());
  }

  private void putLockObject() {
    s3Client.putObject(
        PutObjectRequest.builder()
            .bucket(bucketName)
            .key(s3InstanceLockObjectKey)
            .build(),
        RequestBody.fromString(""));
  }

  private void deleteByPrefix(String prefix) {
    List<ObjectIdentifier> objects = getObjectsByPrefix(prefix)
        .map(o -> ObjectIdentifier.builder()
            .key(o.key())
            .build())
        .toList();

    if (objects.isEmpty()) {
      return;
    }

    s3Client.deleteObjects(DeleteObjectsRequest.builder()
        .bucket(bucketName)
        .delete(Delete.builder()
            .objects(objects)
            .build())
        .build());
  }
}
\$\endgroup\$

2 Answers 2

4
\$\begingroup\$

be explicit about invariants

Write them down.

In the code.


builder pattern

    public DistributedS3LockService build() {
      if (s3Client == null) {
        throw new IllegalArgumentException("s3Client is required");
      }
      if (bucketName == null) {
        throw new IllegalArgumentException("bucketName is required");
      }
      ...

nit: I probably would have required both those attributes in the ctor, so this situation simply never arises. It gets back to class invariants. But that's just me, reasonable people will differ, you were going for a certain pattern. Keep the code as-is.


human interaction with UUIDs

      .replaceAll("-", "");

I kind of like the dashes in there. Why? In practice, when eyeballing N filenames that were all created at same timestamp, I'm going to mentally refer to them by some abbreviated prefix of hex characters. And the dash helps with visually chunking them. But whatever, make no change, clearly it doesn't alter correctness.


  public DistributedS3LockService(

This ctor really needs a /** javadoc */ sentence, explaining that "to come in the front door" caller actually should be looking at Builder instead of looking here.

We anticipate that caller would typically accept the default timeout. It should be very, very hard for callers to accidentally use same prefix with different timeouts. (Should timeout perhaps be embedded within the prefix? Hmmmm....)

The general concern, with any Public API we design, is that not only should it be easy to obtain correct results. It should also be hard for caller to obtain wrong results which he then misinterprets and blindly trusts.

  private static final String INSTANCE_ID = UUID ...

    this.s3InstanceLockObjectKey = "%s_%s".formatted(s3LockObjectPrefix, INSTANCE_ID);

Same concern. Your target audience is callers in separate JVM processes, typically on separate machines. Here, I worry that your two callers might accidentally be two threads running within same JVM, sharing the same INSTANCE_ID. Maybe initialize it within the ctor? Maybe there's some other simple way to signal that caller shouldn't do that, which I've not yet noticed?

If you want to collapse on things like hostname, IP address, or PID, consider making such things part of the s3InstanceLockObjectKey.

I feel that

    ... .formatted(s3LockObjectPrefix, INSTANCE_ID)

should become some version of

    ... .formatted(s3LockObjectPrefix, timestamp, INSTANCE_ID)

even if timestamp has very coarse granularity, perhaps an ISO-8601 day.

Why? Imagine that Bad Things happen, perhaps due to buggy code. And we leave "turd" files behind. Lots of them. And we want to skip them or delete them.

Lexical order is an important part of S3 directory listings. Being able to retrieve strictly the "recent" subset of listings is often an advantage. So I commonly will put stamp before guid in an object name, just in case.

I see your ... o.lastModified().isAfter(Instant.now().minus(5, ... expression. But please understand that client has to retrieve essentially all lock files before applying that filter on client side. Proper design of lexically ordered object names lets you push the filtering to amazon's backend, so the presence of a million ancient turds has no impact. And of course, with pagination it will take quite a few GET requests to obtain a million object names.


race

  public boolean requestLock() {
    putLockObject();
    if (isLeader()) { ...

In process P1 we issue putObject() followed by getObjectsByPrefix(). In process P2 .. Pn we do the same.

Looks like a race to me. Let most of the N processes putObject() simultaneously, and then they all get around to asking for a listing. It's not like we asked a central AWS daemon or redis server to atomically increment a counter. Speed of light is slow enough, and timestamp granularity is coarse enough, that a bunch of hosts could plausibly request a lock at "same" time.

There's a whole literature on this topic. The code offers zero citations. It needs to mention raft or some simpler concensus approach like a bakery. As it stands, the code's not making a persuasive enough argument of correctness to convince me.


Let P1 and P2 be amazon EC2 instances running in West and East coast datacenters. They use just a single bucket.

causality

Any distributed protocol is going to wind up depending on "happens before" relationships, on causality.

Here is an example that happens at roughly noon UTC:

  • at 12:00:01, P1 requests create + write of "one" to S3 object "readme".
  • at 12:00:02, P2 requests create + write of "two" to S3 object "readme".
  • at 12:00:03, P1 reads "readme".

What did P1 read? It's indeterminate. Either of the create requests could have taken a "long" time, perhaps more than one second. Either "one" or "two" could come back.

Here's another example:

  1. P1 requests create + write of "one" to S3 object "readme", and then
  2. P1 sends message "Hi!" to P2.
  3. P2 receives that greeting, and then
  4. P2 requests create + write of "two" to S3 object "readme", and then
  5. P2 sends message "ACK" to P1.
  6. P1 receives that acknowledgement, and then
  7. P1 reads "readme".

Now we know that P1 obtained "two" at the end. Notice that in steps (1.) and (4.) we send request to amazon S3 and await a successful response before continuing.

The greeting and acknowledgement messages established causal happens-before relationships between the steps. The S3 documentation you referenced makes useful guarantees under such circumstances, in contrast to the earlier example where the result is indeterminate.


  public boolean releaseLock() {
    scheduledExecutorService.shutdown();
    try {
      return scheduledExecutorService.awaitTermination(60, TimeUnit.SECONDS);

It seems like this timeout of 1 minute has a relationship to the timeoutPeriod of 5 minutes, and it should be at least that long.

Also, I worry that a "slow to terminate" participant can impact service availability for replacement servers trying to come up to mitigate an outage.


    scheduledExecutorService.scheduleAtFixedRate(this::putLockObject, 0, timeoutPeriod, timeoutUnit);

Wouldn't you like to schedule lock maintenance at, IDK, 0.9 * timeoutPeriod ? (Which would motivate using 300 seconds rather than 5 minutes.) Else we invite a race with freshly booted participants every five minutes.

Also, I worry that we daily can produce more than a hundred versions of the S3 object during normal operations. There's no documentation in this code on appropriate bucket settings to ensure that amazon will age out those ancient versions. Better, this submission should include code which automatically deletes such ancient turds.


I didn't find a compelling correctness argument.

I wouldn't use this in production. I'm sure an S3 approach could be made to work, but incrementing a redis counter seems like much less work to accomplish your goal.

\$\endgroup\$
1
\$\begingroup\$

This answer is not really a code review.

If you are implementing a distributed locking system chances are, you are trying to protect a resource being utilized by more than one nodes/ threads/ processes at any given point of time.

I faced similar a situation also and tried to build a distributed locking system using Redis and tried come up with a lot of edge case scenarios.

But in reality, production use cases will always surprise us.

So instead of spending effort on implementing something new, we leveraged something that is already available out of the box and battle tested.

The solution we used is simple:

  1. Modify your logic to use the pub/sub design pattern.
  2. Create a FIFO queue (AWS) or Ordered Subscription (GCP)
  3. Publish your events the queue and implement a consumer to do your business logic.
  4. The queue provider will make sure, the events published to the queue are processed in order and one at a time.
  5. This approach is a lot simpler and works practically very well.
  6. You just need to take care of visibility timeouts and deadlettering.

I hope this helps.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.