From 5a6a36d22230d6fc8cbe27fc2cc93173319d3c9d Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Thu, 19 Jan 2023 14:38:04 -0800 Subject: [PATCH] `runsc` metric server: Periodically add sandboxes that failed registration. The metric server periodically scans the root directory and picks up all the containers that requested instrumentation. This change is part of a series of changes to support Prometheus-style metrics in `runsc`. PiperOrigin-RevId: 503268994 --- runsc/cmd/metric_server.go | 169 ++++++++++++++++++++++++++++++---- runsc/container/state_file.go | 6 ++ runsc/sandbox/sandbox.go | 11 ++- 3 files changed, 167 insertions(+), 19 deletions(-) diff --git a/runsc/cmd/metric_server.go b/runsc/cmd/metric_server.go index 15cdb8e86..e2f4948b7 100644 --- a/runsc/cmd/metric_server.go +++ b/runsc/cmd/metric_server.go @@ -67,9 +67,10 @@ const ( // servedSandbox is a sandbox that we serve metrics from. // A single metrics server will export data about multiple sandboxes. type servedSandbox struct { - rootContainerID container.FullID - rootDir string - extraLabels map[string]string + rootContainerID container.FullID + rootDir string + metricServerAddr string + extraLabels map[string]string // mu protects the fields below. mu sync.Mutex @@ -108,7 +109,15 @@ func (s *servedSandbox) load() (*sandbox.Sandbox, *prometheus.Verifier, error) { if err != nil { return nil, nil, err } + sandboxMetricAddr := strings.ReplaceAll(cont.Sandbox.MetricServerAddress, "%RUNTIME_ROOT%", s.rootDir) + if sandboxMetricAddr == "" { + return nil, nil, errors.New("sandbox did not request instrumentation") + } + if sandboxMetricAddr != s.metricServerAddr { + return nil, nil, fmt.Errorf("sandbox requested instrumentation by a metric server running at a different address (sandbox wants %q, this metric server serves %q)", sandboxMetricAddr, s.metricServerAddr) + } // Update label data as read from the state file. + // Do not store empty labels. authoritativeLabels := cont.Sandbox.PrometheusLabels() for _, label := range []string{sandbox.SandboxIDLabel, sandbox.PodNameLabel, sandbox.NamespaceLabel} { s.extraLabels[label] = authoritativeLabels[label] @@ -166,6 +175,7 @@ func queryMetrics(ctx context.Context, sand *sandbox.Sandbox, verifier *promethe // MetricServer implements subcommands.Command for the "metric-server" command. type MetricServer struct { rootDir string + address string exporterPrefix string startTime time.Time rand *rand.Rand @@ -188,6 +198,11 @@ type MetricServer struct { // sandboxes is the list of sandboxes we serve metrics for. sandboxes map[container.FullID]*servedSandbox + // lastStateFileStat maps container full IDs to the last observed stat() of their state file. + // This is used to monitor for sandboxes in the background. If a sandbox's state file matches this + // info, we can assume that the last background scan already looked at it. + lastStateFileStat map[container.FullID]os.FileInfo + // numSandboxes counts the number of sandboxes that have ever been registered on this server. // Used to distinguish between the case where this metrics serve has sat there doing nothing // because no sandbox ever registered against it (which is unexpected), vs the case where it has @@ -223,9 +238,40 @@ func (*MetricServer) Usage() string { // SetFlags implements subcommands.Command.SetFlags. func (m *MetricServer) SetFlags(f *flag.FlagSet) {} -// purgeSandboxesLocked removes sandboxes that are no longer running from m.sandboxes. +// sufficientlyEqualStats returns whether the given FileInfo's are sufficiently +// equal to assume the file they represent has not changed between the time +// each FileInfo was obtained. +func sufficientlyEqualStats(s1, s2 os.FileInfo) bool { + if !s1.ModTime().Equal(s2.ModTime()) { + return false + } + if s1.Size() != s2.Size() { + return false + } + statT1, ok1 := s1.Sys().(*syscall.Stat_t) + statT2, ok2 := s2.Sys().(*syscall.Stat_t) + if ok1 != ok2 { + return false + } + if ok1 && ok2 { + if statT1.Dev != statT2.Dev { + return false + } + if statT1.Ino != statT2.Ino { + return false + } + } + return true +} + +// refreshSandboxesLocked removes sandboxes that are no longer running from m.sandboxes, and +// adds sandboxes found in the root directory that do request instrumentation. // Preconditions: m.mu is locked. -func (m *MetricServer) purgeSandboxesLocked() { +func (m *MetricServer) refreshSandboxesLocked() { + if m.shuttingDown { + // Do nothing to avoid log spam. + return + } sandboxIDs, err := container.ListSandboxes(m.rootDir) if err != nil { log.Warningf("Cannot list containers in root directory %s, it has likely gone away: %v.", m.rootDir, err) @@ -247,6 +293,86 @@ func (m *MetricServer) purgeSandboxesLocked() { delete(m.sandboxes, sandboxID) } } + newSandboxIDs := make(map[container.FullID]bool, len(sandboxIDs)) + for _, sid := range sandboxIDs { + if _, found := m.sandboxes[sid]; found { + continue + } + newSandboxIDs[sid] = true + } + for sid := range m.lastStateFileStat { + if _, found := newSandboxIDs[sid]; !found { + delete(m.lastStateFileStat, sid) + } + } + for sid := range newSandboxIDs { + stateFile := container.StateFile{ + RootDir: m.rootDir, + ID: sid, + } + stat, err := stateFile.Stat() + if err != nil { + log.Warningf("Failed to stat() container state file for sandbox %q: %v", sid, err) + continue + } + if existing, found := m.lastStateFileStat[sid]; found { + // We already tried to stat this sandbox but decided not to pick it up. + // Check if the state file changed since. If it didn't, we don't want to + // try again. + if sufficientlyEqualStats(existing, stat) { + continue + } + log.Infof("State file for sandbox %q has changed since we last looked at it; will try to reload it.", sid) + delete(m.lastStateFileStat, sid) + } + // If we get here, we either haven't seen this sandbox before, or we saw it + // and it has disappeared (which means it is new in this iteration), or we + // saw it before but its state file changed. Either way, we want to try + // loading it and see if it wants instrumentation. + cont, err := container.Load(m.rootDir, sid, container.LoadOpts{ + Exact: true, + SkipCheck: true, + TryLock: container.TryAcquire, + RootContainer: true, + }) + if err != nil { + if err == container.ErrStateFileLocked { + // This error is OK and shouldn't generate log spam. The sandbox is probably in the middle + // of being created. + continue + } + log.Warningf("Cannot load state file for sandbox %q: %v", sid, err) + continue + } + // This is redundant with one of the checks performed below in servedSandbox.load(), but this + // avoids log spam for the non-error case of sandboxes that didn't request instrumentation. + sandboxMetricAddr := strings.ReplaceAll(cont.Sandbox.MetricServerAddress, "%RUNTIME_ROOT%", m.rootDir) + if sandboxMetricAddr != m.address { + m.lastStateFileStat[sid] = stat + continue + } + m.numSandboxes++ + served := &servedSandbox{ + rootContainerID: sid, + rootDir: m.rootDir, + metricServerAddr: m.address, + extraLabels: map[string]string{ + sandbox.SandboxIDLabel: sid.SandboxID, + iterationIDLabel: fmt.Sprintf("%d", m.rand.Uint64()), + }, + } + // Best-effort attempt to load the state file instantly. + // This may legitimately fail if it is locked, e.g. during sandbox startup. + // If it fails for any other reason, then the sandbox went away between the time we listed the + // sandboxes and now, so just delete it. + if _, _, err := served.load(); err != nil && err != container.ErrStateFileLocked { + log.Warningf("Sandbox %q cannot be loaded, ignoring it: %v", sid, err) + m.lastStateFileStat[sid] = stat + continue + } + m.sandboxes[sid] = served + log.Infof("Registered new sandbox found in root directory: %q", sid) + } } // httpResult is returned by HTTP handlers. @@ -309,7 +435,7 @@ func (m *MetricServer) serveMetrics(w http.ResponseWriter, req *http.Request) ht ctx, ctxCancel := context.WithTimeout(req.Context(), metricsExportTimeout) defer ctxCancel() m.mu.Lock() - m.purgeSandboxesLocked() + m.refreshSandboxesLocked() numGoroutines := exportParallelGoroutines numSandboxes := len(m.sandboxes) @@ -543,6 +669,23 @@ func logRequest(f func(w http.ResponseWriter, req *http.Request) httpResult) fun } } +// verify is one iteration of verifyLoop. +// It runs in a loop in the background which checks all sandboxes for liveness, tries to load +// their metadata if that hasn't been loaded yet, and tries to pick up new sandboxes that +// failed to register for whatever reason. +func (m *MetricServer) verify(ctx context.Context) { + _, err := container.ListSandboxes(m.rootDir) + m.mu.Lock() + defer m.mu.Unlock() + if err != nil { + log.Warningf("Cannot list sandboxes in root directory %s, it has likely gone away: %v. Server shutting down.", m.rootDir, err) + m.shutdownLocked(ctx) + return + } + m.refreshSandboxesLocked() +} + +// verifyLoop runs in the background and periodically calls verify. func (m *MetricServer) verifyLoop(ctx context.Context) { ticker := time.NewTicker(verifyLoopInterval) defer ticker.Stop() @@ -557,17 +700,7 @@ func (m *MetricServer) verifyLoop(ctx context.Context) { m.mu.Unlock() return case <-ticker.C: - _, listErr := container.ListSandboxes(m.rootDir) - func() { - m.mu.Lock() - defer m.mu.Unlock() - if listErr != nil { - log.Warningf("Cannot list sandboxes in root directory %s, it has likely gone away: %v. Server shutting down.", m.rootDir, listErr) - m.shutdownLocked(ctx) - return - } - m.purgeSandboxesLocked() - }() + m.verify(ctx) } } } @@ -601,7 +734,9 @@ func (m *MetricServer) Execute(ctx context.Context, f *flag.FlagSet, args ...any log.Infof("Metric server address replaced %RUNTIME_ROOT%: %q -> %q", conf.MetricServer, newAddr) conf.MetricServer = newAddr } + m.address = conf.MetricServer m.sandboxes = make(map[container.FullID]*servedSandbox) + m.lastStateFileStat = make(map[container.FullID]os.FileInfo) var listener net.Listener var listenErr error diff --git a/runsc/container/state_file.go b/runsc/container/state_file.go index fc0fb05b7..29b53d6f6 100644 --- a/runsc/container/state_file.go +++ b/runsc/container/state_file.go @@ -374,6 +374,12 @@ func (s *StateFile) SaveLocked(v any) error { return nil } +// Stat returns the result of calling stat() on the state file. +// Doing so does not require locking. +func (s *StateFile) Stat() (os.FileInfo, error) { + return os.Stat(s.statePath()) +} + func (s *StateFile) load(v any, opts LoadOpts) error { if err := s.lock(opts.TryLock); err != nil { return err diff --git a/runsc/sandbox/sandbox.go b/runsc/sandbox/sandbox.go index 7ed4d665b..8756d05b8 100644 --- a/runsc/sandbox/sandbox.go +++ b/runsc/sandbox/sandbox.go @@ -140,6 +140,12 @@ type Sandbox struct { // This is a textproto string of metricpb.MetricRegistration. RegisteredMetrics string `json:"registeredMetrics"` + // MetricServerAddress is the address of the metric server that this sandbox + // intends to export metrics for. + // Only populated if exporting metrics was requested when the sandbox was + // created. + MetricServerAddress string `json:"metricServerAddress"` + // child is set if a sandbox process is a child of the current process. // // This field isn't saved to json, because only a creator of sandbox @@ -213,8 +219,9 @@ func New(conf *config.Config, args *Args) (*Sandbox, error) { CgroupJSON: cgroup.CgroupJSON{ Cgroup: args.Cgroup, }, - UID: -1, // prevent usage before it's set. - GID: -1, // prevent usage before it's set. + UID: -1, // prevent usage before it's set. + GID: -1, // prevent usage before it's set. + MetricServerAddress: conf.MetricServer, } if args.Spec != nil && args.Spec.Annotations != nil { s.PodName = args.Spec.Annotations[podNameAnnotation]