multi: create and provide cgroup manager

Create and provide cgroup manager via envelope for all the weavelets

Signed-off-by: flouthoc <flouthoc.git@gmail.com>

internal/tool/multi/deployer.go

@@ -44,6 +44,7 @@ import (
 	"github.com/ServiceWeaver/weaver/runtime/profiling"
 	"github.com/ServiceWeaver/weaver/runtime/protos"
 	"github.com/ServiceWeaver/weaver/runtime/traces"
+	cgroups "github.com/containerd/cgroups/v3/cgroup2"
 	"github.com/google/uuid"
 	"golang.org/x/exp/maps"
 	"golang.org/x/sync/errgroup"
@@ -55,19 +56,20 @@ const defaultReplication = 2
 
 // A deployer manages an application deployment.
 type deployer struct {
-	ctx          context.Context
-	ctxCancel    context.CancelFunc
-	deploymentId string
-	tmpDir       string // Private directory for this weavelet/envelope
-	config       *MultiConfig
-	started      time.Time
-	logger       *slog.Logger
-	caCert       *x509.Certificate
-	caKey        crypto.PrivateKey
-	running      errgroup.Group
-	logsDB       *logging.FileStore
-	printer      *logging.PrettyPrinter
-	traceDB      *traces.DB
+	ctx           context.Context
+	ctxCancel     context.CancelFunc
+	deploymentId  string
+	tmpDir        string // Private directory for this weavelet/envelope
+	config        *MultiConfig
+	started       time.Time
+	logger        *slog.Logger
+	caCert        *x509.Certificate
+	caKey         crypto.PrivateKey
+	running       errgroup.Group
+	logsDB        *logging.FileStore
+	printer       *logging.PrettyPrinter
+	traceDB       *traces.DB
+	cgroupManager *cgroups.Manager
 
 	// statsProcessor tracks and computes stats to be rendered on the /statusz page.
 	statsProcessor *imetrics.StatsProcessor
@@ -109,6 +111,33 @@ type handler struct {
 
 var _ envelope.EnvelopeHandler = &handler{}
 
+// Ofcourse this will read and parse values from config in final
+// version of this PR.
+func createCgroupManager() (*cgroups.Manager, error) {
+	// Note: In final PR these parmeters will be read and
+	// parsed from the config.
+	group := "/service-weaver-cgroup"
+	groupPath := fmt.Sprintf("%s-%d", group, os.Getpid())
+	var (
+		quota  int64  = 5000
+		period uint64 = 100000
+		weight uint64 = 100
+	)
+	res := cgroups.Resources{
+		CPU: &cgroups.CPU{
+			Weight: &weight,
+			Max:    cgroups.NewCPUMax(&quota, &period),
+			Cpus:   "0",
+			Mems:   "0",
+		},
+	}
+	cgroupManager, err := cgroups.NewManager("/sys/fs/cgroup", groupPath, &res)
+	if err != nil {
+		return nil, fmt.Errorf("Unable to create a new cgroup: %w", err)
+	}
+	return cgroupManager, nil
+}
+
 // newDeployer creates a new deployer. The deployer can be stopped at any
 // time by canceling the passed-in context.
 func newDeployer(ctx context.Context, deploymentId string, config *MultiConfig, tmpDir string) (*deployer, error) {
@@ -136,6 +165,11 @@ func newDeployer(ctx context.Context, deploymentId string, config *MultiConfig,
 		}
 	}
 
+	cgroupManager, err := createCgroupManager()
+	if err != nil {
+		return nil, err
+	}
+
 	// Create the trace saver.
 	traceDB, err := traces.OpenDB(ctx, perfettoFile)
 	if err != nil {
@@ -157,6 +191,7 @@ func newDeployer(ctx context.Context, deploymentId string, config *MultiConfig,
 		deploymentId:   deploymentId,
 		config:         config,
 		started:        time.Now(),
+		cgroupManager:  cgroupManager,
 		proxies:        map[string]*proxyInfo{},
 	}
 
@@ -286,7 +321,10 @@ func (d *deployer) stop(err error) {
 		d.err = err
 	}
 	d.mu.Unlock()
-
+	// Remove any cgroups if created
+	if d.cgroupManager != nil {
+		d.cgroupManager.Delete()
+	}
 	// Cancel the context.
 	d.ctxCancel()
 }
@@ -330,7 +368,8 @@ func (d *deployer) startColocationGroup(g *group) error {
 			InternalAddress: "localhost:0",
 		}
 		e, err := envelope.NewEnvelope(d.ctx, info, d.config.App, envelope.Options{
-			Logger: d.logger,
+			Logger:  d.logger,
+			Manager: d.cgroupManager,
 		})
 		if err != nil {
 			return err