2.1.1 State of the Art

Standard system development and validation techniques are not well suited for addressing the above challenge. Designing cloud storage systems is hard, as the design must take into account wide-area asynchronous communication, concurrency, and fault tolerance. Experimentation with modifications and extensions of an existing system is often impeded by the lack of a precise description at a suitable level of abstraction and by the need to understand and modify large code bases (if available) to test the new design ideas. Furthermore, test-driven system development [2] – where a suite of tests for the planned features are written before development starts, and is used both to give the developer quick feedback during development and as a set of regression tests when new features are added—has traditionally been considered to be unfeasible for ensuring fault tolerance in complex distributed systems due to the lack of tool support for testing large numbers of different scenarios.

It is also very difficult or impossible to obtain high assurance that the cloud storage system satisfies given correctness and performance requirements using traditional validation methods. Real implementations are costly and error-prone to implement and modify for experimentation purposes. Simulation tool implementations require building an additional artifact that cannot be used for much else. Although system executions and simulations can give an idea of the performance of a design, they cannot give any (quantified) assurance on the performance measures. Furthermore, such implementations cannot verify consistency guarantees: Even if we execute the system and analyze the read/write operations log for consistency violations, this would only cover certain scenarios and cannot guarantee the absence of subtle bugs. In addition, nontrivial fault-tolerant storage systems are too complex for “hand proofs” of key properties based on an informal system description. Even if attempted, such proofs can be error-prone, informal, and usually rely on implicit assumptions.

The inadequacy of current design and verification methods for cloud storage systems in industry has also been pointed out by engineers at Amazon in [3] (see also Section 2.6). For example, they conclude that “the standard verification techniques in industry are necessary but not sufficient. We routinely use deep design reviews, code reviews, static code analysis, stress testing, and fault-injection testing but still find that subtle bugs can hide in complex concurrent fault-tolerant systems.”

2.1.2 Vision: Formal Methods for Cloud Storage Systems

Our vision is to use formal methods to design cloud storage systems and to provide high levels of assurance that their designs satisfy given correctness and performance requirements. In a formally based system design and analysis methodology, a mathematical model S describes the system design at the appropriate level of abstraction. This system specification S should be complemented by a formal property specification P that describes mathematically (and therefore precisely) the requirements that the system S should satisfy. Being a mathematical object, the model S can be subjected to mathematical reasoning (preferably fully automated or at least machine-assisted) to guarantee that the design satisfies the properties P. If the mathematical description S is executable, then it can be immediately simulated; there is no need to generate an extra artifact for testing and verification. An executable model can also be subjected to various kinds of model checking analyses that automatically explore all possible system behaviors from a given initial system configuration. From a system developer's perspective, such model checking can be seen as a powerful debugging and testing method that can automatically find subtle “corner case” bugs and that automatically executes a comprehensive “test suite” for complex fault-tolerant systems. We advocate the use of formal methods throughout the design process to quickly and easily explore many design options and to validate designs as early as possible, since errors are increasingly costly the later in the development process they are discovered. Of course, one can also do a postmortem formal analysis of an existing system by defining a formal model of it in order to analyze the system formally; we show the usefulness of such postmortem analysis in Section 2.2.

Performance is as important as correctness for storage systems. Some formal frameworks provide probabilistic or statistical model checking that can give performance assurances with a given confidence level.

What properties should a formal framework have in order to be suitable for developing and analyzing cloud storage systems in an industrial setting? In Ref. [4], Chris Newcombe of Amazon Web Services, the world's largest cloud computing provider, who has used formal methods during the development of key components of Amazon's cloud computing infrastructure, lists key requirements for formal methods to be used in the development of such cloud computing systems in industry. These requirements can be summarized as follows:

1. Expressive languages and powerful tools that can handle very large and complex distributed systems. Complex distributed systems at different levels of abstraction must be expressible without tedious workarounds of key concepts (e.g., time and different forms of communication). This requirement also includes the ability to express and verify complex liveness properties. In addition to automatic methods that help users diagnose bugs, it is also desirable to be able to machine-check proofs of the most critical parts.
2. The method must be easy to learn, apply, and remember, and its tools must be easy to use. The method should have clean simple syntax and semantics, should avoid esoteric concepts, and should use just a few simple language constructs. The author also recommends against distorting the language to make it more accessible, as the effect would be to obscure what is really going on.
3. A single method should be effective for a wide range of problems, and should quickly give useful results with minimal training and reasonable effort. A single method should be useful for many kinds of problems and systems, including data modeling and concurrent algorithms.
4. Modeling and analyzing performance, since performance is almost as important as correctness in industry.

2.1.3 The Rewriting Logic Framework

Satisfying the above requirements is a tall order. We suggest the use of rewriting logic [5] and its associated Maude tool [6], and their extensions, as a suitable framework for formally specifying and analyzing cloud storage systems.

In rewriting logic, data types are defined by algebraic equational specifications. That is, we declare sorts and function symbols; some function symbols are constructors used to define the values of the data type; the others denote defined functions – functions that are defined in a functional programming style using equations. Transitions are defined by rewrite rules of the form if cond, where t and t′ are terms (possibly containing variables) representing local state patterns, and cond is a condition. Rewriting logic is particularly suitable for specifying distributed systems in an object-oriented way, in which case the states are multisets of objects and messages (traveling between the objects), and where an object o of class C with attributes to , having values to , is represented by a term . A rewrite rule

rl [l] : m(O,w)

        < O : C | a1 : x, a2 : O′, a3 : z >

       =>

        < O : C | a1 : x + w, a2 : O′, a3 : z >

        m′(O′,x) .

then defines a family of transitions in which a message m, with parameters O and w, is read and consumed by an object O of class C, the attribute al of the object O is changed to x + w, and a new message m′(O′,x) is generated.

Maude [6] is a specification language and high-performance simulation and model checking tool for rewriting logic. Simulations – which simulate single runs of the system – provide a first quick initial feedback of the design. Maude reachability analysis – which checks whether a certain (un)desired state pattern can be reached from the initial state – and linear temporal logic ( LTL) model checking – which checks whether all possible behaviors from the initial state satisfy a given LTL formula – can be used to analyze all possible behaviors from a given initial configuration.

The Maude tool ecosystem also includes Real-Time Maude [7], which extends Maude to real-time systems, and probabilistic rewrite theories [8], a specification formalism for specifying distributed systems with probabilistic features. A fully probabilistic subset of such theories can be subjected to statistical model checking analysis using the PVeStA tool [9]. Statistical model checking [10] performs randomized simulations until a probabilistic query can be answered (or the value of an expression be estimated) with the desired statistical confidence.

Rewriting logic and Maude address the above requirements as follows:

1. Rewriting logic is an expressive logic in which a wide range of complex concurrent systems, with different forms of communication and at various levels of abstractions, can be modeled in a natural way. In addition, its real-time extension supports the modeling of real-time systems. The Maude tools have been applied to a range of industrial and state-of-the-art academic systems [11,12]. Complex system requirements, including safety and liveness properties, can be specified in Maude using linear temporal logic, which seems to be the most intuitive and easy-to-understand advanced property specification language for system designers [13]. We can also define functions on states to express nontrivial reachability properties.
2. Equations and rewrite rules: These intuitive notions are all that have to be learned. In addition, object-oriented programming is a well-known programming paradigm, which means that Maude's simple model of concurrent objects should be attractive to designers. We have experienced in other projects that system developers find object-oriented Maude specifications easier to read and understand than their own use case descriptions [14], and that students with no previous formal methods background can easily model and analyze complex distributed systems in Maude [15]. The Maude tools provide automatic (push-button) reachability and temporal logic model checking analysis, and simulation for rapid prototyping.
3. As mentioned, this simple and intuitive formalism has been applied to a wide range of systems, and to all aspects of those systems. For example, data types are modeled as equational specification and dynamic behavior is modeled by rewrite rules. Maude simulations and model checking are easy to use and provide useful feedback automatically: Maude's search and LTL model checking provides a counterexample trace if the desired property does not hold.
4. We show in Ref. [16] that randomized Real-Time Maude simulations (of wireless sensor networks) can give performance estimates as good as those of domain-specific simulation tools. More importantly, we can analyze performance measures and provide performance estimations with given confidence levels using probabilistic rewrite theories and statistical model checking; e.g., “I can claim with 90% confidence that at least 75% of the transactions satisfy the property P.” For performance estimation for cloud storage systems, see Sections 2.2, 2.3, and 2.5.

To summarize, a formal executable specification in Maude or one of its extensions allows us to define a single artifact that is, simultaneously, a mathematically precise high-level description of the system design and an executable system model that can be used for rapid prototyping, extensive testing, correctness analysis, and performance estimation.

2.1.4 Summary: Using Formal Methods on Cloud Storage Systems

In this chapter, we summarize some of the work performed at the Assured Cloud Computing Center at the University of Illinois at Urbana-Champaign using Maude and its extensions to formally specify and analyze the correctness and performance of several important industrial cloud storage systems and a state-of-the-art academic one. In particular, we describe the following contributions:

1. Apache Cassandra [17] is a popular open-source industrial key-value data store that only guarantees eventual consistency. We were interested in (i) evaluating a proposed variation of Cassandra, and (ii) analyzing under what circumstances – and how often in practice – Cassandra also provides stronger consistency guarantees, such as read-your-writes or strong consistency. After studying Cassandra's 345,000 lines of code, we first developed a 1000-line Maude specification that captured the main design choices. Standard model checking allowed us to analyze under what conditions Cassandra guarantees strong consistency. By modifying a single function in our Maude model, we obtained a model of our proposed optimization. We subjected both of our models to statistical model checking using PVeStA; this analysis indicated that the proposed optimization did not improve Cassandra's performance. But how reliable are such formal performance estimates? To investigate this question, we modified the Cassandra code to obtain an implementation of the alternative design, and executed both the original Cassandra code and the new system on representative workloads. These experiments showed that PVeStA statistical model checking provides reliable performance estimates. To the best of our knowledge, this was the first time that for key-value stores, model checking results were checked against a real system deployment, especially on performance-related metrics.
2. Megastore [18] is a key part of Google's celebrated cloud infrastructure. Megastore's trade-off between consistency and efficiency is to guarantee consistency only for transactions that access a single entity group. It is obviously interesting to study such a successful cloud storage system. Furthermore, one of us had an idea on how to extend Megastore so that it would also guarantee strong consistency for certain transactions accessing multiple entity groups without sacrificing performance. The first challenge was to develop a detailed formal model of Megastore from the short high-level description in Ref. [18]. We used Maude simulation and model checking throughout the formalization of this complex system until we obtained a model that satisfied all desired properties. This model also provided the first reasonable detailed public description of Megastore. We then developed a formal model of our extension, and estimated the performance of both systems using randomized simulations in Real-Time Maude; these simulations indicated that Megastore and our extension had about the same performance. (Note that such ad hoc randomized simulations do not give a precise level of confidence in the performance estimates.)
3. RAMP [19] is a state-of-the-art academic partitioned data store that provides efficient lightweight transactions that guarantee the simple “read atomicity” consistency property. Reference [19] gives hand proofs of correctness properties and proposes a number of variations of RAMP without giving details. We used Maude to (i) check whether RAMP indeed satisfies the guaranteed properties, and (ii) develop detailed specifications of the different variations of RAMP and check which properties they satisfy.
4. ZooKeeper [20] is a fault-tolerant distributed key/value data store that provides reliable distributed coordination. In Ref. [21] we investigate whether a useful group key management service can be built using ZooKeeper. PVeStA statistical model checking showed that such a ZooKeeper-based service handles faults better than a traditional centralized group key management service, and that it scales to a large number of clients while maintaining low latencies.

To the best of our knowledge, the above-mentioned work at the Assured Cloud Computing Center represents the first published papers on the use of formal methods to model and analyze such a wide swathe of industrial cloud storage systems. Our results are encouraging, but the question arises: Is the use of formal methods feasible in an industrial setting? The recent paper [3] from Amazon tells a story very similar to ours, and formal methods are now a key ingredient in the system development process at Amazon. The Amazon experience is summarized in Section 2.6, which also discusses the formal framework used at Amazon.

The rest of this chapter is organized as follows: Sections 2.2–2.5 summarize our work on Cassandra, Megastore, RAMP, and ZooKeeper, respectively, while Section 2.6 gives an overview of the use of formal methods at Amazon. Section 2.7 discusses related work, and Section 2.8 gives some concluding remarks.

2.3.1 Specifying Megastore

Our specification of Megastore [26] is the first publicly available formalization and reasonably detailed description of Megastore. It contains 56 rewrite rules, of which 37 deal with fault tolerance features. An important point is that even if we had access to Megastore's code base, understanding and extending it would likely have been much more time-consuming than developing our own 56-rule description and simulation model in Real-Time Maude.

To show an example of the specification style, the following shows one (of the smallest) of the 56 rewrite rules:

rl [bufferWriteOperation]  :

  < SID : Site | localTransactions : LOCALTRANS

   < TID : Transaction | operations : write(KEY, VAL) :: OL, writes : WRITEOPS,

     status : idle > >

  =>

  < SID : Site | localTransactions : LOCALTRANS

   < TID : Transaction | operations : OL, writes : WRITEOPS :: write(KEY, VAL) > >.

In this rule, the Site object SID has a number of transaction objects to execute in its localTransactions attribute; one of them is the transaction TID (the variable LOCALTRANS ranges over multisets of objects, and therefore captures all the other objects in the site's transaction set). The next operation that the transaction TID should execute is the write operation write(KEY, VAL), which represents writing the value VAL to the key KEY. The effect of applying this rewrite rule is that the write operation is removed from the transaction's list of operations to perform and is added to its write set writes.

The second rewrite rule we show concerns the validation phase. In Megastore, a replicated transaction log is maintained for each entity group. When a transaction t is ready to commit, its coordinating site s prepares a new log entry for each entity group written by t, does some leader election part of Paxos, and (if the leader election phase is successful) multicasts the new proposed log entry to the other replicating sites. Each recipient of this message must then verify that it has not already granted an accept for the new log position. If so, the recipient replies with an accept message to the originating site.

The following rule shows the part when the replicating site THIS receives the multicasted message acceptAllReq with the proposed new log entry (TID′ LP SID OL) for entity group EG from the coordinator THIS. The site THIS verifies that it has not already granted an accept for that log position. (Since messages could be delayed for a long time, it checks both the transaction log and received proposals). If there are no such conflicts, the site responds with an accept message acceptAllRsp(…) and stores its accept in its proposals attribute. The record (TID′ LP SID OL) represents the candidate log entry, which contains the transaction identifier TID′, the log position LP, the proposed leader site SID, and the list of update operations OL:

crl [rcvAcceptAllReq]   :

   (msg acceptAllReq(TID, EG,  (TID′ LP SID OL), PROPNUM) from SENDER to THIS)

   < THIS : Site |

   entityGroups : EGROUPS

    < EG : EntityGroup | proposals : PROPSET, transactionLog : LEL > >

   =>

   < THIS : Site |

   entityGroups :   EGROUPS

   < EG : EntityGroup |

   proposals : accepted(SENDER,  (TID′ LP SID OL), PROPNUM);   removeProposal(LP, PROPSET) > >

    dly(msg acceptAllRsp(TID, EG, LP, PROPNUM) from THIS to SENDER, T)

    if not (containsLPos(LP, LEL) or hasAcceptedForPosition(LP, PROPSET)) / T;  TS := possibleMessageDelays(THIS, SENDER)  .

Our model assumes that possibleMessageDelays(s₁, s₂) gives the set of possible messaging delays between sites s₁ and s₂, where the union operator _;_ is associative and commutative. The matching equation

T;  TS := possibleMessageDelays(THIS, SENDER)

dly(msg acceptAllRsp(TID, EG, LP, PROPNUM) from THIS to SENDER, T).

2.3.2 Analyzing Megastore

We wanted to analyze both the correctness and the performance of our model of Megastore throughout its development, to catch functional errors and performance bottlenecks quickly. For our analysis, we generated the following two additional models from the main model already described:

1. Our model of Megastore is a real-time model. However, this means that any exhaustive model checking analysis of our model only analyzes those behaviors that are possible within the given timing parameters (for messaging delays, etc.). To exhaustively analyze all possible system behaviors irrespective of particular timing parameters, we generated an untimed model from our real-time model by just ignoring messaging delays in our specification. For example, in this specification, the above rewrite rule becomes (where the parts represented by “…”are as before):

   crl [rcvAcceptAllReq]  :

        (msg acceptAllReq(...) from SENDER to THIS)

        < THIS : Site | entityGroups : EGROUPS  < EG : EntityGroup | ... > >

       =>

       < THIS : Site | entityGroups :   EGROUPS  < EG : EntityGroup | ... > >

        (msg acceptAllRsp(TID, EG, LP, PROPNUM) from THIS to SENDER)

       if not (containsLPos(LP, LEL) or hasAcceptedForPosition(LP, PROPSET)) .

2. For performance estimation purposes, we also defined a real-time model in which certain parameters, such as the messaging delays between two nodes, are selected probabilistically according to a given probability distribution. For example, we used the following probability distribution for the network delays (in milliseconds):

   	30%	30%	30%	10%
   London ↔ Paris	10	15	20	50
   London ↔ New York	30	35	40	100
   Paris ↔ New York	30	35	40	100

   An important difference between our Megastore work and the Cassandra effort described in Section 2.2 is that in the Cassandra work we used (fully) probabilistic rewrite theories and the dedicated statistical model checker PVeStA for the probabilistic analysis. In contrast, our Megastore work stayed within Real-Time Maude and simulated the selection of a value from a distribution by using an ever-changing seed and Maude's built-in function random. In this “lightweight” probabilistic real-time model, we maintain the value n of the seed in an object < seed: Seed | value: n > in the state, and the above rewrite rule is transformed to the following:

   crl [rcvAcceptAllReq]  :

          (msg acceptAllReq(...)  from SENDER to THIS)

          < THIS : Site | entityGroups : EGROUPS  < EG : EntityGroup | ... > >

          < seed : Seed / value : N > =>

        < THIS : Site | entityGroups :   EGROUPS  < EG : EntityGroup | ... > >

          < seed : Seed / value : N + 1 >

          dly(msg acceptAllRsp(...) from THIS to SENDER,

          selectFromDistribution(delayDistribution(THIS, SENDER), random(N)))

          if not (containsLPos(LP, LEL) or hasAcceptedForPosition(LP, PROPSET)) .

   where selectFromDistribution “picks” the appropriate delay value from the distribution delayDistribution(THIS, SENDER) using the random number random(N). While such lightweight probabilistic performance analysis using Real-Time Maude has previously been shown to give performance estimates on par with those of dedicated domain-specific simulation tools, for example, for wireless sensor networks [16], such analysis cannot provide a (quantified) level of confidence in the estimates; for that we need statistical model checking.

      Maude> (mc initMegastore /=u

      ([] isSerializable)

      /  (<> [] (allTransFinished / entityGroupsEqualOrInvalid

                    / transLogsEqualOrInvalid)).)

2.5.1 Zookeeper Background

ZooKeeper is used by many distributed applications, such as Apache Hadoop MapReduce and Apache HBase, and by many companies, including Yahoo and Zynga in their infrastructures. From a bird's eye view, the ZooKeeper system consists of two kinds of entities: servers and clients. All of the servers together form a distributed and fault-tolerant key/value store that the clients may read data from or write data to. In ZooKeeper terminology, each key/value pair is called a znode, and these znodes are then organized into a tree structure similar to that of a standard filesystem. In order to achieve fault tolerance, ZooKeeper requires that a majority of servers acknowledge and log each write to disk before it is committed. Since ZooKeeper requires a majority of servers to be alive and aware of each other in order to operate, updates will not be lost. A detailed description of ZooKeeper design and features can be found in Ref. [20] and in ZooKeeper documentation.⁴

ZooKeeper provides a set of guarantees that are useful for key management applications. Updates to ZooKeeper state are atomic, and once committed they will persist until they are overwritten. Thus, a client will have a consistent view of the system regardless of which server it connects to and is guaranteed to be up-to-date within a certain time-bound. For our purposes, this means updates, once committed, will not be lost. Furthermore, ZooKeeper provides an event-driven notification mechanism through watches. A client sets a watch on a znode that will be notified whenever the znode is changed or deleted. Watches can enable us, for example, to easily propagate key change events to interested clients.

2.5.2 System Design

In our design, if a user wishes to join/leave a group, she will contact the group controller who will: (a) perform the necessary authentication and authorization checks; (b) if the user is authorized, add/remove the user to/from the group; (c) update the group membership state in the ZooKeeper service; (d) generate and update the group key in the ZooKeeper service; and (e) provide any new users with the updated group key and necessary information to connect to ZooKeeper and obtain future key updates.

The system state is stored in ZooKeeper as follows: Each secure group and authorized user is assigned a unique znode storing an encrypted key. Whenever a user joins a group, the znode corresponding to that user is added as a child of the znode representing the group. The znode corresponding to the group also stores the current group key encrypted by the KEK. During periodic group key updates, the old group key is overwritten by the new group key (encrypted by the same KEK). However, when an authorized user joins or leaves a group, the group controller generates a new KEK and distributes it using the client's pairwise keys. Specifically, the group controller updates the value stored at each user's assigned znode with the new KEK encrypted by the client's pairwise key. Since each group member sets a watch on the znode corresponding to itself and its group, it will be notified whenever the group key or the KEK changes.

Thus, the group controller uses the ZooKeeper service to maintain group information and to distribute and update cryptographic keys. Since the group controller's operational state is already saved within the ZooKeeper service (e.g., group member znodes it has updated, last key update time, etc.), if the controller were to fail a backup controller could take over with minimal downtime.

2.5.3 Maude Model

The distributed system state in our model is a multiset structure, populated by objects, messages, and the scheduler. Messages are state fragments passed from one object to another marked with a timestamp, an address, and a payload. The scheduler's purpose is to deliver messages to an object at the time indicated by the message's timestamp and to provide a total ordering of messages in case some have identical timestamps. ZooKeeper system state is modeled by three classes of objects: the ZooKeeper service, servers, and clients. The group key management system also consists of three different classes of objects: the group key service, managers, and clients. All these objects are designed to operate according to the system design discussed above, but we must be careful to clarify a few details.

To simplify the model, we abstracted away many complex – but externally invisible – details of ZooKeeper. We also say a few words about failure modes. We assume that both ZooKeeper servers and group key managers may fail. When a ZooKeeper server fails, it will no longer respond to any messages. After a variable repair timeout, the server will come back online. When any connected clients discover the failure, they will attempt to migrate to another randomly chosen server. Similarly, when a key manager fails, any client waiting on that key manager will time out and try again. A key manager buffers clients' requests and answers them in order. Before answering each request, it saves information about the requested operation to ZooKeeper. If a manager dies, the succeeding manager loads the stored state from ZooKeeper and completes any pending operations before responding to new requests. In order that our model would accurately reflect the performance of a real ZooKeeper cluster, we chose parameters that agree with the data gathered in Ref. [20]. In particular, we set the total latency of a read or write request to ZooKeeper to under 2ms on average (which corresponds to a ZooKeeper system under full load as in Ref. [20]). Note that this average time only occurs in practice if there is no server failure; our model permits servers to fail. We also assume that all of the ZooKeeper servers are located in the same data center, which is the recommended way to deploy a ZooKeeper cluster.

Here we show an example Maude rewrite rule from our specification:

crl [ZKSERVER-PROCESS]   :

     < A : ZKServer | leader: L, live: true, txid: Z, store: S,

         clients: C, requests: R, updates: U, AS >

     {T, A <- process}

    =>

     < A : ZKServer | leader: L, live: true, txid: Z′, store: S′,

         clients: C, requests: null, updates: null, AS >

       M M′

     if (Z′,S′,M)  := commit(C,U,S) / M′  := process(L,R,S′)  .

This rule, specified in Maude, illustrates how a live ZooKeeper server object processes a batch of updates (service internal messages) and requests from clients during a processing cycle. The condition of this rule invokes two auxiliary functions: commit and process, which repeatedly commit key updates and process requests from connected clients. When this rule completes, the server will have a new transaction ID and key/value store, an empty update and request list, and a set of messages M M′ to be sent, where M are notifications to interested clients that keys were updated and M′ are client requests forwarded onto the ZooKeeper leader.

2.5.4 Analysis and Discussion

Our analysis consisted of two experiments. Both were run hundreds of times via PVeStA and average results were collected. The first experiment was designed to test whether saving snapshots of the group key manager's state in the ZooKeeper store could increase the overall reliability of the system. In the experiment, we set the failure rate for the key manager to 50% per experiment, the time to failover from one server to another to 5 s and the experiment duration to 50 s. We then compared the average key manager availability (i.e., the time it is available to distribute keys to clients) between a single key manager and two key managers where they share a common state saved in the ZooKeeper store. In former case, the average availability was 32.3 seconds, whereas in the latter case it was 42.31 s. This represents an availability improvement from 65 to 85%. Of course, we expect a system with replicated state to be more reliable as there is no longer a single point of failure.

Our second experiment was designed to examine whether using ZooKeeper to distribute shared keys is efficient and scalable enough for real-world use. The experiment measured the variations in: (a) the percentage of keys successfully received by group members, and (b) the key distribution latency, as increasing numbers of clients joined a group per second. The percentage of keys successfully received is defined as the total number of keys expected to be received by any client over its lifetime versus the amount actually received; and distribution latency is the average time measured from when a key is generated by the key manager until that key is received by a client. We sampled these parameters at client join rates of once every 4, 2, 1, 0.5, 0.25, and 0.125 s. We kept the same duration as in the first experiment, but picked failure probabilities such that system will have 99.99% availability and specified link latency between the ZooKeeper service and ZooKeeper clients to vary according to a uniform distribution on the interval from 0.05 to 0.25 s.

We present our results in Figures 2.3 and 2.4, where the blue curve corresponds to our initial experiments while the red curve corresponds to a slightly modified model where we added a 2 second wait time between key updates from the key manager. While our initial experiments show that naively using ZooKeeper as a key distribution agent works well, at high client join rates, the key reception rate seems to level out around 96%. This occurs because ZooKeeper can apply key updates internally more quickly than clients can download them; after all the ZooKeeper servers enjoy a high-speed intra-cloud connection. By adding extra latency between key updates, the ZooKeeper servers are forced to wait enough time for the correct keys to propagate to clients. As shown in Figure 2.3, this change achieves a 99% key reception in all cases. On the other hand, key distribution latency remained relatively constant, at around half a second, regardless of the join rate because ZooKeeper can distribute keys at a much higher rate than a key manager can update them [20]. Of course, the artificial latency added in the second round of experiments has a cost; it increases the time required for a client to join or leave the group by the additional wait time.

In essence, our analysis confirmed that a scalable and fault-tolerant key management service can indeed be built using ZooKeeper, settling various doubts raised about the effectiveness of ZooKeeper for key management by an earlier, but considerably less-detailed, model and analysis [31]. This result is not particularly surprising, especially considering that many man-hours would be needed to optimize an actual system. More interestingly, the analysis also showed that system designs may suffer from performance bottlenecks not readily apparent in the original description – highlighting the power of formal modeling and analysis as a method to explore the design space.

2.6.1 Use of Formal Methods

The AWS developers used the approach that we advocate in this chapter: the use of an intuitive, expressive, and executable (sort of) specification language together with model checking for automatic push-button exploration of all possible behaviors.

More precisely, they used Leslie Lamport's specification formalism TLA+ [33] and its associated model checker TLC. The language is based on set theory and has the usual logic and temporal logic operators. In TLA+, a transition T is defined as a logical axiom relating the “current value” of a variable x with the next value x′ of the variable. For example, a transition T that increases the value of x by 1 and the value of sum with x can be defined as . The model checker TLC can analyze invariants and generate random behaviors.

Formal methods were applied on different components of S3, DynamoDB, and other components, and a number of subtle bugs were found.

2.6.2 Outcomes and Experiences

The experience of the AWS engineers was remarkably similar to that generally advocated by the formal methods community. The quotations below are all from Ref. [3].

Model checking finds “corner case” bugs that would be hard to find with standard industrial methods:

• “We have found that standard verification techniques in industry are necessary but not sufficient. We routinely use deep design reviews, static code analysis, stress testing, and fault-injection testing but still find that subtle bugs can hide in complex fault-tolerant systems.”
• “T.R. learned TLA+ and wrote a detailed specification of [components of DynamoDB] in a couple of weeks. […] the model checker found a bug that could lead to losing data if a particular sequence of failures and recovery steps would be interleaved with other processing. This was a very subtle bug; the shortest error trace exhibiting the bug included 35 high-level steps. […] The bug had passed unnoticed through extensive design reviews, code reviews, and testing.”

A formal specification is a valuable precise description of an algorithm:

• “There are at least two major benefits to writing precise design: the author is forced to think more clearly, helping eliminating ‘hand waving,’ and tools can be applied to check for errors in the design, even while it is being written. In contrast, conventional design documents consist of prose, static diagrams, and perhaps pseudo-code in an ad hoc untestable language.”
• “Talk and design documents can be ambiguous or incomplete, and the executable code is much too large to absorb quickly and might not precisely reflect the intended design. In contrast, a formal specification is precise, short, and can be explored and experimented on with tools.”
• “We had been able to capture the essence of a design in a few hundred lines of precise description.”

Formal methods are surprisingly feasible for mainstream software development and give good return on investment:

• “In industry, formal methods have a reputation for requiring a huge amount of training and effort to verify a tiny piece of relatively straightforward code, so the return on investment is justified only in safety-critical domains (such as medical systems and avionics). Our experience with TLA+ shows this perception to be wrong. […] Amazon engineers have used TLA+ on 10 large complex real-world systems. In each, TLA+ has added significant value. […] Amazon now has seven teams using TLA+, with encouragement from senior management and technical leadership. Engineers from entry level to principal have been able to learn TLA+ from scratch and get useful results in two to three weeks.”
• “Using TLA+ in place of traditional proof writing would thus likely have improved time to market, in addition to achieving greater confidence in the system's correctness.”

Quick and easy to experiment with different design choices:

• “We have been able to make innovative performance optimizations […] we would not have dared to do without having model checked those changes. A precise, testable description of a system becomes a what-if tool for designs.”

The paper's conclusions include the following:

“Formal methods are a big success at AWS, helping us to prevent subtle but serious bugs from reaching production, bugs we would not have found using other techniques. They have helped us devise aggressive optimizations to complex algorithms without sacrificing quality. […] seven Amazon teams have used TLA+, all finding value in doing so […] Using TLA+ will improve both time-to-market and quality of our systems. Executive management actively encourages teams to write TLA+ specs for new features and other significant design changes. In annual planning, managers now allocate engineering time to TLA+.”

2.6.3 Limitations

The authors point out that there are two main classes of problems with large distributed systems: bugs and performance degradation when some components slow down, leading to unacceptable response times from a user's perspective. While TLA+ was effective to find bugs, it had one significant limitation in that it was not, or could not be, used to analyze performance degradation.

Why TLA + ? We have advocated using the Maude framework for the specification and analysis of cloud computing storage systems. What are the differences between Maude and its toolset and TLA+ and its model checker? And why did the Amazon engineers use TLA+ instead of, for example, Maude?

On the specification side, Maude supports hierarchical system states, object-oriented specifications, with dynamic object creation deletion, subclasses, and so on, as well as the ability to specify any computable data type as an equational specification. These features do not seem to be supported by TLA+. Maude also has a clear separation between the system specification and the property specification, whereas TLA+ uses the same language for both parts. Hence, the fact that a system S satisfies its property P can be written in TLA+ as the logical implication S → P.

Perhaps the most important difference is that real-time systems can be modeled and analyzed in Real-Time Maude, and that probabilistic rewrite theories can be statistically model checked using PVeStA, whereas TLA+ seems to lack support for the specification and analysis of real-time and probabilistic systems. (Lamport argues that special treatment of real-time systems is not needed: just add a system variable clock that denotes the current time [34].) The lack of support for real-time and probabilistic analysis probably explains why the TLA+ engineers could only use TLA+ and TLC for correctness analysis but not for performance analysis, whereas we have shown that the Maude framework can be used for both aspects.

So, why did the Amazon engineers choose TLA+? The main reason seems to be that TLA+ was developed by one of the most prominent researcher in distributed systems, Leslie Lamport, whose algorithms (like the many versions of Paxos) are key components in today's cloud computing systems:

“C.N. eventually stumbled on a language […] when he found a TLA+ specification in the appendix of a paper on a canonical algorithm in our problem domain—the Paxos consensus algorithm. The fact that TLA+ was created by the designer of such a widely used algorithm gave us confidence that TLA+ would work for real-world systems.”

Indeed, it seems that they did not explore too many formal frameworks:

“When we found [that] TLA+ met those requirements, we stopped evaluating methods.”

2.8.1 The Future

Much work remains ahead. To begin with, one current limitation is that explicit state model checking explores the behaviors starting from a single initial system configuration, and hence – since this can only be done for a finite number of states – it cannot be used to verify that an algorithm is correct for all possible initial system configurations, which can be infinite.

A first approach to obtain a more complete coverage is illustrated by Refs [26,46], where we have extended coverage by model checking all initial system configurations up to n operations, m replicas, and so on. However, naïvely generating initial states can yield a large number of symmetric initial states; greater coverage and symmetry reduction methods should therefore be explored. A second, more powerful approach is to use symbolic model checking techniques, where a possibly infinite number of initial states is described symbolically by formulas in a theory whose satisfiability is decidable by an SMT solver. In the Maude context, this form of symbolic model checking is supported by rewriting modulo SMT [47], which has already been applied to various distributed real-time systems, and by narrowing-based symbolic model checking [48,49]. An obvious next step is to verify properties of cloud storage systems for possibly infinite sets of initial states using this kind of symbolic model checking. A third approach is to take to heart the complementary nature of model checking and theorem proving, so that these methods do help each other when used in combination, as explained below.

In the approach we have presented, design exploration based on formal executable specification comes first. Once a promising design has been identified, fairly exhaustive model checking debugging and verification of such a design can be carried out using: (i) LTL explicit-state model checking, (ii) statistical model checking of quantitative properties, and (iii) symbolic model checking from possibly infinite sets of initial configurations for greater assurance. Since all these methods are automatic, this is relatively easy to do based on the system's formal executable specification with rewrite rules. Only after we are fairly sure that we have obtained a good design and have eliminated many subtle errors by these automatic methods does it become cost-effective to attempt a more labor-intensive deductive verification through theorem proving of properties not yet fully verified. Indeed, some properties may already have been fully verified in an automatic way by symbolic model checking, so that only some additional properties not expressible within the symbolic model checking framework need to be verified deductively. Two compelling reasons for such deductive verification are (i) the safety-critical nature of a system or a system component, and (ii) the high reusability of a component or algorithm, for example, Paxos or Raft, so that the deductive verification effort becomes amortized over many uses.

In the context of rewriting logic verification, a simple logic to deductively verify reachability properties of a distributed system specified as a rewrite theory has been recently developed, namely, constructor-based reachability logic [50], which itself extends and applies to rewrite theories of the original reachability logic in Ref. [51]. In fact, a number of distributed algorithms have already been proved correct in Ref. [50]; and we plan to soon extend and apply the Maude reachability logic tool to verify specific properties of cloud storage systems.

Of course, high reusability of designs is not just a good thing for amortizing verification efforts: It is a good thing in all respects. By breaking distributed system designs into modular, well-understood, and highly reusable components that come with strong correctness guarantees, as advocated by the notion of formal pattern in Ref. [52], unprecedented levels of assurance and of software quality could be achieved for cloud storage systems in the near future. It is for this reason that, from the beginning of our research on the application of formal methods to gain high assurance for cloud-based systems, we have tried to understand cloud system designs as suitable compositions of more basic components providing key functionality in the form of generic distributed algorithms. This is still work in progress. Our medium-term goal is to develop a library of generic components formally specified and verified in Maude as formal patterns in the above sense.

Out of these generic components, different system designs for key-value stores and for cloud storage systems supporting transactions could then be naturally and easily obtained as suitable compositions.