Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commitc09c9b9

Browse files
committed
WIP: agent reinitialization
1 parente562e3c commitc09c9b9

File tree

22 files changed

+990
-426
lines changed

22 files changed

+990
-426
lines changed

‎agent/agent.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,9 @@ import (
3636
"tailscale.com/util/clientmetric"
3737

3838
"cdr.dev/slog"
39+
40+
"github.com/coder/retry"
41+
3942
"github.com/coder/clistat"
4043
"github.com/coder/coder/v2/agent/agentcontainers"
4144
"github.com/coder/coder/v2/agent/agentexec"
@@ -53,7 +56,6 @@ import (
5356
"github.com/coder/coder/v2/tailnet"
5457
tailnetproto"github.com/coder/coder/v2/tailnet/proto"
5558
"github.com/coder/quartz"
56-
"github.com/coder/retry"
5759
)
5860

5961
const (
@@ -365,9 +367,11 @@ func (a *agent) runLoop() {
365367
ifctx.Err()!=nil {
366368
// Context canceled errors may come from websocket pings, so we
367369
// don't want to use `errors.Is(err, context.Canceled)` here.
370+
a.logger.Warn(ctx,"runLoop exited with error",slog.Error(ctx.Err()))
368371
return
369372
}
370373
ifa.isClosed() {
374+
a.logger.Warn(ctx,"runLoop exited because agent is closed")
371375
return
372376
}
373377
iferrors.Is(err,io.EOF) {
@@ -1048,7 +1052,12 @@ func (a *agent) run() (retErr error) {
10481052
returna.statsReporter.reportLoop(ctx,aAPI)
10491053
})
10501054

1051-
returnconnMan.wait()
1055+
err=connMan.wait()
1056+
// TODO: this broke some tests at some point. investigate.
1057+
iferr!=nil {
1058+
a.logger.Warn(context.Background(),"connection manager errored",slog.Error(err))
1059+
}
1060+
returnerr
10521061
}
10531062

10541063
// handleManifest returns a function that fetches and processes the manifest

‎agent/metrics.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ type agentMetrics struct {
2020
// took to run. This is reported once per agent.
2121
startupScriptSeconds*prometheus.GaugeVec
2222
currentConnections*prometheus.GaugeVec
23+
manifestsReceived prometheus.Counter
2324
}
2425

2526
funcnewAgentMetrics(registerer prometheus.Registerer)*agentMetrics {
@@ -54,11 +55,20 @@ func newAgentMetrics(registerer prometheus.Registerer) *agentMetrics {
5455
}, []string{"connection_type"})
5556
registerer.MustRegister(currentConnections)
5657

58+
manifestsReceived:=prometheus.NewCounter(prometheus.CounterOpts{
59+
Namespace:"coderd",
60+
Subsystem:"agentstats",
61+
Name:"manifests_received",
62+
Help:"The number of manifests this agent has received from the control plane.",
63+
})
64+
registerer.MustRegister(manifestsReceived)
65+
5766
return&agentMetrics{
5867
connectionsTotal:connectionsTotal,
5968
reconnectingPTYErrors:reconnectingPTYErrors,
6069
startupScriptSeconds:startupScriptSeconds,
6170
currentConnections:currentConnections,
71+
manifestsReceived:manifestsReceived,
6272
}
6373
}
6474

‎agent/reaper/reaper_unix.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
package reaper
44

55
import (
6+
"fmt"
67
"os"
78
"os/signal"
89
"syscall"
@@ -29,6 +30,10 @@ func catchSignals(pid int, sigs []os.Signal) {
2930
s:=<-sc
3031
sig,ok:=s.(syscall.Signal)
3132
ifok {
33+
// TODO:
34+
// Tried using a logger here but the I/O streams are already closed at this point...
35+
// Why is os.Stderr still working then?
36+
_,_=fmt.Fprintf(os.Stderr,"reaper caught %q signal, killing process %v\n",sig.String(),pid)
3237
_=syscall.Kill(pid,sig)
3338
}
3439
}

‎cli/agent.go‎

Lines changed: 89 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,16 @@ import (
1919
"golang.org/x/xerrors"
2020
"gopkg.in/natefinch/lumberjack.v2"
2121

22+
"github.com/coder/retry"
23+
2224
"github.com/prometheus/client_golang/prometheus"
2325

2426
"cdr.dev/slog"
2527
"cdr.dev/slog/sloggers/sloghuman"
2628
"cdr.dev/slog/sloggers/slogjson"
2729
"cdr.dev/slog/sloggers/slogstackdriver"
30+
"github.com/coder/serpent"
31+
2832
"github.com/coder/coder/v2/agent"
2933
"github.com/coder/coder/v2/agent/agentcontainers"
3034
"github.com/coder/coder/v2/agent/agentexec"
@@ -34,7 +38,6 @@ import (
3438
"github.com/coder/coder/v2/cli/clilog"
3539
"github.com/coder/coder/v2/codersdk"
3640
"github.com/coder/coder/v2/codersdk/agentsdk"
37-
"github.com/coder/serpent"
3841
)
3942

4043
func (r*RootCmd)workspaceAgent()*serpent.Command {
@@ -63,8 +66,10 @@ func (r *RootCmd) workspaceAgent() *serpent.Command {
6366
// This command isn't useful to manually execute.
6467
Hidden:true,
6568
Handler:func(inv*serpent.Invocation)error {
66-
ctx,cancel:=context.WithCancel(inv.Context())
67-
defercancel()
69+
ctx,cancel:=context.WithCancelCause(inv.Context())
70+
deferfunc() {
71+
cancel(xerrors.New("defer"))
72+
}()
6873

6974
var (
7075
ignorePorts=map[int]string{}
@@ -281,7 +286,6 @@ func (r *RootCmd) workspaceAgent() *serpent.Command {
281286
returnxerrors.Errorf("add executable to $PATH: %w",err)
282287
}
283288

284-
prometheusRegistry:=prometheus.NewRegistry()
285289
subsystemsRaw:=inv.Environ.Get(agent.EnvAgentSubsystem)
286290
subsystems:= []codersdk.AgentSubsystem{}
287291
for_,s:=rangestrings.Split(subsystemsRaw,",") {
@@ -328,46 +332,90 @@ func (r *RootCmd) workspaceAgent() *serpent.Command {
328332
containerLister=agentcontainers.NewDocker(execer)
329333
}
330334

331-
agnt:=agent.New(agent.Options{
332-
Client:client,
333-
Logger:logger,
334-
LogDir:logDir,
335-
ScriptDataDir:scriptDataDir,
336-
// #nosec G115 - Safe conversion as tailnet listen port is within uint16 range (0-65535)
337-
TailnetListenPort:uint16(tailnetListenPort),
338-
ExchangeToken:func(ctx context.Context) (string,error) {
339-
ifexchangeToken==nil {
340-
returnclient.SDK.SessionToken(),nil
335+
// TODO: timeout ok?
336+
reinitCtx,reinitCancel:=context.WithTimeout(context.Background(),time.Hour*24)
337+
deferreinitCancel()
338+
reinitEvents:=make(chan agentsdk.ReinitializationResponse)
339+
340+
gofunc() {
341+
// Retry to wait for reinit, main context cancels the retrier.
342+
forretrier:=retry.New(100*time.Millisecond,10*time.Second);retrier.Wait(ctx); {
343+
select {
344+
case<-reinitCtx.Done():
345+
return
346+
default:
341347
}
342-
resp,err:=exchangeToken(ctx)
348+
349+
err:=client.WaitForReinit(reinitCtx,reinitEvents)
343350
iferr!=nil {
344-
return"",err
351+
logger.Error(ctx,"failed to wait for reinit instructions, will retry",slog.Error(err))
345352
}
346-
client.SetSessionToken(resp.SessionToken)
347-
returnresp.SessionToken,nil
348-
},
349-
EnvironmentVariables:environmentVariables,
350-
IgnorePorts:ignorePorts,
351-
SSHMaxTimeout:sshMaxTimeout,
352-
Subsystems:subsystems,
353-
354-
PrometheusRegistry:prometheusRegistry,
355-
BlockFileTransfer:blockFileTransfer,
356-
Execer:execer,
357-
ContainerLister:containerLister,
358-
359-
ExperimentalDevcontainersEnabled:experimentalDevcontainersEnabled,
360-
})
361-
362-
promHandler:=agent.PrometheusMetricsHandler(prometheusRegistry,logger)
363-
prometheusSrvClose:=ServeHandler(ctx,logger,promHandler,prometheusAddress,"prometheus")
364-
deferprometheusSrvClose()
365-
366-
debugSrvClose:=ServeHandler(ctx,logger,agnt.HTTPDebug(),debugAddress,"debug")
367-
deferdebugSrvClose()
368-
369-
<-ctx.Done()
370-
returnagnt.Close()
353+
}
354+
}()
355+
356+
var (
357+
lastErrerror
358+
mustExitbool
359+
)
360+
for {
361+
prometheusRegistry:=prometheus.NewRegistry()
362+
363+
agnt:=agent.New(agent.Options{
364+
Client:client,
365+
Logger:logger,
366+
LogDir:logDir,
367+
ScriptDataDir:scriptDataDir,
368+
// #nosec G115 - Safe conversion as tailnet listen port is within uint16 range (0-65535)
369+
TailnetListenPort:uint16(tailnetListenPort),
370+
ExchangeToken:func(ctx context.Context) (string,error) {
371+
ifexchangeToken==nil {
372+
returnclient.SDK.SessionToken(),nil
373+
}
374+
resp,err:=exchangeToken(ctx)
375+
iferr!=nil {
376+
return"",err
377+
}
378+
client.SetSessionToken(resp.SessionToken)
379+
returnresp.SessionToken,nil
380+
},
381+
EnvironmentVariables:environmentVariables,
382+
IgnorePorts:ignorePorts,
383+
SSHMaxTimeout:sshMaxTimeout,
384+
Subsystems:subsystems,
385+
386+
PrometheusRegistry:prometheusRegistry,
387+
BlockFileTransfer:blockFileTransfer,
388+
Execer:execer,
389+
ContainerLister:containerLister,
390+
ExperimentalDevcontainersEnabled:experimentalDevcontainersEnabled,
391+
})
392+
393+
promHandler:=agent.PrometheusMetricsHandler(prometheusRegistry,logger)
394+
prometheusSrvClose:=ServeHandler(ctx,logger,promHandler,prometheusAddress,"prometheus")
395+
396+
debugSrvClose:=ServeHandler(ctx,logger,agnt.HTTPDebug(),debugAddress,"debug")
397+
398+
select {
399+
case<-ctx.Done():
400+
logger.Warn(ctx,"agent shutting down",slog.Error(ctx.Err()),slog.F("cause",context.Cause(ctx)))
401+
mustExit=true
402+
caseevent:=<-reinitEvents:
403+
logger.Warn(ctx,"agent received instruction to reinitialize",
404+
slog.F("message",event.Message),slog.F("reason",event.Reason))
405+
}
406+
407+
lastErr=agnt.Close()
408+
debugSrvClose()
409+
prometheusSrvClose()
410+
411+
ifmustExit {
412+
reinitCancel()
413+
break
414+
}
415+
416+
logger.Info(ctx,"reinitializing...")
417+
}
418+
returnlastErr
371419
},
372420
}
373421

‎coderd/agentapi/api.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,8 @@ func New(opts Options) *API {
109109
Database:opts.Database,
110110
DerpMapFn:opts.DerpMapFn,
111111
WorkspaceID:opts.WorkspaceID,
112+
Log:opts.Log.Named("manifests"),
113+
Pubsub:opts.Pubsub,
112114
}
113115

114116
api.AnnouncementBannerAPI=&AnnouncementBannerAPI{

‎coderd/agentapi/manifest.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,10 @@ import (
88
"strings"
99
"time"
1010

11+
"cdr.dev/slog"
12+
13+
"github.com/coder/coder/v2/coderd/database/pubsub"
14+
1115
"github.com/google/uuid"
1216
"golang.org/x/sync/errgroup"
1317
"golang.org/x/xerrors"
@@ -35,6 +39,8 @@ type ManifestAPI struct {
3539
AgentFnfunc(context.Context) (database.WorkspaceAgent,error)
3640
Database database.Store
3741
DerpMapFnfunc()*tailcfg.DERPMap
42+
Pubsub pubsub.Pubsub
43+
Log slog.Logger
3844
}
3945

4046
func (a*ManifestAPI)GetManifest(ctx context.Context,_*agentproto.GetManifestRequest) (*agentproto.Manifest,error) {

‎coderd/apidoc/docs.go‎

Lines changed: 45 additions & 0 deletions
Some generated files are not rendered by default. Learn more aboutcustomizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp