github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Address PR comments

Browse Source 
Signed-off-by: Maksym Pavlenko <pavlenko.maksym@gmail.com>
This commit is contained in:
Maksym Pavlenko 2021-11-02 11:19:43 -07:00
parent 6fa1bb4a5c
commit d022fbe789
2 changed files with 30 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
runtime/v2/manager.go
View File

@ -114,7 +114,7 @@ func init() {
} }
func migrateTasks(ic *plugin.InitContext, shimManager *ShimManager) error { func migrateTasks(ic *plugin.InitContext, shimManager *ShimManager) error {
if !shimManager.list.IsEmpty() { if !shimManager.shims.IsEmpty() {
return nil return nil
} }
@ -122,29 +122,29 @@ func migrateTasks(ic *plugin.InitContext, shimManager *ShimManager) error {
// `Root` and `State` dirs expected to be empty at this point (we check that there are no shims loaded above). // `Root` and `State` dirs expected to be empty at this point (we check that there are no shims loaded above).
// If for some they are not empty, these remove calls will fail (`os.Remove` requires a dir to be empty to succeed). // If for some they are not empty, these remove calls will fail (`os.Remove` requires a dir to be empty to succeed).
if err := os.Remove(shimManager.root); err != nil && !os.IsNotExist(err) { if err := os.Remove(shimManager.root); err != nil && !os.IsNotExist(err) {
return err return fmt.Errorf("failed to remove `root` dir: %w", err)
} }
if err := os.Remove(shimManager.state); err != nil && !os.IsNotExist(err) { if err := os.Remove(shimManager.state); err != nil && !os.IsNotExist(err) {
return err return fmt.Errorf("failed to remove `state` dir: %w", err)
} }
if err := os.Rename(ic.Root, shimManager.root); err != nil { if err := os.Rename(ic.Root, shimManager.root); err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
return nil return nil
} }
return fmt.Errorf("failed to migrate task `root` directory") return fmt.Errorf("failed to migrate task `root` directory: %w", err)
} }
if err := os.Rename(ic.State, shimManager.state); err != nil { if err := os.Rename(ic.State, shimManager.state); err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
return nil return nil
} }
return fmt.Errorf("failed to migrate task `state` directory") return fmt.Errorf("failed to migrate task `state` directory: %w", err)
} }
if err := shimManager.loadExistingTasks(ic.Context); err != nil { if err := shimManager.loadExistingTasks(ic.Context); err != nil {
return fmt.Errorf("failed to load tasks after migration") return fmt.Errorf("failed to load tasks after migration: %w", err)
} }
return nil return nil
@ -173,7 +173,7 @@ func NewShimManager(ctx context.Context, config *ManagerConfig) (*ShimManager, e
state: config.State, state: config.State,
containerdAddress: config.Address, containerdAddress: config.Address,
containerdTTRPCAddress: config.TTRPCAddress, containerdTTRPCAddress: config.TTRPCAddress,
list: runtime.NewTaskList(), shims: runtime.NewTaskList(),
events: config.Events, events: config.Events,
containers: config.Store, containers: config.Store,
schedCore: config.SchedCore, schedCore: config.SchedCore,
@ -196,7 +196,7 @@ type ShimManager struct {
containerdAddress string containerdAddress string
containerdTTRPCAddress string containerdTTRPCAddress string
schedCore bool schedCore bool
list *runtime.TaskList shims *runtime.TaskList
events *exchange.Exchange events *exchange.Exchange
containers containers.Store containers containers.Store
} }
@ -235,7 +235,7 @@ func (m *ShimManager) Start(ctx context.Context, id string, opts runtime.CreateO
task: task.NewTaskClient(shim.client), task: task.NewTaskClient(shim.client),
} }
if err := m.list.Add(ctx, shimTask); err != nil { if err := m.shims.Add(ctx, shimTask); err != nil {
return nil, errors.Wrap(err, "failed to add task") return nil, errors.Wrap(err, "failed to add task")
} }
@ -262,12 +262,12 @@ func (m *ShimManager) startShim(ctx context.Context, bundle *Bundle, id string,
shim, err := b.Start(ctx, topts, func() { shim, err := b.Start(ctx, topts, func() {
log.G(ctx).WithField("id", id).Info("shim disconnected") log.G(ctx).WithField("id", id).Info("shim disconnected")
cleanupAfterDeadShim(context.Background(), id, ns, m.list, m.events, b) cleanupAfterDeadShim(context.Background(), id, ns, m.shims, m.events, b)
// Remove self from the runtime task list. Even though the cleanupAfterDeadShim() // Remove self from the runtime task list. Even though the cleanupAfterDeadShim()
// would publish taskExit event, but the shim.Delete() would always failed with ttrpc // would publish taskExit event, but the shim.Delete() would always failed with ttrpc
// disconnect and there is no chance to remove this dead task from runtime task lists. // disconnect and there is no chance to remove this dead task from runtime task lists.
// Thus it's better to delete it here. // Thus it's better to delete it here.
m.list.Delete(ctx, id) m.shims.Delete(ctx, id)
}) })
if err != nil { if err != nil {
return nil, errors.Wrap(err, "start failed") return nil, errors.Wrap(err, "start failed")
@ -282,11 +282,11 @@ func (m *ShimManager) cleanupShim(shim *shim) {
defer cancel() defer cancel()
_ = shim.delete(dctx) _ = shim.delete(dctx)
m.list.Delete(dctx, shim.ID()) m.shims.Delete(dctx, shim.ID())
} }
func (m *ShimManager) Get(ctx context.Context, id string) (ShimProcess, error) { func (m *ShimManager) Get(ctx context.Context, id string) (ShimProcess, error) {
proc, err := m.list.Get(ctx, id) proc, err := m.shims.Get(ctx, id)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -296,14 +296,14 @@ func (m *ShimManager) Get(ctx context.Context, id string) (ShimProcess, error) {
// Delete a runtime task // Delete a runtime task
func (m *ShimManager) Delete(ctx context.Context, id string) error { func (m *ShimManager) Delete(ctx context.Context, id string) error {
proc, err := m.list.Get(ctx, id) proc, err := m.shims.Get(ctx, id)
if err != nil { if err != nil {
return err return err
} }
shimTask := proc.(*shimTask) shimTask := proc.(*shimTask)
err = shimTask.shim.delete(ctx) err = shimTask.shim.delete(ctx)
m.list.Delete(ctx, id) m.shims.Delete(ctx, id)
return err return err
} }
@ -322,13 +322,13 @@ func parsePlatforms(platformStr []string) ([]ocispec.Platform, error) {
// TaskManager wraps task service client on top of shim manager. // TaskManager wraps task service client on top of shim manager.
type TaskManager struct { type TaskManager struct {
shims *ShimManager manager *ShimManager
} }
// NewTaskManager creates a new task manager instance. // NewTaskManager creates a new task manager instance.
func NewTaskManager(shims *ShimManager) *TaskManager { func NewTaskManager(shims *ShimManager) *TaskManager {
return &TaskManager{ return &TaskManager{
shims: shims, manager: shims,
} }
} }
@ -339,11 +339,13 @@ func (m *TaskManager) ID() string {
// Create launches new shim instance and creates new task // Create launches new shim instance and creates new task
func (m *TaskManager) Create(ctx context.Context, taskID string, opts runtime.CreateOpts) (runtime.Task, error) { func (m *TaskManager) Create(ctx context.Context, taskID string, opts runtime.CreateOpts) (runtime.Task, error) {
process, err := m.shims.Start(ctx, taskID, opts) process, err := m.manager.Start(ctx, taskID, opts)
if err != nil { if err != nil {
return nil, errors.Wrap(err, "failed to start shim") return nil, errors.Wrap(err, "failed to start shim")
} }
// Cast to shim task and call task service to create a new container task instance.
// This will not be required once shim service / client implemented.
shim := process.(*shimTask) shim := process.(*shimTask)
t, err := shim.Create(ctx, opts) t, err := shim.Create(ctx, opts)
if err != nil { if err != nil {
@ -351,7 +353,7 @@ func (m *TaskManager) Create(ctx context.Context, taskID string, opts runtime.Cr
defer cancel() defer cancel()
_, errShim := shim.delete(dctx, func(ctx context.Context, id string) { _, errShim := shim.delete(dctx, func(ctx context.Context, id string) {
m.shims.list.Delete(ctx, id) m.manager.shims.Delete(ctx, id)
}) })
if errShim != nil { if errShim != nil {
@ -372,24 +374,24 @@ func (m *TaskManager) Create(ctx context.Context, taskID string, opts runtime.Cr
// Get a specific task // Get a specific task
func (m *TaskManager) Get(ctx context.Context, id string) (runtime.Task, error) { func (m *TaskManager) Get(ctx context.Context, id string) (runtime.Task, error) {
return m.shims.list.Get(ctx, id) return m.manager.shims.Get(ctx, id)
} }
// Tasks lists all tasks // Tasks lists all tasks
func (m *TaskManager) Tasks(ctx context.Context, all bool) ([]runtime.Task, error) { func (m *TaskManager) Tasks(ctx context.Context, all bool) ([]runtime.Task, error) {
return m.shims.list.GetAll(ctx, all) return m.manager.shims.GetAll(ctx, all)
} }
// Delete deletes the task and shim instance // Delete deletes the task and shim instance
func (m *TaskManager) Delete(ctx context.Context, taskID string) (*runtime.Exit, error) { func (m *TaskManager) Delete(ctx context.Context, taskID string) (*runtime.Exit, error) {
item, err := m.shims.list.Get(ctx, taskID) item, err := m.manager.shims.Get(ctx, taskID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
shimTask := item.(*shimTask) shimTask := item.(*shimTask)
exit, err := shimTask.delete(ctx, func(ctx context.Context, id string) { exit, err := shimTask.delete(ctx, func(ctx context.Context, id string) {
m.shims.list.Delete(ctx, id) m.manager.shims.Delete(ctx, id)
}) })
if err != nil { if err != nil {

10
runtime/v2/shim_load.go
View File

@ -107,15 +107,15 @@ func (m *ShimManager) loadShims(ctx context.Context) error {
shim, err := loadShim(ctx, bundle, func() { shim, err := loadShim(ctx, bundle, func() {
log.G(ctx).WithField("id", id).Info("shim disconnected") log.G(ctx).WithField("id", id).Info("shim disconnected")
cleanupAfterDeadShim(context.Background(), id, ns, m.list, m.events, binaryCall) cleanupAfterDeadShim(context.Background(), id, ns, m.shims, m.events, binaryCall)
// Remove self from the runtime task list. // Remove self from the runtime task list.
m.list.Delete(ctx, id) m.shims.Delete(ctx, id)
}) })
if err != nil { if err != nil {
cleanupAfterDeadShim(ctx, id, ns, m.list, m.events, binaryCall) cleanupAfterDeadShim(ctx, id, ns, m.shims, m.events, binaryCall)
continue continue
} }
m.list.Add(ctx, shim) m.shims.Add(ctx, shim)
} }
return nil return nil
} }
@ -133,7 +133,7 @@ func (m *ShimManager) cleanupWorkDirs(ctx context.Context) error {
// if the task was not loaded, cleanup and empty working directory // if the task was not loaded, cleanup and empty working directory
// this can happen on a reboot where /run for the bundle state is cleaned up // this can happen on a reboot where /run for the bundle state is cleaned up
// but that persistent working dir is left // but that persistent working dir is left
if _, err := m.list.Get(ctx, d.Name()); err != nil { if _, err := m.shims.Get(ctx, d.Name()); err != nil {
path := filepath.Join(m.root, ns, d.Name()) path := filepath.Join(m.root, ns, d.Name())
if err := os.RemoveAll(path); err != nil { if err := os.RemoveAll(path); err != nil {
log.G(ctx).WithError(err).Errorf("cleanup working dir %s", path) log.G(ctx).WithError(err).Errorf("cleanup working dir %s", path)