package main import ( "bytes" "encoding/json" "fmt" "log" "net/http" "os" "queues-go/app/routers" client "queues-go/app/storage/mongo" tasks "queues-go/app/storage/mongo/collections" "strconv" "sync" "time" ) func writeMetric(timestamp time.Time, endpoint string, statusCode int, responseTime int, method string) { loc, _ := time.LoadLocation("Europe/Moscow") s := fmt.Sprintf( `{"timestamp":"%s","service":"queues","environment":"%s","endpoint":"%s","status_code":%s,"response_time":%s,"method":"%s"}`, timestamp.In(loc).Format("2006-01-02T15:04:05Z"), os.Getenv("STAGE"), endpoint, strconv.Itoa(statusCode), strconv.Itoa(responseTime), method, ) data := []byte(s) r := bytes.NewReader(data) _, err := http.Post("http://monitoring:1237/api/v1/metrics/endpoint", "application/json", r) if err != nil { log.Printf("Error sending metrics %s", err.Error()) } } func handlerWrapper(f func(*http.Request) (interface{}, int)) func(http.ResponseWriter, *http.Request) { return func(w http.ResponseWriter, r *http.Request) { start := time.Now() response, status := f(r) if response != nil { data, err := json.Marshal(response) if err != nil { log.Printf("Error marshalling response") status = http.StatusInternalServerError } w.WriteHeader(status) w.Write(data) } else { w.WriteHeader(status) } go writeMetric(start, r.URL.Path, status, int(time.Since(start).Milliseconds()), r.Method) log.Printf("%s %d %s", r.URL, status, time.Since(start)) } } func metricProxy(w http.ResponseWriter, r *http.Request) { http.Post("http://monitoring:1237/api/v1/metrics/task", "application/json", r.Body) w.WriteHeader(202) } func writeCount() { for { count, err := tasks.Count() if err != nil { log.Printf("Failed getting docs count: %s", err.Error()) } else { loc, _ := time.LoadLocation("Europe/Moscow") s := fmt.Sprintf( `{"timestamp":"%s","service":"queues","environment":"%s","name":"tasks","count":%s}`, time.Now().In(loc).Format("2006-01-02T15:04:05Z"), os.Getenv("STAGE"), strconv.Itoa(int(*count)), ) data := []byte(s) r := bytes.NewReader(data) _, err := http.Post("http://monitoring:1237/api/v1/metrics/increment", "application/json", r) if err != nil { log.Printf("ERROR %s", err.Error()) } } time.Sleep(time.Second) } } func main() { client.Connect() routers.MutexMap = make(map[string]*sync.Mutex) http.HandleFunc("/api/v1/take", handlerWrapper(routers.Take)) http.HandleFunc("/api/v1/finish", handlerWrapper(routers.Finish)) http.HandleFunc("/api/v1/put", handlerWrapper(routers.Put)) http.HandleFunc("/api/v1/metric", metricProxy) log.Printf("Server started") go writeCount() http.ListenAndServe(":1239", nil) }