Merge pull request #60318 from jiayingz/api-change

pkg/kubelet/apis/deviceplugin/v1beta1/api.proto

@@ -46,6 +46,10 @@ message Empty {
 
 // DevicePlugin is the service advertised by Device Plugins
 service DevicePlugin {
+	// GetDevicePluginOptions returns options to be communicated with Device
+        // Manager
+	rpc GetDevicePluginOptions(Empty) returns (DevicePluginOptions) {}
+
 	// ListAndWatch returns a stream of List of Devices
 	// Whenever a Device state change or a Device disapears, ListAndWatch
 	// returns the new list
@@ -102,6 +106,10 @@ message PreStartContainerResponse {
 // - Allocate allows Device Plugin to run device specific operations on
 //   the Devices requested
 message AllocateRequest {
+	repeated ContainerAllocateRequest container_requests = 1;
+}
+
+message ContainerAllocateRequest {
 	repeated string devicesIDs = 1;
 }
 
@@ -114,6 +122,10 @@ message AllocateRequest {
 // The Device plugin should send a ListAndWatch update and fail the
 // Allocation request
 message AllocateResponse {
+	repeated ContainerAllocateResponse container_responses = 1;
+}
+
+message ContainerAllocateResponse {
   	// List of environment variable to be set in the container to access one of more devices.
 	map<string, string> envs = 1;
 	// Mounts for the container.

pkg/kubelet/cm/devicemanager/device_plugin_stub.go

@@ -130,6 +130,11 @@ func (m *Stub) Register(kubeletEndpoint, resourceName string, preStartContainerF
 	return nil
 }
 
+// GetDevicePluginOptions returns DevicePluginOptions settings for the device plugin.
+func (m *Stub) GetDevicePluginOptions(ctx context.Context, e *pluginapi.Empty) (*pluginapi.DevicePluginOptions, error) {
+	return &pluginapi.DevicePluginOptions{}, nil
+}
+
 // PreStartContainer resets the devices received
 func (m *Stub) PreStartContainer(ctx context.Context, r *pluginapi.PreStartContainerRequest) (*pluginapi.PreStartContainerResponse, error) {
 	log.Printf("PreStartContainer, %+v", r)

pkg/kubelet/cm/devicemanager/endpoint.go

@@ -179,7 +179,9 @@ func (e *endpointImpl) run() {
 // allocate issues Allocate gRPC call to the device plugin.
 func (e *endpointImpl) allocate(devs []string) (*pluginapi.AllocateResponse, error) {
 	return e.client.Allocate(context.Background(), &pluginapi.AllocateRequest{
-		DevicesIDs: devs,
+		ContainerRequests: []*pluginapi.ContainerAllocateRequest{
+			{DevicesIDs: devs},
+		},
 	})
 }
 

pkg/kubelet/cm/devicemanager/endpoint_test.go

@@ -130,24 +130,27 @@ func TestAllocate(t *testing.T) {
 	defer ecleanup(t, p, e)
 
 	resp := new(pluginapi.AllocateResponse)
-	resp.Devices = append(resp.Devices, &pluginapi.DeviceSpec{
+	contResp := new(pluginapi.ContainerAllocateResponse)
+	contResp.Devices = append(contResp.Devices, &pluginapi.DeviceSpec{
 		ContainerPath: "/dev/aaa",
 		HostPath:      "/dev/aaa",
 		Permissions:   "mrw",
 	})
 
-	resp.Devices = append(resp.Devices, &pluginapi.DeviceSpec{
+	contResp.Devices = append(contResp.Devices, &pluginapi.DeviceSpec{
 		ContainerPath: "/dev/bbb",
 		HostPath:      "/dev/bbb",
 		Permissions:   "mrw",
 	})
 
-	resp.Mounts = append(resp.Mounts, &pluginapi.Mount{
+	contResp.Mounts = append(contResp.Mounts, &pluginapi.Mount{
 		ContainerPath: "/container_dir1/file1",
 		HostPath:      "host_dir1/file1",
 		ReadOnly:      true,
 	})
 
+	resp.ContainerResponses = append(resp.ContainerResponses, contResp)
+
 	p.SetAllocFunc(func(r *pluginapi.AllocateRequest, devs map[string]pluginapi.Device) (*pluginapi.AllocateResponse, error) {
 		return resp, nil
 	})

pkg/kubelet/cm/devicemanager/manager.go

@@ -643,6 +643,8 @@ func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Cont
 		}
 
 		devs := allocDevices.UnsortedList()
+		// TODO: refactor this part of code to just append a ContainerAllocationRequest
+		// in a passed in AllocateRequest pointer, and issues a single Allocate call per pod.
 		glog.V(3).Infof("Making allocation request for devices %v for device plugin %s", devs, resource)
 		resp, err := e.allocate(devs)
 		metrics.DevicePluginAllocationLatency.WithLabelValues(resource).Observe(metrics.SinceInMicroseconds(startRPCTime))
@@ -657,7 +659,7 @@ func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Cont
 
 		// Update internal cached podDevices state.
 		m.mutex.Lock()
-		m.podDevices.insert(podUID, contName, resource, allocDevices, resp)
+		m.podDevices.insert(podUID, contName, resource, allocDevices, resp.ContainerResponses[0])
 		m.mutex.Unlock()
 	}
 

pkg/kubelet/cm/devicemanager/manager_test.go

@@ -279,8 +279,8 @@ func constructDevices(devices []string) sets.String {
 	return ret
 }
 
-func constructAllocResp(devices, mounts, envs map[string]string) *pluginapi.AllocateResponse {
-	resp := &pluginapi.AllocateResponse{}
+func constructAllocResp(devices, mounts, envs map[string]string) *pluginapi.ContainerAllocateResponse {
+	resp := &pluginapi.ContainerAllocateResponse{}
 	for k, v := range devices {
 		resp.Devices = append(resp.Devices, &pluginapi.DeviceSpec{
 			HostPath:      k,
@@ -458,7 +458,7 @@ func getTestManager(tmpDir string, activePods ActivePodsFunc, testRes []TestReso
 		if res.resourceName == "domain2.com/resource2" {
 			testManager.endpoints[res.resourceName] = &MockEndpoint{
 				allocateFunc: func(devs []string) (*pluginapi.AllocateResponse, error) {
-					resp := new(pluginapi.AllocateResponse)
+					resp := new(pluginapi.ContainerAllocateResponse)
 					resp.Envs = make(map[string]string)
 					for _, dev := range devs {
 						switch dev {
@@ -469,7 +469,9 @@ func getTestManager(tmpDir string, activePods ActivePodsFunc, testRes []TestReso
 							resp.Envs["key2"] = "val3"
 						}
 					}
-					return resp, nil
+					resps := new(pluginapi.AllocateResponse)
+					resps.ContainerResponses = append(resps.ContainerResponses, resp)
+					return resps, nil
 				},
 			}
 		}
@@ -774,8 +776,10 @@ func TestDevicePreStartContainer(t *testing.T) {
 	as.Contains(initializedDevs, "dev2")
 	as.Equal(len(initializedDevs), len(res1.devs))
 
-	expectedResp, err := allocateStubFunc()([]string{"dev1", "dev2"})
+	expectedResps, err := allocateStubFunc()([]string{"dev1", "dev2"})
 	as.Nil(err)
+	as.Equal(1, len(expectedResps.ContainerResponses))
+	expectedResp := expectedResps.ContainerResponses[0]
 	as.Equal(len(runContainerOpts.Devices), len(expectedResp.Devices))
 	as.Equal(len(runContainerOpts.Mounts), len(expectedResp.Mounts))
 	as.Equal(len(runContainerOpts.Envs), len(expectedResp.Envs))
@@ -783,7 +787,7 @@ func TestDevicePreStartContainer(t *testing.T) {
 
 func allocateStubFunc() func(devs []string) (*pluginapi.AllocateResponse, error) {
 	return func(devs []string) (*pluginapi.AllocateResponse, error) {
-		resp := new(pluginapi.AllocateResponse)
+		resp := new(pluginapi.ContainerAllocateResponse)
 		resp.Envs = make(map[string]string)
 		for _, dev := range devs {
 			switch dev {
@@ -822,6 +826,8 @@ func allocateStubFunc() func(devs []string) (*pluginapi.AllocateResponse, error)
 				resp.Envs["key1"] = "val1"
 			}
 		}
-		return resp, nil
+		resps := new(pluginapi.AllocateResponse)
+		resps.ContainerResponses = append(resps.ContainerResponses, resp)
+		return resps, nil
 	}
 }

pkg/kubelet/cm/devicemanager/pod_devices.go

@@ -28,7 +28,7 @@ type deviceAllocateInfo struct {
 	// deviceIds contains device Ids allocated to this container for the given resourceName.
 	deviceIds sets.String
 	// allocResp contains cached rpc AllocateResponse.
-	allocResp *pluginapi.AllocateResponse
+	allocResp *pluginapi.ContainerAllocateResponse
 }
 
 type resourceAllocateInfo map[string]deviceAllocateInfo // Keyed by resourceName.
@@ -43,7 +43,7 @@ func (pdev podDevices) pods() sets.String {
 	return ret
 }
 
-func (pdev podDevices) insert(podUID, contName, resource string, devices sets.String, resp *pluginapi.AllocateResponse) {
+func (pdev podDevices) insert(podUID, contName, resource string, devices sets.String, resp *pluginapi.ContainerAllocateResponse) {
 	if _, podExists := pdev[podUID]; !podExists {
 		pdev[podUID] = make(containerDevices)
 	}
@@ -168,7 +168,7 @@ func (pdev podDevices) fromCheckpointData(data []podDevicesCheckpointEntry) {
 		for _, devID := range entry.DeviceIDs {
 			devIDs.Insert(devID)
 		}
-		allocResp := &pluginapi.AllocateResponse{}
+		allocResp := &pluginapi.ContainerAllocateResponse{}
 		err := allocResp.Unmarshal(entry.AllocResp)
 		if err != nil {
 			glog.Errorf("Can't unmarshal allocResp for %v %v %v: %v", entry.PodUID, entry.ContainerName, entry.ResourceName, err)

test/e2e_node/device_plugin.go

@@ -229,34 +229,38 @@ func numberOfDevices(node *v1.Node, resourceName string) int64 {
 
 // stubAllocFunc will pass to stub device plugin
 func stubAllocFunc(r *pluginapi.AllocateRequest, devs map[string]pluginapi.Device) (*pluginapi.AllocateResponse, error) {
-	var response pluginapi.AllocateResponse
-	for _, requestID := range r.DevicesIDs {
-		dev, ok := devs[requestID]
-		if !ok {
-			return nil, fmt.Errorf("invalid allocation request with non-existing device %s", requestID)
-		}
+	var responses pluginapi.AllocateResponse
+	for _, req := range r.ContainerRequests {
+		response := &pluginapi.ContainerAllocateResponse{}
+		for _, requestID := range req.DevicesIDs {
+			dev, ok := devs[requestID]
+			if !ok {
+				return nil, fmt.Errorf("invalid allocation request with non-existing device %s", requestID)
+			}
 
-		if dev.Health != pluginapi.Healthy {
-			return nil, fmt.Errorf("invalid allocation request with unhealthy device: %s", requestID)
-		}
+			if dev.Health != pluginapi.Healthy {
+				return nil, fmt.Errorf("invalid allocation request with unhealthy device: %s", requestID)
+			}
 
-		// create fake device file
-		fpath := filepath.Join("/tmp", dev.ID)
+			// create fake device file
+			fpath := filepath.Join("/tmp", dev.ID)
 
-		// clean first
-		os.RemoveAll(fpath)
-		f, err := os.Create(fpath)
-		if err != nil && !os.IsExist(err) {
-			return nil, fmt.Errorf("failed to create fake device file: %s", err)
-		}
+			// clean first
+			os.RemoveAll(fpath)
+			f, err := os.Create(fpath)
+			if err != nil && !os.IsExist(err) {
+				return nil, fmt.Errorf("failed to create fake device file: %s", err)
+			}
 
-		f.Close()
+			f.Close()
 
-		response.Mounts = append(response.Mounts, &pluginapi.Mount{
-			ContainerPath: fpath,
-			HostPath:      fpath,
-		})
+			response.Mounts = append(response.Mounts, &pluginapi.Mount{
+				ContainerPath: fpath,
+				HostPath:      fpath,
+			})
+		}
+		responses.ContainerResponses = append(responses.ContainerResponses, response)
 	}
 
-	return &response, nil
+	return &responses, nil
 }