[ws-daemon] Integrate new CPU limiter

Author: csweichel

components/ws-daemon/pkg/cpulimit/cfs.go

@@ -0,0 +1,103 @@
+// Copyright (c) 2022 Gitpod GmbH. All rights reserved.
+// Licensed under the GNU Affero General Public License (AGPL).
+// See License-AGPL.txt in the project root for license information.
+
+package cpulimit
+
+import (
+	"bufio"
+	"math"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"golang.org/x/xerrors"
+)
+
+// CgroupCFSController controls a cgroup's CFS settings
+type CgroupCFSController string
+
+// Usage returns the cpuacct.usage value of the cgroup
+func (basePath CgroupCFSController) Usage() (usage CPUTime, err error) {
+
+	cpuTimeInNS, err := basePath.readUint64("cpuacct.usage")
+	if err != nil {
+		return 0, xerrors.Errorf("cannot read cpuacct.usage: %w", err)
+	}
+
+	return CPUTime(time.Duration(cpuTimeInNS) * time.Nanosecond), nil
+}
+
+// SetQuota sets a new CFS quota on the cgroup
+func (basePath CgroupCFSController) SetLimit(limit Bandwidth) (changed bool, err error) {
+	p, err := basePath.readUint64("cpu.cfs_period_us")
+	if err != nil {
+		err = xerrors.Errorf("cannot parse CFS period: %w", err)
+		return
+	}
+	period := time.Duration(p) * time.Microsecond
+
+	q, err := basePath.readUint64("cpu.cfs_quota_us")
+	if err != nil {
+		err = xerrors.Errorf("cannot parse CFS quota: %w", err)
+		return
+	}
+	quota := time.Duration(q) * time.Microsecond
+	target := limit.Quota(period)
+	if quota == target {
+		return false, nil
+	}
+
+	err = os.WriteFile(filepath.Join(string(basePath), "cpu.cfs_quota_us"), []byte(strconv.FormatInt(target.Microseconds(), 10)), 0644)
+	if err != nil {
+		return false, xerrors.Errorf("cannot set CFS quota: %w", err)
+	}
+	return true, nil
+}
+
+func (basePath CgroupCFSController) readUint64(path string) (uint64, error) {
+	fn := filepath.Join(string(basePath), path)
+	fc, err := os.ReadFile(fn)
+	if err != nil {
+		return 0, err
+	}
+
+	s := strings.TrimSpace(string(fc))
+	if s == "max" {
+		return math.MaxUint64, nil
+	}
+
+	p, err := strconv.ParseInt(s, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return uint64(p), nil
+}
+
+// NrThrottled returns the number of CFS periods the cgroup was throttled in
+func (basePath CgroupCFSController) NrThrottled() (uint64, error) {
+	f, err := os.Open(filepath.Join(string(basePath), "cpu.stat"))
+	if err != nil {
+		return 0, xerrors.Errorf("cannot read cpu.stat: %w", err)
+	}
+	defer f.Close()
+
+	const prefixNrThrottled = "nr_throttled "
+
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		l := scanner.Text()
+		if !strings.HasPrefix(l, prefixNrThrottled) {
+			continue
+		}
+
+		r, err := strconv.ParseInt(strings.TrimSpace(strings.TrimPrefix(l, prefixNrThrottled)), 10, 64)
+		if err != nil {
+			return 0, xerrors.Errorf("cannot parse cpu.stat: %s: %w", l, err)
+		}
+		return uint64(r), nil
+	}
+	return 0, xerrors.Errorf("cpu.stat did not contain nr_throttled")
+}

components/ws-daemon/pkg/cpulimit/cpulimit.go

@@ -4,12 +4,6 @@
 
 package cpulimit
 
-// TODO(cw):
-//    - introduce metrics that report
-// 			- total throttled period count (per QoS)
-//			- total bandwidth granted (per QoS)
-//			- something that corresponds to the user experience
-
 import (
 	"context"
 	"sort"
@@ -37,7 +31,7 @@ type WorkspaceHistory struct {
 
 func (h *WorkspaceHistory) Usage() CPUTime {
 	if h == nil || h.LastUpdate == nil {
-		panic("usage called before update")
+		return 0
 	}
 	return h.LastUpdate.Usage - h.UsageT0
 }
@@ -60,26 +54,26 @@ func (h *WorkspaceHistory) Throttled() bool {
 }
 
 type DistributorSource func(context.Context) ([]Workspace, error)
-type DistributorSink func(workspaceID string, limit Bandwidth)
+type DistributorSink func(id string, limit Bandwidth, burst bool)
 
-func NewDistributor(source DistributorSource, sink DistributorSink, limiter ResourceLimiter, breakoutLimiter ResourceLimiter, totalBandwidth Bandwidth) *Distributor {
+func NewDistributor(source DistributorSource, sink DistributorSink, limiter ResourceLimiter, burstLimiter ResourceLimiter, totalBandwidth Bandwidth) *Distributor {
 	return &Distributor{
-		Source:          source,
-		Sink:            sink,
-		Limiter:         limiter,
-		BreakoutLimiter: breakoutLimiter,
-		TotalBandwidth:  totalBandwidth,
-		History:         make(map[string]*WorkspaceHistory),
+		Source:         source,
+		Sink:           sink,
+		Limiter:        limiter,
+		BurstLimiter:   burstLimiter,
+		TotalBandwidth: totalBandwidth,
+		History:        make(map[string]*WorkspaceHistory),
 	}
 }
 
 type Distributor struct {
 	Source DistributorSource
 	Sink   DistributorSink
 
-	History         map[string]*WorkspaceHistory
-	Limiter         ResourceLimiter
-	BreakoutLimiter ResourceLimiter
+	History      map[string]*WorkspaceHistory
+	Limiter      ResourceLimiter
+	BurstLimiter ResourceLimiter
 
 	// TotalBandwidth is the total CPU time available in nanoseconds per second
 	TotalBandwidth Bandwidth
@@ -90,7 +84,7 @@ type Distributor struct {
 }
 
 type DistributorDebug struct {
-	BandwidthAvail, BandwidthUsed, BandwidthBreakout Bandwidth
+	BandwidthAvail, BandwidthUsed, BandwidthBurst Bandwidth
 }
 
 // Run starts a ticker which repeatedly calls Tick until the context is canceled.
@@ -175,7 +169,7 @@ func (d *Distributor) Tick(dt time.Duration) (DistributorDebug, error) {
 	d.LastTickUsage = totalUsage
 
 	// enforce limits
-	var breakoutBandwidth Bandwidth
+	var burstBandwidth Bandwidth
 	for _, id := range wsOrder {
 		ws := d.History[id]
 		limit := d.Limiter.Limit(ws.Usage())
@@ -184,24 +178,25 @@ func (d *Distributor) Tick(dt time.Duration) (DistributorDebug, error) {
 		// and there's still some bandwidth left to give, let's act as if had
 		// never spent any CPU time and assume the workspace will spend their
 		// entire bandwidth at once.
+		var burst bool
 		if totalBandwidth < d.TotalBandwidth && ws.Throttled() {
-			limit = d.BreakoutLimiter.Limit(ws.Usage())
+			limit = d.BurstLimiter.Limit(ws.Usage())
 
 			// We assume the workspace is going to use as much as their limit allows.
 			// This might not be true, because their process which consumed so much CPU
 			// may have ended by now.
 			totalBandwidth += limit
 
-			breakoutBandwidth += limit
+			burstBandwidth += limit
 		}
 
-		d.Sink(id, limit)
+		d.Sink(id, limit, burst)
 	}
 
 	return DistributorDebug{
-		BandwidthAvail:    d.TotalBandwidth,
-		BandwidthUsed:     totalBandwidth,
-		BandwidthBreakout: breakoutBandwidth,
+		BandwidthAvail: d.TotalBandwidth,
+		BandwidthUsed:  totalBandwidth,
+		BandwidthBurst: burstBandwidth,
 	}, nil
 }
 

components/ws-daemon/pkg/cpulimit/cpulimit_test.go

@@ -221,7 +221,7 @@ func (n *Node) Source(context.Context) ([]cpulimit.Workspace, error) {
 }
 
 // Sink acts as sink for a distributor
-func (n *Node) Sink(id string, limit cpulimit.Bandwidth) {
+func (n *Node) Sink(id string, limit cpulimit.Bandwidth, burst bool) {
 	n.State[id].Limit = limit
 }
 
@@ -238,7 +238,7 @@ func (n *Node) Dump(out io.Writer, t time.Duration, dbg cpulimit.DistributorDebu
 		if actualRate > limit {
 			actualRate = limit
 		}
-		fmt.Fprintf(out, "%d,%s,%d,%d,%d,%d,%d,%d,%d,%d,%d\n", t, c.ID(), c.Rate(t), state.Throttled, time.Duration(state.Usage).Milliseconds(), state.Limit, actualRate, totalCapacity, n.bandwidthUsed, n.bandwidthReq, dbg.BandwidthBreakout)
+		fmt.Fprintf(out, "%d,%s,%d,%d,%d,%d,%d,%d,%d,%d,%d\n", t, c.ID(), c.Rate(t), state.Throttled, time.Duration(state.Usage).Milliseconds(), state.Limit, actualRate, totalCapacity, n.bandwidthUsed, n.bandwidthReq, dbg.BandwidthBurst)
 	}
 }