Forem: Tobias

Goroutines and Channels: Concurrency Patterns in Go

Tobias — Wed, 11 Dec 2024 06:34:34 +0000

Concurrency allows us to handle multiple tasks independently from each other. Goroutines are a simple way to process multiple tasks independently. In this post we progressively enhance a http handler which accepts files an explore various concurrency patterns in Go utilizing channels and the sync package.

Setup

Before getting into concurrency patterns let’s set the stage. Imagine we have a HTTP handler that accepts multiple files via form and processes the files in some way.


func processFile(file multipart.File) {
   // do something with the file
   fmt.Println("Processing file...")
   time.Sleep(100 * time.Millisecond) // Simulating file processing time
}
func UploadHandler(w http.ResponseWriter, r *http.Request) {
   // limit to 10mb 
   if err := r.ParseMultipartForm(10 << 20); err != nil {
       http.Error(w, "Unable to parse form", http.StatusBadRequest)
       return 
   }
   // iterate through all files and process them sequentially 
   for _, file := range r.MultipartForm.File["files"] {
       f, err := file.Open()
       if err != nil {
          http.Error(w, "Unable to read file", http.StatusInternalServerError)
          return
       }
       processFile(f)
       f.Close()
   }
}

In the example above we receive files from a form and process them sequentially. If 10 files are uploaded it would take 1 second to complete the process and send a response to the client.
When handling many files this can become a bottleneck, however with Go's concurrency support we can easily solve this issue.

Wait Groups

To solve this we can process files concurrently. To spawn a new goroutine we can prefix a function call with the go keyword e.g. go processFile(f). However as goroutines are non blocking the handler might return before the process is finished, leaving files possibly unprocessed or returning an incorrect state. To wait for the processing of all files we can utilize sync.WaitGroup.
A WaitGroup waits for a number of goroutines to finish, for each goroutine we spawn we additionally should increase the counter in the WaitGroup this can be done with the Add function. When a goroutine is finished Done should be called so the counter is decreased by one. Before returning from the function Wait should be called which is blocking until the counter of the WaitGroup is 0.

func UploadHandler(w http.ResponseWriter, r *http.Request) {
   if err := r.ParseMultipartForm(10 << 20); err != nil {
       http.Error(w, "Unable to parse form", http.StatusBadRequest)
       return
   }

   // create WaitGroup 
   var wg sync.WaitGroup 
   for _, file := range r.MultipartForm.File["files"] {
       f, err := file.Open()
       if err != nil {
          http.Error(w, "Unable to read file", http.StatusInternalServerError)
          return
       }

       wg.Add(1) // Add goroutine to the WaitGroup by incrementing the WaitGroup counter, this should be called before starting a goroutine
       // Process file concurrently
       go func(file multipart.File) {
           defer wg.Done() // decrement the counter by calling Done, utilize defer to guarantee that Done is called. 
           defer file.Close()
           processFile(f)
       }(f)
   }

   // Wait for all goroutines to complete
   wg.Wait()
   fmt.Fprintln(w, "All files processed successfully!")
}

Now for each uploaded file a new goroutine is spawned this could overwhelm the system. One solution is to limit the number of spawned goroutines.

Limiting Concurrency With A Semaphore

A Semaphore is just a variable we can use to control access to common resources by multiple threads or in this case goroutines.

In Go we can utilize buffered channels to implement a semaphore.

Channels

Before getting into the implementation let's look at what channels are and the difference between buffered and unbuffered channels.

Channels are a pipe through which we can send and receive data to communicate safely between go routines.
Channels must be created with the make function.

ch := make(chan int)

Channels have a special operator <- which is utilized to send or read from a channel.
Having the operator point at the channel ch <- 1 sends data to the channel, if the arrow points away from the channel <-ch the value will be received. Send and receive operations are blocking by default this means each operation will wait until the other side is ready.

The animation visualizes a producer sending the value 1 through an unbuffered channel and the consumer reading from the channel.

If the producer can send events faster than the consumer can handle then we have the option to utilize a buffered channel to queue up multiple messages without blocking the producer until the buffer is full. At the same time the consumer can handle the messages at its own pace.

ch := make(chan int, 2)

In this example the producer can send up to two items without blocking. When the capacity of the buffer is reached the producer will block until the consumer handled at least one message.

Back to the initial problem we want to limit the amount of goroutines processing files concurrently. To do this we can utilize buffered channels.

const maxConcurrentUploads = 5

func UploadHandler(w http.ResponseWriter, r *http.Request) {
   if err := r.ParseMultipartForm(10 << 20); err != nil {
       http.Error(w, "Unable to parse form", http.StatusBadRequest)
       return
   }

   semaphore := make(chan struct{}, maxConcurrentUploads) // channel used as a semaphore

   var wg sync.WaitGroup
   for _, file := range r.MultipartForm.File["files"] {
       f, err := file.Open()
       if err != nil {
          http.Error(w, "Unable to read file", http.StatusInternalServerError)
          return
       }

       wg.Add(1)
       go func(file multipart.File) {
           defer wg.Done()
           semaphore <- struct{}{} // Acquire slot by writing to the channel
           processFile(file)
           defer file.Close()
           <-semaphore // Release slot by reading from the channel
       }(f)
   }

   // Wait for all goroutines to complete
   wg.Wait()
   fmt.Fprintln(w, "All files processed successfully!")
}

In this example we added a buffered channel with a capacity of 5, this allows us to process 5 files concurrently and limit strain on the system.

But what if not all files are equal? We might can reliably predict that different file types or file size require more resources to process. In this case we can utilize a weighted semaphore.

Weighted Semaphore

Simply put with a weighted semaphore we can assign more resources to a single tasks. Go already provides an implementation for a weighted semaphore within the extend sync package.

const maxConcurrentUploads = 5

func UploadHandler(w http.ResponseWriter, r *http.Request) {
    if err := r.ParseMultipartForm(10 << 20); err != nil {
        http.Error(w, "Unable to parse form", http.StatusBadRequest)
        return
    }

    // Create a semaphore with a max of maxConcurrentUploads slots
    sem := semaphore.NewWeighted(int64(maxConcurrentUploads))

    var wg sync.WaitGroup
    for _, file := range r.MultipartForm.File["files"] {
        f, err := file.Open()
        if err != nil {
            http.Error(w, "Unable to read file", http.StatusInternalServerError)
            return
        }

        wg.Add(1)
        go func(file multipart.File, filename string) {
            defer wg.Done()
            slotsNeeded := getSlotsForFile(filename)

            // Acquire the semaphore with the needed slots
            if err := sem.Acquire(r.Context(), int64(slotsNeeded)); err != nil {
                http.Error(w, "Failed to acquire semaphore", http.StatusInternalServerError)
                return
            }
            defer sem.Release(int64(slotsNeeded)) // Release the slots after processing

            processFile(file)
            defer file.Close()
        }(f, file.Filename)
    }

    // Wait for all goroutines to finish
    wg.Wait()
    fmt.Fprintln(w, "All files processed successfully!")
}

func getSlotsForFile(filename string) int {
    ext := strings.ToLower(getFileExtension(filename))
    if ext == "pdf" {
        return 2
    }
    return 1
}

func getFileExtension(filename string) string {
    if idx := strings.LastIndex(filename, "."); idx != -1 {
        return filename[idx+1:]
    }
    return ""
}

In this version we created a weighted semaphore with 5 slots, if only images are uploaded for example the process handles 5 images concurrently, however if a PDF is uploaded 2 slots are acquired, which would reduce amount of files which can be handled concurrently.

Conclusion

We explored a few concurrency patterns in Go, utilizing sync.WaitGroup and semaphores to control the number of concurrent tasks. However there are more tools available, we could utilize channels to create a worker pool, add timeouts or use fan in/out pattern.
Additionally error handling is an important aspect which was mostly left out for simplicity.
One way to handle errors would be utilized channels to aggregate errors and handle them after all goroutines are done.

Go also provides a errgroup.Group which is related to sync.WaitGroups but adds handling of tasks which return errors.
The package can be found in the extend sync package.

Comparing Two RESTful APIs With JMeter

Tobias — Thu, 23 Mar 2023 02:51:13 +0000

JMeter is commonly used for performance testing, but it can do more than that. In this post, we explore how to compare two RESTful APIs.

Comparing API Results

Sometimes, you might need to compare the results of two RESTful API responses.

To do this, you should first create a Thread Group and add a Simple Logic Controller.

Within the Simple Controller, you can now add the two HTTP requests that you want to compare.
You should have a structure similar to the following:

For simplicity, let’s assume both endpoints return a JSON structure with a unique identifier, and it’s enough to compare if all the identifiers are present and in order.

A JSON extractor can be used to run JSON Path expressions against a JSON response.
A variety of options in the JSON extractor:

Variable Names: this is a required field and needed for later
JSON Path expression: again a required field, here we describe the value we’d like to extract
Match No: you can adjust which values to extract,
- -1 to extract all
- 0 for random extraction
- N (any number) to extract the result at position N
Compute concatenation var: this will concatenate results by using , as separator

As you can see we will set the Variable name to v1id, we try to extract id's with $..id, match any result by setting matching no. to -1 and concatenate the results.

You should repeat the same step for the second HTTP request, but be aware to change the variable name or you will overwrite the results.

To Compare the two results we can add a JSR223Sampler (groovy sampler) to the Simple Controller

Add the following code to the sampler to compare the results of the JSON extractor



if(vars.get("v1id_ALL") != vars.get("v2id_ALL")){
    log.error("got: "+vars.get("v2id_ALL")+" \n want: "+vars.get("v1id_ALL"))
    SampleResult.setSuccessful(false)
}

the sampler should be last in the simple controller to not be executed before the HTTP requests

Within a View Result Tree Listener you can see the sampler failing if id's don't match. You can also see a error message in the jmeter log file if you want to see the exact result.

But there is more we can compare.
We might be interested in the response time, maybe our new API should be faster than the previous version.
To do this we can add a JSR223Sampler to each HTTP requests with the following code



vars.put("responseTimeV1", String.valueOf(prev.getTime()));

And we can then again compare the two variables in the sampler of the simple controller.

You can do more with JMeter than just the traditional performance testing. You could even add functional tests within the above test plan. This allows you to manage all your tests with one tool, making your test setup hopefully simpler by using less tools.

Test Fixtures in Go

Tobias — Mon, 13 Mar 2023 03:56:15 +0000

When testing, we often need some kind of test data. In Go, test data can often come in the form of structs. In this post, we explore various methods of creating reusable test data, and I will also introduce a library I am currently working on to assist you in the process of creating test data.

Creating Test Data

Let’s assume we have a structure as follows:

type Book struct{
   ID int
   Title string
}

type Author struct {
   ID int
   Name string
   Books []Book
}

func GetAuthor() Author {
// return an author
}

In most common IDEs, you can generate unit tests for a function. In most cases, a table-driven test is then generated, similar to the following:

func TestGetAuthor(t *testing.T) {
    tests := []struct {
        name string
        want Author
    }{
        // TODO: Add test cases.
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            if got := GetAuthor(); !reflect.DeepEqual(got, tt.want) {
                t.Errorf("GetAuthor() = %v, want %v", got, tt.want)
            }
        })
    }
}

Now, you could just initialize the Author in each test case, making it a simple and straightforward approach. But this is somewhat time-consuming and repetitive. And as soon as Author changes, you have to fix every test.

Instead of initializing the struct directly within the test, we can create a builder function with functional parameters.

type Opts func(*Author)
func BuildAuthor(options ...opts) Author {
  // we can set some default values here
  author := &Author{ID: 1}
  for _,option := range options {
     option(author)
  }
  return *Author
}

// usage
someAuthor := BuildAuthor(func(a *Author){ a.Name="Name"})

With this option, we can create reusable test data and can easily call BuildAuthor within the test cases.

Because I found myself creating similar code to the above, I’ve created a library to assist in creating the test data.

Introduction to the Library

The motivation behind testcraft was to remove some of the boilerplate needed during the test data creation and add a few little helper functions. A simple example as the above could look like this:

authorFactory := testcraft.NewFactory(Author{}).Attr(func(a *Author) error {
   a.ID = 1
   a.Name="Name"
   return nil
})
someAuthor,err := authorFactory.Build()
// or use MustBuild, MustBuild panics on error
someAuthor := authorFactory.MustBuild()

This is basically the same code as the previously shown BuildAuthor function.

But there is more you can do with testcraft. Let’s look at some of the helpers. Having always the same ID is somewhat unrealistic. To get a sequence of IDs, you can create a Sequence:

// NewSequencer can take any of the following types
// int, int32, int64, int8, float32, float64
authorSeq := NewSequencer(1)

// in the previously declared factory, we can now use
a.ID = authorSeq.Next()

Anytime Build is called, the ID will now increase by 1. The increment can be changed with userSeq.SetIncrement(2).

If you just need some data and don’t care about the values, there is a Randomize function, which initializes the struct with random values:

authorSeq := NewSequencer(1)
authorFactory := testcraft.NewFactory(Author{}).Attr(func(a *Author) error {
   a.ID = authorSeq.Next()
   a.Name="Name"
   return nil
})
// Randomize ignores the previously set values
randomAuthor, err := authorFactory.Randomize() 
// RandomizeWithAttrs will apply the previously set attributes and randomize the rest.
randomAuthor2, err := authorFactory.RandomizeWithAttrs()

Next, let's have a look at how to combine factories and use the Multiple function:

// As a reminder, here are the structs again.
type Book struct{
   ID int
   Title string
}

type Author struct {
   ID int
   Name string
   Books []Book
}
// create sequence for books
bookSeq := NewSequencer(1)
// define book factory
bookFactory := testcraft.NewFactory(Book{}).Attr(func(b *Book) error{
   b.ID = bookSeq.Next()
})
// create sequence for author
authorSeq := NewSequencer(1)
// define author factory
authorFactory := testcraft.NewFactory(Author{}).Attr(func(a *Author) error {
   a.ID = authorSeq.Next()
   // // AlphanumericBetween returns a random string of alphanumeric characters between min(3) and max(10).
   a.Name = datagen.AlphanumericBetween(3,10)
   // Multiple creates a slice of Book with 5 elements.
   a.Books = testcraft.Multiple(5, func(i int) Book {
     return bookFactory.MustRandomizeWithAttrs() 
     // functions prefixed with Must panic on error
   })
   return nil
})

In the above code, you also get a first glimpse of the datagen package as we generate an alphanumeric string between 3 and 10 characters.

I think the datagen package could still use some work, but for some basic work it’s good to go.

I am still actively working on testcraft and add a few more features and improve the data generation package.

An Introduction to Go Ast - Generating SQL Queries Based on Structs

Tobias — Wed, 01 Mar 2023 03:48:32 +0000

The Go standard library includes an 'ast' package, which provides a set of tools for working with the abstract syntax tree (AST). The AST represents the structure of a Go program in a tree-like data structure and can be used to analyze, modify, or generate code programmatically. In this post, we investigate the possibility of analyzing structs and generating a SQL query based on them.

Goal

Our goal involves generating a SQL query based on a given struct with specific formatting requirements. The struct name should be pluralized and lowercase, while the field names should be in snake_case and also lowercase.

For example, given the following struct

type User struct {
    ID             int
    Name           string
    InvitationCode string
}

The generated SQL query should be an INSERT statement into a table named users with columns id, name, and invitation_code. The values to be inserted should be represented using placeholders in the query, such as :id, :name, and :invitation_code.

Formatting Requirements

Let's create some helper functions to normalize the structs and field names.
The first requirement we look at is to transform UpperCamelCase into snake_case

We can do this by simply iterating over the string and anytime we see a uppercase character and the previous character is lower case we insert a underscore _

func SnakeCase(s string) string {
    var str strings.Builder
    var prev rune
    for i, r := range s {
        // check if we should insert a underscore
        if i > 0 && unicode.IsUpper(r) && unicode.IsLower(prev) {
            str.WriteRune('_')
        }
        // lower case all characters
        str.WriteRune(unicode.ToLower(r))
        prev = r
    }
    return str.String()
}

Next up we want to pluralize the string,
this could be done in similar fashion to this

func Pluralize(s string) string {
    if strings.HasSuffix(s, "y") {
        return strings.TrimSuffix(s, "y") + "ies"
    }
    return s + "s"
}

As a little helper function lets add a NormalizeTableName() function which combines the above into one

// normalize the struct name to lowercase, pluralize it and apply snakeCase
// for example, User -> users, ReviewPost -> review_posts
func normalizeTableName(name string) string {
    return pluralize(snakeCase(name))
}

With those helper functions out of the way let's have a look on how to parse and analyse the file.
To do this we can use the go/ast package, the ast package provides us with a set of tools for working with the abstract syntax tree.

File Parsing

The Go standard library comes with a handy package to parse Go source code files and generate a AST based on it.

func ExtractStructs(filePath string) []QueryBuilder {
    // create a file set, this keeps track of file positions for error messages and other diagnostic output
    fset := token.NewFileSet()
    // parse the whole file including errors
    parsedAST, err := parser.ParseFile(fset, filePath, nil, parser.AllErrors)
}

After parsing the file and checking for potential errors we use the go/ast package to inspect the abstract syntax tree.

Inspect traverses an AST in depth first order. If the function returns true, Inspect is called recursively.

ast.Inspect(parsedAST, func(n ast.Node) bool {
   return true
})

In our case we are only interested if our node of type ast.TypeSpec and ast.StructType.
TypeSpec is needed to get the actual struct name and ast.StructType let's us iterate over the struct fields

ast.Inspect(parsedAST, func(n ast.Node) bool {
    // try to convert n to ast.TypeSpec
    if typeSpec, isTypeSpec := n.(*ast.TypeSpec); isTypeSpec {
        s, isStructType := typeSpec.Type.(*ast.StructType)
        // check if conversion was successful
        if !isStructType {
            return true
        }
        // get the struct name
        structName := normalizeTableName(typeSpec.Name.Name)
        // get Fields helper function
        fields := getFields(s)
    }
    return true
})

As you've probably noticed there is another helper function getFields
The implementation looks like this.

func getFields(s *ast.StructType) []string {
    fields := make([]string, len(s.Fields.List))
    for i, field := range s.Fields.List {
        if len(field.Names) == 0 {
            continue
        }
        fields[i] = SnakeCase(field.Names[0].Name)
    }
    return fields
}

With this function we got all we need. We are able to extract the struct name and the field names additionally apply snake_case and pluralization respectively.

Query Generation

As we already extracted all the needed data we can introduce a simple function to generate the needed queries. To keep this nice and clean we use another struct which defines the needed data for a query and a struct method to return a insert query.

// define the QueryBuilder struct
type QueryBuilder struct {
    TableName string
    Fields    []string
}
// Generate a insert query
func (q QueryBuilder) InsertQuery() string {
    return fmt.Sprintf("INSERT INTO %s (%s) VALUES (:%s)", strings.ToLower(q.TableName), strings.Join(q.Fields, ", "), strings.Join(q.Fields, ", :"))
}
// add more implementation for the needed queries

and at last let's update the ExtractStructs function and combine everything

func ExtractStructs(filePath string) []QueryBuilder {
    fset := token.NewFileSet()
    parsedAST, err := parser.ParseFile(fset, filePath, nil, parser.AllErrors)
    if err != nil {
        fmt.Println(err)
        return nil
    }

    // Find all struct declarations
    var structs []QueryBuilder
    ast.Inspect(parsedAST, func(n ast.Node) bool {
        // try to convert n to ast.TypeSpec
        if typeSpec, isTypeSpec := n.(*ast.TypeSpec); isTypeSpec {
            s, isStructType := typeSpec.Type.(*ast.StructType)
            // check if conversion was successful
            if !isStructType {
                return true
            }
            // get the struct name
            structName := typeSpec.Name.Name
            // get Fields helper function
            fields := getFields(s)
            structs = append(structs, QueryBuilder{TableName: normalizeTableName(structName), Fields: fields})
        }
        return true
    })
    return structs
}

You can find the full code example here

A CI/CD Pipeline With GitHub Actions for Go

Tobias — Mon, 27 Feb 2023 22:39:49 +0000

Continuous Integration/Continuous Deployment (CI/CD) is an essential practice for modern software development. It helps developers to streamline the software development process by automating testing, building, and deploying their applications. GitHub Actions is a powerful tool for setting up CI/CD pipelines for your projects. In this post, we will explore how to set up a CI/CD pipeline using GitHub Actions for Go, which will help you to deploy your applications quickly and reliably. We'll cover the steps needed to create a pipeline that deploys to a staging and production environment from the same branch.

Creating the Staging Pipeline

First we utilize golangci-lint, this helps us to identify potential bugs, errors and stylistic issues.
After that we will run unit tests and start the build and deployment process. For demonstration purposes we will perform a straightforward Docker build and tag command as our build process.

name: Build

on:
  push:
    branches: 
      - main
  pull_request:
    branches: 
      - main

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Set up Go
      uses: actions/setup-go@v3
      with:
        go-version: 1.20
    - name: golangci-lint
      uses: golangci/golangci-lint-action@v3
      with:
        version: latest
    - name: Test
      run: go test -v ./...
    - name: Build
      run: docker build -t latest .

To enhance the above process, I recommend setting up notifications to a slack channel or a commonly used tool in your environment. This eliminates the need to constantly monitor the GitHub Action. Additionally, a step similar to this can be added to sent notifications on job failure.

 - name: Notify on failure
   if: failure()
   run: # do something

Releasing to Production

As mentioned before we wan't to do this within the same branch. To do this we will create two jobs, one which creates a release and tags the latest commit and another one which is triggered by the tag to start the deployment process.
This needs a bit of preparation, because GitHub Actions can't trigger another action.
To achieve this you can head to the branches settings and add a deploy key.
For the rest of the article I assume the key is named COMMIT_KEY.

Let' have a look at the change log generation action.

name: Deploy Production

on:
  workflow_dispatch:

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout Code
        uses: actions/checkout@v3
        with:
          fetch-depth: 0
          ssh-key: "${{ secrets.COMMIT_KEY }}" # set deploy key here
      - name: Create Release Tag
        id: release-tag
        # for demonstration purposes GitHub run number is used
        run: echo "tag=r-$(printf $GITHUB_RUN_NUMBER)" >> $GITHUB_OUTPUT
      - name: Find last successful workflow 
        uses: SamhammerAG/last-successful-build-action@v2
        with:
          branch: "main"
          workflow: "Build"
          verify: true
      - name: Push Tag
        run: |
          git tag ${{ steps.release-tag.outputs.tag }} ${{ github.sha }}
          git push origin ${{ steps.release-tag.outputs.tag }}
      - name: Create Release
        uses: ncipollo/release-action@v1
        with:
          allowUpdates: true
          draft: false
          name: ${{ github.ref_name }}
          tag: ${{ steps.release-tag.outputs.tag }}
          body: # create a change log 
          token: ${{ secrets.GITHUB_TOKEN }}

In this action we will fetch the commit of the last successful run of our deploy to staging action. Because of this we can assume that our code is in a deployable state and we can skip running tests again. If you are using conventional commits you could use this action to generate the body for the release.

Lastly, we have an action that initiates the release to the production environment triggered by a push event specifically targeting tags.

name: Release Production

on:
  push:
    tags:
      - '' # use regex to limit to certain tags
jobs:
  build:
    runs-on: ubuntu-latest
  steps:
    - uses: actions/checkout@v3
      with:
        ref: ${{ github.ref }} # set the ref
    - name: Build and Deploy # add your deployment process here
      run: docker build -t latest .

In conclusion, setting up a CI/CD pipeline with GitHub Actions for Go is a relatively simple process that can bring significant benefits to your development workflow. By automating the build, test, and deployment processes, you can reduce errors, improve efficiency, and free up your time to focus on more important tasks. With the examples provided in this post, you should have a good starting point for setting up your own CI/CD pipeline with GitHub Actions and Go. Always test and refine your pipeline to ensure that it meets your specific needs and requirements.