| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "log" |
| 7 | "net/http" |
| 8 | "net/http/httptest" |
| 9 | "path/filepath" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/analytics" |
| 16 | "github.com/carlosframework/platform/internal/apperr" |
| 17 | "github.com/carlosframework/platform/internal/applog" |
| 18 | "github.com/carlosframework/platform/internal/edge" |
| 19 | "github.com/carlosframework/platform/internal/registry" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | "github.com/carlosframework/platform/internal/usage" |
| 22 | ) |
| 23 | |
| 24 | // keyLog records every key a shipper touches, so a test can see a key that |
| 25 | // a directory-backed store would have quietly normalised ("a//b" is "a/b" |
| 26 | // on disk). |
| 27 | type keyLog struct { |
| 28 | release.Store |
| 29 | mu sync.Mutex |
| 30 | touched []string |
| 31 | } |
| 32 | |
| 33 | func (k *keyLog) note(key string) { |
| 34 | k.mu.Lock() |
| 35 | k.touched = append(k.touched, key) |
| 36 | k.mu.Unlock() |
| 37 | } |
| 38 | |
| 39 | func (k *keyLog) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 40 | k.note("put " + key) |
| 41 | return k.Store.Put(ctx, key, body, ifMatch) |
| 42 | } |
| 43 | |
| 44 | func (k *keyLog) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 45 | k.note("get " + key) |
| 46 | return k.Store.Get(ctx, key) |
| 47 | } |
| 48 | |
| 49 | func (k *keyLog) List(ctx context.Context, prefix string) ([]string, error) { |
| 50 | k.note("list " + prefix) |
| 51 | return k.Store.List(ctx, prefix) |
| 52 | } |
| 53 | |
| 54 | func (k *keyLog) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| 55 | k.note("listdirs " + prefix) |
| 56 | return k.Store.ListDirs(ctx, prefix) |
| 57 | } |
| 58 | |
| 59 | func (k *keyLog) Delete(ctx context.Context, key string) error { |
| 60 | k.note("delete " + key) |
| 61 | return k.Store.Delete(ctx, key) |
| 62 | } |
| 63 | |
| 64 | func (k *keyLog) all() []string { |
| 65 | k.mu.Lock() |
| 66 | defer k.mu.Unlock() |
| 67 | return append([]string(nil), k.touched...) |
| 68 | } |
| 69 | |
| 70 | // captureShipLog swaps the standard logger's output for the test and forgets |
| 71 | // which apps unshippable has reported, so its once-only line is observable. |
| 72 | func captureShipLog(t *testing.T) *bytes.Buffer { |
| 73 | t.Helper() |
| 74 | unshippableSeen.Range(func(k, _ any) bool { unshippableSeen.Delete(k); return true }) |
| 75 | var buf bytes.Buffer |
| 76 | prev, flags := log.Writer(), log.Flags() |
| 77 | log.SetOutput(&buf) |
| 78 | t.Cleanup(func() { log.SetOutput(prev); log.SetFlags(flags) }) |
| 79 | return &buf |
| 80 | } |
| 81 | |
| 82 | // TestErrshipAppLessRouteWritesNothing is the dev-bucket regression end to |
| 83 | // end. The route is the shape found on leo-climbing: an instance row with |
| 84 | // an account, no app, and a TCP upstream nobody listens on. Every request |
| 85 | // is a refused 502 that the edge files under (bab, ""). The errors shipper |
| 86 | // must write nothing for it (it once wrote apps/bab//errors/...), prune |
| 87 | // nothing, keep shipping the app beside it, and say so once. |
| 88 | func TestErrshipAppLessRouteWritesNothing(t *testing.T) { |
| 89 | dead := httptest.NewServer(http.NotFoundHandler()) |
| 90 | deadAddr := dead.Listener.Addr().String() |
| 91 | dead.Close() // nothing listens here now: every dial is refused |
| 92 | |
| 93 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 94 | if err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | for _, rt := range []registry.Route{ |
| 98 | {Host: "proof.example.com", Kind: "instance", Account: "bab", App: "", Addr: deadAddr, Status: "ready", Channel: "stable"}, |
| 99 | {Host: "web.example.com", Kind: "service", Account: "bab", App: "web", Addr: deadAddr, Status: "ready", Version: "v1"}, |
| 100 | } { |
| 101 | if err := reg.Upsert(rt); err != nil { |
| 102 | t.Fatal(err) |
| 103 | } |
| 104 | } |
| 105 | e, err := edge.New(reg, nil) |
| 106 | if err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | sink := apperr.NewSink("leo-climbing", time.Now) |
| 110 | e.EnableErrors(sink) |
| 111 | srv := httptest.NewServer(e) |
| 112 | t.Cleanup(srv.Close) |
| 113 | get := func(host string) { |
| 114 | t.Helper() |
| 115 | req, _ := http.NewRequest("GET", srv.URL+"/", nil) |
| 116 | req.Host = host |
| 117 | resp, err := http.DefaultClient.Do(req) |
| 118 | if err != nil { |
| 119 | t.Fatal(err) |
| 120 | } |
| 121 | resp.Body.Close() |
| 122 | if resp.StatusCode != http.StatusBadGateway { |
| 123 | t.Fatalf("%s answered %d, want a refused 502", host, resp.StatusCode) |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | logs := captureShipLog(t) |
| 128 | ctx := context.Background() |
| 129 | s := &keyLog{Store: release.NewDirStore(t.TempDir())} |
| 130 | pending := map[applog.AppKey][]apperr.Record{} |
| 131 | for round := 0; round < 2; round++ { |
| 132 | get("proof.example.com") |
| 133 | get("web.example.com") |
| 134 | errshipGather(sink, pending) |
| 135 | if _, ok := pending[applog.AppKey{Acct: "bab", App: ""}]; !ok { |
| 136 | t.Fatalf("round %d: the edge did not file the app-less route under (bab, \"\"); pending = %v", round, pending) |
| 137 | } |
| 138 | errshipFlush(ctx, s, pending) |
| 139 | if len(pending) != 0 { |
| 140 | t.Fatalf("round %d: pending after flush = %v; a refused key must drop, not hold forever", round, pending) |
| 141 | } |
| 142 | } |
| 143 | for _, k := range errshipKnownApps(pending, sink) { |
| 144 | chunkPrune(ctx, s, applog.FamilyErrors, "leo-climbing", k, time.Now()) |
| 145 | } |
| 146 | |
| 147 | var webPuts int |
| 148 | for _, op := range s.all() { |
| 149 | if strings.Contains(op, "//") { |
| 150 | t.Fatalf("touched a key with an empty segment: %s", op) |
| 151 | } |
| 152 | if strings.HasPrefix(op, "put apps/bab/web/errors/leo-climbing/") { |
| 153 | webPuts++ |
| 154 | } |
| 155 | } |
| 156 | if webPuts != 2 { |
| 157 | t.Fatalf("web chunks written = %d, want 2 (one per round): %v", webPuts, s.all()) |
| 158 | } |
| 159 | if n := strings.Count(logs.String(), `errors: not shipping for account "bab" app ""`); n != 1 { |
| 160 | t.Fatalf("refusal logged %d times, want once:\n%s", n, logs.String()) |
| 161 | } |
| 162 | } |
| 163 | |
| 164 | // TestLogshipAndAnalyticsDropAppLessRecords: the logs and analytics |
| 165 | // shippers take the same refusal: drop, never re-queue, write nothing. |
| 166 | func TestLogshipAndAnalyticsDropAppLessRecords(t *testing.T) { |
| 167 | logs := captureShipLog(t) |
| 168 | ctx := context.Background() |
| 169 | s := &keyLog{Store: release.NewDirStore(t.TempDir())} |
| 170 | bad := applog.AppKey{Acct: "bab", App: ""} |
| 171 | |
| 172 | pending := map[applog.AppKey][]applog.Record{bad: {{TS: time.Now(), Box: "box1", Seq: 1, Msg: "x"}}} |
| 173 | logshipFlush(ctx, s, pending) |
| 174 | if len(pending) != 0 { |
| 175 | t.Fatalf("logship pending = %v", pending) |
| 176 | } |
| 177 | chunkPrune(ctx, s, applog.FamilyLogs, "box1", bad, time.Now()) |
| 178 | |
| 179 | now := time.Date(2026, 10, 16, 6, 46, 0, 0, time.UTC) |
| 180 | am := analytics.NewMeter(t.TempDir(), "box1", func() time.Time { return now }) |
| 181 | am.Observe(analytics.Obs{Acct: "bab", App: "", Host: "proof.example.com", Method: "GET", Status: 502, Path: "/", When: now}) |
| 182 | am.Observe(analytics.Obs{Acct: "bab", App: "web", Host: "web.example.com", Method: "GET", Status: 200, ContentType: "text/html", Path: "/", UA: "Mozilla/5.0 Safari", ClientIP: "1.2.3.4", Bytes: 10, When: now}) |
| 183 | none := func() map[string]usage.EdgeRow { return nil } |
| 184 | for i := 0; i < 2; i++ { // a second flush proves the bad delta was not re-queued |
| 185 | if err := usageFlush(ctx, s, s, "box1", none, am, now); err != nil { |
| 186 | t.Fatal(err) |
| 187 | } |
| 188 | } |
| 189 | |
| 190 | var webDay bool |
| 191 | for _, op := range s.all() { |
| 192 | if strings.Contains(op, "//") { |
| 193 | t.Fatalf("touched a key with an empty segment: %s", op) |
| 194 | } |
| 195 | if op == "put apps/bab/web/analytics/box1/days/2026-10-16.json" { |
| 196 | webDay = true |
| 197 | } |
| 198 | } |
| 199 | if !webDay { |
| 200 | t.Fatalf("the valid app's day file was not written: %v", s.all()) |
| 201 | } |
| 202 | out := logs.String() |
| 203 | if strings.Count(out, `logs: not shipping for account "bab" app ""`) != 1 || strings.Count(out, `analytics: not shipping for account "bab" app ""`) != 1 { |
| 204 | t.Fatalf("want one refusal line per family:\n%s", out) |
| 205 | } |
| 206 | if strings.Contains(out, "re-queued") { |
| 207 | t.Fatalf("a refused analytics delta was re-queued:\n%s", out) |
| 208 | } |
| 209 | } |
| 210 | |
amadan