open-nomad/plugins/csi/testing/client.go

180 lines
7.6 KiB
Go
Raw Normal View History

CSI Plugin Registration (#6555) This changeset implements the initial registration and fingerprinting of CSI Plugins as part of #5378. At a high level, it introduces the following: * A `csi_plugin` stanza as part of a Nomad task configuration, to allow a task to expose that it is a plugin. * A new task runner hook: `csi_plugin_supervisor`. This hook does two things. When the `csi_plugin` stanza is detected, it will automatically configure the plugin task to receive bidirectional mounts to the CSI intermediary directory. At runtime, it will then perform an initial heartbeat of the plugin and handle submitting it to the new `dynamicplugins.Registry` for further use by the client, and then run a lightweight heartbeat loop that will emit task events when health changes. * The `dynamicplugins.Registry` for handling plugins that run as Nomad tasks, in contrast to the existing catalog that requires `go-plugin` type plugins and to know the plugin configuration in advance. * The `csimanager` which fingerprints CSI plugins, in a similar way to `drivermanager` and `devicemanager`. It currently only fingerprints the NodeID from the plugin, and assumes that all plugins are monolithic. Missing features * We do not use the live updates of the `dynamicplugin` registry in the `csimanager` yet. * We do not deregister the plugins from the client when they shutdown yet, they just become indefinitely marked as unhealthy. This is deliberate until we figure out how we should manage deploying new versions of plugins/transitioning them.
2019-10-22 13:20:26 +00:00
package testing
import (
"context"
"fmt"
CSI Plugin Registration (#6555) This changeset implements the initial registration and fingerprinting of CSI Plugins as part of #5378. At a high level, it introduces the following: * A `csi_plugin` stanza as part of a Nomad task configuration, to allow a task to expose that it is a plugin. * A new task runner hook: `csi_plugin_supervisor`. This hook does two things. When the `csi_plugin` stanza is detected, it will automatically configure the plugin task to receive bidirectional mounts to the CSI intermediary directory. At runtime, it will then perform an initial heartbeat of the plugin and handle submitting it to the new `dynamicplugins.Registry` for further use by the client, and then run a lightweight heartbeat loop that will emit task events when health changes. * The `dynamicplugins.Registry` for handling plugins that run as Nomad tasks, in contrast to the existing catalog that requires `go-plugin` type plugins and to know the plugin configuration in advance. * The `csimanager` which fingerprints CSI plugins, in a similar way to `drivermanager` and `devicemanager`. It currently only fingerprints the NodeID from the plugin, and assumes that all plugins are monolithic. Missing features * We do not use the live updates of the `dynamicplugin` registry in the `csimanager` yet. * We do not deregister the plugins from the client when they shutdown yet, they just become indefinitely marked as unhealthy. This is deliberate until we figure out how we should manage deploying new versions of plugins/transitioning them.
2019-10-22 13:20:26 +00:00
csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
"google.golang.org/grpc"
)
// IdentityClient is a CSI identity client used for testing
type IdentityClient struct {
NextErr error
NextPluginInfo *csipbv1.GetPluginInfoResponse
NextPluginCapabilities *csipbv1.GetPluginCapabilitiesResponse
NextPluginProbe *csipbv1.ProbeResponse
}
// NewIdentityClient returns a new IdentityClient
func NewIdentityClient() *IdentityClient {
return &IdentityClient{}
}
func (f *IdentityClient) Reset() {
f.NextErr = nil
f.NextPluginInfo = nil
f.NextPluginCapabilities = nil
f.NextPluginProbe = nil
}
// GetPluginInfo returns plugin info
func (f *IdentityClient) GetPluginInfo(ctx context.Context, in *csipbv1.GetPluginInfoRequest, opts ...grpc.CallOption) (*csipbv1.GetPluginInfoResponse, error) {
return f.NextPluginInfo, f.NextErr
}
// GetPluginCapabilities implements csi method
func (f *IdentityClient) GetPluginCapabilities(ctx context.Context, in *csipbv1.GetPluginCapabilitiesRequest, opts ...grpc.CallOption) (*csipbv1.GetPluginCapabilitiesResponse, error) {
return f.NextPluginCapabilities, f.NextErr
}
// Probe implements csi method
func (f *IdentityClient) Probe(ctx context.Context, in *csipbv1.ProbeRequest, opts ...grpc.CallOption) (*csipbv1.ProbeResponse, error) {
return f.NextPluginProbe, f.NextErr
}
// ControllerClient is a CSI controller client used for testing
type ControllerClient struct {
NextErr error
NextCapabilitiesResponse *csipbv1.ControllerGetCapabilitiesResponse
NextPublishVolumeResponse *csipbv1.ControllerPublishVolumeResponse
NextUnpublishVolumeResponse *csipbv1.ControllerUnpublishVolumeResponse
NextValidateVolumeCapabilitiesResponse *csipbv1.ValidateVolumeCapabilitiesResponse
NextCreateVolumeResponse *csipbv1.CreateVolumeResponse
NextDeleteVolumeResponse *csipbv1.DeleteVolumeResponse
NextListVolumesResponse *csipbv1.ListVolumesResponse
2021-04-01 15:16:52 +00:00
NextCreateSnapshotResponse *csipbv1.CreateSnapshotResponse
NextDeleteSnapshotResponse *csipbv1.DeleteSnapshotResponse
NextListSnapshotsResponse *csipbv1.ListSnapshotsResponse
}
// NewControllerClient returns a new ControllerClient
func NewControllerClient() *ControllerClient {
return &ControllerClient{}
}
func (f *ControllerClient) Reset() {
f.NextErr = nil
f.NextCapabilitiesResponse = nil
f.NextPublishVolumeResponse = nil
f.NextUnpublishVolumeResponse = nil
f.NextValidateVolumeCapabilitiesResponse = nil
f.NextCreateVolumeResponse = nil
f.NextDeleteVolumeResponse = nil
f.NextListVolumesResponse = nil
2021-04-01 15:16:52 +00:00
f.NextCreateSnapshotResponse = nil
f.NextDeleteSnapshotResponse = nil
f.NextListSnapshotsResponse = nil
}
func (c *ControllerClient) ControllerGetCapabilities(ctx context.Context, in *csipbv1.ControllerGetCapabilitiesRequest, opts ...grpc.CallOption) (*csipbv1.ControllerGetCapabilitiesResponse, error) {
return c.NextCapabilitiesResponse, c.NextErr
}
func (c *ControllerClient) ControllerPublishVolume(ctx context.Context, in *csipbv1.ControllerPublishVolumeRequest, opts ...grpc.CallOption) (*csipbv1.ControllerPublishVolumeResponse, error) {
return c.NextPublishVolumeResponse, c.NextErr
}
func (c *ControllerClient) ControllerUnpublishVolume(ctx context.Context, in *csipbv1.ControllerUnpublishVolumeRequest, opts ...grpc.CallOption) (*csipbv1.ControllerUnpublishVolumeResponse, error) {
return c.NextUnpublishVolumeResponse, c.NextErr
}
func (c *ControllerClient) ValidateVolumeCapabilities(ctx context.Context, in *csipbv1.ValidateVolumeCapabilitiesRequest, opts ...grpc.CallOption) (*csipbv1.ValidateVolumeCapabilitiesResponse, error) {
return c.NextValidateVolumeCapabilitiesResponse, c.NextErr
}
2019-12-18 12:18:07 +00:00
func (c *ControllerClient) CreateVolume(ctx context.Context, in *csipbv1.CreateVolumeRequest, opts ...grpc.CallOption) (*csipbv1.CreateVolumeResponse, error) {
if in.VolumeContentSource != nil {
if in.VolumeContentSource.Type == nil || (in.VolumeContentSource.Type ==
&csipbv1.VolumeContentSource_Volume{
Volume: &csipbv1.VolumeContentSource_VolumeSource{VolumeId: ""},
}) || (in.VolumeContentSource.Type ==
&csipbv1.VolumeContentSource_Snapshot{
Snapshot: &csipbv1.VolumeContentSource_SnapshotSource{SnapshotId: ""},
}) {
return nil, fmt.Errorf("empty content source should be nil")
}
}
return c.NextCreateVolumeResponse, c.NextErr
}
func (c *ControllerClient) DeleteVolume(ctx context.Context, in *csipbv1.DeleteVolumeRequest, opts ...grpc.CallOption) (*csipbv1.DeleteVolumeResponse, error) {
return c.NextDeleteVolumeResponse, c.NextErr
}
func (c *ControllerClient) ListVolumes(ctx context.Context, in *csipbv1.ListVolumesRequest, opts ...grpc.CallOption) (*csipbv1.ListVolumesResponse, error) {
return c.NextListVolumesResponse, c.NextErr
}
2021-04-01 15:16:52 +00:00
func (c *ControllerClient) CreateSnapshot(ctx context.Context, in *csipbv1.CreateSnapshotRequest, opts ...grpc.CallOption) (*csipbv1.CreateSnapshotResponse, error) {
return c.NextCreateSnapshotResponse, c.NextErr
}
func (c *ControllerClient) DeleteSnapshot(ctx context.Context, in *csipbv1.DeleteSnapshotRequest, opts ...grpc.CallOption) (*csipbv1.DeleteSnapshotResponse, error) {
return c.NextDeleteSnapshotResponse, c.NextErr
}
func (c *ControllerClient) ListSnapshots(ctx context.Context, in *csipbv1.ListSnapshotsRequest, opts ...grpc.CallOption) (*csipbv1.ListSnapshotsResponse, error) {
return c.NextListSnapshotsResponse, c.NextErr
}
2019-12-18 12:18:07 +00:00
// NodeClient is a CSI Node client used for testing
type NodeClient struct {
NextErr error
NextCapabilitiesResponse *csipbv1.NodeGetCapabilitiesResponse
NextGetInfoResponse *csipbv1.NodeGetInfoResponse
NextStageVolumeResponse *csipbv1.NodeStageVolumeResponse
NextUnstageVolumeResponse *csipbv1.NodeUnstageVolumeResponse
NextPublishVolumeResponse *csipbv1.NodePublishVolumeResponse
NextUnpublishVolumeResponse *csipbv1.NodeUnpublishVolumeResponse
2019-12-18 12:18:07 +00:00
}
2020-01-24 16:20:23 +00:00
// NewNodeClient returns a new stub NodeClient
2019-12-18 12:18:07 +00:00
func NewNodeClient() *NodeClient {
return &NodeClient{}
}
func (f *NodeClient) Reset() {
f.NextErr = nil
f.NextCapabilitiesResponse = nil
f.NextGetInfoResponse = nil
f.NextStageVolumeResponse = nil
f.NextUnstageVolumeResponse = nil
2020-01-24 16:20:23 +00:00
f.NextPublishVolumeResponse = nil
f.NextUnpublishVolumeResponse = nil
2019-12-18 12:18:07 +00:00
}
func (c *NodeClient) NodeGetCapabilities(ctx context.Context, in *csipbv1.NodeGetCapabilitiesRequest, opts ...grpc.CallOption) (*csipbv1.NodeGetCapabilitiesResponse, error) {
return c.NextCapabilitiesResponse, c.NextErr
}
func (c *NodeClient) NodeGetInfo(ctx context.Context, in *csipbv1.NodeGetInfoRequest, opts ...grpc.CallOption) (*csipbv1.NodeGetInfoResponse, error) {
return c.NextGetInfoResponse, c.NextErr
}
func (c *NodeClient) NodeStageVolume(ctx context.Context, in *csipbv1.NodeStageVolumeRequest, opts ...grpc.CallOption) (*csipbv1.NodeStageVolumeResponse, error) {
return c.NextStageVolumeResponse, c.NextErr
}
func (c *NodeClient) NodeUnstageVolume(ctx context.Context, in *csipbv1.NodeUnstageVolumeRequest, opts ...grpc.CallOption) (*csipbv1.NodeUnstageVolumeResponse, error) {
return c.NextUnstageVolumeResponse, c.NextErr
}
2020-01-24 16:20:23 +00:00
func (c *NodeClient) NodePublishVolume(ctx context.Context, in *csipbv1.NodePublishVolumeRequest, opts ...grpc.CallOption) (*csipbv1.NodePublishVolumeResponse, error) {
return c.NextPublishVolumeResponse, c.NextErr
}
func (c *NodeClient) NodeUnpublishVolume(ctx context.Context, in *csipbv1.NodeUnpublishVolumeRequest, opts ...grpc.CallOption) (*csipbv1.NodeUnpublishVolumeResponse, error) {
return c.NextUnpublishVolumeResponse, c.NextErr
}