Forem: maxwellhertz

A Go Microservices Demo using Go kit And API Gateway Pattern

maxwellhertz — Fri, 31 Jan 2020 10:24:51 +0000

Go kit is a popular Go microservices framework. I've found it quite interesting but lacking clear and detailed examples that a learner can follow. The official hello-world tutorial stringsvc really confused me when I tried to implement stringsvc3. After some struggles, I realized that all this example wanted to do was to "simulate" an API gateway, which was kinda impractical in the real world. Therefore, I combined stringsvc3 with apigateway to create a more practical microservices application. I hope this article may help you if you want to write a useful demo with Go kit.

In this article, I will focus on the implementation of API gateway rather than basic concepts in Go kit such as endpoint, transport or service.

API Gateway Based On Service Discovery

Service discovery is an essential component of a microservices architecture. In a nutshell, we don't need to bother choosing which instance of a service to use because the service discovery system will do this under the hood.

In this example, I will use Consul to implement a very simple client-side service discovery system.

Service Registration

Firstly, download the source code of stringsvc3, remove the file proxying.go and delete the relevant code in main.go.

// We don't need this anymore.
// svc = proxyingMiddleware(context.Background(), *proxy, logger)(svc)

Then, we can register the stringsvc using Go kit's sd package.

package main

import (
    ....

    "github.com/go-kit/kit/sd/consul"
    "github.com/hashicorp/consul/api"
)

func main() {
    ...

    // Build consul client and register services.
    // Specify the information of an instance.
    asr := api.AgentServiceRegistration{
        // Every service instance must have an unique ID.
        ID:      fmt.Sprintf("%v%v/%v", host, listen, prefix),
        Name:    serviceName,
        // These two values are the location of an instance.
        Address: host,
        Port:    port,
    }
    consulConfig := api.DefaultConfig()
    // We can get the address of consul server from environment variale or a config file.
    if len(consulServer) > 0 {
        consulConfig.Address = consulServer
    }
    consulClient, err := api.NewClient(consulConfig)
    if err != nil {
        logger.Log("err", err)
        os.Exit(1)
    }
    sdClient := consul.NewClient(consulClient)
    registar := consul.NewRegistrar(sdClient, &asr, logger)
    registar.Register()
    // According to the official doc of Go kit, 
    // it's important to call registar.Deregister() before the program exits.
    defer registar.Deregister()

    ...
}

It's pretty simple to register services in Consul, the service registry. You can do some additional configurations like some tags for the service. Further reading: Go consul API's Godoc.

I am gonna deploy this demo using Docker Compose later so right now I want to build a Docker image of stringsvc (I have converted this app into a Go modules project in order to manage dependencies more conveniently).

FROM golang:latest as builder

ENV GO111MODULE=on

ENV GOPROXY=https://goproxy.cn,direct

RUN mkdir /app

WORKDIR /app

COPY . .

RUN CGO_ENABLED=0 GOOS=linux go build -o out

FROM alpine:latest

RUN mkdir /app

WORKDIR /app

COPY --from=builder /app/out .

CMD ["./out"]

Service Discovery Client - API Gateway

Create a new Go modules project called stringclient that will act as the service discovery client i.e. the API gateway. In main.go file, we first need to build a consul instancer that yields instances for a service, which has been registered in Consul:

    // Build instancer.
    consulConfig := api.DefaultConfig()
    if len(consulServer) > 0 {
        consulConfig.Address = consulServer
    }
    consulClient, err := api.NewClient(consulConfig)
    if err != nil {
        logger.Log("err", err)
        os.Exit(1)
    }
    client := consul.NewClient(consulClient)
    instancer := consul.NewInstancer(client, logger, serviceName, []string{}, true)

I think the last two parameters of consul.NewInstancer might be confusing (at least I was confused in the very first place). The 4th parameter, tags, indicates only those services with these tags will be returned. This parameter is useful only when you tag your services when you register them. Otherwise, just pass an empty slice of string. As for the last parameter passingOnly, only instances where both the service and any proxy are healthy will be returned if it is true.

At last, we just need to create some endpoints for stringsvc and run the client.

    // uppercase endpoint
    // Create an endpointer that subscribes to the instancer.
    uppercaseEndpointer := sd.NewEndpointer(instancer, serviceFactoryBuilder(uppercasePath, "POST", encodeRequest, decodeResponseFuncBuilder(uppercaseResponse{})), logger)
    // Use round-robin load balancing.
    // Set retry policy.
    uppercaseEndpoint := lb.Retry(3, 3*time.Second, lb.NewRoundRobin(uppercaseEndpointer))
    http.Handle(uppercasePath, httptransport.NewServer(
        uppercaseEndpoint,
        decodeRequestFuncBuilder(uppercaseRequest{}),
        encodeResponse,
    ))
    // count endpoint
    countEndPointer := sd.NewEndpointer(instancer, serviceFactoryBuilder(countPath, "POST", encodeRequest, decodeResponseFuncBuilder(countResponse{})), logger)
    countEndPoint := lb.Retry(3, 3*time.Second, lb.NewRoundRobin(countEndPointer))
    http.Handle(countPath, httptransport.NewServer(
        countEndPoint,
        decodeRequestFuncBuilder(countRequest{}),
        encodeResponse,
    ))

    logger.Log("err", http.ListenAndServe(":8080", nil))

In the above code, sd.NewEndpointer creates an endpointer that subscribes to the instancer. The instancer can retrieve healthy instances from Consul. What the endpointer does is to fetching an instance from the "instance pool" created by the instancer and use a factory function to convert it to a client endpoint. In this case, I define a factory builder function to construct factory function based on some parameters like relative path, HTTP method, etc.

func serviceFactoryBuilder(path string, method string, enc httptransport.EncodeRequestFunc, dec httptransport.DecodeResponseFunc) sd.Factory {
    // instance (host:port) is the location of an instance.
    return func(instance string) (e endpoint.Endpoint, closer io.Closer, err error) {
        httpPrefix := "http://"
        if !strings.HasPrefix(instance, httpPrefix) {
            instance = httpPrefix + instance
        }

        tgt, err := url.Parse(instance)
        if err != nil {
            return nil, nil, err
        }
        tgt.Path = path

        return httptransport.NewClient(method, tgt, enc, dec).Endpoint(), nil, nil
    }
}

Of course, build the client program into a Docker image using the same Dockerfile in the last section.

Deployment And Test

Like I said before, I would use Docker Compose to deploy this demo.

version: "3.7"

services:
  consul:
    image: consul
    command: agent -server -bootstrap -ui -client=0.0.0.0
    ports:
      - 8500:8500
      - 8600:8600/udp
    networks: 
      - gokit
  stringsvc1:
    image: stringsvc
    depends_on: 
      - consul
    ports:
      - 8001
    networks:
      - gokit
  stringsvc2:
    image: stringsvc
    depends_on: 
      - consul
    ports:
      - 8002
    networks:
      - gokit
  stringsvc3:
    image: stringsvc
    depends_on: 
      - consul
    ports:
      - 8003
    networks:
      - gokit
  stringclient:
    image: stringclient
    depends_on: 
      - consul
    ports:
      - 8080:8080
    networks:
      - gokit

networks:
  gokit:

I created 3 instances of stringsvc called stringsvc1, stringsvc2 and stringsvc3 respectively. I also created a client that will be the API gateway. After everything is ready, we can test the client:

$ curl -d '{"s": "foo"}' http://localhost:8080/stringsvc/uppercase
{"v":"FOO"}

$ curl -d '{"s": "foo"}' http://localhost:8080/stringsvc/count
{"v":"3"}

Great! It works! There is still one more thing I would like to show you before ending this article:

$ for s in foo bar baz ; do curl -d"{\"s\":\"$s\"}" localhost:8080/stringsvc/uppercase ; done
{"v":"FOO"}
{"v":"BAR"}
{"v":"BAZ"}

If we have a look at the logs, we will find something like this:

stringsvc2_1    | listen=:8002 caller=logging.go:22 method=uppercase input=foo output=FOO err=null took=629ns
stringsvc3_1    | listen=:8003 caller=logging.go:22 method=uppercase input=bar output=BAR err=null took=967ns
stringsvc1_1    | listen=:8001 caller=logging.go:22 method=uppercase input=baz output=BAZ err=null took=646ns

3 instances of the same services were called one by one and this was because we used the round-robin load balancing policy when we created the endpoint.

uppercaseEndpoint := lb.Retry(3, 3*time.Second, lb.NewRoundRobin(uppercaseEndpointer))

Read the complete source code here.

Something fun about `defer`

maxwellhertz — Wed, 02 Oct 2019 07:56:28 +0000

I read this in Effective Go:

The arguments to the deferred function (which include the receiver if the function is a method) are evaluated when the defer executes, not when the call executes. Besides avoiding worries about variables changing values as the function executes, this means that a single deferred call site can defer multiple function executions.

At first I got confused: what does that mean by saying "The arguments to the deferred function (which include the receiver if the function is a method) are evaluated when the defer executes, not when the call executes." ? I continued reading this passage and I found this example:

func trace(s string) string {
    fmt.Println("entering:", s)
    return s
}

func un(s string) {
    fmt.Println("leaving:", s)
}

func a() {
    defer un(trace("a"))
    fmt.Println("in a")
}

func b() {
    defer un(trace("b"))
    fmt.Println("in b")
    a()
}

func main() {
    b()
}

// output:
// entering: b
// in b
// entering: a
// in a
// leaving: a
// leaving: b

OMG I got more confused. WHY was "entering" printed before "in" ??? I read the passage again and again and finally I thought I revealed what happened under the hood.

defer will acutally postpone the most outer function if we try to defer multiple functioin executions. In this case, the most outer function is un which needs a string parameter so trace will be executed to evaluate this parameter. In other words, the deferred function's parameters will be evaluaed when it is deferred. Therefore, "entering" is printed before "in".

Let's try a more complicated example and I'm pretty sure you know the drill:

func main() {
    b()
}

func entering(s string) string{
    fmt.Println("entering:", s)
    return s
}

func entered(s string) string {
    fmt.Println("entered:", s)
    return s
}

func leaving(s string) {
    fmt.Println("leaving:", s)
}

func a() {
    defer leaving(entered(entering("a")))
    fmt.Println("in a")
}

func b() {
    defer leaving(entered(entering("b")))
    fmt.Println("in b")
    a()
}

Tutorial: Integrate Gin with Cabsin

maxwellhertz — Wed, 02 Oct 2019 01:37:20 +0000

I've been working on a Java project recently in which our team uses Apache Shiro to do some authentication and authorization stuff. Now I just wonder if there exists a framework like Shiro in Go's world? After some search, I found Casbin. It's also a powerful and interesting library but maybe it's kinda hard for a newbie to write a demo from sractch.

I completed a basic web app based on Gin and Casbin after struggling for a whole afternoon. This tutorial is a simple replay. Hope it will help. :)

Structure of Our Project

root/
    main.go              # entry point of application                       
    handler/             # Gin handler functions
    middleware/          # Gin middlewares
    config/              # some configuration files like Casbin's rbac_model.conf
    component/           # global components like GORM DB instance

Initialize DB Connection And Cache

Create a file called persistence.go in component in which we will initialize DB connection using GORM and cache using BigCache.

import (
    "fmt"
    "github.com/allegro/bigcache"
    _ "github.com/go-sql-driver/mysql"
    "github.com/jinzhu/gorm"
    "time"
)

var (
    DB           *gorm.DB
    GlobalCache  *bigcache.BigCache
)

func init() {
    // Connect to DB
    var err error
    DB, err = gorm.Open("mysql", "your_db_url")
    if err != nil {
        panic(fmt.Sprintf("failed to connect to DB: %v", err))
    }

    // Initialize cache
    GlobalCache, err = bigcache.NewBigCache(bigcache.DefaultConfig(30 * time.Minute)) // Set expire time to 30 mins
    if err != nil {
        panic(fmt.Sprintf("failed to initialize cahce: %v", err))
    }
}

In our application, we will store Casbin's policies in DB (which we will talk about soon) and store current user in cache.

Configure Casbin

Model Configuration File

Ar first you may find some concepts in Casbin quite confusing. The first one is its model configuration file. I don't want to talk too much about it here (cuz I don't get it very well yet 😣) so I'm gonna give a simple example which is quite specific to our application. We will control a user's request based on
his role, which is called RBAC aka Role-based access control. Therefore we will create a rbac_model.conf in config directory.

[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act

A model configuration file tells Casbin how to determine if a user has some qualifications. In the above example, we just declare some stuff:

r = sub, obj, act defines that a limited request will consist of 3 parts: *sub*ject - user, *obj*ect - URL or more generally resource and *act*ion - operation.
p = sub, obj, act defines the format of policy. For example, admin, data, write means All admins can write data.
e = some(where (p.eft == allow)) means that a user can do something as long as there is a defined policy which allows him to do so.
g = _, _ defines the format of definition of user's role. For example, Alice, admin indicates Alice is an admin.
m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act defines the workflow of authorization: check user's role -> check the resource which user is trying to access -> check what user wants to do.

Note that there are at least four sections in a model configuration file: request_definition, policy_definition, policy_effect and matchers. Sometimes we don't do RBAC so role_definition section is not necessary.

Policies

Let's say we have some policies and user groups like this:

p, user, data, read
p, admin, data, read
p, admin, data, write
g, Alice, admin
g, Bob, user

Here we firstly define 3 policies:

All users can only read data.
All admins can read data.
All admins can also write data.

Then we assign roles to user:

Alice is an admin.
Bob is a user.

Thus Alice has full control over data while Bob can only read data. He will be blocked if he wants to write data.

Casbin allows us to simply store all policies in a CSV file and this is the most basic way. But this time we will store them in a MySQL DB. Casbin ususally stores policies in a table named casbin_rule and it will create this table automatically if not existed. In our case, the structure of table casbin_rule will look like this:

CREATE TABLE casbin_rule (
    p_type VARCHAR(100),
    v0 VARCHAR(100),
    v1 VARCHAR(100),
    v2 VARCHAR(100)
);

Add a policy:

INSERT INTO casbin_rule VALUES('p', 'user', 'data', 'read');

Add a user group:

INSERT INTO casbin_rule(p_type, v0, v1) VALUES('g', 'Bob', 'user');

Implement Gin Handler Functions

At first, we will implement the logic of user login.

// handler/user_handler.go

func Login(c *gin.Context) {
    username, password := c.PostForm("username"), c.PostForm("password")
        // Authentication
        // blahblah...

    // Generate random session id
    u, err := uuid.NewRandom()
    if err != nil {
        log.Fatal(err)
    }
    sessionId := fmt.Sprintf("%s-%s", u.String(), username)
    // Store current subject in cache
    component.GlobalCache.Set(sessionId, []byte(username))
    // Send cache key back to client in cookie
    c.SetCookie("current_subject", sessionId, 30*60, "/resource", "", false, true)
    c.JSON(200, component.RestResponse{Code: 1, Message:username + " logged in successfully"})
}

If a user has been identified, we need to store current user (or subject) in cache. In fact, what we do here is the same that Shiro stores current subject in session. Don't forget to send cache's key (or you can call it session id) back to client side.

Note that Shiro will do authentication stuff for us while Casbin just leaves that to us. So we have to implement authentication logic ourselves.

Don't forget to provide handlers for users to access resource:

// handler/resource_handler.go

func ReadResource(c *gin.Context) {
        // some stuff
        // blahblah...

    c.JSON(200, component.RestResponse{Code: 1, Message: "read resource successfully", Data: "resource"})
}

func WriteResource(c *gin.Context) {
        // some stuff
        // blahblah...

    c.JSON(200, component.RestResponse{Code: 1, Message: "write resource successfully", Data: "resource"})
}

After this, we should register these functions and start our application in main.go:

// main.go

var (
    router *gin.Engine
)

func init() {
    // Initialize gin router
    router = gin.Default()
    corsConfig := cors.DefaultConfig()
    corsConfig.AllowAllOrigins = true
    corsConfig.AllowCredentials = true
    router.Use(cors.New(corsConfig)) // CORS configuraion
    router.POST("/user/login", handler.Login)
    router.GET("/resource", handler.ReadResource)
    router.POST("/resource", handler.WriteResource)
}

func main() {
    defer component.DB.Close()

    // Start our application
    err := router.Run(":8081")
    if err != nil {
        panic(fmt.Sprintf("failed to start gin engin: %v", err))
    }
    log.Println("application is now running...")
}

Ok, almost done! The last piece and most important part of our application is to secure our API by RBAC.

Enforce Casbin Policies

Load Policies From DB

The first problem is: how can we load policies from DB dynamically? We can do this using Casbin Adapters. More specifically, we will use Gorm Adapter here.

The first step is to initialize an adapter with existing GORM instance:

// main.go

func init() {
    // Initialize  casbin adapter
    adapter, err := gormadapter.NewAdapterByDB(component.DB)
    if err != nil {
        panic(fmt.Sprintf("failed to initialize casbin adapter: %v", err))
    }

    // Initialize gin router
    router = gin.Default()
    corsConfig := cors.DefaultConfig()
    corsConfig.AllowAllOrigins = true
    corsConfig.AllowCredentials = true
    router.Use(cors.New(corsConfig)) // CORS configuraion
        router.POST("/user/login", handler.Login)
        router.GET("/resource", handler.ReadResource)
        router.POST("/resource", handler.WriteResource)
}

Apparently, we should force policies to control access to resource before any relevant handler functions are called. In my opinion, an elegant way to do this is utilizing Gin's middlewares and grouping routes. Firstly, let's define a middleware in which our policies will be enforced. I think the code showed below is self-explanatory:

// middleware/access_control.go

// Authorize determines if current subject has been authorized to take an action on an object.
func Authorize(obj string, act string, adapter *gormadapter.Adapter) gin.HandlerFunc {
    return func(c *gin.Context) {
                // Get current user/subject
        val, existed := c.Get("current_subject")
        if !existed {
            c.AbortWithStatusJSON(401, component.RestResponse{Message: "user hasn't logged in yet"})
            return
        }
        // Casbin enforces policy
        ok, err := enforce(val.(string), obj, act, adapter)
        if err != nil {
            log.Println(err)
            c.AbortWithStatusJSON(500, component.RestResponse{Message: "error occurred when authorizing user"})
            return
        }
        if !ok {
            c.AbortWithStatusJSON(403, component.RestResponse{Message: "forbidden"})
            return
        }
        c.Next()
    }
}

func enforce(sub string, obj string, act string, adapter *gormadapter.Adapter) (bool, error) {
        // Load model configuration file and policy store adapter
    enforcer, err := casbin.NewEnforcer("config/rbac_model.conf", adapter)
    if err != nil {
        return false, fmt.Errorf("failed to create casbin enforcer: %w", err)
    }
    // Load policies from DB dynamically
    err = enforcer.LoadPolicy()
    if err != nil {
        return false, fmt.Errorf("failed to load policy from DB: %w", err)
    }
        // Verify
    ok, err := enforcer.Enforce(sub, obj, act)
    return ok, err
}

At last, group all routes needed to secure and use our middleware:

// main.go

func init() {
    // Initialize  casbin adapter
    adapter, err := gormadapter.NewAdapterByDB(component.DB)
    if err != nil {
        panic(fmt.Sprintf("failed to initialize casbin adapter: %v", err))
    }

    // Initialize Gin router
    router = gin.Default()
    corsConfig := cors.DefaultConfig()
    corsConfig.AllowAllOrigins = true
    corsConfig.AllowCredentials = true
    router.Use(cors.New(corsConfig)) // CORS configuraion
        router.POST("/user/login", handler.Login)
        // Secure our API
    resource := router.Group("/api")
    {
        resource.GET("/resource", middleware.Authorize("resource", "read", adapter), handler.ReadResource)
        resource.POST("/resource", middleware.Authorize("resource", "write", adapter), handler.WriteResource)
    }
}

Boom! All are set to go! If a client doesn't log in firstly, he will be denied when he tries to GET /api/resource or POST /api/resource; if he is just a user, he will also be blocked when he wants to write data.

Source code