mirror of https://github.com/chubin/wttr.in
Igor Chubin
5 years ago
parent
a35b1ea350
commit
e34dbc0f8b
@ -0,0 +1,79 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/robfig/cron"
|
||||
)
|
||||
|
||||
var peakRequest30 sync.Map
|
||||
var peakRequest60 sync.Map
|
||||
|
||||
func initPeakHandling() {
|
||||
c := cron.New()
|
||||
// cronTime := fmt.Sprintf("%d,%d * * * *", 30-prefetchInterval/60, 60-prefetchInterval/60)
|
||||
c.AddFunc("24 * * * *", prefetchPeakRequests30)
|
||||
c.AddFunc("54 * * * *", prefetchPeakRequests60)
|
||||
c.Start()
|
||||
}
|
||||
|
||||
func savePeakRequest(cacheDigest string, r *http.Request) {
|
||||
_, min, _ := time.Now().Clock()
|
||||
if min == 30 {
|
||||
peakRequest30.Store(cacheDigest, *r)
|
||||
} else if min == 0 {
|
||||
peakRequest60.Store(cacheDigest, *r)
|
||||
}
|
||||
}
|
||||
|
||||
func prefetchRequest(r *http.Request) {
|
||||
processRequest(r)
|
||||
}
|
||||
|
||||
func syncMapLen(sm *sync.Map) int {
|
||||
count := 0
|
||||
|
||||
f := func(key, value interface{}) bool {
|
||||
|
||||
// Not really certain about this part, don't know for sure
|
||||
// if this is a good check for an entry's existence
|
||||
if key == "" {
|
||||
return false
|
||||
}
|
||||
count++
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
sm.Range(f)
|
||||
|
||||
return count
|
||||
}
|
||||
|
||||
func prefetchPeakRequests(peakRequestMap *sync.Map) {
|
||||
peakRequestLen := syncMapLen(peakRequestMap)
|
||||
log.Printf("PREFETCH: Prefetching %d requests\n", peakRequestLen)
|
||||
if peakRequestLen == 0 {
|
||||
return
|
||||
}
|
||||
sleepBetweenRequests := time.Duration(prefetchInterval*1000/peakRequestLen) * time.Millisecond
|
||||
peakRequestMap.Range(func(key interface{}, value interface{}) bool {
|
||||
r := value.(http.Request)
|
||||
log.Printf("Prefetching %s\n", key)
|
||||
prefetchRequest(&r)
|
||||
peakRequestMap.Delete(key)
|
||||
time.Sleep(sleepBetweenRequests)
|
||||
return true
|
||||
})
|
||||
}
|
||||
|
||||
func prefetchPeakRequests30() {
|
||||
prefetchPeakRequests(&peakRequest30)
|
||||
}
|
||||
|
||||
func prefetchPeakRequests60() {
|
||||
prefetchPeakRequests(&peakRequest60)
|
||||
}
|
||||
@ -0,0 +1,130 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
"math/rand"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
func processRequest(r *http.Request) responseWithHeader {
|
||||
var response responseWithHeader
|
||||
|
||||
foundInCache := false
|
||||
cacheDigest := getCacheDigest(r)
|
||||
|
||||
savePeakRequest(cacheDigest, r)
|
||||
|
||||
cacheBody, ok := lruCache.Get(cacheDigest)
|
||||
if ok {
|
||||
cacheEntry := cacheBody.(responseWithHeader)
|
||||
|
||||
// if after all attempts we still have no answer,
|
||||
// we try to make the query on our own
|
||||
for attempts := 0; attempts < 300; attempts++ {
|
||||
if !ok || !cacheEntry.InProgress {
|
||||
break
|
||||
}
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
cacheBody, ok = lruCache.Get(cacheDigest)
|
||||
cacheEntry = cacheBody.(responseWithHeader)
|
||||
}
|
||||
if cacheEntry.InProgress {
|
||||
log.Printf("TIMEOUT: %s\n", cacheDigest)
|
||||
}
|
||||
if ok && !cacheEntry.InProgress && cacheEntry.Expires.After(time.Now()) {
|
||||
response = cacheEntry
|
||||
foundInCache = true
|
||||
}
|
||||
}
|
||||
|
||||
if !foundInCache {
|
||||
lruCache.Add(cacheDigest, responseWithHeader{InProgress: true})
|
||||
response = get(r)
|
||||
if response.StatusCode == 200 || response.StatusCode == 304 {
|
||||
lruCache.Add(cacheDigest, response)
|
||||
} else {
|
||||
log.Printf("REMOVE: %d response for %s from cache\n", response.StatusCode, cacheDigest)
|
||||
lruCache.Remove(cacheDigest)
|
||||
}
|
||||
}
|
||||
return response
|
||||
}
|
||||
|
||||
func get(req *http.Request) responseWithHeader {
|
||||
|
||||
client := &http.Client{}
|
||||
|
||||
queryURL := fmt.Sprintf("http://%s%s", req.Host, req.RequestURI)
|
||||
|
||||
proxyReq, err := http.NewRequest(req.Method, queryURL, req.Body)
|
||||
if err != nil {
|
||||
log.Printf("Request: %s\n", err)
|
||||
}
|
||||
|
||||
// proxyReq.Header.Set("Host", req.Host)
|
||||
// proxyReq.Header.Set("X-Forwarded-For", req.RemoteAddr)
|
||||
|
||||
for header, values := range req.Header {
|
||||
for _, value := range values {
|
||||
proxyReq.Header.Add(header, value)
|
||||
}
|
||||
}
|
||||
|
||||
res, err := client.Do(proxyReq)
|
||||
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
body, err := ioutil.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
log.Println(err)
|
||||
}
|
||||
|
||||
return responseWithHeader{
|
||||
InProgress: false,
|
||||
Expires: time.Now().Add(time.Duration(randInt(1000, 1500)) * time.Second),
|
||||
Body: body,
|
||||
Header: res.Header,
|
||||
StatusCode: res.StatusCode,
|
||||
}
|
||||
}
|
||||
|
||||
// implementation of the cache.get_signature of original wttr.in
|
||||
func getCacheDigest(req *http.Request) string {
|
||||
|
||||
userAgent := req.Header.Get("User-Agent")
|
||||
|
||||
queryHost := req.Host
|
||||
queryString := req.RequestURI
|
||||
|
||||
clientIPAddress := readUserIP(req)
|
||||
|
||||
lang := req.Header.Get("Accept-Language")
|
||||
|
||||
return fmt.Sprintf("%s:%s%s:%s:%s", userAgent, queryHost, queryString, clientIPAddress, lang)
|
||||
}
|
||||
|
||||
func readUserIP(r *http.Request) string {
|
||||
IPAddress := r.Header.Get("X-Real-Ip")
|
||||
if IPAddress == "" {
|
||||
IPAddress = r.Header.Get("X-Forwarded-For")
|
||||
}
|
||||
if IPAddress == "" {
|
||||
IPAddress = r.RemoteAddr
|
||||
var err error
|
||||
IPAddress, _, err = net.SplitHostPort(IPAddress)
|
||||
if err != nil {
|
||||
log.Printf("ERROR: userip: %q is not IP:port\n", IPAddress)
|
||||
}
|
||||
}
|
||||
return IPAddress
|
||||
}
|
||||
|
||||
func randInt(min int, max int) int {
|
||||
return min + rand.Intn(max-min)
|
||||
}
|
||||
@ -0,0 +1,40 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type safeCounter struct {
|
||||
v map[int]int
|
||||
mux sync.Mutex
|
||||
}
|
||||
|
||||
func (c *safeCounter) inc(key int) {
|
||||
c.mux.Lock()
|
||||
c.v[key]++
|
||||
c.mux.Unlock()
|
||||
}
|
||||
|
||||
// func (c *safeCounter) val(key int) int {
|
||||
// c.mux.Lock()
|
||||
// defer c.mux.Unlock()
|
||||
// return c.v[key]
|
||||
// }
|
||||
//
|
||||
// func (c *safeCounter) reset(key int) int {
|
||||
// c.mux.Lock()
|
||||
// defer c.mux.Unlock()
|
||||
// result := c.v[key]
|
||||
// c.v[key] = 0
|
||||
// return result
|
||||
// }
|
||||
|
||||
var queriesPerMinute safeCounter
|
||||
|
||||
func printStat() {
|
||||
_, min, _ := time.Now().Clock()
|
||||
queriesPerMinute.inc(min)
|
||||
log.Printf("Processed %d requests\n", min)
|
||||
}
|
||||
Loading…
Reference in new issue