Giannis Tzouros
email: tzouros@aueb.gr

Vana Kalogeraki

Fed-DIC: Diagonally Interleaved Coding in a Federated Cloud Environment

Coping with failures in modern distributed storage systems that handle massive volumes of heterogeneous and potentially rapidly changing data, has become a very important challenge. A common practice is to utilize fault tolerance methods like Replication and Erasure Coding for maximizing data availability. However, while erasure codes provide better fault tolerance compared to replication with a more affordable storage overhead, they frequently suffer from high reconstruction cost as they require to access all available nodes when a data block needs to be repaired, and also can repair up to a limited number of unavailable data blocks, depending on the number of the code's parity block capabilities. Furthermore, storing and placing the encoded data in the federated storage system also remains a challenge. In this paper we present Fed-DIC, a framework which combines Diagonally Interleaved Coding on client devices at the edge of the network with organized storage of encoded data in a federated cloud system comprised of multiple independent storage clusters. The erasure coding operations are performed on client devices at the edge while they interact with the federated cloud to store the encoded data. We describe how our solution integrates the functionality of federated clouds alongside erasure coding implemented on edge devices for maximizing data availability and we evaluate the working and benefits of our approach in terms of read access cost, data availability, storage overhead, load balancing and network bandwidth rate compared to popular Replication and Erasure Coding schemes.

Introduction

In recent years, the management and preservation of big data has become a vital challenge in distributed storage systems. Failures, unreliable nodes and components are inevitable and such failures can lead to permanent data loss and overall system slowdowns. To guarantee availability, distributed storage systems typically rely on two fault tolerance methods: (1) Replication, where multiple copies of the data are made, and (2) Erasure Coding, where data is stored in the form of smaller data blocks which are distributed across a set of different storage nodes.

Replication based algorithms as those utilized in Amazon Dynamo [START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF], Google File System (GFS) [START_REF] Ghemawat | The Google file system[END_REF][START_REF] Wang | Formalizing Google File System[END_REF], Hadoop Distributed File System (HDFS) [START_REF] Shvachko | The hadoop distributed file system[END_REF][START_REF] Karun | A review on Hadoop-HDFS infrastructure extensions[END_REF] are widely utilized. These can help tolerate a high permanent failure rate as they provide the simplest form of redundancy by creating replicas from which systems can retrieve the lost data blocks, but cannot easily cope with bursts of failures. Furthermore, replication introduces a massive storage overhead as the size of the created replicas is equal to the size of their original data e.g. 3-way replication occupies 3 times the volume of the original data block in order to provide fault tolerance.

On the other hand, Erasure Coding [START_REF]Erasure coding vs. Replication: A quantitative comparison[END_REF] can provide higher redundancy while also offering a significant improvement in storage overhead compared to replication. For example, a 3-way replication creates 3 replicas of a data block and causes a 3x storage overhead for providing fault tolerance, while an erasure code can provide the same services for half the storage overhead or even lower by creating smaller parity blocks that can retrieve lost data more efficiently than full-sized replicas. Thus, Erasure codes are more storage affordable than replication but their reliability is limited to the number of parity blocks for repairing erasures. For example, an erasure code that creates 3 parity chunks cannot fix a data block with 4 or more unavailable or lost chunks.

Yet the most critical challenge with erasure coding is that it suffers from high reconstruction cost as it needs to access multiple blocks stored across different sets of storage nodes or racks (groups of nodes inside a distributed system) in order to retrieve lost data [START_REF] Li | D 3 : Deterministic Data Distribution for Efficient Data Reconstruction in Erasure-Coded Distributed Storage Systems[END_REF], leading to high read access and network bandwidth latency. The majority of the distributed file systems deploy random block placement [START_REF] Ambade | Hadoop Block Placement Policy For Different File Formats[END_REF] and one block per rack policies [START_REF] Sathiamoorthy | Xoring elephants: Novel erasure codes for big data[END_REF][START_REF] Muralidhar | f4: Facebook's Warm {BLOB} Storage System[END_REF] to achieve optimized reliability and load balancing for stored encoded data. However, storing data across multiple nodes and/or racks can lead to higher read and network access costs among nodes and racks during the repairing processes. For example, in the worst case, repairing a corrupted or unavailable block in a node may require traversing all nodes across different racks, causing a heavy amount of data traffic among nodes and racks. Also, in a typical cross-rack storage, the user does not have any control over the placement of the data blocks across different racks, limiting the ability of the system to tolerate a higher average failure rate.

To reduce the cost of accessing multiple nodes or racks, file systems can keep metadata records regarding the topology of the encoded data codewords (groups that contain original data blocks alongside their parity blocks) in private nodes. However, the placement of the metadata files among the system's nodes is also challenging. For example, storing a codeword in a small group of nodes while keeping metadata about the data blocks scattered throughout the public clouds instead of a specific storage node [START_REF] Dobre | Hybris: Robust hybrid cloud storage[END_REF], will also require to traverse all nodes at worst in order to recover any failed data inside the codeword. This problem leads to high cross-node read and network access costs, despite the use of metadata.

In this paper we propose Fed-DIC (Federated cloud Diagonally Interleaved Coding), a novel compression framework deployed on an edge-cloud infrastructure where client devices perform the coding operation and they interact with the federated cloud to store the encoded data. Fed-DIC's compression approach is based on diagonal interleaved erasure coding that offers improved data availability while reducing read access costs in a federated cloud environment. It employs a variation of diagonally interleaved codes on streaming data organized as a grid of input records. Specifically, the grid content is interleaved into groups that diagonally span across the grid, and then the interleaved groups of data are encoded using a simple Reed-Solomon (RS) erasure code. Next, our framework organizes the encoded data into batches based on the number of clusters in the federated cloud and places each batch to a different cluster in the cloud, while keeping a metadata index of the locations of each stored data stream. The benefit is that Fed-DIC will only access the cluster with the requested data records and retrieve the correspondent diagonals, enabling the system to efficiently extract the corresponding records.

Fed-DIC has multiple benefits: it maximizes the availability of the encoded data by ordering input data into smaller groups, based on diagonally interleaved coding, and encoding each group using the erasure coding technique. Furthermore, it supports efficient archival and balances the load by storing each version of the streaming data array in a rotational basis among the storage nodes, e.g. if we have an infrastructure with 3 file clusters, for the first version of the data array, the first batch of diagonals is stored on the first node cluster, the second batch on the second node cluster and the third batch on the third cluster. For the second version of the array, the first batch of diagonals is stored on the second cluster, the second batch on the third cluster and the third batch on the first cluster and so on. We present an approach how multiple storage usage can optimize read access costs while keeping data availability and low bandwidth cost for retrieving data by utilizing multiple storage clusters in the same cloud environment instead of storing data in a single cluster. We illustrate the effectiveness of our approach with an extended experimental evaluation in terms of read access cost, data availability, storage overhead, load balancing and network bandwidth rate compared to popular Replication and Erasure Coding schemes.

Background

In this section we provide some background material regarding the technologies that we utilize at Fed-DIC: the Federated Cloud environment, Erasure Coding and Diagonally Interleaved Coding.

Federated Storage Systems

Many large-scale distributed computing organizations that need to store and maintain continuous amounts of data deploy distributed storage systems, such as HDFS [START_REF] Shvachko | The hadoop distributed file system[END_REF][START_REF] Karun | A review on Hadoop-HDFS infrastructure extensions[END_REF], GFS [START_REF] Ghemawat | The Google file system[END_REF][START_REF] Wang | Formalizing Google File System[END_REF] (which were mentioned above), Ceph [START_REF] Weil | Ceph: A scalable, high-performance distributed file system[END_REF], Microsoft Azure [START_REF] Chappell | Introducing Windows Azure[END_REF][START_REF] Calder | Windows Azure Storage: a highly available cloud storage service with strong consistency[END_REF], Amazon S3 [START_REF] Amazon | Amazon web services[END_REF], Alluxio [START_REF] Chang | The performance analysis of cache architecture based on Alluxio over virtualized infrastructure[END_REF] etc., which comprise multiple nodes, often organized into groups called racks. Currently, most of these systems write and store large data as blocks of fixed size, which are distributed almost evenly among the system's nodes using random block placement or load balancing policies. In each system, one of the nodes operates as the master node e.g. the NameNode in HDFS, that keeps a record of the file directories and redirects client requests toward the storage API for opening, copying or deleting a file. However, these policies are limited as they depend on the size of the data stored in the systems as well as the policies followed by the specific storage nodes (e.g., load balancing policies). Our framework assumes the deployment of multiple HDFS clusters within the federated cloud environment, each comprising a different master node and storage layer. The client edge device can communicate with each of the master nodes with a different interface in order to store different groups of data into separate HDFS clusters.

Erasure Codes

Distributed systems deploy erasure codes as a storage-efficient alternative to replication so as to guarantee fault tolerance and data availability for their stored data. Erasure codes are a form of Forward Error Correction (FEC) codes that can achieve fault tolerance in the communication between a sender and a receiver by adding redundant information in a message; this enables the detection and correction of errors without the need for re-transmission. For instance, a sender node encodes a file with erasure coding and generates a data codeword or a stripe containing original and redundant parity data. Next, the sender node sends sequentially the blocks of the encoded stripe to a receiver node. In its turn, the receiver node detects whether there is a sufficient number of available blocks in order to decode them into their original content. If no original blocks are received, the parity blocks can repair them up to a finite range.

The most commonly used erasure code algorithm is the Reed-Solomon (RS), a maximum distance separable code (MDS) which is expressed as a pair of parameters (b,k) (RS(b,k)) where b is the number of input chunks on a data block and k is the number of parity chunks created by the erasure code. The parity chunks are generated by utilizing Cauchy or Vandermonde matrices over a GF (2 m ) Galois Field, where 2 m is the number of elements in the field and m is the word size of encoding. The code constructs a matrix of size k × d which contains values from the GF (2 m ) field that correspond to the dimensions of the matrix and represent the positions of the input chunks. Next, the RS code derives an inverse k ×k submatrix from the previous matrix. The original matrix is multiplied by the inverse submatrix in order to convert the top square of the former into a k × k identity matrix which will keep the content of the original data chunks unaltered during the encoding and decoding processes. The result is a stripe of length n = b + k, that contains the b chunks of the original data and the k parity chunks generated by the code. RS is k-fault tolerant due to the fact that the original data can be recovered for up to k lost chunks. In other words, while replication needs to copy and store the original data n + 1 times, erasure codes only require to store the data n-k n times, which costs considerably less compared to replication. Reed-Solomon codes are also characterized by linearity [START_REF] Macwilliams | The theory of error-correcting codes[END_REF]. In other words, they perform linear coding operations based on the Galois field arithmetic. More formally, given an (b,k) code, let B 1 , B 2 , ..., B b be the b original data chunks and P 1 , P 2 , ..., P k be the k parity chunks. Each parity chunk P j (0 < j < k) can be expressed as

P j = b i=1 (c ji • B i ), where c ji ⊂ GF (2 m
) is a coding coefficient specified by the RS code for computing P j .

This technique is limited as the redundancy provided by simple RS codes can repair up to k unavailable nodes. If there are more than k chunk erasures, the code will not be able to fully repair their original data. Our framework tries to deal with limited redundancy by deploying a more advanced erasure coding technique based on Reed-Solomon and Diagonally Interleaved Coding, the latter of which we describe in the next section.

Diagonally Interleaved Coding

Leong et al. have studied a burst erasure model in [START_REF] Leong | On coding for real-time streaming under packet erasures[END_REF], where all erasure patterns with limited-length burst erasures are admissible so that they can construct an asymptotically optimal convolutional code that achieves maximum message size for all available patterns. This code involves stripes derived from one or more data messages interleaved in a diagonal order.

For a set of parameters (c, d, k), where c is the interval between input messages, d is the total number of symbols in the encoded message (original data and parity symbols) and k is the number of generated parity symbols, an input message is equally split into a vector of c columns and d -k rows. Next, tables of blank or null symbols are placed around the message table that represent non-existent messages before and after the input message. The symbols of the entire table are interleaved in diagonal pattern, forming well-defined diagonals containing at least one symbol from the input message. Finally, a systematic block code is used to create k parity symbols for every diagonal, thus constructing a convolutional code with d -1 diagonal stripes that can repair up to k lost symbols in each diagonal and span across d consecutive time steps. As a result, diagonally interleaved codes are able to handle an extended number of erasure bursts in one message and allow smaller erasures to be fixed without accessing massive amounts of data. In figure 1 we illustrate with an example how diagonally interleaved coding is applied for a single data block.

The process of splitting an input message into a vector can be applied only if the input data is organized in single data stripes. To optimize data availability, our framework uses a derived version of diagonally interleaved coding that takes as input data organized in a grid and interleaves all content into diagonals before encoding them with a Reed-Solomon code.

Challenges

In this section we present the challenges of existing schemes and how we propose to address them in our Fed-DIC framework.

High read access and network bandwidth costs during data retrieval. One major challenge in typical cloud environments is the lack of user-oriented control in data distribution and storage. Most cloud systems store data blocks in randomly chosen nodes and nodes within racks in their clusters without balancing the load. For example, a system that uses an RS(b, k) to encode its streamed data, will distribute the d = b+k chunks of the generated codeword to d different nodes in a random order. However, in cases of node failures, the system needs to retrieve data from other nodes within the rack or even across racks to retrieve parity data, leading to high read access costs and network overhead, which can considerably slow down the repair process.

Fed-DIC deals with this problem by uploading and distributing the encoded data to a federated cloud with multiple autonomous Hadoop clusters in the same network, each with a unique NameNode. To retrieve a particular data record, the framework keeps a metadata file containing the locations of the stored encoded data. The metadata file is created and can be accessed by the edge device in order to locate the requested data record and retrieve it faster with a significantly reduced read access latency, limited to the cluster where the specific data record is stored, without the need to traverse all nodes or maintain scattered metadata among nodes or clusters. Fed-DIC's topology in terms of the stored data among the clusters of the federated cloud, combined with the reduced storage size of the data chunks generated from its encoding process, provide significantly smaller read access costs and transfer bandwidth overhead for nodes in the cloud.

Limited data availability. Distributed systems deploy erasure coding methods to achieve higher redundancy than replication with more affordable storage cost.

However, the availability provided by simple erasure codes such as Reed-Solomon codes for the encoded data is restricted to the number of parity chunks generated by the code. More specifically, a Reed-Solomon code that creates k parity data chunks from b original data chunks (RS(b, k)) can repair up to k failures between the original or parity data. If there are more than k unavailable or failed chunks in the stripe, the RS code will not be able to restore the data back to their original state.

To deal with this challenge, several advanced erasure codes have been presented, including alpha entanglement codes [START_REF] Estrada-Galinanes | Alpha Entanglement Codes: Practical Erasure Codes to Archive Data in Unreliable Environments[END_REF] and diagonally interleaved codes [START_REF] Leong | On coding for real-time streaming under packet erasures[END_REF]. Fed-DIC uses a variation of diagonally interleaved coding on a group of streaming data containing input records from multiple sensor groups (columns) across multiple days (rows). The array data are interleaved diagonally and encoded with multiple parity chunks for each arranged diagonal pattern, achieving higher data availability and greater repairing range than conventional erasure coding methods.

Load balancing unreliability. Most large-scale distributed systems deploy load balancing policies for node distribution or utilizing one-node-per-rack [START_REF] Sathiamoorthy | Xoring elephants: Novel erasure codes for big data[END_REF][START_REF] Muralidhar | f4: Facebook's Warm {BLOB} Storage System[END_REF][START_REF] Ambade | Hadoop Block Placement Policy For Different File Formats[END_REF] to balance the storage load across the cluster. However, most load balancing policies require the use of sophisticated techniques which may lead to load imbalances among nodes, especially when the number of the data chunks in a stripe exceeds the number of nodes that comprise a cluster.

Fed-DIC groups the encoded data diagonals into batches before they are stored to multiple node clusters in a non-random order. If the user decides to upload a data array and store it over the old one, the framework rotates the directions of the clusters in which the new batches will be stored in order to achieve good load balancing.

Design of the Fed-DIC Framework

To deal with the above problems of conventional erasure coding on federated clouds, we designed and developed Fed-DIC (Federated cloud Diagonally Interleaved Coding), a framework that utilizes diagonal interleaving and erasure coding on streaming data records in a federated edge cloud environment. The goal of our framework is to reduce the read access cost and network overhead caused by accessing multiple nodes in a federated cloud while maximizing data availability for the data stored in the federated cloud environment. Fed-DIC also supports load balancing by storing multiple versions of the data records among clusters in a rotational order, while keeping storage availability, using the techniques we have developed and its API for distributing the data and balancing them across the clusters. In cases of high load in a cluster due to data congestion or unavailable nodes, Fed-DIC can reconfigure the number of batches and the content size of each batch in order to achieve load balancing by storing data to a smaller number of clusters with more nodes and larger storage space.

Fig. 2: The architecture of Fed-DIC, which comprises the client devices, where all operations are performed, the network hub, which connects the clients with the cloud and the federated cloud, which contains multiple independent storage clusters.

Framework Architecture

As illustrated in figure 2 Fed-DIC comprises three main components: the client side (edge devices), a federated cloud comprising multiple independent clusters where each cluster consists of multiple independent nodes, and a network hub that connects the two other components through the network. The client devices are operated by the user and provide six services: (1) The Interleaver module which re-orders the input data set into a grid and interleaves them into diagonal groups, (2) the Coder module which encodes all diagonal groups prior to the uploading process and decodes received diagonal stripes containing user-requested data, (3) the Destination module which splits the encoded stripes into batches and configures the order of destination clusters where the batches will be stored, (4) the Hadoop Service which communicates with the NameNodes of each cluster in order to upload the diagonal stripe batches, (5) the Metadata service which creates a metadata index file during the upload process and provides a query interface for the user during the retrieval process, and (6) the Extractor module which searches through a received diagonal stripe in order to extract the data record requested by the user and store it to a new file.

Our framework works as follows: A client takes as input a set of streaming data records and organises them into a grid of D columns and G rows. The data records in the grid are re-ordered into C = D + G -1 diagonal groups, which are then encoded with Reed-Solomon, generating up to k parity chunks per diagonal using an 8-bit Galois Field. Next, Fed-DIC groups the diagonal stripes into H batches and stores each batch into a different cluster in the federated cloud. Simultaneously, the client creates a metadata file that contains information for each stored data record: The day the record was created, the group of sensors that generated the record and the diagonal stripe in which the record was interleaved. To retrieve selected data records, the client receives user-created request queries about data records and communicates with their correspondent clusters to download the stripes that include the records so as to extract their contents in output files. To upload a new version of the already stored data while archiving the older versions, the cluster destinations are rotated in a stack order by setting the first cluster destination at the position of the last cluster destination in a circular pattern. In that way, Fed-DIC achieves not only the maintenance of multiple versions, but also load balancing throughout all clusters within the federated cloud. If Fed-DIC kept uploading newer versions into the same clusters each time, there could have been inconsistencies between the clusters. Especially, the first and last clusters in the cloud would have smaller data load than the other clusters.

Read Access Latency

The read access cost for a data query q from a group of Q queries, is given by the sum of the access time a client needs to traverse l lines in the metadata file to find the requested data, the latency needed to access any h clusters that contain the data (h ≤ H) and the search delay caused by any missing d data chunks in a cluster. The probability p i shows if a chunk is available for transferring. If p i = 0, the chunk is missing. This is computed by the following formula:

T q = l • r md + h • r h + d i=1 ((1 -p i ) • t m )
where r md is the time a client needs to read a line from the metadata file, r h is the time to access a cluster in the federated cloud and t m is the search delay caused by missing data chunks in the cloud. The read access latency L q for downloading and extracting a requested data query q is given by the access cost T q which was computed previously, plus the time required to download all available d chunks in the diagonal stripe that includes the data using an internet connection of B bandwidth and the computation time T dec q a client needs to decode the diagonal stripe so as to extract the result. The formula for the overall query storage latency is given below:

L q = T q + d i=1 (pi•tp) B + T dec q
where t p is the elapsed time for an available data chunk to be transferred from the federated cloud to a client device. Similarly, the total read access latency L Q is the sum of the read access latency for all Q queries:

L Q = Q q=1 (L q )
The read access latency for erasure coding is computed in a similar way to L q , with the only difference that the metadata access time is not taken into account.

Data Loss percentage

When stored chunks are missing or unavailable in the federated cloud due to failures or nodes being disabled in the cloud's clusters, erasure codes try to utilize any available parity chunks in order to reconstruct the damaged encoded file. However, if a decent amount of chunks are not available in a cluster, there may be permanent loss of the original data, due to the number of available data chunks being insufficient for use with erasure codes. The data loss percentage D C of a fault tolerance method is measured by the fraction of the probability p i of a data chunk c i being available with the total number of data chunks in the entire cloud, subtracted from 1, as follows:

D C = (1 - C i (pi•ci) C ) • 100

Framework API

Fed-DIC provides an API with the following four operations: Encode() This operation interleaves the input data set into D diagonal data groups of varied length. Then, it merges data in each group into new data blocks so as to be encoded with a unique Reed-Solomon erasure code.

Store() This operation groups the encoded diagonal data stripes into H batches containing an equal number of (D/H) codewords in each batch and communicates with all the NameNodes of the federated cloud in order to upload and store each batch in a different cluster, while keeping track of the data locations and information in a metadata file stored in the client devices. The metadata file can be shared and backed up in all clients in order to avoid any corruptions. If, for any reason, the cloud changes the location of its clusters, the clients need to update the metadata accordingly. However, a small non-significant access overhead may occur in the case that the client device that performs the Store() process becomes unavailable and the metadata have to be accessed from another client. Due to the integrity of our private client nodes, the probability of this situation is extremely rare, so it is not considered when measuring the read access latency.

Retrieve() This method provides an interface to the user for entering multiple queries regarding a data record the user aims to retrieve. Once the user issues his queries, the method searches for each requested data record the diagonal stripe in which it is included and downloads it accessing immediately the corresponding storage cluster. Decode() Once the clients receive the diagonal stripes with the data requested by the user, this operation decodes any available chunks in a stripe into its original merged data block and extracts the requested result from the block before deleting it.

Uploading and Downloading Algorithms

We describe the two main algorithms implemented by our framework: Storing data to the federated cloud: A client takes as input the data records to be uploaded, these correspond to G sensor data groups over a time period R days, stored in .csv files. The client invokes the Encode() operation to organize the content into a grid with dimensions G×R, where its elements are interleaved into C dynamic diagonal arrays of varied length (as shown in Figure 1). Records are inserted into the grid according to the day and sensor group indicated on the record. Starting with the record of the last sensor group during the first day, the client forms a diagonal line from bottom right to top left and inserts any existing grid elements in the diagonal line, into a dynamic diagonal array. The diagonal arrays span through the entire grid with the last one containing only the record of the first sensor group during the last day. Next, in each diagonal array, the data in the elements are merged into a single data object and encoded using a (b, k) Reed-Solomon code. The encoding process splits each merged data object into equally sized b chunks and generates k parity chunks, creating a stripe of length d = b + k. Next, the client uses the operation Store() to group the diagonal stripes into H batches containing an equal number of (C/H) stripes in each batch and to upload the batches into the different clusters of the federated cloud by communicating with every NameNode within the cloud. Once the NameNode of a cluster receives the data, it distributes the chunks in random order to its nodes. During the storage process, the clients write and store metadata records about the stored data, their version, the date and sensor group as well as the number of the diagonal stripes they belong to. The metadata file helps the edge devices to access the stored data faster and more easily by reducing the access costs among the HDFS clusters. The distribution of the batches is performed in a sequential way. For example, in a federated cloud of F clusters, the first data batch is stored into the first cluster and so on until the last batch is stored in the F -th cluster. When the user wants to upload a new version of the data over the already stored versions, the clients swap the order of the cluster destinations by placing the first cluster destination right after the last cluster destination of the older version in a Last In, First Out (LIFO) order. In our example, for the second version of our data, the first batch will be uploaded into the F -th cluster, the second one to the first cluster and so on with the last cluster being uploaded to the (F -1)-th cluster. The way data records are stored in Fed-DIC enables us to traverse 1-2 clusters at most to recover any data segment. Whereas, conventional (b, k) Reed-Solomon would merge r 1 with every other record in the input into a single data block, split it into b original chunks and encode it using a Galois Field matrix to generate k parity chunks which are distributed to the cloud via Hadoop. Thus, even if a small part of data must be recovered, the data encoded with RS need to be restored in their entirety, which may require traversing all clusters in the cloud, incurring a heavy read access cost.

Retrieving data from the federated cloud: The clients provide an interface to the user awaiting response queries. When the user issues a query, the clients gather all entered queries into a list array and use Retrieve() to search through the metadata file generated from the uploading process for the diagonal stripes where the query data are stored. For every entry in the query list, the client connects to the correspondent cluster to download the diagonal stripe with the requested data. If the edge device fails to download sufficient amount of chunks for restoring the stripe into its original data, it informs the user that the queried data from that diagonal stripe cannot be recovered. If it receives enough chunks from the stripe, it deploys Decode() to restore the diagonal stripe using RS(b, k) back to its original content. Finally, the clients search through the recovered data objects for the requested record entries and extract them as a result. When there are multiple concurrent requests from users, the clients schedule the requests to the hub in multiple rows according to the source cluster of the requested data and return the result for the oldest request each time.

Experimental Evaluation

In this section we evaluate Fed-DIC in terms of data loss, maximum transfer network rate and storage overhead, compared to the Replication and conventional For memory and disk allocation reasons, the VMs are running across 2 real desktop machines: Our client device and a second machine with the same hardware specifications as the first, which is connected to the same network. 8 VMs are running on each machine, connected to the same network as the client machines using a bridged adapter. Our setup is restricted to the equipment and network availability in our local computing and communication environment, however the algorithms we have developed can adapt well to accommodate larger clusters with thousands of nodes by modifying the number of batches in which the encoded data will be grouped as well as the content size in each batch. Also, we can set the batches to be stored in clusters with higher reliability within a large cloud. Our data set for the experiments is a collection of transport values obtained from SCATS sensors that are deployed in the Dublin Smart City [START_REF]SCATS[END_REF]. This data set contains a huge amount of records with information regarding the specific sensor that captured the snapshot and its capture date; the data needs to be stored and maintained in the cloud to be further analyzed by the human operators (i.e., to identify congested streets and entire geographical areas over time). Fed-DIC is responsible to store and recover this data to and from the cloud. Our first experiment involves the total read latency of recovering data with Fed-DIC(7,4) compared to Reed-Solomon [START_REF] Li | D 3 : Deterministic Data Distribution for Efficient Data Reconstruction in Erasure-Coded Distributed Storage Systems[END_REF][START_REF] Shvachko | The hadoop distributed file system[END_REF]. For RS [START_REF] Li | D 3 : Deterministic Data Distribution for Efficient Data Reconstruction in Erasure-Coded Distributed Storage Systems[END_REF][START_REF] Shvachko | The hadoop distributed file system[END_REF] we merge the input to store and retrieve the data successfully utilizing only 3 clusters. However, in order to take advantage of the entire experimental environment (4-cluster cloud system with a total of 16 nodes) we utilize all 4 clusters for both techniques, to avoid load imbalances (data distributed in 3 clusters, while the 4th cluster is unused) and minimize the impact on the data loss percentage (in cases of failures). Due to the data chunks spanning across all 4 clusters, a simple decoding process with RS takes almost 20 seconds to complete, as seen in figure 3. This happens due to the clients having to access all 4 clusters in order to download all the chunks needed for recovering the stripe's original data. Even if we request a small portion of the encoded data, Reed-Solomon has no built-in features that allow us to retrieve a specific part of data, so it will still have to retrieve and decode the entire file content in order to give us an output. Our Fed-DIC technique on the other hand, reduces the total access latency by returning only requested parts of the stored data instead of the entire data content by accessing 1 to 2 clusters at most. For 1 to 4 queries for data inside the same cluster, Fed-DIC achieves at least 60% lower read access latency compared to RS. Even in the case that we request 2 data queries that are stored in two different clusters, Fed-DIC still reads the data in a shorter time compared to RS.

Our second experiment evaluates the reliability between 3-way Replication, RS [START_REF] Li | D 3 : Deterministic Data Distribution for Efficient Data Reconstruction in Erasure-Coded Distributed Storage Systems[END_REF][START_REF] Shvachko | The hadoop distributed file system[END_REF] and Fed-DIC [START_REF] Li | D 3 : Deterministic Data Distribution for Efficient Data Reconstruction in Erasure-Coded Distributed Storage Systems[END_REF][START_REF] Shvachko | The hadoop distributed file system[END_REF] in the data loss scenario. We performed 3 runs of experiments. As figure 4 indicates, due to its organized multi-cluster storage policies, Fed-DIC manages to achieve lower data loss rates than RS. Even when only up to 40% of the nodes are available in the federated cloud, Fed-DIC may be able to maintain a sufficient number of chunks in some diagonal stripes, which allows it to restore a portion the original data. The next experiment we evaluate the storage overhead and the maximum network transfer rate between these fault tolerance methods. As figure 5a shows, Replication stores the entire data content inside the cluster without splitting it, causing a large storage overhead even for single blocks, compared to a chunk produced by simple Erasure Coding and Fed-DIC. In figure 5b we present the total storage overhead for all three methods. 3-Way replication occupies a massive portion of the storage with all 3 replicas combined, while all chunks generated by Erasure Coding and Fed-DIC produce lower overheads, with the latter occupying slightly less storage than erasure coding due to the varied sizes of the chunks. We also measured the rates during data transferring using performance monitoring programs included with Lubuntu OS. As seen in figure 6 due to the size of the replicas, Replication severely burdens the network with a high transfer rate of 1.2 MBps, followed by Erasure Coding with a transfer rate of 900 KBps. Fed-DIC operates with smaller data transfers and thus provides smaller and less burdening network data rates when transferring one or multiple queried data records.

Finally, figure 7 shows the load balancing achieved in the three fault tolerance methods between 4 4-node clusters, while uploading 4 different data streams with similar sizes. Due to the random distribution of replicas and chunks in the HDFS cloud, Replication and client-side Reed-Solomon erasure codes are very inconsistent in terms of load balancing. Specifically, a majority of data may be stored to one cluster, while other clusters store less data, even though Erasure Coding seems more consistent than Replication. It is worth to note that we do not consider HDFS server-side erasure coding since it requires a code with higher parameters, which generates a number of chunks equal to the number of nodes in a single cluster. Meanwhile, Fed-DIC, using the rotational stack policy for cluster destinations described previously, it can store new streams in the federated cloud's clusters in a different order for every stream. Since our framework stores data of different size in each cluster in every uploading process, it can maintain an almost perfect load balance between H clusters for each H uploaded streams. For example, in figure 7 for every 4 streams uploaded in the

Related Work

Several approaches over the last decade have been proposed for improving read access costs and the reliability of erasure coding in cloud storage environments.

In particular, a method that drastically improves read access costs and data reconstruction in erasure coded storage systems is Deterministic Data Distribution, or D 3 for short [START_REF] Li | D 3 : Deterministic Data Distribution for Efficient Data Reconstruction in Erasure-Coded Distributed Storage Systems[END_REF]. D 3 maximizes the reliability of a storage system and reduces cross rack repair traffic by utilizing deterministic distribution of data blocks across the storage system. D 3 uses orthogonal arrays to define the data layout in which the data will be distributed across multiple racks, ignoring the one block per rack placement, while balancing the load among nodes across the system's racks. This implementation works on single HDFS clusters with multiple racks but it does not seem to support federated clouds or other systems with independent clusters, unlike our approach with Fed-DIC. Even if we modify D 3 to support multiple clusters, the clusters need to contain a certain number of nodes in order to apply server-side erasure coding, whereas in Fed-DIC, erasure coding is performed by the client devices. Simple erasure codes provide efficient fault tolerance but their reliability is restricted to the parameters set by the user. Advanced erasure coding techniques like Alpha entanglement codes by Estrada et al. [START_REF] Estrada-Galinanes | Alpha Entanglement Codes: Practical Erasure Codes to Archive Data in Unreliable Environments[END_REF], increase the reliability and the integrity of a system compared to normal Reed-Solomon codes by entangling old and new data blocks and creating robust, flexible meshes of interdependent data with multiple redundancy paths. Also in the Ring framework for key-value stores (KVS) [START_REF] Taranov | Fast and strongly-consistent peritem resilience in Key-Value Stores[END_REF], Taranov et al. introduce Stretched Reed-Solomon (SRS) codes which support a single key-to-node mapping for multiple resilience levels. These lead to higher and more expanded reliability compared to conventional Reed-Solomon codes. However, this work is only restricted to key-value stores and is not available to conventional databases for use. Also, unlike our work, the reliability ranges of SRS are limited only to the parameters of specific key-tonode mappings.

Hybris [START_REF] Dobre | Hybris: Robust hybrid cloud storage[END_REF] by Dobre et al. is a hybrid cloud storage system that scatters data across multiple unreliable or inconsistent public clouds, and it stores and replicates metadata information within trusted private nodes. The metadata are related to the data scattered across the public clouds, providing easier access and strong consistency for the data, as well as improved system performance and storage costs compared to existing multi-cloud storage systems. In our case, Fed-DIC uses metadata containing information about the topology of data stored in a federated cloud so that the client can connect immediately to the cluster that contains a requested portion of the data, thus drastically reducing the read access cost in these systems compared to simple erasure codes.

Conclusion

In this paper, we presented Fed-DIC, our framework that integrates Diagonal Interleaved Coding with organized storage of the encoded data in a federated cloud environment. Our framework takes as input data organized in a grid, interleaves them into diagonal stripes that are encoded using a Reed-Solomon erasure code. The encoded diagonal stripes are grouped into batches which are stored to different clusters in the cloud. The user issues queries to retrieve portions of the data without the need for the clients to access every cluster in the cloud, thus reducing the access cost compared to other methods like Replication and simple Erasure Codes. Our experimental evaluations illustrate the benefits of our framework compared to other fault tolerance methods in terms of total read access latency, data loss percentage, maximum network transfer rate, storage overhead and load balancing. For future work, one direction we are following is to deploy Fed-DIC in a federated environment with different hardware equipment where we plan to evaluate the working and benefits as well as the corresponding costs of our approach when different types of equipment are utilized.

Fig. 1 :

 1 Fig. 1: A pictorial representation of diagonally interleaved coding for an input message with (c, d, a) = (2, 5, 2). The data blocks are rearranged into diagonals and each diagonal is encoded into stripes (D 1 ...D 4 ) by the systematic code. B Symbols in time steps from -1 to 0 and from 3 to 6 are assumed to have zero, null or non-positive values, and they are not part of the input message.

Fig. 3 :Fig. 4 :

 34 Fig. 3: Read access latency for Reed-Solomon and Fed-DIC (multiple queries)

Fig. 5 :

 5 Fig. 5: Storage overhead for Replication, Erasure Coding, and Fed-DIC

Fig. 6 :

 6 Fig. 6: Comparing Replication, Erasure Coding and Fed-DIC in terms of maximum network transfer rate (single record retrieval and 1 diagonal retrieval)

Fig. 7 :

 7 Fig. 7: Load balancing between 4 file clusters for all fault methods including Fed-DIC

Acknowledgment. This research has been supported by the Computer Systems and Communications Laboratory at AUEB. The authors would like to thank Dr. Davide Frey for shepherding the paper.


Benjamin Taubmann

Hans P Reiser

Towards hypervisor support for enhancing the performance of virtual machine introspection

Virtual machine introspection (VMI) is the process of external monitoring of virtual machines. Previous work has demonstrated that VMI can contribute to the security of cloud environments and distributed systems, as it enables, for example, stealthy intrusion detection. One of the biggest challenges for applying VMI in production environments is the performance overhead that certain tracing operations impose on the monitored virtual machine. In this paper, we show how this performance overhead can be significantly minimized by incorporating minor extensions for VMI operations into the hypervisor. In a proof-ofconcept implementation, we demonstrate that the pre-processing of VMI events in the Xen hypervisor reduces the monitoring overhead for the use case of VMI-based process-bound monitoring by a factor of 18.

Introduction

Virtual machine introspection (VMI) is the process of analyzing the state of a virtual machine from outside, i.e., the perspective of the hypervisor [START_REF] Garfinkel | A Virtual Machine Introspection Based Architecture for Intrusion Detection[END_REF]. Based on the external view, VMI-based monitoring has certain properties that make it appealing for many application scenarios, including solutions that aim to enhance the security in cloud environments and distributed systems [START_REF] Hebbal | Virtual Machine Introspection: Techniques and Applications[END_REF][START_REF] Jain | SoK: Introspections on Trust and the Semantic Gap[END_REF]. Hence, it is not surprising that the first application for VMI was an intrusion detection system [START_REF] Garfinkel | A Virtual Machine Introspection Based Architecture for Intrusion Detection[END_REF]. Those properties are: isolation between the monitoring and the monitored system, an untampered view on the system state, and the stealthiness of monitoring.

In the following we use the term production virtual machine for the virtual machine that is monitored via VMI and is performing the normal operations. The term monitoring virtual machine is used for the virtual machine that analyzes the production virtual machine using VMI [START_REF] Taubmann | CloudPhylactor: Harnessing mandatory access control for virtual machine introspection in cloud data centers[END_REF]. The focus of this paper is on the Xen architecture. Hence, the term monitoring virtual machine refers to either the Dom0 of Xen or any other virtual machine with the permissions to perform VMI-based operations. The VMI application that performs the analysis does not run directly in the hypervisor.

There are two forms of virtual machine introspection: asynchronous and synchronous. The main difference between these operation modes is how the analysis is started [START_REF] Jain | SoK: Introspections on Trust and the Semantic Gap[END_REF]. We speak of asynchronous virtual machine introspection if the analysis is started by external events, such as timers that periodically retrieve the system state. This is useful to regularly observe the system state and retrieve information that is in memory for a longer time frame, such as the list of running processes. Synchronous virtual machine introspection is triggered by sensitive operations in the control flow of the monitored system, e.g., software breakpoints. This can be required to retrieve ephemeral information that resides in main memory only for a short period, such as the parameters of function calls.

In this paper we use the term performance impact to quantify the overhead induced on the execution in the production virtual machine due to VMI-based operations. The overhead is mainly caused by the fact that the production virtual machine is paused for the analysis, which has three different reasons. The first one concerns the case when the production virtual machine and monitoring virtual machine share the same physical CPU core. In this case, the production virtual machine gets paused when the monitoring system is running. This overhead factor can be reduced by using an additional CPU core or by minimizing the amount of time required for the analysis.

The second reason is that the production virtual machine must be paused to prevent that the analyzed data structure (e.g., the process list) is modified during the analysis. If the analysis were performed on a running system, memory contents changing concurrently to the analysis could result in incorrect information. In order to obtain reliable information, it is, therefore, necessary to perform the analysis in a non-modifying state in order to retrieve a consistent view of the system state. A common approach to address this problem is to take a snapshot of the contents in main memory and then run the analysis on it. However, taking a snapshot of the main memory may also require to pause the production virtual machine. Klemperer et al. [START_REF] Klemperer | High-performance memory snapshotting for real-time, consistent, hypervisor-based monitors[END_REF] addressed this problem by implementing a snapshot mechanism that uses a copy-on-write strategy.

The third reason is that the production virtual machine is paused due to synchronous VMI-based mechanisms, e.g., when a breakpoint is invoked and the control flow of the production virtual machine traps to the monitoring virtual machine. The impact on the performance of the production virtual machine, in this case, depends on how often the monitored virtual machine is interrupted and the time how long the monitored system is paused for the analysis.

While the first two causes of performance overhead have adequate solutions, the third reason is an open problem that severely limits the applicability of synchronous VMI-based monitoring in practical use cases. In this paper, we focus on how to minimize the performance overhead in this third case and discuss mechanisms to enable efficient VMI-based monitoring. In detail, we discuss an approach that reduces the overhead of process-bound system call monitoring. Minimizing the overhead of synchronous VMI-based monitoring is important to implement VMI-based security solutions in production environments that tolerate only a minimal performance impact.

The contributions of this paper are:

a presentation of different approaches that aim to minimize the performance impact of synchronous virtual machine introspection;

a proof-of-concept implementation that minimizes the performance impact of process-bound monitoring; the evaluation of the proof-of-concept implementation.

The outline of the paper is as follows: Section 2 introduces the most important technologies used by virtual machine introspection and the most common mechanisms of synchronous virtual machine introspection. Section 3 discusses approaches on how to minimize the overhead of synchronous VMI-based monitoring. In Section 4, we discuss a prototype implementation that aims at minimizing the monitoring overhead of process-bound system call tracing. In Section 5, we measure the performance gain of the prototype. Section 6 discusses related approaches and Section 7 concludes the paper.

Virtual Machine Introspection

This section describes the technologies that are used for virtual machine introspection and synchronous monitoring.

Hardware Requirements

Synchronous virtual machine introspection requires that the CPU is able to trap to the hypervisor in order to monitor the execution of a virtual machine. For example, the hardware virtualization instruction set of Intel processors (Intel VT-x) allows trapping to the hypervisor (VM-Exit) when certain sensitive instructions are invoked in a virtual machine [START_REF]Intel Corporation: Intel ® 64 and IA-32 Architectures Software Developer's Manual[END_REF][START_REF] Uhlig | Intel virtualization technology[END_REF]. The virtual-machine control structure (VMCS) defines for each virtual machine in which case it should trap to the hypervisor (VMX root mode). While the hypervisor handles the event, the virtual machine that invoked the sensitive operation is paused.

Hypervisor Support

The hypervisor has an important role for virtual machine introspection. For asynchronous introspection, the hypervisor must give the monitoring virtual machine the permissions to access the main memory of the production virtual machine. This can be done for example by mapping the memory pages from the production virtual machine into the address space of the monitoring virtual machine. For synchronous monitoring, the hypervisor needs to forward the information about traps to the monitoring virtual machine so that it can run analysis operations. The concepts presented in this paper are mostly hypervisor independent. However, the focus of this paper is on the design of the Xen architecture.

Xen [START_REF] Barham | Xen and the art of virtualization[END_REF] is a bare-metal hypervisor that supports the hardware virtualization instruction set of Intel. The relevant components for VMI-based operations using Xen are depicted in Figure 1. The Dom0 is the virtual machine with the most permissions and can access the main memory of all virtual machines running on this hypervisor. Thus, most VMI-based approaches implement the VMI application in Dom0. However, with the Xen security modules (XSM) it is possible to grant other virtual machines than the Dom0 the permissions required to perform VMI-based operations [START_REF] Shi | A Disjunctive VMI Model Based on XSM[END_REF][START_REF] Taubmann | CloudPhylactor: Harnessing mandatory access control for virtual machine introspection in cloud data centers[END_REF]. This has certain advantages and can be useful in cloud computing when it is necessary to restrict the VMI permissions of a monitoring virtual machine to the production virtual machine of a specific user.

Asynchronous VMI-based operations can be implemented with Xen by mapping the relevant memory pages of a virtual machine to the monitoring virtual machine, e.g., the Dom0. The functions that map memory pages are implemented in the libxencontrol (libxc) library and use Xen hypercalls to instruct the hypervisor. For synchronous VMI-based operations, sensitive operations in the production virtual machine cause a trap to the hypervisor. Then, the monitoring module of the hypervisor becomes active and sends the corresponding VMI event via the event channel to the VMI application in the Dom0/monitoring virtual machine. While the event is processed in the monitoring virtual machine, the production virtual machine is paused.

VMI Applications

LibVMI [START_REF] Payne | Simplifying virtual machine introspection using libvmi[END_REF][START_REF] Payne | Secure and flexible monitoring of virtual machines[END_REF] is the defacto standard library that helps to build VMI-based monitoring applications. LibVMI provides the API to perform VMI-based operations on Xen and KVM. This includes, for example, functions for reading and writing content from the main memory and functions for translating virtual and physical addresses. In addition, LibVMI supports synchronous monitoring with Xen. For this purpose, it provides primitives for listening on VMI-events generated by the Xen hypervisor.

Basic Synchronous VMI Methods

The main advantages of synchronous virtual machine introspection are that it can be used to monitor the control flow of virtual machines and to extract ephemeral information such as parameters of function calls. The mechanisms of synchronous monitoring utilize the concepts of hardware virtualization that allow trapping to the hypervisor under certain conditions, e.g., when sensitive instructions are invoked. In the following, we explain the most common VMIbased monitoring techniques: breakpoints, the monitoring of the write access to CPU control registers, and system call tracing.

Breakpoints A common approach to monitoring the control flow using virtual machine introspection is to insert software breakpoints, by using the INT3 instructions that trap to the hypervisor when they are invoked. One approach to implement a software breakpoint mechanism for virtual machine introspection requires the following eight steps. First, the software breakpoint is inserted at the function that should be monitored, either by changing the content in memory or by manipulating the page tables1 . Second, the production virtual machine invokes the function and the CPU traps to the hypervisor. Third, the hypervisor handles the trap caused by the INT3 instruction, creates a VMI monitoring event and sends it via the Xen event channel to the virtual machine handling the event. Fourth, the monitoring virtual machine gets active and the VMI application receives the new event. Fifth, the VMI application analysis the state of the production virtual machine and re-inserts the original instruction. The VMI application finishes the investigation and tells the hypervisor that the production virtual machine should perform a single-step and execute the original instruction. Sixth, the CPU is configured to run the original instruction in the context of the production virtual machine in single-step mode. Afterward, the CPU traps again to the hypervisor. Seventh, steps four and five are repeated with the difference that the VMI application inserts the original breakpoint. Eight, the VMI application tells the hypervisor to resume the production virtual machine.

In total, this procedure requires switching twice between the monitoring virtual machine and the production virtual machine. First, to handle the breakpoint, and second, to reinsert it after the original instruction is executed in single-step mode. Since a software breakpoint actually traps first to the Xen hypervisor and is then handled by a monitoring virtual machine, eight VM Exit/Entry-based context switches are required to handle a single breakpoint (see Figure 1). Also, since the VMI application is usually implemented as a userspace application in monitoring virtual machine, additional context switches between the kernel of monitoring virtual machine and the VMI applications are required. The actual number of required context switches depends on many factors, such as the scheduler's decision and how many processes are active in the monitoring virtual machine.

Access to Control Registers Modifications to the control registers of a CPU can change the general behavior of the system. Hence, observing the changes is a valuable mechanism of virtual machine introspection. The most common example is the monitoring of modifications of the CR3 register. In the Intel architecture, the CR3 register holds a pointer to the directory table base (DTB) of a process. Whenever a process is dispatched by the scheduler of the operating system, it updates the contents of the CR3 register to point to the DTB of the next process. Thus, monitoring changes of the CR3 register can be used to be informed whenever a new process becomes active.

System Call Tracing

Software breakpoints can be used to monitor the invocation of system calls. To do so, the breakpoint is set on the first instruction of the system call handler function. Since the operating system kernel is used by all processes, this means that all invocations of system calls of all processes are monitored when a breakpoint is set on a system call handler. Thus, if only the system calls of one process should be monitored, this approach can have unnecessary high overhead.

There are two strategies to obtain the system call invocations of a specific process. The first one is to use post-processing. This means that the monitoring application itself filters the events in order to obtain only those system calls that belong to a process with a specific process identifier (PID). This, of course, does not have any positive impact on the performance at run-time. The second strategy is to additionally monitor modifications of the CR3 register, which indicate that a different process in the production virtual machine is being dispatched. Every time a new process is dispatched, the breakpoints can be inserted/removed from main memory so that the tracing is only active when specific processes are running. Sentanoe et al. [START_REF] Sentanoe | Sarracenia: Enhancing the Performance and Stealthiness of SSH Honeypots Using Virtual Machine Introspection[END_REF] call this approach process-bound monitoring.

Improving the Performance

The methods of synchronous virtual machine introspection have a severe performance impact on the production virtual machine. One big part of that problem is the long chain of subsequent operations that are required to handle a VMI event in the hypervisor and the monitoring virtual machine while the production virtual machine is paused. Those steps are processed for each VMI event, even if the VMI application just disregards the event because it does not match a certain filter.

The processing of a breakpoint in the VMI application can be organized into three processing layers. On the lowest layer, the breakpoint is handled with the steps explained in Section 2.4. Then on the layer above, the VMI-events of a breakpoint are filtered and processed in the VMI application, e.g., when only breakpoints of a specific process should be monitored VMI events can be filtered based on the value stored in the CR3 register. On the highest layer, additional data from the production virtual machine, e.g., the parameters passed to a function are extracted.

In the following, we discuss possible approaches that aim at minimizing the VMI performance overhead. All of these approaches achieve this goal by reducing the time needed by the data extraction routine and the time during which the monitored virtual machine is paused.

Pre-filtering of events: In some cases, only a few of the VMI events are necessary for a specific use case. For example, in the case of process-bound monitoring, many events are generated by changes in the CR3 register. However, only events for changes that either write the value of the DTB of the process to be monitored to CR3 or replace that value with a different value are required to enable/disable the breakpoints. All other CR3-based events (e.g., when a context switch from a not monitored to another not monitored process occurs) don't need to be observed. Nevertheless, in the usual approach to monitor CR3 events, a context switch to the monitoring virtual machine is required in all cases, even if the VMI application simply ignores irrelevant events.

One approach to solve this problem is to filter events at the hypervisor level. For example, CR3 events that are captured because of the production virtual machine changing the register from oldDTB to newDTB could be forwarded to the userspace VMI application only if either oldDTB or newDTB match the DTB of the process to be monitored.

Breakpoint mechanism: A fast and efficient breakpoint mechanism helps to minimize the tracing overhead. Since breakpoints can occur frequently, small optimizations can make a big difference. Hence, the implementation of breakpoints should be as optimized as possible so that it requires as little time as possible.

Moreover, the breakpoint mechanism should be implemented into the hypervisor. This eliminates the need to switch to the monitoring virtual machine after a single-step operation was executed as the restoration of the original instruction is independent of the VMI application logic. So instead of switching to the Pre-processing of events: Another approach to reducing the number of context switches of VMI-based monitoring is to integrate parts of the analysis into the hypervisor so that it is not necessary to always switch to the monitoring virtual machine. However, hypervisors should be designed to be very minimal and contain only the functions to manage virtual machines. Integrating additional VMI code would violate this principle. Thus, we propose to run only very basic information extraction routines that are necessary to access ephemeral data in the context of the hypervisor. Complex analysis of this data should still occur in the context of the monitoring virtual machine. With that approach, it should be possible to minimize the time when production virtual machine must be paused, i.e., the time to extract ephemeral data. The extracted data must then be sent to the VMI-application in the monitoring virtual machine so that it can process the data asynchronous to the execution of the production virtual machine.

Prototype

In the following, we describe how the performance impact of process-bound VMIbased monitoring can be minimized by implementing basic pre-filtering primi-tives into the hypervisor. The implemented architecture is depicted in Figure 2. This goal of process-bound monitoring is to decrease the monitoring overhead, especially, when the production virtual machine runs many applications that are not relevant to trace for the analysis. As described in Section 2.4, one way to implement process-bound monitoring is to intercept write access to the CR3 register and en-/disable the breakpoints based on the fact which process becomes active. The problem of this approach is that the operating system updates the CR3 registers frequently. Hence, depending on how often the operating system changes the active process, many VMI-events must be processed by the VMI application. However, most of those events are irrelevant, because the VMIapplication does not need to be informed about context switches between two processes that are not monitored.

To tackle that problem, our prototype implements a basic pre-filtering mechanism in the Xen hypervisor. The VMI application can configure the filter so that only CR3 events are forwarded to the VMI application when the current or the new value of the CR3 register matches the DTB of the monitored process. The filter can be configured at run-time using hypercalls from the monitoring virtual machine. To achieve that, we extend the hypercall API of Xen to activate or disable the monitoring and set values of CR3 events that should be forwarded. With this approach, we eliminate the need for many context switches to the monitoring virtual machine.

The changes to the Xen hypervisor are essentially the extension of the hypercall and the filtering mechanism. Both modifications require only a few lines of code in the Xen hypervisor. Additionally, the libxc used by the VMI application must be extended to support the introduced parameter of the hypercall. Finally, the VMI application must be extended to use the new CR3 filtering.

Evaluation

In this section, we evaluate the performance improvement of our prototype implementation for process-bound monitoring. In our evaluation, we examine whether our prototype that pre-filters VMI events in the hypervisor can help to reduce the performance overhead. For that purpose, we use a similar use case as Sentanoe et al. for the Sarracenia Honeypot [START_REF] Sentanoe | Sarracenia: Enhancing the Performance and Stealthiness of SSH Honeypots Using Virtual Machine Introspection[END_REF], which aims to reconstruct the inputs and outputs of an attacker in a bash terminal session. Hence, for this use case, it is sufficient to monitor only the read, write and exec system calls invoked by the main bash process of an SSH session.

We obtain the performance impact of tracing by measuring the time that is consumed by extracting the zip file of the jansson library2 and compiling the source code. Both the extracting of the zip file and the compilation process require the invocation of many system calls. Most of them are not important for monitoring the attacker's behavior as they belong to the compilation process. Thus, process-bound monitoring of the main bash process that only starts the extraction and compilation should help to reduce the monitoring cost. Table 1: Measurements of the time, the number of intercepted system calls (read, write, open, close and exec) and CR3 events using four different monitoring mechanisms

In our evaluation, we run this use case while monitoring with four different mechanisms:

1. No monitoring 2. Monitoring the system calls of all running processes 3. Monitoring the system calls of the bash process only. The tracing is based on the fact which process is currently running. The CR3 events are not prefiltered in the hypervisor. This is similar to the approach of Sentanoe et al. 4. Monitoring the system calls of the bash only by en-/disabling the tracing based on the fact which process is currently running. The CR3 events are pre-filtered in the hypervisor.

The measurements are executed on an HP Elitebook 820 G4 with an Intel(R) Core(TM) i5-7200U CPU @ 2.50GHz processor. The production virtual machine has one CPU core and 256 MB of main memory. The VMI application is running in the Dom0 and is implemented to monitor the read, write, open, close and exec system call of the bash process of an SSH session that runs the extraction and compilation. The monitoring of the system calls is implemented by placing a breakpoint on the corresponding system call handler function.

The results of our measurements are depicted in Table 1. The second row shows the average execution time of the compilation process that was executed ten times. The execution time is measured in the production virtual machine using the Linux time command. We use the elapsed real-time to quantify the run-time of the unpacking and compilation process over all ten iterations. The third and fourth rows provide the number of monitored system calls and CR3 events over all ten iterations.

At first glance, the approach of process-bound monitoring of system calls appears to be beneficial in order to reduce the impact of monitoring. However, the measurements of monitoring method three show that the required monitoring of write access to the CR3 register in the VMI application almost annihilates this positive effect. This is because the CR3 events are more frequent than system call events. Nevertheless, we measure a slight improvement in performance. The results of the method four show that implementing basic analysis mechanisms into the hypervisor can significantly reduce the impact of VMI-based monitoring. Based on method four we can estimate the overhead for a single CR3 event by dividing the overhead through the number of CR3 events. From this calculation, we get an overhead of about 11ms per CR3 event. This long time period can be explained with the different layers in which the event is processed (see Figure 1).

To sum up, the process of pre-filtering CR3 events decreases the run-time of the execution with process bound tracing from 148.92 s to 8.31 s, which is about 18 times faster. Additionally, these measurements show that integrating basic VMI primitives into the hypervisor can reduce the performance overhead and help to make VMI applicable in production environments. Hence, there is ample room for performance improvement in the current Xen mechanisms for VMI that needs to be addressed by future research. To further improve the performance of synchronous VMI-based monitoring, we, therefore, suggest that more VMI monitoring primitives should be implemented in the Xen hypervisor. Nevertheless, by adding VMI-based functionality to the hypervisor, an additional access control mechanism should be added to the hypervisor. Additionally, if code is loaded to the hypervisor, it must be verified that it does not affect the reliability and security of the overall system.

Related Work

The field of dynamic control flow instrumentation has a long history and in the last decades has been many different approaches that tackle the performance overhead. In the following, we want to discuss the most important related approaches. DTrace [START_REF] Cantrill | Dynamic instrumentation of production systems[END_REF] is a tracing framework for the Solaris operating system. One of the main objectives of DTrace is to have no impact on the performance when the tracing is not implemented. This is implemented by inserting no-operation (nop) instruction at possible probing points. When the monitoring is activated, the nop instructions are replaced by jumps to the analysis routines. The monitoring actions can be implemented by users in the D programming language.

Our prototype approach follows the same approach as the eBPF filters in the Linux kernel [START_REF] Fleming | A thorough introduction to eBPF[END_REF][START_REF] Mccanne | The BSD Packet Filter: A New Architecture for User-level Packet Capture[END_REF], which have a similar concept as DTrace. The core concept of eBPF is to run application-specific tracing code in the high privileged Linux kernel in order to monitor functions in user and kernel space. This approach helps to minimize (synchronous) context switches to the user space that stalls the execution. The features of eBPF filters are very advanced and the Linux kernel is already able to run small programs, whereas our prototype for VMI-based tracing currently only allows pre-filtering tracing events in the Xen hypervisor.

Tuzel et al. [START_REF] Tuzel | Who watches the watcher? Detecting hypervisor introspection from unprivileged guests[END_REF] analyzed the performance impact of VMI-based monitoring techniques in detail. The focus of their paper is to analyze to which extent VMI-based monitoring techniques can be detected from the production virtual machine. The result of their conducted study is that VMI-based monitoring in their setup is not stealthy to applications running in the production virtual machine. The focus of our paper is to decrease the monitoring overhead to make it applicable in production environments but not to make VMI-based monitoring completely stealthy.

Klemperer [START_REF] Klemperer | High-performance memory snapshotting for real-time, consistent, hypervisor-based monitors[END_REF] proposed to use copy-on-write-based snapshots in order to perform the analysis on a non-changing system state. For this purpose, they extended the KVM hypervisor and hook instructions that alter memory pages during the normal execution. If a page is modified they copy the original version of the page for the snapshot. After the analysis is finished, the snapshot gets deleted and changes to memory pages are not monitored anymore. This approach is probably most effective when larger parts of main memory must be analyzed. For the extraction of function call parameters when a breakpoint is invoked this approach is too expensive.

Drakvuf [START_REF] Lengyel | Scalability, fidelity and stealth in the DRAKVUF dynamic malware analysis system[END_REF] is a VMI framework that uses libvmi and Xen and is mainly designed for dynamic analysis of Windows malware. It uses the altp2m approach [START_REF] Lengyel | Stealthy monitoring with Xen altp2m[END_REF] to implement software breakpoints. The advantage of this approach is that instead of replacing the original instruction with the INT3 instruction in memory, it creates a new memory page with the software breakpoint. Depending on whether a process in the production virtual machine is reading from that page or executing an instruction the memory page is swapped in the page tables. The rVMI framework designed by Pfoh and Vogl [START_REF] Pfoh | rVMI -A new Paradigm for Full System Analysis[END_REF] is similar to Drakvuf. However, it uses a patched version of KVM and Qemu instead of Xen3 . The communication between the VMI application and Qemu is established via the QMP interface of Qemu.

Westphal et al. [START_REF] Westphal | Vmi-pl: A monitoring language for virtual platforms using virtual machine introspection[END_REF] define a VMI monitoring language that supports the most common methods for VMI. Their prototype is implemented for the VMware KVM hypervisor. Similarly to our proposed approach, they execute VMI analysis scripts in the domain of the KVM hypervisor.

Bushouse et al. [START_REF] Bushouse | Furnace: Self-service Tenant VMI for the Cloud[END_REF] implement VMI-based monitoring into Linux containers running in the Dom0 of Xen instead of running the VMI-application in a monitoring virtual machine. This should not have any positive beneficial impact on the performance.

Conclusions

By implementing a pre-filtering approach in the hypervisor, we significantly reduce the impact of VMI-based monitoring on the production virtual machine. Our prototype reduces the monitoring overhead for process-bound monitoring by a factor of 18. We, therefore, conclude that more complex event processing in the hypervisor, such as the implementation of a breakpoint mechanism, can help to solve the problem of minimizing the performance impact of VMI-based tracing.

Fig. 1 :

 1 Fig. 1: VMI architecture using Xen and Libvmi. The arrows show the control flow of VM Exit/Entry-based context switches when a sensitive instruction in the production VM triggers a VMI event. The numbered items illustrate that synchronous tracing with breakpoints requires two iterations, the first to intercept the breakpoint and single-step the original instruction, the second to re-insert the breakpoint and resume the production VM.

Fig. 2 :

 2 Fig. 2: Our prototype architecture that pre-filters CR3 events in the hypervisor

The approach of manipulating the page tables uses the altp2m feature of Xen[START_REF] Lengyel | Stealthy monitoring with Xen altp2m[END_REF]. This approach does not modify the original memory page. Instead, it creates a new memory page that includes the software breakpoint. By switching the content in the memory tables it can activate/deactivate the breakpoint.

https://github.com/akheron/jansson

Currently, the KVM hypervisor does not support synchronous VMI-based monitoring out of the box. To use synchronous VMI-based monitoring mechanisms, the KVM hypervisor must be patched

Acknowledgment

This work has been supported by the German Research Foundation (DFG) in the project ARADIA (RE 3590/3-1).


Gerhard Goos

Elisa Bertino

Wen Gao

Bernhard Steffen

Gerhard Woeginger

Anne Remke

Valerio Schiavoni

Foreword

The 15th International Federated Conference on Distributed Computing Techniques (DisCoTec 2020) took place during June 15-19, 2020. It was organized by the Department of Computer Science at the University of Malta, but was held online due to the abnormal circumstances worldwide affecting physical travel.

The DisCoTec series is one of the major events sponsored by the International Federation for Information Processing (IFIP). It comprises three conferences: Together, these conferences cover a broad spectrum of distributed computing subjects, ranging from theoretical foundations and formal description techniques to systems research issues. As is customary, the event also included several plenary sessions in addition to the individual sessions of each conference, that gathered attendants from the three conferences. These included joint invited speaker sessions and a joint session for the best papers from the respective three conferences.

Associated with the federated event, two satellite events took place:

• The 13th International Workshop on Interaction and Concurrency Experience (ICE 2020) • The First International Workshop on Foundations of Consensus and Distributed Ledgers (FOCODILE 2020)

I would like to thank the Program Committee chairs of the different events for their help and cooperation during the preparation of the conference, and the Steering Committee and Advisory Boards of DisCoTec and their conferences for their guidance and support. The organization of DisCoTec 2020 was only possible thanks to the dedicated work of the Organizing Committee, including Davide Basile and Francisco "Kiko" Fernández Reyes (publicity chairs), Antonis Achilleos, Duncan Paul Attard, and Ornela Dardha (workshop chairs), Lucienne Bugeja (logistics and finances), as well as all the students and colleagues who volunteered their time to help. Finally, I would like to thank IFIP WG6.1 for sponsoring this event, Springer's Lecture Notes in Computer Science team for their support and sponsorship, EasyChair for providing the reviewing framework, and the University of Malta for providing the support and infrastructure to host the event.

June 2020

Adrian Francalanza

Preface

This volume contains the papers presented at the 20th IFIP International Conference on Distributed Applications and Interoperable Systems (DAIS 2020), sponsored by the IFIP (International Federation for Information Processing) and organized by the IFIP WG6.1. The DAIS conference series addresses all practical and conceptual aspects of distributed applications, including their design, modeling, implementation, and operation, the supporting middleware, appropriate software engineering methodologies and tools, as well as experimental studies and applications. DAIS 2020 was meant to be held during June 15-19, 2020, in Valletta, Malta, as part of DisCoTec, the 15th International Federated Conference on Distributed Computing Techniques. However, due to the COVID-19 pandemic, the organizers decided to turn the conference into a virtual event to be held completely online.

There were 30 initial abstract registrations for DAIS, which were then followed by 17 full papers. Each submission was reviewed by up to three Program Committee (PC) members. The review process included an in-depth discussion phase, during which the merits of all papers were discussed by the PC. The committee decided to accept ten full papers, one short paper, and one invited paper.

Accepted papers address challenges in multiple application areas, including system support for machine learning, security and privacy issues, experimental reproducibility and fault-tolerance, as well novel networking approaches for future network generations. Researchers continue the trend of focusing on trusted execution environments, for instance in the case of database systems. Instead, we notice fewer research efforts devoted to blockchain topics.

The virtual conference, especially during these last months full of unpredictable events, was made possible by the work and cooperation of many people working in several committees and organizations, all of which are listed in these proceedings. In particular, we are grateful to the Program Committee members for their commitment and thorough reviews and for their active participation in the discussion phase, and all the external reviewers for their help in evaluating submissions. Finally, we also thankful to the DisCoTec general chair, Adriano Francalanza, and the DAIS Steering Committee chair, Rui Oliveira, for their constant availability, support, and guidance.

June 2020

Anne Remke Valerio Schiavoni