service_internal.go

Documentation: github.com/Microsoft/hcsshim/cmd/containerd-shim-runhcs-v1

     1  //go:build windows
     2  
     3  package main
     4  
     5  import (
     6  	"context"
     7  	"encoding/json"
     8  	"fmt"
     9  	"os"
    10  	"path/filepath"
    11  	"strings"
    12  
    13  	runhcsopts "github.com/Microsoft/hcsshim/cmd/containerd-shim-runhcs-v1/options"
    14  	"github.com/Microsoft/hcsshim/internal/extendedtask"
    15  	"github.com/Microsoft/hcsshim/internal/oci"
    16  	"github.com/Microsoft/hcsshim/internal/shimdiag"
    17  	containerd_v1_types "github.com/containerd/containerd/api/types/task"
    18  	"github.com/containerd/containerd/errdefs"
    19  	"github.com/containerd/containerd/mount"
    20  	"github.com/containerd/containerd/runtime/v2/task"
    21  	"github.com/containerd/typeurl"
    22  	google_protobuf1 "github.com/gogo/protobuf/types"
    23  	"github.com/opencontainers/runtime-spec/specs-go"
    24  	"github.com/pkg/errors"
    25  )
    26  
    27  var empty = &google_protobuf1.Empty{}
    28  
    29  // getPod returns the pod this shim is tracking or else returns `nil`. It is the
    30  // callers responsibility to verify that `s.isSandbox == true` before calling
    31  // this method.
    32  //
    33  // If `pod==nil` returns `errdefs.ErrFailedPrecondition`.
    34  func (s *service) getPod() (shimPod, error) {
    35  	raw := s.taskOrPod.Load()
    36  	if raw == nil {
    37  		return nil, errors.Wrapf(errdefs.ErrFailedPrecondition, "task with id: '%s' must be created first", s.tid)
    38  	}
    39  	return raw.(shimPod), nil
    40  }
    41  
    42  // getTask returns a task matching `tid` or else returns `nil`. This properly
    43  // handles a task in a pod or a singular task shim.
    44  //
    45  // If `tid` is not found will return `errdefs.ErrNotFound`.
    46  func (s *service) getTask(tid string) (shimTask, error) {
    47  	raw := s.taskOrPod.Load()
    48  	if raw == nil {
    49  		return nil, errors.Wrapf(errdefs.ErrNotFound, "task with id: '%s' not found", tid)
    50  	}
    51  	if s.isSandbox {
    52  		p := raw.(shimPod)
    53  		return p.GetTask(tid)
    54  	}
    55  	// When its not a sandbox only the init task is a valid id.
    56  	if s.tid != tid {
    57  		return nil, errors.Wrapf(errdefs.ErrNotFound, "task with id: '%s' not found", tid)
    58  	}
    59  	return raw.(shimTask), nil
    60  }
    61  
    62  func (s *service) stateInternal(ctx context.Context, req *task.StateRequest) (*task.StateResponse, error) {
    63  	t, err := s.getTask(req.ID)
    64  	if err != nil {
    65  		return nil, err
    66  	}
    67  	e, err := t.GetExec(req.ExecID)
    68  	if err != nil {
    69  		return nil, err
    70  	}
    71  	return e.Status(), nil
    72  }
    73  
    74  func (s *service) createInternal(ctx context.Context, req *task.CreateTaskRequest) (*task.CreateTaskResponse, error) {
    75  	setupDebuggerEvent()
    76  
    77  	shimOpts := &runhcsopts.Options{}
    78  	if req.Options != nil {
    79  		v, err := typeurl.UnmarshalAny(req.Options)
    80  		if err != nil {
    81  			return nil, err
    82  		}
    83  		shimOpts = v.(*runhcsopts.Options)
    84  	}
    85  
    86  	var spec specs.Spec
    87  	f, err := os.Open(filepath.Join(req.Bundle, "config.json"))
    88  	if err != nil {
    89  		return nil, err
    90  	}
    91  	if err := json.NewDecoder(f).Decode(&spec); err != nil {
    92  		f.Close()
    93  		return nil, err
    94  	}
    95  	f.Close()
    96  
    97  	spec = oci.UpdateSpecFromOptions(spec, shimOpts)
    98  	//expand annotations after defaults have been loaded in from options
    99  	err = oci.ProcessAnnotations(ctx, &spec)
   100  	// since annotation expansion is used to toggle security features
   101  	// raise it rather than suppress and move on
   102  	if err != nil {
   103  		return nil, errors.Wrap(err, "unable to process OCI Spec annotations")
   104  	}
   105  
   106  	// If sandbox isolation is set to hypervisor, make sure the HyperV option
   107  	// is filled in. This lessens the burden on Containerd to parse our shims
   108  	// options if we can set this ourselves.
   109  	if shimOpts.SandboxIsolation == runhcsopts.Options_HYPERVISOR {
   110  		if spec.Windows == nil {
   111  			spec.Windows = &specs.Windows{}
   112  		}
   113  		if spec.Windows.HyperV == nil {
   114  			spec.Windows.HyperV = &specs.WindowsHyperV{}
   115  		}
   116  	}
   117  
   118  	if len(req.Rootfs) == 0 {
   119  		// If no mounts are passed via the snapshotter its the callers full
   120  		// responsibility to manage the storage. Just move on without affecting
   121  		// the config.json at all.
   122  		if spec.Windows == nil || len(spec.Windows.LayerFolders) < 2 {
   123  			return nil, errors.Wrap(errdefs.ErrFailedPrecondition, "no Windows.LayerFolders found in oci spec")
   124  		}
   125  	} else if len(req.Rootfs) != 1 {
   126  		return nil, errors.Wrap(errdefs.ErrFailedPrecondition, "Rootfs does not contain exactly 1 mount for the root file system")
   127  	} else {
   128  		m := req.Rootfs[0]
   129  		if m.Type != "windows-layer" && m.Type != "lcow-layer" {
   130  			return nil, errors.Wrapf(errdefs.ErrFailedPrecondition, "unsupported mount type '%s'", m.Type)
   131  		}
   132  
   133  		// parentLayerPaths are passed in layerN, layerN-1, ..., layer 0
   134  		//
   135  		// The OCI spec expects:
   136  		//   layerN, layerN-1, ..., layer0, scratch
   137  		var parentLayerPaths []string
   138  		for _, option := range m.Options {
   139  			if strings.HasPrefix(option, mount.ParentLayerPathsFlag) {
   140  				err := json.Unmarshal([]byte(option[len(mount.ParentLayerPathsFlag):]), &parentLayerPaths)
   141  				if err != nil {
   142  					return nil, errors.Wrapf(errdefs.ErrFailedPrecondition, "failed to unmarshal parent layer paths from mount: %v", err)
   143  				}
   144  			}
   145  		}
   146  
   147  		// This is a Windows Argon make sure that we have a Root filled in.
   148  		if spec.Windows.HyperV == nil {
   149  			if spec.Root == nil {
   150  				spec.Root = &specs.Root{}
   151  			}
   152  		}
   153  
   154  		// Append the parents
   155  		spec.Windows.LayerFolders = append(spec.Windows.LayerFolders, parentLayerPaths...)
   156  		// Append the scratch
   157  		spec.Windows.LayerFolders = append(spec.Windows.LayerFolders, m.Source)
   158  	}
   159  
   160  	if req.Terminal && req.Stderr != "" {
   161  		return nil, errors.Wrap(errdefs.ErrFailedPrecondition, "if using terminal, stderr must be empty")
   162  	}
   163  
   164  	resp := &task.CreateTaskResponse{}
   165  	s.cl.Lock()
   166  	if s.isSandbox {
   167  		pod, err := s.getPod()
   168  		if err == nil {
   169  			// The POD sandbox was previously created. Unlock and forward to the POD
   170  			s.cl.Unlock()
   171  			t, err := pod.CreateTask(ctx, req, &spec)
   172  			if err != nil {
   173  				return nil, err
   174  			}
   175  			e, _ := t.GetExec("")
   176  			resp.Pid = uint32(e.Pid())
   177  			return resp, nil
   178  		}
   179  		pod, err = createPod(ctx, s.events, req, &spec)
   180  		if err != nil {
   181  			s.cl.Unlock()
   182  			return nil, err
   183  		}
   184  		t, _ := pod.GetTask(req.ID)
   185  		e, _ := t.GetExec("")
   186  		resp.Pid = uint32(e.Pid())
   187  		s.taskOrPod.Store(pod)
   188  	} else {
   189  		t, err := newHcsStandaloneTask(ctx, s.events, req, &spec)
   190  		if err != nil {
   191  			s.cl.Unlock()
   192  			return nil, err
   193  		}
   194  		e, _ := t.GetExec("")
   195  		resp.Pid = uint32(e.Pid())
   196  		s.taskOrPod.Store(t)
   197  	}
   198  	s.cl.Unlock()
   199  	return resp, nil
   200  }
   201  
   202  func (s *service) startInternal(ctx context.Context, req *task.StartRequest) (*task.StartResponse, error) {
   203  	t, err := s.getTask(req.ID)
   204  	if err != nil {
   205  		return nil, err
   206  	}
   207  	e, err := t.GetExec(req.ExecID)
   208  	if err != nil {
   209  		return nil, err
   210  	}
   211  	err = e.Start(ctx)
   212  	if err != nil {
   213  		return nil, err
   214  	}
   215  	return &task.StartResponse{
   216  		Pid: uint32(e.Pid()),
   217  	}, nil
   218  }
   219  
   220  func (s *service) deleteInternal(ctx context.Context, req *task.DeleteRequest) (*task.DeleteResponse, error) {
   221  	t, err := s.getTask(req.ID)
   222  	if err != nil {
   223  		return nil, err
   224  	}
   225  
   226  	pid, exitStatus, exitedAt, err := t.DeleteExec(ctx, req.ExecID)
   227  	if err != nil {
   228  		return nil, err
   229  	}
   230  
   231  	// if the delete is for a task and not an exec, remove the pod sandbox's reference to the task
   232  	if s.isSandbox && req.ExecID == "" {
   233  		p, err := s.getPod()
   234  		if err != nil {
   235  			return nil, errors.Wrapf(err, "could not get pod %q to delete task %q", s.tid, req.ID)
   236  		}
   237  		err = p.DeleteTask(ctx, req.ID)
   238  		if err != nil {
   239  			return nil, fmt.Errorf("could not delete task %q in pod %q: %w", req.ID, s.tid, err)
   240  		}
   241  	}
   242  	// TODO: check if the pod's workload tasks is empty, and, if so, reset p.taskOrPod to nil
   243  
   244  	return &task.DeleteResponse{
   245  		Pid:        uint32(pid),
   246  		ExitStatus: exitStatus,
   247  		ExitedAt:   exitedAt,
   248  	}, nil
   249  }
   250  
   251  func (s *service) pidsInternal(ctx context.Context, req *task.PidsRequest) (*task.PidsResponse, error) {
   252  	t, err := s.getTask(req.ID)
   253  	if err != nil {
   254  		return nil, err
   255  	}
   256  	pids, err := t.Pids(ctx)
   257  	if err != nil {
   258  		return nil, err
   259  	}
   260  	processes := make([]*containerd_v1_types.ProcessInfo, len(pids))
   261  	for i, p := range pids {
   262  		a, err := typeurl.MarshalAny(&p)
   263  		if err != nil {
   264  			return nil, errors.Wrapf(err, "failed to marshal ProcessDetails for process: %s, task: %s", p.ExecID, req.ID)
   265  		}
   266  		proc := &containerd_v1_types.ProcessInfo{
   267  			Pid:  p.ProcessID,
   268  			Info: a,
   269  		}
   270  		processes[i] = proc
   271  	}
   272  	return &task.PidsResponse{
   273  		Processes: processes,
   274  	}, nil
   275  }
   276  
   277  func (s *service) pauseInternal(ctx context.Context, req *task.PauseRequest) (*google_protobuf1.Empty, error) {
   278  	/*
   279  		s.events <- cdevent{
   280  			topic: runtime.TaskPausedEventTopic,
   281  			event: &eventstypes.TaskPaused{
   282  				req.ID,
   283  			},
   284  		}
   285  	*/
   286  	return nil, errdefs.ErrNotImplemented
   287  }
   288  
   289  func (s *service) resumeInternal(ctx context.Context, req *task.ResumeRequest) (*google_protobuf1.Empty, error) {
   290  	/*
   291  		s.events <- cdevent{
   292  			topic: runtime.TaskResumedEventTopic,
   293  			event: &eventstypes.TaskResumed{
   294  				req.ID,
   295  			},
   296  		}
   297  	*/
   298  	return nil, errdefs.ErrNotImplemented
   299  }
   300  
   301  func (s *service) checkpointInternal(ctx context.Context, req *task.CheckpointTaskRequest) (*google_protobuf1.Empty, error) {
   302  	return nil, errdefs.ErrNotImplemented
   303  }
   304  
   305  func (s *service) killInternal(ctx context.Context, req *task.KillRequest) (*google_protobuf1.Empty, error) {
   306  	if s.isSandbox {
   307  		pod, err := s.getPod()
   308  		if err != nil {
   309  			return nil, errors.Wrapf(errdefs.ErrNotFound, "%v: task with id: '%s' not found", err, req.ID)
   310  		}
   311  		// Send it to the POD and let it cascade on its own through all tasks.
   312  		err = pod.KillTask(ctx, req.ID, req.ExecID, req.Signal, req.All)
   313  		if err != nil {
   314  			return nil, err
   315  		}
   316  		return empty, nil
   317  	}
   318  	t, err := s.getTask(req.ID)
   319  	if err != nil {
   320  		return nil, err
   321  	}
   322  	// Send it to the task and let it cascade on its own through all exec's
   323  	err = t.KillExec(ctx, req.ExecID, req.Signal, req.All)
   324  	if err != nil {
   325  		return nil, err
   326  	}
   327  	return empty, nil
   328  }
   329  
   330  func (s *service) execInternal(ctx context.Context, req *task.ExecProcessRequest) (*google_protobuf1.Empty, error) {
   331  	t, err := s.getTask(req.ID)
   332  	if err != nil {
   333  		return nil, err
   334  	}
   335  	if req.Terminal && req.Stderr != "" {
   336  		return nil, errors.Wrap(errdefs.ErrFailedPrecondition, "if using terminal, stderr must be empty")
   337  	}
   338  	var spec specs.Process
   339  	if err := json.Unmarshal(req.Spec.Value, &spec); err != nil {
   340  		return nil, errors.Wrap(err, "request.Spec was not oci process")
   341  	}
   342  	err = t.CreateExec(ctx, req, &spec)
   343  	if err != nil {
   344  		return nil, err
   345  	}
   346  	return empty, nil
   347  }
   348  
   349  func (s *service) diagExecInHostInternal(ctx context.Context, req *shimdiag.ExecProcessRequest) (*shimdiag.ExecProcessResponse, error) {
   350  	if req.Terminal && req.Stderr != "" {
   351  		return nil, errors.Wrap(errdefs.ErrFailedPrecondition, "if using terminal, stderr must be empty")
   352  	}
   353  	t, err := s.getTask(s.tid)
   354  	if err != nil {
   355  		return nil, err
   356  	}
   357  	ec, err := t.ExecInHost(ctx, req)
   358  	if err != nil {
   359  		return nil, err
   360  	}
   361  	return &shimdiag.ExecProcessResponse{ExitCode: int32(ec)}, nil
   362  }
   363  
   364  func (s *service) diagShareInternal(ctx context.Context, req *shimdiag.ShareRequest) (*shimdiag.ShareResponse, error) {
   365  	t, err := s.getTask(s.tid)
   366  	if err != nil {
   367  		return nil, err
   368  	}
   369  	if err := t.Share(ctx, req); err != nil {
   370  		return nil, err
   371  	}
   372  	return &shimdiag.ShareResponse{}, nil
   373  }
   374  
   375  func (s *service) diagListExecs(task shimTask) ([]*shimdiag.Exec, error) {
   376  	var sdExecs []*shimdiag.Exec
   377  	execs, err := task.ListExecs()
   378  	if err != nil {
   379  		return nil, err
   380  	}
   381  	for _, exec := range execs {
   382  		sdExecs = append(sdExecs, &shimdiag.Exec{ID: exec.ID(), State: string(exec.State())})
   383  	}
   384  	return sdExecs, nil
   385  }
   386  
   387  func (s *service) diagTasksInternal(ctx context.Context, req *shimdiag.TasksRequest) (_ *shimdiag.TasksResponse, err error) {
   388  	raw := s.taskOrPod.Load()
   389  	if raw == nil {
   390  		return nil, errors.Wrapf(errdefs.ErrNotFound, "task with id: '%s' not found", s.tid)
   391  	}
   392  
   393  	resp := &shimdiag.TasksResponse{}
   394  	if s.isSandbox {
   395  		p, ok := raw.(shimPod)
   396  		if !ok {
   397  			return nil, errors.New("failed to convert task to pod")
   398  		}
   399  
   400  		tasks, err := p.ListTasks()
   401  		if err != nil {
   402  			return nil, err
   403  		}
   404  
   405  		for _, task := range tasks {
   406  			t := &shimdiag.Task{ID: task.ID()}
   407  			if req.Execs {
   408  				t.Execs, err = s.diagListExecs(task)
   409  				if err != nil {
   410  					return nil, err
   411  				}
   412  			}
   413  			resp.Tasks = append(resp.Tasks, t)
   414  		}
   415  		return resp, nil
   416  	}
   417  
   418  	t, ok := raw.(shimTask)
   419  	if !ok {
   420  		return nil, errors.New("failed to convert task to 'shimTask'")
   421  	}
   422  
   423  	task := &shimdiag.Task{ID: t.ID()}
   424  	if req.Execs {
   425  		task.Execs, err = s.diagListExecs(t)
   426  		if err != nil {
   427  			return nil, err
   428  		}
   429  	}
   430  
   431  	resp.Tasks = []*shimdiag.Task{task}
   432  	return resp, nil
   433  }
   434  
   435  func (s *service) resizePtyInternal(ctx context.Context, req *task.ResizePtyRequest) (*google_protobuf1.Empty, error) {
   436  	t, err := s.getTask(req.ID)
   437  	if err != nil {
   438  		return nil, err
   439  	}
   440  	e, err := t.GetExec(req.ExecID)
   441  	if err != nil {
   442  		return nil, err
   443  	}
   444  	err = e.ResizePty(ctx, req.Width, req.Height)
   445  	if err != nil {
   446  		return nil, err
   447  	}
   448  	return empty, nil
   449  }
   450  
   451  func (s *service) closeIOInternal(ctx context.Context, req *task.CloseIORequest) (*google_protobuf1.Empty, error) {
   452  	t, err := s.getTask(req.ID)
   453  	if err != nil {
   454  		return nil, err
   455  	}
   456  	e, err := t.GetExec(req.ExecID)
   457  	if err != nil {
   458  		return nil, err
   459  	}
   460  	err = e.CloseIO(ctx, req.Stdin)
   461  	if err != nil {
   462  		return nil, err
   463  	}
   464  	return empty, nil
   465  }
   466  
   467  func (s *service) updateInternal(ctx context.Context, req *task.UpdateTaskRequest) (*google_protobuf1.Empty, error) {
   468  	if req.Resources == nil {
   469  		return nil, errors.Wrapf(errdefs.ErrInvalidArgument, "resources cannot be empty, updating container %s resources failed", req.ID)
   470  	}
   471  	t, err := s.getTask(req.ID)
   472  	if err != nil {
   473  		return nil, err
   474  	}
   475  	if err := t.Update(ctx, req); err != nil {
   476  		return nil, err
   477  	}
   478  	return empty, nil
   479  }
   480  
   481  func (s *service) waitInternal(ctx context.Context, req *task.WaitRequest) (*task.WaitResponse, error) {
   482  	t, err := s.getTask(req.ID)
   483  	if err != nil {
   484  		return nil, err
   485  	}
   486  	var state *task.StateResponse
   487  	if req.ExecID != "" {
   488  		e, err := t.GetExec(req.ExecID)
   489  		if err != nil {
   490  			return nil, err
   491  		}
   492  		state = e.Wait()
   493  	} else {
   494  		state = t.Wait()
   495  	}
   496  	return &task.WaitResponse{
   497  		ExitStatus: state.ExitStatus,
   498  		ExitedAt:   state.ExitedAt,
   499  	}, nil
   500  }
   501  
   502  func (s *service) statsInternal(ctx context.Context, req *task.StatsRequest) (*task.StatsResponse, error) {
   503  	t, err := s.getTask(req.ID)
   504  	if err != nil {
   505  		return nil, err
   506  	}
   507  	stats, err := t.Stats(ctx)
   508  	if err != nil {
   509  		return nil, err
   510  	}
   511  	any, err := typeurl.MarshalAny(stats)
   512  	if err != nil {
   513  		return nil, errors.Wrapf(err, "failed to marshal Statistics for task: %s", req.ID)
   514  	}
   515  	return &task.StatsResponse{Stats: any}, nil
   516  }
   517  
   518  func (s *service) connectInternal(ctx context.Context, req *task.ConnectRequest) (*task.ConnectResponse, error) {
   519  	// We treat the shim/task as the same pid on the Windows host.
   520  	pid := uint32(os.Getpid())
   521  	return &task.ConnectResponse{
   522  		ShimPid: pid,
   523  		TaskPid: pid,
   524  	}, nil
   525  }
   526  
   527  func (s *service) shutdownInternal(ctx context.Context, req *task.ShutdownRequest) (*google_protobuf1.Empty, error) {
   528  	// Because a pod shim hosts multiple tasks only the init task can issue the
   529  	// shutdown request.
   530  	if req.ID != s.tid {
   531  		return empty, nil
   532  	}
   533  
   534  	s.shutdownOnce.Do(func() {
   535  		// TODO: should taskOrPod be deleted/set to nil?
   536  		// TODO: is there any extra leftovers of the shimTask/Pod to clean? ie: verify all handles are closed?
   537  		s.gracefulShutdown = !req.Now
   538  		close(s.shutdown)
   539  	})
   540  
   541  	return empty, nil
   542  }
   543  
   544  func (s *service) computeProcessorInfoInternal(ctx context.Context, req *extendedtask.ComputeProcessorInfoRequest) (*extendedtask.ComputeProcessorInfoResponse, error) {
   545  	t, err := s.getTask(req.ID)
   546  	if err != nil {
   547  		return nil, err
   548  	}
   549  	info, err := t.ProcessorInfo(ctx)
   550  	if err != nil {
   551  		return nil, err
   552  	}
   553  	return &extendedtask.ComputeProcessorInfoResponse{
   554  		Count: info.count,
   555  	}, nil
   556  }
   557
View as plain text