From 2ce059fcad23f62bd3f32c285f00290070efb737 Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Thu, 19 Jan 2023 13:43:03 -0800 Subject: [PATCH] gVisor: Implement `runsc metric-server` which serves Prometheus metrics. The metrics server implements the following interface: - GET `/metrics`: Serves Prometheus metrics. - POST `/runsc-metrics`: Contains administrative endpoints. All of them require the `root` argument to be specified, and match the one that the server expects. This is used to avoid confusing multiple instances of the `runsc` metrics server. - POST `/runsc-metrics/healthcheck`: Used by clients to verify that the server is running and is the expected metric server. This change has no tests, but coverage is provided in a later change that provides an end-to-end container tests that the metric server works and exports data faithfully. This change is part of a series of changes to support Prometheus-style metrics in `runsc`. Doing so requires making several seemingly-odd design decisions, due to the following architectural constraints: - Prometheus requires an HTTP server serving the `/metrics` endpoint. - For performance reasons, the `runsc boot` process cannot run the `netpoller` goroutine. - Since we don't want to write our own HTTP server implementation, this means the HTTP endpoint has to be served by a separate process that remains running during the lifetime of the container. - The `runsc boot` process is untrusted. - This means we cannot trust metrics data that comes out of the Sentry. Therefore, there needs to be an elaborate dance where we pre-register metric metadata before starting any untrusted workload. Then, the server relaying the metric data must verify the validity of metric values against this metric metadata. This avoids leaking metrics, cardinality blow-ups, and other such DoS vectors. - This feature needs to be easy-to-use in a typical Docker setting. - This means having the ability to just say `--metrics-server=localhost:1337` in the `runsc` runtime entry in `/etc/docker/daemon.json` and have that Just Work(TM), even when multiple containers are running. - Since only one process may listen on a port at a given time, this means the metric server needs to be able to multiplex requests out to multiple running sandboxes, and remain alive for the entire duration of either of these sandboxes. - For this reason, the metrics server runs *outside* of the usual per-container cgroups. - This also saves system resources by not running one server per sandbox. - The metrics server must be exposed to the outside world, and cannot assume that its clients are trustworthy. - For this reason, a metrics server is bound to a runtime root directory, and double-checks all that the sandboxes it is asked to follow actually exist in this root directory. PiperOrigin-RevId: 503254345 --- runsc/cli/main.go | 1 + runsc/cmd/BUILD | 4 + runsc/cmd/metric_export.go | 2 +- runsc/cmd/metric_server.go | 671 ++++++++++++++++++++++++++++++ runsc/config/config.go | 11 + runsc/config/flags.go | 1 + runsc/container/container.go | 10 +- runsc/container/container_test.go | 4 +- runsc/container/state_file.go | 47 ++- runsc/sandbox/BUILD | 2 + runsc/sandbox/sandbox.go | 69 ++- 11 files changed, 805 insertions(+), 17 deletions(-) create mode 100644 runsc/cmd/metric_server.go diff --git a/runsc/cli/main.go b/runsc/cli/main.go index 198f41958..1ac26ea07 100644 --- a/runsc/cli/main.go +++ b/runsc/cli/main.go @@ -103,6 +103,7 @@ func Main(version string) { subcommands.Register(new(cmd.Boot), internalGroup) subcommands.Register(new(cmd.Gofer), internalGroup) subcommands.Register(new(cmd.Umount), internalGroup) + subcommands.Register(new(cmd.MetricServer), internalGroup) // Register with the main command line. config.RegisterFlags(flag.CommandLine) diff --git a/runsc/cmd/BUILD b/runsc/cmd/BUILD index 7fc4f150c..2a9c06cd6 100644 --- a/runsc/cmd/BUILD +++ b/runsc/cmd/BUILD @@ -22,6 +22,7 @@ go_library( "kill.go", "list.go", "metric_export.go", + "metric_server.go", "mitigate.go", "mitigate_extras.go", "path.go", @@ -48,6 +49,7 @@ go_library( ], deps = [ "//pkg/abi/linux", + "//pkg/atomicbitops", "//pkg/coretag", "//pkg/coverage", "//pkg/log", @@ -59,6 +61,7 @@ go_library( "//pkg/sentry/platform", "//pkg/state/pretty", "//pkg/state/statefile", + "//pkg/sync", "//pkg/unet", "//pkg/urpc", "//runsc/boot", @@ -71,6 +74,7 @@ go_library( "//runsc/fsgofer/filter", "//runsc/mitigate", "//runsc/profile", + "//runsc/sandbox", "//runsc/specutils", "@com_github_google_subcommands//:go_default_library", "@com_github_opencontainers_runtime_spec//specs-go:go_default_library", diff --git a/runsc/cmd/metric_export.go b/runsc/cmd/metric_export.go index a58cd8563..ec0ea73a4 100644 --- a/runsc/cmd/metric_export.go +++ b/runsc/cmd/metric_export.go @@ -74,7 +74,7 @@ func (m *MetricExport) Execute(ctx context.Context, f *flag.FlagSet, args ...any }, map[*prometheus.Snapshot]prometheus.SnapshotExportOptions{ snapshot: { ExporterPrefix: conf.MetricExporterPrefix, - ExtraLabels: map[string]string{"sandbox": cont.Sandbox.ID}, + ExtraLabels: cont.Sandbox.PrometheusLabels(), }, }) if err != nil { diff --git a/runsc/cmd/metric_server.go b/runsc/cmd/metric_server.go new file mode 100644 index 000000000..15cdb8e86 --- /dev/null +++ b/runsc/cmd/metric_server.go @@ -0,0 +1,671 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cmd + +import ( + "context" + "errors" + "fmt" + "io" + "math/rand" + "net" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "github.com/google/subcommands" + "gvisor.dev/gvisor/pkg/atomicbitops" + "gvisor.dev/gvisor/pkg/log" + "gvisor.dev/gvisor/pkg/prometheus" + "gvisor.dev/gvisor/pkg/sync" + "gvisor.dev/gvisor/runsc/cmd/util" + "gvisor.dev/gvisor/runsc/config" + "gvisor.dev/gvisor/runsc/container" + "gvisor.dev/gvisor/runsc/flag" + "gvisor.dev/gvisor/runsc/sandbox" +) + +const ( + // verifyLoopInterval is the interval at which we check whether there are any sandboxes we need + // to serve metrics for. If there are none, the server exits. + verifyLoopInterval = 20 * time.Second + + // httpTimeout is the timeout used for all connect/read/write operations of the HTTP server. + httpTimeout = 1 * time.Minute + + // metricsExportTimeout is the maximum amount of time that the metrics export process should take. + metricsExportTimeout = 30 * time.Second + + // metricsExportPerSandboxTimeout is the maximum amount of time that we wait on any individual + // sandbox when exporting its metrics. + metricsExportPerSandboxTimeout = 8 * time.Second + + // exportParallelGoroutines is the maximum number of goroutines spawned during metrics export. + exportParallelGoroutines = 8 +) + +// Prometheus label names. +const ( + iterationIDLabel = "iterationid" +) + +// 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 + + // mu protects the fields below. + mu sync.Mutex + + // sandbox is the sandbox being monitored. + // Once set, it is immutable. + sandbox *sandbox.Sandbox + + // verifier allows verifying the data integrity of the metrics we get from this sandbox. + // It is not always initialized when the sandbox is discovered, but rather upon first metrics + // access to the sandbox. Metric registration data is loaded from the root container's + // state file. + // The server needs to load this registration data before any data from this sandbox is + // served to HTTP clients. If there is no metric registration data within the Container + // data, then metrics were not requested for this sandbox, and this servedSandbox should + // be deleted from the server. + // Once set, it is immutable. + verifier *prometheus.Verifier +} + +// load loads the sandbox being monitored and initializes its metric verifier. +// If it returns an error other than container.ErrStateFileLocked, the sandbox is either +// non-existent, or has not requested instrumentation to be enabled, or does not have +// valid metric registration data. In any of these cases, the sandbox should be removed +// from this metrics server. +func (s *servedSandbox) load() (*sandbox.Sandbox, *prometheus.Verifier, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.sandbox == nil { + cont, err := container.Load(s.rootDir, s.rootContainerID, container.LoadOpts{ + Exact: true, + SkipCheck: true, + TryLock: container.TryAcquire, + RootContainer: true, + }) + if err != nil { + return nil, nil, err + } + // Update label data as read from the state file. + authoritativeLabels := cont.Sandbox.PrometheusLabels() + for _, label := range []string{sandbox.SandboxIDLabel, sandbox.PodNameLabel, sandbox.NamespaceLabel} { + s.extraLabels[label] = authoritativeLabels[label] + if s.extraLabels[label] == "" { + delete(s.extraLabels, label) + } + } + s.sandbox = cont.Sandbox + } + if s.verifier == nil { + registeredMetrics, err := s.sandbox.GetRegisteredMetrics() + if err != nil { + return nil, nil, err + } + verifier, err := prometheus.NewVerifier(registeredMetrics) + if err != nil { + return nil, nil, err + } + s.verifier = verifier + } + return s.sandbox, s.verifier, nil +} + +// queryMetrics queries the sandbox for metrics data. +func queryMetrics(ctx context.Context, sand *sandbox.Sandbox, verifier *prometheus.Verifier) (*prometheus.Snapshot, error) { + ch := make(chan struct { + snapshot *prometheus.Snapshot + err error + }, 1) + defer close(ch) + go func() { + snapshot, err := sand.ExportMetrics() + select { + case ch <- struct { + snapshot *prometheus.Snapshot + err error + }{snapshot, err}: + default: + } + }() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case ret := <-ch: + if ret.err != nil { + return nil, ret.err + } + if err := verifier.Verify(ret.snapshot); err != nil { + return nil, err + } + return ret.snapshot, nil + } +} + +// MetricServer implements subcommands.Command for the "metric-server" command. +type MetricServer struct { + rootDir string + exporterPrefix string + startTime time.Time + rand *rand.Rand + srv http.Server + + // Size of the map of written metrics during the last /metrics export. Initially zero. + // Used to efficiently reallocate a map of the right size during the next export. + lastMetricsWrittenSize atomicbitops.Uint32 + + // mu protects the fields below. + mu sync.Mutex + + // udsPath is a path to a Unix Domain Socket file on which the server is bound and which it owns. + // This socket file will be deleted on server shutdown. + // This field is not set if binding to a network port, or when the UDS already existed prior to + // being bound by us (i.e. its ownership isn't ours), such that it isn't deleted in this case. + // The field is unset once the file is succesfully removed. + udsPath string + + // sandboxes is the list of sandboxes we serve metrics for. + sandboxes map[container.FullID]*servedSandbox + + // 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 + // done a good job serving sandbox metrics and it's time for it to gracefully die as there are no + // more sandboxes to serve. + // Also exported as a metric of total number of sandboxes started. + numSandboxes int64 + + // shuttingDown is flipped to true when the server shutdown process has started. + // Used to deal with race conditions where a sandbox is trying to register after the server has + // already started to go to sleep. + shuttingDown bool + + // shutdownCh is written to when receiving the signal to shut down gracefully. + shutdownCh chan os.Signal +} + +// Name implements subcommands.Command.Name. +func (*MetricServer) Name() string { + return "metric-server" +} + +// Synopsis implements subcommands.Command.Synopsis. +func (*MetricServer) Synopsis() string { + return "implements Prometheus metrics HTTP endpoint" +} + +// Usage implements subcommands.Command.Usage. +func (*MetricServer) Usage() string { + return `-root= -metric-server= [-metric-exporter-prefix=] metric-server` +} + +// SetFlags implements subcommands.Command.SetFlags. +func (m *MetricServer) SetFlags(f *flag.FlagSet) {} + +// purgeSandboxesLocked removes sandboxes that are no longer running from m.sandboxes. +// Preconditions: m.mu is locked. +func (m *MetricServer) purgeSandboxesLocked() { + 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) + return + } + for sandboxID, sandbox := range m.sandboxes { + found := false + for _, sid := range sandboxIDs { + if sid == sandboxID { + found = true + break + } + } + if !found { + log.Warningf("Sandbox %s no longer exists but did not explicitly unregister. Removing it.", sandboxID) + delete(m.sandboxes, sandboxID) + } else if _, _, err := sandbox.load(); err != nil && err != container.ErrStateFileLocked { + log.Warningf("Sandbox %s cannot be loaded, deleting it: %v", sandboxID, err) + delete(m.sandboxes, sandboxID) + } + } +} + +// httpResult is returned by HTTP handlers. +type httpResult struct { + code int + err error +} + +// httpOK is the "everything went fine" HTTP result. +var httpOK = httpResult{code: http.StatusOK} + +// serveIndex serves the index page. +func (m *MetricServer) serveIndex(w http.ResponseWriter, req *http.Request) httpResult { + if req.URL.Path != "/" { + return httpResult{http.StatusNotFound, errors.New("path not found")} + } + fmt.Fprintf(w, "runsc metrics") + fmt.Fprintf(w, "

You have reached the runsc metrics server page!

") + fmt.Fprintf(w, `

To see actual metric data, head over to /metrics.

`) + fmt.Fprintf(w, "") + return httpOK +} + +// Metrics generated by the metrics server itself. +var ( + sandboxPresenceMetric = prometheus.Metric{ + Name: "sandbox_presence", + Type: prometheus.TypeGauge, + Help: "Boolean metric set to 1 for each known sandbox.", + } + sandboxRunningMetric = prometheus.Metric{ + Name: "sandbox_running", + Type: prometheus.TypeGauge, + Help: "Boolean metric set to 1 for each running sandbox.", + } + numRunningSandboxesMetric = prometheus.Metric{ + Name: "num_sandboxes_running", + Type: prometheus.TypeGauge, + Help: "Number of sandboxes running at present.", + } + numCannotExportSandboxesMetric = prometheus.Metric{ + Name: "num_sandboxes_broken_metrics", + Type: prometheus.TypeGauge, + Help: "Number of sandboxes from which we cannot export metrics.", + } + numTotalSandboxesMetric = prometheus.Metric{ + Name: "num_sandboxes_total", + Type: prometheus.TypeCounter, + Help: "Counter of sandboxes that have ever been started.", + } + processStartTimeMetric = prometheus.Metric{ + Name: "process_start_time_seconds", + Type: prometheus.TypeGauge, + Help: "Unix timestamp at which the process started. Used by Prometheus for counter resets.", + } +) + +// serveMetrics serves metrics requests. +func (m *MetricServer) serveMetrics(w http.ResponseWriter, req *http.Request) httpResult { + ctx, ctxCancel := context.WithTimeout(req.Context(), metricsExportTimeout) + defer ctxCancel() + m.mu.Lock() + m.purgeSandboxesLocked() + + numGoroutines := exportParallelGoroutines + numSandboxes := len(m.sandboxes) + if numSandboxes < numGoroutines { + numGoroutines = numSandboxes + } + + // First, load all the sandboxes in parallel. We need to do this while m.mu is held. + loadSandboxCh := make(chan *servedSandbox, numSandboxes) + type sandboxLoadResult struct { + served *servedSandbox + sandbox *sandbox.Sandbox + verifier *prometheus.Verifier + err error + } + loadedSandboxesCh := make(chan sandboxLoadResult, numSandboxes) + loadedSandboxes := make([]sandboxLoadResult, 0, numSandboxes) + for i := 0; i < numGoroutines; i++ { + go func() { + for served := range loadSandboxCh { + sand, verifier, err := served.load() + loadedSandboxesCh <- sandboxLoadResult{served, sand, verifier, err} + } + }() + } + for _, sandbox := range m.sandboxes { + loadSandboxCh <- sandbox + } + close(loadSandboxCh) + for i := 0; i < numSandboxes; i++ { + loadedSandboxes = append(loadedSandboxes, <-loadedSandboxesCh) + } + close(loadedSandboxesCh) + m.mu.Unlock() + + // Now iterate over all sandboxes. + // Important: This must be done in random order. + // A malicious/compromised sandbox may decide to stall when being asked for metrics. + // If at least `numGoroutines` sandboxes do this, this will starve other sandboxes + // from having their metrics exported, because all the goroutines will be stuck on + // the stalled sandboxes. + // One way to completely avoid this would be to spawn one goroutine per + // sandbox, but this can amount to ~hundreds of goroutines, which is not desirable + // for the metrics server. + // Another way would be to have a very strict timeout on each sandbox's export + // process, but in some cases a busy sandbox will take more than a decisecond + // or so to export its data, so this would miss some data from legitimate (but + // slow) sandboxes. + // Instead, we take a middle-of-the-road approach: we use a timeout that's not + // too strict but still ensures we make forward progress away from stalled + // sandboxes, and we also iterate across sandboxes in a different random order at + // each export. This ensures that all sandboxes eventually get a fair chance of + // being part of the "first `numGoroutines` sandboxes in line" to get their + // metric data loaded, such that a client repeatedly scraping metrics will + // eventually get data from each sandbox. + + // Used to prevent goroutines from accessing the shared variables below. + var metricsMu sync.Mutex + + // Meta-metrics keep track of metrics to export about the metrics server itself. + type metaMetrics struct { + numRunningSandboxes int64 + numCannotExportSandboxes int64 + } + meta := metaMetrics{} // Protected by metricsMu. + selfMetrics := prometheus.NewSnapshot() // Protected by metricsMu. + + ctxDeadline, ok := ctx.Deadline() + if !ok { + panic("context had no deadline, this should never happen as it was created with a timeout") + } + exportStartTime := time.Now() + requestTimeLeft := ctxDeadline.Sub(exportStartTime) + perSandboxTime := requestTimeLeft + if numSandboxes != 0 { + perSandboxTime = requestTimeLeft / time.Duration(numSandboxes) + } + if perSandboxTime < metricsExportPerSandboxTimeout { + perSandboxTime = metricsExportPerSandboxTimeout + } + loadedSandboxCh := make(chan sandboxLoadResult, numSandboxes) + type snapshotAndOptions struct { + snapshot *prometheus.Snapshot + options prometheus.SnapshotExportOptions + } + snapshotCh := make(chan snapshotAndOptions, numSandboxes) + var wg sync.WaitGroup + wg.Add(numGoroutines) + for i := 0; i < numGoroutines; i++ { + go func(metricsMu *sync.Mutex, meta *metaMetrics, selfMetrics *prometheus.Snapshot) { + defer wg.Done() + for s := range loadedSandboxCh { + served, sand, verifier, err := s.served, s.sandbox, s.verifier, s.err + isRunning := false + var snapshot *prometheus.Snapshot + if err == nil { + queryCtx, queryCtxCancel := context.WithTimeout(ctx, perSandboxTime) + snapshot, err = queryMetrics(queryCtx, sand, verifier) + queryCtxCancel() + isRunning = sand.IsRunning() + } + func() { + metricsMu.Lock() + defer metricsMu.Unlock() + selfMetrics.Add(prometheus.LabeledIntData(&sandboxPresenceMetric, served.extraLabels, 1)) + sandboxRunning := int64(0) + if isRunning { + sandboxRunning = 1 + } + selfMetrics.Add(prometheus.LabeledIntData(&sandboxRunningMetric, served.extraLabels, sandboxRunning)) + if err != nil && !isRunning { + // The sandbox either hasn't started running yet, or it ran and has gone away between the + // start of the function and now. It is normal that metrics are not exported for this + // sandbox in this case, so do not report this as an error. + return + } + if err != nil { + meta.numRunningSandboxes++ + meta.numCannotExportSandboxes++ + log.Warningf("Could not export metrics from sandbox %s: %v", served.rootContainerID.SandboxID, err) + return + } + meta.numRunningSandboxes++ + snapshotCh <- snapshotAndOptions{ + snapshot: snapshot, + options: prometheus.SnapshotExportOptions{ + ExporterPrefix: m.exporterPrefix, + ExtraLabels: served.extraLabels, + }, + } + }() + } + }(&metricsMu, &meta, selfMetrics) + } + // Feed the channel in random order: + for _, sandboxIndex := range rand.Perm(len(loadedSandboxes)) { + loadedSandboxCh <- loadedSandboxes[sandboxIndex] + } + close(loadedSandboxCh) + + // Meanwhile, build the map of all snapshots we will be rendering. + snapshotsToOptions := make(map[*prometheus.Snapshot]prometheus.SnapshotExportOptions, numSandboxes+2) + snapshotsToOptions[selfMetrics] = prometheus.SnapshotExportOptions{ + ExporterPrefix: fmt.Sprintf("%smeta_", m.exporterPrefix), + } + processMetrics := prometheus.NewSnapshot() + processMetrics.Add(prometheus.NewFloatData(&processStartTimeMetric, float64(m.startTime.Unix())+(float64(m.startTime.Nanosecond())/1e9))) + snapshotsToOptions[processMetrics] = prometheus.SnapshotExportOptions{ + // These metrics must be written without any prefix. + } + + // Aggregate all the snapshots from the sandboxes. + wg.Wait() + close(snapshotCh) + for snapshotAndOptions := range snapshotCh { + snapshotsToOptions[snapshotAndOptions.snapshot] = snapshotAndOptions.options + } + + // Write out all data. + lastMetricsWrittenSize := int(m.lastMetricsWrittenSize.Load()) + metricsWritten := make(map[string]bool, lastMetricsWrittenSize) + written, err := prometheus.Write(w, prometheus.ExportOptions{ + CommentHeader: fmt.Sprintf("Data for runsc metric server exporting data for sandboxes in root directory %s", m.rootDir), + MetricsWritten: metricsWritten, + }, snapshotsToOptions) + if err != nil { + if written == 0 { + return httpResult{http.StatusServiceUnavailable, err} + } + // Note that we cannot return an HTTP error here because we have already started writing a + // response, which means we've already responded with a 200 OK status code. + // This probably means the client closed the connection before we could finish writing. + return httpOK + } + if lastMetricsWrittenSize < len(metricsWritten) { + m.lastMetricsWrittenSize.CompareAndSwap(uint32(lastMetricsWrittenSize), uint32(len(metricsWritten))) + } + return httpOK +} + +// serveHealthCheck serves the healthcheck endpoint. +// Returns a response prefixed by "runsc-metrics:OK" on success. +// Clients can use this to assert that they are talking to the metrics server, as opposed to some +// other random HTTP server. +func (m *MetricServer) serveHealthCheck(w http.ResponseWriter, req *http.Request) httpResult { + m.mu.Lock() + defer m.mu.Unlock() + if m.shuttingDown { + return httpResult{http.StatusServiceUnavailable, errors.New("server is shutting down already")} + } + if err := req.ParseForm(); err != nil { + return httpResult{http.StatusBadRequest, err} + } + rootDir := req.Form.Get("root") + if rootDir != m.rootDir { + return httpResult{http.StatusBadRequest, fmt.Errorf("this metric server is configured to serve root directory: %s", m.rootDir)} + } + w.WriteHeader(http.StatusOK) + io.WriteString(w, "runsc-metrics:OK") + return httpOK +} + +// shutdownLocked shuts down the server. It assumes mu is held. +func (m *MetricServer) shutdownLocked(ctx context.Context) { + log.Infof("Server shutting down.") + m.shuttingDown = true + if m.udsPath != "" { + if err := os.Remove(m.udsPath); err != nil { + log.Warningf("Cannot remove UDS at %s: %v", m.udsPath, err) + } else { + m.udsPath = "" + } + } + m.srv.Shutdown(ctx) +} + +// logRequest wraps an HTTP handler and adds logging to it. +func logRequest(f func(w http.ResponseWriter, req *http.Request) httpResult) func(w http.ResponseWriter, req *http.Request) { + return func(w http.ResponseWriter, req *http.Request) { + log.Infof("Request: %s %s", req.Method, req.URL.Path) + defer func() { + if r := recover(); r != nil { + log.Warningf("Request: %s %s: Panic:\n%v", req.Method, req.URL.Path, r) + } + }() + result := f(w, req) + if result.err != nil { + http.Error(w, result.err.Error(), result.code) + log.Warningf("Request: %s %s: Failed with HTTP code %d: %v", req.Method, req.URL.Path, result.code, result.err) + } + } +} + +func (m *MetricServer) verifyLoop(ctx context.Context) { + ticker := time.NewTicker(verifyLoopInterval) + defer ticker.Stop() + for ctx.Err() == nil { + select { + case <-ctx.Done(): + return + case <-m.shutdownCh: + log.Infof("Received interrupt signal, shutting down server.") + m.mu.Lock() + m.shutdownLocked(ctx) + 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() + }() + } + } +} + +// Execute implements subcommands.Command.Execute. +func (m *MetricServer) Execute(ctx context.Context, f *flag.FlagSet, args ...any) subcommands.ExitStatus { + ctx, ctxCancel := context.WithCancel(ctx) + defer ctxCancel() + + if f.NArg() != 0 { + f.Usage() + return subcommands.ExitUsageError + } + conf := args[0].(*config.Config) + if conf.MetricServer == "" || conf.RootDir == "" { + f.Usage() + return subcommands.ExitUsageError + } + if strings.Contains(conf.MetricServer, "%ID%") { + return util.Errorf("Metric server address contains '%%ID%%': %v. This should have been replaced by the parent process.", conf.MetricServer) + } + if _, err := container.ListSandboxes(conf.RootDir); err != nil { + return util.Errorf("Invalid root directory %q: tried to list sandboxes within it and got: %v", conf.RootDir, err) + } + m.startTime = time.Now() + m.rand = rand.New(rand.NewSource(m.startTime.UnixNano())) + m.rootDir = conf.RootDir + m.exporterPrefix = conf.MetricExporterPrefix + if strings.Contains(conf.MetricServer, "%RUNTIME_ROOT%") { + newAddr := strings.ReplaceAll(conf.MetricServer, "%RUNTIME_ROOT%", m.rootDir) + log.Infof("Metric server address replaced %RUNTIME_ROOT%: %q -> %q", conf.MetricServer, newAddr) + conf.MetricServer = newAddr + } + m.sandboxes = make(map[container.FullID]*servedSandbox) + + var listener net.Listener + var listenErr error + if strings.HasPrefix(conf.MetricServer, fmt.Sprintf("%c", os.PathSeparator)) { + beforeBindSt, beforeBindErr := os.Stat(conf.MetricServer) + if listener, listenErr = (&net.ListenConfig{}).Listen(ctx, "unix", conf.MetricServer); listenErr != nil { + return util.Errorf("Cannot listen on unix domain socket %q: %v", conf.MetricServer, listenErr) + } + afterBindSt, afterBindErr := os.Stat(conf.MetricServer) + if afterBindErr != nil { + return util.Errorf("Cannot stat our own unix domain socket %q: %v", conf.MetricServer, afterBindErr) + } + ownUDS := true + if beforeBindErr == nil && beforeBindSt.Mode() == afterBindSt.Mode() { + // Socket file existed and was a socket prior to us binding to it. + if beforeBindSt.Sys() != nil && afterBindSt.Sys() != nil { + beforeSt, beforeStOk := beforeBindSt.Sys().(*syscall.Stat_t) + afterSt, afterStOk := beforeBindSt.Sys().(*syscall.Stat_t) + if beforeStOk && afterStOk && beforeSt.Dev == afterSt.Dev && beforeSt.Ino == afterSt.Ino { + // Socket file is the same before and after binding, so we should not consider ourselves + // the owner of it. + ownUDS = false + } + } + } + if ownUDS { + log.Infof("Bound on socket file %s which we own. As such, this socket file will be deleted on server shutdown.", conf.MetricServer) + m.udsPath = conf.MetricServer + defer os.Remove(m.udsPath) + os.Chmod(m.udsPath, 0777) + } else { + log.Infof("Bound on socket file %s which existed prior to this server's existence. As such, it will not be deleted on server shutdown.", conf.MetricServer) + } + } else { + if strings.HasPrefix(conf.MetricServer, ":") { + log.Warningf("Binding on all interfaces. This will allow anyone to list all containers on your machine!") + } + if listener, listenErr = (&net.ListenConfig{}).Listen(ctx, "tcp", conf.MetricServer); listenErr != nil { + return util.Errorf("Cannot listen on TCP address %q: %v", conf.MetricServer, listenErr) + } + } + + mux := http.NewServeMux() + mux.HandleFunc("/runsc-metrics/healthcheck", logRequest(m.serveHealthCheck)) + mux.HandleFunc("/metrics", logRequest(m.serveMetrics)) + mux.HandleFunc("/", logRequest(m.serveIndex)) + m.srv.Handler = mux + m.srv.ReadTimeout = httpTimeout + m.srv.WriteTimeout = httpTimeout + m.shutdownCh = make(chan os.Signal, 1) + log.Infof("Server serving on %s for root directory %s.", conf.MetricServer, conf.RootDir) + + go m.verifyLoop(ctx) + signal.Notify(m.shutdownCh, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT) + serveErr := m.srv.Serve(listener) + log.Infof("Server has stopped accepting requests.") + m.mu.Lock() + defer m.mu.Unlock() + if serveErr != nil { + if serveErr == http.ErrServerClosed { + return subcommands.ExitSuccess + } + return util.Errorf("Cannot serve on address %s: %v", conf.MetricServer, serveErr) + } + // Per documentation, http.Server.Serve can never return a nil error, so this is not a success. + return util.Errorf("HTTP server Serve() did not return expected error") +} diff --git a/runsc/config/config.go b/runsc/config/config.go index f8023c63b..b0d3bb599 100644 --- a/runsc/config/config.go +++ b/runsc/config/config.go @@ -139,6 +139,17 @@ type Config struct { // If unset, a sane platform-specific default will be used. PlatformDevicePath string `flag:"platform_device_path"` + // MetricServer, if set, indicates that metrics should be exported on this address. + // This may either be 1) "addr:port" to export metrics on a specific network interface address, + // 2) ":port" for exporting metrics on all addresses, or 3) an absolute path to a Unix Domain + // Socket. + // The substring "%ID%" will be replaced by the container ID, and "%RUNTIME_ROOT%" by the root. + // This flag must be specified *both* as part of the `runsc metric-server` arguments (so that the + // metric server knows which address to bind to), and as part of the `runsc create` arguments (as + // an indication that the container being created wishes that its metrics should be exported). + // The value of this flag must also match across the two command lines. + MetricServer string `flag:"metric-server"` + // MetricExporterPrefix is added as prefix to all metric names. // It is used to follow Prometheus's exporter convention, whereby all metric names should be // prefixed by a name meaningfully identifying the software exporting the metric. diff --git a/runsc/config/flags.go b/runsc/config/flags.go index 3000b1c5b..1f3cbf6df 100644 --- a/runsc/config/flags.go +++ b/runsc/config/flags.go @@ -52,6 +52,7 @@ func RegisterFlags(flagSet *flag.FlagSet) { flagSet.String("traceback", "system", "golang runtime's traceback level") // Metrics flags. + flagSet.String("metric-server", "", "if set, export metrics on this address. This may either be 1) 'addr:port' to export metrics on a specific network interface address, 2) ':port' for exporting metrics on all interfaces, or 3) an absolute path to a Unix Domain Socket. The substring '%ID%' will be replaced by the container ID, and '%RUNTIME_ROOT%' by the root. This flag must be specified in both `runsc metric-server` and `runsc create`, and their values must match.") flagSet.String("metric-exporter-prefix", "runsc_", "prefix for all metric names, following Prometheus exporter convention") // Debugging flags: strace related diff --git a/runsc/container/container.go b/runsc/container/container.go index 632ed98e8..757ba7d5a 100644 --- a/runsc/container/container.go +++ b/runsc/container/container.go @@ -379,7 +379,7 @@ func New(conf *config.Config, args Args) (*Container, error) { func (c *Container) Start(conf *config.Config) error { log.Debugf("Start container, cid: %s", c.ID) - if err := c.Saver.lock(); err != nil { + if err := c.Saver.lock(BlockAcquire); err != nil { return err } unlock := cleanup.Make(c.Saver.UnlockOrDie) @@ -463,7 +463,7 @@ func (c *Container) Start(conf *config.Config) error { // to restore a container from its state file. func (c *Container) Restore(spec *specs.Spec, conf *config.Config, restoreFile string) error { log.Debugf("Restore container, cid: %s", c.ID) - if err := c.Saver.lock(); err != nil { + if err := c.Saver.lock(BlockAcquire); err != nil { return err } defer c.Saver.UnlockOrDie() @@ -649,7 +649,7 @@ func (c *Container) Checkpoint(f *os.File) error { // The call only succeeds if the container's status is created or running. func (c *Container) Pause() error { log.Debugf("Pausing container, cid: %s", c.ID) - if err := c.Saver.lock(); err != nil { + if err := c.Saver.lock(BlockAcquire); err != nil { return err } defer c.Saver.UnlockOrDie() @@ -669,7 +669,7 @@ func (c *Container) Pause() error { // The call only succeeds if the container's status is paused. func (c *Container) Resume() error { log.Debugf("Resuming container, cid: %s", c.ID) - if err := c.Saver.lock(); err != nil { + if err := c.Saver.lock(BlockAcquire); err != nil { return err } defer c.Saver.UnlockOrDie() @@ -710,7 +710,7 @@ func (c *Container) Processes() ([]*control.Process, error) { func (c *Container) Destroy() error { log.Debugf("Destroy container, cid: %s", c.ID) - if err := c.Saver.lock(); err != nil { + if err := c.Saver.lock(BlockAcquire); err != nil { return err } defer func() { diff --git a/runsc/container/container_test.go b/runsc/container/container_test.go index d57cb4830..6efce8949 100644 --- a/runsc/container/container_test.go +++ b/runsc/container/container_test.go @@ -2670,7 +2670,7 @@ func TestSaveSystemdCgroup(t *testing.T) { defer cont.Destroy() cont.CompatCgroup = cgroup.CgroupJSON{Cgroup: cgroup.CreateMockSystemdCgroup()} - if err := cont.Saver.lock(); err != nil { + if err := cont.Saver.lock(BlockAcquire); err != nil { t.Fatalf("cannot lock container metadata file: %v", err) } if err := cont.saveLocked(); err != nil { @@ -2678,7 +2678,7 @@ func TestSaveSystemdCgroup(t *testing.T) { } cont.Saver.unlock() loadCont := Container{} - cont.Saver.load(&loadCont) + cont.Saver.load(&loadCont, LoadOpts{}) if !reflect.DeepEqual(cont.CompatCgroup, loadCont.CompatCgroup) { t.Errorf("CompatCgroup not properly saved: want %v, got %v", cont.CompatCgroup, loadCont.CompatCgroup) } diff --git a/runsc/container/state_file.go b/runsc/container/state_file.go index 96eb63959..fc0fb05b7 100644 --- a/runsc/container/state_file.go +++ b/runsc/container/state_file.go @@ -16,6 +16,7 @@ package container import ( "encoding/json" + "errors" "fmt" "io/ioutil" "os" @@ -31,6 +32,21 @@ import ( const stateFileExtension = "state" +// ErrStateFileLocked is returned by Load() when the state file is locked +// and TryLock is enabled. +var ErrStateFileLocked = errors.New("state file locked") + +// TryLock represents whether we should block waiting for the lock to be acquired or not. +type TryLock bool + +const ( + // BlockAcquire means we will block until the lock can be acquired. + BlockAcquire TryLock = false + + // TryAcquire means we will fail fast if the lock cannot be acquired. + TryAcquire TryLock = true +) + // LoadOpts provides options for Load()ing a container. type LoadOpts struct { // Exact tells whether the search should be exact. See Load() for more. @@ -39,6 +55,11 @@ type LoadOpts struct { // SkipCheck tells Load() to skip checking if container is runnning. SkipCheck bool + // TryLock tells Load() to fail if the container state file cannot be locked, + // as opposed to blocking until it is available. + // When the state file cannot be locked, it will error with ErrStateFileLocked. + TryLock TryLock + // RootContainer when true matches the search only with the root container of // a sandbox. This is used when looking for a sandbox given that root // container and sandbox share the same ID. @@ -74,7 +95,7 @@ func Load(rootDir string, id FullID, opts LoadOpts) (*Container, error) { defer state.close() c := &Container{} - if err := state.load(c); err != nil { + if err := state.load(c, opts); err != nil { if os.IsNotExist(err) { // Preserve error so that callers can distinguish 'not found' errors. return nil, err @@ -272,13 +293,23 @@ type StateFile struct { } // lock globally locks all locking operations for the container. -func (s *StateFile) lock() error { +func (s *StateFile) lock(tryLock TryLock) error { s.once.Do(func() { s.flock = flock.New(s.lockPath()) }) - if err := s.flock.Lock(); err != nil { - return fmt.Errorf("acquiring lock on %q: %v", s.flock, err) + if tryLock { + gotLock, err := s.flock.TryLock() + if err != nil { + return fmt.Errorf("acquiring lock on %q: %v", s.flock, err) + } + if !gotLock { + return ErrStateFileLocked + } + } else { + if err := s.flock.Lock(); err != nil { + return fmt.Errorf("acquiring lock on %q: %v", s.flock, err) + } } return nil } @@ -287,7 +318,7 @@ func (s *StateFile) lock() error { // is done to ensure that more than one creation didn't race to create // containers with the same ID. func (s *StateFile) LockForNew() error { - if err := s.lock(); err != nil { + if err := s.lock(BlockAcquire); err != nil { return err } @@ -327,7 +358,7 @@ func (s *StateFile) UnlockOrDie() { // SaveLocked saves 'v' to the state file. // -// Preconditions: lock() must been called before. +// Preconditions: lock(*) must been called before. func (s *StateFile) SaveLocked(v any) error { if !s.flock.Locked() { panic("saveLocked called without lock held") @@ -343,8 +374,8 @@ func (s *StateFile) SaveLocked(v any) error { return nil } -func (s *StateFile) load(v any) error { - if err := s.lock(); err != nil { +func (s *StateFile) load(v any, opts LoadOpts) error { + if err := s.lock(opts.TryLock); err != nil { return err } defer s.UnlockOrDie() diff --git a/runsc/sandbox/BUILD b/runsc/sandbox/BUILD index 6f0420f93..975b4cff4 100644 --- a/runsc/sandbox/BUILD +++ b/runsc/sandbox/BUILD @@ -23,6 +23,7 @@ go_library( "//pkg/control/server", "//pkg/coverage", "//pkg/log", + "//pkg/metric:metric_go_proto", "//pkg/prometheus", "//pkg/sentry/control", "//pkg/sentry/platform", @@ -44,6 +45,7 @@ go_library( "@com_github_opencontainers_runtime_spec//specs-go:go_default_library", "@com_github_syndtr_gocapability//capability:go_default_library", "@com_github_vishvananda_netlink//:go_default_library", + "@org_golang_google_protobuf//encoding/prototext:go_default_library", "@org_golang_x_sys//unix:go_default_library", ], ) diff --git a/runsc/sandbox/sandbox.go b/runsc/sandbox/sandbox.go index 5e9b855e2..7ed4d665b 100644 --- a/runsc/sandbox/sandbox.go +++ b/runsc/sandbox/sandbox.go @@ -33,12 +33,14 @@ import ( specs "github.com/opencontainers/runtime-spec/specs-go" "github.com/syndtr/gocapability/capability" "golang.org/x/sys/unix" + "google.golang.org/protobuf/encoding/prototext" "gvisor.dev/gvisor/pkg/atomicbitops" "gvisor.dev/gvisor/pkg/cleanup" "gvisor.dev/gvisor/pkg/control/client" "gvisor.dev/gvisor/pkg/control/server" "gvisor.dev/gvisor/pkg/coverage" "gvisor.dev/gvisor/pkg/log" + metricpb "gvisor.dev/gvisor/pkg/metric/metric_go_proto" "gvisor.dev/gvisor/pkg/prometheus" "gvisor.dev/gvisor/pkg/sentry/control" "gvisor.dev/gvisor/pkg/sentry/platform" @@ -54,6 +56,16 @@ import ( "gvisor.dev/gvisor/runsc/specutils" ) +const ( + // namespaceAnnotation is a pod annotation populated by containerd. + // It contains the name of the pod that a sandbox is in when running in Kubernetes. + podNameAnnotation = "io.kubernetes.cri.sandbox-name" + + // namespaceAnnotation is a pod annotation populated by containerd. + // It contains the namespace of the pod that a sandbox is in when running in Kubernetes. + namespaceAnnotation = "io.kubernetes.cri.sandbox-namespace" +) + // pid is an atomic type that implements JSON marshal/unmarshal interfaces. type pid struct { val atomicbitops.Int64 @@ -96,6 +108,14 @@ type Sandbox struct { // ID as the first container run in the sandbox. ID string `json:"id"` + // PodName is the name of the Kubernetes Pod (if any) that this sandbox + // represents. Unset if not running under containerd or Kubernetes. + PodName string `json:"podName"` + + // Namespace is the Kubernetes namespace (if any) of the pod that this + // sandbox represents. Unset if not running under containerd or Kubernetes. + Namespace string `json:"namespace"` + // Pid is the pid of the running sandbox. May be 0 if the sandbox // is not running. Pid pid `json:"pid"` @@ -113,6 +133,13 @@ type Sandbox struct { // started, before it may be modified. OriginalOOMScoreAdj int `json:"originalOomScoreAdj"` + // RegisteredMetrics is the set of metrics registered in the sandbox. + // Used for verifying metric data integrity after containers are started. + // Only populated if exporting metrics was requested when the sandbox was + // created. + // This is a textproto string of metricpb.MetricRegistration. + RegisteredMetrics string `json:"registeredMetrics"` + // 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 @@ -189,6 +216,10 @@ func New(conf *config.Config, args *Args) (*Sandbox, error) { UID: -1, // prevent usage before it's set. GID: -1, // prevent usage before it's set. } + if args.Spec != nil && args.Spec.Annotations != nil { + s.PodName = args.Spec.Annotations[podNameAnnotation] + s.Namespace = args.Spec.Annotations[namespaceAnnotation] + } // The Cleanup object cleans up partially created sandboxes when an error // occurs. Any errors occurring during cleanup itself are ignored. c := cleanup.Make(func() { @@ -1072,7 +1103,23 @@ func (s *Sandbox) UsageFD() (*control.MemoryUsageRecord, error) { return control.NewMemoryUsageRecord(*m.FilePayload.Files[0], *m.FilePayload.Files[1]) } -// ExportMetrics writes Prometheus-formatted metrics data to the given io.Writer. +// GetRegisteredMetrics returns metric registration data from the sandbox. +// This data is meant to be used as a way to sanity-check any exported metrics data during the +// lifetime of the sandbox in order to avoid a compromised sandbox from being able to produce +// bogus metrics. +// This returns an error if the sandbox has not requested instrumentation during creation time. +func (s *Sandbox) GetRegisteredMetrics() (*metricpb.MetricRegistration, error) { + if s.RegisteredMetrics == "" { + return nil, errors.New("sandbox did not request instrumentation when it was created") + } + registeredMetrics := &metricpb.MetricRegistration{} + if err := prototext.Unmarshal([]byte(s.RegisteredMetrics), registeredMetrics); err != nil { + return nil, err + } + return registeredMetrics, nil +} + +// ExportMetrics returns a snapshot of metric values from the sandbox in Prometheus format. func (s *Sandbox) ExportMetrics() (*prometheus.Snapshot, error) { log.Debugf("Metrics export sandbox %q", s.ID) data := &control.MetricsExportData{} @@ -1082,6 +1129,26 @@ func (s *Sandbox) ExportMetrics() (*prometheus.Snapshot, error) { return data.Snapshot, nil } +// Label names used to identify each sandbox. +const ( + SandboxIDLabel = "sandbox" + PodNameLabel = "pod" + NamespaceLabel = "namespace" +) + +// PrometheusLabels returns a set of Prometheus labels that identifies the sandbox. +func (s *Sandbox) PrometheusLabels() map[string]string { + labels := make(map[string]string, 3) + labels[SandboxIDLabel] = s.ID + if s.PodName != "" { + labels[PodNameLabel] = s.PodName + } + if s.Namespace != "" { + labels[NamespaceLabel] = s.Namespace + } + return labels +} + // IsRunning returns true if the sandbox or gofer process is running. func (s *Sandbox) IsRunning() bool { pid := s.Pid.load()