Microservices

In this article, I will cover some of the concept of microservices architecture which is becoming popular for building modern application. Microservices is building application as a small set of services. Each of these services run under its own process and communicates with other processes using protocols such as HTTP, AMQP, HTTP2. Each of these microservice implements a specific end to end domain or business capability and each of them developed autonomously and deployed independently. Some of these services can have their own databases and these microservices are managed by another service or tool called API management which acts as a facade and in turn provides a single-entry point for various client like web or mobile. One of the ways to design the microservices is using Bounded context pattern which is part of the DDD (domain driven design). A bounded context encapsulates the details of a single domain or business capability and could be composed of several physical services, but not vice versa

Domain model in DDD

Value: Domain objects which describes a domain aspect. These objects do not have an identity.

Entity: Domain Objects which have the identity and can be persisted.

Aggregates: Group of domain objects which can be treated as single unit or consistent together.

For more details refer the link here

Communication in microservices

Synchronous request/response communication: HTTP is synchronous protocol and hence this kind of communication is used between client to microservices or client to API Management services. The invocation of the microservices can be as simple as getting the URL of the service and invoking the service through HTTP client as shown below.

public Task<string> GetStringAsync(string uri)
        {
           var requestMessage = new HttpRequestMessage(HttpMethod.Get, uri);
           var response = await _client.SendAsync(requestMessage);
           return await response.Content.ReadAsStringAsync();
        }

One of the drawbacks of synchronous communication is that we cannot use it for communication between microservices as a failure in one services will propagate to the entire system which will bring down the system. In addition to asynchronous communication, we can use pattern like Retry and Circuit Breaker pattern to build resilient microservices and client applications.

Asynchronous communication: AMQP which is the asynchronous protocol is used for communication between microservices and are of two types

Single: Every request will be processed by exactly one receiver. Considering the scenario where we have the basket service and an order service. When the user do a check out of items in the basket service it should be converted to order in order service

Below shows the snippet of the code used for

·        Specifying a single receiver or subscription

·        Publishing the message on the service bus

·        Receiving the message from the service bus

Subscription: We can use the RabbitMQ service bus custom implementation

var eventBus = app.ApplicationServices.GetRequiredService<BuildingBlocks.EventBus.Abstractions.IEventBus>()
eventBus.Subscribe<UserCheckoutAcceptedIntegrationEvent, IIntegrationEventHandler<UserCheckoutAcceptedIntegrationEvent>>();

 Publishing: We publish the user check out event onto the service bus

Publishing: We publish the user check out event onto the service bus
[Route("checkout")]
[HttpPost]
 public async Task<IActionResult> Checkout([FromBody]BasketCheckout basketCheckout, [FromHeader(Name = "x-requestid")] string requestId)
        {
            var userId = _identitySvc.GetUserIdentity();
            basketCheckout.RequestId = (Guid.TryParse(requestId, out Guid guid) && guid != Guid.Empty) ?
                guid : basketCheckout.RequestId;
            var basket = await _repository.GetBasketAsync(userId);
            if (basket == null)
            {
                return BadRequest();
            }


            var eventMessage = new UserCheckoutAcceptedIntegrationEvent(userId, basketCheckout.City, basketCheckout.Street,
                basketCheckout.State, basketCheckout.Country, basketCheckout.ZipCode, basketCheckout.CardNumber, basketCheckout.CardHolderName,
                basketCheckout.CardExpiration, basketCheckout.CardSecurityNumber, basketCheckout.CardTypeId, basketCheckout.Buyer, basketCheckout.RequestId, basket);


            // Once basket is checkout, sends an integration event to
            // ordering.api to convert basket to order and proceeds with
            // order creation process

            _eventBus.Publish(eventMessage);            
            return Accepted();

        }

Receiving: We handle the message and create new events and convert into an order.

public class UserCheckoutAcceptedIntegrationEventHandler : IIntegrationEventHandler<UserCheckoutAcceptedIntegrationEvent>

    {
        private readonly IMediator _mediator;
        private readonly ILoggerFactory _logger;
        private readonly IOrderingIntegrationEventService _orderingIntegrationEventService;
        public UserCheckoutAcceptedIntegrationEventHandler(IMediator mediator,
            ILoggerFactory logger, IOrderingIntegrationEventService orderingIntegrationEventService)
        {

            _mediator = mediator ?? throw new ArgumentNullException(nameof(mediator));
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
            _orderingIntegrationEventService = orderingIntegrationEventService ?? throw new ArgumentNullException(nameof(orderingIntegrationEventService));

        }

        public async Task Handle(UserCheckoutAcceptedIntegrationEvent eventMsg)
        {

            var result = false;
            // Send Integration event to clean basket once basket is converted to Order and before starting with the order creation process
            var orderStartedIntegrationEvent = new OrderStartedIntegrationEvent(eventMsg.UserId);
            await _orderingIntegrationEventService.PublishThroughEventBusAsync(orderStartedIntegrationEvent);
             if (eventMsg.RequestId != Guid.Empty)
            {
                var createOrderCommand = new CreateOrderCommand(eventMsg.Basket.Items, eventMsg.UserId, eventMsg.City, eventMsg.Street, 
                    eventMsg.State, eventMsg.Country, eventMsg.ZipCode,
                    eventMsg.CardNumber, eventMsg.CardHolderName, eventMsg.CardExpiration,
                    eventMsg.CardSecurityNumber, eventMsg.CardTypeId);
                var requestCreateOrder = new IdentifiedCommand<CreateOrderCommand, bool>(createOrderCommand, eventMsg.RequestId);
                result = await _mediator.Send(requestCreateOrder);

            }            
   
        }

    }

Multiple: Every request will be processed by one or multiple receivers. This uses Publish/Subscribe mechanism which uses services bus for communication and can be implemented using the tools like RabbitMQ and Azure bus.

In this type of communication, Microservices publish an event when it updates a business entity on to the services bus and other microservices which are subscribing to those events will update its own business entity upon receiving those events. We use these events to implement business transactions which span multiple services. This is called as eventual consistency between those services. Consider a scenario where the basket service need to subscribe both for the changes in the product price (so that the basket is updated with the current price) and whenever the order is created (so that the basket is cleared)

Below shows the snippet of the code used for

·        Specifying multiple receiver or subscription

·        Publishing the message on the service bus

·        Receiving the message from the service bus

Subscription:

var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();
eventBus.Subscribe<ProductPriceChangedIntegrationEvent, ProductPriceChangedIntegrationEventHandler>();
eventBus.Subscribe<OrderStartedIntegrationEvent, OrderStartedIntegrationEventHandler>();

Publishing: We publish the change in the product price on to the service bus.

//PUT api/v1/[controller]/items
        [Route("items")]
        [HttpPut]
        public async Task<IActionResult> UpdateProduct([FromBody]CatalogItem productToUpdate)
        {
            var catalogItem = await _catalogContext.CatalogItems
                .SingleOrDefaultAsync(i => i.Id == productToUpdate.Id);
            if (catalogItem == null)
            {
                return NotFound(new { Message = $"Item with id {productToUpdate.Id} not found." });
            }

            var oldPrice = catalogItem.Price;
            var raiseProductPriceChangedEvent = oldPrice != productToUpdate.Price;

            // Update current product
            catalogItem = productToUpdate;
            _catalogContext.CatalogItems.Update(catalogItem);

            if (raiseProductPriceChangedEvent) // Save product's data and publish integration event through the Event Bus if price has changed
            {
                //Create Integration Event to be published through the Event Bus
                var priceChangedEvent = new ProductPriceChangedIntegrationEvent(catalogItem.Id, productToUpdate.Price, oldPrice);
                // Publish through the Event Bus and mark the saved event as published
                await _catalogIntegrationEventService.PublishThroughEventBusAsync(priceChangedEvent);
            }
            else // Just save the updated product because the Product's Price hasn't changed.
            {
                await _catalogContext.SaveChangesAsync();
            }
            return CreatedAtAction(nameof(GetItemById), new { id = productToUpdate.Id }, null);
        }

Receiving: We handle the message and update the price in the basket.

public class ProductPriceChangedIntegrationEventHandler : IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>
    {
        private readonly IBasketRepository _repository;
        public ProductPriceChangedIntegrationEventHandler(IBasketRepository repository)
        {
            _repository = repository ?? throw new ArgumentNullException(nameof(repository));
        }
        public async Task Handle(ProductPriceChangedIntegrationEvent @event)
        {
            var userIds = _repository.GetUsers();
    
            foreach (var id in userIds)
            {
                var basket = await _repository.GetBasketAsync(id);
                await UpdatePriceInBasketItems(@event.ProductId, @event.NewPrice, @event.OldPrice, basket);                      
            }
        }
        private async Task UpdatePriceInBasketItems(int productId, decimal newPrice, decimal oldPrice, CustomerBasket basket)
        {
            var itemsToUpdate = basket?.Items?.Where(x => int.Parse(x.ProductId) == productId).ToList();
            if (itemsToUpdate != null)
            {
                foreach (var item in itemsToUpdate)
                {
                    if(item.UnitPrice == oldPrice)
                    { 
                        var originalPrice = item.UnitPrice;
                        item.UnitPrice = newPrice;
                        item.OldUnitPrice = originalPrice;
                    }
                }
                await _repository.UpdateBasketAsync(basket);
            }         
        }
    }

CAP theorem: It states that you cannot build a system which is highly available, strong consistent and tolerant to failures. So, we should choose two of these three properties

In microservices-based architecture we choose availability and tolerance and less emphasize on strong consistency.

The challenge with the above scenario (two phase commit) of updating the business entity and publishing the events to other microservices though provide eventual consistent, but there will scenario when the data will become not consistent Ex: the server crashes after the business entity was updated and before the event was published to event bus or there were some issues with the event bus. From the previous example, the system crashes when the price event change has happened, but before the event is published to service bus

The other challenge would be though we can scale these microservices, the database at the backend will have a single model for both read and write and hence we will not be able to scale read and write independently.

To overcome the challenge of two phase commit and database scaling challenges, we will consider Event Sourcing & CQRS

Event Sourcing & CQRS: We approach building the system in a unique way compared to the traditional approach

·        We think the input to the system as commands which are then converted to different events and we store these events instead of the state of the system and we compute the state of events by replaying the events.

·        We have a different data models optimized(performant) for writing and reading the data and data is made eventual consistent between both by sending message to service bus. (For reading we can use NoSQL database and for writing we can use SQL database)

So broadly we can think of this system as

1)     Commands

2)     Event Sourcing

3)     Query

Considering a scenario where we need to capture the location of a team member and take an action when member of the same team is near by

Commands:

·        It takes the input or the command and converts the command to event

·        Emit the event on a service bus

//Covert the command to event
MemberLocationRecordedEvent locationRecordedEvent = converter.CommandToEvent(locationReport);
locationRecordedEvent.TeamID = teamServiceClient.GetTeamForMember(locationReport.MemberID);
 //Emit the event
eventEmitter.EmitLocationRecordedEvent(locationRecordedEvent);
 

Events Sourcing:

·        It obtains the message from the event stream by subscribing to the service bus and write the events to an event store

·        It might process the stream of messages and generate new events or messages on to the services bus

//Get the message details
var memberLocations = locationCache.GetMemberLocations(mlre.TeamID);

//Process the message or events and generate new events
ICollection<ProximityDetectedEvent> proximityEvents =
proximityDetector.DetectProximityEvents(mlre, memberLocations, 30.0f);
                foreach (var proximityEvent in proximityEvents) {
                    eventEmitter.EmitProximityDetectedEvent(proximityEvent);
                }
//Store the events or messages
locationCache.Put(mlre.TeamID, new MemberLocation { MemberID = mlre.MemberID, Location = new GpsCoordinate {
                    Latitude = mlre.Latitude, Longitude = mlre.Longitude
                } });

Query:

·        This will provide information about the events that happened in the system by querying the event store. We can create the state based on these events. 

//Query the store and get all the values for a team
public IList<MemberLocation> GetMemberLocations(Guid teamId)
{
    IDatabase db = connection.GetDatabase();
    RedisValue[] vals = db.HashValues(teamId.ToString());
    return ConvertRedisValsToLocationList(vals);
}

One of the questions that arise when we create a state based on replaying the events would if it takes time to generate the state every time based on the events. This is solved by creating a snapshot of the state and replaying only the events that had happened after the snapshot.