Blog and maybe other stuff

   One of the key principles of economics is supply and demand. The pricing of a good is directly related to the amount available and the desire to purchase the good. If there are too few of an object that is highly desired, the provider can charge a higher price. Were as, if there is too many of an item, the provider would want to lower the price in order to make the item more buyable. Amazon's spot instances apply this principle.

   Amazon owns a vast amount of server farms with many different types of servers. So many that they have some instances sitting idle, doing nothing. If they remain unused, Amazon has no gain from these servers. Nor is there a public need for these servers, otherwise they would be in use. So their value, based on the supply and demand, is very low. Instead of wasting these machines, Amazon decided to make them available to the public at a different price scale.

   Normally, when renting an ec2 instance from AWS, there is a set price you pay for on demand access to an instance. The better the criteria of the hardware, the more it costs. Amazon EC2 Spot Instances is a system for bidding on Amazon's spare instances. According to the EC2 Spot website, "Since Spot instances are often available at a discount compared to On-Demand pricing, you can significantly reduce the cost of running your applications." The way the pricing for spot instances works is that you set a cost you are willing to pay for a type of instance. If the price for that instance is lower than your bid you are able to get one. But you don't pay what you bid; you pay what the instances are valued at, based on the pricing of each hour.

   The main reason to use spot instances is that you get the same instances for a cheaper or equal price. If you are debating between getting an on demand or a spot instance and decide to go with a spot instance, it makes most sense to bid at least the value of the on demand instance. First of all, you were willing to pay that much to begin with if you had gone with an on demand instance. Secondly, it gives you more leeway if the price goes up. From my experience, the value of the spot instances is usually around 20 percent of the on demand price. So, if the price rises a little, you don't have your instances taken down. Thirdly, you most likely won't pay the on demand price anyway. Like stated earlier, you don't pay what you bid; you pay the value of the spot instance. Finally, if the value does go over the on demand price, your instances will be taken down, because you only get the instances if they are valued below your bid. To counteract this, you can always bring up an on demand instance. Therefore, you will never be paying more than what you would have paid for an on demand instance.

   One of the pains of spot versus on-demand instances is that Amazon will shutdown the spot instances when the value exceeds the bid. The processes that run on the spot instances have to be able to deal with a shut down at anytime. This is why Amazon recommends spot instances run temporary or short running processes. If the process is short lived then there is less time to be affected by a price change and less of a threat or risk of being taken down. Otherwise, AWS gives a two minute warning for when they are going to shut down an instance. Unfortunately, the best way to listen to this warning is to have a separate process querying amazon every X minutes to see if the instance is marked as going down. If the processes require zero down time, it is recommended to use dedicated instances, either instead or in addition. It can take minutes to bring up new instances, so spot is unreliable for zero down time.

   Amazon provides a very helpful tool for managing your spot instances, called spot fleet. Spot fleet has a view of the prices of the different kinds of instances. It also allows for diversification and scaling of instances in groups. Diversification is very important because it makes having spot instances available more likely. There are two means of diversification.

   The first kind of diversification is having multiple kinds of instance types. Choosing different types of instances lowers the likelihood that all your instances will be taken down at the same time. If only one instance type is selected and then that instance type becomes in high demand, all the spot instances may be taken down. If multiple instances types are selected, if one of the instance types becomes in high demand, those instances will be scaled down and another type can be brought up in its stead. Furthermore, there are enough instance types similar or higher in performance from different EC2 families (i.e. m, r, t) that you can swap them out, almost insignificantly.

   The other type of diversification is by availability zones. It is possible that all the instances of one type in one availability zone will be taken. It is less likely that all the instances of that type in all availability zone in the region will be unavailable. Specifying multiple availability zones increases the chances of scaling up another instance in the region and maintaining the amount of instances desired. By utilizing both types of diversification, the likelihood of having no spot instances available is very low.

   Spot instances can also be put in load balancers for serving requests. In a normal situation of an on demand instance, the instance would be put in an autoscale group and then into the elastic load balancer. Spot fleet can be used as the autoscale group. Each fleet is a group of instances that can be diversified and has a minimum, maximum, and desired capacity for scaling purposes. These instances can be grouped together with an EC2 group name, and not an autoscale group name because they are not actually in an autoscale group.

    Scalability is a hard task to accomplish. To attain constant and consistent availability and accessibility. Very rarely can a database offer such amenities on a large scale. In some circumstances, just storing the data in a database is not be enough. What we tried to provide was a scalable solution for accessing our data. What we ended up getting was an analytics system.

   The first step for us achieving scalability was changing databases. We had most of our data stored on MongoDB. The way MongoDB was sharding our data sets could not support our large read and write throughputs. One of the reasons we migrated to DynamoDB was because its throughput is configurable. We could scale the throughput as we needed it. Switching to DynamoDB gave us more control over the availability of our data.

   The second step was to make the data more accessible. Besides for storing the data in DynamoDB, we decided to cache the data in MySQL. This way we would hit DynamoDB less and need less provisioned throughput. In turn, it would be cheaper because DynamoDB charges by throughput capacity. But besides the money benefits, we could prevent throttling because we would be hitting single shards too often.

   The problem with storing the data just in DynamoDB is that newer data tends to have similar keys and get sharded together. The way DynamoDB shards the data is by the indexed keys (makes sense, doesn't it). This causes the newer data, on single shards, to get hit more often and causes throttling. For instance, when a user registers they need to set or get their Glide contact list. This causes many hits on that user's contact documents, referenced by his personal ID. As opposed to an older user, whose shard does not get hit as often because they are not using the app as much or aren’t uploading as many contacts at one time. Since the throughput is divided throughout all shards, much of the provisioning is wasted on the cold data. Therefore, we aimed to separate the newer data from the older data, so we could keep our throughput low and have the data still be accessible.

   Whenever we would insert or update a record in DynamoDB we would insert a record in MySQL. The Mysql tables that we would cache the hot data in would be created in advance by a script running on a singleton every hour. Usually it would create tables to be ready for up to a day in advance, which would host an hour’s worth of data in each. But the script was configurable for how many tables and their time-buckets. There was one table for inserts and one table for updates per time-bucket. After the time-bucket passes, the data would be considered cold.

   When the cold data tables were no longer in use, we would send the data to Amazon’s S3 for starage. Then, using Amazon Redshifts copy command, we would insert the records from the insert tables into Redshift. Afterword, we would apply all the updates in the update tables to the records in Redshift. This enabled our analytics department to have a near duplicate of the production dataset within a reasonable amount of time.

   There were several shortcomings with this system. The first is that it created two points for inconsistency. We didn’t want to slow down the requests, so we would run the queries to DynamoDB and MySQL in parallel. If we successfully inserted into DynamoDB, there was still the possibility that we would fail to insert it into MySQL. Similarly, we could succeed at creating the record in MySQL and get throttled on DynamoDB’s insert. This made it very unreliable for a caching layer for production use.

   Secondly, we were caching a noSQL schema in a SQL database. If we added a field in DynamoDB we needed to create new tables or we would miss data. We could not just add  the column because that locks the table and would cost time on the requests. We could add the column and wait a day to start using that field. Or, we could delete the tables that were created in advance and have the script create new ones on the next run with the new schema. Either way, it’s a hassle to maintain.

   Thirdly, Redshift doesn’t have a restriction on duplicate keys. We could have the same data over and over and not even know it. For instance, if we would have a record inserted in one time-buckets table in MySQL and then insert it again in another time-bucket, for whatever reason, it would show up twice in Redshift. Or, if we ran the copy command twice on the same file, the database would have no problem holding the same data twice. This was a third possibility for inconsistencies and could ruin analytic data.

   Although we did have this hot table system, we never implemented using it as a caching layer. It only acted as a means for getting analytic data. We would keep adding and removing fields based on whatever the analytics department felt they needed at the time. The duplicate code for inserts and updates was cluttering up the code. Furthermore, recently there has been a goal of cutting server teams expenses. These tables in MySQL were just duplicate data that we already had in DynamoDB, so it had no real reason to exist. And then Amazon announced their DynamoDB streams feature.

   With DynamoDB streams we could get item inserts and updates in near real time and insert it into Redshift, or S3, right away. A DyamoDB stream is, “an ordered flow of information about changes to items in an Amazon DynamoDB table.” Whenever a document in DynamoDB is inserted or updated a new record is inserted into the steam. This record contains the action taking place and the data that is being inserted or updated. We use Amazon Kinesis Firehose to going over each of our table’s streams and insert and update the data in Redshift.

   Every company needs to know how it is performing. As a small company, we only needed a simple tool to see how the app was being used. As the company grew, this was not enough. We needed a better understanding of our users' actions. The following is the evolution of our analytics system.

   My first task as an intern with Glide was to create a way for our founders to get a better idea of our metrics. It had to be easy to understand and be up to date; with frequent updates. So I created the hourlies, to gather and display our metrics. The hourlies was a fairly simple system. It was an aggregation of the events that occurred, grouped by hour.

   On every action done by the server or background processes, we would increment a counter. If we needed to have actions that were unique per user, we would add the userId to a set. Then, when aggregating, we would get the length of the set for the count. We chose to store the counters in Redis because its increment operations are atomic. The structures for the counters was a hash with keys specific for each hour and a field for each action. The sets also had a part of the key specific for the hour of the actions performed. Redis's set data structure allowed for efficient storage of unique userIds and getting a count of them when aggregating. Also, because Redis is all in memory, it would be quick to do many insert and increment operations and not hider the processing speed.

   Then, on the hour, plus five extra minutes for insurance, a singleton process would iterate over the last hours' data and aggregate them. One of the lessons we learned was to not only go over the last hour but previous hours also, in case we missed some data for whatever reason. After we were done with each hour's aggregation, we would use Redis's increment operation, but use a negative value, to decrement. We decremented and not deleted the hour's counters because one time there was a mistake and we had two singletons running, an oxymoron I know. The two processes running the aggregation script got the same data and therefore incremented twice. By decrementing instead, we would still have double the value, but only until the next hour. When the script would run again the next hour, it would get a negative value for that hour from the Redis counter and increment that negative number to the hour that was doubled. Finally, after seventy two hours we would delete the hash; confident that we processed that data.

   The database we decided to store the permanent aggregated objects was MongoDB. The Nosql database structure allowed us to add and remove actions whenever we saw fit. Furthermore, the JSON structure of the queries made it extremely easy to compose the upsert queries from the Redis data. If we wanted an object in MongoDB, we would just add a "." to the field name in Redis and this would become an object with the fields in JSON. Also, like Redis, MongoDB's increment operation is atomic which we would use in the upsert, which is a term for "insert or update". An insert would overwrite the data and an update when the data doesn't exist wouldn't work. The simple change of a boolean, setting upsert to true, made the storage easy. For the CRUD operations, MongoDB was great for insert, but it was also great for reading.

   MongoDB also has a great way for storing documents in time-buckets, meaning sorting each document into one or more time slots. The way to do this is have an index on an array of strings. MongoDB’s documentation writes, "to index a field that holds an array value, MongoDB creates an index key for each element in the array." Therefore, we can use each string in this array to represent a "bucket" that the hourly should be counted in. To make it time relevant, we stored an element in the array for the year, month, day, and hour that this hourly was aggregated for. For example, in order to search for everything aggregated in the current year, it would already be indexed, and easily retrieved by querying where the array field equals the string we have made to represent the year bucket. (The comparison is not intuitive, but it works.) These time-buckets allowed our front-end graph generator to sort the data in order to display it by hour, day, month, or year; whatever the person wanted, with a fast response rates.

   But, as the company grew, the system needed to grow with it. We needed more actions with more granular data. We needed to know who was doing what, when, how, why. The simplicity of how often and how much was not enough to make the educated decisions that needed to be made. So in parallel with the hourlies, we started using AWS's Kinesis service.

   Now, for every action we would write a line of JSON to a file. The JSON could contain anything we needed it to, similar to the reason we used nosql. Each line would be required to have an action number and an action time. This way we would know what was happening when. Any additional information could be used for grouping. For instance, client type, client version, or the user's registration date. Other information could give us ideas about how each group of users use the app. By providing recipient ids, or chat ids, on message actions, we can see what kind of people our users like to talk to, or how often to each person do users contact, or frequency of response. By using Kinesis, we had the opportunity to learn a lot more about our application.

   We had to get our actions to the Kinesis queue, but Amazon charges per kinesis upload in units of twenty five kilobytes. In Amazon's pricing example they state, "As our record is 35KB, each record contains two PUT Payload Units". Even though the put request didn't contain a full fifty kilobytes, it still counted as two units. Those missing fifteen kilobytes were being paid for, but not utilized. Therefore, we wanted to be batch the requests into multiple of twenty five kilobyte chunks to save money.

   The way we accomplished this was by appending each action line to a file in a designated folder for Kinesis. Each file name would have the current second in it, generated from a unix timestamp, so that the file would never become too big. This assumes we don't do one megabyte of actions in one second on one machine. If a file for that second didn't exist, the file would be created when writing to it. Then, on each instance which wrote actions, such as the servers and background processes, there would be a process to send the data to Kinesis.

   The goal of this process was to combine the individual files for the seconds into a single file about fifty kilobytes, without exceeding, and to upload them the the Kinesis stream. This process would first list the file names in the kinesis directory. For each of the file names returned, it would get the size of the data in that file. It would check if the data size added to whatever data is currently in the upload file would be less than fifty kilobytes. If it would be greater, we would then make a request to Kinesis to put the upload file in the stream, and erase the upload file on success. On failure, we would try again to add a seconds file to it and find that it was too big and retry. Finally, when the upload file was cleared, the process could continue to create an upload file and append to it until it got to fifty kilobytes.

   Just like Redis, the Kinesis stream is also ephemeral. On the Kinesis pricing page it is written, "Amazon Kinesis stores your data for 24 hours by default. You can choose to increase the data retention period of your stream to up to 7 days." Seven days of retention does not give much time for analysis, (especially if we would have to iterate over the entire stream whenever we wanted to do it, which would be a ridiculous idea in the first place). Therefore, we needed to put the data somewhere more permanent, like we did in MongoDB with the hourlies. Somewhere we could query a lot of data in a short amount of time.

   Amazon's Redshift database is an SQL columnar data warehouse. Being an SQL database means that we can relate our actions to each other. This makes it easy to store and analyze the data. A columnar database enables it to access the data we want, instead of getting all the data and then discarding what is unnecessary. Like Wikipedia says, "by storing data in columns rather than rows, the database can more precisely access the data it needs to answer a query rather than scanning and discarding unwanted data in rows. Query performance is often increased as a result, particularly in very large data sets." Redshift is also made to hold a lot of data, their pricing page even advertises scaling for petabytes of storage. Our analytics team would greatly benefit from the speed of a columnar database, the size allowance of a data warehouse, and the convenience of being able to use SQL.

   The way we get the data from Kinesis to Redshift is through a Kinesis consumer. A Kinesis consumer is a process that iterates over a Kinesis stream. Our consumer inserts each record into Redshift based on the action in the record. The actions dictate which table to put the record in. If the record has an action that has to do with messages, it goes in the message actions table. If the action is a thread action we put it in the thread actions table. So, whenever we want to create a new action, we need to associate it to a Redshift table, otherwise our Kinesis consumer will disregard it. Now that we have our actions in Redshift we can do all the analytics we want.

    This month Glide is focused on saving money. One of the ways is by moving off of Mysql which we have hosted on Amazon. We have a whole system for A/B testing invite text copies using Mysql. The task I was placed with was to move that system to any other service that will save us money. I decided to go with Amazon’s DynamoDB. DynamoDB has certain limitations which required some additional systems, like S3 and redis. And even with those other services, we are still saving money.

    Migrating the texts over from Mysql to DynamoDB was fairly simple. The Ids were simple numbers, so they were able to stay the same as the Hash for each document. Any other field we queried by got a global secondary index, namely the languageCode. In Mysql we were storing a bunch of boolean values to dictate which types of invite each text would apply to. Mysql being a sql database meant that we needed to store false and null values. Since DynamoDB is NoSQL, we didn’t have to migrate any of the false or null values. This meant each text entry was smaller and easier to view. So the migration went over smoothly and all the text copies were now in dynamo.

    In the old system, using Mysql, our servers would query every five minutes for all the texts. It would then put the invites into a JSON object. This object maps the languages, as the keys, to arrays for the different types of invites. We then iterate over each text and push it into each invite type that applies to it. This way, when a client comes with a specific language for a specific invite we can grab from the array one of the copies to send. We could very well use the same system using the JSON object for our new system because it doesn’t harm anything. But querying from DynamoDB is much different than querying from Mysql.

    DynamoDB uses a throughput system where it rations reads and writes that are set by the user. From Amazon’s DynamoDB pricing website, “Amazon DynamoDB lets you specify the request throughput you want your table to be able to achieve (your table’s “throughput capacity”). Behind the scenes, the service handles the provisioning of resources to achieve the requested throughput rate. Rather than asking you to think about instances, hardware, memory, and other factors that could affect your throughput rate, we simply ask you to provision the throughput level you want to achieve and we handle the rest.” What they don’t tell you here is that if the throughput get used up, Amazon will throttle your requests and return an error if it does not succeed on their retries. The query for getting all the texts would require a full table scan, running from multiple machines and processes. This will utilize a lot of throughput with no guarantee that we will not be throttled. This problem of throughput and the fact that this table is not being updated often gave reason for a caching layer.

    We decided on Storing a JSON file on Amazon’s S3 that will have all the enabled texts. With S3’s reliability, we are not concerned with all the processes making requests for this one file. Also, with the speed from within Amazon, from our ec2 instances, it is very fast. All that remained for fetching the data was to know when to update the file on S3.

    Redis, an in memory data store, was the perfect way to check quickly if an update was needed. Whenever the DynamoDB table would be updated, it would update a key in redis with the current time. A singleton process would check every five minutes if the value is greater than the time it last uploaded the texts to S3. If the DynamoDB update time was more recent, the process would scan the entire texts table once with a scan filter where enabled is true. The process would append all texts returned to a local array. It may have had to make several requests to dynamo to build up this array, but that was fine because in the meantime the servers would use the current file on S3. When the process finished scanning the table, it would then upload the JSON array as the body of the S3 file.

    On the callback from S3, it would update another key in Redis for the upload time of the S3. This S3 upload time would be used on startup of any new singleton, so that we don’t upload to S3 even on startup if it is not necessary. The processes that use the texts also check the Redis key that is updated when there is an upload to S3 every five minutes. If the time in redis is more recent than the last time they fetched then it indicates to them that they need to get the file from S3 and update their local texts object. Checking Redis and fetching from S3 solved DynamoDB’s throughput problem. But there were other challenges that needed overcoming.

    Creating new texts posed its own issue. Before we had texts on Mysql, we had them on MongoDB. MongoDB didn’t scale well enough for our other purposes, so the company wanted to get off of it altogether. Something that remained from when we were on mongo was how we generated ids. We needed small ids because we were appending them to the invite text to know on clickthroughs which text was being affective. So, we decided that we would increment a counter for the ids. In Mysql you can simply have an auto incrementing id. But, since we came from MongoDB the way we were doing it was a bit convoluted. We would fetch the last message, sorted by id, and increment that by one and use it for the new text’s id. DynamoDB on the other hand doesn’t have a query like this. One cannot query by id to get the biggest. One would have to scan the entire table and calculate locally which id is the greatest.

    Once again, it’s Redis to the rescue. We have one more key in redis to represent the current greatest text id. Then, when we want to create a new text document, we just increment this counter in redis with a single incr command, and the returned value is our new id. To start off the counter, we just needed to scan the table once and store that value. There is a condition on the update so that it checks that the id does not exist when being created. This is a failsafe, in case this Redis value gets erased or changed, so that we don’t overwrite existing texts and mess up the metrics. Because Redis is in memory, it is all non-volatile and means it can be erased and not be recovered. Of course, the script for the scan to find the greatest id is saved in case we need to run it again in this event.

    The goal was to get a bunch of data on S3 securely and efficiently. The zip command line tool had a convenient way of securing and compressing by including a password in the bash command line. Furthermore, AWS’s Lambda allowed us to process the data we need and not worry about a local server limitations that come with doing it locally (i.e. disconnecting from the internet, limited bandwidth, going from AWS dbs to local computer speeds). It was so simple of course there needed to be a problem.

    Using a node script we were spawning a child process for the zip command. The spawn command was “spawn('zip', ['-P', password, PATH + '.zip', '-j', '-r', filePath]);”. Simple enough; zip this file path into a file at PATH.zip with a password. But the child process kept reporting onError an error of “{ [Error: spawn ENOENT] code: 'ENOENT', errno: 'ENOENT', syscall: 'spawn' }”. The files we were trying to zip were in the /tmp folder. AWS Lambda gives access to the /tmp folder with a limited amount of space of 512MB, which we were definitely under. So that wasn’t part of the issue.

    An “ENOENT” error would seem to mean that it can’t find the path. However when the script was run locally it worked fine. So we knew that the script was pointing to the /tmp folder and that locally it could access it. We also tried to do other commands in the /tmp folder, like creating folders, writing files and listing files there, and we were not getting back any errors on these. Therefore this error must mean something else.

    The next logical step is for this error to mean that it could not find the zip command. This seemed very suspicious because most if not all linux and windows based machines have the zip command pre-installed. But, just to be sure, we got the instance ami that AWS Lambda runs from http://docs.aws.amazon.com/lambda/latest/dg/current-supported-versions.html, and we started up our own instance in us-east-1 to see if it has the zip command. And, or course, we found that it did. So we opened a ticket with the Amazon support team.