Scale By The Bay 2018: Scott Kidder, Video Access-Log Processing with Apache Flink
Recording: Scale By The Bay 2018: Scott Kidder, Video Access-Log Processing with Apache Flink
you so welcome thank you for joining me today my name is Scott Carey I'm a principal software engineer at MUX here in San Francisco actually just a block from where we are today so today I'll be sharing with you our video access log processing system that we've built using Apache flink so here's the agenda for the talk begin with a quick background on MUX just for context then I'll follow it up with a web video primer explaining how most web video is delivered then we'll cover some of the challenges with using a utility billing model for billing based on video usage and then we'll cover the details of the log processing system that we built at MUX and if we have time I'd love to take any questions that you might have so background on MUX in 2016 we launched MUX data product which is an analytic service for video it provides insights into quality of experience issues for video playback on websites or mobile apps so much data is very easily integrated into any video delivery application exposed as metrics like buffering playback errors and more so our experiences building MUX data and operating it I've led us to see that it's still very difficult to deliver Internet video well and provide a good experience so earlier this year we launched a MUX video which makes publishing internet video as simple as making a REST API call you just provide a ctrl for your source video and will automatically select the optimal video encoding settings and package it for delivery and we also provide no commitment utility billing which means you only pay for the video that you store in the amount of video they've actually delivered which is the subject of this talk it requires processing the x-axis logs to actually make that possible so web video primer so how does web video even work what is happening in the background when you visit twitch Netflix or YouTube and start playing back video what sort of requests are being made by your browser smart TV or mobile your mobile app so playback is always initiated by the video player most adaptive web video is delivered using a protocol called HTTP Live Streaming or HLS which was developed by Apple and initially released in 2009 so HLS relies on m3u8 manifests which were originally designed as just generic playlists but they've been extended to support video streaming as well so the sequence begins with your video player making a request for the master manifest the master manifest describes all the different renditions that are available for the video that you intend to play each rendition might have a different video resolution bitrate language or some other distinguishing feature the player then selects the optimal rendition for your current playback circumstances next the player requests the selected rendition manifest which again is a three-way playlist which includes a sequence of URLs for video chunks as well as their timing information to facilitate seeking within the video lastly the video player begins requesting video chunks so they can be decoded and played back if the player decides to change renditions perhaps because the network bandwidth is increased or decreased then it will request a different rendition manifest and resume playing a different set of video chunks so which systems are involved in this chain of requests so you begin with the playback device on the far left which makes a request for the manifest and the video chunks through the content delivery network of the CDN and in the event of a CDN cache miss it'll then for the request back to the the video origin so MUX performs just-in-time transcoding of video which means that in the case of a cache miss will actually read the high quality video mezzanine chunks from storage a request time transcode them and stream the transcoded bytes out to the content delivery network using HTTP chunked transfer encoding back to the client and cache them at the CDN to service future requests so this is very different from how most video services worked in the past they typically transcode all the renditions up front which is expensive both in terms of compute and storage especially for user-generated content that might not ever be played so in this case we're only transcoding and storing renditions for content that is actually accessed so as I mentioned earlier utility billing requires us to know exactly how much video is served we do that by processing the CDN access logs so what are some of the challenges of doing utility billing so it basically just means paying for what you use just like the electricity usage that you pay for at home its metered the bulk of your bill is based on what you use if you use more you pay more if you use less you pay less so applying this model to video is very appealing to people who are working maybe on a new project where they don't necessarily know upfront what sort of volume they're going to be dealing with and therefore don't want to commit to an expensive pricing plan upfront so in order to build based on video usage we need to enrich each video access log record with details about the corresponding video we do this by querying our asset index database we need information about the video chunk that was requested and attributes of the video rendition like the resolution bitrate frame rate etc we also need to be able to avoid falling victim to a thundering heard type scenario so CDNs do an excellent job of shielding our origin from requests for manifests and chunks in the case of viral content however the CDN will also provide us with a log record for every single request that was made whether it was absorbed by the CDN or came back to our origin so as our CDN log processing is processing this these huge spikes in logs we need to make sure that it doesn't negatively impact other internal services back at our origin during the enrichment process additionally the logs we receive from the CDNs are not ordered this is mostly an artifact of how the CDN s aggregate their own logs from their many points of presence around the world before making those logs available to us there could also be several hours of delay from the time when a manifest or chunk request is made and when it eventually is delivered to us so we calculate hourly usage details by adding the duration of video that was served for a chunk to a an hourly window which I'll cover later so these windows are scoped by customer account as well as by video asset so we can say for a given hour and actly how much video was served for a customer or for a specific video so let's look at the log processing system that we built at MUX so the first step is initiated by the act of washing video so video player makes a request goes to the CDN may or may not go back to the origin depending on if that content is cached with step 2 we actually get down to the details of handling logs so we've got in this diagram you can maybe see in the upper right there we've got log collectors which sit behind a load balancer that permits inbound connections on HTTP and syslog so many of our CDN partners will stream logs to us as they're generated additional CDN partners that we're currently trying to integrate with right now have api's that we can pull for CDM logs either way those two mechanisms of getting access logs result in logs being written to a kafka raw logs topic so our CDN log collectors are written and go most of the services that we run at MUX are written in go the log in Richmond app is really kind of the exception to the rule so our collectors run as doctor container docker containers in kubernetes cluster the collectors themselves require very little CPU or memory resources so we can scale them horizontally very easily and cost-effectively and they've got a simple task which is to parse the CDM log messages from either comma separated value format or JSON what however whatever format is used by the CDM into a pretty buff based message format that we use for representing all CDN access logs so the serialized protobuf messages are then written to a Kafka topic for raw log records so Kafka is probably something that you guys are all familiar with it's just a high-performance distributed log that's really ideal for storing access logs like this and as usage increases we can scale the number of Kafka brokers or the number of partitions associated with the raw logs topic 2 deal with increased write throughput requirements so next we'll introduce one of my favorite parts of the architecture is what's actually referenced by this talk which is the log enrichment app so from the Kafka raw logs topic those records feed into a flink login Richman app so we need to enrich the log records with details about the video asset and this can all be done asynchronously that's why the collectors are simple stateless services that get the log records quickly write them to Kafka and then we can asynchronously enrich them so the Apache flink Patchi flank is a platform for building stream processing applications in Java or Scala so this log enrichment app reads from the raw logs topic performs log enrichment and then writes the enriched logs to a second Kafka topic that has a much longer retention period so the raw logs topic we only retain those those records for seven days but are enriched logs we keep for much much longer on the order of many months to support debugging operations auditing as well as data mining so the the log enrichment flink application has its State distributed across a variable number of flink task managers and flink makes it very easy to scale the the amount of resources that you want to devote to a job simply by changing the degree of parallelism specified when you're submitting a job so actually I had to do this just this morning which I'll come to again later in the talk so enrichment consists of adding asset details network details geographic information about the client that made each manifest or chunk request so ask the details that are particularly relevant to billing include the source the source video duration and the rendition settings so the rendition settings also include the chunk target duration so HTTP Live Streaming requires that each video chunk have a target duration which varies depending on the the use case so live video generally has a smaller target duration in our case is about two seconds on demand content has a larger target duration which is like four seconds and that's because you have different latency requirements and so knowing the target duration helps inform the calculation of how many seconds of video were served for a particular chunk request so we also identify the network ASN or autonomous system number of the internet router that sat between the client and the CDN this is important because it helps us identify slow or poorly performing Internet routers and potentially perform dynamic CDN selection so we support multiple CDNs and identifying CDNs that are performing better or worse in different geographies helps us at the time of rendition manifest requests select the optimal CD ami as we're rendering that rendition manifest that has all those chunky urls we could choose to deliver those chunks through one CDN or another and those can have performance cost differences so our asset details are kept in a cockroach DB cluster that spans multiple data centers and the asset information about the video is mostly immutable the only thing that really changes over the life of an asset after it becomes ready is it transitioning from ready to deleted it the the video details generally don't change much and we want to avoid overwhelming our cockroach DB cluster with reads for data that doesn't change much so our fling cap when is performing log enrichment will first query a Redis in memory cache for asset details and only in the event of a Redis cache miss will it query our asset index which will then proxy that request to cockroach DB transaction support in flank and Kafka is also an important feature that we take advantage of so flink log enrichment app will periodically check point at state to HDFS so we do that every five minutes and if the blank app is restarted unexpectedly maybe one of the task managers crashes or one of the external services that we need to interact with return some sort of error code that we cannot recover from the flink app will then be resubmitted or restarted automatically and it will resume from that earlier checkpoint which results in us reading messages from an earlier position in the kafka of raw logs topic this leads to reprocessing of messages and messages being written again to the enriched logs Kafka topic but this time as part of a separate transaction so the transaction that was in progress when the flink app crashed will never be committed and therefore downstream consumers on that enriched logs topic will ignore those messages so downstream consumers only process committed Kafka messages this is important because we don't want to process enrich log records multiple times aka like double double billing or both you know double charging so we have those exactly once processing guarantees courtesy of flank and Kafka which is very cool and we also have the log enrichment app right performance metrics minute-by-minute to an influx DB database that drives many of our graph fauna dashboards for internal monitoring so the chart that's shown here is something that I added just this morning because we had a new customer that launched it's a it's a self signup service and so we can literally get bombarded with huge volumes of in incoming video being ingested as well as video being delivered without any warning so this huge spike in volume led to a huge influx of CDN access logs and we had to scaled a number of flink task managers to process these logs in order to keep up but it didn't result in any any change in the the calculations made for the purpose of billing made by the flanca it simply ran slower which is the desired behavior so lastly utility billing usage calculations so we've got a second flink app that reads these in rich log records from the enrich logs Kafka topic it uses a vent time processing to assign a log record to a one-hour window that covers the time of the request when it was made so we support up to three hours of allowed lateness this means that we don't consider a one-hour window either for calculations for across a customer environment or for a specific video asset we don't consider a one-hour window to be final until we've received a log record that is at least three hours later than the most recent edge of the window so this is essential to processing out of order and late CDM log messages and so Flinx support for watermarks and allowed lateness really is what makes this pretty easy to achieve and so once a one-hour window has been finalized we'll pass it on to our Postgres billing database so we have we maintain the total number of seconds a video served for each customer and for each video asset in a given hour and calculating billing totals or billing reports for our customers as is as simple as summing up the number of seconds of video served for a customer environment over the the given time range so in summary billing on video usage requires large volume processing large volumes of logs Apache flank and Kafka have made a great foundation for building scalable stream processing applications and their support for event time processing transactions and allowed lateness have been critical to the the work that we've done so thank you again for your time and I'd love to answer any questions that you might have [Applause] yes yeah yeah so your question was how did we determine the the right amount of time for a loud lateness three hours so that's based on the the current CD ends and we integrate with so they have a guarantee that they'll provide us logs within three hours so as we begin to integrate with more CD ends and maybe they don't provide as timely logs we'll have to expand that and the the implication there is that we won't be able to finalize those windows until much later yeah thank you actually because we're using a vent time processing it can take longer it can take longer than three hours we're not using processing time the the time is an artifact if link event time versus processing time so the the processing within the flink application can take much longer than three hours and still provide determinates deterministic results yeah so if link will maintain an accumulator that a window to accumulator that is good for a period until we see a message that is more than that allowed lateness in the future so it's not it's not relative to the current wall clock time it's relative to the time on messages that are flowing through the flink out yeah so we don't provide we don't support auto-scaling at this time so if we determine that we need more CPU or memory resources then we'll save we'll do an atomic cancel with save point on the fling cap which writes out the state to HDFS and then resubmit it with a different parallelism value maybe having it run on more or less task managers and flink will then automatically redistribute the state across the new task managers that's exactly right yeah so we went from you know certain number of task managers and we added added more task managers and just how to run across a larger larger cluster yes yes that's a good question so you could end up having soaring state within the the flink application is kind of discouraged because flank manages its own objects in memory and if you're then starting to kind of pollute that a little bit with your own cache then it could conflict with Flinx own memory management as I understand it and it's running a separate reddish Redis cache is cheap and easy so yes yes what do we do in the with the scaling of the log collectors yeah so we've got that's a good question so we've got a service called run scope that periodically tries to play video through our system and it provides a complete loop as far as like ingesting video playing video so we expect to see access logs for those those those video views so if we don't receive those video views coming through the log collectors and into the flink app then we'll we'll page yeah and then I'll get woken up but Ted I don't want to get work man yes yeah so with our CD ends we require them to individually tag each CD and access log record with a UUID so we'll try to do do D duping using the UUID that's provided in the CDN access log record