Skip to content

Attachment Streaming

Jump to bottom Edit New page
Karim Vidhani edited this page May 17, 2016 · 54 revisions

Contents

Goals

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

Inspiration

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

New Wire Format

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

Sample Traditional Rest.li Wire Format

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"]}}
   }
}

Wire Format for Attachment Streaming

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

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.

Creating Attachments

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 Attachments

Reading attachments is a multi-step callback driven process which allows developers to asynchronously walk through each attachment. It begins with a top level RestLiAttachmentReader and a SingleRestLiAttachmentReader for each individual attachment encountered. This applies whether a client is reading a server’s response attachments or a server reading a client’s incoming request attachments.

There are two callbacks involved, one for the RestLiAttachmentReader and one for the SingleRestLiAttachmentReader. The relevant interfaces are as follows:


/**
 * 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. At some point in time in the future, the RestLiAttachmentReaderCallback will be invoked on RestLiAttachmentReaderCallback#onNewAttachment(SingleRestLiAttachmentReader).

The process then continues for each attachment as developers must register a SingleRestLiAttachmentReaderCallback with the provided SingleRestLiAttachmentReader. Once registered, the SingleRestLiAttachmentReader can then be told to produce attachment data via SingleRestLiAttachmentReader#requestAttachmentData(). Once this is invoked, at some point in time in the future, the SingleRestLiAttachmentReaderCallback will be invoked on SingleRestLiAttachmentReaderCallback#onAttachmentDataAvailable(ByteString) representing the data for the reader to consume. Once the attachment data is consumed, another call may be made to SingleRestLiAttachmentReader#requestAttachmentData() thereby driving through all the data in that attachment.

The Javadocs provided for each class are clear and should provide technical details as to how to use each API. Additional features, such as attachment draining and exception handling, are described in great detail.

Chaining (Proxying) Attachments across services

Rest.li streaming supports the ability to proxy attachments meaning that:

  • A server can take an incoming request and send one or more of its attachments as a request further downstream.
  • A server can take attachments from a response to a downstream request, and then send them back to the original request.
  • Clients and servers can coalesce multiple attachments from different sources.
  • Useful for observer or authentication patterns.

Here is an outline of what they may look like across multiple services:

Add a custom footer

Clone this wiki locally