From b385e92d9474dff2307400a6d55ad3cf0a0774b2 Mon Sep 17 00:00:00 2001
From: bergzhao <13611129507@163.com>
Date: Tue, 21 Apr 2020 21:48:59 +0800
Subject: [PATCH] =?UTF-8?q?feature:=20=E6=94=AF=E6=8C=81=E5=AE=B9=E5=99=A8?=
 =?UTF-8?q?=E6=89=A9=E5=B1=95=E8=B5=84=E6=BA=90=E8=B0=83=E5=BA=A6=E6=9C=BA?=
 =?UTF-8?q?=E5=88=B6=20#424?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Makefile                                      |   2 +
 bcs-common/pkg/mesosdriver/client.go          |   6 +-
 .../container/cni/cni_pod.go                  |  30 +-
 .../container/cnm/cnm_pod.go                  |  54 ++--
 .../bcs-container-executor/container/task.go  |  50 ++--
 .../device-plugin-manager/plugin-manager.go   |  43 ++-
 .../src/manager/sched/offer/offerpool.go      |   9 +-
 .../src/manager/sched/offer/offers_test.go    | 259 ------------------
 .../manager/sched/scheduler/task_operate.go   |  10 +-
 .../sched/scheduler/trans_launchapp.go        |   6 +-
 .../sched/scheduler/trans_reschedule.go       |   7 +-
 .../manager/sched/scheduler/trans_scaleapp.go |   6 +-
 .../sched/scheduler/trans_updateapp.go        |   6 +-
 .../src/manager/sched/task/task.go            |   6 +-
 .../bcs-scheduler/src/types/application.go    |  12 +-
 .../cpuset-device/container.go                | 210 +++++++++++---
 .../cpuset-device/manager.go                  |  45 +--
 bcs-services/bcs-cpuset-device/types/types.go |  25 +-
 18 files changed, 340 insertions(+), 446 deletions(-)
 delete mode 100644 bcs-mesos/bcs-scheduler/src/manager/sched/offer/offers_test.go

diff --git a/Makefile b/Makefile
index 0f889c2575..f3ef7a0f8d 100644
--- a/Makefile
+++ b/Makefile
@@ -68,6 +68,8 @@ pre:
 	@echo "git tag: ${GITTAG}"
 	mkdir -p ${PACKAGEPATH}
 	mkdir -p ${EXPORTPATH}
+	go fmt ./...
+	cd ./scripts && chmod +x vet.sh && ./vet.sh
 
 api:pre
 	mkdir -p ${PACKAGEPATH}/bcs-services
diff --git a/bcs-common/pkg/mesosdriver/client.go b/bcs-common/pkg/mesosdriver/client.go
index 263f128524..594c452110 100644
--- a/bcs-common/pkg/mesosdriver/client.go
+++ b/bcs-common/pkg/mesosdriver/client.go
@@ -51,7 +51,7 @@ func NewMesosPlatform(conf *Config) (*MesosDriverClient, error) {
 	//init http client
 	m.cli = httpclient.NewHttpClient()
 	//if https
-	if m.conf.ClientCert!=nil && m.conf.ClientCert.IsSSL {
+	if m.conf.ClientCert != nil && m.conf.ClientCert.IsSSL {
 		blog.Infof("NetworkDetection http client cert ssl")
 		m.cli.SetTlsVerity(m.conf.ClientCert.CAFile, m.conf.ClientCert.CertFile, m.conf.ClientCert.KeyFile,
 			m.conf.ClientCert.CertPasswd)
@@ -84,7 +84,7 @@ func (m *MesosDriverClient) getModuleAddr(clusterid string) (string, error) {
 
 //update agent external resources
 func (m *MesosDriverClient) UpdateAgentExtendedResources(er *commtypes.ExtendedResource) error {
-	by,_ := json.Marshal(er)
+	by, _ := json.Marshal(er)
 	_, err := m.requestMesosApiserver(m.conf.ClusterId, http.MethodPut, "agentsettings/extendedresource", by)
 	if err != nil {
 		blog.Errorf("update agent %s external resources error %s", er.InnerIP, err.Error())
@@ -107,7 +107,7 @@ func (m *MesosDriverClient) requestMesosApiserver(clusterid, method, url string,
 	}
 	uri := fmt.Sprintf("%s/mesosdriver/v4/%s", addr, url)
 	m.cli.SetHeader("BCS-ClusterID", clusterid)
-	blog.V(3).Infof("request %s body(%s)",uri, string(payload))
+	blog.V(3).Infof("request %s body(%s)", uri, string(payload))
 
 	var by []byte
 	switch method {
diff --git a/bcs-mesos/bcs-container-executor/container/cni/cni_pod.go b/bcs-mesos/bcs-container-executor/container/cni/cni_pod.go
index a3ba719c53..7f9ea827e7 100644
--- a/bcs-mesos/bcs-container-executor/container/cni/cni_pod.go
+++ b/bcs-mesos/bcs-container-executor/container/cni/cni_pod.go
@@ -94,7 +94,7 @@ func NewPod(operator container.Container, tasks []*container.BcsContainerTask,
 		conTasks:         taskMap,
 		networkTaskId:    tasks[0].TaskId,
 		runningContainer: make(map[string]*container.BcsContainerInfo),
-		pluginManager: device_plugin_manager.NewDevicePluginManager(),
+		pluginManager:    device_plugin_manager.NewDevicePluginManager(),
 	}
 	if len(tasks[0].NetworkIPAddr) != 0 {
 		//ip injected by executor
@@ -138,7 +138,7 @@ type CNIPod struct {
 	networkTaskId string
 	netImage      string
 	//device plugin manager
-	pluginManager    *device_plugin_manager.DevicePluginManager
+	pluginManager *device_plugin_manager.DevicePluginManager
 }
 
 //IsHealthy check pod is healthy
@@ -254,7 +254,7 @@ func (p *CNIPod) Init() error {
 		HostName:       p.cniHostName,
 	}
 	p.netTask.Resource = &bcstypes.Resource{
-		Cpus:   1,
+		Cpus: 1,
 	}
 
 	netflag := container.BcsKV{
@@ -390,32 +390,29 @@ func (p *CNIPod) Start() error {
 		task.HostName = ""
 		var extendedErr error
 		//if task contains extended resources, need connect device plugin to allocate resources
-		for _,ex :=range task.ExtendedResources {
+		for _, ex := range task.ExtendedResources {
 			logs.Infof("task %s contains extended resource %s, then allocate it", task.TaskId, ex.Name)
-			deviceIds,err := p.pluginManager.ListAndWatch(ex)
-			if err!=nil {
-				logs.Errorf("task %s ListAndWatch extended resources %s failed, err: %s\n",
+			deviceIds, err := p.pluginManager.ListAndWatch(ex)
+			if err != nil {
+				extendedErr = fmt.Errorf("task %s ListAndWatch extended resources %s failed, err: %s\n",
 					task.TaskId, ex.Name, err.Error())
-				extendedErr = err
 				break
 			}
 
 			//allocate device
-			if len(deviceIds)<int(ex.Value) {
+			if len(deviceIds) < int(ex.Value) {
 				extendedErr = fmt.Errorf("extended resources %s Capacity %d, not enough", ex.Name, len(deviceIds))
-				logs.Errorf(extendedErr.Error())
 				break
 			}
-			envs,err := p.pluginManager.Allocate(ex, deviceIds[:int(ex.Value)])
-			if err!=nil {
-				logs.Errorf("task %s extended resources %s Allocate deviceIds(%v) failed, err: %s\n",
+			envs, err := p.pluginManager.Allocate(ex, deviceIds[:int(ex.Value)])
+			if err != nil {
+				extendedErr = fmt.Errorf("task %s extended resources %s Allocate deviceIds(%v) failed, err: %s\n",
 					task.TaskId, ex.Name, deviceIds[:int(ex.Value)], err.Error())
-				extendedErr = err
 				break
 			}
 
 			//append response docker envs to task.envs
-			for k,v :=range envs {
+			for k, v := range envs {
 				kv := container.BcsKV{
 					Key:   k,
 					Value: v,
@@ -425,7 +422,8 @@ func (p *CNIPod) Start() error {
 		}
 
 		//if allocate extended resource failed, then return and exit
-		if extendedErr!=nil {
+		if extendedErr != nil {
+			logs.Errorf(extendedErr.Error())
 			task.RuntimeConf.Status = container.ContainerStatus_EXITED
 			task.RuntimeConf.Message = extendedErr.Error()
 			p.startFailedStop(extendedErr)
diff --git a/bcs-mesos/bcs-container-executor/container/cnm/cnm_pod.go b/bcs-mesos/bcs-container-executor/container/cnm/cnm_pod.go
index 0fc0ddbc9b..abd512b352 100644
--- a/bcs-mesos/bcs-container-executor/container/cnm/cnm_pod.go
+++ b/bcs-mesos/bcs-container-executor/container/cnm/cnm_pod.go
@@ -92,7 +92,7 @@ type DockerPod struct {
 	conTasks         map[string]*container.BcsContainerTask //task for running containers, key is taskID
 	runningContainer map[string]*container.BcsContainerInfo //running container Name list for monitor
 	//device plugin manager
-	pluginManager    *device_plugin_manager.DevicePluginManager
+	pluginManager *device_plugin_manager.DevicePluginManager
 }
 
 //IsHealthy check pod is healthy
@@ -195,32 +195,29 @@ func (p *DockerPod) Init() error {
 	container.EnvOperCopy(p.netTask)
 	var extendedErr error
 	//if task contains extended resources, need connect device plugin to allocate resources
-	for _,ex :=range p.netTask.ExtendedResources {
+	for _, ex := range p.netTask.ExtendedResources {
 		logs.Infof("task %s contains extended resource %s, then allocate it", p.netTask.TaskId, ex.Name)
-		deviceIds,err := p.pluginManager.ListAndWatch(ex)
-		if err!=nil {
-			logs.Errorf("task %s ListAndWatch extended resources %s failed, err: %s\n",
+		deviceIds, err := p.pluginManager.ListAndWatch(ex)
+		if err != nil {
+			extendedErr = fmt.Errorf("task %s ListAndWatch extended resources %s failed, err: %s\n",
 				p.netTask.TaskId, ex.Name, err.Error())
-			extendedErr = err
 			break
 		}
 
 		//allocate device
-		if len(deviceIds)<int(ex.Value) {
+		if len(deviceIds) < int(ex.Value) {
 			extendedErr = fmt.Errorf("extended resources %s Capacity %d, not enough", ex.Name, len(deviceIds))
-			logs.Errorf(extendedErr.Error())
 			break
 		}
-		envs,err := p.pluginManager.Allocate(ex, deviceIds[:int(ex.Value)])
-		if err!=nil {
-			logs.Errorf("task %s extended resources %s Allocate deviceIds(%v) failed, err: %s\n",
+		envs, err := p.pluginManager.Allocate(ex, deviceIds[:int(ex.Value)])
+		if err != nil {
+			extendedErr = fmt.Errorf("task %s extended resources %s Allocate deviceIds(%v) failed, err: %s\n",
 				p.netTask.TaskId, ex.Name, deviceIds[:int(ex.Value)], err.Error())
-			extendedErr = err
 			break
 		}
 
 		//append response docker envs to task.envs
-		for k,v :=range envs {
+		for k, v := range envs {
 			kv := container.BcsKV{
 				Key:   k,
 				Value: v,
@@ -228,6 +225,13 @@ func (p *DockerPod) Init() error {
 			p.netTask.Env = append(p.netTask.Env, kv)
 		}
 	}
+	//if allocate extended resource failed, then return and exit
+	if extendedErr != nil {
+		logs.Errorf(extendedErr.Error())
+		p.status = container.PodStatus_FAILED
+		p.message = extendedErr.Error()
+		return extendedErr
+	}
 
 	//fix(developerJim): all containers in pod can not create PortMappings separately,
 	// so we need to copy all PortMappings from other containers to Network
@@ -322,32 +326,29 @@ func (p *DockerPod) Start() error {
 		container.EnvOperCopy(task)
 		var extendedErr error
 		//if task contains extended resources, need connect device plugin to allocate resources
-		for _,ex :=range task.ExtendedResources {
+		for _, ex := range task.ExtendedResources {
 			blog.Infof("task %s contains extended resource %s, then allocate it", task.TaskId, ex.Name)
-			deviceIds,err := p.pluginManager.ListAndWatch(ex)
-			if err!=nil {
-				logs.Errorf("task %s ListAndWatch extended resources %s failed, err: %s\n",
+			deviceIds, err := p.pluginManager.ListAndWatch(ex)
+			if err != nil {
+				extendedErr = fmt.Errorf("task %s ListAndWatch extended resources %s failed, err: %s\n",
 					task.TaskId, ex.Name, err.Error())
-				extendedErr = err
 				break
 			}
 
 			//allocate device
-			if len(deviceIds)<int(ex.Value) {
+			if len(deviceIds) < int(ex.Value) {
 				extendedErr = fmt.Errorf("extended resources %s Capacity %d, not enough", ex.Name, len(deviceIds))
-				logs.Errorf(extendedErr.Error())
 				break
 			}
-			envs,err := p.pluginManager.Allocate(ex, deviceIds[:int(ex.Value)])
-			if err!=nil {
-				logs.Errorf("task %s extended resources %s Allocate deviceIds(%v) failed, err: %s\n",
+			envs, err := p.pluginManager.Allocate(ex, deviceIds[:int(ex.Value)])
+			if err != nil {
+				extendedErr = fmt.Errorf("task %s extended resources %s Allocate deviceIds(%v) failed, err: %s\n",
 					task.TaskId, ex.Name, deviceIds[:int(ex.Value)], err.Error())
-				extendedErr = err
 				break
 			}
 
 			//append response docker envs to task.envs
-			for k,v :=range envs {
+			for k, v := range envs {
 				kv := container.BcsKV{
 					Key:   k,
 					Value: v,
@@ -357,7 +358,8 @@ func (p *DockerPod) Start() error {
 		}
 
 		//if allocate extended resource failed, then return and exit
-		if extendedErr!=nil {
+		if extendedErr != nil {
+			logs.Errorf(extendedErr.Error())
 			task.RuntimeConf.Status = container.ContainerStatus_EXITED
 			task.RuntimeConf.Message = extendedErr.Error()
 			p.startFailedStop(extendedErr)
diff --git a/bcs-mesos/bcs-container-executor/container/task.go b/bcs-mesos/bcs-container-executor/container/task.go
index a960275545..e3f8491aad 100644
--- a/bcs-mesos/bcs-container-executor/container/task.go
+++ b/bcs-mesos/bcs-container-executor/container/task.go
@@ -46,32 +46,32 @@ type BcsVolume struct {
 
 //BcsContainerTask task info for running container
 type BcsContainerTask struct {
-	Name            string                 //container name
-	Image           string                 //container image
-	HostName        string                 //container hostname
-	Hosts           []string               //host:ip pair for /etc/hosts in container
-	Command         string                 //container command
-	Args            []string               //args for command
-	Env             []BcsKV                //environments
-	Volums          []BcsVolume            //host path mount
-	NetworkName     string                 //container network name
-	NetworkIPAddr   string                 //container ip address request
-	ForcePullImage  bool                   //pull image every time
-	OOMKillDisabled bool                   //OOM kill feature, default is true
-	AutoRemove      bool                   //remove container when exit, default false
-	Ulimits         []BcsKV                //ulimit for docker parameter
-	ShmSize         int64                  //docker hostconfig shm size, 1 = 1B
-	Privileged      bool                   //setting container privileged
-	PublishAllPorts bool                   //publish all ports in container
-	PortBindings    map[string]BcsPort     //port for container reflection, only useful for docker bridge
-	Labels          []BcsKV                //label for container
-	Resource        *bcstypes.Resource     //container resource request
-	LimitResource   *bcstypes.Resource     // container resource limit
+	Name              string                       //container name
+	Image             string                       //container image
+	HostName          string                       //container hostname
+	Hosts             []string                     //host:ip pair for /etc/hosts in container
+	Command           string                       //container command
+	Args              []string                     //args for command
+	Env               []BcsKV                      //environments
+	Volums            []BcsVolume                  //host path mount
+	NetworkName       string                       //container network name
+	NetworkIPAddr     string                       //container ip address request
+	ForcePullImage    bool                         //pull image every time
+	OOMKillDisabled   bool                         //OOM kill feature, default is true
+	AutoRemove        bool                         //remove container when exit, default false
+	Ulimits           []BcsKV                      //ulimit for docker parameter
+	ShmSize           int64                        //docker hostconfig shm size, 1 = 1B
+	Privileged        bool                         //setting container privileged
+	PublishAllPorts   bool                         //publish all ports in container
+	PortBindings      map[string]BcsPort           //port for container reflection, only useful for docker bridge
+	Labels            []BcsKV                      //label for container
+	Resource          *bcstypes.Resource           //container resource request
+	LimitResource     *bcstypes.Resource           // container resource limit
 	ExtendedResources []*comtypes.ExtendedResource //extended resources
-	BcsMessages     []*bcstypes.BcsMessage //bcs define message
-	RuntimeConf     *BcsContainerInfo      //container runtime info
-	HealthCheck     healthcheck.Checker    //for health check
-	KillPolicy      int                    //kill policy timeout, unit is seconds
+	BcsMessages       []*bcstypes.BcsMessage       //bcs define message
+	RuntimeConf       *BcsContainerInfo            //container runtime info
+	HealthCheck       healthcheck.Checker          //for health check
+	KillPolicy        int                          //kill policy timeout, unit is seconds
 	//container network flow limit args
 	NetLimit *comtypes.NetLimit
 	TaskId   string
diff --git a/bcs-mesos/bcs-container-executor/device-plugin-manager/plugin-manager.go b/bcs-mesos/bcs-container-executor/device-plugin-manager/plugin-manager.go
index 88c2c9bb1a..5532c0f9c3 100644
--- a/bcs-mesos/bcs-container-executor/device-plugin-manager/plugin-manager.go
+++ b/bcs-mesos/bcs-container-executor/device-plugin-manager/plugin-manager.go
@@ -21,23 +21,22 @@ import (
 	"bk-bcs/bcs-common/common/blog"
 	comtypes "bk-bcs/bcs-common/common/types"
 
-	"google.golang.org/grpc"
 	"golang.org/x/net/context"
+	"google.golang.org/grpc"
 	pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
 )
 
 type DevicePluginManager struct {
-
 }
 
 //new DevicePluginManager function
-func NewDevicePluginManager()*DevicePluginManager{
+func NewDevicePluginManager() *DevicePluginManager {
 	return &DevicePluginManager{}
 }
 
 //request device plugin to listandwatch device list ids
 //return deviceIds, examples: ["cpuset0","cpuset1","cpuset2"...]
-func (m *DevicePluginManager) ListAndWatch(ex *comtypes.ExtendedResource)([]string, error){
+func (m *DevicePluginManager) ListAndWatch(ex *comtypes.ExtendedResource) ([]string, error) {
 	//connect grpc socket
 	conn, err := m.dial(ex.Socket, 5*time.Second)
 	if err != nil {
@@ -49,25 +48,25 @@ func (m *DevicePluginManager) ListAndWatch(ex *comtypes.ExtendedResource)([]stri
 	client := pluginapi.NewDevicePluginClient(conn)
 	listAndWatchClient, err := client.ListAndWatch(context.Background(), &pluginapi.Empty{})
 	if err != nil {
-		blog.Errorf("extended resource %s ListAndWatch failed: %s",ex.Name, err.Error())
+		blog.Errorf("extended resource %s ListAndWatch failed: %s", ex.Name, err.Error())
 		return nil, err
 	}
-	response,err := listAndWatchClient.Recv()
+	response, err := listAndWatchClient.Recv()
 	if err != nil {
-		blog.Errorf("extended resource %s ListAndWatch receive message failed: %s",ex.Name, err.Error())
+		blog.Errorf("extended resource %s ListAndWatch receive message failed: %s", ex.Name, err.Error())
 		return nil, err
 	}
-	deviceIds := make([]string,0,len(response.Devices))
-	for _,device :=range response.Devices {
+	deviceIds := make([]string, 0, len(response.Devices))
+	for _, device := range response.Devices {
 		deviceIds = append(deviceIds, device.ID)
 	}
-	blog.Infof("extended resource %s ListAndWatch success, devices(%s)",ex.Name,deviceIds)
+	blog.Infof("extended resource %s ListAndWatch success, devices(%s)", ex.Name, deviceIds)
 	return deviceIds, nil
 }
 
 //request deviceplugin to allocate extended resources
 //before create container call the function
-func (m *DevicePluginManager) Allocate(ex *comtypes.ExtendedResource, deviceIds []string)(map[string]string,error){
+func (m *DevicePluginManager) Allocate(ex *comtypes.ExtendedResource, deviceIds []string) (map[string]string, error) {
 	//connect grpc socket
 	conn, err := m.dial(ex.Socket, 5*time.Second)
 	if err != nil {
@@ -78,30 +77,30 @@ func (m *DevicePluginManager) Allocate(ex *comtypes.ExtendedResource, deviceIds
 
 	client := pluginapi.NewDevicePluginClient(conn)
 	in := &pluginapi.AllocateRequest{
-		ContainerRequests: make([]*pluginapi.ContainerAllocateRequest,0),
+		ContainerRequests: make([]*pluginapi.ContainerAllocateRequest, 0),
 	}
-	req := &pluginapi.ContainerAllocateRequest{DevicesIDs:deviceIds}
+	req := &pluginapi.ContainerAllocateRequest{DevicesIDs: deviceIds}
 	in.ContainerRequests = append(in.ContainerRequests, req)
 	response, err := client.Allocate(context.Background(), in)
 	if err != nil {
-		blog.Errorf("extended resource %s Allocate devices(%v) failed: %s",ex.Name, deviceIds, err.Error())
+		blog.Errorf("extended resource %s Allocate devices(%v) failed: %s", ex.Name, deviceIds, err.Error())
 		return nil, err
 	}
-	if len(response.ContainerResponses)==0 {
+	if len(response.ContainerResponses) == 0 {
 		err = fmt.Errorf("ContainerResponses is empty")
-		blog.Errorf("extended resource %s Allocate devices(%v) failed: %s",ex.Name, deviceIds, err.Error())
+		blog.Errorf("extended resource %s Allocate devices(%v) failed: %s", ex.Name, deviceIds, err.Error())
 		return nil, err
 	}
 
 	//the envs are appended when the container is created
 	//Some Settings of device plugin are done according to these docker envs
-	blog.Infof("extended resource %s Allocate success, envs(%v)",ex.Name,response.ContainerResponses[0].Envs)
+	blog.Infof("extended resource %s Allocate success, envs(%v)", ex.Name, response.ContainerResponses[0].Envs)
 	return response.ContainerResponses[0].Envs, nil
 }
 
 //request deviceplugin to allocate extended resources
 //before create container call the function
-func (m *DevicePluginManager) PreStartContainer(ex *comtypes.ExtendedResource, deviceIds []string)error{
+func (m *DevicePluginManager) PreStartContainer(ex *comtypes.ExtendedResource, deviceIds []string) error {
 	//connect grpc socket
 	conn, err := m.dial(ex.Socket, 5*time.Second)
 	if err != nil {
@@ -111,14 +110,14 @@ func (m *DevicePluginManager) PreStartContainer(ex *comtypes.ExtendedResource, d
 	defer conn.Close()
 
 	client := pluginapi.NewDevicePluginClient(conn)
-	in := &pluginapi.PreStartContainerRequest{DevicesIDs:deviceIds}
+	in := &pluginapi.PreStartContainerRequest{DevicesIDs: deviceIds}
 	_, err = client.PreStartContainer(context.Background(), in)
 	if err != nil {
-		blog.Errorf("extended resource %s Allocate devices(%v) failed: %s",ex.Name, deviceIds, err.Error())
+		blog.Errorf("extended resource %s Allocate devices(%v) failed: %s", ex.Name, deviceIds, err.Error())
 		return err
 	}
 
-	blog.Infof("extended resource %s PreStartContainer success",ex.Name)
+	blog.Infof("extended resource %s PreStartContainer success", ex.Name)
 	return nil
 }
 
@@ -136,4 +135,4 @@ func (m *DevicePluginManager) dial(unixSocketPath string, timeout time.Duration)
 	}
 
 	return conn, nil
-}
\ No newline at end of file
+}
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/offer/offerpool.go b/bcs-mesos/bcs-scheduler/src/manager/sched/offer/offerpool.go
index 70b9d629f5..9d6e73f8c9 100644
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/offer/offerpool.go
+++ b/bcs-mesos/bcs-scheduler/src/manager/sched/offer/offerpool.go
@@ -577,7 +577,7 @@ func (p *offerPool) setInnerOffersAttributes(offers []*mesos.Offer) {
 				InnerIP: ip,
 			}
 			err = p.store.SaveAgentSetting(setting)
-			if err!=nil {
+			if err != nil {
 				blog.Errorf("save agentsetting %s error %s", ip, err.Error())
 			}
 			continue
@@ -763,8 +763,9 @@ func (p *offerPool) addOfferAttributes(offer *mesos.Offer, agentSetting *commtyp
 			blog.Errorf("FetchTaskGroup %s failed: %s", id, err.Error())
 			continue
 		}
-		if pod.Status!=types.TASKGROUP_STATUS_RUNNING {
-			blog.V(3).Infof("taskgroup %s status %s ")
+		if pod.Status != types.TASKGROUP_STATUS_RUNNING {
+			blog.V(3).Infof("taskgroup %s status %s, and continue", pod.ID, pod.Status)
+			continue
 		}
 
 		pods = append(pods, pod)
@@ -784,7 +785,7 @@ func (p *offerPool) addOfferAttributes(offer *mesos.Offer, agentSetting *commtyp
 			}
 		}
 	}
-	by,_ := json.Marshal(allocatedResources)
+	by, _ := json.Marshal(allocatedResources)
 	blog.Infof("extended resources %s", string(by))
 	//extended resources, agentsetting have total extended resources
 	for _, ex := range agentSetting.ExtendedResources {
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/offer/offers_test.go b/bcs-mesos/bcs-scheduler/src/manager/sched/offer/offers_test.go
deleted file mode 100644
index 500cd406c6..0000000000
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/offer/offers_test.go
+++ /dev/null
@@ -1,259 +0,0 @@
-/*
- * Tencent is pleased to support the open source community by making Blueking Container Service available.
- * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
- * Licensed under the MIT License (the "License"); you may not use this file except
- * in compliance with the License. You may obtain a copy of the License at
- * http://opensource.org/licenses/MIT
- * Unless required by applicable law or agreed to in writing, software distributed under
- * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
- * either express or implied. See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package offer
-
-import (
-	"fmt"
-	"net/http"
-	"testing"
-	"time"
-
-	typesplugin "bk-bcs/bcs-common/common/plugin"
-	commtype "bk-bcs/bcs-common/common/types"
-	"bk-bcs/bcs-mesos/bcs-scheduler/src/mesosproto/mesos"
-	"bk-bcs/bcs-mesos/bcs-scheduler/src/types"
-
-	"github.com/golang/protobuf/proto"
-	"github.com/stretchr/testify/assert"
-)
-
-const (
-	LostSlaveOfferId     = "lostoffer1"
-	LostSlaveHostname    = "lostofferhostname1"
-	DefaultSleepDuration = 50 * time.Millisecond
-
-	OfferLifePeriod      = 60
-	LostSlaveGracePeriod = 1
-)
-
-type scheduler struct{}
-
-func (s *scheduler) GetHostAttributes(para *typesplugin.HostPluginParameter) (map[string]*typesplugin.HostAttributes, error) {
-	attrs := make(map[string]*typesplugin.HostAttributes, 0)
-
-	attr := &typesplugin.Attribute{
-		Name:   "ip-resources",
-		Type:   typesplugin.ValueScalar,
-		Scalar: typesplugin.Value_Scalar{Value: 1},
-	}
-
-	hostAttr := &typesplugin.HostAttributes{
-		Ip:         "127.0.0.1",
-		Attributes: make([]*typesplugin.Attribute, 0),
-	}
-
-	hostAttr.Attributes = append(hostAttr.Attributes, attr)
-	attrs["127.0.0.1"] = hostAttr
-
-	return attrs, nil
-}
-
-func (s *scheduler) FetchAgentSetting(string) (*commtype.BcsClusterAgentSetting, error) {
-	return nil, nil
-}
-
-func (s *scheduler) FetchAgentSchedInfo(string) (*types.AgentSchedInfo, error) {
-	return nil, nil
-}
-
-func (s *scheduler) GetClusterId() string {
-	return "BCS-TESTBCSTEST01-10001"
-}
-
-func (s *scheduler) DeclineResource(*string) (*http.Response, error) {
-	return nil, nil
-}
-
-var offerpool OfferPool
-var offerids []string
-
-func init() {
-	para := &OfferPara{
-		Sched:                &scheduler{},
-		OfferlifePeriod:      OfferLifePeriod,
-		LostSlaveGracePeriod: LostSlaveGracePeriod,
-	}
-
-	offerpool = NewOfferPool(para)
-
-	offerids = []string{"offer0", "offer1", "offer2", "offer3", "offer4", "offer5",
-		"offer6", "offer7", "offer8", "offer9"}
-}
-
-func getOffers() []*mesos.Offer {
-	offers := make([]*mesos.Offer, 0)
-
-	for i := 0; i < 10; i++ {
-		resource := &mesos.Resource{
-			Name: proto.String("cpus"),
-			Type: mesos.Value_SCALAR.Enum(),
-			Scalar: &mesos.Value_Scalar{
-				Value: proto.Float64(float64(10 - i)),
-			},
-		}
-
-		attr := &mesos.Attribute{
-			Name: proto.String("InnerIP"),
-			Type: mesos.Value_TEXT.Enum(),
-			Text: &mesos.Value_Text{
-				Value: proto.String("127.0.0.1"),
-			},
-		}
-
-		offer := &mesos.Offer{
-			Id: &mesos.OfferID{
-				Value: proto.String(offerids[i]),
-			},
-			FrameworkId: nil,
-			AgentId: &mesos.AgentID{
-				Value: proto.String(fmt.Sprintf("agentid.%d", i)),
-			},
-			Hostname:   proto.String(fmt.Sprintf("hostname.%d", i)),
-			Resources:  []*mesos.Resource{resource},
-			Attributes: []*mesos.Attribute{attr},
-		}
-
-		offers = append(offers, offer)
-	}
-
-	return offers
-}
-
-func getLostOffer() *mesos.Offer {
-	resource := &mesos.Resource{
-		Name: proto.String("cpus"),
-		Type: mesos.Value_SCALAR.Enum(),
-		Scalar: &mesos.Value_Scalar{
-			Value: proto.Float64(float64(1)),
-		},
-	}
-
-	attr := &mesos.Attribute{
-		Name: proto.String("InnerIP"),
-		Type: mesos.Value_TEXT.Enum(),
-		Text: &mesos.Value_Text{
-			Value: proto.String("127.0.0.1"),
-		},
-	}
-
-	offer := &mesos.Offer{
-		Id: &mesos.OfferID{
-			Value: proto.String(LostSlaveOfferId),
-		},
-		FrameworkId: nil,
-		AgentId: &mesos.AgentID{
-			Value: proto.String(fmt.Sprintf("agentid.1")),
-		},
-		Hostname:   proto.String(LostSlaveHostname),
-		Resources:  []*mesos.Resource{resource},
-		Attributes: []*mesos.Attribute{attr},
-	}
-
-	return offer
-}
-
-func TestAddOffers(t *testing.T) {
-	//get offers and add these offers in offer pool
-	offers := getOffers()
-	offerpool.AddOffers(offers)
-
-	// because func AddOffers is asynchronous, so here sleep 1 seconds
-	time.Sleep(time.Second)
-
-	// get first offer
-	offer := offerpool.GetFirstOffer()
-	assert.NotNil(t, offer)
-
-	length := len(offerids)
-	for i, offerid := range offerids {
-		assert.Equal(t, offerid, offer.Offer.GetId().GetValue())
-		offer = offerpool.GetNextOffer(offer)
-
-		if i+1 != length {
-			assert.NotNil(t, offer)
-		} else {
-			assert.Nil(t, offer)
-		}
-	}
-}
-
-func TestLostSlaveOffer(t *testing.T) {
-	//add lost slave, then add this slave offer
-	offerpool.AddLostSlave(LostSlaveHostname)
-	offerpool.AddOffers([]*mesos.Offer{getLostOffer()})
-
-	// because func AddOffers is asynchronous, so sleep some times
-	time.Sleep(DefaultSleepDuration)
-
-	//because the slave just lost, so this slave's offer can't be added in
-	//offer pool, the slave need LostSlaveGracePeriod seconds to recover
-	offer := offerpool.GetFirstOffer()
-	assert.NotNil(t, offer)
-	assert.NotEqual(t, LostSlaveOfferId, offer.Offer.GetId().GetValue())
-
-	length := offerpool.GetOffersLength()
-	for i := 0; i < length; i++ {
-		offer = offerpool.GetNextOffer(offer)
-
-		if i+1 != length {
-			assert.NotNil(t, offer)
-			assert.NotEqual(t, LostSlaveOfferId, offer.Offer.GetId().GetValue())
-		} else {
-			assert.Nil(t, offer)
-		}
-	}
-
-	//the lost slave need LostSlaveGracePeriod times to recover,so sleep the
-	//times, then we can add this slave offer in offer pool
-	time.Sleep((LostSlaveGracePeriod + 1) * time.Second)
-
-	offerpool.AddOffers([]*mesos.Offer{getLostOffer()})
-	time.Sleep(DefaultSleepDuration)
-
-	offer = offerpool.GetFirstOffer()
-	assert.NotNil(t, offer)
-
-	length = offerpool.GetOffersLength()
-	var ok bool
-	var lostOffer *Offer
-
-	for i := 0; i < length; i++ {
-		if offer.Offer.GetId().GetValue() == LostSlaveOfferId {
-			ok = true
-			lostOffer = offer
-			break
-		}
-
-		offer = offerpool.GetNextOffer(offer)
-		if i+1 != length {
-			assert.NotNil(t, offer)
-		} else {
-			assert.Nil(t, offer)
-		}
-	}
-
-	offerpool.UseOffer(lostOffer)
-	assert.Equal(t, ok, true)
-}
-
-func TestGetAllOffers(t *testing.T) {
-	offers := offerpool.GetAllOffers()
-	length := offerpool.GetOffersLength()
-
-	for i, offer := range offers {
-		assert.Equal(t, offerids[i], offer.offerId)
-	}
-
-	assert.Equal(t, length, len(offers))
-}
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/task_operate.go b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/task_operate.go
index 8325678286..774ff4c7b9 100644
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/task_operate.go
+++ b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/task_operate.go
@@ -219,7 +219,7 @@ func (s *Scheduler) DeleteTaskGroup(app *types.Application, taskGroup *types.Tas
 
 	//update agentsetting taskgroup index info
 	nodeIp := taskGroup.GetAgentIp()
-	if nodeIp=="" {
+	if nodeIp == "" {
 		blog.Errorf("taskgroup %s don't have nodeIp", taskGroup.ID)
 		return nil
 	}
@@ -227,12 +227,12 @@ func (s *Scheduler) DeleteTaskGroup(app *types.Application, taskGroup *types.Tas
 	util.Lock.Lock(bcstype.BcsClusterAgentSetting{}, nodeIp)
 	defer util.Lock.UnLock(bcstype.BcsClusterAgentSetting{}, nodeIp)
 
-	agentsetting,err := s.store.FetchAgentSetting(nodeIp)
-	if err!=nil {
+	agentsetting, err := s.store.FetchAgentSetting(nodeIp)
+	if err != nil {
 		blog.Errorf("fetch agentsetting %s failed: %s", nodeIp, err.Error())
 		return nil
 	}
-	if agentsetting==nil {
+	if agentsetting == nil {
 		blog.Errorf("fetch agentsetting %s Not found", nodeIp)
 		return nil
 	}
@@ -248,7 +248,7 @@ func (s *Scheduler) DeleteTaskGroup(app *types.Application, taskGroup *types.Tas
 	}
 	agentsetting.Pods = append(agentsetting.Pods[:delete], agentsetting.Pods[delete+1:]...)
 	err = s.store.SaveAgentSetting(agentsetting)
-	if err!=nil {
+	if err != nil {
 		blog.Errorf("save agentsetting %s failed: %s", nodeIp, err.Error())
 	}
 
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_launchapp.go b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_launchapp.go
index f5f3741c24..bf7aa0b8b4 100644
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_launchapp.go
+++ b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_launchapp.go
@@ -196,11 +196,11 @@ func (s *Scheduler) doLaunchTrans(trans *Transaction, outOffer *offer.Offer, sta
 		//lock agentsetting
 		util.Lock.Lock(commtypes.BcsClusterAgentSetting{}, taskGroup.GetAgentIp())
 		//update agentsettings taskgroup index info
-		agentsetting,_ := s.store.FetchAgentSetting(taskGroup.GetAgentIp())
-		if agentsetting!=nil {
+		agentsetting, _ := s.store.FetchAgentSetting(taskGroup.GetAgentIp())
+		if agentsetting != nil {
 			agentsetting.Pods = append(agentsetting.Pods, taskGroup.ID)
 			err := s.store.SaveAgentSetting(agentsetting)
-			if err!=nil {
+			if err != nil {
 				blog.Errorf("save agentsetting %s pods error %s", agentsetting.InnerIP, err.Error())
 			}
 		} else {
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_reschedule.go b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_reschedule.go
index 796475acfd..4c4c5e3007 100644
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_reschedule.go
+++ b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_reschedule.go
@@ -16,7 +16,6 @@
 package scheduler
 
 import (
-
 	"net/http"
 	"time"
 
@@ -283,11 +282,11 @@ func (s *Scheduler) doRescheduleTrans(trans *Transaction, outOffer *offer.Offer,
 	//lock agentsetting
 	util.Lock.Lock(commtypes.BcsClusterAgentSetting{}, newTaskGroup.GetAgentIp())
 	//update agentsettings taskgroup index info
-	agentsetting,_ := s.store.FetchAgentSetting(newTaskGroup.GetAgentIp())
-	if agentsetting!=nil {
+	agentsetting, _ := s.store.FetchAgentSetting(newTaskGroup.GetAgentIp())
+	if agentsetting != nil {
 		agentsetting.Pods = append(agentsetting.Pods, newTaskGroup.ID)
 		err := s.store.SaveAgentSetting(agentsetting)
-		if err!=nil {
+		if err != nil {
 			blog.Errorf("save agentsetting %s pods error %s", agentsetting.InnerIP, err.Error())
 		}
 	} else {
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_scaleapp.go b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_scaleapp.go
index ea7710b896..cf26531f03 100644
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_scaleapp.go
+++ b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_scaleapp.go
@@ -263,11 +263,11 @@ func (s *Scheduler) doScaleUpAppTrans(trans *Transaction, outOffer *offer.Offer,
 		//lock agentsetting
 		util.Lock.Lock(commtypes.BcsClusterAgentSetting{}, taskGroup.GetAgentIp())
 		//update agentsettings taskgroup index info
-		agentsetting,_ := s.store.FetchAgentSetting(taskGroup.GetAgentIp())
-		if agentsetting!=nil {
+		agentsetting, _ := s.store.FetchAgentSetting(taskGroup.GetAgentIp())
+		if agentsetting != nil {
 			agentsetting.Pods = append(agentsetting.Pods, taskGroup.ID)
 			err := s.store.SaveAgentSetting(agentsetting)
-			if err!=nil {
+			if err != nil {
 				blog.Errorf("save agentsetting %s pods error %s", agentsetting.InnerIP, err.Error())
 			}
 		} else {
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_updateapp.go b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_updateapp.go
index 41098a5152..70348a1cd5 100644
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_updateapp.go
+++ b/bcs-mesos/bcs-scheduler/src/manager/sched/scheduler/trans_updateapp.go
@@ -220,11 +220,11 @@ func (s *Scheduler) doUpdateTrans(trans *Transaction, outOffer *offer.Offer, sta
 		//lock agentsetting
 		util.Lock.Lock(commtypes.BcsClusterAgentSetting{}, newTaskGroup.GetAgentIp())
 		//update agentsettings taskgroup index info
-		agentsetting,_ := s.store.FetchAgentSetting(newTaskGroup.GetAgentIp())
-		if agentsetting!=nil {
+		agentsetting, _ := s.store.FetchAgentSetting(newTaskGroup.GetAgentIp())
+		if agentsetting != nil {
 			agentsetting.Pods = append(agentsetting.Pods, newTaskGroup.ID)
 			err := s.store.SaveAgentSetting(agentsetting)
-			if err!=nil {
+			if err != nil {
 				blog.Errorf("save agentsetting %s pods error %s", agentsetting.InnerIP, err.Error())
 			}
 		} else {
diff --git a/bcs-mesos/bcs-scheduler/src/manager/sched/task/task.go b/bcs-mesos/bcs-scheduler/src/manager/sched/task/task.go
index bcdfa894ef..7857de9169 100644
--- a/bcs-mesos/bcs-scheduler/src/manager/sched/task/task.go
+++ b/bcs-mesos/bcs-scheduler/src/manager/sched/task/task.go
@@ -1308,9 +1308,9 @@ func CreateTaskGroupInfo(offer *mesos.Offer, version *types.Version, resources [
 		task.AgentHostname = *offer.Hostname
 		task.AgentIPAddress, _ = offerP.GetOfferIp(offer)
 		//if task contains extended resources, then set device plugin socket address int it
-		for _,ex :=range task.DataClass.ExtendedResources {
-			for _,re :=range offer.GetResources() {
-				if re.GetName()==ex.Name {
+		for _, ex := range task.DataClass.ExtendedResources {
+			for _, re := range offer.GetResources() {
+				if re.GetName() == ex.Name {
 					//device plugin socket setted in role parameter
 					ex.Socket = re.GetRole()
 				}
diff --git a/bcs-mesos/bcs-scheduler/src/types/application.go b/bcs-mesos/bcs-scheduler/src/types/application.go
index f62595676f..9a9cbc8af1 100644
--- a/bcs-mesos/bcs-scheduler/src/types/application.go
+++ b/bcs-mesos/bcs-scheduler/src/types/application.go
@@ -370,13 +370,13 @@ func (version *Version) AllResource() *Resource {
 
 //Container for Version
 type Container struct {
-	Type              string
-	Docker            *Docker
-	Volumes           []*Volume
-	Resources         *Resource
-	LimitResoures     *Resource
+	Type          string
+	Docker        *Docker
+	Volumes       []*Volume
+	Resources     *Resource
+	LimitResoures *Resource
 	//ExtendedResources []*commtypes.ExtendedResource
-	DataClass         *DataClass
+	DataClass *DataClass
 
 	ConfigMaps []commtypes.ConfigMap
 	Secrets    []commtypes.Secret
diff --git a/bcs-services/bcs-cpuset-device/cpuset-device/container.go b/bcs-services/bcs-cpuset-device/cpuset-device/container.go
index 28e1330c24..e163fa822b 100644
--- a/bcs-services/bcs-cpuset-device/cpuset-device/container.go
+++ b/bcs-services/bcs-cpuset-device/cpuset-device/container.go
@@ -14,82 +14,202 @@
 package cpuset_device
 
 import (
+	"fmt"
+	"os"
 	"strings"
+	"time"
 
 	"bk-bcs/bcs-common/common/blog"
 
 	docker "github.com/fsouza/go-dockerclient"
 )
 
+const CgroupCpusetRoot = "/sys/fs/cgroup/cpuset/docker"
+
+func (c *CpusetDevicePlugin) loopUpdateCpusetNodes() {
+	for {
+		time.Sleep(time.Minute * 10)
+		c.updateCpusetNodes()
+	}
+}
+
 func (c *CpusetDevicePlugin) updateCpusetNodes() error {
-	//cpuset env, example: node:0;cpuset:0,1,2,3
-	envs, err := c.listContainersCpusetEnvs()
+	c.lockNodes()
+	defer c.unlockNodes()
+
+	containers, err := c.client.ListContainers(docker.ListContainersOptions{})
 	if err != nil {
+		blog.Errorf("ListContainers failed: %s", err.Error())
 		return err
 	}
 
-	for _, env := range envs {
-		//node:0;cpuset:0,1,2,3
-		values := strings.Split(env, ";")
-		//node:0, nv[0]=node, nv[1]=0
-		nv := strings.Split(values[0], ":")
-		node, ok := c.nodes[nv[1]]
+	for _, node := range c.nodes {
+		node.AllocatedCpuset = make([]string, 0)
+	}
+	//traversal container for get allocated cpusets
+	for _, container := range containers {
+		info, err := c.client.InspectContainer(container.ID)
+		if err != nil {
+			blog.Errorf("inspect container %s failed %s, then continue", container.ID, err.Error())
+			continue
+		}
+		if info.State.Status != "running" {
+			blog.Infof("container %s status %s, and continue", container.ID, info.State.Status)
+			continue
+		}
+
+		//example:
+		//node=0
+		//cpusets=0,1,2,3
+		node, cpusets := c.getContainerCpusetInfo(info)
+		if node == "" || cpusets == "" {
+			continue
+		}
+
+		mNode, ok := c.nodes[node]
 		if !ok {
-			blog.Errorf("env %s invalid, node %s not found", env, nv[1])
+			blog.Errorf("container %s invalid, node %s not found", info.ID, node)
 			continue
 		}
-		//cpuset:0,1,2,3
-		//cv[0]=cpuset, cv[1]=0,1,2,3
-		cv := strings.Split(values[1], ":")
-		sets := strings.Split(cv[1], ",")
-		node.AllocatedCpuset = append(node.AllocatedCpuset, sets...)
+		mNode.AllocatedCpuset = append(mNode.AllocatedCpuset, strings.Split(cpusets, ",")...)
 	}
-	blog.Infof("update cpuset nodes success")
 
 	return nil
 }
 
-func (c *CpusetDevicePlugin) listContainersCpusetEnvs() ([]string, error) {
-	containers, err := c.client.ListContainers(docker.ListContainersOptions{})
+//start listen docker api event
+func (s *CpusetDevicePlugin) listenerDockerEvent() {
+	listener := make(chan *docker.APIEvents)
+	err := s.client.AddEventListener(listener)
 	if err != nil {
-		blog.Errorf("ListContainers failed: %s", err.Error())
-		return nil, err
+		blog.Errorf("listen docker event error %s", err.Error())
+		os.Exit(1)
 	}
 
-	//traversal container for get allocated cpusets
-	envs := make([]string, 0)
-	for _, container := range containers {
-		if container.Status != "running" {
-			blog.Infof("container %s status %s, and continue", container.ID, container.Status)
-			continue
-		}
-		info, err := c.client.InspectContainer(container.ID)
+	defer func() {
+		err = s.client.RemoveEventListener(listener)
 		if err != nil {
-			blog.Errorf("inspect container %s failed %s, then continue", container.ID, err.Error())
-			continue
+			blog.Errorf("remove docker event error  %s", err.Error())
 		}
+	}()
 
-		//cpuset env, example: node:0;cpuset:0,1,2,3
-		var envValue string
-		//docker env format []string
-		//example: []string{"k1=v1","k2=v2"...}
-		for _, o := range info.Config.Env {
-			//envs[0] is key, envs[1] is value
-			envs := strings.Split(o, "=")
-			if envs[0] == EnvBkbcsAllocateCpuset {
-				envValue = envs[1]
+	for {
+		var msg *docker.APIEvents
+		select {
+		case msg = <-listener:
+			blog.Infof("receive docker event action %s container %s", msg.Action, msg.ID)
+			c, err := s.client.InspectContainer(msg.ID)
+			if err != nil {
+				blog.Errorf("inspect container %s error %s", msg.ID, err.Error())
 				break
 			}
+
+			switch msg.Action {
+			//start container
+			case "start":
+				//set container group cpuset
+				s.setContainerCpuset(c)
+
+			// stop container
+			case "stop", "die":
+				//release container cpuset resources
+				s.releaseCpuset(c)
+			}
 		}
-		//if container don't contain bkbcs_allocate_cpuset env, then continue
-		if envValue == "" {
-			blog.Infof("container %s don't contain bkbcs_allocate_cpuset env, then continue", container.ID)
-			continue
+	}
+}
+
+//return
+func (s *CpusetDevicePlugin) getContainerCpusetInfo(c *docker.Container) (string, string) {
+	//cpuset env, example: node:0;cpuset:0,1,2,3
+	var envValue string
+	//docker env format []string
+	//example: []string{"k1=v1","k2=v2"...}
+	for _, o := range c.Config.Env {
+		//envs[0] is key, envs[1] is value
+		envs := strings.Split(o, "=")
+		if envs[0] == EnvBkbcsAllocateCpuset {
+			envValue = envs[1]
+			break
 		}
+	}
+	//if container don't contain bkbcs_allocate_cpuset env, then continue
+	if envValue == "" {
+		blog.Infof("container %s don't contain bkbcs_allocate_cpuset env, then continue", c.ID)
+		return "", ""
+	}
+	blog.Infof("container %s contains env(%s=%s)", c.ID, EnvBkbcsAllocateCpuset, envValue)
+
+	//node:0;cpuset:0,1,2,3
+	values := strings.Split(envValue, ";")
+	//node:0, nv[0]=node, nv[1]=0
+	nv := strings.Split(values[0], ":")
+	// 0
+	node := nv[1]
+	//cpuset:0,1,2,3
+	//cv[0]=cpuset, cv[1]=0,1,2,3
+	cv := strings.Split(values[1], ":")
+	//0,1,2,3
+	cpusets := cv[1]
+
+	return node, cpusets
+}
 
-		blog.Infof("container %s contains env(%s=%s)", container.ID, EnvBkbcsAllocateCpuset, envValue)
-		envs = append(envs, envValue)
+func (s *CpusetDevicePlugin) setContainerCpuset(c *docker.Container) {
+	node, cpusets := s.getContainerCpusetInfo(c)
+	if node == "" || cpusets == "" {
+		return
+	}
+	//set container cgroup cpuset.cpus、cpuset.mems
+	cpus := fmt.Sprintf("%s/%s/cpuset.cpus", CgroupCpusetRoot, c.ID)
+	fcpus, err := os.Create(cpus)
+	if err != nil {
+		blog.Errorf("open file %s error %s", cpus, err.Error())
+		return
+	}
+	defer fcpus.Close()
+	_, err = fcpus.WriteString(cpusets)
+	if err != nil {
+		blog.Errorf("write file %s error %s", cpus, err.Error())
+		return
 	}
 
-	return envs, nil
+	mems := fmt.Sprintf("%s/%s/cpuset.mems", CgroupCpusetRoot, c.ID)
+	fmems, err := os.Create(mems)
+	if err != nil {
+		blog.Errorf("open file %s error %s", cpus, err.Error())
+		return
+	}
+	defer fmems.Close()
+	_, err = fmems.WriteString(node)
+	if err != nil {
+		blog.Errorf("write file %s error %s", mems, err.Error())
+	}
+	blog.Infof("set container %s cpuset(%s) success", c.ID, cpusets)
+}
+
+func (s *CpusetDevicePlugin) releaseCpuset(c *docker.Container) {
+	node, cpusets := s.getContainerCpusetInfo(c)
+	if node == "" || cpusets == "" {
+		return
+	}
+
+	mNode, ok := s.nodes[node]
+	if !ok {
+		blog.Errorf("container %s node %s not found", c.ID, node)
+		return
+	}
+	mNode.ReleaseCpuset(strings.Split(cpusets, ","))
+}
+
+func (s *CpusetDevicePlugin) lockNodes() {
+	for _, node := range s.nodes {
+		node.Lock()
+	}
+}
+
+func (s *CpusetDevicePlugin) unlockNodes() {
+	for _, node := range s.nodes {
+		node.Unlock()
+	}
 }
diff --git a/bcs-services/bcs-cpuset-device/cpuset-device/manager.go b/bcs-services/bcs-cpuset-device/cpuset-device/manager.go
index e4d6c0b793..b95c1e65c7 100644
--- a/bcs-services/bcs-cpuset-device/cpuset-device/manager.go
+++ b/bcs-services/bcs-cpuset-device/cpuset-device/manager.go
@@ -100,6 +100,11 @@ func (c *CpusetDevicePlugin) Start() error {
 		return err
 	}
 
+	//watch docker create&stop event, handler container cpuset resources
+	go c.listenerDockerEvent()
+	//loop list containers to update cpuset node info
+	go c.loopUpdateCpusetNodes()
+
 	//start grpc server
 	err = c.serve()
 	if err != nil {
@@ -108,20 +113,20 @@ func (c *CpusetDevicePlugin) Start() error {
 	blog.Infof("grpc serve on %s success", c.cpusetSocket)
 
 	//if device plugin in k8s cluster, then register device plugin info to kubelet
-	if c.conf.Engine=="k8s" {
+	if c.conf.Engine == "k8s" {
 		err = c.register()
 		if err != nil {
 			blog.Errorf("register kubelet failed: %s", err.Error())
 			return err
 		}
 		blog.Infof("Registered device plugin for '%s' with Kubelet", c.resourceName)
-	//else device plugin in mesos cluster, then report extended resources info to mesos scheduler
-	}else {
+		//else device plugin in mesos cluster, then report extended resources info to mesos scheduler
+	} else {
 		err = c.reportMesosSchedulerExtendedResources()
-		if err!=nil {
+		if err != nil {
 			return err
 		}
-		blog.Infof("device plugin %s report mesos scheduler extended resources success",c.resourceName)
+		blog.Infof("device plugin %s report mesos scheduler extended resources success", c.resourceName)
 	}
 
 	return nil
@@ -214,7 +219,7 @@ func (c *CpusetDevicePlugin) initCpusetDevice() error {
 			blog.Errorf(err.Error())
 			return err
 		}
-		blog.Infof("numa node %d cpus(%v)", node, cpusets)
+		blog.Infof("numa node %s cpus(%v)", node, cpusets)
 
 		o := &types.CpusetNode{
 			Id:              node,
@@ -223,7 +228,7 @@ func (c *CpusetDevicePlugin) initCpusetDevice() error {
 		}
 		for _, cpuset := range cpusets {
 			device := &pluginapi.Device{
-				ID:     fmt.Sprintf("node%d:cpus%d", node, cpuset),
+				ID:     fmt.Sprintf("node:%s;cpuset:%s", node, cpuset),
 				Health: "Healthy",
 			}
 
@@ -237,16 +242,16 @@ func (c *CpusetDevicePlugin) initCpusetDevice() error {
 	return nil
 }
 
-func (c *CpusetDevicePlugin) reportMesosSchedulerExtendedResources()error{
+func (c *CpusetDevicePlugin) reportMesosSchedulerExtendedResources() error {
 	conf := &mesosdriver.Config{
-		ZkAddr: c.conf.BcsZk,
+		ZkAddr:     c.conf.BcsZk,
 		ClientCert: c.conf.ClientCert,
-		ClusterId: c.conf.ClusterId,
+		ClusterId:  c.conf.ClusterId,
 	}
 
-	client,err := mesosdriver.NewMesosPlatform(conf)
-	if err!=nil {
-		blog.Errorf("NewMesosPlatform failed: %s",err.Error())
+	client, err := mesosdriver.NewMesosPlatform(conf)
+	if err != nil {
+		blog.Errorf("NewMesosPlatform failed: %s", err.Error())
 		return err
 	}
 	ex := &commtypes.ExtendedResource{
@@ -256,8 +261,8 @@ func (c *CpusetDevicePlugin) reportMesosSchedulerExtendedResources()error{
 		Socket:   c.cpusetSocket,
 	}
 	err = client.UpdateAgentExtendedResources(ex)
-	if err!=nil {
-		blog.Errorf("Update Agent ExtendedResources failed: %s",err.Error())
+	if err != nil {
+		blog.Errorf("Update Agent ExtendedResources failed: %s", err.Error())
 		return err
 	}
 	return nil
@@ -279,7 +284,7 @@ func (c *CpusetDevicePlugin) Allocate(ctx context.Context, reqs *pluginapi.Alloc
 
 	responses := pluginapi.AllocateResponse{}
 	for _, req := range reqs.ContainerRequests {
-		blog.Infof("request allocate devices(%v)",req.DevicesIDs)
+		blog.Infof("request allocate devices(%v)", req.DevicesIDs)
 		var mnode *types.CpusetNode
 		for _, node := range c.nodes {
 			if node.Capacity() >= len(req.DevicesIDs) {
@@ -291,10 +296,14 @@ func (c *CpusetDevicePlugin) Allocate(ctx context.Context, reqs *pluginapi.Alloc
 		if mnode == nil {
 			return nil, fmt.Errorf("no enough cpuset to allocated container")
 		}
-		cpuset, _ := mnode.AllocateCpuset(len(req.DevicesIDs))
+		cpuset, err := mnode.AllocateCpuset(len(req.DevicesIDs))
+		if err != nil {
+			blog.Errorf(err.Error())
+			return nil, err
+		}
 		response := pluginapi.ContainerAllocateResponse{
 			Envs: map[string]string{
-				EnvBkbcsAllocateCpuset: strings.Join(cpuset, ","),
+				EnvBkbcsAllocateCpuset: fmt.Sprintf("node:%s;cpuset:%s", mnode.Id, strings.Join(cpuset, ",")),
 			},
 		}
 
diff --git a/bcs-services/bcs-cpuset-device/types/types.go b/bcs-services/bcs-cpuset-device/types/types.go
index e5614f8d32..b8bb6e3b12 100644
--- a/bcs-services/bcs-cpuset-device/types/types.go
+++ b/bcs-services/bcs-cpuset-device/types/types.go
@@ -63,6 +63,29 @@ func (c *CpusetNode) AllocateCpuset(number int) ([]string, error) {
 		}
 	}
 
-	blog.Infof("node %s allocate cpuset(%v), and AllocatedCpuset(%v)", c.Id, cpuset, c.AllocatedCpuset)
+	blog.Infof("node %s allocate cpuset(%v), and AllocatedCpuset(%v) AllCpuset(%v)",
+		c.Id, cpuset, c.AllocatedCpuset, c.Cpuset)
 	return cpuset, nil
 }
+
+func (c *CpusetNode) ReleaseCpuset(cpuset []string) {
+	c.Lock()
+	defer c.Unlock()
+
+	allocated := make([]string, 0)
+	for _, o := range c.AllocatedCpuset {
+		release := false
+		for _, r := range cpuset {
+			if o == r {
+				release = true
+				break
+			}
+		}
+		if !release {
+			allocated = append(allocated, o)
+		}
+	}
+	c.AllocatedCpuset = allocated
+	blog.Infof("node %s release cpuset(%v), and AllocatedCpuset(%v) AllCpuset(%v)",
+		c.Id, cpuset, c.AllocatedCpuset, c.Cpuset)
+}