increase retention time and reduce cluster hits

internal/apicache/transport.go

@@ -20,23 +20,26 @@ type APICacheTransport struct {
 
 // RoundTrip wraps http.DefaultTransport.RoundTrip to provide stats and handle error rates.
 func (t *APICacheTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-
 	// Loop until success
 	tries := 0
 	for {
 		// Tickup retry counter
+		sleep := time.Second * time.Duration(tries) * 2
 		tries++
 
 		// Time our response
 		start := time.Now()
-
-		// Run the request.
-		res, httperr := t.Transport.RoundTrip(req)
-
 		metricAPICalls.With(
 			prometheus.Labels{"host": req.Host},
 		).Observe(float64(time.Since(start).Nanoseconds()) / float64(time.Millisecond))
 
+		// Run the request.
+		res, err := t.Transport.RoundTrip(req)
+		if err != nil {
+			time.Sleep(time.Second * 10) // Uncatchable errors
+			return res, err
+		}
+
 		// We got a response
 		if res != nil {
 			// Get the ESI error information
@@ -53,9 +56,7 @@ func (t *APICacheTransport) RoundTrip(req *http.Request) (*http.Response, error)
 			// Tick up and log any errors
 			if res.StatusCode >= 400 {
 				metricAPIErrors.Inc()
-
 				// Backoff
-				sleep := 60 * time.Second
 				if res.StatusCode == 401 { // Something went really wrong
 					sleep = 120 * time.Second
 				} else if esiRateLimiter { // Sleep based on error rate.
@@ -66,11 +67,10 @@ func (t *APICacheTransport) RoundTrip(req *http.Request) (*http.Response, error)
 				if sleep < time.Second {
 					sleep = time.Second + (time.Duration(rand.Float32()) * time.Second)
 				}
-				if sleep > time.Second*60 {
-					sleep = time.Second * 60
+				if sleep > time.Second*120 {
+					sleep = time.Second * 120
 				}
 
-				time.Sleep(sleep)
 				log.Printf("Try: %d Sleep: %d St: %d Res: %s Tok: %s - %s\n", tries, sleep/time.Second, res.StatusCode, resetS, tokensS, req.URL)
 
 				// Dump data for important errors // !esiRateLimiter &&
@@ -85,22 +85,20 @@ func (t *APICacheTransport) RoundTrip(req *http.Request) (*http.Response, error)
 						fmt.Printf("%s\n\n", dump)
 						log.Printf("Giving up %d %s\n", res.StatusCode, req.URL)
 					}
-					return res, httperr
+					time.Sleep(sleep)
+					return res, err
 				}
-
 			}
 
-			if tries > 10 && res.StatusCode >= 400 {
+			if tries > 10 {
 				log.Printf("Too many tries %d %s\n", res.StatusCode, req.URL)
 				return res, err
 			}
-
-		} else {
-			return res, httperr
 		}
 		if res.StatusCode >= 200 && res.StatusCode < 400 {
-			return res, httperr
+			return res, err
 		}
+		time.Sleep(sleep)
 	}
 }
 

kubernetes/infrastructure/monitoring.yaml

@@ -67,7 +67,7 @@ spec:
         - "/bin/prometheus"
         args:
         - "--config.file=/etc/prometheus/prometheus.yml"
-        - "--storage.tsdb.retention=1d"
+        - "--storage.tsdb.retention=90d"
         ports:
         - containerPort: 9090
           protocol: TCP

services/hammer/hammer.go

@@ -23,7 +23,7 @@ import (
 )
 
 // NUM_WORKERS number of concurrent workers
-const NUM_WORKERS = 20
+const NUM_WORKERS = 5
 
 // Hammer completes work handling CCP ESI and other API.
 type Hammer struct {

services/hammer/killmail.go

@@ -1,6 +1,7 @@
 package hammer
 
 import (
+	"context"
 	"log"
 
 	"github.com/antihax/evedata/internal/datapackages"
@@ -15,7 +16,18 @@ func killmailConsumer(s *Hammer, parameter interface{}) {
 	hash := parameters[0].(string)
 	id := int32(parameters[1].(int))
 
-	kill, _, err := s.esi.ESI.KillmailsApi.GetKillmailsKillmailIdKillmailHash(nil, hash, id, nil)
+	known := s.inQueue.CheckWorkCompleted("evedata_known_kills", id)
+	if known {
+		return
+	}
+
+	kill, _, err := s.esi.ESI.KillmailsApi.GetKillmailsKillmailIdKillmailHash(context.Background(), hash, id, nil)
+	if err != nil {
+		log.Println(err)
+		return
+	}
+
+	s.inQueue.SetWorkCompleted("evedata_known_kills", id)
 	if err != nil {
 		log.Println(err)
 		return
@@ -65,6 +77,5 @@ func killmailConsumer(s *Hammer, parameter interface{}) {
 			log.Println(err)
 			return
 		}
-
 	}
 }

services/hammer/war.go

@@ -20,8 +20,8 @@ func warConsumer(s *Hammer, parameter interface{}) {
 	}
 
 	// if the war ended, market it finished
-	if war.Finished.IsZero() == false && war.Finished.Before(time.Now().UTC()) {
-		err = s.inQueue.SetWorkCompleted("evedata_known_kills", int64(id))
+	if war.Finished.IsZero() && war.Finished.Before(time.Now().UTC()) {
+		err = s.inQueue.SetWorkCompleted("evedata_war_finished", int64(id))
 		if err != nil {
 			log.Println(err)
 		}