inserter.go

package main
import (
        "flag"
        "log"
        "time"
        "sync"
        "github.com/valyala/fasthttp"
)
var (
        kittenhouse = flag.Bool("kittenhouse", false, "Whether or not to send INSERTs in kittenhouse format")
        clickhouse = flag.Bool("clickhouse-async", false, "Whether or not to use clickhouse-async")
        persistent  = flag.Bool("persistent", false, "(for kittenhouse) Whether or not use persistent mode")
        addr        = flag.String("addr", "127.0.0.1:8124", "Address of bulk inserter")
        requests = flag.Int("requests", 200, "number of requests to send")
        concurrency = flag.Int("concurrency", 5000, "concurrency of the requests")
)
func main() {
        flag.Parse()
        table := `InsertTest_buffer(id)`
        const N = 200
        postURL := "http://" + (*addr) + "/?query=INSERT%20INTO%20" + table + "%20VALUES"
        if *clickhouse {
        postURL = "http://" + (*addr) + "/?wait_for_async_insert=0&async_insert=1&query=INSERT%20INTO%20" + table + "%20VALUES"
}
        bodyStr := `(1)`
        body := []byte(bodyStr)
        //      inserted := 0
        start := time.Now()
        var wg sync.WaitGroup
        cl := &fasthttp.Client{
                MaxConnsPerHost: 50000,
        }
        log.Printf("Sending %d requests to %q with concurrency %d", *requests, postURL, *concurrency)
        for j := 0; j < *concurrency; j++ {
                wg.Add(1)
                go func() {
                        for i := 0; i < *requests; i++ {
                                req := fasthttp.AcquireRequest()
                                resp := fasthttp.AcquireResponse()
                                req.Reset()
                                resp.Reset()
                                req.SetBodyRaw(body)
                                req.SetRequestURI(postURL)
                                req.Header.SetMethod("POST")
                                err := cl.Do(req, resp)
                                if err != nil {
                                        log.Fatalf("Request failed: %v", err)
                                }
                        }
                        wg.Done()
                }()
        }
        wg.Wait()
        log.Printf("QPS: %.1f (for %s)", float64((*concurrency) * (*requests))/(float64(time.Since(start))/float64(time.Second)), time.Since(start))
}