stress_server_linux.go

Documentation: github.com/chaos-mesh/chaos-mesh/pkg/chaosdaemon

     1  // Copyright 2021 Chaos Mesh Authors.
     2  //
     3  // Licensed under the Apache License, Version 2.0 (the "License");
     4  // you may not use this file except in compliance with the License.
     5  // You may obtain a copy of the License at
     6  //
     7  // http://www.apache.org/licenses/LICENSE-2.0
     8  //
     9  // Unless required by applicable law or agreed to in writing, software
    10  // distributed under the License is distributed on an "AS IS" BASIS,
    11  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12  // See the License for the specific language governing permissions and
    13  // limitations under the License.
    14  //
    15  
    16  package chaosdaemon
    17  
    18  import (
    19  	"context"
    20  	"strconv"
    21  	"strings"
    22  	"syscall"
    23  	"time"
    24  
    25  	"github.com/golang/protobuf/ptypes/empty"
    26  
    27  	"github.com/chaos-mesh/chaos-mesh/pkg/bpm"
    28  	"github.com/chaos-mesh/chaos-mesh/pkg/chaosdaemon/cgroups"
    29  	pb "github.com/chaos-mesh/chaos-mesh/pkg/chaosdaemon/pb"
    30  	"github.com/chaos-mesh/chaos-mesh/pkg/chaosdaemon/util"
    31  )
    32  
    33  func (s *DaemonServer) ExecStressors(ctx context.Context,
    34  	req *pb.ExecStressRequest) (*pb.ExecStressResponse, error) {
    35  	log := s.getLoggerFromContext(ctx)
    36  	log.Info("Executing stressors", "request", req)
    37  
    38  	// cpuStressors
    39  	cpuProc, err := s.ExecCPUStressors(ctx, req)
    40  	if err != nil {
    41  		// this error would be resolved by grpc server framework, it's the top level to print it.
    42  		s.rootLogger.Error(err, "exec cpu stressors", "containerID", req.Target, "cpuStressors", req.CpuStressors)
    43  		return nil, err
    44  	}
    45  
    46  	// memoryStressor
    47  	memoryProc, err := s.ExecMemoryStressors(ctx, req)
    48  	if err != nil {
    49  		return nil, err
    50  	}
    51  
    52  	resp := new(pb.ExecStressResponse)
    53  	if cpuProc != nil {
    54  		resp.CpuInstance = strconv.Itoa(cpuProc.Pair.Pid)
    55  		resp.CpuStartTime = cpuProc.Pair.CreateTime
    56  		resp.CpuInstanceUid = cpuProc.Uid
    57  	}
    58  	if memoryProc != nil {
    59  		resp.MemoryInstance = strconv.Itoa(memoryProc.Pair.Pid)
    60  		resp.MemoryStartTime = memoryProc.Pair.CreateTime
    61  		resp.MemoryInstanceUid = memoryProc.Uid
    62  	}
    63  
    64  	return resp, nil
    65  }
    66  
    67  func (s *DaemonServer) CancelStressors(ctx context.Context,
    68  	req *pb.CancelStressRequest) (*empty.Empty, error) {
    69  	log := s.getLoggerFromContext(ctx)
    70  	CpuPid, err := strconv.Atoi(req.CpuInstance)
    71  	if req.CpuInstance != "" && err != nil {
    72  		return nil, err
    73  	}
    74  
    75  	MemoryPid, err := strconv.Atoi(req.MemoryInstance)
    76  	if req.MemoryInstance != "" && err != nil {
    77  		return nil, err
    78  	}
    79  
    80  	if req.CpuInstanceUid == "" && CpuPid != 0 {
    81  		if uid, ok := s.backgroundProcessManager.GetUID(bpm.ProcessPair{Pid: CpuPid, CreateTime: req.CpuStartTime}); ok {
    82  			req.CpuInstanceUid = uid
    83  		}
    84  	}
    85  
    86  	if req.MemoryInstanceUid == "" && MemoryPid != 0 {
    87  		if uid, ok := s.backgroundProcessManager.GetUID(bpm.ProcessPair{Pid: MemoryPid, CreateTime: req.MemoryStartTime}); ok {
    88  			req.MemoryInstanceUid = uid
    89  		}
    90  	}
    91  
    92  	log.Info("Canceling stressors", "request", req)
    93  
    94  	if req.CpuInstanceUid != "" {
    95  		err = s.backgroundProcessManager.KillBackgroundProcess(ctx, req.CpuInstanceUid)
    96  		if err != nil {
    97  			return nil, err
    98  		}
    99  	}
   100  
   101  	if req.MemoryInstanceUid != "" {
   102  		err = s.backgroundProcessManager.KillBackgroundProcess(ctx, req.MemoryInstanceUid)
   103  		if err != nil {
   104  			return nil, err
   105  		}
   106  	}
   107  
   108  	log.Info("killing stressor successfully")
   109  	return &empty.Empty{}, nil
   110  }
   111  
   112  func (s *DaemonServer) ExecCPUStressors(ctx context.Context,
   113  	req *pb.ExecStressRequest) (*bpm.Process, error) {
   114  	log := s.getLoggerFromContext(ctx)
   115  	if req.CpuStressors == "" {
   116  		return nil, nil
   117  	}
   118  	pid, err := s.crClient.GetPidFromContainerID(ctx, req.Target)
   119  	if err != nil {
   120  		return nil, err
   121  	}
   122  
   123  	attachCGroup, err := cgroups.GetAttacherForPID(int(pid))
   124  	if err != nil {
   125  		return nil, err
   126  	}
   127  
   128  	processBuilder := bpm.DefaultProcessBuilder("stress-ng", strings.Fields(req.CpuStressors)...).
   129  		EnablePause()
   130  	if req.EnterNS {
   131  		processBuilder = processBuilder.SetNS(pid, bpm.PidNS)
   132  	}
   133  	cmd := processBuilder.Build(ctx)
   134  
   135  	proc, err := s.backgroundProcessManager.StartProcess(ctx, cmd)
   136  	if err != nil {
   137  		return nil, err
   138  	}
   139  	log.Info("Start stress-ng successfully", "command", cmd, "pid", proc.Pair.Pid, "uid", proc.Uid)
   140  
   141  	if err = attachCGroup.AttachProcess(proc.Pair.Pid); err != nil {
   142  		if kerr := cmd.Process.Kill(); kerr != nil {
   143  			log.Error(kerr, "kill stress-ng failed", "request", req)
   144  		}
   145  		return nil, err
   146  	}
   147  
   148  	for {
   149  		// TODO: find a better way to resume pause process
   150  		if err := cmd.Process.Signal(syscall.SIGCONT); err != nil {
   151  			return nil, err
   152  		}
   153  
   154  		log.Info("send signal to resume process")
   155  		time.Sleep(time.Millisecond)
   156  
   157  		// TODO: integrate the resume process into the bpm
   158  		comm, err := util.ReadCommName(cmd.Process.Pid)
   159  		if err != nil {
   160  			return nil, err
   161  		}
   162  		if comm != "pause\n" {
   163  			log.Info("pause has been resumed", "comm", comm)
   164  			break
   165  		}
   166  		log.Info("the process hasn't resumed, step into the following loop", "comm", comm)
   167  	}
   168  
   169  	return proc, nil
   170  }
   171  
   172  func (s *DaemonServer) ExecMemoryStressors(ctx context.Context,
   173  	req *pb.ExecStressRequest) (*bpm.Process, error) {
   174  	log := s.getLoggerFromContext(ctx)
   175  	if req.MemoryStressors == "" {
   176  		return nil, nil
   177  	}
   178  	pid, err := s.crClient.GetPidFromContainerID(ctx, req.Target)
   179  	if err != nil {
   180  		return nil, err
   181  	}
   182  
   183  	attachCGroup, err := cgroups.GetAttacherForPID(int(pid))
   184  	if err != nil {
   185  		return nil, err
   186  	}
   187  
   188  	processBuilder := bpm.DefaultProcessBuilder("memStress", strings.Fields(req.MemoryStressors)...).
   189  		EnablePause()
   190  
   191  	if req.OomScoreAdj != 0 {
   192  		processBuilder = processBuilder.SetOOMScoreAdj(int(req.OomScoreAdj))
   193  	}
   194  	if req.EnterNS {
   195  		processBuilder = processBuilder.SetNS(pid, bpm.PidNS)
   196  	}
   197  	cmd := processBuilder.Build(ctx)
   198  
   199  	proc, err := s.backgroundProcessManager.StartProcess(ctx, cmd)
   200  	if err != nil {
   201  		return nil, err
   202  	}
   203  	log.Info("Start memStress successfully", "command", cmd, "pid", proc.Pair.Pid, "uid", proc.Uid)
   204  
   205  	if err = attachCGroup.AttachProcess(proc.Pair.Pid); err != nil {
   206  		if kerr := cmd.Process.Kill(); kerr != nil {
   207  			log.Error(kerr, "kill memStress failed", "request", req)
   208  		}
   209  		return nil, err
   210  	}
   211  
   212  	for {
   213  		// TODO: find a better way to resume pause process
   214  		if err := cmd.Process.Signal(syscall.SIGCONT); err != nil {
   215  			return nil, err
   216  		}
   217  
   218  		log.Info("send signal to resume process")
   219  		time.Sleep(time.Millisecond)
   220  		comm, err := util.ReadCommName(proc.Pair.Pid)
   221  
   222  		if err != nil {
   223  			return nil, err
   224  		}
   225  		if comm != "pause\n" {
   226  			log.Info("pause has been resumed", "comm", comm)
   227  			break
   228  		}
   229  		log.Info("the process hasn't resumed, step into the following loop", "comm", comm)
   230  	}
   231  
   232  	return proc, nil
   233  }
   234