| 1 | package relay |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | "testing" |
| 9 | "time" |
| 10 | ) |
| 11 | |
| 12 | func TestMeterFlushAppendsDeltas(t *testing.T) { |
| 13 | s := newMemStore(t) |
| 14 | now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC) |
| 15 | m := NewMeter("box1", func() time.Time { return now }, nil) |
| 16 | id := Identity{"abc", "meet"} |
| 17 | m.Add(id, 100, 200, 5) |
| 18 | if err := m.Flush(context.Background(), s); err != nil { |
| 19 | t.Fatal(err) |
| 20 | } |
| 21 | m.Add(id, 10, 20, 0) |
| 22 | if err := m.Flush(context.Background(), s); err != nil { |
| 23 | t.Fatal(err) |
| 24 | } |
| 25 | body, _, _ := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12")) |
| 26 | var c MeterChunk |
| 27 | json.Unmarshal(body, &c) |
| 28 | if len(c.Samples) != 2 || c.Samples[0].Apps["abc/meet"].In != 100 || c.Samples[1].Apps["abc/meet"].Out != 20 { |
| 29 | t.Fatalf("chunk = %+v", c) |
| 30 | } |
| 31 | } |
| 32 | |
| 33 | func TestLoadWindowSpansTwoHours(t *testing.T) { |
| 34 | s := newMemStore(t) |
| 35 | now := time.Date(2026, 9, 7, 12, 1, 0, 0, time.UTC) |
| 36 | m := NewMeter("box1", func() time.Time { return now.Add(-30 * time.Minute) }, nil) // 11:31 |
| 37 | m.Add(Identity{"abc", "meet"}, 7, 0, 0) |
| 38 | m.Flush(context.Background(), s) |
| 39 | samples, err := LoadWindow(context.Background(), s, "box1", now, nil) |
| 40 | if err != nil || len(samples) != 1 || samples[0].Apps["abc/meet"].In != 7 { |
| 41 | t.Fatalf("window = %+v %v", samples, err) |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | // TestMeterFlushPreservesCorruptChunk checks that a Flush landing on a |
| 46 | // chunk that fails to unmarshal doesn't just clobber it: the unreadable |
| 47 | // bytes are preserved as a ".corrupt-<unix>" sibling, a fresh chunk is |
| 48 | // written in the real key's place, and the corruption is logged. |
| 49 | func TestMeterFlushPreservesCorruptChunk(t *testing.T) { |
| 50 | s := newMemStore(t) |
| 51 | now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC) |
| 52 | key := MeterKey("box1", "2026-09-07-12") |
| 53 | if _, err := s.Put(context.Background(), key, []byte("not json"), ""); err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | |
| 57 | var logs []string |
| 58 | m := NewMeter("box1", func() time.Time { return now }, func(format string, args ...any) { |
| 59 | logs = append(logs, fmt.Sprintf(format, args...)) |
| 60 | }) |
| 61 | m.Add(Identity{"abc", "meet"}, 1, 2, 0) |
| 62 | if err := m.Flush(context.Background(), s); err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | |
| 66 | body, _, err := s.Get(context.Background(), key) |
| 67 | if err != nil { |
| 68 | t.Fatal(err) |
| 69 | } |
| 70 | var c MeterChunk |
| 71 | if err := json.Unmarshal(body, &c); err != nil { |
| 72 | t.Fatalf("chunk at %s still unreadable: %v", key, err) |
| 73 | } |
| 74 | if len(c.Samples) != 1 { |
| 75 | t.Fatalf("chunk = %+v", c) |
| 76 | } |
| 77 | |
| 78 | keys, err := s.List(context.Background(), "metrics/relay/box1/") |
| 79 | if err != nil { |
| 80 | t.Fatal(err) |
| 81 | } |
| 82 | found := false |
| 83 | for _, k := range keys { |
| 84 | if strings.Contains(k, ".corrupt-") { |
| 85 | found = true |
| 86 | } |
| 87 | } |
| 88 | if !found { |
| 89 | t.Fatalf("no .corrupt- sibling among %v", keys) |
| 90 | } |
| 91 | if len(logs) == 0 { |
| 92 | t.Fatal("expected a log message about the corrupt chunk") |
| 93 | } |
| 94 | } |
| 95 | |
| 96 | // TestMeterEvictsDeadApps checks that an app with nothing left to report |
| 97 | // (zero deltas, zero allocations) drops out of the meter's live set — and |
| 98 | // out of the sample — after being flushed once, while an app still holding |
| 99 | // open allocations keeps being reported even with zero deltas. |
| 100 | func TestMeterEvictsDeadApps(t *testing.T) { |
| 101 | s := newMemStore(t) |
| 102 | now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC) |
| 103 | m := NewMeter("box1", func() time.Time { return now }, nil) |
| 104 | a := Identity{"abc", "meet"} |
| 105 | b := Identity{"abc", "docs"} |
| 106 | m.Add(a, 1, 2, 0) |
| 107 | m.Add(b, 3, 4, 0) |
| 108 | m.SetAllocs(a, 1) |
| 109 | m.SetAllocs(b, 0) |
| 110 | |
| 111 | if err := m.Flush(context.Background(), s); err != nil { |
| 112 | t.Fatal(err) |
| 113 | } |
| 114 | if err := m.Flush(context.Background(), s); err != nil { |
| 115 | t.Fatal(err) |
| 116 | } |
| 117 | |
| 118 | body, _, err := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12")) |
| 119 | if err != nil { |
| 120 | t.Fatal(err) |
| 121 | } |
| 122 | var c MeterChunk |
| 123 | if err := json.Unmarshal(body, &c); err != nil { |
| 124 | t.Fatal(err) |
| 125 | } |
| 126 | if len(c.Samples) != 2 { |
| 127 | t.Fatalf("chunk = %+v", c) |
| 128 | } |
| 129 | |
| 130 | second := c.Samples[1] |
| 131 | appA, ok := second.Apps["abc/meet"] |
| 132 | if !ok || appA.Allocs != 1 || appA.In != 0 || appA.Out != 0 || appA.Rejected != 0 { |
| 133 | t.Fatalf("second sample abc/meet = %+v ok=%v", appA, ok) |
| 134 | } |
| 135 | if _, ok := second.Apps["abc/docs"]; ok { |
| 136 | t.Fatalf("second sample still has dead app abc/docs: %+v", second.Apps) |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | // TestMeterFlushOfAnIdleRelayWritesNothing: a sample naming no app carries |
| 141 | // no information, and a relay box is idle almost all of the time — left |
| 142 | // unguarded, the flush loop writes 1,440 empty rows a day per box, each one |
| 143 | // a read, a write, a CAS and a row every reader of the hour has to walk. An |
| 144 | // hour in which nothing happened is best represented by no object at all. |
| 145 | func TestMeterFlushOfAnIdleRelayWritesNothing(t *testing.T) { |
| 146 | s := newMemStore(t) |
| 147 | now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC) |
| 148 | m := NewMeter("box1", func() time.Time { return now }, nil) |
| 149 | for i := 0; i < 2; i++ { |
| 150 | if err := m.Flush(context.Background(), s); err != nil { |
| 151 | t.Fatalf("flush %d: %v", i, err) |
| 152 | } |
| 153 | } |
| 154 | keys, err := s.List(context.Background(), "metrics/relay/box1/") |
| 155 | if err != nil { |
| 156 | t.Fatal(err) |
| 157 | } |
| 158 | if len(keys) != 0 { |
| 159 | t.Fatalf("an idle relay wrote %v, want nothing", keys) |
| 160 | } |
| 161 | // And the first byte that IS forwarded still lands, so this is a skip, |
| 162 | // not a mute. |
| 163 | m.Add(Identity{"abc", "meet"}, 1, 0, 0) |
| 164 | if err := m.Flush(context.Background(), s); err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | if keys, _ := s.List(context.Background(), "metrics/relay/box1/"); len(keys) != 1 { |
| 168 | t.Fatalf("after traffic, keys = %v, want the hour's one chunk", keys) |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | // TestSetAllocsAllZeroesDepartedApps is the leak test. SetAllocs alone can |
| 173 | // only ever raise a gauge, because the ids that need lowering are precisely |
| 174 | // the ones a census of live allocations no longer names — so an app whose |
| 175 | // last call ended would report a phantom allocation in every sample from |
| 176 | // then on and would never be evicted, growing m.cur without bound in a |
| 177 | // process the unit caps at 256 MiB. |
| 178 | func TestSetAllocsAllZeroesDepartedApps(t *testing.T) { |
| 179 | s := newMemStore(t) |
| 180 | now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC) |
| 181 | m := NewMeter("box1", func() time.Time { return now }, nil) |
| 182 | a := Identity{"abc", "meet"} |
| 183 | b := Identity{"abc", "docs"} |
| 184 | |
| 185 | // Both live and forwarding. |
| 186 | m.Add(a, 10, 20, 0) |
| 187 | m.Add(b, 30, 40, 0) |
| 188 | m.SetAllocsAll(map[Identity]int{a: 1, b: 2}) |
| 189 | if err := m.Flush(context.Background(), s); err != nil { |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | // b's call ends part way through the next minute: it forwarded some |
| 193 | // bytes and then went, so the census no longer names it. |
| 194 | m.Add(b, 5, 5, 0) |
| 195 | m.SetAllocsAll(map[Identity]int{a: 1}) |
| 196 | if err := m.Flush(context.Background(), s); err != nil { |
| 197 | t.Fatal(err) |
| 198 | } |
| 199 | // And a third flush finds nothing left to say about b at all. |
| 200 | m.SetAllocsAll(map[Identity]int{a: 1}) |
| 201 | if err := m.Flush(context.Background(), s); err != nil { |
| 202 | t.Fatal(err) |
| 203 | } |
| 204 | |
| 205 | body, _, err := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12")) |
| 206 | if err != nil { |
| 207 | t.Fatal(err) |
| 208 | } |
| 209 | var c MeterChunk |
| 210 | if err := json.Unmarshal(body, &c); err != nil { |
| 211 | t.Fatal(err) |
| 212 | } |
| 213 | if len(c.Samples) != 3 { |
| 214 | t.Fatalf("want 3 samples, got %+v", c.Samples) |
| 215 | } |
| 216 | if got := c.Samples[0].Apps["abc/docs"].Allocs; got != 2 { |
| 217 | t.Errorf("first sample abc/docs allocs = %d, want 2", got) |
| 218 | } |
| 219 | // The gauge came back DOWN. Without the census this would still read 2, |
| 220 | // which is the whole finding. |
| 221 | if got, ok := c.Samples[1].Apps["abc/docs"]; !ok || got.Allocs != 0 || got.In != 5 { |
| 222 | t.Errorf("second sample abc/docs = %+v ok=%v, want its last 5 bytes with allocs 0", got, ok) |
| 223 | } |
| 224 | if _, ok := c.Samples[2].Apps["abc/docs"]; ok { |
| 225 | t.Errorf("third sample still carries the departed app: %+v", c.Samples[2].Apps) |
| 226 | } |
| 227 | if got := c.Samples[2].Apps["abc/meet"].Allocs; got != 1 { |
| 228 | t.Errorf("third sample abc/meet allocs = %d, want the live app kept at 1", got) |
| 229 | } |
| 230 | // And it is out of the meter's live set, not merely out of the sample: |
| 231 | // the gauge that never fell was a leak as much as a lie. |
| 232 | m.mu.Lock() |
| 233 | n := len(m.cur) |
| 234 | m.mu.Unlock() |
| 235 | if n != 1 { |
| 236 | t.Errorf("meter still carries %d apps, want only the live one", n) |
| 237 | } |
| 238 | } |
| 239 | |
amadan