m3db_http_client.go

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "io/ioutil"
    "math/rand"
    "net/http"
    "sync"
    "time"
)

const (
    factorDegree = 200
    batchSize    = 500

    servers = 50000

    concurrency = 20
)

func main() {
    start := time.Now()
    bench()
    d := time.Now().Sub(start).Seconds()
    t := factorDegree * batchSize
    fmt.Printf("%d %.3f %.3f\n", t, d, float64(t)/d)
}

type Tag struct {
    Name  string `json:"name"`
    Value string `json:"value"`
}

type Datapoint struct {
    Timestamp int64   `json:"timestamp"`
    Value     float64 `json:"value"`
}

type Point struct {
    Namespace string    `json:"namespace"`
    ID        string    `json:"id"`
    Tags      []Tag     `json:"tags"`
    Datapoint Datapoint `json:"Datapoint"`
}

func bench() {
    client := &http.Client{
        Transport: &http.Transport{
            MaxIdleConns:    1000,
            MaxConnsPerHost: 1000,
            IdleConnTimeout: 90 * time.Second,
        },
    }

    ch := make(chan *Point, concurrency)
    var wg sync.WaitGroup
    for i := 0; i < concurrency; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for point := range ch {
                buf := new(bytes.Buffer)

                enc := json.NewEncoder(buf)
                enc.Encode(point)

                // write it here
                resp, err := client.Post("http://localhost:9003/writetagged", "application/json", buf)
                if err != nil {
                    fmt.Println(err)
                } else if resp.StatusCode != 200 {
                    fmt.Println(resp)
                }

                if resp != nil && resp.Body != nil {
                    io.CopyN(ioutil.Discard, resp.Body, 512)
                    resp.Body.Close()
                }
            }
        }()
    }

    count := 0
    for i := 0; i < factorDegree; i++ {
        for j := 0; j < batchSize; j++ {
            count++
            if count%1000000 == 0 {
                fmt.Println(count)
            }

            p := &Point{
                Namespace: "metrics",
                ID:        fmt.Sprintf("server.alpha.%d", rand.Intn(1000000)),
                Tags: []Tag{
                    {Name: "unit", Value: fmt.Sprintf("1m%d", rand.Intn(6))},
                    {Name: "entityType", Value: "system"},
                    {Name: "entityId", Value: "system"},
                    {Name: "serverId", Value: fmt.Sprintf("VNPM%d", rand.Uint32()%servers)},
                    {Name: "group", Value: "default"},
                },
                Datapoint: Datapoint{
                    Timestamp: time.Now().Unix(),
                    Value:     rand.Float64(),
                },
            }
            ch <- p
        }
    }
    close(ch)

    wg.Wait()
}