go project - add a migration and rename the dockerfiles so that Go is the primary
This commit is contained in:
@@ -0,0 +1,94 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ErrOutOfSequence is returned when a heartbeat timestamp is earlier than
|
||||
// the latest accepted heartbeat for that device.
|
||||
var ErrOutOfSequence = errors.New("heartbeat out of sequence")
|
||||
|
||||
// heartbeatHistory tracks one device's heartbeat timeline.
|
||||
// Unexported: only Store methods expose behavior.
|
||||
type heartbeatHistory struct {
|
||||
mu sync.Mutex
|
||||
count int64
|
||||
first time.Time
|
||||
last time.Time
|
||||
}
|
||||
|
||||
func (h *heartbeatHistory) add(sentAt time.Time) error {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
if h.first.IsZero() {
|
||||
h.first = sentAt
|
||||
h.last = sentAt
|
||||
h.count = 1
|
||||
return nil
|
||||
}
|
||||
|
||||
// Equal timestamps are allowed (matches C#: sentAt < LastHeartBeat only).
|
||||
if sentAt.Before(h.last) {
|
||||
return ErrOutOfSequence
|
||||
}
|
||||
|
||||
h.last = sentAt
|
||||
h.count++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *heartbeatHistory) uptime() float64 {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
if h.first.IsZero() {
|
||||
return 0
|
||||
}
|
||||
|
||||
minutes := h.last.Sub(h.first).Minutes()
|
||||
// Single heartbeat (or identical timestamps): treat span as one minute.
|
||||
if minutes == 0 {
|
||||
minutes = 1
|
||||
}
|
||||
|
||||
return float64(h.count) / minutes * 100
|
||||
}
|
||||
|
||||
// historyFor returns the per-device history, creating it if needed.
|
||||
// The store-level lock only covers map insert/lookup; callers then use
|
||||
// the history's own lock for mutations/reads.
|
||||
func (s *Store) historyFor(id DeviceID) *heartbeatHistory {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
h, ok := s.heartbeats[id]
|
||||
if !ok {
|
||||
h = &heartbeatHistory{}
|
||||
s.heartbeats[id] = h
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// AddHeartbeat records a heartbeat for deviceID.
|
||||
// Returns ErrOutOfSequence if sentAt is earlier than the last accepted beat.
|
||||
func (s *Store) AddHeartbeat(deviceID DeviceID, sentAt time.Time) error {
|
||||
return s.historyFor(deviceID).add(sentAt)
|
||||
}
|
||||
|
||||
// Uptime returns uptime percent for deviceID:
|
||||
//
|
||||
// (heartbeatCount / minutesBetweenFirstAndLast) * 100
|
||||
//
|
||||
// Unknown / never-seen devices return 0. A zero-length span is treated as 1 minute.
|
||||
func (s *Store) Uptime(deviceID DeviceID) float64 {
|
||||
s.mu.Lock()
|
||||
h, ok := s.heartbeats[deviceID]
|
||||
s.mu.Unlock()
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
return h.uptime()
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestAddHeartbeat_FirstSetsFirstLastAndCount(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
t0 := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
if err := s.AddHeartbeat(id, t0); err != nil {
|
||||
t.Fatalf("AddHeartbeat: %v", err)
|
||||
}
|
||||
if got := s.Uptime(id); got != 100 {
|
||||
t.Fatalf("uptime = %v, want 100 (1 beat / 1 minute)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddHeartbeat_LaterUpdatesLast(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
t0 := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
|
||||
t1 := t0.Add(time.Minute)
|
||||
|
||||
_ = s.AddHeartbeat(id, t0)
|
||||
if err := s.AddHeartbeat(id, t1); err != nil {
|
||||
t.Fatalf("AddHeartbeat: %v", err)
|
||||
}
|
||||
// (2 beats / 1 minute) * 100
|
||||
if got := s.Uptime(id); got != 200 {
|
||||
t.Fatalf("uptime = %v, want 200", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddHeartbeat_SameTimestampAccepted(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
t0 := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
_ = s.AddHeartbeat(id, t0)
|
||||
if err := s.AddHeartbeat(id, t0); err != nil {
|
||||
t.Fatalf("same timestamp should be accepted: %v", err)
|
||||
}
|
||||
// span still zero → 1 minute floor; 2 beats → 200
|
||||
if got := s.Uptime(id); got != 200 {
|
||||
t.Fatalf("uptime = %v, want 200", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddHeartbeat_OutOfSequence(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
t0 := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
|
||||
t1 := t0.Add(time.Minute)
|
||||
|
||||
_ = s.AddHeartbeat(id, t1)
|
||||
err := s.AddHeartbeat(id, t0)
|
||||
if !errors.Is(err, ErrOutOfSequence) {
|
||||
t.Fatalf("err = %v, want ErrOutOfSequence", err)
|
||||
}
|
||||
// rejected beat must not affect uptime: still 1 beat → 100
|
||||
if got := s.Uptime(id); got != 100 {
|
||||
t.Fatalf("uptime = %v, want 100", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUptime_TwoHeartbeatsOverTwoMinutes(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
t0 := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
_ = s.AddHeartbeat(id, t0)
|
||||
_ = s.AddHeartbeat(id, t0.Add(2*time.Minute))
|
||||
|
||||
if got := s.Uptime(id); got != 100 {
|
||||
t.Fatalf("uptime = %v, want 100", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUptime_UnknownDevice(t *testing.T) {
|
||||
s := New(nil)
|
||||
if got := s.Uptime(DeviceID("missing")); got != 0 {
|
||||
t.Fatalf("uptime = %v, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUptime_IsPerDevice(t *testing.T) {
|
||||
s := New(nil)
|
||||
a := DeviceID("device-1")
|
||||
b := DeviceID("device-2")
|
||||
t0 := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
_ = s.AddHeartbeat(a, t0)
|
||||
_ = s.AddHeartbeat(a, t0.Add(time.Minute))
|
||||
_ = s.AddHeartbeat(b, t0)
|
||||
|
||||
if got := s.Uptime(a); got != 200 {
|
||||
t.Fatalf("device-1 uptime = %v, want 200", got)
|
||||
}
|
||||
if got := s.Uptime(b); got != 100 {
|
||||
t.Fatalf("device-2 uptime = %v, want 100 (single beat = 1 minute)", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
package store
|
||||
|
||||
import "sync"
|
||||
|
||||
// DeviceID is a typed device identifier.
|
||||
// Using a named type (instead of plain string) makes call sites clearer
|
||||
// and prevents accidentally mixing unrelated string values.
|
||||
type DeviceID string
|
||||
|
||||
// Store holds in-memory fleet metrics for known devices.
|
||||
type Store struct {
|
||||
known map[DeviceID]struct{}
|
||||
|
||||
// mu protects the heartbeats/uploads maps (insert/lookup of entries only).
|
||||
// Each per-device history has its own mutex for mutations.
|
||||
mu sync.Mutex
|
||||
heartbeats map[DeviceID]*heartbeatHistory
|
||||
uploads map[DeviceID]*uploadHistory
|
||||
}
|
||||
|
||||
// New creates a Store seeded with the given known device IDs.
|
||||
func New(known []DeviceID) *Store {
|
||||
m := make(map[DeviceID]struct{}, len(known))
|
||||
for _, id := range known {
|
||||
m[id] = struct{}{}
|
||||
}
|
||||
return &Store{
|
||||
known: m,
|
||||
heartbeats: make(map[DeviceID]*heartbeatHistory),
|
||||
uploads: make(map[DeviceID]*uploadHistory),
|
||||
}
|
||||
}
|
||||
|
||||
// Known reports whether deviceID was present in the startup device list.
|
||||
func (s *Store) Known(deviceID DeviceID) bool {
|
||||
_, ok := s.known[deviceID]
|
||||
return ok
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// uploadHistory tracks running total/count of upload durations for one device.
|
||||
type uploadHistory struct {
|
||||
mu sync.Mutex
|
||||
total time.Duration
|
||||
count int64
|
||||
}
|
||||
|
||||
func (u *uploadHistory) add(d time.Duration) {
|
||||
u.mu.Lock()
|
||||
defer u.mu.Unlock()
|
||||
u.total += d
|
||||
u.count++
|
||||
}
|
||||
|
||||
func (u *uploadHistory) average() time.Duration {
|
||||
u.mu.Lock()
|
||||
defer u.mu.Unlock()
|
||||
if u.count == 0 {
|
||||
return 0
|
||||
}
|
||||
// Truncating integer division — same idea as C# BigInteger total / count.
|
||||
return u.total / time.Duration(u.count)
|
||||
}
|
||||
|
||||
func (s *Store) uploadFor(id DeviceID) *uploadHistory {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
u, ok := s.uploads[id]
|
||||
if !ok {
|
||||
u = &uploadHistory{}
|
||||
s.uploads[id] = u
|
||||
}
|
||||
return u
|
||||
}
|
||||
|
||||
// AddUpload records an upload duration (nanoseconds as time.Duration) for deviceID.
|
||||
// sent_at from the API is intentionally ignored here; the HTTP layer still accepts it.
|
||||
func (s *Store) AddUpload(deviceID DeviceID, uploadTime time.Duration) {
|
||||
s.uploadFor(deviceID).add(uploadTime)
|
||||
}
|
||||
|
||||
// AverageUploadTime returns the mean upload duration for deviceID.
|
||||
// Formatting for the API (duration.String()) belongs at the HTTP edge.
|
||||
// Unknown / never-seen devices return 0.
|
||||
func (s *Store) AverageUploadTime(deviceID DeviceID) time.Duration {
|
||||
s.mu.Lock()
|
||||
u, ok := s.uploads[deviceID]
|
||||
s.mu.Unlock()
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
return u.average()
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestAddUpload_AverageOfOne(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
d := 197331667813 * time.Nanosecond // ~3m17.331667813s
|
||||
|
||||
s.AddUpload(id, d)
|
||||
|
||||
if got := s.AverageUploadTime(id); got != d {
|
||||
t.Fatalf("avg = %v, want %v", got, d)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAddUpload_AverageOfMany(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
|
||||
s.AddUpload(id, 100*time.Nanosecond)
|
||||
s.AddUpload(id, 200*time.Nanosecond)
|
||||
s.AddUpload(id, 300*time.Nanosecond)
|
||||
|
||||
// (100+200+300)/3 = 200
|
||||
if got := s.AverageUploadTime(id); got != 200*time.Nanosecond {
|
||||
t.Fatalf("avg = %v, want 200ns", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAverageUploadTime_UnknownDevice(t *testing.T) {
|
||||
s := New(nil)
|
||||
if got := s.AverageUploadTime(DeviceID("missing")); got != 0 {
|
||||
t.Fatalf("avg = %v, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAverageUploadTime_IsPerDevice(t *testing.T) {
|
||||
s := New(nil)
|
||||
a := DeviceID("device-1")
|
||||
b := DeviceID("device-2")
|
||||
|
||||
s.AddUpload(a, 100*time.Nanosecond)
|
||||
s.AddUpload(a, 300*time.Nanosecond)
|
||||
s.AddUpload(b, 50*time.Nanosecond)
|
||||
|
||||
if got := s.AverageUploadTime(a); got != 200*time.Nanosecond {
|
||||
t.Fatalf("device-1 avg = %v, want 200ns", got)
|
||||
}
|
||||
if got := s.AverageUploadTime(b); got != 50*time.Nanosecond {
|
||||
t.Fatalf("device-2 avg = %v, want 50ns", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAverageUploadTime_TruncatingDivision(t *testing.T) {
|
||||
s := New(nil)
|
||||
id := DeviceID("device-1")
|
||||
|
||||
// 10+11 = 21; 21/2 = 10 in integer division
|
||||
s.AddUpload(id, 10*time.Nanosecond)
|
||||
s.AddUpload(id, 11*time.Nanosecond)
|
||||
|
||||
if got := s.AverageUploadTime(id); got != 10*time.Nanosecond {
|
||||
t.Fatalf("avg = %v, want 10ns (truncating)", got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user