DEV Community: Ramón Berrutti

gRPC Streaming: Best Practices and Performance Insights

Ramón Berrutti — Sat, 30 Nov 2024 20:56:44 +0000

Introduction

gRPC streaming allows protobuf messages to be streamed from client to server, server to client, or bidirectionally.
This powerful feature can be used to build real-time applications such as chat applications, real-time monitoring dashboards, and more.

In this article, we will explore how to use gRPC streaming correctly.

Prerequisites

Basic knowledge of gRPC
Basic knowledge of Go programming language (The sample code is written in Go, but the concept can be applied to other languages as well)
The code examples are available on GitHub

Good Practices

Let's check the good practices for using gRPC streaming:

Use unary request for unary request

One common mistake is to use streaming for unary requests.
For example, consider the following gRPC service definition:

service MyService {
  rpc GetSomething (SomethingRequest) returns (stream SomethingResponse) {}
}

If the client only needs to send one request and receive one response,
You don't need to use streaming. Instead, we can define the service as follows:

service MyService {
  rpc GetSomething (SomethingRequest) returns (SomethingResponse) {}
}

By using streaming for unary requests, we are adding unnecessary complexity
to the code, which can make it harder to understand and maintain and not
gaining any benefits from using streaming.

Golang code example comparing unary request and streaming request:

Unary request:

type somethingUnary struct {
    pb.UnimplementedSomethingUnaryServer
}

func (s *somethingUnary) GetSomething(ctx context.Context, req *pb.SomethingRequest) (*pb.SomethingResponse, error) {
    return &pb.SomethingResponse{
        Message: "Hello " + req.Name,
    }, nil
}

func TestSomethingUnary(t *testing.T) {
    conn := newServer(t, func(s grpc.ServiceRegistrar) {
        pb.RegisterSomethingUnaryServer(s, &somethingUnary{})
    })

    client := pb.NewSomethingUnaryClient(conn)

    response, err := client.GetSomething(
        context.Background(),
        &pb.SomethingRequest{
            Name: "test",
        },
    )
    if err != nil {
        t.Fatalf("failed to get something: %v", err)
    }

    if response.Message != "Hello test" {
        t.Errorf("unexpected response: %v", response.Message)
    }
}

Streaming unary request:

type somethingStream struct {
    pb.UnimplementedSomethingStreamServer
}

func (s *somethingStream) GetSomething(req *pb.SomethingRequest, stream pb.SomethingStream_GetSomethingServer) error {
    if err := stream.Send(&pb.SomethingResponse{
        Message: "Hello " + req.Name,
    }); err != nil {
        return err
    }

    return nil
}

func TestSomethingStream(t *testing.T) {
    conn := newServer(t, func(s grpc.ServiceRegistrar) {
        pb.RegisterSomethingStreamServer(s, &somethingStream{})
    })

    client := pb.NewSomethingStreamClient(conn)

    stream, err := client.GetSomething(
        context.Background(),
        &pb.SomethingRequest{
            Name: "test",
        },
    )
    if err != nil {
        t.Fatalf("failed to get something stream: %v", err)
    }

    response, err := stream.Recv()
    if err != nil {
        t.Fatalf("failed to receive response: %v", err)
    }

    if response.Message != "Hello test" {
        t.Errorf("unexpected response: %v", response.Message)
    }
}

As we can see, the code for unary requests is simpler and easier to understand
than the code for streaming requests.

Sending multiple documents at once if we can

Let's compare these two service definitions:

service BookStore {
  rpc ListBooks(ListBooksRequest) returns (stream Book) {}
}

service BookStoreBatch {
  rpc ListBooks(ListBooksRequest) returns (stream ListBooksResponse) {}
}

message ListBooksResponse {
  repeated Book books = 1;
}

BookStore streams one book at a time, while BookStoreBatch streams multiple books simultaneously.

If the client needs to list all books, it is more efficient to use BookStoreBatch
because it reduces the number of round trips between the client and the server.

Let's see the Golang code example for BookStore and BookStoreBatch:

BookStore:

type bookStore struct {
    pb.UnimplementedBookStoreServer
}

func (s *bookStore) ListBooks(req *pb.ListBooksRequest, stream pb.BookStore_ListBooksServer) error {
    for _, b := range bookStoreData {
        if b.Author == req.Author {
            if err := stream.Send(&pb.Book{
                Title:           b.Title,
                Author:          b.Author,
                PublicationYear: int32(b.PublicationYear),
                Genre:           b.Genre,
            }); err != nil {
                return err
            }
        }
    }
    return nil
}

func TestBookStore_ListBooks(t *testing.T) {
    conn := newServer(t, func(s grpc.ServiceRegistrar) {
        pb.RegisterBookStoreServer(s, &bookStore{})
    })

    client := pb.NewBookStoreClient(conn)

    stream, err := client.ListBooks(
        context.Background(),
        &pb.ListBooksRequest{
            Author: charlesDickens,
        },
    )
    if err != nil {
        t.Fatalf("failed to list books: %v", err)
    }

    books := []*pb.Book{}
    for {
        book, err := stream.Recv()
        if err != nil {
            break
        }
        books = append(books, book)
    }

    if len(books) != charlesDickensBooks {
        t.Errorf("unexpected number of books: %d", len(books))
    }
}

BookStoreBatch:

type bookStoreBatch struct {
    pb.UnimplementedBookStoreBatchServer
}

func (s *bookStoreBatch) ListBooks(req *pb.ListBooksRequest, stream pb.BookStoreBatch_ListBooksServer) error {
    const batchSize = 10
    books := make([]*pb.Book, 0, batchSize)
    for _, b := range bookStoreData {
        if b.Author == req.Author {
            books = append(books, &pb.Book{
                Title:           b.Title,
                Author:          b.Author,
                PublicationYear: int32(b.PublicationYear),
                Genre:           b.Genre,
            })

            if len(books) == batchSize {
                if err := stream.Send(&pb.ListBooksResponse{
                    Books: books,
                }); err != nil {
                    return err
                }
                books = books[:0]
            }
        }
    }

    if len(books) > 0 {
        if err := stream.Send(&pb.ListBooksResponse{
            Books: books,
        }); err != nil {
            return nil
        }
    }

    return nil
}

func TestBookStoreBatch_ListBooks(t *testing.T) {
    conn := newServer(t, func(s grpc.ServiceRegistrar) {
        pb.RegisterBookStoreBatchServer(s, &bookStoreBatch{})
    })

    client := pb.NewBookStoreBatchClient(conn)

    stream, err := client.ListBooks(
        context.Background(),
        &pb.ListBooksRequest{
            Author: charlesDickens,
        },
    )
    if err != nil {
        t.Fatalf("failed to list books: %v", err)
    }

    books := []*pb.Book{}
    for {
        response, err := stream.Recv()
        if err != nil {
            break
        }
        books = append(books, response.Books...)
    }

    if len(books) != charlesDickensBooks {
        t.Errorf("unexpected number of books: %d", len(books))
    }
}

From the code above, it needs to be clarified which one is better.
Let's run a benchmark to see the difference:

BookStore benchmark:

func BenchmarkBookStore_ListBooks(b *testing.B) {
    conn := newServer(b, func(s grpc.ServiceRegistrar) {
        pb.RegisterBookStoreServer(s, &bookStore{})
    })

    client := pb.NewBookStoreClient(conn)

    var benchInnerBooks []*pb.Book
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        stream, err := client.ListBooks(
            context.Background(),
            &pb.ListBooksRequest{
                Author: charlesDickens,
            },
        )
        if err != nil {
            b.Fatalf("failed to list books: %v", err)
        }

        books := []*pb.Book{}
        for {
            book, err := stream.Recv()
            if err != nil {
                break
            }
            books = append(books, book)
        }

        benchInnerBooks = books
    }

    benchBooks = benchInnerBooks
}

BookStoreBatch benchmark:

func BenchmarkBookStoreBatch_ListBooks(b *testing.B) {
    conn := newServer(b, func(s grpc.ServiceRegistrar) {
        pb.RegisterBookStoreBatchServer(s, &bookStoreBatch{})
    })

    client := pb.NewBookStoreBatchClient(conn)

    var benchInnerBooks []*pb.Book
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        stream, err := client.ListBooks(
            context.Background(),
            &pb.ListBooksRequest{
                Author: charlesDickens,
            },
        )
        if err != nil {
            b.Fatalf("failed to list books: %v", err)
        }

        books := []*pb.Book{}
        for {
            response, err := stream.Recv()
            if err != nil {
                break
            }
            books = append(books, response.Books...)
        }

        benchInnerBooks = books
    }

    benchBooks = benchInnerBooks
}

Benchmark results:

BenchmarkBookStore_ListBooks
BenchmarkBookStore_ListBooks-12                      732           1647454 ns/op           85974 B/op       1989 allocs/op
BenchmarkBookStoreBatch_ListBooks
BenchmarkBookStoreBatch_ListBooks-12                1202            937491 ns/op           61098 B/op        853 allocs/op

What an improvement! BookStoreBatch is faster than BookStore by a factor of 1.75x.

But why is BookStoreBatch faster than BookStore?

Each time that the server sends a message stream.Send() to the client, needs to
encode the message and send it over the network. By sending multiple documents
at once, we reduce the number of times that the server needs to encode and send
the message, which improves the performance not only for the server but also
for the client that needs to decode the message.

In the above example, the batch size is set to 10, but the client can adjust it based on the network conditions and the size of the documents.

Use bidirectional streaming to control the flow

The bookstore example returns all the books and finishes the stream, but if the client
needs to watch for events in real-time (e.g., sensors), the use of bidirectional
streaming is the right choice.

Bidirectional streams are a bit tricky because both the client and the server
can send and receive messages at the same time. Hopefully, golang will make it easy
to work with concurrency like this.

As mentioned, a sensor can be an excellent example of bidirectional streaming.
The watch function allows the client to decide which sensors to watch and request
the current value if needed.

Let's take a look at the following protobuf definition:

service Sensor {
  rpc Watch(stream WatchRequest) returns (stream WatchResponse) {}
}

message WatchRequest {
  oneof request {
    WatchCreateRequest create_request = 1;
    WatchCancelRequest cancel_request = 2;
    WatchNowRequest now_request = 3;
  }
}

message WatchCreateRequest {
  // sensor_id contains the sensor id to watch.
  string sensor_id = 1;
}

message WatchCancelRequest {
  // sensor_id contains the sensor id to cancel.
  string sensor_id = 1;
}

message WatchNowRequest {
  // sensor_id contains the sensor id to get the current value.
  string sensor_id = 1;
}

message WatchResponse {
  // sensor_id contains the sensor id for the current response.
  string sensor_id = 1;
  // created is true if the watch was created successfully.
  bool created = 2;
  // canceleted is true if the watch was canceled successfully or if the creation failed.
  bool canceleted = 3;
  // error contains the error message if something went wrong.
  string error = 4;
  // timestamp contains the timestamp of the value.
  google.protobuf.Timestamp timestamp = 5;
  // value contains the value of the sensor.
  int32 value = 6;
}

The request message is not only a stream of messages but also a message that can
contain different types of requests. The oneof directive allows us to define a
field that can contain only one of the specified types.

The golang code for the sensor will ignore, but you can found it here

serverStream wraps the stream and the sensor data to make it easier to work with.

type serverStream struct {
    s           *sensorService         // Service
    stream      pb.Sensor_WatchServer  // Stream
    sendCh      chan *pb.WatchResponse // Control channel
    sensorCh    chan sensorData        // Data channel
    sensorWatch map[string]int         // Map of sensor id to watch id
}

As noted before, the server can send and receive messages at the same time, one
function will handle the incoming messages and another function will handle the
outgoing messages.

Receiving messages:

func (ss *serverStream) recvLoop() error {
    defer ss.close()
    for {
        req, err := ss.stream.Recv()
        if errors.Is(err, io.EOF) {
            return nil
        }
        if err != nil {
            return err
        }

        switch req := req.Request.(type) {
        case *pb.WatchRequest_CreateRequest:
            // IGNORE VALIDATION (check the full code)

            // create a channel to send data to the client
            id := sensor.watch(ss.sensorCh)
            ss.sensorWatch[sensorId] = id

            // send created message
            ss.sendCh <- &pb.WatchResponse{
                SensorId: sensorId,
                Created:  true,
            }

        case *pb.WatchRequest_CancelRequest:
            // IGNORE VALIDATION (check the full code)

            // cancel the watch
            ss.s.sensors[sensorId].cancel(id)
            delete(ss.sensorWatch, sensorId)

            ss.sendCh <- &pb.WatchResponse{
                SensorId:   sensorId,
                Canceleted: true,
            }

        case *pb.WatchRequest_NowRequest:
            // IGNORE VALIDATION (check the full code)

            // send current value
            ss.sendCh <- &pb.WatchResponse{
                SensorId:  sensorId,
                Timestamp: timestamppb.Now(),
                Value:     int32(sensor.read()),
            }
        }
    }
}

The switch statement is used to handle the different types of requests and decide
what to do with each request. It's important to leave the recvLoop function only
to read and don't send messages to the client for this reason we have the sendLoop
that will read the messages from the control channel and send it to the client.

Sending messages:

func (ss *serverStream) sendLoop() {
    for {
        select {
        case m, ok := <-ss.sendCh:
            if !ok {
                return
            }

            // send message
            if err := ss.stream.Send(m); err != nil {
                return
            }

        case data, ok := <-ss.sensorCh:
            if !ok {
                return
            }

            // send data
            if err := ss.stream.Send(&pb.WatchResponse{
                SensorId:  data.id,
                Timestamp: timestamppb.New(data.time),
                Value:     int32(data.val),
            }); err != nil {
                return
            }

        case <-ss.stream.Context().Done():
            return
        }
    }
}

The sendLoop function reads both the control channel and the data channel and sends
the messages to the client. If the stream is closed, the function will return.

Finally, a happy path test for the sensor service:

func TestSensor(t *testing.T) {
    conn := newServer(t, func(s grpc.ServiceRegistrar) {
        pb.RegisterSensorServer(s, &sensorService{
            sensors: newSensors(),
        })
    })

    client := pb.NewSensorClient(conn)

    stream, err := client.Watch(context.Background())
    if err != nil {
        t.Fatalf("failed to watch: %v", err)
    }

    response := make(chan *pb.WatchResponse)
    // Go routine to read from the stream
    go func() {
        defer close(response)
        for {
            resp, err := stream.Recv()
            if errors.Is(err, io.EOF) {
                return
            }
            if err != nil {
                return
            }
            response <- resp
        }
    }()

    createRequest(t, stream, "temp")
    waitUntilCreated(t, response, "temp")
    waitForSensorData(t, response, "temp")

    createRequest(t, stream, "pres")
    waitUntilCreated(t, response, "pres")
    waitForSensorData(t, response, "pres")

    waitForSensorData(t, response, "temp")
    waitForSensorData(t, response, "pres")

    // invalid sensor
    createRequest(t, stream, "invalid")
    waitUntilCanceled(t, response, "invalid")

    nowRequest(t, stream, "light")
    waitForSensorData(t, response, "light")
    // Wait for 2 seconds to make sure we don't receive any data for light
    waitForNoSensorData(t, response, "light", 2*time.Second)

    cancelRequest(t, stream, "temp")
    waitUntilCanceled(t, response, "temp")

    waitForSensorData(t, response, "pres")
    // Wait for 2 seconds to make sure we don't receive any data for temp
    waitForNoSensorData(t, response, "temp", 2*time.Second)

    err = stream.CloseSend()
    if err != nil {
        t.Fatalf("failed to close send: %v", err)
    }
}

From the test above, we can see that the client can create, cancel, and get the current
value of a sensor. The client can also watch multiple sensors at the same time.

Challenge Yourself

Implement a chat application using gRPC streaming.
Modify the sensor service to send multiple values at once to save round trips.
Sniff the network traffic to see the difference between unary request and streaming request.

Conclusion

gRPC streaming is a versatile and powerful tool for building real-time applications.
By following best practices like using streaming only when necessary, batching data efficiently, and leveraging bidirectional streaming wisely, developers can maximize performance
and maintain code simplicity.
While gRPC streaming introduces complexity, its benefits far outweigh the challenges
when applied thoughtfully.

Stay in touch

If you have any questions or feedback, feel free to reach out to me on LinkedIn.

Build your Service Mesh: Admission Controller

Ramón Berrutti — Sat, 22 Jun 2024 20:26:49 +0000

Creating the Admission Controller

Before kube-apiserver persists the object and later be scheduled to a node,
the Admission Controller can validate and mutate the object.

The Mutation Admission Controller is going to mutate the pods that have the the
annotation diy-service-mesh: true.

Let's dig into how the Admission Controller works.

Admission Controller Flow

kube-controller or kubectl sends a request to the kube-apiserver to create a pod.
The kube-apiserver sends the request to the Admission Controller. In this case the proxy-injector.
The proxy-injector returns the mutated patch to the kube-apiserver.
Kube-apiserver persists the object in the etcd if the object is valid.
Kube-scheduler will schedule the pod to a node.
Kube-scheduler returns an available node to the kube-apiserver or an error if the pod can't be scheduled.
The kube-apiserver will store the object in the etcd with the selected node.
The kubelet in the selected node will create the pod in the container runtime.

Admission Controller Code

Full code of the Admission Controller: injector

The mutate function processes the AdmissionReview object and returns the AdmissionResponse object with the mutated patch.

func mutate(ar *admissionv1.AdmissionReview) *admissionv1.AdmissionResponse {
    req := ar.Request
    // Ignore all requests other than pod creation.
    if req.Operation != admissionv1.Create || req.Kind.Kind != "Pod" {
        return &admissionv1.AdmissionResponse{
            UID:     req.UID,
            Allowed: true,
        }
    }

    var pod v1.Pod
    // Unmarshal the raw object to the pod.
    if err := json.Unmarshal(req.Object.Raw, &pod); err != nil {
        return &admissionv1.AdmissionResponse{
            UID: req.UID,
            Result: &metav1.Status{
                Message: err.Error(),
            },
        }
    }

    // Check if the pod contains the inject annotation.
    if v, ok := pod.Annotations["diy-service-mesh/inject"]; !ok || strings.ToLower(v) != "true" {
        return &admissionv1.AdmissionResponse{
            UID:     req.UID,
            Allowed: true,
        }
    }

    // Add the initContainer to the pod.
    pod.Spec.InitContainers = append(pod.Spec.InitContainers, v1.Container{
        Name:            "proxy-init",
        Image:           os.Getenv("IMAGE_TO_DEPLOY_PROXY_INIT"),
        ImagePullPolicy: v1.PullAlways,
        SecurityContext: &v1.SecurityContext{
            Capabilities: &v1.Capabilities{
                Add:  []v1.Capability{"NET_ADMIN", "NET_RAW"},
                Drop: []v1.Capability{"ALL"},
            },
        },
    })

    // Add the sidecar container to the pod.
    pod.Spec.Containers = append(pod.Spec.Containers, v1.Container{
        Name:            "proxy",
        Image:           os.Getenv("IMAGE_TO_DEPLOY_PROXY"),
        ImagePullPolicy: v1.PullAlways,
        SecurityContext: &v1.SecurityContext{
            RunAsUser:    func(i int64) *int64 { return &i }(1337),
            RunAsNonRoot: func(b bool) *bool { return &b }(true),
        },
    })

    patch := []map[string]any{
        {
            "op":    "replace",
            "path":  "/spec/initContainers",
            "value": pod.Spec.InitContainers,
        },
        {
            "op":    "replace",
            "path":  "/spec/containers",
            "value": pod.Spec.Containers,
        },
    }

    podBytes, err := json.Marshal(patch)
    if err != nil {
        return &admissionv1.AdmissionResponse{
            UID: req.UID,
            Result: &metav1.Status{
                Message: err.Error(),
            },
        }
    }

    patchType := admissionv1.PatchTypeJSONPatch
    return &admissionv1.AdmissionResponse{
        UID:     req.UID,
        Allowed: true,
        AuditAnnotations: map[string]string{
            "proxy-injected": "true",
        },
        Patch:     podBytes,
        PatchType: &patchType,
    }
}

IMAGE_TO_DEPLOY_PROXY_INIT and IMAGE_TO_DEPLOY_PROXY are environment variables that tilt will update with the last proxy-init and proxy image respectively.

For complex patch use thi library: https://github.com/evanphx/json-patch

Deploying the Admission Controller

MutatingWebhookConfiguration tells the kube-apiserver to send the pod creation requests to the injector.

apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
  name: service-mesh-injector-webhook
webhooks:
- name: service-mesh-injector.service-mesh.svc
  clientConfig:
    service:
      name: service-mesh-injector
      namespace: service-mesh
      path: "/inject"
  rules:
  - operations: ["CREATE"]
    apiGroups: [""]
    apiVersions: ["v1"]
    resources: ["pods"]
  admissionReviewVersions: ["v1"]
  sideEffects: None
  timeoutSeconds: 5

Some important points:

caBundle in the clientConfig is missing. This is a necessary field because the kube-apiserver only calls the webhook if the certificate is valid.
Two jobs, injector-admission-create injector-admission-patch are going generate the certificates and patch the MutatingWebhookConfiguration with the caBundle.
The rules options allow to filter the objects that are going to be sent to the injector.

The file injector.yaml contains the nesesary resources
including the Service Account, Role, RoleBinding, ClusterRole, ClusterRoleBinding,
Service, Deployment and the Job to generate the certificates.

Testing the Admission Controller

Let's modify the http-client and http-server deployments to add the annotation diy-service-mesh/inject: "true".

spec:
  replicas: 1
  selector:
    matchLabels:
      app: http-client
  template:
    metadata:
      labels:
        app: http-client
      annotations:
        diy-service-mesh/inject: "true"
    spec:

Important: the annotation needs to be added to the pod template and not to the deployment.

Build your Service Mesh: The Proxy

Ramón Berrutti — Tue, 18 Jun 2024 22:56:15 +0000

DIY Service Mesh

This is a Do-It-Yourself Service Mesh, a simple tutorial for understanding
the internals of a service mesh. This project aims to provide a simple,
easy-to-understand reference implementation of a service mesh, which can be used
to learn about the various concepts and technologies used by a service mesh like Linkerd.

What are you going to learn?

Build a simple proxy and add service mesh features to it.
Use Netfilter to intercept and modify network packets.
Create a simple control plane to manage the service mesh.
Use gRPC to communicate between the proxy and the control plane.
Create an Admission Controller to validate and mutate Kubernetes resources.
Certificate generation flow and mTLS between the services.
How HTTP/2 works and how to use it with gRPC to balance the traffic between the services.
Add useful features like circuit breaking, retries, timeouts, and load balancing.
Add metrics and tracing to the service mesh with OpenTelemetry.
Canary deployments.

Some considerations

Only for learning propose, not a production-ready service mesh.
The proxy will print many logs to understand what is happening.
Use IPTables instead of Nftables for simplicity.
Keep the code as simple as possible to make it easy to understand.
Some Golang errors are ignored for simplicity.
Everything will be in the same repository to make it easier to understand the project.

What is going to be built?

The following components are going to be built:

proxy-init: Configure the network namespace of the pod.
proxy: This is the data plane of the service mesh, which is responsible for intercepting and modifying network packets.
controller: This is the control plane of the service mesh, which is responsible for configuring the data plane.
injector: This is an Admission Controller for Kubernetes, which mutates each pod that needs to be meshed.
samples apps: Four simple applications will communicate with each other. (http-client, http-server, grpc-client, grpc-server)

Tools and Running the project

kind to create a Kubernetes cluster locally.
Tilt to run the project and watch for changes.
Buf to lint and generate the Protobuf/gRPC code.
Docker to build the Docker images.
k9s to interact with the Kubernetes cluster. (Optional)

To start all the components, run the following command:

kind create cluster
tilt up

Tilt will build all the images and deploy all the components to the Kubernetes cluster.
All the images are created by the Dockerfile in the root directory.

The main branch contains the WIP version of the project.

Architecture

The architecture of the service mesh is composed of the following components:

Creating the HTTP applications

http-client: This application is going to be called the http-server service.
http-server: This application will be called by the http-client service.

In the next steps, grpc-client and grpc-server will be created.

http-client:

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer cancel()
    n := 0

    endpoint := os.Getenv("ENDPOINT")
    if endpoint == "" {
        endpoint = "http://http-server.http-server.svc.cluster.local./hello"
    }

    httpClient := &http.Client{
        Timeout: 5 * time.Second,
    }

    // This application will call the endpoint every second
    ticker := time.NewTicker(time.Second)
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
            if err != nil {
                panic(err)
            }

            resp, err := httpClient.Do(req)
            if err != nil {
                panic(err)
            }

            dump, err := httputil.DumpResponse(resp, true)
            if err != nil {
                panic(err)
            }
            resp.Body.Close()

            n++
            fmt.Printf("Response #%d\n", n)
            fmt.Println(string(dump))
        }
    }
}

http-server:

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer cancel()

    failRate, _ := strconv.Atoi(os.Getenv("FAIL_RATE"))

    n := uint64(0)
    hostname := os.Getenv("HOSTNAME")
    version := os.Getenv("VERSION")

    var b bytes.Buffer
    b.WriteString("Hello from the http-server service! Hostname: ")
    b.WriteString(hostname)
    b.WriteString(" Version: ")
    b.WriteString(version)

    mux := http.NewServeMux()
    mux.HandleFunc("/hello", func(w http.ResponseWriter, r *http.Request) {
        // Dump the request
        dump, _ := httputil.DumpRequest(r, true)
        fmt.Printf("Request #%d\n", atomic.AddUint64(&n, 1))
        fmt.Println(string(dump))
        fmt.Println("---")

        // Simulate failure
        if failRate > 0 {
            // Get a random number between 0 and 100
            n := rand.Intn(100)
            if n < failRate {
                http.Error(w, "Internal server error", http.StatusInternalServerError)
                fmt.Println("Failed to process request")
                return
            }
        }

        w.Header().Set("Content-Type", "text/plain")
        w.WriteHeader(http.StatusOK)
        w.Write(b.Bytes())
    })

    server := &http.Server{
        Addr:    ":8080",
        Handler: mux,
    }

    g, ctx := errgroup.WithContext(ctx)
    g.Go(func() error {
        <-ctx.Done()
        return server.Shutdown(context.Background())
    })

    g.Go(func() error {
        return server.ListenAndServe()
    })

    if err := g.Wait(); err != nil {
        if err != http.ErrServerClosed {
            panic(err)
        }
    }
}

The http-server service is going to respond with a message that contains the
hostname and the version of the service.
Failures will be simulated by setting the FAIL_RATE environment variable.

Each service is going to be deployed in a different namespace:

http-client: http-client Deployment in the http-client namespace: http-client.yaml
http-server: http-server Deployment in the http-server namespace: http-server.yaml

Testing the service mesh

Logs for http-client and http-server pods to see the communication between the services.

http-client logs:

Response #311
HTTP/1.1 200 OK
Content-Length: 71
Content-Type: text/plain
Date: Sat, 08 Jun 2024 19:38:27 GMT

Hello from http-server service! Hostname: http-server-799c77dc9b-56lmd Version: 1.0

http-server logs:

Request #171
GET /hello HTTP/1.1
Host: http-server.http-server.svc.cluster.local.
Accept-Encoding: gzip
User-Agent: Go-http-client/1.1

Implementing the proxy to intercept the HTTP/1.1 requests and responses.

Why need a proxy?

The proxy will intercept all of the inbound and outbound traffic of the services. (except explicitly ignored)
Linkerd has a Rust based proxy called linkerd2-proxy, and Istio has a C++ based proxy called Envoy,
which is a very powerful proxy with a lot of features.

The proxy code is going to be very simple and will be similar to linkerd2-proxy functionalities.

For now, the proxy will listen on two ports: one for inbound traffic and another for outbound traffic.

4000 for the inbound traffic.
5000 for the outbound traffic.

This is a basic proxy implementation that intercepts HTTP requests and HTTP responses.

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
    defer cancel()

    g, ctx := errgroup.WithContext(ctx)
    // Inbound connection
    g.Go(func() error {
        return listen(ctx, ":4000", handleInboundConnection)
    })

    // Outbound connection
    g.Go(func() error {
        return listen(ctx, ":5000", handleOutboundConnection)
    })

    if err := g.Wait(); err != nil {
        panic(err)
    }
}

func listen(ctx context.Context, addr string, accept func(net.Conn)) error {
    l, err := net.Listen("tcp", addr)
    if err != nil {
        return fmt.Errorf("failed to listen: %w", err)
    }
    defer l.Close()
    go func() {
        <-ctx.Done()
        l.Close()
    }()

    for {
        conn, err := l.Accept()
        if err != nil {
            return fmt.Errorf("failed to accept: %w", err)
        }

        go accept(conn)
    }
}

The listen function listens on the port and calls the accept function when a connection is established.

handleInboundConnection

All the inbound traffic is going to be intercepted by the proxy, print the request, forward the request
to the local destination port and print the response.

func handleInboundConnection(c net.Conn) {
    defer c.Close()

    // Get the original destination
    _, port, err := getOriginalDestination(c)
    if err != nil {
        fmt.Printf("Failed to get original destination: %v\n", err)
        return
    }

    fmt.Printf("Inbound connection from %s to port: %d\n", c.RemoteAddr(), port)

    // Read the request
    req, err := http.ReadRequest(bufio.NewReader(c))
    if err != nil {
        fmt.Printf("Failed to read request: %v\n", err)
        return
    }

    // Call the local service port
    upstream, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", port))
    if err != nil {
        fmt.Printf("Failed to dial: %v\n", err)
        return
    }
    defer upstream.Close()

    // Write the request
    if err := req.Write(io.MultiWriter(upstream, os.Stdout)); err != nil {
        fmt.Printf("Failed to write request: %v\n", err)
        return
    }

    // Read the response
    resp, err := http.ReadResponse(bufio.NewReader(upstream), req)
    if err != nil {
        fmt.Printf("Failed to read response: %v\n", err)
        return
    }
    defer resp.Body.Close()

    // Write the response
    if err := resp.Write(io.MultiWriter(c, os.Stdout)); err != nil {
        fmt.Printf("Failed to write response: %v\n", err)
        return
    }

    // Add a newline for better readability
    fmt.Println()
}

The handleInboundConnection function first reads the destination port to our service,
iptables will set the destination port using the SO_ORIGINAL_DST socket option.
The function getOriginalDestination returns the original destination of the TCP connection,
check the code to see how it works. (This is a Linux specific feature)

After that, read the request, forward the request to the local service port, read
the response and send it back to the client.

For visibility, print the request and response using io.MultiWriter to write to the connection and stdout.

handleOutboundConnection

The outbound look very similar to the inbound, but forward the request to the target service.

func handleOutboundConnection(c net.Conn) {
    defer c.Close()

    // Get the original destination
    ip, port, err := getOriginalDestination(c)
    if err != nil {
        fmt.Printf("Failed to get original destination: %v\n", err)
        return
    }

    fmt.Printf("Outbound connection to %s:%d\n", ip, port)

    // Read the request
    req, err := http.ReadRequest(bufio.NewReader(c))
    if err != nil {
        fmt.Printf("Failed to read request: %v\n", err)
        return
    }

    // Call the external service ip:port
    upstream, err := net.Dial("tcp", fmt.Sprintf("%s:%d", ip, port))
    if err != nil {
        fmt.Printf("Failed to dial: %v\n", err)
        return
    }
    defer upstream.Close()

    // Write the request
    if err := req.Write(io.MultiWriter(upstream, os.Stdout)); err != nil {
        fmt.Printf("Failed to write request: %v\n", err)
        return
    }

    // Read the response
    resp, err := http.ReadResponse(bufio.NewReader(upstream), req)
    if err != nil {
        fmt.Printf("Failed to read response: %v\n", err)
        return
    }
    defer resp.Body.Close()

    // Write the response
    if err := resp.Write(io.MultiWriter(c, os.Stdout)); err != nil {
        fmt.Printf("Failed to write response: %v\n", err)
        return
    }

    // Add a newline for better readability
    fmt.Println()
}

As can be seen, the only difference is in how the external service is called.

    // Call the external service ip:port
    upstream, err := net.Dial("tcp", fmt.Sprintf("%s:%d", ip, port))
    if err != nil {
        fmt.Printf("Failed to dial: %v\n", err)
        return
    }
    defer upstream.Close()

It is important to note that the service resolves the DNS, so only the IP and the port need to be provided.

How are the connections intercepted?

Kubernetes Pod Networking Understanding

Each kubernetes pod shares the same network between the containers, so the localhost is the same for all the containers in the pod.

initContainers

Kubernetes has a feature called initContainers, which is a container that runs before the main containers starts. These containers need to finish before the main containers starts.

iptables

The iptables is a powerful tool to manage Netfilter in Linux, it can be used to intercept and modify the network packets.

Before our http-client and http-server containers starts, the proxy-init is going to configure the Netfilter to redirect
all the traffic to the proxy inbounds and outbounds ports.

func main() {
    // Configure the proxy
    commands := []*exec.Cmd{
        // Default accept for all nat chains
        exec.Command("iptables", "-t", "nat", "-P", "PREROUTING", "ACCEPT"),
        exec.Command("iptables", "-t", "nat", "-P", "INPUT", "ACCEPT"),
        exec.Command("iptables", "-t", "nat", "-P", "OUTPUT", "ACCEPT"),
        exec.Command("iptables", "-t", "nat", "-P", "POSTROUTING", "ACCEPT"),

        // Create custom chains so is possible jump to them
        exec.Command("iptables", "-t", "nat", "-N", "PROXY_INBOUND"),
        exec.Command("iptables", "-t", "nat", "-N", "PROXY_OUTBOUND"),

        // Jump to custom chains, if something is not matched, will return to the default chains.
        exec.Command("iptables", "-t", "nat", "-A", "PREROUTING", "-p", "tcp", "-j", "PROXY_INBOUND"),
        exec.Command("iptables", "-t", "nat", "-A", "OUTPUT", "-p", "tcp", "-j", "PROXY_OUTBOUND"),

        // Set rules for custom chains: PROXY_INBOUND, redirect all inbound traffic to port 4000
        exec.Command("iptables", "-t", "nat", "-A", "PROXY_INBOUND", "-p", "tcp", "-j", "REDIRECT", "--to-port", "4000"),

        // Set rules for custom chains: PROXY_OUTBOUND
        // Ignore traffic between the containers.
        exec.Command("iptables", "-t", "nat", "-A", "PROXY_OUTBOUND", "-o", "lo", "-j", "RETURN"),
        exec.Command("iptables", "-t", "nat", "-A", "PROXY_OUTBOUND", "-d", "127.0.0.1/32", "-j", "RETURN"),

        // Ignore outbound traffic from the proxy container.
        exec.Command("iptables", "-t", "nat", "-A", "PROXY_OUTBOUND", "-m", "owner", "--uid-owner", "1337", "-j", "RETURN"),

        // Redirect all the outbound traffic to port 5000
        exec.Command("iptables", "-t", "nat", "-A", "PROXY_OUTBOUND", "-p", "tcp", "-j", "REDIRECT", "--to-port", "5000"),
    }

    for _, cmd := range commands {
        if err := cmd.Run(); err != nil {
            panic(fmt.Sprintf("failed to run command %s: %v\n", cmd.String(), err))
        }
    }

    fmt.Println("Proxy initialized successfully!")
}

Some important points:

To allow outbound traffic from the proxy container, the iptables option --uid-owner is used.
The use of custom chains is to make it easier to understand the rules and allow to return to the default chains if the rule is not matched.
The REDIRECT option is used to redirect the traffic to the proxy and is the responsable to add the SO_ORIGINAL_DST information to the socket.

Adding the proxy and proxy-init containers to the deployments:

    spec:
      initContainers:
      - name: proxy-init
        image: diy-sm-proxy-init
        imagePullPolicy: IfNotPresent
        securityContext:
          capabilities:
            add:
              - NET_ADMIN
              - NET_RAW
            drop:
              - ALL
      containers:
      - name: proxy
        image: diy-sm-proxy
        imagePullPolicy: IfNotPresent
        securityContext:
          runAsUser: 1337
      - name: http-client
        image: diy-sm-http-client
        imagePullPolicy: IfNotPresent

The same configuration is going to be applied to the http-server deployment.

Some important points:

proxy-init run as init container that is going to configure the network namespace of the pod and exit.
NET_ADMIN and NET_RAW are linux capabilities that are necessary to configure the Netfilter, without these capabilities iptables can't call the system calls to configure the Netfilter.
Using the runAsUser: 1337 in the proxy container is very important so the proxy traffic to outside of the pod is allowed.

Logs output for the proxy and the applications

http-server logs:

proxy Inbound connection from 10.244.0.115:60296 to port: 8080
http-server Request #13
http-server GET /hello HTTP/1.1
http-server Host: http-server.http-server.svc.cluster.local.
http-server Accept-Encoding: gzip
http-server User-Agent: Go-http-client/1.1
proxy GET /hello HTTP/1.1
proxy Host: http-server.http-server.svc.cluster.local.
proxy User-Agent: Go-http-client/1.1
proxy Accept-Encoding: gzip
proxy
proxy HTTP/1.1 200 OK
proxy Content-Length: 86
proxy Content-Type: text/plain
proxy Date: Mon, 17 Jun 2024 20:58:46 GMT
proxy
proxy Hello from the http-server service! Hostname: http-server-c6f4776bb-mmw2d Version: 1.0

http-client logs:

proxy Outbound connection to 10.96.182.169:80
proxy GET /hello HTTP/1.1
proxy Host: http-server.http-server.svc.cluster.local.
proxy User-Agent: Go-http-client/1.1
proxy Accept-Encoding: gzip
proxy
proxy HTTP/1.1 200 OK
proxy Content-Length: 86
proxy Content-Type: text/plain
proxy Date: Mon, 17 Jun 2024 21:04:53 GMT
proxy
proxy Hello from the http-server service! Hostname: http-server-c6f4776bb-slpdf Version: 1.0
http-client Response #16
http-client HTTP/1.1 200 OK
http-client Content-Length: 86
http-client Content-Type: text/plain
http-client Date: Mon, 17 Jun 2024 21:04:53 GMT
http-client
http-client Hello from the http-server service! Hostname: http-server-c6f4776bb-slpdf Version: 1.0

Adding manually the proxy-init and proxy containers?

Doing this is not so practical, right?

Kubernetes has a feature called Admission Controller, which is a webhook that can validate and mutate the objects before they are persisted in the etcd.
Learn more about it here.

But in the next steps, a Mutation Admission Controller will be created to inject the proxy-init and proxy containers into the pods that need to be meshed.

For the next part, the focus will be to create the Admission Controller to inject the proxy-init and the proxy.

Contact and Github Project

The github project contains the WIP of the project.

Feel free to contact me: https://www.linkedin.com/in/ramonberrutti/

MultiPlayer Game with Nats KV Store And WebSocket [Part 1]

Ramón Berrutti — Wed, 13 Mar 2024 22:41:11 +0000

What are you going to learn?

The power of Nats Key-Value Store.
Create a simple game using Nats. In this case, the Jokenpo game.
Nats Authorizations and Permissions.
Play the game through the browser connecting to Nats server using Websocket.

Jokenpo Game

Jokenpo, also known as Rock-Paper-Scissors, is a simple hand game between two players. The game has three possible moves: rock, paper, and scissors.

The rules are the follows:

Rock beats scissors: If one player chooses rock and the other chooses scissors, the player who chose rock wins.
Scissor beats paper: If one player chooses scissors and the other chooses paper, the player who chose scissor wins.
Paper beats rock: If one player chooses paper and the other chooses rock, the player who chose paper wins.

The game is typically played by counting to three and simultaneously revealing the chosen move.

The winner of each round is determined based on the rules mentioned above.

Game Architecture

A single Nats server with a small Golang application is used to handle the game logic.

In the future, Nats may allow us to run some Lua scripts inside the server.

There are two approaches to storing the game state:

Single KV store for all the games.
Single KV store for each game.

For this tutorial, a single KV store for all the games, so the game is simple enough.

Our model for the KV will consist of the following keys:

$KV.jokenpo.{game_id}: Store all the information about the game in JSON format.
$KV.jokenpo.{game_id}.choice1: Store the choice of player 1.
$KV.jokenpo.{game_id}.choice2: Store the choice of player 2.

In future versions of Nats, hashmaps will be allowed, and the game state could be stored in a single key without serializing the data as JSON.

Each KV store is a stream, and a consumer can be created to listen to the changes in the game state.

This is similar to the KV Watch option, but multiple applications can pull messages and acknowledge the message explicitly.

The choice1 and choice2 suffix keys store the players' choices, which are used to determine the game's winner.

Let's do it.

Initial structure for our game state.

// JokenpoGameState represents the game state.
type JokenpoGameState struct {
 Chose1 bool `json:"chose1"`
 Chose2 bool `json:"chose2"`

 Win1 int `json:"wins1"`
 Win2 int `json:"wins2"`
}

Creating the KV Store

nats kv add jokenpo

 kv, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
  Bucket:  "jokenpo",
 })

For now, create the game using the Nats CLI

nats kv create jokenpo game-1 '{}'
echo -n "" | nats kv create jokenpo game-1.choice1
echo -n "" | nats kv create jokenpo game-1.choice2

The pipe is needed here to create an empty value for the choice keys because the Nats CLI does not offer this option.

The durable consumer processes each player's choices:

  nats consumer add --filter "\$KV.jokenpo.*.*" --ack explicit --deliver all --pull --defaults  KV_jokenpo choices-consumer

 con, err := js.CreateOrUpdateConsumer(ctx, fmt.Sprintf("KV_%s", kv.Bucket()), jetstream.ConsumerConfig{
  Durable:       "choices-consumer",
  DeliverPolicy: jetstream.DeliverAllPolicy,
  AckPolicy:     jetstream.AckExplicitPolicy,
  FilterSubject: fmt.Sprintf("$KV.%s.*.*", kv.Bucket()),
 })

Finally and not less importantly, our consumer handler processes the player's choices:

Use the subject to get the game id and the player.
When a player makes a choice, update the game state and check if both players have chosen an option.
If both players choose an option, calculate the winner, update the game state by incrementing the winner's wins, and reset the choices.

sub, err := con.Consume(func(msg jetstream.Msg) {
  ack := true
  defer func() {
   if !ack {
    msg.Nak()
   } else {
    msg.Ack()
   }
  }()
  choice := string(msg.Data())

  // Ignore empty choices and invalid options.
  if choice == "" || (choice != "rock" && choice != "paper" && choice != "scissors") {
   return
  }

  // Get the original key:
  key := strings.TrimPrefix(msg.Subject(), subPrefix)

  // Grab the gameId and the player from the key.
  gameId, slotId := getGameIdAndPlayerSlot(key)
  if gameId == "" || slotId == "" {
   return
  }

  // Get the game state.
  entry, err := kv.Get(ctx, gameId)
  if err != nil {
   return
  }

  // Unmarshal the game state.
  var state JokenpoGameState
  if err := json.Unmarshal(entry.Value(), &state); err != nil {
   return
  }

  // Check if both players have chosen.
  if slotId == "choice1" {
   state.Chose1 = true
  } else {
   state.Chose2 = true
  }

  // Update the game state when a player has chosen.
  encodeState, err := json.Marshal(state)
  if err != nil {
   ack = false
   return
  }
  rev, err := kv.Update(ctx, gameId, encodeState, entry.Revision())
  if err != nil {
   ack = false
   return
  }

  // If both players have chosen, calculate the winner.
  if state.Chose1 && state.Chose2 {
   // Get the other player's choice.
   otherSlot := "choice1"
   if slotId == "choice1" {
    otherSlot = "choice2"
   }

   otherChoiceEntry, err := kv.Get(ctx, fmt.Sprintf("%s.%s", gameId, otherSlot))
   if err != nil {
    ack = false
    return
   }
   otherChoice := string(otherChoiceEntry.Value())
   if otherChoice == "" {
    ack = false
    return
   }

   // Calculate the winner.
   winner := ""
   switch {
   case choice == otherChoice:
    winner = "draw"
   case choice == "rock" && otherChoice == "scissors",
    choice == "scissors" && otherChoice == "paper",
    choice == "paper" && otherChoice == "rock":
    winner = slotId
   default:
    winner = otherSlot
   }

   // Update the game state with the winner.
   if winner == "choice1" {
    state.Win1++
   } else if winner == "choice2" {
    state.Win2++
   }

   state.Chose1 = false
   state.Chose2 = false

   encodeState, err := json.Marshal(state)
   if err != nil {
    ack = false
    return
   }

   if _, err := kv.Update(ctx, gameId, encodeState, rev); err != nil {
    ack = false
    return
   }

   // Reset the choices.
   _, _ = kv.Put(ctx, fmt.Sprintf("%s.choice1", gameId), []byte(""))
   _, _ = kv.Put(ctx, fmt.Sprintf("%s.choice2", gameId), []byte(""))
  }
 })

Testing the Game

Watch for changes in the game state in a terminal with Nats CLI.

nats kv watch jokenpo game-1

Choose an option for each player in another terminal:

nats kv put jokenpo game-1.choice1 rock
nats kv put jokenpo game-1.choice2 scissors

The game state updates, and the winner of the game is player 1.

PUT jokenpo > game-1: {}
PUT jokenpo > game-1: {"chose1":true,"chose2":false,"wins1":0,"wins2":0}
PUT jokenpo > game-1: {"chose1":true,"chose2":true,"wins1":0,"wins2":0}
PUT jokenpo > game-1: {"chose1":false,"chose2":false,"wins1":1,"wins2":0}

Choose a new option for each player:

nats kv put jokenpo game-1.choice1 rock
nats kv put jokenpo game-1.choice2 paper

There are new updates for the game state, and the winner of the game is player 2.

PUT jokenpo > game-1: {"chose1":true,"chose2":false,"wins1":1,"wins2":0}
PUT jokenpo > game-1: {"chose1":true,"chose2":true,"wins1":1,"wins2":0}
PUT jokenpo > game-1: {"chose1":false,"chose2":false,"wins1":1,"wins2":1}

In the following parts:

Create a simple web application to play the game using Websockets to connect to the Nats server because using the Nats CLI is not fun.
Prevent race conditions with the game state and the player's choices. HINT: Revisions.
Add some features such as Rounds, Chat Rooms, and more.

Contact Me

Any suggestions feel free to reach me on LinkedIn: https://www.linkedin.com/in/ramonberrutti/

DEV Community: Ramón Berrutti

gRPC Streaming: Best Practices and Performance Insights

Introduction

Prerequisites

Good Practices

Use unary request for unary request

Sending multiple documents at once if we can

Use bidirectional streaming to control the flow

Challenge Yourself

Conclusion

Stay in touch

Build your Service Mesh: Admission Controller

Creating the Admission Controller

Admission Controller Flow

Admission Controller Code

Deploying the Admission Controller

Testing the Admission Controller

Build your Service Mesh: The Proxy

DIY Service Mesh

What are you going to learn?

Some considerations

What is going to be built?

Tools and Running the project

Architecture

Creating the HTTP applications

Testing the service mesh

Implementing the proxy to intercept the HTTP/1.1 requests and responses.

Why need a proxy?

handleInboundConnection

handleOutboundConnection

How are the connections intercepted?

Kubernetes Pod Networking Understanding

initContainers

iptables

Adding the proxy and proxy-init containers to the deployments:

Logs output for the proxy and the applications

Adding manually the proxy-init and proxy containers?

Next

Contact and Github Project

MultiPlayer Game with Nats KV Store And WebSocket [Part 1]

What are you going to learn?

Jokenpo Game

The rules are the follows:

Game Architecture

Let's do it.

Testing the Game

In the following parts:

Contact Me