DEV Community: Manav Kushwaha

Learning Microservices With Go(Part 5). Asynchronous Communication. (Kafka)

Manav Kushwaha — Thu, 29 Feb 2024 03:37:39 +0000

Intro

This post is the fifth part of the "Learning Microservices with Go". I'm writing these posts as I'm learning the concepts. If you haven't checked out the part 4, here's the link for it: Part 4.
Link for Git Repo: Github

In last part, we added gRPC code to our project. We used gRPC logic for synchronous communication with different services.

In this part we'll be learning about asynchronous communication, what are it's benefits, where to use it and also it's implementation (using Kafka).

Let's start!

SYNCHRONOUS VS ASYNCHRONOUS COMMUNICATION

You might be asking what's the difference between the two types of communication.

For synchronous communication, the sender sends a request and waits for the reply from the receiver. Only if the reply comes(or request times out) does the sender continue with it's work.
For asynchronous communication, the sender sends a request and doesn't wait for the reply from the receiver(it only waits for the confirmation that the message was sent successfully). The receiver can these choose to send the reply(or not send any reply) based on the implementation.

An analogy is like a phone call vs email.
Both parties need to be sending and receiving actively in case of phone call, but in case of email, the receiver can reply at it's own pace or not even reply.

BENEFITS AND CHALLENGES TO ASYNCHRONOUS COMMUNICATION

Benefits

Streamlined approach to processing messages:
- Eg: A service processing messages and sending them, can focus on processing and not focus on the response from other services.
- It just needs confirmation that the message was sent.
Decouple sending and processing requests:
- Eg: When one service requests to download a video from a server, it may take minutes/hours to do that.
- If we use the asynchronous model, it can just send the task to the server, get the acknowledgement from the server. The service can continue it's work and the downloading can happen asynchronously in the background.
- Then when the task is complete, it'll receive a notification of completion.
Better Load Balancing:
- Some applications don't have even request loads and may be prone to spikes.
- Having synchronous communication would require it to handle all of the requests at the same time.
- This would make the service unable to handle requests properly, also affecting the clients. It might cause delays in response.
- Eg: Trying to handle thousands of calls at the same time.

PLACES TO USE ASYNC COMMUNICATION

Some tasks do require immediate response and hence synchronous communication needs to be used. Hence we should know when to use Async. Here are some places where async communication is recommended:

Long running tasks
eg: Video processing. The caller wouldn't need to wait and would be notified when the task is completed or failed.
Send once, processed by multiple components
Certain types of messages, eg: status reports can be processed by multiple independent components.
Instead of sending it to each client independently, the message can get published to a component, able to be consumed by everyone.
High performance sequential tasks:
Some operations when performed sequentially and/or (in batches) are more effecient.
Eg: Writing to HDD. If performed sequentially, result in lower latency(increase performance).
Async communication provides these improvements as the server can process the incoming request in it's own speed and process tasks one after the other.

CHALLENGES

There are some challenges to using the asynchronous communication also. Some of these are:

Complex Error Handling
If any message is lost, we need to think about many cases, was the message sent properly? was the message recieved? did the response get lost?
If we're using synchronous communication, we'd have immediately known if this happened and could have retried/took other measures.
Additional Components for Message delivery
Certain use cases require additional software for delivering the message.
These software also perform additional operations eg: message batching, storing.
This causes increase in the complexity in the system.
Async data flow may seem non-intuitive and more complex
Unlike the synchronous communcation, where every request is followed by a response, in Async, there might not be any response.
Or the response may require additional steps to be taken.

TECHNIQUES AND PATTERNS OF ASYNC COMMUNICATION

Lets describe some patterns and concepts that'll help us in the asynchronous interactions.

Message Broker

Message broker is an intermediary component. It can be used for:

Message delivery: Delivery of message to one or more servers.
Message transformation: Transform the message to a format that's consumable by other services.
Message aggregation: Combines multiple message to send a single message improving efficiency in delivery/processing.
Message routing: Routing the messages to their appropriate destinations using some set rules.

Eg: Sending an email, the mail platform(eg: Google for gmail) would be acting like message broker.
We don't need to know how to send the message to other person. Google takes care of it.

Guarantees for message delivery:
A message broker can provide different types of guarantees for message delivery. These may be:

Atleast once: Message delivered. May deliver more than once in case of failure. Broker just resends in case of failure.
Exactly once: Message delivered exactly once. Broker needs to perform additional checks for exactly once delivery.
Atmost once: Message delivered 0 or 1 time.

Based on the guarantees, we can also classify the message broker as Lossy or Lossless.

Lossy: When message broker can sometimes(in failures) lose the messages. Eg: Atmost-once.
Lossless: When message broker will not lose the messages. Eg: Atleast, exactly once.

Publisher-Subcriber model

This is a model of communication between microservices(not limited to that) where every microservices can publish messages or subscribe to the relavant ones.
Eg: Following someone on twitter will make you subscribe to them. Whenever they publish(post) something, you'll get that in your feed.

The pub-sub model provides a flexible solution for sending messages in a system where the messages can be produced and consumed by multiple components.
The publisher doesn't need to worry about the delivery of the messages and about any failures.
Also the subscriber can just subscribe to relavant message and get them delivered without needing to talk to the publishers directly.

APACHE KAFKA

Apache Kafka is an open source message broker system. It was developed at LinkedIn. Very widely used around the world.

Kafka Basics

The component that produces messages is called a producer.
Messages are published in a sequential order to objects called topics.
Each message in a topic, has a unique numerical offset in it.
Kafka provides the APIs for consuming messages for consuming existing topics(that component is called Consumer)
Topics can also be partitioned to allow multiple consumers to consume from them(eg: Parallel data processing)

Benefits

High write and read throughput: Kafka is optimized for highly performant read & writes. It does this by doing as many sequential writes and reads as possible, allowing it to make use of HDD, as well as sequentially sending large data over the network.
Scalability: Using topic partitioning, devs can achieve parallel processing of their data.
Flexible durability: Devs can configure policies for storing data in Kafka, eg: message retention. Messages can be configured to be stored for a fixed duration(eg: 1day) or indefinitely(till storage is available)

ADOPTING KAFKA FOR OUR MOVIE-APP

We want to now work with a ratings provider(eg: IMDB), which will provide ratings for a movie to us.
We need to ingest these ratings along with the rating API that we'd created earlier. These two can work side by side.
We can make this communication*async*as we don't need to process this bulk information immediately.
Our rating service could do this periodically or even immediately when there's new data published.
We'll use Apache Kafka as our message broker.

First we'll implement:

Application to produce the ratings data.
Consumption of data from Apache Kafka and storing those ratings in our ratins repository.

Let's assume that the data provider will provide us JSON formatted data(Most external APIs use JSON format).

[
 {"userId":"105","recordId":"1","recordType":1,"value":5,"providerId":"test-provider","eventType":"put"},
 {"userId":"105","recordId":"2","recordType":1,"value":4,"providerId":"test-provider","eventType":"put"}
]

We'll start by adding a Go model for these rating records to src/rating/pkg/model/rating.go file which already contains the rating structure.

// RatingEvent defines an event containing a rating record
type RatingEvent struct {
    RecordID   RecordID        `json:"recordId"`
    RecordType RecordType      `json:"recordType"`
    UserID     UserID          `json:"userId"`
    Value      RatingValue     `json:"ratingValue"`
    EventType  RatingEventType `json:"eventType"`
}

// RatingEventType defines a type of rating event
type RatingEventType string

// Rating Event Types
const (
    RatingEventTypePut    = RatingEventType("put")
    RatingEventTypeDelete = RatingEventType("delete")
)

Now we'll add an example application, that'll read the data from the given file and publish it to Kafka.
Create a file cmd/ratingingester/main.go with the content:

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/confluentinc/confluent-kafka-go/kafka"
    "movieexample.com/rating/pkg/model"
)

// The main function creates a Kafka producer, reads rating events from a file, and sends them to a Kafka topic.
func main() {
    fmt.Println("Creating a Kafka producer")

    producer, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost"})
    if err != nil {
        log.Fatalf("Failed to create producer: %s\n", err)
    }
    defer producer.Close()

    const fileName = "ratingsdata.json"
    log.Println("Reading data from file", fileName)

    ratingEvents, err := readRatingEvents(fileName)
    log.Printf("RatingData read from file: %v\n", ratingEvents)
    if err != nil {
        log.Fatalf("Failed to read rating events: %s\n", err)
    }

    const topic = "ratings"
    log.Println("Sending rating events to topic", topic)

    if err := produceRatingEvents(producer, topic, ratingEvents); err != nil {
        panic(err)
    }

    const timeout = 10 * time.Second
    log.Printf("Waiting for %s to send rating events\n", timeout)

    producer.Flush(int(timeout.Milliseconds()))
}

func readRatingEvents(fileName string) ([]model.RatingEvent, error) {
    f, err := os.Open(fileName)
    if err != nil {
        return nil, err
    }
    defer f.Close()

    var ratingEvents []model.RatingEvent
    if err := json.NewDecoder(f).Decode(&ratingEvents); err != nil {
        return nil, err
    }

    return ratingEvents, nil
}

func produceRatingEvents(producer *kafka.Producer, topic string, ratingEvents []model.RatingEvent) error {
    for _, ratingEvent := range ratingEvents {
        encodedEvent, err := json.Marshal(ratingEvent)
        if err != nil {
            return err
        }

        message := &kafka.Message{
            TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
            Value:          encodedEvent,
        }
        if err := producer.Produce(message, nil); err != nil {
            return err
        }
    }
    return nil
}

We initialized a Kafka.Producer via NewProducer, read the file having data, produced rating events to Kafka.

Note: We're using a Kafka client for Go for this. See the import in the above snippet.

Key points about the code that we just wrote:

We create a kafka producer by creating a new producer and using the localhost as the kafka address.
When producing the kafka events, we specified the topic(ratings) to which the events are produced to. So, if we send a message to a particular topic, only the services subscribed to that topic will get that message.
We flush out the messages at the end of our program to make sure that all the messages are sent to the Kafka.

Also add the ratingsdata.json file in the same directory with the following content:

[
    {
        "userId":"105",
        "recordId":"1",
        "recordType": "movie",
        "ratingValue": 5,
        "providerId":"test-provider",
        "eventType":"put"
    },
    {
        "userId":"105",
        "recordId":"2",
        "recordType": "movie",
        "ratingValue": 4,
        "providerId":"test-provider",
        "eventType":"put"
    }
]

Now we've completed the logic for publishing events to the Kafka. We'll now write logic for consuming these events.
Create a file rating/internal/ingester/kafka/ingester.go with the following content:

package kafka

import (
    "context"
    "encoding/json"
    "log"

    "github.com/confluentinc/confluent-kafka-go/kafka"
    "movieexample.com/rating/pkg/model"
)

// Ingester defines a Kafka ingester
type Ingester struct {
    consumer *kafka.Consumer
    topic    string
}

// NewIngester creates a new Kafka ingester
func NewIngester(adrr string, groupID string, topic string) (*Ingester, error) {
    consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers": adrr,
        "group.id":          groupID,
    })
    if err != nil {
        return nil, err
    }
    return &Ingester{consumer, topic}, nil
}

// Ingest starts ingesting from Kafka and returns a channel of rating events
// representing the data consumed from topics
func (i *Ingester) Ingest(ctx context.Context) (<-chan model.RatingEvent, error) {
    if err := i.consumer.SubscribeTopics([]string{i.topic}, nil); err != nil {
        return nil, err
    }

    ch := make(chan model.RatingEvent, 1)
    go func() {
        for {
            select {
            case <-ctx.Done():
                close(ch)
                i.consumer.Close()
            default:
            }

            // Waiting indefinitely
            msg, err := i.consumer.ReadMessage(-1)
            if err != nil {
                log.Println("Consumer error:", err)
                continue
            }
            var event model.RatingEvent
            if err := json.Unmarshal(msg.Value, &event); err != nil {
                log.Println("Failed to unmarshal event:", err)
                continue
            }
            ch <- event
        }
    }()

    return ch, nil
}

We implemented a NewIngester function to create a new Kafka ingester. This will consume from a topic in background and then return a Go channel with the rating events.

To use this ingester we'll add to our rating/internal/controller/rating/controller.go

type ratingIngester interface {
    Ingest(ctx context.Context) (<-chan model.RatingEvent, error)
}

// Controller defines a rating service controller
type Controller struct {
    repo     ratingRepository
    ingester ratingIngester
}

// New creates a rating service controller
func New(repo ratingRepository, ingester ratingIngester) *Controller {
    return &Controller{repo, ingester}
}

We'll also modify the rating/cmd/main.go to use this new Controller struct.

    repo := memory.New()
    ingester, err := kafka.NewIngester("localhost", "rating", "ratings")
    if err != nil {
        log.Fatalf("Failed to create ingester: %s\n", err)
        ingester = nil
    }
    ctrl := rating.New(repo, ingester)

    // Start the ingester in a separate go routine
    // This will keep trying to ingest the rating events from the kafka topic every 10 seconds
    go func() {
        for {
            ratingEventChannel, err := ingester.Ingest(ctx)
            if err != nil {
                log.Fatalf("Failed to ingest: %s\n", err)
            }
            for ratingEvent := range ratingEventChannel {
                // Put the rating event in the repository
                recordId := ratingEvent.RecordID
                recordType := ratingEvent.RecordType
                rating := &model.Rating{
                    RecordID:   ratingEvent.RecordID,
                    RecordType: ratingEvent.RecordType,
                    UserID:     ratingEvent.UserID,
                    Value:      ratingEvent.Value,
                }
                log.Printf("Putting rating: %v for recordId: %s recordType: %s\n", rating, recordId, recordType)
                err := ctrl.PutRating(ctx, recordId, recordType, rating)
                if err != nil {
                    log.Fatalf("Failed to put rating: %s\n", err)
                }
            }
            time.Sleep(10 * time.Second)
        }
    }()

Now we've completed the modifications and our rating service is now able to consume the data using both synchronous and asynchronous communication.

But we still haven't installed Kafka. To do this we'll be using a docker image of Kafka.

Pull the docker image of Kafka. Note: The latest version might be different as to what it now. Check out the website for latest version.

docker pull apache/kafka:3.7.0

Then run it using

docker run -p 9092:9092 apache/kafka:3.7.0

This will get the Kafka up and running.

Testing

Now we'll test our implementation.

First run all the services and the service discovery consul container as we've done multiple times previously.
Now first add the movie metadata.
If you now try to get the rating using GetMovieDetails, it'll show you No rating found for the record.

Now we'll run the rating ingester. Go to the rating ingester directory(cmd/ratingingester/) and run go run main.go. This will publish the rating events to the Kafka.

These rating events will automatically be added to our repository because our rating controller checks for any new rating event every 10 seconds.

Now let's do the same postman request of getting movie details. This time the result is a bit different.

We can see that the rating is the same as that present in the file because there is only one rating record in the repository.

Tadaaaa, we've completed this part. We learnt about asynchronous communication, it's benefits, drawbacks, use-cases. Also we incorporated Kafka in our movie-app to see what we'd learnt in practical way.

If you found something useful, do Like and Share. It really helps the article.

Here's the github repo: Github

See you!!

Learning Microservices with Go(Part 4). GRPC (Synchronous Communication)

Manav Kushwaha — Wed, 14 Feb 2024 02:42:25 +0000

Intro

This post is the fourth part of the "Learning Microservices with Go". I'm writing these posts as I'm learning the concepts. If you haven't checked out the part 3, here's the link for it: Part 3.
Link for Git Repo: Github

In part 3, we explored serialization, various serialization formats, their comparison, implemented some simple benchmarks to measure the performance.

In this part, we'll be adding gRPC code to our project which uses Protocol Buffers, that we'd learnt about in the last part. We'll use gRPC logic for communication with different services instead of REST APIs which we're using earlier.

We'll also be testing the services using Postman. This was the first time I used postman to test a non-http service. Everyday you learn something new!!

Let's start!

Synchronous Communication

Synchronous communication is way for applications(eg: microservices) to interact over the network using the request-response model. There is a request that goes to the other application and it waits for the response from the other side. It waits till the response comes or the request times out. This way of interaction is used by many protocols eg: HTTP.

Also, there are some existing Remote Procedure Call(RPC) framework and libraries that can help you achieve this task. What's an RPC? RPC allows one machine on the network to call a function(subroutine) on another machine as if it was called natively on the other machine.

For our use case, we'll be using RPCs to call methods of a service from another service. We'll be taking help of gRPC for this task.

gRPC

gRPC is an RPC framework that was developed by Google. It used HTTP/2 as the transport protocol and Protobufs as the serialization format.

It essentially allows the user to define client side and server side code for the services. The code generated can be used by other services to call the specified method on this service.
Also it's language agnostic i.e. you can use different language for each service that best suits that service.

gRPC is one of the most widely known RPC framework used for building microservice architecture.

Defining Service using Protobufs

Let's now see how to define a service using Protobufs. We'll also be defining code for the gRPC framework, which will allow us to talk with other services.

First install the gRPC plugin for Protobuf. You can do this by:

go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

Now we'll add some code to the api/movie.proto file that we created in the last part.

service MetadataService {
    rpc GetMetadata(GetMetadataRequest) returns (GetMetadataResponse);

    rpc PutMetadata(PutMetadataRequest) returns (PutMetadataResponse);
}
message GetMetadataRequest {
    string movie_id = 1;
}
message GetMetadataResponse {
    Metadata metadata = 1;
}

message PutMetadataRequest {
    Metadata metadata = 1;
}
message PutMetadataResponse {
}

The code that we added defined our MetadataService. We already have the Metadata structure which we defined in the previous part that we can reuse.

Let's explain the code that we added.
We defined a rpc named GetMetadata that accepts GetMetadataRequest as it's parameter and outputs GetMetadataResponse. We then defined these two messages below the request function. Note:

It's good practice to create a new structure for both a request and a response. Eg: GetMetadataRequest, GetMetadataResponse.
It's good to follow consistent naming for all the endpoints. We'll prefix the request and response with the function name for all our endpoints.

Similarly for the PutMetadata.

Now defining the RatingService.

service RatingService {
    rpc GetAggregatedRating(GetAggregatedRatingRequest) returns (GetAggregatedRatingResponse);
    rpc PutRating(PutRatingRequest) returns (PutRatingResponse);
}

message GetAggregatedRatingRequest {
    string record_id = 1;
    string record_type = 2;
}

message GetAggregatedRatingResponse {
    double rating_value = 1;
}


message PutRatingRequest {
    string user_id = 1;
    string record_id = 2;
    int32 record_type = 3;
    int32 record_value = 4;
}
message PutRatingResponse {
}

Similarly add the code for the MovieService

service MovieService {
    rpc GetMovieDetails(GetMovieDetailsRequest) returns (GetMovieDetailsResponse);
}

message GetMovieDetailsRequest {
    string movie_id = 1;
}

message GetMovieDetailsResponse {
    MovieDetails movie_details = 1;
}

Now our movie.proto file includes the structure definitions and the API definitions for the
services.

We'll run the following command in the root directory to generate the code for these definitions.

protoc -I=api --go_out=. --go-grpc_out=. movie.proto

-I=api defines the input file path
--go_out=. defines the output path of the protobuf structure code
--go-grpc_out defines the output path of the service code.
movie.proto specifies the protobuf file containing definitions.

After execution, we'll find a movie_grpc.pb.go file in the /gen directory containing the code for the services.

Here's a small screenshot of what it'll look like.

We'll use this generated code in our Go services.

Generated Model vs Internal model

So currently we're having two versions of our model data structures. 1. The internal ones in metadata/pkg/model 2. The newly generated ones in gen directory.
Some of you might think that having two similar structures is redundant. (I also thought the same earlier). But, we need to understand that these structures differ in the their purpose.

Internal Model: The structures that we create to be used across our code base like in repository, controller etc.
Generated Model: The structures generated by tools like protoc compiler. These should only be used for serialization ie for transferring the data or storing the serialized data.

We should not use the generated structures across the application for the following reasons:

Coupling between application and serialization format. Switching to other serialization would requre full codebase changes.
Generated code (function definitions etc) may vary from version to version. This might cause your code which is using the generated code to break.
Generated Code is harder to use. Eg: In case of protobufs the field values are optional. Hence it'll require having multiple field != nil checks across the project.

Hence, we'll need some mapping logic to convert the data from one model to other. We'll add a file called mapper.go in the directory metadata/pkg/model

package model

import "movieexample.com/gen"

// MetadataToProto converts a Metadata struct into a generated proto counterpart
func MetadataToProto(m *Metadata) *gen.Metadata {
    return &gen.Metadata{
        Id:          m.ID,
        Title:       m.Title,
        Description: m.Description,
        Director:    m.Director,
    }
}

// MetadataFromProto converts a generated proto counterpart into a Metadata struct
func MetadataFromProto(m *gen.Metadata) *Metadata {
    return &Metadata{
        ID:          m.Id,
        Title:       m.Title,
        Description: m.Description,
        Director:    m.Director,
    }
}

Implementing clients and gateways

We'll now plugin the generated code into our microservices. This will help us to switch our communication protocol from the JSON based HTTP to Protobufs based gRPC calls.

In metadata/internal/handler add grpc directory and add grpc.go file.

package grpc

import (
    "context"
    "errors"
    "log"

    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "movieexample.com/gen"
    "movieexample.com/metadata/internal/controller/metadata"
    "movieexample.com/metadata/pkg/model"
)

// Handler defines a movie metadata gRPC handler
type Handler struct {
    gen.UnimplementedMetadataServiceServer
    svc *metadata.Controller
}

// New creates a new movie metadata gRPC handler
func New(ctrl *metadata.Controller) *Handler {
    return &Handler{svc: ctrl}
}

// GetMetadata returns movie metadata by id
func (h *Handler) GetMetadata(ctx context.Context, req *gen.GetMetadataRequest) (*gen.GetMetadataResponse, error) {
    if req == nil || req.MovieId == "" {
        return nil, status.Errorf(codes.InvalidArgument, "nil req or empty id")
    }

    m, err := h.svc.Get(ctx, req.MovieId)
    if err != nil && errors.Is(err, metadata.ErrNotFound) {
        log.Printf("GetMetadata failed: Err: %v", err)
        return nil, status.Errorf(codes.NotFound, err.Error())
    } else if err != nil {
        return nil, status.Errorf(codes.Internal, err.Error())
    }

    return &gen.GetMetadataResponse{Metadata: model.MetadataToProto(m)}, nil
}

Let's see the key points regarding the implementation:

Our Handler implemented the function in the same format as the one defined in the generated MetadataServiceServer interface.
We use the MetadataToProto mapping function to transform our internal structures into generated ones.
We embed the UnimplementedMetadataServiceServer into our Handler. This is required by the Protobufs compiler.

Now we'll update the metadata/cmd/main.go file. Most of the code is from the previous parts

import (
    grpcHandler "movieexample.com/metadata/internal/handler/grpc"
)

func main() {
    // The service will be running on port "port"
    var port int
    flag.IntVar(&port, "port", 8081, "API Handler port")
    flag.Parse()
    log.Printf("Starting the movie metadata service on port %d", port)

    registery, err := consul.NewRegistery("localhost:8500")
    if err != nil {
        panic(err)
    }
    ctx := context.Background()
    instanceID := discovery.GenerateInstanceID(serviceName)

    if err := registery.Register(ctx, instanceID, serviceName, fmt.Sprintf("localhost:%d", port)); err != nil {
        panic(err)
    }

    go func() {
        for {
            if err := registery.HealthCheck(instanceID, serviceName); err != nil {
                log.Println("Failed to report healthy state: ", err.Error())
            }
            time.Sleep(1 * time.Second)
        }
    }()
    defer registery.Deregister(ctx, instanceID, serviceName)

    repo := memory.New()

        // Updated code

    svc := metadata.New(repo)
    h := grpcHandler.New(svc)
    lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%v", port))
    if err != nil {
        log.Fatal("failed to listen: %v", err)
    }
    srv := grpc.NewServer()
    gen.RegisterMetadataServiceServer(srv, h)
    if err := srv.Serve(lis); err != nil {
        panic(err)
    }
}

We instantiated our gRPC server and started listening for requests on it. Rest of the code is same as before.

Let's do the same with the rating service.
Create a file grpc.go in /rating/internal/handler/grpc

package grpc

import (
    "context"
    "errors"

    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "movieexample.com/gen"
    "movieexample.com/rating/internal/controller/rating"
    "movieexample.com/rating/pkg/model"
)

// Handler defines a gRPC rating API handler
type Handler struct {
    gen.UnimplementedRatingServiceServer
    svc *rating.Controller
}

// New creates a new gRPC rating API handler
func New(ctrl *rating.Controller) *Handler {
    return &Handler{svc: ctrl}
}

// GetAggregateRating returns the aggregated rating for a record.
func (h *Handler) GetAggregateRating(ctx context.Context, req *gen.GetAggregatedRatingRequest) (*gen.GetAggregatedRatingResponse, error) {
    if req.RecordId == "" || req.RecordType == "" {
        return nil, status.Errorf(codes.InvalidArgument, "empty record id or record type")
    }
    val, err := h.svc.GetAggregatedRatings(ctx, model.RecordID(req.RecordId), model.RecordType(req.RecordType))
    if err != nil && errors.Is(err, rating.ErrNotFound) {
        return nil, status.Errorf(codes.NotFound, err.Error())
    } else if err != nil {
        return nil, status.Errorf(codes.Internal, err.Error())
    }

    return &gen.GetAggregatedRatingResponse{RatingValue: val}, nil
}

// PutRating writes a rating for a given record.
func (h *Handler) PutRating(ctx context.Context, req *gen.PutRatingRequest) (*gen.PutRatingResponse, error) {
    if req == nil || req.RecordId == "" || req.UserId == "" {
        return nil, status.Errorf(codes.InvalidArgument, "nil req or empty user id or record id")
    }
    rating := &model.Rating{
        RecordID:   model.RecordID(req.RecordId),
        RecordType: model.RecordType(req.RecordType),
        UserID:     model.UserID(req.UserId),
        Value:      model.RatingValue(req.RecordValue),
    }
    if err := h.svc.PutRating(ctx, model.RecordID(req.RecordId), model.RecordType(req.RecordType), rating); err != nil {
        return nil, err
    }
    return &gen.PutRatingResponse{}, nil
}

Now similar to the metadata service, we'll update the main.go for rating i.e. rating/cmd/main.go

import (
    grpcHandler "movieexample.com/rating/internal/handler/grpc"
)

func main() {
    var port int
    flag.IntVar(&port, "port", 8082, "API Handler port")
    flag.Parse()
    fmt.Printf("Starting the movie rating service on port %d", port)

    registery, err := consul.NewRegistery("localhost:8500")
    if err != nil {
        panic(err)
    }

    ctx := context.Background()
    instanceID := discovery.GenerateInstanceID(serviceName)

    if err := registery.Register(ctx, instanceID, serviceName, fmt.Sprintf("localhost:%d", port)); err != nil {
        panic(err)
    }

    go func() {
        for {
            if err := registery.HealthCheck(instanceID, serviceName); err != nil {
                log.Println("Failed to report healthy state: ", err.Error())
            }
            time.Sleep(1 * time.Second)
        }
    }()
    defer registery.Deregister(ctx, instanceID, serviceName)

    repo := memory.New()
    ctrl := rating.New(repo)

    // New Code

    h := grpcHandler.New(ctrl)

    lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%v", port))
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    srv := grpc.NewServer()
    gen.RegisterRatingServiceServer(srv, h)
    srv.Serve(lis)
}

Note: There are other imports that I've not put in the code as it'll make this part even longer.

Now add a handler file for movie service. Add movie/internal/handler/grpc/grpc.go

package grpc

import (
    "context"
    "log"

    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "movieexample.com/gen"
    "movieexample.com/metadata/pkg/model"
    "movieexample.com/movie/internal/controller/movie"
)

type Handler struct {
    gen.UnimplementedMovieServiceServer
    ctrl *movie.Controller
}

func New(ctrl *movie.Controller) *Handler {
    return &Handler{ctrl: ctrl}
}

func (h *Handler) GetMovieDetails(ctx context.Context, req *gen.GetMovieDetailsRequest) (*gen.GetMovieDetailsResponse, error) {
    log.Printf("GetMovieDetails called: %v", req.MovieId)

    if req == nil || req.MovieId == "" {
        return nil, status.Errorf(codes.InvalidArgument, "nil req or empty id")
    }
    m, err := h.ctrl.Get(ctx, req.MovieId)

    // Handle the errors returned by the grpc response
    if e, ok := status.FromError(err); ok {
        switch e.Code() {
        case codes.NotFound:
            return nil, status.Errorf(codes.NotFound, err.Error())
        default:
            return nil, status.Errorf(codes.Internal, err.Error())
        }
    }

    return &gen.GetMovieDetailsResponse{
        MovieDetails: &gen.MovieDetails{
            Rating:   *m.Rating,
            Metadata: model.MetadataToProto(&m.Metadata),
        },
    }, nil
}

In the previous few steps, we've added logic on the server side to handle the client requests. We'll now add code that'll connect to that logic from the client side.

First, we'll add a file internal/grpcutil/grpcutil.go. This will contain the logic for connecting to any service. We're extracting that logic to a function in this file.

package grpcutil

import (
    "context"
    "math/rand"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "movieexample.com/pkg/discovery"
)

// ServiceConnection attemps to select a random service instance and returns a gRPC connection to it.
func ServiceConnection(ctx context.Context, serviceName string, registery discovery.Registery) (*grpc.ClientConn, error) {
    addrs, err := registery.Discover(ctx, serviceName)
    if err != nil {
        return nil, err
    }

    return grpc.Dial(addrs[rand.Intn(len(addrs))], grpc.WithTransportCredentials(insecure.NewCredentials()))
}

This function will pick a random instance of the service and connect to it.

Now we'll create gateway for our services.
Create a file as movie/internal/gateway/metdata/grpc/metadata.go
Add the following code

package grpc

import (
    "context"

    "movieexample.com/gen"
    "movieexample.com/internal/grpcutil"
    "movieexample.com/metadata/pkg/model"
    "movieexample.com/pkg/discovery"
)

// Gateway defines a gRPC gateway for movie metadata service
type Gateway struct {
    registry discovery.Registery
}

// New creates a new gRPC gateway for movie metadata service
func New(registry discovery.Registery) *Gateway {
    return &Gateway{registry: registry}
}

// Get retrieves movie metadata by movie id
func (g *Gateway) Get(ctx context.Context, id string) (*model.Metadata, error) {
    // Create a gRPC connection to the movie metadata service
    conn, err := grpcutil.ServiceConnection(ctx, "metadata", g.registry)
    if err != nil {
        return nil, err
    }
    defer conn.Close()

    client := gen.NewMetadataServiceClient(conn)
    resp, err := client.GetMetadata(ctx, &gen.GetMetadataRequest{MovieId: id})
    if err != nil {
        return nil, err
    }
    return model.MetadataFromProto(resp.Metadata), nil
}

Let's mention some key points:

We'll be using the grpcutil.ServiceConnection to connect to any service. This is the method that we'd created recently.
We converted the metadata from Protobufs to model type using MetadataFromProto function.

Now we'll create a gateway for the rating service. Create a file /movie/internal/gateway/rating/grpc/rating.go

package grpc

import (
    "context"

    "movieexample.com/gen"
    "movieexample.com/internal/grpcutil"
    "movieexample.com/pkg/discovery"
    "movieexample.com/rating/pkg/model"
)

// Gateway defines a gRPC gateway for rating service
type Gateway struct {
    registry discovery.Registery
}

// New creates a new gRPC gateway for rating service
func New(registry discovery.Registery) *Gateway {
    return &Gateway{registry}
}

// GetAggregatedRating returns the aggregated rating for a record or ErrNotFound if there are no ratings for it.
func (g *Gateway) GetAggregatedRating(ctx context.Context, recordID model.RecordID, recordType model.RecordType) (float64, error) {
    conn, err := grpcutil.ServiceConnection(ctx, "rating", g.registry)
    if err != nil {
        return 0, err
    }
    defer conn.Close()

    client := gen.NewRatingServiceClient(conn)
    resp, err := client.GetAggregatedRating(ctx, &gen.GetAggregatedRatingRequest{RecordId: string(recordID), RecordType: string(recordType)})
    if err != nil {
        return 0, err
    }
    return resp.RatingValue, nil
}

// PutRating adds a rating for a record
func (g *Gateway) PutRating(ctx context.Context, recordID model.RecordID, recordType model.RecordType, userID model.UserID, value model.RatingValue) error {
    conn, err := grpcutil.ServiceConnection(ctx, "rating", g.registry)
    if err != nil {
        return err
    }
    defer conn.Close()

    client := gen.NewRatingServiceClient(conn)
    putRatingRequest := &gen.PutRatingRequest{
        UserId:      string(userID),
        RecordId:    string(recordID),
        RecordType:  string(recordType),
        RecordValue: int32(value),
    }

    _, err = client.PutRating(ctx, putRatingRequest)
    if err != nil {
        return err
    }
    return nil
}

Now let's finally change the main function of the movie service.

func main() {
    var port int
    flag.IntVar(&port, "port", 8083, "Port to listen on")
    flag.Parse()

    // Register with consul
    registery, err := consul.NewRegistery("localhost:8500")
    if err != nil {
        panic(err)
    }

    ctx := context.Background()
    instanceID := discovery.GenerateInstanceID(serviceName)

    if err := registery.Register(ctx, instanceID, serviceName, fmt.Sprintf("localhost:%d", port)); err != nil {
        panic(err)
    }

    go func() {
        for {
            if err := registery.HealthCheck(instanceID, serviceName); err != nil {
                log.Println("Failed to report healthy state: ", err.Error())
            }
            time.Sleep(1 * time.Second)
        }
    }()
    defer registery.Deregister(ctx, instanceID, serviceName)

    log.Printf("Starting the movie service at port: %d\n", port)
    metadataGateway := metadataGateway.New(registery)
    ratingGateway := ratingGateway.New(registery)
    ctrl := movie.New(ratingGateway, metadataGateway)

    // New Code

    h := grpcHanlder.New(ctrl)
    lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%v", port))
    if err != nil {
        log.Fatal("failed to listen: err")
    }

    srv := grpc.NewServer()
    reflection.Register(srv)
    gen.RegisterMovieServiceServer(srv, h)
    if err := srv.Serve(lis); err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
}

Note: We also registered our server with the reflection package. This allows us to use API request tools(eg: Postman) and get information about the available methods.

We've now completed the coding part. We'll now be testing our implementation.

Testing

Run the main.go in each directory. Also remember to run the consul docker container before that. I added the command to a file docker-compose in the root directory.

services:
  dev-consul:
    image: hashicorp/consul
    command: agent -server -ui -node=server-1 -bootstrap-expect=1 -client=0.0.0.0
    ports:
      - 8500:8500
      - 8600:8600/udp

Also added a Makefile

consul:
    @docker compose up -d

You can now just run make consul and your consul service will be up.

Now after you've started all the services. Open the Postman app. Select a new request and select grpc request type.

Final Request can be:

On sending the request you'll be seeing the following:

We can see that the response is correct. We are sending the Not Found code as the response as we'd defined in our code.

Tadaaaaa. We've completed this part!!! This was a challenging part for me personally. Probably due to many things that I saw for the first time.
I also felt that JSON should be used if you're not thinking about scaling and implementing some service which is not going to have very high traffic. JSON is trivial and is easy to implement and debug.
But Protobufs give a high advantage over JSON when optimizing is important.

Here's the github repo: Github

Learning Microservices with Go(Part 3). Serialization

Manav Kushwaha — Mon, 05 Feb 2024 12:11:45 +0000

Intro

This post is the third part of the "Learning Microservices with Go". I'm writing these posts as I'm learning the concepts. If you haven't checked out the part 2, here's the link for it: Part 2.
Link for Git Repo: Github

In part 2, we implemented the service discovery for our movie app. It allowed us to create multiple instances of our services and run them without hard-coding the ports and addresses.

For this part, we'll be exploring serialization, various serialization formats, their comparison, implementation and some simple benchmarks.

The basics of serialization

Serialization is the process of converting data, into some format that which allows us to store it, transfer it and deconstruct it back.

Uses Of Serialization

There are 2 primary use cases for doing serialization:

Transferring data between services and acting as the common language between them.
Encoding and decoding the data for storage. Allowing to store complex data structures as strings or as bytes.

If you try to think, you've probably used one serialization format quite a bit. We ourselves used it in communication with the HTTP API endpoints. And that is ...... JSON.
JSON allowed us to transform the data when sending back responses from the endpoint and decode them on other side.

In Go, we have annotations that help the JSON encoder to transform our object into an output.

// Metadata defines the movie metadata.
type Metadata struct {
  ID          string `json:"id"`
  Title       string `json:"title"`
  Description string `json:"description"`
  Director    string `json:"director"`
}

Let's try encoding an instance of this Metadata structure.

Metadata{
  ID:          "1",
  Title:       "Saving Private Ryan",
  Description: "Movie about WW2",
  Director:    "Steven Spielberg",
}

The result after encoding will be:

{"id":"1","title":"Saving Private Ryan","description":"Movie about WW2","director":"Steven Spielberg"}

Some more use cases of serialization:

Store configuration: Eg: tsconfig.json is used to specify the typescript configurations in many projects.
Storing record in Database: Many databases use JSON to store arbitrary data. Eg: Key-Value databases require converting the value to byte arrays. JSON is often used for this.

JSON

JSON is the most widely used serialization format.

Most of the languages have tools/libraries/packages to handle encoding/decoding JSON.
It has a great browser support. All major browsers including developer tools support working with JSON in the browser.
It's having an easier readability and it's easier to use during development and debugging.

Limitations of JSON:

Size: JSON is not a size efficient format. There are formats having lower encoded sizes for the same data.
Speed: Similar to the size, the encoding and decoding speed is not the fastest against some other serialization protocols.

Serialization Protocols

Some of the most popular/used formats(other than JSON) are:

XML
YAML
Apache Thrift
Protocol Buffers

Let's see brief overview and differences between them.

XML
XML was one of the earliest serialization format. It represents data as a tree of nodes. You'll find it similar to the HTML structure. Our metadata struct would be encoded as:

<Metadata><ID>1</ID><Title>Saving Private Ryan</
Title><Description>Movie about WW2</
Description><Director>Steven Spielberg</Director></Metadata>

One downside of XML is the size of the output. You can see that the output is longer than JSON.

YAML
Yaml is one of the most popular serialization format. It's designed to be much more human readable. Our metadata struct would be encoded as:

metadata:
  id: 123
  title: Saving Private Ryan
  description: Movie about WW2
  director: Steven Spielberg

The formats that we've discussed so far are primarily used for defining and serializing arbitrary types of data.
Some solutions are more focused towards using the serialization format mainly for communication between services. For eg:

Apache Thrift
Apache Thrift is a combination of serialization and a communication protocol. It was created by Facebook but became open source later.
Unlike the json, xml, it requires you to define your structures in .thrift files.

struct Metadata {
  1: string id,
  2: string title,
  3: string description,
  4: string director
}

Thrift code generator uses the thrift file to generate code for structures in chosen programming language along with encoding, decoding logic.

Thrift has a higher encoding, decoding speed compared to JSON, XML and also the serialized data can be 30-50% smaller in size than XML, JSON.
Few disadvantages that thrift has is the lack of official documentation. Most of the documentation is unofficial. The serialized data is not readable. Thus there also has been a low adoption rate. And there are more popular and efficient formats.

Protocol Buffers
Protocol buffers or protobufs is a serialization format that was created at google more than 20 years ago. It became open source later. It has some benefits like:

Small data output size along with high speed in serialization and deserialization.
It allows to define data structures that can be used both by client and server code in multiple languages.
It's having official support by Google and widespread use in microservices.

Using Protocol Buffers

We'll use protocol buffers now and see how it compares to other formats.

We'll first install the protocol buffer compiler and the go plugin for the same:

Download the latest version from the Website and follow the Readme.

You can also use the apt repository(for Debian based Linux). NOTE: The version of the compiler might be old if you install using apt.

sudo apt update
sudo apt install protobuf-compiler

To generate Go code from the protobuf compiler, we'll install the Go plugin for protobuf.

go install google.golang.org/protobuf/cmd/protoc-gen-go@latest

Now we have all the tools needed.

Create an api directory in the root and add movie.proto file under it with the following content. File: /api/movie.proto

syntax = "proto3";
option go_package = "/gen";
message Metadata {
  string id = 1;
  string title = 2;
  string description = 3;
  string director = 4;
}

message MovieDetails {
  float rating = 1;
  Metadata metadata = 2;
}

In the code above, we first added the protobuf syntax version we're using. Then we defined the output path of the code that'll be generated based on this file.
The rest is the two structures similar to what we defined in the first part of the series. Link

Now we'll generate the Go code from this file. In the root dir, run the following command.

protoc -I=api --go_out=. movie.proto

If the command is successful you'll see a gen directory at the root. The directory will have a file called movie.pb.go. It includes our structures and code to serialize and deserialize them.

For the above command,

-I=api specifies the input directory for the protobuf file
--go_out=. specifies the output directory as the current directory for the generated go code.
movie.proto is the protobuf file name which has the actual content.

Note: The protobuf file itself adds the generated code to the /gen package. Hence the final destination of the files would be /gen instead of the root.

Here's what we achieved. We created a movie.proto file that defines our data schema. It provides these benefits to us:

Explicit schema definition: Our schema is decoupled from the code and is independent now.
Code Generation: Our schema is now converted to code via code generation. We'll send data between services via the code in next part.
Cross-language support: We can generate the code for any language. If there's a service that's using another language, we can just generate the structures for it. Earlier it wasn't possible as the structures were coupled to Go.

Benchmarking

We'll do a quick benchmark between XML, JSON, Protobufs on the basis of size of serialized data and speed of encoding.

Encoding Size Benchmark

Add the following contents to a file /cmd/sizecompare/main.go

package main

import (
    "encoding/json"
    "encoding/xml"
    "fmt"

    "google.golang.org/protobuf/proto"
    "movieexample.com/gen"
    "movieexample.com/metadata/pkg/model"
)

// This is the model struct from the metadata package
var metadata = &model.Metadata{
    ID:          "1",
    Title:       "Saving Private Ryan",
    Description: "Movie about WW2",
    Director:    "Steven Spielberg",
}

// This is the generated struct from the protobuf definition
var genMetadata = &gen.Metadata{
    Id:          "1",
    Title:       "Saving Private Ryan",
    Description: "Movie about WW2",
    Director:    "Steven Spielberg",
}

func main() {
    jsonBytes, err := serializeToJSON(metadata)
    if err != nil {
        panic(err)
    }
    xmlBytes, err := serializeToXML(metadata)
    if err != nil {
        panic(err)
    }
    protoBytes, err := serializeToProto(genMetadata)
    if err != nil {
        panic(err)
    }
    fmt.Printf("JSON size:\t%dB\n", len(jsonBytes))
    fmt.Printf("XML size:\t%dB\n", len(xmlBytes))
    fmt.Printf("Proto size:\t%dB\n", len(protoBytes))
}

Now let's add implemtation of the functions used


func serializeToJSON(m *model.Metadata) ([]byte, error) {
    return json.Marshal(m)
}
func serializeToXML(m *model.Metadata) ([]byte, error) {
    return xml.Marshal(m)
}
func serializeToProto(m *gen.Metadata) ([]byte, error) {
    return proto.Marshal(m)
}

Do go mod tidy to clean up imports.

Now let's do go run main.go inside the directory with this file.

And the results are....

We see that the XML encoding is about 40% bigger than the JSON one. And protobufs is around 40% smaller than JSON. This is just an illustration of how switching to protobufs can just reduce the data that's being sent over the network.

Serialization speed benchmark

For this, we use the testing package from the Go library which will do an automated performance test and measure how fast the operation is.

Create a file main_test.go in the same directory and add the following content to it.

package main

import (
    "testing"
)

// Benchmark functions need to be in a file ending with _test.go
// and have a function name starting with Benchmark

func BenchmarkSerializeToJSON(b *testing.B) {
    for i := 0; i < b.N; i++ {
        serializeToJSON(metadata)
    }
}
func BenchmarkSerializeToXML(b *testing.B) {
    for i := 0; i < b.N; i++ {
        serializeToXML(metadata)
    }
}
func BenchmarkSerializeToProto(b *testing.B) {
    for i := 0; i < b.N; i++ {
        serializeToProto(genMetadata)
    }
}

Every function must run the target code b.N times. During benchmark execution, b.N is adjusted until the benchmark function lasts long enough to be timed reliably.

Now we'll run the test by running: go test -bench=.

And the results for the speed benchmark are....

The output means that the loop ran 2552851 times at a speed of 501.4 ns per loop for the case of JSON. And similarly for others.

We see that JSON serialization was almost 2 times slower than the Protobufs. XML serialization however had almost an order of difference with Protobufs.

NOTE: Be careful of these comparisions. The benchmark was for the serialization operations and hence had a big difference. In some applications, you may not be performance bound by serialization and instead could be doing other heavy work. For that case you'll see a lower difference in performance.
Just a good to have in mind when seeing any benchmark is that everything is context dependent.

Tadaaa. We've completed understanding serialization and compared the different protocols for serialization.

In the next part we'll continue to use protobufs and will implement their use in our services.

Please do like the post if you found it helpful and learned something.
Checkout my twitter at: manavkush
Link for Git Repo: Github
Link for Reference Book: Book

See you next post.

Learning Microservices with Go(Part 2). Service Discovery

Manav Kushwaha — Tue, 30 Jan 2024 14:58:23 +0000

This post is the second part of the "Learning Microservices with Go". I'm writing these posts as I'm learning the concepts. If you haven't checked out the part 1, here's the link for it: Part 1.
Link for Git Repo: Github

In part 1, we created the directory structure for our movie app. We created 3 services namely metadata, rating, movie services.

The problem that we're facing was that these services were running, interacting with each other using hard-coded addresses and ports. This works fine only till we run all the services on the same machine. We also need to be able to create multiple instances of any service which is currently not possible(This is important if we want to build a real application which can scale. It'll give us high availability and more fault tolerance).

To tackle this, we will use a service discovery. You may ask what's a service discovery. Service discovery addresses multiple problems:

1. How to find instance(s) of a particular service.
2. How to dynamically add or remove an instance of any service.
3. How to handle unresponsive instances of any service.

In simplicity, service discovery can be viewed as a registry, storing the information about the available services. It has the following features:

1. Adding(Registering) an instance of a service.
2. Removing(Deregistering) an instance of a service.
3. Get a list of all instances of a service. (list of network addresses of the instances)

Service Discovery Models

The service discovery can be implemented in two ways.

Client-side service discovery: The application directly accesses the registry directly using a registry client.

Server-side service discovery: The application access the registry via a load balancer, which forwards the request to the available instances.

Some differences:

In case of the client side service discovery, the logic for calling the instance needs to be with the client and be implemented for every application. If this logic is not correctly configured, the application may end up calling only 1 instance of the service, leading to overload of the instance and under-utilization of other instances.
In case of the server-side service discovery, the logic for calling the instance is with the load balancer. The load balancer gets the list of instances from the service discovery and decides which instance should the request be directed to and forwards the request to the chosen instance. This decouples the logic for service directory interaction from the application.

In this series, we'll be covering the client-side discovery models only as they're simpler to implement and don't require setting up of load balancers.

Tools available for service discovery

Hashicorp Consul: This has been a very simple solution for service discovery for some years. It has a simple API, which includes:
- PUT /catalog/register: Registers a service instance.
- PUT /catalog/deregister: Deregisters a service instance.
- GET /catalog/services: Gets the instances of the service that are available.
Kubernetes: Kubernetes is a very widely used container orchestration tool. It's used to manage, deploy, scale containers(which often contain microservices). It has a feature to register and find services running on it. It allows for a client-side discovery model as well as pluging a load balancer for server-side discovery model.

We'll use the hashicorp consul for this tutorial. We'll learn to use Kubernetes in future parts of the series.

Coding the service discovery

We'll add the service directory to the pkg folder as it'd be accessed by all the services to discover other services.

Create a file pkg/discovery/discovery.go. Add the following code to that.

package discovery

import (
    "context"
    "errors"
    "fmt"
    "math/rand"
    "time"
)

type Registery interface {
    // Register a service with the registry.
    Register(ctx context.Context, instanceID string, serviceName string, hostPort string) error
    // Deregister a service with the registry.
    Deregister(ctx context.Context, instanceID string, serviceName string) error
    // Discover a service with the registry.
    Discover(ctx context.Context, serviceName string) ([]string, error)
    // HealthCheck a service with the registry.
    HealthCheck(instanceID string, serviceName string) error
}

// ErrNotFound is returned when no service addresses are found.
var ErrNotFound = errors.New("no service addresses found")

// GenerateInstanceID generates a psuedo-random service instance identifier, using a service name. Suffixed by dash and number
func GenerateInstanceID(serviceName string) string {
    return fmt.Sprintf("%s-%d", serviceName, rand.New(rand.NewSource(time.Now().UnixNano())).Int())
}

What we added was an interface of a service discovery. We added that so as if in future if we switch our service discovery tool(eg: from Consul to some other tool), then we won't be making changes all over the code. The newer discovery could use the same interface.
Also similar to the way we have a memory repository in the last part, we can have a memory discovery as well. It would provide us ability to test our discovery.

In memory service discovery

Create the file pkg/discovery/memory/memory.go and add the following code.

package memory

import (
    "context"
    "errors"
    "sync"
    "time"

    "movieexample.com/pkg/discovery"
)

type serviceNameType string
type instanceIDType string

type Registery struct {
    sync.RWMutex
    serviceAddrs map[serviceNameType]map[instanceIDType]*serviceInstance
}

type serviceInstance struct {
    hostPort   string
    lastActive time.Time
}

// NewRegistry creates a new in-memory registry instance.
func NewRegistry() *Registery {
    return &Registery{
        serviceAddrs: make(map[serviceNameType]map[instanceIDType]*serviceInstance),
    }
}

// Register creates a service record in the registry
func (r *Registery) Register(ctx context.Context, instanceID string, serviceName string, hostPort string) error {
    r.Lock()
    defer r.Unlock()

    if _, ok := r.serviceAddrs[serviceNameType(serviceName)]; !ok {
        r.serviceAddrs[serviceNameType(serviceName)] = map[instanceIDType]*serviceInstance{}
    }
    r.serviceAddrs[serviceNameType(serviceName)][instanceIDType(instanceID)] = &serviceInstance{
        hostPort:   hostPort,
        lastActive: time.Now(),
    }
    return nil
}

// Deregister removes a service record from the registry
func (r *Registery) Deregister(ctx context.Context, instanceID string, serviceName string) error {
    r.Lock()
    defer r.Unlock()

    if _, ok := r.serviceAddrs[serviceNameType(serviceName)]; !ok {
        return discovery.ErrNotFound
    }

    delete(r.serviceAddrs[serviceNameType(serviceName)], instanceIDType(instanceID))
    return nil
}

// HealthCheck marks a service instance as active
func (r *Registery) HealthCheck(instanceID string, serviceName string) error {
    r.Lock()
    defer r.Unlock()

    if _, ok := r.serviceAddrs[serviceNameType(serviceName)]; !ok {
        return errors.New("service not registered yet")
    }

    if _, ok := r.serviceAddrs[serviceNameType(serviceName)][instanceIDType(instanceID)]; !ok {
        return errors.New("service instance not registered yet")
    }

    r.serviceAddrs[serviceNameType(serviceName)][instanceIDType(instanceID)].lastActive = time.Now()
    return nil
}

// Discover returns a list of service instances from the registry
func (r *Registery) Discover(ctx context.Context, serviceName string) ([]string, error) {
    r.RLock()
    defer r.RUnlock()

    if len(r.serviceAddrs[serviceNameType(serviceName)]) == 0 {
        return nil, discovery.ErrNotFound
    }
    var res []string

    for _, v := range r.serviceAddrs[serviceNameType(serviceName)] {
        if time.Since(v.lastActive) > 5*time.Second {
            continue
        }

        res = append(res, v.hostPort)
    }
    return res, nil
}

We can see that the in-memory service discovery implements all the methods of the Registry interface we had defined earlier. We used a map of map to store the service addresses. sync.RWMutex is used to allow concurrent read, writes to the map.
Now we can use this implementation wherever we want.

Consul based service discovery

Create a file pkg/discovery/consul/consul.go
Add the following code to it.

package consul

import (
    "context"
    "errors"
    "fmt"
    "strconv"
    "strings"

    consul "github.com/hashicorp/consul/api"
)

// Registery defines a consul based service registry
type Registery struct {
    client *consul.Client
}

// NewRegistery creates a new consul registry instance.
// Addr is the address where the consul agent is running
func NewRegistery(addr string) (*Registery, error) {
    config := consul.DefaultConfig()
    config.Address = addr
    client, err := consul.NewClient(config)
    if err != nil {
        return nil, err
    }
    return &Registery{client: client}, nil
}

// Register creates a service record in the registry
func (r *Registery) Register(ctx context.Context, instanceID string, serviceName string, hostPort string) error {
    parts := strings.Split(hostPort, ":")
    if len(parts) != 2 {
        return errors.New("invalid host:port format. Eg: localhost:8081")
    }
    port, err := strconv.Atoi(parts[1])
    if err != nil {
        return err
    }
    host := parts[0]

    err = r.client.Agent().ServiceRegister(&consul.AgentServiceRegistration{
        Address: host,
        Port:    port,
        ID:      instanceID,
        Name:    serviceName,
        Check: &consul.AgentServiceCheck{
            CheckID: instanceID,
            TTL:     "5s",
        },
    })
    return err
}

// Deregister removes a service record from the registry
func (r *Registery) Deregister(ctx context.Context, instanceID string, _ string) error {
    err := r.client.Agent().ServiceDeregister(instanceID)
    return err
}

// HealthCheck is a push mechanism to update the health status of a service instance
func (r *Registery) HealthCheck(instanceID string, _ string) error {
    err := r.client.Agent().UpdateTTL(instanceID, "", "pass")
    return err
}

// Discover returns a list of addresses of active instances of the given service
func (r *Registery) Discover(ctx context.Context, serviceName string) ([]string, error) {
    entries, _, err := r.client.Health().Service(serviceName, "", true, nil)
    if err != nil {
        return nil, err
    }
    var instances []string
    for _, entry := range entries {
        instances = append(instances, fmt.Sprintf("%s:%d", entry.Service.Address, entry.Service.Port))
    }
    return instances, nil
}

As we're using consul which is an external library, do remember to do go mod tidy in the src directory.

Modifying files to use service discovery

Change the two gateways that we'd defined earlier for metadata, rating services

type Gateway struct {
    registry discovery.Registry
}

Changing the New function

func New(registry discovery.Registry) *Gateway {
    return &Gateway{registry}
}

Now the gateways require registry when being created. Change the Get function of the rating gateway as the following:

    addrs, err := g.registery.Discover(ctx, "rating")
    if err != nil {
        return 0, err
    }
    if len(addrs) == 0 {
        return 0, fmt.Errorf("no rating service instances available")
    }

    url := "http://" + addrs[rand.Intn(len(addrs))] + "/rating"
    log.Printf("Calling rating service. Request: GET %s\n", url)

    req, err := http.NewRequest(http.MethodGet, url, nil)

Earlier we're calling a pre-configured address. Now we'll get a list of addresses from the service discovery. We choose a random instance from the list.

Update the metadata service in the same way.

Now, we'll update the main functions of our services so that they can register and deregister to the service discovery.

Update the rating service main function as:

func main() {
    var port int
    flag.IntVar(&port, "port", 8082, "API Handler port")
    flag.Parse()
    fmt.Printf("Starting the movie rating service on port %d", port)

    registery, err := consul.NewRegistery("localhost:8500")
    if err != nil {
        panic(err)
    }

    ctx := context.Background()
    instanceID := discovery.GenerateInstanceID(serviceName)

    if err := registery.Register(ctx, instanceID, serviceName, fmt.Sprintf("localhost:%d", port)); err != nil {
        panic(err)
    }

    go func() {
        for {
            if err := registery.HealthCheck(instanceID, serviceName); err != nil {
                log.Println("Failed to report healthy state: ", err.Error())
            }
            time.Sleep(1 * time.Second)
        }
    }()
    defer registery.Deregister(ctx, instanceID, serviceName)

    repo := memory.New()
    ctrl := rating.New(repo)
    h := httpHandler.New(ctrl)

    http.HandleFunc("/rating", http.HandlerFunc(h.Handle))
    if err := http.ListenAndServe(fmt.Sprintf(":%d", port), nil); err != nil {
        panic(err)
    }
}

Do the same for the metadata service.

The final thing that we need to do is to update the main function of the movie service.

func main() {
    var port int
    flag.IntVar(&port, "port", 8083, "Port to listen on")
    flag.Parse()

    // Register with consul
    registery, err := consul.NewRegistery("localhost:8500")
    if err != nil {
        panic(err)
    }

    ctx := context.Background()
    instanceID := discovery.GenerateInstanceID(serviceName)

    if err := registery.Register(ctx, instanceID, serviceName, fmt.Sprintf("localhost:%d", port)); err != nil {
        panic(err)
    }

    go func() {
        for {
            if err := registery.HealthCheck(instanceID, serviceName); err != nil {
                log.Println("Failed to report healthy state: ", err.Error())
            }
            time.Sleep(1 * time.Second)
        }
    }()
    defer registery.Deregister(ctx, instanceID, serviceName)

    log.Printf("Starting the movie service at port: %d\n", port)
    metadataGateway := metadataGateway.New(registery)
    ratingGateway := ratingGateway.New(registery)

    ctrl := movie.New(ratingGateway, metadataGateway)
    h := httpHandler.New(ctrl)
    http.Handle("/movie", http.HandlerFunc(h.GetMovieDetails))
    if err := http.ListenAndServe(fmt.Sprintf(":%d", port), nil); err != nil {
        panic(err)
    }
}

We've completed the code for implementing service discovery.

Testing

Now we'll need to run the Consul app to allow our services to use it. We can do this by using Docker. Run the following command. This will spin up consul in a docker container.

 docker run \                                                                                                                                                                 
    -d \
    -p 8500:8500 \
    -p 8600:8600/udp \
    --name=dev-consul \
    hashicorp/consul agent -server -ui \
-node=server-1 -bootstrap-expect=1 -client=0.0.0.0

We'll now run each microservice inside each cmd directory.
go run *.go
Now going to the consul web ui running at localhost:8500, you'll see list of services that are active.
Try running multiple instances of any of the services by:
go run *.go --port <Port>
Remember to use ports that are not already being used.
Test the API by sending the request to the movie service.
curl -v localhost:8083/movie?id=1

You'll find the following in the shell running movie service.

2024/01/28 10:19:55 Calling metadata service. Request: GET http://localhost:8081/metadata

This means that the movie service was able to interact with the service discovery and then direct the request to the correct address for the service.

Tadaaa. We've completed the implementation of service discovery.

In the next part we'll learn serialization(used to encode the data transferred between the services).

Do like the post if you found it helpful and learned something.
Checkout my twitter at: manavkush
Link for Git Repo: Github
Link for Reference Book: Book

See you next post.

Learning Microservices with Go: The Journey begins

Manav Kushwaha — Sun, 28 Jan 2024 04:56:09 +0000

Intro
This article is about my ongoing journey of learning microservices along with Golang. This is the first of the series of articles I'll try writing to document my journey.

The Journey Begins
Recently I was working on a chat-application backend rewrite in Go. I'd already made that app using NextJS previous year as my personal project. After that, I'd tried Go and fallen in love with it.
I started to build small programs, started reading about Go through videos, articles. I also got a book for learning more about go itself called Learning Go. Going through the book was a great decision, I started to understand the fundamentals of the language.
Very soon, I leaped into the task with very little knowledge about structuring and scaffolding the project. Very soon I went into making bad design choices and ended up with bad code.
This is a necessary step I feel. You learn more from the mistakes. So, for the next try, I tried exploring about the common project structure that's used in Go. I quickly found out about a standard structure that I found common in many open-source projects that I saw. I followed the principles and got into building the second version of the app. This time the things went smoother. The structure made me write better code and I got much further.
The mistake that I made was that even though I was trying to follow the structure, my logic was interlinked with the other parts of code(eg: Database interaction layer was also having app logic). This intertwined code caused me to accept that I need to follow the guidelines that I'd read previously.
I had also tried using a repository called go-blueprint by Melkey, a well-known creator and streamer on Youtube and Twitch.

I tried using the echo framework and a postgres db. It was a bit new for me to understand the default structure that I'd gotten from the go-blueprint. There were concepts like context, services, etc which I was not really comfortable with. I really didn't need to know these when starting building something, but I was just curious to learn. The blueprint structure to me was seeming to be taking care of the microservice architecture also. I figured this would be because Go is extensively used with microservices. This was one thing that I'd not been very comfortable with. So as always, I just jumped into the fire.

This was the point that I searched for some resources for Go and found a book called Microservices with Go.

Learning the basic structure

I'm finding the book a good one to code along. By this article I've read and coded along and made my first app with microservices.
The book started with explaining about the directory structure and uses of them. I was already having knowledge about them from my previous unsuccessful attempts.

Then we started making a simple movie app. The application was simple, there would be functionality to get information about any movie. Also we would be able to rate the movie itself.

The book focused on differentiating and separating the responsibilities first.

Storing the movie info(metadata) would be a onetime thing. We'd generally not be modifying the data once the movie details were entered.
Storing the ratings would be different as there would be more ratings coming in overtime and there would be a need for aggregation for the ratings.

This separation of ways that we're using and storing the data is important to consider when thinking about services.
Another thing that the book did was to focus on extensibility. Instead of storing just the movie ratings, we could store multiple types of ratings(there might be a need of actor ratings in the near future). And to implement it was just adding a rating type field while storing the ratings. Such tradeoffs are considered beneficial as they allow us to be open to future requirement changes with minor changes without going overboard.

So we considered the final structure as the following:

The metadata and rating service could be independent and could maintain different storage solutions. The movie service could just be using the metadata, rating services to get the results for any movie.

The book then dives into a code along, one service and one file at a time. While creating each file, it tells the idiomatic ways of doing things. It also leaves room for upgrading our service which we'll do in further parts(eg: currently we're using http for communicating between the services, we'll also look at integrating RPCs further ahead.)

The final structure for the repository can be found here

The chapter ends with a basic working microservice application.

Currently all the services are using ports of a single machine. They will not work if they're run on seperate machines. We'll solve the problem in the upcoming parts of the series.

It was fun building my first microservice app. Will post the next part asap as I'm done learning it.

See you there!!!
Byee

How Tech Youtubers saved my Software Developer Career.

Manav Kushwaha — Tue, 21 Nov 2023 18:01:41 +0000

This might seem to you a clickbait title. It very well might be. Only further reading may reveal.

So, let me tell a bit about myself. Throughout my college time, I'd focused more on learning Valorant lineups than learning software engineering. I was (maybe still am?) what people call ... a Javascript Andy.
I had known C++ from my attempts at competitive programming in earlier years. Apart from that, I'd made few clone projects using MERN stack (religiously following guides from udemy courses).
I can still remember during interviews people asked me why had I used MongoDb. Words that'd been sketched into my mind by blindly following the tutorials came out instantly. "For better scaling".

Years went by and here I was in my last semester in college. I'd gotten a job offer as a software developer, and was working in the same company as an intern for the semester.

During my internship, I wanted to explore my options. But the scarcity in tech jobs had already been going on and most of them were having significant years of experience in their requirements.
Also remember those projects that I'd told you about... Yeah, I forgot most of the technical stuff. (You can say... following the guides without exploring more, wasn't such a great idea)

Somehow during this time, I came across a youtube channel called theo-t3.gg. It was my saving grace.
One of the first video on the channel that I watched was about choosing the right database. I watched that whole video along with my beautiful girlfriend (better software engineer than me).
We both had an epiphany seeing that. All that talks about scaling and stuff had clouded our judgement about relational dbs. SQL is really good. Watching similar videos on other topics made me realize that my knowledge had been very shallow. This was the onset of my journey.

We both started working on a project together. This was the first time I'd set on making a project that wasn't following a guide...

That's a lot of progress...

There was also a major shift in the attitude. Like the initial rush that'd been when we'd taken up computer science, learning about this computer world. It all had come back!!!

Another thing about me was that I'd always used VSCode up until that time. The only place I'd seen vim was during the streams of competitive programmers that would almost always use vim.
I'd always go and say why do people even use vim when "Modern IDEs are magic".

Skill Issue.

That was what was actual reason. I'd seen people be so fast with vim, but the only thing that I got to know after almost 4 years, was to exit vim. (Forgetting -m flag in git commit).

Watching theo changed my recommendation of youtube. I was enlightened by the Primeagen himself (in youtube shorts, where he had just created The Best Macro of All Time).

Thus began my neovim era. Every day I was getting more hyped by watching Prime just being able to code so fast. I had been vim-pilled.
I tried out my newly installed neovim. I began the vim tutor and after about an hour of it, I was bored. I was so slow. And that dream of becoming fast was too ahead.

I switched back to vscode. But I hadn't stopped watching the Chad prime everyday just flexing him vim skills on me.

The thing that finally got me back completely was .... Kickstart.
I watched the effective IDE video of Teej, and after sometime, my neovim setup was complete.
Teej went on to explain the different settings, but I skipped it instantly. Same habbit of throwing the manual away was still with me after all these years.

Kickstart got me into customizing neovim step by step. I didn't knew anything about Lazy, Mason, Treesitter, Telescope.
The first config that I changed was to enable custom plugins. Customizing neovim step by step to my flavour gradually helped me understand things.

And all this time I'd forced myself to use neovim. So naturally my movement also became faster. I was on the same level as I'd been before switching due to better movements. I've gotten so used to it that I can't ever imagine of going anywhere else.

Apart from the editor wars, I had also started watching more and more tech content. My Javascript bubble had already been burst by Prime. So I was expanding my horizon. I explored Go. I tried Svelte. I used HTMX. I'd always been relectant, but now it's the curiosity that drives me.

Remember all this happened in a span of less than 5-6 months. This progress was so much more than what I'd made in like the last 2 years. Doing all this has also boosted my confidence. I am knowing more things, and am able to indulge in more discussions than any of my friends combined.
I am learning, reading(or let other people read and make a video of themselves and then watch that video. "Same thing") everyday. Also the memes are the best.

My journey has just started as a developer. But the change that I've noticed in myself is huge. Instead of just going with the flow in my career, I've taken charge of it. I'm actively buidling stuff, learning things which is helping me grow much more rapidly than before. And I'd really thank all the people who've helped me through my journey.
I've added some of them in a list below.

Thanks for making software engineering interesting again!!!

Some of the people that've helped me in this journey:

My Girlfriend ( always there supporting me. Also she's beautiful)
Primeagen ( having a great content that you love to wait for everyday)
Theo ( restarting my learning journey in tech and being one of the best explainers)
Josh Tried Coding (Great content on typescript/react w)
Teej ( Catapulted me into the neovim world.)
Devops Toolbox ( Really cool channel that taught me alot about cli tools like tmux and also neovim plugins)
Nerd Academy ( Great explainations. Used for Golang learning)

How did you find my first article? Any suggestions? Do connect with me on LinkedIn or Twitter