Forem: Ryo Yamaoka

Library for easy implementation of instant load testing

Ryo Yamaoka — Tue, 27 Sep 2022 15:38:35 +0000

Motivation

There are many load testing tools in the world, and most of them are highly functional, with the ability to create scenarios and send requests from multiple nodes. Of course that's great, but

I would like to check if the newly implemented API is meeting the target RPS.
I want to see if there is a significant impact on performance before and after adding indexes.

In the case of simple requirements, such as the need to see if there is a significant impact on performance before and after adding indexes, I think it is a bit tedious and time-consuming to set up the tool. On the other hand, if you try to implement an instant tool by yourself

Processing of test parameter acquisition
Performance-aware implementation
Control of concurrency
Statistics output

As you can see, there are many troublesome implementations that require a certain amount of effort. Or, if you try to use a simpler testing tool, it may not be able to reach your itch.

I have created the Otchkiss in the hope that a library that can easily implement load testing tools for such cases would make it easier and more convenient to perform load testing.

https://github.com/ryo-yamaoka/otchkiss

It's read as "Ochikisu" and origin is the machine gun manufacturer Hotchkiss because this tool does not have as much firepower as the Gatling.

The lose of the H at the beginning is not a typo, but a French reading, and furthermore, the intention is to use a different string of letters just in case something might be bad trademark-wise.

Feel of using

$ go run ./sample/...

[Setting]
* warm up time:   5s
* duration:       1s
* max concurrent: 1

[Request]
* total:      90
* succeeded:  90
* failed:     0
* error rate: 0 %
* RPS:        90

[Latency]
* max: 11.0 ms
* min: 10.0 ms
* avg: 10.9 ms
* med: 11.0 ms
* 99th percentile: 11.0 ms
* 90th percentile: 11.0 ms

How to use

Basics

If you want to try using it for a while, you can get a feel for it by looking at the sample, but the following steps are the basic flow.

Implement the Requester interface
Pass the interface implementation to Otchkiss and call Start()
Call Report() to output the results
Build the above code and make it into a CLI tool

https://github.com/ryo-yamaoka/otchkiss/blob/main/sample/main.go

If implemented using otchkiss.New(), the following command line options can be used to change the configuration values at runtime, so there is no need to rebuild even in the case of a trial -> main test.

-p: Run goroutine as many times as specified here and make requests in parallel.
-d: The execution time of the test.
-w: Time to warm up the requestor. Requests will send for the time specified here, but will not be reflected in the test results.

It is difficult to understand -d and -w, but the time when the request actually keeps requesting to the test target is d + w, and the time specified by -d is the time to be aggregated as the test result.

Specify the number of concurrency and execution time

As mentioned above, the basic implementation uses the value specified on the command line or the default value, but if you wish to fix this value, you can hard-code it.

If you want to fix this value, you can hardcode it. In this case, you can use a sub-package github.com/ryo-yamaoka/otchkiss/setting and put the value into the configuration struct otchkiss.FromConfig() .

    s := &setting.Setting{
        MaxConcurrent: 10,
        RunDuration: 60 * time.Second,
        WarmUpTime: 10 * time.Second,
    }
    ot, err = otchkiss.FromConfig(&SampleRequester{}, s, 1000)

Performance tuning

Otchkiss sends many parallel requests and keeps track of their success or failure and latency, but internally it uses slices.

As many of you may know from blog posts, Go's slices behave in such a way that when the allocated memory is insufficient, they allocate a larger memory area.

FYI: Go Slices: usage and internals

Therefore, frequent allocations may affect performance (i.e., RPS). On the other hand, since memory is finite, it is not advisable to allocate too large an amount.

The default allocation is for 100,000,000 requests, but since the optimal value may differ for each user, this can also be specified. (Incidentally, if you specify 1 billion as shown in the following code, several GB of memory will be consumed at the moment of startup.)

    ot, err = otchkiss.FromConfig(&SampleRequester{}, s, 1000*1000*1000)

Change the format in which results are displayed

The default display is as shown at the beginning of this document, but you can use otchkiss.TemplateReport() to output in your preferred format. The only thing you need to do is to pass the Go template itself, so it is not particularly difficult to specify.

The parameters are all the ones used in the default template below.

https://github.com/ryo-yamaoka/otchkiss/blob/main/template.go

TODO

Histogram display
Visual enhancements such as time series graphs

Multiple error handling in Go

Ryo Yamaoka — Tue, 27 Sep 2022 15:14:32 +0000

Go has the feature that concurrency can be easily described by goroutine (Though it is not always easy to get it to worked correctly).

The problem at this point is error handling, which is a major issue in actual application implementation.

If an error occurs in one of the goroutines, we want to suspend the other goroutines.
Wait until each goroutine finishes processing, and handle each error individually.

I think that these requirements often occur.
Of course, it is possible to implement these processes on your own, but there are libraries that are suitable for each of these requirements, and we would like to introduce them here along with sample notation.

Cases and usage for which errgroup is suitable

The golang.org/x/sync/errgroup provided as an official Go experimental package focuses primarily on handling single errors and is suitable for use similar to the standard sync package or for implementing processes that require interruption on error.

Continue to the end even if an error occurs

Group structure and pass the actual function to the .Go() method to start the goroutine, and similarly, .Wait() can be used to wait for the goroutine and retrieve errors. Unlike the sync package, the number of goroutine launches is handled internally, so you don't need to be particularly aware of it. In this usage, it is like a convenient sync package.

In the following implementation, the first goroutine displays 1 and returns an error after 100 ms of waiting, and the second displays 2 and returns an error after 200 ms of waiting.

Since the specification of errgroup is that it only returns the first error that occurs, the second and subsequent errors are lost and cannot be handled.

And since goroutine waits until all processing is completed, both signals 1 and 2, which are signals that processing has been completed, are displayed, but only errgroup: err1, which was returned first, is displayed for errors.

func errGroup() {
    var eg errgroup.Group

    eg.Go(func() error {
        time.Sleep(100 * time.Millisecond)
        fmt.Println("1")
        return fmt.Errorf("errgroup: err1")
    })
    eg.Go(func() error {
        time.Sleep(200 * time.Millisecond)
        fmt.Println("2")
        return fmt.Errorf("errgroup: err2")
    })

    if err := eg.Wait(); err != nil {
        fmt.Println(err.Error())
    }
}

$ go run main.go 
1
2
errgroup: err1

Interruption of other goroutine

In the previous example, it was a simple wait + retrieval of prior errors, similar to the standard sync implementation, but it is also possible to suspend a goroutine that has not generated any errors using context. In practical use, this mechanism may be useful, for example, in cases where multiple APIs are being queried simultaneously, but if any one of them fails, the whole process will be interrupted.

The following implementation incorporates context handling into the previous process to allow interruption. 100ms or 200ms wait and display 1 or 2 as a sign of completion is not changed, but ctx.Done() is also waiting at the same time, so if cancel() is invoked, it will be forced to interrupt at that point.

When initialized with errgroup.WithContext(), it uses context.WithCancel(ctx) internally to return the context to the caller. Furthermore, it keeps a cancel function, which is triggered when an error occurs.

https://cs.opensource.google/go/x/sync/+/7fc1605a:errgroup/errgroup.go;l=45

When executed, the first goroutine prints 1 and returns an error after 100ms of waiting. Since cancel() is executed at the same time, the second goroutine aborts execution before the end of 200ms, resulting in a context canceled.

func errGroupCtx() {
    ctx := context.Background()
    eg, ctx := errgroup.WithContext(ctx)

    eg.Go(func() error {
        select {
        case <-time.After(100 * time.Millisecond):
            fmt.Println("1")
            return fmt.Errorf("errgroupctx: err1")
        case <-ctx.Done():
            fmt.Println("errgroupctx: ctx done1", ctx.Err())
            return ctx.Err()
        }
    })
    eg.Go(func() error {
        select {
        case <-time.After(200 * time.Millisecond):
            fmt.Println("2")
            return nil
        case <-ctx.Done():
            fmt.Println("errgroupctx: ctx done2", ctx.Err())
            return ctx.Err()
        }
    })

    if err := eg.Wait(); err != nil {
        fmt.Println(err.Error())
    }
}

$ go run main.go
1
errgroupctx: ctx done2 context canceled
errgroupctx: err1

Cases and usage for which go-multierror is suitable

Unlike errgroup, github.com/hashicorp/go-multierror cannot be interrupted by context. However, it is useful for use cases where you need to check all errors and handle them carefully, because you can keep all errors.

Summarize of all errors that occur in goroutine.

go-multierror.Group type is used to control goroutine in the same way as for errgroup.

The following code returns an error after waiting 100ms or 200ms as in the previous example, but unlike the errgroup example, the second error is also logged. To retrieve the second error, refer to the .Errors member and get by the range statement.

The go-multierror.Error type returned by .Wait(), which is a goroutine Wait(), can be returned as a normal error because it implements the error interface. The .Error() method is also kindly designed to output the error in a somewhat human readable format (formatting can also be specified, though this is omitted this time).

import (
    multierror "github.com/hashicorp/go-multierror"
)

func goMultierror() {
    var meg multierror.Group

    meg.Go(func() error {
        time.Sleep(100 * time.Millisecond)
        return fmt.Errorf("multierror: err1")
    })
    meg.Go(func() error {
        time.Sleep(200 * time.Millisecond)
        return fmt.Errorf("multierror: err2")
    })

    merr := meg.Wait()
    for _, err := range merr.Errors {
        fmt.Println(err.Error())
    }
    fmt.Println(merr.Error())
}

multierror: err1
multierror: err2
2 errors occurred:
        * multierror: err1
        * multierror: err2

To handle multiple errors at once

Although it is out of the scope of concurrency processing, it is possible to simply append errors like a slice instead of the standard error wrapping and keep the errors in an easy-to-handle format.

Even if the error to be appended is an original type, it can be handled with errors.As errors.Is of the standard package if the Unwrap method is implemented (of course, the same applies to errors returned from goroutine).

Furthermore, if a function returns a go-multierror.Error type as an error type, you can slice and dice it out by making a type assertion.

type ErrX struct {
    err error
    msg string
}

func (ex *ErrX) Error() string {
    return ex.msg
}

func (ex *ErrX) Unwrap() error {
    return ex.err
}

type ErrY struct {
    err error
    msg string
}

func (ey *ErrY) Error() string {
    return ey.msg
}

func (ey *ErrY) Unwrap() error {
    return ey.err
}

func goMultierrorAppend() {
    var err error

    errX := &ErrX{err: nil, msg: "multierror-append: err1"}
    errY := &ErrY{err: nil, msg: "multierror-append: err2"}
    err = multierror.Append(err, errX)
    err = multierror.Append(err, errY)

    merr, ok := err.(*multierror.Error)
    if !ok {
        fmt.Println("failed to assert multierror")
        return
    }
    for _, err := range merr.Errors {
        fmt.Println(err.Error())
    }
    fmt.Println(merr.Error())

    fmt.Println(errors.As(err, &errX))
    fmt.Println(errors.Is(err, errY))
}

multierror-append: err1
multierror-append: err2
2 errors occurred:
        * multierror-append: err1
        * multierror-append: err2


true
true

Conclusion

If you do not need to handle all errors, or if you want to suspend goroutine on error, errgroup is suitable.
go-multierror is suitable if you want to handle all errors that occur.

Summary of Spanner Change Streams

Ryo Yamaoka — Wed, 29 Jun 2022 14:05:19 +0000

Spanner Change Streams was recently released by GA, then here is a summary of what we found out about it.

What is Spanner Change Streams?

Change Streams is a mechanism to obtain the contents of data changes (INSERT, UPDATE, DELETE) made to Spanner in near real-time.

Replication of changes to BigQuery, etc. for analysis
Messaging to Pub/Sub, etc. triggered by data changes
Save changes to GCS and use as audit logs

Since Spanner did not have this kind of change detection functionality until now, it was impossible to implement the Transaction log tailing pattern when sending events triggered by data updates, for example, and instead it was necessary to implement the Transactional outbox pattern on the application side.

Delivery Integrity

The writes to the internal table for delivery by Change Streams are done in the same transaction as the application writes, so delivery is guaranteed as soon as the commit is successful.

The details are described below, but since the data is localized so that writing to the internal table does not cause delays, the impact on performance is probably not a concern.

The delivered values include table name, column name, column data type, timestamp, transaction ID, and other metadata, as well as modified values and PK. If you want to retrieve the entire record, you must use Spanner's Stale Read from the PK and commit timestamp.

Scalability of Change Streams

Spanner stores real data in units of splits on Colossus (distributed storage), and when the number of splits exceeds a certain size, the data is automatically divided into smaller units to achieve scalability.

Change Streams also follows the same mechanism, whereby a Change Stream Partition is tied to a single Split, and this Partition is linked to the increase or decrease of the Split to ensure scalability. In addition, the partitions are arranged to be localized and tied to the Split, like data interleaving, for the fastest possible processing.

picture from here: https://cloud.google.com/spanner/docs/change-streams/details#change_stream_partitions

Enabling

Putting in the settings itself is relatively simple and can be done by flowing DDL similar to indexes and unique constraints.
The scope is

database as a whole
per table
per column

For example, the following is a sample configuration for each scope.

CREATE TABLE Users (
    UserID STRING(MAX) NOT NULL,
    Name STRING(MAX) NOT NULL,
    Age INT64 NOT NULL,
) PRIMARY KEY (UserID);

CREATE TABLE Items (
    ItemID STRING(MAX) NOT NULL,
    Name STRING(MAX) NOT NULL,
) PRIMARY KEY (ItemID);

CREATE CHANGE STREAM EverythingStream FOR ALL;

CREATE CHANGE STREAM UsersTableStream FOR Users;

CREATE CHANGE STREAM UsersNameAndItemsNameColumnsStream FOR Users(Name), Items(Name);

Usage Methods

The official documentation shows how to use Dataflow and how to use Spanner directly by calling the Spanner API, but it seems that Dataflow is the most important one, and several templates are provided that seem to be practical.

However, Dataflow seems to be the real deal, and there are several templates that seem to be practical. For example, when the aforementioned partitioning of a partition occurs, a record notifying that fact is returned, so it may be necessary to start a new process based on the contents of that notification to acquire the stream from another table. Conversely, when a merge of partitions (=merge of splits) occurs, it may be necessary to delete unnecessary processes and stop receiving streams.

Naturally, Dataflow templates have all of this covered and are readily available for correct operation, so it would be wise to consider using Dataflow unless there is a good reason to do so.

When I first saw the news, I was expecting to be able to easily realize transactional log tailing from Go, but it was not that simple.

I touched it a little bit.

However, I tried to touch it a little from Go, so I'm pasting the code as a memorial service. It took some trial and error, but I was able to run it from Go's client library without directly touching the API.

This code is just a simple like "I got something from Change Streams! Yeahh 🥰". This code is not a reference code at all, because it is not useful at all, and it has more holes than anthill in the savanna.

The schema is the same as the one I used in the "Enable" section.

package main

import (
    "context"
    "fmt"
    "os"
    "time"

    "cloud.google.com/go/spanner"
    "google.golang.org/api/iterator"
)

func main() {
    if err := run(); err != nil {
        fmt.Fprintln(os.Stderr, err)
    }
}

func run() error {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    dsn := fmt.Sprintf("projects/%s/instances/%s/databases/%s", os.Getenv("SPANNER_PROJECT_ID"), os.Getenv("SPANNER_INSTANCE_ID"), os.Getenv("SPANNER_DATABASE_ID"))
    cli, err := spanner.NewClient(ctx, dsn)
    if err != nil {
        return fmt.Errorf("failed to init spanner client: %w, dsn: %s", err, dsn)
    }

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            now := time.Now().UTC().Format(time.RFC3339)
            stmt := spanner.NewStatement(
                `SELECT ChangeRecord FROM READ_EverythingStream ( ` +
                    `  start_timestamp => "` + now + `", ` +
                    `  end_timestamp => NULL, ` +
                    `  partition_token => NULL, ` +
                    `  heartbeat_milliseconds => 1000` +
                    `)`,
            )
            iter := cli.Single().Query(ctx, stmt)
            defer iter.Stop()
            for {
                row, err := iter.Next()
                if err != nil {
                    if err == iterator.Done {
                        <-time.After(1 * time.Second)
                        break
                    }
                    return err
                }
                fmt.Println(row.String())
            }
        }
    }
}

$ go run main.go 
{fields: [name:"ChangeRecord" type:{code:ARRAY array_element_type:{code:STRUCT struct_type:{fields:{name:"data_change_record" type:{code:ARRAY array_element_type:{code:STRUCT struct_type:{fields:{name:"commit_timestamp" type:{code:TIMESTAMP}} fields:{name:"record_sequence" type:{code:STRING}} fields:{name:"server_trans
（略）
context deadline exceeded