Reflection on Designing Distributed System

Just finished reading “Designing Distributed Systems: Patterns and Paradigms for Scalable, Reliable Services” by Brendan Burns. I enjoyed reading the book. I think this book is going to add value to people who has recently get their hands on docker and orchestration tools like kubernetes. It is going to empower those developers/devops/admins by helping them to learn more about containerization design patterns. This book is only 150+ pages and it discusses wide range of tools and techniques with hands on example but I would definitely recommend this book for the theory and discussion rather than the practical aspect of it.

Containers establish boundaries that helps to provide separation of concerns around specific resources. When the scaling is not requirement and application runs on a single node, but the application is being run only on a single node, containerising still is going to be better choice because not only it helps us to scale but also it can be used to organise and maintain better.

Sometime in an application, we have tasks that runs in the background and other tasks that are served to endusers. Often the tasks that serves endusers are more important. We can prioritise those tasks in our containers by scaling those containers in a node appropriately.

When engineering teams starts to scale, for better engineering performance, usually a team of developers comprises of six to eight people. When more engineers join, usually it increases the number of teams. A better engineering performance, maintainability, accountability and ownership structure can be implemented by assigning the responsibility of a containerised application to a team. Additionally, often some of the components, if factored properly, are reusable modules that can be used by many teams. 

Single node patterns:

  • Sidecar Pattern: Sidecar containers, are a group of containers that runs together and stayed together on the same node, and they share processes, ports and so on. It can be a very useful when dealing with legacy services or 3rd party maintained closed source application about which we have very little knowledge. In sidecar pattern, unencrypted traffic is only sent via the local loopback adapter inside the container group, that makes the data and communication between them safe.
  • Ambassador Pattern: In ambassador pattern the two containers are tightly linked in a symbiotic pairing that is scheduled to a single machine. Ambassador can be used as proxy server. In a sharded system, ambassador pattern can be used to route to a Shard. Ambassador can be used for Service Broker as well.
  • Adapters: Although every application is different, they still have some things in common. They need to be logged, monitored, maintained, etc. Adapter patter takes advantage of this use case and implemented in a sidecar pattern. This common use case can be modularized and use as sidecar container. Popular use case can be accessing and collecting desired logs, monitoring performance by periodic executing a command, etc via a container that runs as sidecar.

Serving Patterns:

Replicated Load-Balanced Services: In replicated load balanced services, the same data is being replicated across multiple server and a service is being used as a load balancer.

Sharded Services: When the size of data becomes huge for a single machine, the data is splitted and distributed across multiple machine. A router is placed in front of the service to identify which shard to route to to fetch that data. Usually a carefully chosen hashing key is used to identify where the data is located and which server to route to. For better performance and reliability, replication and sharding is used together.

Scatter/Gather: Scatter/gather is quite useful when there is a large amount of mostly independent processing that is needed to handle a particular request. Scatter/gather can be seen as sharding the computation necessary to service the request, rather than sharding the data.

When there is more data that a single machine can hold or process in time, it becomes necessary to introduce sharding by splitting the data across multiple devices. Then when it comes to querying, it is going to require query to all the root notes then the partial responses from the nodes returns to the root node and that root node merges all of the responses together to form a comprehensive response for the user. The limitation of this approach is, the response time is dependent on lowest performing node, thus increased parallelism doesn’t always speed things up because of overhead or straggler on each node. Also mathematically speaking, 99th percentile latency on 5 node for individual requests becomes a 95th percentile latency for our complete scatter/gather system. And it only gets worse from there: if we scatter out to 100 leaves, then we are more or less guaranteeing that our overall latency for all requests will be 2 seconds. 

A single replica means that if it fails, all scatter/gather requests will fail for the duration that the shard is unavailable because all requests are required to be processed by all leaf nodes in the scatter/gather pattern. This risk is being addressed by adding multiple replica of a node.

Functions and Event-Driven Processing 

Function level decoupling: In microservice approach we splitted our application in small parts and managed by a team. FaaS takes it to the next level. It forces to strongly decouple each function of service.

 There is a common saying that 80% of the daily users are going to be using 20% of the features. For the rest of 80% features that are not being used extensively, can be an ideal candidate for FaaS that charges based on number of requests so we are only paying for the time when your service is actively serving requested and we are always paying for processor cycles that is largely sitting around waiting for a user request. 

The function in FaaS, dynamically spuns up in response to a user request while the user is waiting, the need to load a lot of detail may significantly impact the latency that the user perceives while interacting with your service and this loading cost can be amortized across a large number of requests. 

This approach also has many limitations, the function communicates with each other via network and each function instance cannot have high local memory. So maybe it is a good fit for responding to temporal events, but it is still not sufficient infrastructure for generic background processing and the requiring states needs to be stored in a storage service which adds more complexity in code.

Because of this highly decoupled nature, from the other functions, there is no real representation of the dependencies or interactions between different functions, it becomes difficult to find bugs. Also the cost of a bug related to infinity loop is high.

Batch Computational Patterns: for reliable, long-running server applications. This section describes patterns for batch processing. In contrast to long- running applications, batch processes are expected to only run for a short period of time. 

Work Queue Systems: In the containerised work queue, there are two interfaces: the source container interface, which provides a stream of work items that need processing, and the worker container interface, which knows how to actually process a work item. The work processor can be implemented by a container that processes Shared Work Queue where items in the queue as batch operation or it can be implemented in multi-worker pattern which transforms a collection of different worker containers into a single unified container that implements the worker interface, yet delegates the actual work to a collection of different, reusable containers.

Depending on workload, shared work queued containers needs to be scaled accordingly. Keeping one or few workers running when the frequency of works coming is very low can be overkill. Depending on the workload for a task we can also consider implementing multi-worker pattern using kubernetes jobs to implement work processor. We can easily write a dynamic job creator, which is going to spawn a new job when there is a new item in the queue.

Event-Driven Batch Processing: Work queues are great for enabling individual transformations of one input to one output. However, there are a number of batch applications where you want to perform more than a single action, or you may need to generate multiple different outputs from a single data input. In these cases, you start to link work queues together so that the output of one work queue becomes the input to one or more other work queues, and so on. This forms a series of processing steps that respond to events, with the events being the completion of the preceding step in the work queue that came before it.

In an complex event driven system, often the event needs to go through many steps where data needs to divided into few queues, sometime it requires to merged to get an output.

Copier: The job of a copier is to take a single stream of work items and duplicate it out into two or more identical streams. 

Filter: The role of a filter is to reduce a stream of work items to a smaller stream of work items by filtering out work items that don’t meet particular criteria. 

Splitter: divide them into two separate work queues without dropping any of them. 

Sharder: divide up a single queue into an evenly divided collection of work items based upon some sort of shard‐ ing function

Merger: the job of a merger is to take two different work queues and turn them into a single work queue. 

Coordinated Batch processing:

To achieve more complex event driven architecture, we are going to need more coordinated batch processing techniques.

Join (or Barrier Synchronization) : In merge, it does not ensure that a complete dataset is present prior to the beginning of processing but when it comes to join, it does not complete until all of the work items that are processed. It reduces the parallelism that is possible in the batch workflow, and thus increases the overall latency of running the workflow. 

Reduce : Rather than waiting until all data has been processed, it optimistically merge together all of the parallel data items into a single comprehensive representation of the full set. In order to produce a complete output, this process is repeated untill all of the data is processed, but the ability to begin early means that the batch com‐ putation executes more quickly overall.

Reflecton on Building Microservices Designing Fine Grained Systems

Recently I have been reading “Building Microservices Designing Fine Grained Systems” by Sam Newman. Indeed it is a must read book if you are trying to move toward microservices. This book is going to influence and reflect a lot at my work, so I was thinking I should definitely write a reflection blog. This is what I am attempting to do here. Obviously my point of view will differ in terms of many things, I will be explaining my take aways from the book and I will be trying to reflect my point of view and experience as well. Sam Newman is not a fanboy like me, so he expressed his caution while using docker, but I am a fanboy of docker/kubernetes so obviously I would have whole different perspective how docker makes managing orchestrating microservices easier or where and how docker/kubernetes fits in. Maybe I am going to write about it in another blog.

I agree with Sam Newman when he compared our software industry with fashion industry. We are easily dominated by tech trends. We tend to follow what other big organization is doing. But it has its reasons as well. This big organizations who are setting trends, they are doing a lot of experiments, and publishing their findings and some time publishing few open source tools along with it, as a biproduct of their work and smaller organization like us, most of the time we don’t get the luxury to experiment in that large scale, we don’t have the luxury to innovate, but we can easily become the consumer of those tools and we can easily take leverage of the knowledge that has been shared with us. So we take the easy route.

One of the key thing about microservices is that, every service needs to have some form of autonomy, it can depend on other service for data or information that is not within the boundary or context,  but it has to have its own data source that it owns, where it can store and retrieve data that is within the boundary or the context of the service. It can take leverage the best programming language that matches with its goal, it can use the database that matches with its need. Independence of tooling is a great feature, it reduces the tech debts in a very positive way. That’s probably the main selling point of microservices.

At a service oriented architecture, maybe it can be a common practice to directly access database that other service is primarily using, the reason is very simple I need a data that the other service has, so I am accessing the data that I need. But what happens when the owner of the service change its schema for representing their data better? The other service that is consuming the data of that service locks up, the other services which were consuming the data won’t be able to consume the data that it absolutely needs. So maybe, we need less coupling in our services then what we are already got used when we were using service oriented architecture. The best way that we can think of, maybe we can do that using REST APIs, when we would need a data from other service we would request the data using API calls to the service who owns the data and we don’t know the internal details about how the data is being stored. We need the data not its internal representation. Less coupling allows us to change the internal representation of the things without impacting the expected result that it needs to produce. The import term here is, it allows us to improve and change without going through huge hassles and long meetings with stakeholders who runs other services that will be affected for the change in your internal representation. Not only database, usually the system that has been designed to make RPC calls to change or to collect resources of other services are also prone to this problem.

When it is microservice we are talking about, it is easy to imagine that there is a huge code base with millions of lines of code that does pretty much everything on its own, it is struggling with load and the codebase is getting out of hand and almost unmanageable and we don’t want that so we want to move out of it. We want to split this huge monolith into few manageable services with manageable loads. Now dividing a monolith is not always an easy task. Setting a boundary is often harder but every service needs to have a well defined boundary. When dividing monolith we can follow a simple rule of thumb that says code that changes together will stay together which is known as high cohesion and the codes that does not change together may split apart. Then we can further split it based on domain, functionality, special technical needs. For example maybe a facebook like friends network can leverage a graph structured database, maybe neo4j or graphql in that case it deserves to be separated as a different service that use the appropriate technical tools that it needs.

Service to service communication in microservice is tricky, many people are using trusted network based security where the idea is, we try out best (invests more) to secure our parameter/border, it is hard to get into the parameter but once you are inside the parameter we trust you with everything. For organization like this few extra header that defines username is enough. But there are some other organization they are not being able to guarantee with 100% accuracy that their parameters are secured, there are so many hacking techniques as well that allow them to breach the boundary pretending to be someone else. So maybe we need to maintain some form of security and privacy even when we are communicating inside our secured parameter, from one service to another service. All internal communication can be encrypted using some form of HTTPS which encrypts the message and verify that the message source and destinations are accurate. Some organizations are relying on JWT as it is very light weight but still every service needs to have the capability to verify that token. A shared token has to be shared between the services and it has to be done using secured fashion.

When deciding between orchestration and choreography, choreography should be the best pattern because otherwise if we try to orchestrate by one service, we are basically overpowering it compared to other services that eventually it will start becoming a god service that will keep dominating other services all the time. We don’t want service to service domination. We want equal distribution of load and responsibility and almost equal importance. Also it has its problems as well, because in a microservice world new services start to appear everyday and old services gets discarded everyday it becomes real hassle for the owner of the god service to track them and to add and remove those service calls to our god service. Rather than a service who is making rest api calls to notify other services to do certain things, we can fire events and any other service who might be interested on that event will subscribe to that event and play its part when the event is triggered. If new service comes in, it is easier for it as well to subscribe to that event. It can be implemented using kafka or maybe Amazon SQS.

In microservice world it is extremely important to isolate the bug, the bug from one service tends to cripple other services. So it often becomes difficult to identify the real bug in appropriate service. So we need to have proper testing mechanism that that will isolate a service while testing and at the time of testing other api calls that it makes to other services needed to be mocked so it needs to have proper setup for isolated testing. So we would servers that is capable to imitate the ideal behavior of other service. When testing, it would be wise to have another environment that has its own set of database, in that way we would be able to test how resource creation is responding on a particular service. We talked about isolated testing, but integration testings are also important. As in micro services, there are so many moving pieces, maybe a service is calling an api that has been removed or the url has changed for some reason, (maybe the owner of the service has forgot to inform other stakeholdsrs) in that case the service that is calling the api, it won’t be able to perform what it was supposed to do. So integration tests are important, it is going to test when integrated together it is still performing the way it should be performing.

Every service needs to get monitored and there has to be a set of action that would take place when a service is down. For example maybe there can be a sudden bug introduced in one service causing huge CPU or memory can make the service go offline. It can affect the other service that is running on that machine. It is usually recommended that one service is being hosted on one machine so that the failure on one service does not cascade the problem to another service. There also should a well placed rule or circuit breaker on when are we going to consider a service is down and when are we going to stop sending traffic to a service. Suppose one of are service is down, in that case another thing to note here is that is it going to be wise to show the whole system is down when there is only one service that is down out of maybe 100s of services. Instead of doing that we can replace a portion of website with something else. For example when the ordering service of an ecommerce is down. It won’t be wise to shut the whole website down, we can let user visit the site and rather than letting them click “place order” button we can show them “call our customer care to place order”. So if implemented wisely micro service can reduce our embarrassment. Netflix has implemented an opensource smart router called zulus that can open and close a feature to group of users while the rest of the users get that is default. It can come handy time to time as well. There are many tools out there to monitor servers, from opensource world we can use nagios, or we can use 3rd party monitoring tools like datadog, new relic, cloudwatch and the options are many.

When talking about monitoring, we also need to be conscious that sometime our app is going to pass all the health checks yet it will produce a lot of error. For such situations maybe we need to monitor the logs, maybe a centralized log aggregation and monitoring system (famous ELK stack) can be helpful. As we are dealing with microservice, I am sure we have so many servers and it won’t be wise to poke in every server to collect logs when there is crisis, a log aggregation platform is a necessity, that’s how I want to put it. Our log aggregation system needs to be intelligent to fire an alarm when it reaches a certain number of error messages. Depending on the severity of error message, maybe it is time to let the circuit breaker know that we don’t want any traffic on that service until it has been sorted out.

In microservices we are dealing with 100s and 1000s of services, so there will always be some sort of deployment going on, either it is on this service or that service. If DevOps culture is being adopted small changes will be incouraged to go on production. The philosophy is if a bug is introduced due to small change, it will be easier to spot and easier to fix compared to a huge chunk of code. As each services are interconnected, if a service go down even for a second other services gets affected so zero downtime is a must. We can use blue green deployment or cannery deployment strategy to achieve zero downtime thingy and it is absolutely important. If something goes wrong in a deployment there has to have a well defined strategy to rollback to previous build. It is better if it has a CI/CD pipeline attached with it. Chef, Puppet like server automation tools comes handy because it helps us to automate the monotonous tasks. When automation scripts are being written, it is being tested over and over again, it is faster and more reliable, same goes for rollback.

Micorservices makes security much easier, as we know from CIA triangle, we can’t have everything. Often when we want to ensure security, we will need to slow our systems down as  a trade off. Now, as we are splitting our services, does all our services needs equal amount of security be ensured (read, does all our services needs get slow?). Often some services need more security than other. So microservices will give us opportunity to use the write security where security is needed, so it won’t slow the whole server down, when a specific service needs special security measure.

Sometime DNS management gets troublesome. It takes hours to get everything propagated across the servers. It can When we are dealing with interconnected services (and there are 100s of them) it will be really troublesome if server to server DNS records are not being reflected the same way, so we can have our own service discovery mechanism, we can have our own DNS server. We can take advantage of Zookeeper, Consul, Eureka and so on.

Scaling microservices can be challenging some time. It is said that people use 20% of the feature 80% of the time, in other word 80% of the feature are never been used. So when we have a heavy load, linearly increasing all the servers most of the time will be a wrong choice. So we need to setup more complex scaling techniques that needs to be implemented on each service separately. Our load balancers are getting smarter everyday so it won’t be too hard to implement.

Newman has pointed out few interesting things about implementation of microservices, one of the key point he has mentioned is about Conway’s Law. If the organization is not organized, and works independently in terms of team to department and collaboratively with other department like a microservice, it would be challenging for that organization to adopt microservices. So it makes me wonder maybe for smaller startup, maybe microservices are not the solution they should look for. Also when it comes to the ownership of service, in smaller organization maybe the same person or team is going to own all of the services. It is going to be a huge overhead as well.