| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log" |
| 7 | "net/http" |
| 8 | "os" |
| 9 | "strconv" |
| 10 | "strings" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/aws/aws-sdk-go-v2/aws" |
| 14 | "github.com/aws/aws-sdk-go-v2/service/cloudwatch" |
| 15 | "github.com/aws/aws-sdk-go-v2/service/cloudwatch/types" |
| 16 | "github.com/carlosframework/platform/internal/rootaws" |
| 17 | ) |
| 18 | |
| 19 | // memwatch (memory-pressure spec §2.4) publishes two CloudWatch metrics a |
| 20 | // minute so a box that is running out of memory pages someone BEFORE the |
| 21 | // kernel starts thrashing — the state the flagship sat in for fifteen |
| 22 | // minutes on 2026-08-28 with every AWS-side check green. The status |
| 23 | // snapshot already carries memory to the bucket every five minutes, but a |
| 24 | // bucket write alarms nobody; CloudWatch does. |
| 25 | // |
| 26 | // Carlos/Host MemAvailablePercent MemAvailable / MemTotal, /proc/meminfo |
| 27 | // Carlos/Host MemoryPressureSome60 some avg60 from /proc/pressure/memory |
| 28 | // |
| 29 | // Both keyed by InstanceId (IMDS). The region-host module's alarm reads the |
| 30 | // first; the second is the early warning — it climbs minutes before |
| 31 | // MemAvailable hits the floor. |
| 32 | // |
| 33 | // Per-box, not a housekeeper duty: every box reports itself. Armed only by |
| 34 | // CARLOS_HOST_METRICS=cloudwatch in host.env, so a Pi or a customer fleet |
| 35 | // box never tries to reach an API it has no credentials for. |
| 36 | |
| 37 | const ( |
| 38 | memwatchNamespace = "Carlos/Host" |
| 39 | memwatchPeriod = time.Minute |
| 40 | ) |
| 41 | |
| 42 | // metricPutter is the one CloudWatch call this file makes. |
| 43 | type metricPutter interface { |
| 44 | PutMetricData(context.Context, *cloudwatch.PutMetricDataInput, ...func(*cloudwatch.Options)) (*cloudwatch.PutMetricDataOutput, error) |
| 45 | } |
| 46 | |
| 47 | // putMetricFunc is the metricPutter over one bound method value. Storing the |
| 48 | // *cloudwatch.Client itself in the interface would make the linker keep every |
| 49 | // exported CloudWatch method once html/template is linked in (discussion |
| 50 | // #275); a method value keeps PutMetricData alone. |
| 51 | type putMetricFunc func(context.Context, *cloudwatch.PutMetricDataInput, ...func(*cloudwatch.Options)) (*cloudwatch.PutMetricDataOutput, error) |
| 52 | |
| 53 | func (f putMetricFunc) PutMetricData(ctx context.Context, in *cloudwatch.PutMetricDataInput, opts ...func(*cloudwatch.Options)) (*cloudwatch.PutMetricDataOutput, error) { |
| 54 | return f(ctx, in, opts...) |
| 55 | } |
| 56 | |
| 57 | type memwatch struct { |
| 58 | put metricPutter |
| 59 | instanceID string |
| 60 | mem func() (*int64, *int64) // MemTotal, MemAvailable in bytes |
| 61 | pressure func() (string, error) // /proc/pressure/memory content |
| 62 | logf func(string, ...any) |
| 63 | // lastPutErr is the text of the put failure last logged, "" while |
| 64 | // puts succeed. Failures log on CHANGE (and recovery logs once), not |
| 65 | // once a minute: a box rolled before its IAM grant lands would |
| 66 | // otherwise write 1,440 identical lines a day. |
| 67 | lastPutErr string |
| 68 | } |
| 69 | |
| 70 | // memwatchArmed is the host.env gate, split out so the three answers it |
| 71 | // can give are table-testable: armed, off (unset), or off with a value |
| 72 | // this duty doesn't understand — the last is the one an operator wants to |
| 73 | // see in the journal, because they typed it. |
| 74 | func memwatchArmed(getenv func(string) string) (armed bool, why string) { |
| 75 | switch v := strings.TrimSpace(getenv("CARLOS_HOST_METRICS")); v { |
| 76 | case "cloudwatch": |
| 77 | return true, "" |
| 78 | case "": |
| 79 | return false, "CARLOS_HOST_METRICS unset" |
| 80 | default: |
| 81 | return false, "CARLOS_HOST_METRICS=" + strconv.Quote(v) + ` (the only value is "cloudwatch")` |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | // memwatchLoop arms the duty when host.env asks for it. One log line either |
| 86 | // way; a box that can't resolve its instance id or build a client logs why |
| 87 | // and stays quiet rather than retrying — the operator set the env, so a |
| 88 | // silent duty is a bug they'd want to see at arm time, in the journal. |
| 89 | // Process-lifetime, like statusLoop: nothing stops the ticker. |
| 90 | func memwatchLoop() { |
| 91 | if armed, why := memwatchArmed(os.Getenv); !armed { |
| 92 | log.Printf("agent: memwatch disabled (%s)", why) |
| 93 | return |
| 94 | } |
| 95 | ctx := context.Background() |
| 96 | // Region comes from the default chain — AWS_REGION in host.env, else |
| 97 | // IMDS. A compatible-store deployment that points AWS_REGION at its |
| 98 | // own object store (internal/lease/endpoint.go's note) would aim the |
| 99 | // CloudWatch client at nonsense, which is one more reason this duty |
| 100 | // is opt-in per box rather than on wherever credentials resolve. |
| 101 | cfg, err := rootaws.LoadDefaultConfig(ctx) |
| 102 | if err != nil { |
| 103 | log.Printf("agent: memwatch disabled: load aws config: %v", err) |
| 104 | return |
| 105 | } |
| 106 | // The metric goes to the box's OWN region, not host.env's AWS_REGION: |
| 107 | // the flagship sets AWS_REGION=eu-west-1 fleet-wide (it names the |
| 108 | // deployment bucket's region), and on the first roll all four boxes |
| 109 | // dutifully published into Ireland — while the region-host module |
| 110 | // creates each box's alarm in the box's region, so three alarms would |
| 111 | // have watched an empty namespace forever. IMDS placement wins; "" (not |
| 112 | // on EC2) keeps the chain's answer. |
| 113 | if r := readPlacementRegion(); r != "" { |
| 114 | cfg.Region = r |
| 115 | } |
| 116 | id, err := imdsInstanceID() |
| 117 | if err != nil { |
| 118 | log.Printf("agent: memwatch disabled: instance id: %v", err) |
| 119 | return |
| 120 | } |
| 121 | w := memwatch{ |
| 122 | put: putMetricFunc(cloudwatch.NewFromConfig(cfg).PutMetricData), |
| 123 | instanceID: id, |
| 124 | mem: readMemBytes, |
| 125 | pressure: readPressureFile, |
| 126 | logf: log.Printf, |
| 127 | } |
| 128 | log.Printf("agent: memwatch armed (%s in %s, every %s, instance %s)", memwatchNamespace, cfg.Region, memwatchPeriod, id) |
| 129 | for now := range time.Tick(memwatchPeriod) { |
| 130 | w.tick(ctx, now) |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | // tick reads, builds, puts. A failed read skips the tick (a zero would look |
| 135 | // like a full box); a failed put is logged when it first happens or changes, |
| 136 | // and the next tick tries again. |
| 137 | func (w *memwatch) tick(ctx context.Context, now time.Time) { |
| 138 | total, avail := w.mem() |
| 139 | if total == nil || avail == nil || *total <= 0 { |
| 140 | w.logf("memwatch: /proc/meminfo unreadable this tick — skipped") |
| 141 | return |
| 142 | } |
| 143 | availPct := float64(*avail) / float64(*total) * 100 |
| 144 | psi, hasPSI := 0.0, false |
| 145 | if s, err := w.pressure(); err == nil { |
| 146 | psi, hasPSI = parsePressureSome60(s) |
| 147 | } |
| 148 | ctx, cancel := context.WithTimeout(ctx, 20*time.Second) |
| 149 | defer cancel() |
| 150 | _, err := w.put.PutMetricData(ctx, memwatchData(w.instanceID, availPct, psi, hasPSI, now)) |
| 151 | switch { |
| 152 | case err != nil && err.Error() != w.lastPutErr: |
| 153 | w.lastPutErr = err.Error() |
| 154 | w.logf("memwatch: put: %v (logged once until it changes)", err) |
| 155 | case err == nil && w.lastPutErr != "": |
| 156 | w.lastPutErr = "" |
| 157 | w.logf("memwatch: put recovered") |
| 158 | } |
| 159 | } |
| 160 | |
| 161 | // memwatchData is the pure half: the exact PutMetricData input for one |
| 162 | // tick. Standard resolution — a one-minute period at standard resolution |
| 163 | // is free-tier territory; high resolution is not and buys nothing for a |
| 164 | // three-minute alarm. |
| 165 | func memwatchData(instanceID string, availPct, psi float64, hasPSI bool, now time.Time) *cloudwatch.PutMetricDataInput { |
| 166 | dims := []types.Dimension{{Name: aws.String("InstanceId"), Value: aws.String(instanceID)}} |
| 167 | data := []types.MetricDatum{{ |
| 168 | MetricName: aws.String("MemAvailablePercent"), |
| 169 | Dimensions: dims, |
| 170 | Timestamp: aws.Time(now), |
| 171 | Unit: types.StandardUnitPercent, |
| 172 | Value: aws.Float64(availPct), |
| 173 | }} |
| 174 | if hasPSI { |
| 175 | data = append(data, types.MetricDatum{ |
| 176 | MetricName: aws.String("MemoryPressureSome60"), |
| 177 | Dimensions: dims, |
| 178 | Timestamp: aws.Time(now), |
| 179 | Unit: types.StandardUnitPercent, |
| 180 | Value: aws.Float64(psi), |
| 181 | }) |
| 182 | } |
| 183 | return &cloudwatch.PutMetricDataInput{Namespace: aws.String(memwatchNamespace), MetricData: data} |
| 184 | } |
| 185 | |
| 186 | // imdsInstanceID asks the metadata service which instance this is — the |
| 187 | // dimension the alarm keys on — through the same IMDSv2 token dance |
| 188 | // readPlacementRegion uses (imdsToken/imdsLeaf). The IMDS uid-gate |
| 189 | // (harden-host) lets the carlos user through by construction. |
| 190 | func imdsInstanceID() (string, error) { |
| 191 | client := &http.Client{Timeout: 2 * time.Second} |
| 192 | token := imdsToken(client) |
| 193 | if token == "" { |
| 194 | return "", errors.New("no IMDSv2 token (not on EC2, or IMDS unreachable)") |
| 195 | } |
| 196 | id, code := imdsLeaf(client, token, "/latest/meta-data/instance-id") |
| 197 | if code != http.StatusOK || id == "" { |
| 198 | return "", errors.New("instance-id leaf did not answer 200") |
| 199 | } |
| 200 | return id, nil |
| 201 | } |
| 202 | |
| 203 | // readPressureFile is the /proc/pressure/memory read; absent on a kernel |
| 204 | // without PSI (CONFIG_PSI) — the tick then publishes MemAvailable alone. |
| 205 | func readPressureFile() (string, error) { |
| 206 | b, err := os.ReadFile("/proc/pressure/memory") |
| 207 | return string(b), err |
| 208 | } |
| 209 | |
| 210 | // parsePressureSome60 pulls "some ... avg60=N" out of /proc/pressure/memory. |
| 211 | // Not-ok for any shape it doesn't recognise: a parse bug must not report a |
| 212 | // calm box. |
| 213 | func parsePressureSome60(s string) (float64, bool) { |
| 214 | for _, line := range strings.Split(s, "\n") { |
| 215 | fields := strings.Fields(line) |
| 216 | if len(fields) == 0 || fields[0] != "some" { |
| 217 | continue |
| 218 | } |
| 219 | for _, f := range fields[1:] { |
| 220 | if v, ok := strings.CutPrefix(f, "avg60="); ok { |
| 221 | n, err := strconv.ParseFloat(v, 64) |
| 222 | return n, err == nil |
| 223 | } |
| 224 | } |
| 225 | return 0, false |
| 226 | } |
| 227 | return 0, false |
| 228 | } |
| 229 | |
amadan