endpoint.go 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191
  1. /*
  2. Copyright 2017 The Kubernetes Authors.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package devicemanager
  14. import (
  15. "context"
  16. "fmt"
  17. "net"
  18. "sync"
  19. "time"
  20. "google.golang.org/grpc"
  21. "k8s.io/klog"
  22. pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1"
  23. )
  24. // endpoint maps to a single registered device plugin. It is responsible
  25. // for managing gRPC communications with the device plugin and caching
  26. // device states reported by the device plugin.
  27. type endpoint interface {
  28. run()
  29. stop()
  30. allocate(devs []string) (*pluginapi.AllocateResponse, error)
  31. preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error)
  32. callback(resourceName string, devices []pluginapi.Device)
  33. isStopped() bool
  34. stopGracePeriodExpired() bool
  35. }
  36. type endpointImpl struct {
  37. client pluginapi.DevicePluginClient
  38. clientConn *grpc.ClientConn
  39. socketPath string
  40. resourceName string
  41. stopTime time.Time
  42. mutex sync.Mutex
  43. cb monitorCallback
  44. }
  45. // newEndpointImpl creates a new endpoint for the given resourceName.
  46. // This is to be used during normal device plugin registration.
  47. func newEndpointImpl(socketPath, resourceName string, callback monitorCallback) (*endpointImpl, error) {
  48. client, c, err := dial(socketPath)
  49. if err != nil {
  50. klog.Errorf("Can't create new endpoint with path %s err %v", socketPath, err)
  51. return nil, err
  52. }
  53. return &endpointImpl{
  54. client: client,
  55. clientConn: c,
  56. socketPath: socketPath,
  57. resourceName: resourceName,
  58. cb: callback,
  59. }, nil
  60. }
  61. // newStoppedEndpointImpl creates a new endpoint for the given resourceName with stopTime set.
  62. // This is to be used during Kubelet restart, before the actual device plugin re-registers.
  63. func newStoppedEndpointImpl(resourceName string) *endpointImpl {
  64. return &endpointImpl{
  65. resourceName: resourceName,
  66. stopTime: time.Now(),
  67. }
  68. }
  69. func (e *endpointImpl) callback(resourceName string, devices []pluginapi.Device) {
  70. e.cb(resourceName, devices)
  71. }
  72. // run initializes ListAndWatch gRPC call for the device plugin and
  73. // blocks on receiving ListAndWatch gRPC stream updates. Each ListAndWatch
  74. // stream update contains a new list of device states.
  75. // It then issues a callback to pass this information to the device manager which
  76. // will adjust the resource available information accordingly.
  77. func (e *endpointImpl) run() {
  78. stream, err := e.client.ListAndWatch(context.Background(), &pluginapi.Empty{})
  79. if err != nil {
  80. klog.Errorf(errListAndWatch, e.resourceName, err)
  81. return
  82. }
  83. for {
  84. response, err := stream.Recv()
  85. if err != nil {
  86. klog.Errorf(errListAndWatch, e.resourceName, err)
  87. return
  88. }
  89. devs := response.Devices
  90. klog.V(2).Infof("State pushed for device plugin %s", e.resourceName)
  91. var newDevs []pluginapi.Device
  92. for _, d := range devs {
  93. newDevs = append(newDevs, *d)
  94. }
  95. e.callback(e.resourceName, newDevs)
  96. }
  97. }
  98. func (e *endpointImpl) isStopped() bool {
  99. e.mutex.Lock()
  100. defer e.mutex.Unlock()
  101. return !e.stopTime.IsZero()
  102. }
  103. func (e *endpointImpl) stopGracePeriodExpired() bool {
  104. e.mutex.Lock()
  105. defer e.mutex.Unlock()
  106. return !e.stopTime.IsZero() && time.Since(e.stopTime) > endpointStopGracePeriod
  107. }
  108. // used for testing only
  109. func (e *endpointImpl) setStopTime(t time.Time) {
  110. e.mutex.Lock()
  111. defer e.mutex.Unlock()
  112. e.stopTime = t
  113. }
  114. // allocate issues Allocate gRPC call to the device plugin.
  115. func (e *endpointImpl) allocate(devs []string) (*pluginapi.AllocateResponse, error) {
  116. if e.isStopped() {
  117. return nil, fmt.Errorf(errEndpointStopped, e)
  118. }
  119. return e.client.Allocate(context.Background(), &pluginapi.AllocateRequest{
  120. ContainerRequests: []*pluginapi.ContainerAllocateRequest{
  121. {DevicesIDs: devs},
  122. },
  123. })
  124. }
  125. // preStartContainer issues PreStartContainer gRPC call to the device plugin.
  126. func (e *endpointImpl) preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error) {
  127. if e.isStopped() {
  128. return nil, fmt.Errorf(errEndpointStopped, e)
  129. }
  130. ctx, cancel := context.WithTimeout(context.Background(), pluginapi.KubeletPreStartContainerRPCTimeoutInSecs*time.Second)
  131. defer cancel()
  132. return e.client.PreStartContainer(ctx, &pluginapi.PreStartContainerRequest{
  133. DevicesIDs: devs,
  134. })
  135. }
  136. func (e *endpointImpl) stop() {
  137. e.mutex.Lock()
  138. defer e.mutex.Unlock()
  139. if e.clientConn != nil {
  140. e.clientConn.Close()
  141. }
  142. e.stopTime = time.Now()
  143. }
  144. // dial establishes the gRPC communication with the registered device plugin. https://godoc.org/google.golang.org/grpc#Dial
  145. func dial(unixSocketPath string) (pluginapi.DevicePluginClient, *grpc.ClientConn, error) {
  146. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  147. defer cancel()
  148. c, err := grpc.DialContext(ctx, unixSocketPath, grpc.WithInsecure(), grpc.WithBlock(),
  149. grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
  150. return (&net.Dialer{}).DialContext(ctx, "unix", addr)
  151. }),
  152. )
  153. if err != nil {
  154. return nil, nil, fmt.Errorf(errFailedToDialDevicePlugin+" %v", err)
  155. }
  156. return pluginapi.NewDevicePluginClient(c), c, nil
  157. }