Merge pull request #17 from voyagegroup/mimic_team_c

Mimic team c

main.go

@@ -1,19 +1,72 @@
 package main
 
 import (
+	"database/sql"
 	"fmt"
 	"log"
 	"net/http"
+	"strings"
+	"time"
 
 	"os"
 
-	"github.com/go-sql-driver/mysql"
+    "github.com/go-sql-driver/mysql"
+	"github.com/carlescere/scheduler"
 	_ "github.com/go-sql-driver/mysql"
-	sqltrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/database/sql"
 	httptrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/net/http"
+    sqltrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/database/sql"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
 )
 
+type EventLog struct {
+	At    time.Time
+	Name  string
+	Value string
+}
+
+func insertChunk(valueStrings []string, valueArgs [](interface{}), db *sql.DB) {
+	stmt := fmt.Sprintf("INSERT INTO eventlog(at, name, value) VALUES %s", strings.Join(valueStrings, ","))
+	_, e := db.Exec(stmt, valueArgs...)
+	if e != nil {
+		panic(e.Error())
+	}
+}
+
+func insert(resc chan EventLog, db *sql.DB) {
+	const chunkSize = 1000
+
+	valueStrings := []string{}
+	valueArgs := [](interface{}){}
+
+LOOP:
+	for {
+		select {
+		case eventLog, ok := <-resc:
+			if ok {
+				valueStrings = append(valueStrings, "(?, ?, ?)")
+				valueArgs = append(valueArgs, fmt.Sprintf("%s", eventLog.At))
+				valueArgs = append(valueArgs, eventLog.Name)
+				valueArgs = append(valueArgs, eventLog.Value)
+				if len(valueStrings) >= chunkSize {
+					insertChunk(valueStrings, valueArgs, db)
+					valueStrings = nil
+					valueArgs = nil
+				}
+			} else {
+				panic("resc is closed!!!")
+			}
+		default:
+			break LOOP
+		}
+	}
+
+	if len(valueStrings) == 0 {
+		return
+	}
+
+	insertChunk(valueStrings, valueArgs, db)
+}
+
 func main() {
 	tracer.Start(tracer.WithServiceName("test-go"))
 	defer tracer.Stop()
@@ -23,28 +76,44 @@ func main() {
 		dataSourceName = "root:password@tcp(127.0.0.1:13306)/hakaru"
 	}
 
-	maxConnections := 66
-	numInstance := 15
-
-	sqltrace.Register("mysql", &mysql.MySQLDriver{}, sqltrace.WithServiceName("my-db"))
-	db, err := sqltrace.Open("mysql", dataSourceName)
+    sqltrace.Register("mysql", &mysql.MySQLDriver{}, sqltrace.WithServiceName("my-db"))
+    db, err := sqltrace.Open("mysql", dataSourceName)
 	if err != nil {
 		panic(err.Error())
 	}
 	defer db.Close()
+
+	maxConnections := 66
+	numInstance := 15
 	db.SetMaxOpenConns(maxConnections / numInstance)
 
-	stmt, e := db.Prepare("INSERT INTO eventlog(at, name, value) values(NOW(), ?, ?)")
+	resc := make(chan EventLog, 200000)
+
+	_, e := scheduler.Every(10).Seconds().NotImmediately().Run(func() {
+		insert(resc, db)
+	})
+
+	if e != nil {
+		panic(err.Error())
+	}
+
+	jst, e := time.LoadLocation("Asia/Tokyo")
+
 	if e != nil {
 		panic(e.Error())
 	}
-	defer stmt.Close()
 
 	hakaruHandler := func(w http.ResponseWriter, r *http.Request) {
 		name := r.URL.Query().Get("name")
 		value := r.URL.Query().Get("value")
 
-		_, _ = stmt.Exec(name, value)
+		now := time.Now().In(jst)
+
+		resc <- EventLog{
+			At:    now,
+			Name:  name,
+			Value: value,
+		}
 
 		origin := r.Header.Get("Origin")
 		if origin != "" {
@@ -59,12 +128,7 @@ func main() {
 
 	mux := httptrace.NewServeMux()
 	mux.HandleFunc("/hakaru", hakaruHandler)
-	// http.HandleFunc("/hakaru", hakaruHandler)
-	// http.HandleFunc("/ok", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(200) })
-	mux.HandleFunc("/ok", func(w http.ResponseWriter, r *http.Request) {
-		w.WriteHeader(200)
-		fmt.Fprintln(w, "v1")
-	})
+	mux.HandleFunc("/ok", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(200) })
 
 	// start server
 	if err := http.ListenAndServe(":8081", mux); err != nil {