Pragmatic Dictator

Those specifying requirements often express them without consideration for the passing of time, assuming that actions are instantaneous. A naive development team with limited experience in distributed systems will then make the classic mistake of attempting to implement those requirements to the letter. This can lead to a bunch of undesirable outcomes including:

• Brittleness in the face of failure.
• High cost solutions.
• Poor scaling properties.
• Disappointment as the expectations of the requirements source aren’t met.

Consider a system where we have two (network) hops to an observer and one hop to the initiator of an action (assuming uniform network latency for each hop). Potentially for every two actions there will be a single observation. Thus each observation of the system is out of date by the time it reaches the observer.

Administrative actions can suffer similar problems, in that it could take several hops for the request to arrive at the system. A user may be only one hop away and could be performing many operations in the time it takes for one of our actions to reach the system. For example if we wish to block a user, whilst our request is in transit they might perform several operations.

Things are made worse by network failures which can further delay or prevent execution of an action and slow down the rate of updates to an observer.

How then do we account for these troubles when specifying requirements? By qualifying them with appropriate SLA’s. In the example above, appropriate SLA’s might include:

• Time for propagation of an administrative action.
• Maximum acceptable time after the action is triggered for a user to be blocked.

SLA’s such as the above:

1. Help us to identify appropriate solutions (e.g. do we need to pay for multiple independent routes between data-centres).
2. Allow us to make appropriate use of asynchronous operations and eventual consistency.

Since SLA’s have significant impact on the way in which a requirement will be implemented it is essential to perform appropriate expectation management, discussing and communicating the implications with the requirements source, they cannot be solely the domain of techies. Remember also that in many situations customers prefer availability over consistency.

Neglecting to account for failure is an age old problem. Consider this common error (Purify anybody?):

#include <stdio.h>
#include <stdlib.h>
struct rhubarb {
  int aVal;
  int anotherVal;
  char* aString;
};
......
  struct rhubarb* mystruct;
  mystruct = malloc(sizeof(struct rhubarb));
  mystruct->aVal = 55;
......

Of course the following code should have been included after the malloc:

/*
  If memory wasn't allocated, do something appropriate.
*/
if (mystruct == NULL) {
  .....
}

An equivalent mistake is easily possible when building a distributed system in http or RMI by ignoring error codes or exceptions that are designed to communicate failures that we ought to handle. It’s similarly easy to ignore latency, or implement brittle and dumb retry logic or assume something is reliable (like a message queue) when it isn’t. Many have managed to concoct systems with http that breach the idempotent “constraints” of REST and whilst Erlang provides link() and receive timeouts, we’re not forced to use them.

In essence there is no way to ensure developers do the right thing in a single-process or distributed context. No technology, tool or design approach can prevent developers from making poor implementation decisions which limits the value in re-hashing (Steve, Steve and Stu) RPC rights and wrongs.

I believe the best chance we have for doing distributed right is not by providing some de-facto standard toolset, rather it’s through education^[1] and mentoring to encourage the correct mindset. Such a mindset allows a developer building a distributed system to choose the most appropriate tools and use them right.

[1] Material to be covered would be substantially broader then the fallacies, failure handling, latency and should probably include: logical time, FLP, failure detectors, global snapshots and Paxos.

Amazon has had a few problems of late, one of the more interesting ones being something S3 users encountered. It took Amazon a little while to identify the root cause:

We’ve isolated this issue to a single load balancer that was brought into service at 10:55pm PDT on Friday, 6/20. It was taken out of service at 11am PDT Sunday, 6/22. While it was in service it handled a small fraction of Amazon S3′s total requests in the US. Intermittently, under load, it was corrupting single bytes in the byte stream.

Perhaps they had anticipated this scenario as the S3 API features explicit support for software-level check-summing via MD5:

For all PUT requests, Amazon S3 computes its own MD5, stores it with the object, and then returns the computed MD5 as part of the PUT response code in the ETag. By validating the ETag returned in the response, customers can verify that Amazon S3 received the correct bytes even if the Content MD5 header wasn’t specified in the PUT request. Because network transmission errors can occur at any point between the customer and Amazon S3, we recommend that all customers use the Content-MD5 header and/or validate the ETag returned on a PUT request to ensure that the object was correctly transmitted. This is a best practice that we’ll emphasize more heavily in our documentation to help customers build applications that can handle this situation.

Some developers were surprised that any of this was necessary, expecting TCP/UDP checksums to be sufficient however Stevens points out in TCP/IP Illustrated Vol I:

Takeaways:

1. Not all types of failure are binary – working or not working.
2. Leaving the responsibility of data-safety to software layers further down the stack may not be best.
3. Mechanisms for failure handling must be embedded in APIs.

It seems it’s generally accepted^[1] that SOA means breaking up your system into a set of co-operating components partitioned by business process. If you’re not doing that, you’re not doing SOA. It never ceases to amaze me how we get so zealous about fixed methods for architecting a system. I suspect it’s because we’d like to believe that architecture (and much of the act of development) can be done with fixed rules, cookie cutter style, get your catalog of patterns and technology, apply them – job done. The ultimate embodiment of this behaviour is deployment of a piece of technology in the belief that once the integration is complete the system has radically shifted in terms of it’s architecture (e.g. deploying an ESB suddenly makes your system SOA).

So if the fixed methods of SOA are thrown out and technology is not the solution, how do we build a system? Let’s first consider some of the things we’d like from our architecture:

1. Avoid integration via the database – otherwise data coupling will cripple us
2. Support for granular updates – taking down the whole system is not desirable
3. Fast rollback of changes – in case an update breaks
4. In-production testing – there’s no substitute for real traffic in tests
5. Minimal shared resources such as storage – so should there be an outage, impact is minimised
6. Horizontal scaling – more boxes equals more power
7. Support for scalable development – dev teams should be able to act in isolation most of the time
8. Support for appropriate CAP tradeoffs – making everything consistent can be bad for availability

Although we wish to avoid coupling via the database, the reality is that our code still requires access to the data in some form or another. The best we can do under this circumstance is to limit the amount of code that directly accesses the data. We achieve this by vertically slicing (as opposed to horizontal sharding) our data and consolidating the code that is most closely related to it (e.g. performs updates) into a single encapsulated unit. All other access to the data must go via the code element of its associated unit (note that one needn’t always go to a unit for the data, it’s perfectly acceptable to cache).

In this way we limit the impact of data-schema changes to it’s associated unit, other parts of the system need not be concerned but there’s still some work to do. If the code within a unit were to be co-located within all processes containing code that wishes to make use of it, we’d need to restart all those processes when we wish to deploy a new version of that code (for whatever reason). Such a deployment model also encourages several bad habits:

1. Ignoring the remoteness of the data – it’s hidden behind some form of interface and it’s tempting to attempt to hide failure behind that interface
2. Focus on synchronous method calls – it’s natural for a developer to write synchronous method calls when the code being called looks local (note that method calls can support asynchronous behaviours)

To avoid these issues, we deploy each unit in it’s own process accessed via some network endpoint that dependants use to interact with it thus:

• Each unit can now easily be allocated it’s own independent storage, apply it’s own sharding policy etc.
• The network endpoint can support multiple protocol versions or we can opt to terminate multiple network endpoints onto a unit, a powerful primitive for supporting several versions of a remote interface simultaneously.
• The network endpoint can be terminated onto some form of load balancer or custom routing implementation (which might be part of the code within the unit itself perhaps because it’s P2P based) facilitating horizontal scaling, hot upgrades, A/B testing, in-production tests etc.
• Each unit can be assigned to a development team and much work can be done independently of development efforts elsewhere, making for less contention in development.
• Each unit can implement whatever CAP tradeoff makes sense.

If we arrange for the network endpoint of each unit to be discovered dynamically at runtime we gain the ability to move our units around (e.g. for DR reasons) and have means for our system to dynamically knit itself together reducing configuration issues. Such an arrangement can also make it easier to deal with ordered startup issues (where some set of things must be available before others).

Of course it’s not all good news, we will have to manage our desire for ACID guarantees because many of the mechanisms (such as two-phase commit) for achieving this in a distributed system are fraught with problems. Fortunately, people have been thinking about this for a while. We’ll also have to take care of the fallacies but even this has some positive aspects as failure and upgrade in some cases can be considered the same (noting that abstractions for message passing, failure detectors and the like can be implemented in many languages, not just Erlang).

So what remoting approaches might we use? REST/http, WS-*, RMI, CORBA, messages, custom protocol – whatever is suitable for our situation (noting that some choices impact the means by which we can handle evolution of protocols etc). What guidelines might we follow in determining how to split our code and data? There are a number of different approaches including:

1. Considering similarities in consistency, availability and partitioning (CAP) requirements
2. Data access localities
3. Data relationships
4. Jurisdictional requirements
5. Roles and responsibilities (at coarser level than OO)
6. Features (e.g. recommendations)
7. Business processes
8. Constituent elements of an overall business process

Most systems likely require a combination of these rather than one fixed approach, taste and gut instinct count for a lot. And what might we call these units I speak of? I prefer to call them services as do a few other people but there’s no doubt that’ll be confusing, have to think of something else…….

[1] I know that Steve might well argue otherwise.

I mentioned a while back that one could exploit DNS to ease some of the common static configuration issues around hostnames, ports etc. What follows is a simple outline solution, we’ve moved a long way beyond this at Betfair but the details will have to remain secret for now (sorry).

Let’s assume that we have several different releases in testing at any one time such that we wish to segment our development/testing systems into separate enclaves (each handling a separate release) and may wish to add more enclaves over time. Assume also that production is an enclave in its own right.

Firstly we define a set of logical hostnames that refer to the significant components of our system such as databases, file servers etc. Other elements such as webservers are probably independent and not referenced from other parts of the system and thus do not need names. These logical hostnames are what feature in our configuration files and do not need to change from enclave to enclave because we are going to use DNS to map from these logical hostnames to real physical machines.

Thus we want is a separate namespace for hosts in each of these enclaves so as to prevent leakage. To that end we map each namespace onto a separate domain within our DNS setup.

[Note our DNS setup would typically consist of a set of servers that maintain records for our own internal domains and possibly forward other requests for say external web address to other servers.]

Each enclave therefore has:

1. A separate namespace represented as a unique domain
2. A set of services deployed onto physical machines
3. A mapping from logical machine names to physical machine names (or IP addresses)
4. A collection of configuration files all referencing logical machine names

Each domain (namespace) contains the logical to physical mapping of machines for its associated enclave. Each domain can be a separate zone and is thus kept in a separate file read by our DNS master. This allows us to maintain a template file which can be quickly edited to create a new domain (namespace). Thus whenever we wish to create a new enclave we setup a new zone, containing the definition of a new domain which is the namespace for that enclave.

To actually resolve a logical hostname we must ensure that it is concatenated with the domain appropriate to the enclave’s namespace. Before discussing options, note that each machine will be allocated to an enclave and must be configured accordingly which we can exploit to our advantage:

1. Simple configuration – ensure that the application has access to the domain to concatenate. This could be done via command-line argument but better is to source it from a well-known file on the machine which could be setup as part of allocating it to an enclave.
2. Default search domain – any name not fully qualified has the default search domain appended to it. This default is typically part of the resolver configuration of the operating system and again can be setup as part of allocating a machine to an enclave.

Missing from the above is the handling of ports which might change from one enclave to the next. This can be tackled with a similar logical/physical mapping approach but must be based on the use of DNS SRV records rather than simple hostname mappings. The JDK provides little help out of the box for querying these records so something like dnsjava will be required.

Technorati Tags: distributed systems, dns, testing