1package configsource
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "log/slog"
9 "sync"
10 "time"
11
12 "tug/internal/platform/config"
13 "tug/internal/platform/configsrc"
14 "tug/internal/storage"
15)
16
17type Store interface {
18 ConfigCommit(ctx context.Context) (storage.Checkpoint, error)
19 CompareAndSwapConfigCommit(ctx context.Context, expected storage.Checkpoint, next string) error
20 AppendAudit(ctx context.Context, input storage.AppendAuditInput) (storage.AuditEvent, error)
21}
22
23type Source interface {
24 Step(ctx context.Context, in configsrc.Input) (configsrc.Result, error)
25 Pin(head string) error
26 Read(commit string) ([]byte, error)
27}
28
29type Metrics interface {
30 Applied(commit string) error
31 Fetched(at time.Time)
32 RollbackRefused()
33}
34
35type Setters struct {
36 SessionTTLs func(ttl time.Duration, hostTTL map[string]time.Duration) error
37 AllowedHD func(hd string)
38 DefaultGroups func(ctx context.Context, defaults, referenced []string) error
39}
40
41type Config struct {
42 ConfigPath, SecretsPath string
43 Bootstrap *config.Config
44 Store Store
45 Source Source
46 Logger *slog.Logger
47 Now func() time.Time
48 Restart func(keys []string)
49}
50
51var (
52 ErrRefused = errors.New("configsource: the configuration was refused")
53 ErrUnavailable = errors.New("configsource: the checkpoint is unreadable and the remote unreachable")
54
55 errPinFailed = errors.New("configsource: the accepted head could not be pinned")
56)
57
58const (
59 sourceGit = "git"
60 startTimeout = 15 * time.Second
61 pollInterval = 5 * time.Minute
62 kindConfigChanged = "config-changed"
63 actionApplied = "config.applied"
64 actionRefused = "config.refused"
65 auditActor = "system"
66 auditSubject = "config"
67)
68
69type refusalKey struct {
70 kind, head string
71}
72
73type Controller struct {
74 cfg Config
75 boot *config.Config
76 git bool
77 logger *slog.Logger
78 now func() time.Time
79 tick func(time.Duration) (<-chan time.Time, func())
80
81 stepMu sync.Mutex
82 running *config.Config
83 audited map[refusalKey]bool
84 restarting bool
85
86 mu sync.Mutex
87 set Setters
88 metrics Metrics
89 bound bool
90 commit string
91 fetchedAt time.Time
92 rollbacks int
93}
94
95func New(cfg Config) (*Controller, error) {
96 if cfg.Bootstrap == nil {
97 return nil, errors.New("configsource: the bootstrap configuration is required")
98 }
99 if cfg.Restart == nil {
100 return nil, errors.New("configsource: the restart request is required")
101 }
102 git := cfg.Bootstrap.Source == sourceGit
103 if git && (cfg.Source == nil || cfg.Store == nil || cfg.Bootstrap.ConfigGit == nil) {
104 return nil, errors.New("configsource: the git source needs a source, a store and [config.git]")
105 }
106 logger := cfg.Logger
107 if logger == nil {
108 logger = slog.New(slog.DiscardHandler)
109 }
110 now := cfg.Now
111 if now == nil {
112 now = time.Now
113 }
114 return &Controller{cfg: cfg, boot: cfg.Bootstrap, git: git, logger: logger, now: now, tick: newTicker, audited: map[refusalKey]bool{}}, nil
115}
116
117func newTicker(d time.Duration) (<-chan time.Time, func()) {
118 t := time.NewTicker(d)
119 return t.C, t.Stop
120}
121
122func (c *Controller) Start(ctx context.Context) (*config.Config, error) {
123 c.stepMu.Lock()
124 defer c.stepMu.Unlock()
125 if !c.git {
126 c.running = c.boot
127 return c.boot, nil
128 }
129 k, err := c.cfg.Store.ConfigCommit(ctx)
130 if err != nil {
131 return nil, err
132 }
133 stepCtx, cancel := context.WithTimeout(ctx, startTimeout)
134 res, err := c.cfg.Source.Step(stepCtx, c.input(k, startTimeout))
135 cancel()
136 if err != nil {
137 return nil, err
138 }
139 c.noteFetched(res)
140 c.noteRepaired(res)
141 switch res.Outcome {
142 case configsrc.Accepted:
143 if cand, ok := c.candidate(ctx, res); ok {
144 err := c.advance(ctx, k, res.Head)
145 if err == nil {
146 c.recordApplied(ctx, res.Head, config.Changes{})
147 return c.started(cand, res.Head, origin(res.Head)), nil
148 }
149 if !nothingMoved(err) {
150 return nil, err
151 }
152 }
153 case configsrc.Refused:
154 c.refused(ctx, res.Kind, res.Head, res.Cause, nil)
155 }
156 if !k.Set {
157 return c.startFromDisk()
158 }
159 return c.startFromCheckpoint(k.Value, res)
160}
161
162func (c *Controller) startFromDisk() (*config.Config, error) {
163 secrets, err := config.CheckSecrets(c.boot, c.cfg.SecretsPath)
164 if err != nil {
165 return nil, fmt.Errorf("%w: %w", ErrRefused, err)
166 }
167 cfg := *c.boot
168 cfg.Secrets = secrets
169 return c.started(&cfg, "", c.cfg.ConfigPath), nil
170}
171
172func (c *Controller) startFromCheckpoint(k string, res configsrc.Result) (*config.Config, error) {
173 data, err := c.cfg.Source.Read(k)
174 if err != nil {
175 c.logger.Error("config checkpoint unreadable", "checkpoint", short(k), "kind", res.Kind, "err", err)
176 if res.Outcome == configsrc.Refused && res.Kind == configsrc.KindCheckpointMissing {
177 return nil, fmt.Errorf("%w: %s: %w", ErrRefused, configsrc.KindCheckpointMissing, err)
178 }
179 return nil, fmt.Errorf("%w: %w", ErrUnavailable, err)
180 }
181 cfg, err := config.LoadCandidate(origin(k), data, c.cfg.SecretsPath)
182 if err != nil {
183 return nil, fmt.Errorf("%w: %w", ErrRefused, err)
184 }
185 if err := c.cfg.Source.Pin(k); err != nil {
186 c.logger.Warn("config pin failed", "head", short(k), "err", err)
187 }
188 return c.started(cfg, k, origin(k)), nil
189}
190
191func (c *Controller) started(cfg *config.Config, commit, from string) *config.Config {
192 c.running = cfg
193 c.setCommit(commit)
194 c.logger.Info("config running", "source", sourceGit, "commit", commit, "from", from)
195 return cfg
196}
197
198func (c *Controller) input(k storage.Checkpoint, timeout time.Duration) configsrc.Input {
199 g := c.boot.ConfigGit
200 return configsrc.Input{
201 Checkpoint: k.Value,
202 Remote: configsrc.Remote{
203 URL: g.Remote,
204 Branch: g.Branch,
205 DeployKey: c.boot.Secrets.ConfigDeployKey.Bytes(),
206 KnownHosts: g.KnownHostsData,
207 },
208 Signers: g.Signers,
209 Timeout: timeout,
210 }
211}
212
213func (c *Controller) candidate(ctx context.Context, res configsrc.Result) (*config.Config, bool) {
214 cand, err := config.LoadCandidate(origin(res.Head), res.Config, c.cfg.SecretsPath)
215 if err != nil {
216 c.refused(ctx, configsrc.KindInvalid, res.Head, nil, err)
217 return nil, false
218 }
219 if !cand.SameSource(c.boot) {
220 c.refused(ctx, kindConfigChanged, res.Head, nil, nil)
221 return nil, false
222 }
223 return cand, true
224}
225
226func (c *Controller) advance(ctx context.Context, k storage.Checkpoint, head string) error {
227 if err := c.cfg.Source.Pin(head); err != nil {
228 c.logger.Warn("config pin failed", "head", short(head), "err", err)
229 return fmt.Errorf("%w: %w", errPinFailed, err)
230 }
231 if err := c.cfg.Store.CompareAndSwapConfigCommit(ctx, k, head); err != nil {
232 if errors.Is(err, storage.ErrConflict) {
233 c.logger.Warn("config checkpoint conflict", "remote", c.boot.ConfigGit.Remote, "head", short(head))
234 }
235 return err
236 }
237 return nil
238}
239
240func nothingMoved(err error) bool {
241 return errors.Is(err, storage.ErrConflict) || errors.Is(err, errPinFailed)
242}
243
244func (c *Controller) refused(ctx context.Context, kind, head string, cause, problems error) {
245 remote := c.boot.ConfigGit.Remote
246 attrs := []any{"kind", kind, "remote", remote, "head", short(head)}
247 if kind == configsrc.KindUnreachable {
248 attrs = append(attrs, "timeout", configsrc.IsTimeout(cause))
249 }
250 if lines := problemLines(problems); lines != nil {
251 attrs = append(attrs, "problems", lines)
252 }
253 c.logger.Warn("config refused", attrs...)
254 if kind == configsrc.KindUnreachable {
255 return
256 }
257 key := refusalKey{kind, head}
258 if c.audited[key] {
259 return
260 }
261 c.audited[key] = true
262 if kind == configsrc.KindRewound || kind == configsrc.KindDiverged {
263 c.noteRollback()
264 }
265 c.audit(ctx, actionRefused, map[string]any{"kind": kind, "remote": remote, "head": short(head)})
266}
267
268func (c *Controller) recordApplied(ctx context.Context, commit string, ch config.Changes) {
269 hot, restart := list(ch.Hot), list(ch.Restart)
270 if !c.git {
271 c.logger.Info("config applied", "hot", hot, "restart", restart)
272 return
273 }
274 c.logger.Info("config applied", "commit", commit, "hot", hot, "restart", restart)
275 c.audit(ctx, actionApplied, map[string]any{"commit": commit, "hot": hot, "restart": restart})
276}
277
278func (c *Controller) audit(ctx context.Context, action string, detail map[string]any) {
279 data, err := json.Marshal(detail)
280 if err == nil {
281 _, err = c.cfg.Store.AppendAudit(ctx, storage.AppendAuditInput{Actor: auditActor, Action: action, Subject: auditSubject, Detail: string(data)})
282 }
283 if err != nil {
284 c.logger.Error("config audit failed", "action", action, "err", err)
285 }
286}
287
288func (c *Controller) Bind(set Setters, m Metrics) {
289 c.mu.Lock()
290 defer c.mu.Unlock()
291 c.set, c.metrics, c.bound = set, m, true
292 if m == nil {
293 return
294 }
295 if !c.fetchedAt.IsZero() {
296 m.Fetched(c.fetchedAt)
297 }
298 for ; c.rollbacks > 0; c.rollbacks-- {
299 m.RollbackRefused()
300 }
301 if c.commit != "" {
302 c.applied(c.commit)
303 }
304}
305
306func (c *Controller) setters() Setters {
307 c.mu.Lock()
308 defer c.mu.Unlock()
309 return c.set
310}
311
312func (c *Controller) setCommit(commit string) {
313 c.mu.Lock()
314 defer c.mu.Unlock()
315 c.commit = commit
316 if c.bound && commit != "" {
317 c.applied(commit)
318 }
319}
320
321func (c *Controller) applied(commit string) {
322 if c.metrics == nil {
323 return
324 }
325 if err := c.metrics.Applied(commit); err != nil {
326 c.logger.Warn("config metric refused the commit", "commit", commit, "err", err)
327 }
328}
329
330func (c *Controller) noteRepaired(res configsrc.Result) {
331 if res.Repaired {
332 c.logger.Warn("config clone repaired", "remote", c.boot.ConfigGit.Remote, "head", short(res.Head))
333 }
334}
335
336func (c *Controller) noteFetched(res configsrc.Result) {
337 if !res.Fetched {
338 return
339 }
340 at := c.now()
341 c.mu.Lock()
342 defer c.mu.Unlock()
343 if !c.bound {
344 c.fetchedAt = at
345 return
346 }
347 if c.metrics != nil {
348 c.metrics.Fetched(at)
349 }
350}
351
352func (c *Controller) noteRollback() {
353 c.mu.Lock()
354 defer c.mu.Unlock()
355 if !c.bound {
356 c.rollbacks++
357 return
358 }
359 if c.metrics != nil {
360 c.metrics.RollbackRefused()
361 }
362}
363
364func origin(commit string) string {
365 return "git:" + short(commit) + ":tug.toml"
366}
367
368func short(commit string) string {
369 if len(commit) > 12 {
370 return commit[:12]
371 }
372 return commit
373}
374
375func list(s []string) []string {
376 if s == nil {
377 return []string{}
378 }
379 return s
380}
381
382func problemLines(err error) []string {
383 var cfgErr *config.Error
384 if !errors.As(err, &cfgErr) {
385 return nil
386 }
387 lines := make([]string, len(cfgErr.Problems))
388 for i, p := range cfgErr.Problems {
389 lines[i] = p.String()
390 }
391 return lines
392}