-
Notifications
You must be signed in to change notification settings - Fork 0
Attachment Streaming
- Goals
- Inspiration
- New Wire Format
- PDSC Modeling
- Writing API
- Reading API
- Chaining API
- Examples
- Good Programming Practice
- Additional Developer Notes
- Streaming Attachments without Rest.li
- Future Enhancements
Rest.li is the high performance platform on which web services operate at LinkedIn. As our company moved closer to the formal adoption of large unstructured blobs of data, such as media, we needed a highly performant way to move all this data around. Therefore it became apparent that we needed to perform an overhaul of our service-to-service architecture.
The goals of Rest.li attachment streaming therefore are the following:
- Provide the ability to pass large blobs of bytes around our data centers between multiple services seamlessly
- No one service should hold the entire payload in memory at once
- Fully asynchronous and event driven
- Zero copy write and read for high performance
- Leverage existing web services infrastructure, namely R2/D2/Rest.li
- Allow multiple blobs to be sent in a single request or response
- Establish a wire format that can foster a high adoption rate for external members and platforms
- Solve the immediate business need of our services interacting with our custom distributed object store – LinkedIn’s version of S3 – Ambry
- Provide clean and intuitive async APIs for our engineers
Rest.li attachment streaming is inspired by the Reactive Streaming Manifesto
Therefore attachment streaming in Rest.li, from the bottom (R2) to the top, is based on the following:
- Allow processing a potentially unbounded number of elements
- Ensure that data elements are handled in sequence
- Asynchronously pass elements between components
- Mandatory non-blocking backpressure from the bottom (TCP) up
- The reader should never be forced to buffer data
In order to support attachment streaming it became apparent that a modification to the current wire protocol was needed. This was for a number of reasons:
- Current wire formats are JSON and PSON – do not work well for expressing binary attachments
- Multiple attachments need to be supported
- Traditional JSON/PSON payload needs to be fully read in before attachments are read in
- Arbitrarily large attachments need to be supported
- Each attachment needs some metadata associated with it
- Desirable to have a wire format that is set up for easy adoption by external consumers and platforms
- Multipart/MIME best suits our requirements – fully conform to the RFC
POST /widgets?action=purge HTTP/1.1
Content-Type: application/json
{
"reason": "spam",
"purgedByAdminId": 1
}POST /widgets?ids=List(1,2) HTTP/1.1
Content-Type: application/json
X-RestLi-Method: BATCH_PARTIAL_UPDATE
{
"entities": {
"1": {"patch": { "$set": { "name":"Sam"}}},
"2": {"patch": { "$delete": ["name"]}}
}
}Note that the current wire format described above will still be supported as there are no backward incompatible changes being made.
However for clients sending requests with attachments present or for servers responding with attachments, the wire protocol will change.
If attachments are present in either a request or a response, the content type becomes multipart/related. If a client can handle attachments back from a server, then an accept type of multipart/related is also added to the Accept header.
For example:
PUT /widgets?ids=List(1,2,3)
HTTP/1.1 X-RestLi-Method: BATCH_UPDATE
Content-Type: multipart/related; boundary=--km6cltxBQgkYRIwT8lAgFGfNV0AmQFwDB
Accept: multipart/related; application/json
--km6cltxBQgkYRIwT8lAgFGfNV0AmQFwDB
Content-Type: application/json
{
"entities": {
"1": {
"widgetName": "Trebuchet",
"myVideo": "cid:725c0319-b1f1-4b9c-b618-7ee9468870f0"
},
"2": {
"widgetName": "Gear",
"myVideo": "cid:a4d4133b-0546-4f7b-8104-ffdd644168c6"
}
"3": {
"widgetName": "Slider",
"myVideo": "cid:725c0319-b1f1-4b9c-b618-7ee9468870f0"
}
}
}
--km6cltxBQgkYRIwT8lAgFGfNV0AmQFwDB
Content-ID: <725c0319-b1f1-4b9c-b618-7ee9468870f0>
binary data…..
--km6cltxBQgkYRIwT8lAgFGfNV0AmQFwDB
Content-ID: <a4d4133b-0546-4f7b-8104-ffdd644168c6>
binary data…..
--km6cltxBQgkYRIwT8lAgFGfNV0AmQFwDB--
Note that the regular Rest.li payload becomes the first part in a multipart/related envelope. Each attachment then becomes its own subsequent part separated by the multipart boundary. Since each part in a multipart envelope can have its own headers, the Content-Type header from the regular payload now appears as a header in the first part. Both JSON and PSON are supported as valid Content-Types for the first part.
Each part after the regular Rest.li represents a blob of data attached to the request or response. Each attachment part has a Content-ID header which uniquely identifies that attachment in the payload. References to this unique identifier should then be placed as fields in the JSON (RecordTemplate backing) payload as pointers to the blobs in the attachments. In this particular example we have two attachments. Trebuchet and Slider both point to the same attachment while Gear points to the other attachment.
PDSC modeling will not change for streaming, with the exception of the following recommendation. This recommendation is simply to serve as a visual cue and does not impact the generated RecordTemplates or the processing of a streaming request or response.
Our recommendation is that anytime you have a field in a PDSC referencing an attachment, that a key value pair of attachment is present. This should further be expanded to include documentation mentioning that this field represents a pointer to an attachment. Once again it is important to note that the purpose of these fields is simply to convey that an attachment could be present.
{
"type" : "record",
"name" : "Greeting",
"namespace" : "com.linkedin.greetings.api",
"doc" : "A greeting",
"fields" : [
{
"name" : "id",
"type" : "long"
},
{
"name" : "content",
"type" : "string",
"attachment" : true,
"doc" : "Type 1 UUID representing a video attachment"
}
]
}In terms of the actual data supplied at runtime we suggest using Type 1 UUIDs.
Technical Details on Type 1 UUIDs
The reason for suggesting Type 1 UUIDs is because it provides the best guarantee of producing a globally unique identifier. The is important since, as shown further below, attachments can be coalesced from different machines/services which may lead to an identifier collision.
In order to create an attachment, developers must implement the following interface(s):
/**
* Represents a custom data source that can serve as an attachment.
*/
public interface RestLiAttachmentDataSourceWriter extends Writer
{
/**
* Denotes a unique identifier for this attachment. It is recommended to choose
* identifiers with a high degree of uniqueness, such as Type 1 UUIDs.
* For most use cases there should be a corresponding String field in a PDSC
* to indicate affiliation.
*
* @return the {@link java.lang.String} representing this attachment.
*/
public String getAttachmentID();
}
You’ll notice this extends Writer which is defined as the following:
/**
* Writer is the producer of data for an EntityStream.
*/
public interface Writer
{
/**
* This is called when a Reader is set for the EntityStream.
*
* @param wh the handle to write data to the EntityStream.
*/
void onInit(final WriteHandle wh);
/**
* Invoked when it it possible to write data.
*
* This method will be invoked the first time as soon as data can be written to the WriteHandle.
* Subsequent invocations will only occur if a call to {@link WriteHandle#remaining()} has returned 0
* and it has since become possible to write data.
*/
void onWritePossible();
/**
* Invoked when the entity stream is aborted.
* Usually writer could do clean up to release any resource it has acquired.
*
* @param e the throwable that caused the entity stream to abort
*/
void onAbort(Throwable e);
}
The Writer class leverages a class called WriteHandle whose interface is as follows:
/**
* This is the handle to write data to an EntityStream.
*/
public interface WriteHandle
{
/**
* This writes data into the EntityStream. This call may have no effect if the stream has been aborted
* @param data the data chunk to be written
* @throws java.lang.IllegalStateException if remaining capacity is 0, or done() or error() has been called
* @throws java.lang.IllegalStateException if called after done() or error() has been called
*/
void write(final ByteString data);
/**
* Signals that Writer has finished writing.
* This call has no effect if the stream has been aborted or done() or error() has been called
*/
void done();
/**
* Signals that the Writer has encountered an error.
* This call has no effect if the stream has been aborted or done() or error() has been called
* @param throwable the cause of the error.
*/
void error(final Throwable throwable);
/**
* Returns the remaining capacity in number of data chunks. Always returns 0 if the stream is aborted or
* finished with done() or error()
*
* @return the remaining capacity in number of data chunks
*/
int remaining();
}
These are the essential interfaces to keep in mind when defining an attachment as they represent how your custom data source will be asked to produce both the metadata as well as the raw bytes for your attachment.
When it is time for a RestLiAttachmentDataSourceWriter to produce data, it will first be invoked on RestLiAttachmentDataSourceWriter#getAttachmentID(). Implementations should return a unique identifier that should be the same identifier placed in the strongly typed RecordTemplate payload as described earlier.
Next, the attachment will be invoked on onInit(WriteHandle). The provided WriteHandle is the object that will be used to perform the actual writing of bytes later, so implementations should save a reference to it.
Subsequently, at some point in time in the future, the attachment will be invoked on Writer#onWritePossible(). It is as this point that implementations should write raw bytes on WriteHandle#write(ByteString). The amount of times that the writer may write will be based on what is returned from WriteHandle#remaining. Once the number of writes remaining has been honored, then again at some time in the future the attachment will be invoked again on code>Writer#onWritePossible(). Then the attachment simply repeats the logic above. The Javadoc is clear about this behavior for developers to follow.
The size of the chunk written is up to the developer but keep in mind that the larger the chunks written, the more memory that may be used by the application at any given time. Furthermore, in order to minimize copies, developers should use ByteString#unsafeWrap(byte[]) to wrap byte arrays that need to be written out.
Reading attachment reading is performed via a RestLiAttachmentReader. This applies whether a client is reading a server’s response attachments or a server reading a client’s incoming request attachments. Reading attachments using the RestLiAttachmentReader is a multi-step callback driven process.
The relevant interfaces are the following:
/**
* Used to register with {@link com.linkedin.restli.common.attachments.RestLiAttachmentReader} to asynchronously
* drive through the reading of multiple attachments.
*/
public interface RestLiAttachmentReaderCallback
{
/**
* Invoked (at some time in the future) upon a registration with a {@link RestLiAttachmentReader}.
* Also invoked when previous attachments are finished and new attachments are available.
*
* @param singleRestLiAttachmentReader the {@link RestLiAttachmentReader.SingleRestLiAttachmentReader}
* which can be used to walk through this attachment.
*/
public void onNewAttachment(RestLiAttachmentReader.SingleRestLiAttachmentReader singleRestLiAttachmentReader);
/**
* Invoked when this reader is finished which means all attachments have been consumed.
*/
public void onFinished();
/**
* Invoked as a result of calling {@link RestLiAttachmentReader#drainAllAttachments()}.
* This will be invoked at some time in the future when all the attachments in this reader have been drained.
*/
public void onDrainComplete();
/**
* Invoked when there was an error reading attachments.
*
* @param throwable the Throwable that caused this to happen.
*/
public void onStreamError(Throwable throwable);
}
/**
* Used to register with {@link com.linkedin.restli.common.attachments.RestLiAttachmentReader.SingleRestLiAttachmentReader}
* to asynchronously drive through the reading of a single attachment.
*/
public interface SingleRestLiAttachmentReaderCallback
{
/**
* Invoked when data is available to be read on the attachment.
*
* @param attachmentData the {@link com.linkedin.data.ByteString} representing the current window of attachment data.
*/
public void onAttachmentDataAvailable(ByteString attachmentData);
/**
* Invoked when the current attachment is finished being read.
*/
public void onFinished();
/**
* Invoked when the current attachment is finished being drained.
*/
public void onDrainComplete();
/**
* Invoked when there was an error reading the attachments.
*
* @param throwable the Throwable that caused this to happen.
*/
public void onAttachmentError(Throwable throwable);
}
The process begins by registering a callback of type RestLiAttachmentReaderCallback as shown above with the provided RestLiAttachmentReader.
Reading of attachments is performed via a RestLiAttachmentReader
Begin by registering a callback with the RestLiAttachmentReader
This callback will be invoked for every single attachment encountered
Reading of a single attachment is performed via a SingleRestLiAttachmentReader
Register another callback with every SingleRestLiAttachmentReader to drive through it’s attachment data
actual client side and server side APIs
Before going into details, one note on terminology. There is a big overlap between Map and Object types (in general). In many languages, they are treated interchangeably. You can think of an object as a map, where map key is field name and map value is a field value. Moreover, JSON does not distinguish between the two. Thus, this document describes only how to define filters for Map data type. All definitions related to Map type also apply to Object type.
There are two types of structural filters users might want to express:
- Positive mask: select only specified fields from the object
- Negative mask: remove specified fields from the object
It is important to note that even though the Rest.li framework can process negative masks for negative projections, the generated client builders do NOT allow such negative masks to be created. Only positive masks can be created by the generated request builders and hence all projection use cases within LinkedIn use positive projections.
The documentation listed below here for positive and negative masks is provided simply for reference purposes. At some point in the future, it may be possible that the Rest.li framework supports negative projection.
Structural filtering can be expressed as a JSON or any equivalent in memory representation. Ability of expressing filters as JSON objects is beneficial because it can be easily processed by clients and is language agnostic.
Positive mask is a JSON object, with number 1 assigned to selected fields, for example:
{
"person": {
"phone": 1,
"firstname": 1,
"lastname": 1,
"current_position": {
"job_title": 1
}
}
}
is a mask, which selects only phone, first name, last name and current position’s job title from User Profile.
Negative mask is a JSON object, with number 0 assigned to selected fields, for example:
{
"profile": {
"phone": 0
}
}
is a mask, which hides phone number in the User Profile.
Syntax for array is the following:
"array_field": {
"$start": 10,
"$count": 15,
"$*": {
(...)
}
}
where $* is a wildcard mask, applied to every element in Array. $start and $count are optional fields which specify range of array that will be returned. First element of an array has index 0. Semantic of $start and $count is intuitive:
- $start specify first element of an array, which will be returned
- $count specifies how many elements, starting from $start will be returned
Specifying range in an array is treated as a positive mask e.g. it is equivalent to selecting elements of array using positive mask.
If mask contains only $start, then $count is implicitly evaluated to Integer.MAX_INT.
If mask contains only $count, then $start is implicitly evaluated to 0.
If entire array needs to be masked, then syntax is simply:
"array_field": 1
for positive mask or
"array_field": 0
for negative mask.
Syntax for the map is the following:
"map_field": {
"$*": {
/* mask for every values in the map */
},
"key1": {
/* mask for value of key1 */
},
"key2": {
/* mask for value of key1 */
}
}
}
where $* is a wildcard mask, applied to every value in Map.
If mask for map contains both wildcard $* and values for specific key e.g. key1, then mask, which will be applied to the value of key1 is a composition of wildcard mask and mask for key1. See next chapters for semantics of mask composition.
If entire value for a specific key needs to be masked, syntax is:
"map_field": {
"key1": 1,
"key2": 1
}
for positive mask and:
"map_field": {
"key1": 0,
"key2": 0
}
for negative mask.
Finally, if entire map needs to be masked, syntax is:
"map_field": 1
for positive mask and:
"map_field": 0
for negative mask.
Mask syntax defines meta-fields: $, $start and
{
"$$field": 1
}
is a mask, which selects only field with name “$field”.
Positive mask is used in situations where user wants to limit the results only to specified fields (most likely for performance reasons).
If positive mask is applied to an object, all fields not specified in the mask will be permanently removed from the object, leaving only fields specified in the mask.
Negative mask is used in situations where user wants to restrict access to some fields. One use case might be applying security policies on the server side, where certain confidential fields are removed from objects. Other use case is that client might want to fetch entire object excluding some specified fields e.g. for performance reasons – see REST API section for use case example.
If negative mask is applied to an object, all fields specified in the mask will be permanently removed from the object, leaving all other fields untouched in the object.
In case all fields of an object get removed by the negative mask, then the result is an empty object. Similarly, if object does not have any of fields specified in positive mask, the result is an empty object.
Two main use cases for mask composition are:
- Server side service implementation wants to apply various security policies on top of positive mask obtained from request; for performance reasons it is better to compose masks and apply them in one go, especially if object is big
- Assembly-like engine wants to compose multiple positive masks, send single request to fetch data and return result to requesters; with proposed mask composition semantics this would be possible
In terms of JSON representation, composition of 2 or more masks of same kind is just union of objects that JSON represents.
Composition of two positive masks is mask equivalent to sum of fields requested in separate masks.
The following table summarizes the semantics of composition of two positive masks (1 means that field exists in positive mask, – means that it does not exist in mask, v in result means that field exist in an object after applying mask, – means that field is missing):
| positive mask | positive mask | result |
|---|---|---|
| – | – | – |
| – | 1 | v |
| 1 | – | v |
| 1 | 1 | v |
The following table summarizes the semantics of composition of two negative masks (0 means that field exists in negative mask, – means that it does not exist in mask, v in result means that field exist in an object after applying mask, – means that field is missing):
| negative mask | negative mask | result |
|---|---|---|
| – | – | v |
| – | 0 | – |
| 0 | – | – |
| 0 | 0 | – |
For example, composition of two positive masks:
mask1:
{
"a": 1,
"c": 1
}
mask2:
{
"b": 1,
"d": 1
}
composition of mask1 and mask2:
{
"a": 1,
"b": 1,
"c": 1,
"d": 1
}
Case of array’s $start and $count meta-fields requires extra explanation. Since $start and $count are treated as a positive mask, for example selection of sub-range of array, composition of two ranges is smallest range that contains both ranges:
mask1:
"array_field": {
"$start": 15,
"$count": 20,
"$*": {
(...)
}
}
mask2:
"array_field": {
"$start": 20,
"$count": 30,
"$*": {
(...)
}
}
composition of mask1 and mask2:
"array_field": {
"$start": 15,
"$count": 35,
"$*": {
/* composition of masks from mask1 and mask2 */
(...)
}
}
another example, when ranges are disjoint:
mask1:
"array_field": {
"$start": 10
"$count": 5,
"$*": {
(...)
}
}
mask2:
"array_field": {
"$start": 20,
"$count": 5,
"$*": {
(...)
}
}
composition of mask1 and mask2:
"array_field": {
"$start": 10,
"$count": 15,
"$*": {
/* composition of masks from mask1 and mask2 */
(...)
}
}
Intuitively the output of composition of positive and negative masks is as if at first, positive mask was applied and after that, negative mask was applied.
The following table summarizes the semantics of composition of positive and negative masks (1 means that field exists in positive mask, 0 means that field exists in negative mask, v in result means that field exist in an object after applying mask, – means that field is missing):
| positive mask | negative mask | result |
|---|---|---|
| – | – | – |
| – | 0 | – |
| 1 | – | v |
| 1 | 0 | – |
Few examples:
mask1 (positive):
{
"a": 1,
"b": 1
}
mask2 (negative):
{
"b": 0,
"c": 0
}
input object:
{
"a": "value1",
"b": "value2",
"c": "value3",
"d": "value4"
}
result after applying composition of mask1 and mask2:
{
"a": "value1"
}
It is possible that simple mask (0 or 1) is composed with a complex mask, represented as an object.
Since negative mask (0) has the higher priority than positive mask, then result of composition of mask equal to 0 with any other mask is 0 e.g. composition of mask:
{
"a": 0
}
with
{
"a": {
"$*": 1
"b": 0,
}
}
is equal to:
{
"a": 0
}
The semantics of positive mask is: select this field and all it’s children. Hence, the following two masks are semantically equivalent:
{
"a": 1
}
{
"a": {
"$*": 1
}
}
In other words, positive mask is recursive. If positive mask was not recursive, then it would not be possible to express the following filter: "select field “a” and all it’s children" without prior knowledge of all available fields in field “a”.
Composition of positive mask with a complex mask propagates positive mask recursively e.g. composition of:
{
"a": 1
}
with:
{
"a": {
"b": 0
}
}
yields result:
{
"a": {
"$*": 1,
"b": 0
}
}
If complex mask already contains wildcard mask, then $*=1 is recursively pushed down to the wildcard of the wildcard e.g. composition of positive mask:
{
"profile": 1
}
with negative mask:
{
"profile": {
"$*": {
"password": 0
}
}
}
yields result:
{
"profile": {
"$*": {
"$*": 1,
"password": 0
}
}
}
The reason why pushing down the
It means that the following masks are equivalent:
{
"a": 1
}
and
{
"a": {
"$*": 1
}
}
Mask composition is commutative. It means that it doesn’t matter in which order masks are composed. However, application of masks, which are or were built using positive masks, on the object is not associative operation. Consider the following example:
mask1:
{
"a": 1
}
mask2:
{
"b": 1
}
input object:
{
"a": "value1",
"b": "value2"
}
If we first compose masks and apply mask obtained from composition on input object the result will contain both fields “a” and “b”. However, if we first apply mask1 on input object and then apply mask2 on the result, then final result will be empty object.
REST API uses syntax similar to LinkedIn public REST API extended with concept of negative masks. The need for concept of negative mask is required to accommodate the following scenario.
Let’s assume that User Profile contains a field “emails” which contains mailbox contents in it. Client might want to request User Profile data (without knowing exactly which fields are available) but definitely without “emails” field, which might be huge. The notion of not needing to know which fields are available allows backward compatible evolution of data model without modification of all clients. This use case might appear in mid-tier, where service retrieves objects, do calculations/modify them and return to requesters. It would be beneficial if mid-tier services were oblivious (to some extent) to data model evolution.
The reasons for choosing LinkedIn public API syntax in REST API, instead of encoding JSON are the following:
- It is concise, so filters can be written by hand when needed (for example, during development and testing).
- It is familiar to LinkedIn public API users.
- It is expressive.
Internally, filter expressions and JSON filter expressions will be parsed into the same data structure. This will allow composition of filters (for example, security policies). REST API will not be fully compatible with original LinkeIn public API syntax.
Proposed syntax will be used only in REST interface. Composition of simple masks can generate complicated masks. Pegasus will automatically generate correct expression and will be able to parse it. The only time, when developer will deal with projections syntax is during experimenting, debugging and so on.
For example, the following JSON positive mask:
{
"person": {
"firstname": 1,
"lastname": 1
}
}
would translate to the following expression:
:(person:(firstname,lastname))
Projections are passed as a value of ‘fields’ request parameter in URL e.g.
http://host:port/context/resource/id?fields=person:(firstname,lastname)
Please note that, as mentioned above, negative projections should not be used.
The syntax for array representation is the following:
"array_field": {
"$start": 10,
"$count": 15,
"$*": {
"field1": 1,
"field2": 1
}
}
will be represented as:
array_field:($*:(field1,field2),$start=10,$count=15)
The syntax for map representation follows similarity between Object and Map data structure. The syntax is the following, for mask on map:
"map_field": {
"$*": {
"field1": 1
},
"key1": {
"field2": 1
},
"key2": {
"field3": 1
}
}
}
will be represented as:
map_field:($*:(field1),key1:(field2),key2:(field3))
Java client should use helper classes in:
com.linkedin.data.transform.filter.request
to build and manipulate projections.