package main import ( "archive/zip" "bytes" "flag" "fmt" "io" "io/ioutil" "log" "math/rand" "net/http" "os" "os/exec" "path" "strconv" "strings" "sync/atomic" "time" "github.com/gomodule/redigo/redis" "github.com/tidwall/gjson" "github.com/tidwall/tile38/internal/server" ) const tile38Port = 9191 const httpPort = 9292 const dir = "data" var tile38Addr string var httpAddr string var wd string var minX float64 var minY float64 var maxX float64 var maxY float64 var pool = &redis.Pool{ MaxIdle: 3, IdleTimeout: 240 * time.Second, Dial: func() (redis.Conn, error) { return redis.Dial("tcp", tile38Addr) }, } var providedTile38 bool var providedHTTP bool const blank = false const hookServer = true var logf *os.File func main() { flag.StringVar(&tile38Addr, "tile38", "", "Tile38 address, leave blank to start a new server") flag.StringVar(&httpAddr, "hook", "", "Hook HTTP url, leave blank to start a new server") flag.Parse() log.Println("mockfill-107 (Github #107: Memory leak)") if tile38Addr == "" { tile38Addr = "127.0.0.1:" + strconv.FormatInt(int64(tile38Port), 10) } else { providedTile38 = true } if httpAddr == "" { httpAddr = "http://127.0.0.1:" + strconv.FormatInt(int64(httpPort), 10) + "/hook" } else { providedHTTP = true } var err error wd, err = os.Getwd() if err != nil { log.Fatal(err) } logf, err = os.Create("log") if err != nil { log.Fatal(err) } defer logf.Close() if !providedTile38 { copyAOF() go startTile38Server() } if !providedHTTP { if hookServer { go startHookServer() } } go waitForServers(func() { log.Printf("servers ready") logServer("START") setPoints() logServer("DONE") }) select {} return } func startTile38Server() { log.Println("start tile38 server") err := server.Serve("localhost", tile38Port, "data", false, "") if err != nil { log.Fatal(err) } } func startHookServer() { log.Println("start hook server") http.HandleFunc("/ping", func(w http.ResponseWriter, _ *http.Request) { io.WriteString(w, "pong") }) http.HandleFunc("/hook", func(w http.ResponseWriter, req *http.Request) { data, err := ioutil.ReadAll(req.Body) if err != nil { log.Fatal(err) } log.Println(string(data)) }) err := http.ListenAndServe(fmt.Sprintf("127.0.0.1:%d", httpPort), nil) if err != nil { log.Fatal(err) } } func waitForServers(cb func()) { log.Println("wait for servers") var err error start := time.Now() for { if time.Since(start) > time.Second*5 { log.Fatal("connection failed:", err) } func() { conn := pool.Get() defer conn.Close() var s string s, err = redis.String(conn.Do("PING")) if err != nil { return } if s != "PONG" { log.Fatalf("expected '%v', got '%v'", "PONG", s) } }() if err == nil { break } time.Sleep(time.Second / 5) } if hookServer { start = time.Now() for { if time.Since(start) > time.Second*5 { log.Fatal("connection failed:", err) } func() { var resp *http.Response resp, err = http.Get(httpAddr + "/notreal") if err != nil { return } defer resp.Body.Close() if resp.StatusCode != 200 && resp.StatusCode != 404 { log.Fatalf("expected '%v', got '%v'", "200 or 404", resp.StatusCode) } }() if err == nil { break } time.Sleep(time.Second / 5) } } cb() } func downloadAOF() { log.Println("downloading aof") resp, err := http.Get("https://github.com/tidwall/tile38/files/675225/appendonly.aof.zip") if err != nil { log.Fatal(err) } defer resp.Body.Close() body, err := ioutil.ReadAll(resp.Body) if err != nil { log.Fatal(err) } rd, err := zip.NewReader(bytes.NewReader(body), int64(len(body))) if err != nil { log.Fatal(err) } for _, f := range rd.File { if path.Ext(f.Name) == ".aof" { rc, err := f.Open() if err != nil { log.Fatal(err) } defer rc.Close() data, err := ioutil.ReadAll(rc) if err != nil { log.Fatal(err) } err = ioutil.WriteFile(path.Join(wd, "appendonly.aof"), data, 0666) if err != nil { log.Fatal(err) } return } } log.Fatal("invalid appendonly.aof.zip") } func copyAOF() { if err := os.RemoveAll(path.Join(wd, "data")); err != nil { log.Fatal(err) } if err := os.MkdirAll(path.Join(wd, "data"), 0777); err != nil { log.Fatal(err) } fin, err := os.Open(path.Join(wd, "appendonly.aof")) if err != nil { if os.IsNotExist(err) { downloadAOF() fin, err = os.Open(path.Join(wd, "appendonly.aof")) if err != nil { log.Fatal(err) } } else { log.Fatal(err) } } defer fin.Close() log.Println("load aof") fout, err := os.Create(path.Join(wd, "data", "appendonly.aof")) if err != nil { log.Fatal(err) } defer fout.Close() data, err := ioutil.ReadAll(fin) if err != nil { log.Fatal(err) } rep := httpAddr rep = "$" + strconv.FormatInt(int64(len(rep)), 10) + "\r\n" + rep + "\r\n" data = bytes.Replace(data, []byte("$23\r\nhttp://172.17.0.1:9999/\r\n"), []byte(rep), -1) if blank { data = nil } if _, err := fout.Write(data); err != nil { log.Fatal(err) } } func respGet(resp interface{}, idx ...int) interface{} { for i := 0; i < len(idx); i++ { arr, _ := redis.Values(resp, nil) resp = arr[idx[i]] } return resp } type PSAUX struct { User string PID int CPU float64 Mem float64 VSZ int RSS int TTY string Stat string Start string Time string Command string } func atoi(s string) int { n, _ := strconv.ParseInt(s, 10, 64) return int(n) } func atof(s string) float64 { n, _ := strconv.ParseFloat(s, 64) return float64(n) } func psaux(pid int) PSAUX { var res []byte res, err := exec.Command("ps", "ux", "-p", strconv.FormatInt(int64(pid), 10)).CombinedOutput() if err != nil { return PSAUX{} } pids := strconv.FormatInt(int64(pid), 10) for _, line := range strings.Split(string(res), "\n") { var words []string for _, word := range strings.Split(line, " ") { if word != "" { words = append(words, word) } } if len(words) >= 11 { if words[1] == pids { return PSAUX{ User: words[0], PID: atoi(words[1]), CPU: atof(words[2]), Mem: atof(words[3]), VSZ: atoi(words[4]), RSS: atoi(words[5]), TTY: words[6], Stat: words[7], Start: words[8], Time: words[9], Command: words[10], } } } } return PSAUX{} } func respGetFloat(resp interface{}, idx ...int) float64 { resp = respGet(resp, idx...) f, _ := redis.Float64(resp, nil) return f } func logServer(tag string) { conn := pool.Get() defer conn.Close() _, err := conn.Do("OUTPUT", "json") if err != nil { log.Fatal(err) } _, err = redis.String(conn.Do("GC")) if err != nil { log.Fatal(err) } json, err := redis.String(conn.Do("SERVER")) if err != nil { log.Fatal(err) } _, err = conn.Do("OUTPUT", "resp") if err != nil { log.Fatal(err) } rss := float64(psaux(int(gjson.Get(json, "stats.pid").Int())).RSS) / 1024 heapSize := gjson.Get(json, "stats.heap_size").Float() / 1024 / 1024 heapReleased := gjson.Get(json, "stats.heap_released").Float() / 1024 / 1024 fmt.Fprintf(logf, "%s %10.2f MB (heap) %10.2f MB (released) %10.2f MB (system)\n", time.Now().Format("2006-01-02T15:04:05Z07:00"), heapSize, heapReleased, rss) } func setPoints() { go func() { var i int for range time.NewTicker(time.Second * 1).C { logServer(fmt.Sprintf("SECOND-%d", i*1)) i++ } }() rand.Seed(time.Now().UnixNano()) n := 1000000 ex := time.Second * 10 log.Printf("time to pump data (%d points, expires %s)", n, ex) conn := pool.Get() defer conn.Close() if blank { minX = -124.40959167480469 minY = 32.53415298461914 maxX = -114.13121032714844 maxY = 42.009521484375 } else { resp, err := conn.Do("bounds", "boundies") if err != nil { log.Fatal(err) } minX = respGetFloat(resp, 0, 0) minY = respGetFloat(resp, 0, 1) maxX = respGetFloat(resp, 1, 0) maxY = respGetFloat(resp, 1, 1) } log.Printf("bbox: [[%.4f,%.4f],[%.4f,%.4f]]\n", minX, minY, maxX, maxY) var idx uint64 for i := 0; i < 4; i++ { go func() { conn := pool.Get() defer conn.Close() for i := 0; i < n; i++ { atomic.AddUint64(&idx, 1) id := fmt.Sprintf("person:%d", idx) x := rand.Float64()*(maxX-minX) + minX y := rand.Float64()*(maxY-minY) + minY ok, err := redis.String(conn.Do("SET", "people", id, "EX", float64(ex/time.Second), "POINT", y, x)) if err != nil { log.Fatal(err) } if ok != "OK" { log.Fatalf("expected 'OK', got '%v", ok) } log.Printf("SET people %v EX %v POINT %v %v", id, float64(ex/time.Second), y, x) } }() } select {} }