Add in-memory cache #5

Merged
adoralaura merged 6 commits from feature-in-memory-cache into main 2024-04-19 11:55:25 +02:00
8 changed files with 211 additions and 9 deletions

View file

@ -0,0 +1,9 @@
when:
- event: push
steps:
test:
image: golang:1.22-bullseye
commands:
- go mod download
- go test -v ./...

3
Makefile Normal file
View file

@ -0,0 +1,3 @@
test:
go test ./...

93
cache/cache.go vendored Normal file
View file

@ -0,0 +1,93 @@
/*
Copyright 2024 Adora Laura Kalb <adora@lila.network>
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cache
import (
"net/netip"
"sync"
"time"
)
var (
GlobalScoreCache *ScoreCache
cacheExpiredAfter = 5 * time.Minute
)
type ServerScore struct {
expiresAt time.Time
Score float64
}
type ScoreCache struct {
stop chan struct{}
mu sync.RWMutex
scores map[netip.Addr]ServerScore
}
type CacheMissError struct{}
func (m *CacheMissError) Error() string {
return "User is not in cache!"
}
func newCacheMissError() *CacheMissError {
return &CacheMissError{}
}
func NewScoreCache() *ScoreCache {
lc := &ScoreCache{
scores: make(map[netip.Addr]ServerScore),
stop: make(chan struct{}),
}
return lc
}
func (sc *ScoreCache) Add(score float64, ip netip.Addr, ts time.Time) {
ssc := ServerScore{Score: score, expiresAt: ts.Add(cacheExpiredAfter)}
sc.mu.Lock()
sc.scores[ip] = ssc
sc.mu.Unlock()
}
func (lc *ScoreCache) Get(ip netip.Addr) (ServerScore, error) {
now := time.Now()
lc.mu.RLock()
cachedScore, ok := lc.scores[ip]
if !ok {
lc.mu.RUnlock()
return ServerScore{}, newCacheMissError()
}
lc.mu.RUnlock()
if now.After(cachedScore.expiresAt) {
lc.delete(ip)
return ServerScore{}, newCacheMissError()
}
return cachedScore, nil
}
func (lc *ScoreCache) delete(ip netip.Addr) {
lc.mu.Lock()
delete(lc.scores, ip)
lc.mu.Unlock()
}

49
cache/cache_test.go vendored Normal file
View file

@ -0,0 +1,49 @@
/*
Copyright 2024 Adora Laura Kalb <adora@lila.network>
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cache
import (
"net/netip"
"testing"
"time"
)
func TestCacheGet(t *testing.T) {
const testFloat = 1.23456
cache := NewScoreCache()
cache.Add(testFloat, netip.MustParseAddr("1.2.3.4"), time.Now().Add(5*time.Minute))
score, _ := cache.Get(netip.MustParseAddr("1.2.3.4"))
if score.Score != testFloat {
t.Fatalf("cache.Get(\"1.2.3.4\") = %f, want %f", score.Score, testFloat)
}
}
func TestCacheMiss(t *testing.T) {
now := time.Now()
const testFloat = 1.23456
expiredTime := now.Add(-(time.Minute * 10))
cache := NewScoreCache()
cache.Add(testFloat, netip.MustParseAddr("1.2.3.4"), expiredTime)
_, err := cache.Get(netip.MustParseAddr("1.2.3.4"))
if err == nil {
t.Fatalf("cache.Get(\"1.2.3.4\") = got nil, want error")
}
}

View file

@ -27,6 +27,7 @@ import (
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/prometheus/client_golang/prometheus"
"golang.adora.codes/ntppool-exporter/cache"
"golang.adora.codes/ntppool-exporter/models"
)
@ -61,7 +62,24 @@ func (c Collector) Collect(ch chan<- prometheus.Metric) {
}
func (c Collector) collect(ch chan<- prometheus.Metric, logger log.Logger) {
httpError := false
score, err := cache.GlobalScoreCache.Get(c.target)
if err == nil {
level.Debug(logger).Log("msg", "Serving score from cache",
"server", c.target.String(), "score", score.Score)
serverScoreMetric := prometheus.NewDesc("ntppool_server_score",
"Shows the server score currently assigned at ntppool.org",
nil, nil)
m1 := prometheus.MustNewConstMetric(serverScoreMetric, prometheus.GaugeValue, score.Score)
m1 = prometheus.NewMetricWithTimestamp(time.Now(), m1)
ch <- m1
return
}
level.Debug(logger).Log("msg", "Cache miss, querying API",
"server", c.target.String())
var serverApiScore float64 = 0
const apiEndpoint = "https://www.ntppool.org/scores/"
const apiQuery = "/json?limit=1&monitor=24"
@ -73,7 +91,7 @@ func (c Collector) collect(ch chan<- prometheus.Metric, logger log.Logger) {
if err != nil {
level.Error(logger).Log("msg", "Error sending HTTP request", "url", url, "message", err)
httpError = true
return
}
res, err := client.Do(req)
if err != nil {
@ -97,9 +115,11 @@ func (c Collector) collect(ch chan<- prometheus.Metric, logger log.Logger) {
return
}
if !httpError {
serverApiScore = response.History[0].Score
}
cache.GlobalScoreCache.Add(serverApiScore, c.target, time.Now())
level.Debug(logger).Log("msg", "Added score to cache",
"server", c.target.String(), "score", score.Score)
// TODO: Test or delete
serverScoreMetric := prometheus.NewDesc("ntppool_server_score",

25
helpers/helpers.go Normal file
View file

@ -0,0 +1,25 @@
/*
Copyright 2024 Adora Laura Kalb <adora@lila.network>
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package helpers
import (
"time"
)
func UnixToTime(i int64) time.Time {
return time.Unix(i, 0)
}

View file

@ -33,6 +33,7 @@ import (
"github.com/prometheus/common/version"
"github.com/prometheus/exporter-toolkit/web"
webflag "github.com/prometheus/exporter-toolkit/web/kingpinflag"
"golang.adora.codes/ntppool-exporter/cache"
"golang.adora.codes/ntppool-exporter/collector"
)
@ -95,6 +96,8 @@ func main() {
kingpin.Parse()
logger := promlog.New(promlogConfig)
cache.GlobalScoreCache = cache.NewScoreCache()
level.Info(logger).Log("msg", "Starting ntppool_exporter", "version", version.Info())
level.Info(logger).Log("build_context", version.BuildContext())

View file

@ -34,7 +34,7 @@ type ApiResponse struct {
}
type ApiResponseHistory struct {
Timestamp int `json:"ts"`
TimestampInt int64 `json:"ts"`
Step int `json:"step"`
Score float64 `json:"score"`
MonitorID int `json:"monitor_id"`