Node Driver Registrar 源码分析

// cmd/csi-node-driver-registrar/main.go// Command line flagsvar (  connectionTimeout       = flag.Duration("connection-timeout", 0, "The --connection-timeout flag is deprecated")  csiAddress              = flag.String("csi-address", "/run/csi/socket", "Path of the CSI driver socket that the node-driver-registrar will connect to.")  kubeletRegistrationPath = flag.String("kubelet-registration-path", "", "Path of the CSI driver socket on the Kubernetes host machine.")  showVersion             = flag.Bool("version", false, "Show version.")  version                 = "unknown"
  // List of supported versions  supportedVersions = []string{"1.0.0"})

    containers:        - name: driver-registrar-rbd          # This is necessary only for systems with SELinux, where          # non-privileged sidecar containers cannot access unix domain socket          # created by privileged CSI driver container.          securityContext:            privileged: true          image: quay.io/k8scsi/csi/csi-node-driver-registrar:v1.3.0          args:            - "--v=5"            - "--csi-address=/csi/csi.sock"            - "--kubelet-registration-path=/var/lib/kubelet/plugins/rbd.csi.ceph.com/csi.sock"          lifecycle:            preStop:              exec:                command: [                  "/bin/sh", "-c",                  "rm -rf /registration/rbd.csi.ceph.com \                  /registration/rbd.csi.ceph.com-reg.sock"                ]          env:            - name: KUBE_NODE_NAME              valueFrom:                fieldRef:                  fieldPath: spec.nodeName          volumeMounts:            - name: socket-dir              mountPath: /csi            - name: registration-dir              mountPath: /registration          imagePullPolicy: "Always"    volumes:        - name: socket-dir          hostPath:            path: /var/lib/kubelet/plugins/rbd.csi.ceph.com            type: DirectoryOrCreate        - name: registration-dir          hostPath:            path: /var/lib/kubelet/plugins_registry/            type: Directory

// cmd/csi-node-driver-registrar/main.gofunc main() {  klog.InitFlags(nil)  flag.Set("logtostderr", "true")  flag.Parse()
  if *kubeletRegistrationPath == "" {    klog.Error("kubelet-registration-path is a required parameter")    os.Exit(1)  }
  if *showVersion {    fmt.Println(os.Args[0], version)    return  }  klog.Infof("Version: %s", version)
  if *connectionTimeout != 0 {    klog.Warning("--connection-timeout is deprecated and will have no effect")  }
  // Once https://github.com/container-storage-interface/spec/issues/159 is  // resolved, if plugin does not support PUBLISH_UNPUBLISH_VOLUME, then we  // can skip adding mapping to "csi.volume.kubernetes.io/nodeid" annotation.
  klog.V(1).Infof("Attempting to open a gRPC connection with: %q", *csiAddress)  csiConn, err := connection.Connect(*csiAddress)  if err != nil {    klog.Errorf("error connecting to CSI driver: %v", err)    os.Exit(1)  }
  klog.V(1).Infof("Calling CSI driver to discover driver name")  ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)  defer cancel()
  csiDriverName, err := csirpc.GetDriverName(ctx, csiConn)  if err != nil {    klog.Errorf("error retreiving CSI driver name: %v", err)    os.Exit(1)  }
  klog.V(2).Infof("CSI driver name: %q", csiDriverName)
  // Run forever  nodeRegister(csiDriverName)}

// cmd/csi-node-driver-registrar/node_register.gofunc nodeRegister(  csiDriverName string,) {  // When kubeletRegistrationPath is specified then driver-registrar ONLY acts  // as gRPC server which replies to registration requests initiated by kubelet's  // pluginswatcher infrastructure. Node labeling is done by kubelet's csi code.  registrar := newRegistrationServer(csiDriverName, *kubeletRegistrationPath, supportedVersions)  socketPath := fmt.Sprintf("/registration/%s-reg.sock", csiDriverName)  if err := util.CleanupSocketFile(socketPath); err != nil {    klog.Errorf("%+v", err)    os.Exit(1)  }
  var oldmask int  if runtime.GOOS == "linux" {    // Default to only user accessible socket, caller can open up later if desired    oldmask, _ = util.Umask(0077)  }
  klog.Infof("Starting Registration Server at: %s\n", socketPath)  lis, err := net.Listen("unix", socketPath)  if err != nil {    klog.Errorf("failed to listen on socket: %s with error: %+v", socketPath, err)    os.Exit(1)  }  if runtime.GOOS == "linux" {    util.Umask(oldmask)  }  klog.Infof("Registration Server started at: %s\n", socketPath)  grpcServer := grpc.NewServer()  // Registers kubelet plugin watcher api.  registerapi.RegisterRegistrationServer(grpcServer, registrar)
  // Starts service  if err := grpcServer.Serve(lis); err != nil {    klog.Errorf("Registration Server stopped serving: %v", err)    os.Exit(1)  }  // If gRPC server is gracefully shutdown, exit  os.Exit(0)}

// cmd/csi-node-driver-registrar/main.go// NewregistrationServer returns an initialized registrationServer instancefunc newRegistrationServer(driverName string, endpoint string, versions []string) registerapi.RegistrationServer {  return &registrationServer{    driverName: driverName,    endpoint:   endpoint,    version:    versions,  }}

// GetInfo is the RPC invoked by plugin watcherfunc (e registrationServer) GetInfo(ctx context.Context, req *registerapi.InfoRequest) (*registerapi.PluginInfo, error) {  klog.Infof("Received GetInfo call: %+v", req)  return &registerapi.PluginInfo{    Type:              registerapi.CSIPlugin,    Name:              e.driverName,    Endpoint:          e.endpoint,    SupportedVersions: e.version,  }, nil}

// GetInfo is the RPC invoked by plugin watcherfunc (e registrationServer) NotifyRegistrationStatus(ctx context.Context, status *registerapi.RegistrationStatus) (*registerapi.RegistrationStatusResponse, error) {  klog.Infof("Received NotifyRegistrationStatus call: %+v", status)  if !status.PluginRegistered {    klog.Errorf("Registration process failed with error: %+v, restarting registration container.", status.Error)    os.Exit(1)  }
  return &registerapi.RegistrationStatusResponse{}, nil}