426c26d7c0
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.
192 lines
4.7 KiB
Go
192 lines
4.7 KiB
Go
package csi
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"testing"
|
|
|
|
csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
|
|
"github.com/golang/protobuf/ptypes/wrappers"
|
|
fake "github.com/hashicorp/nomad/plugins/csi/testing"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func newTestClient() (*fake.IdentityClient, CSIPlugin) {
|
|
ic := &fake.IdentityClient{}
|
|
client := &client{
|
|
identityClient: ic,
|
|
}
|
|
|
|
return ic, client
|
|
}
|
|
|
|
func TestClient_RPC_PluginProbe(t *testing.T) {
|
|
cases := []struct {
|
|
Name string
|
|
ResponseErr error
|
|
ProbeResponse *csipbv1.ProbeResponse
|
|
ExpectedResponse bool
|
|
ExpectedErr error
|
|
}{
|
|
{
|
|
Name: "handles underlying grpc errors",
|
|
ResponseErr: fmt.Errorf("some grpc error"),
|
|
ExpectedErr: fmt.Errorf("some grpc error"),
|
|
},
|
|
{
|
|
Name: "returns false for ready when the provider returns false",
|
|
ProbeResponse: &csipbv1.ProbeResponse{
|
|
Ready: &wrappers.BoolValue{Value: false},
|
|
},
|
|
ExpectedResponse: false,
|
|
},
|
|
{
|
|
Name: "returns true for ready when the provider returns true",
|
|
ProbeResponse: &csipbv1.ProbeResponse{
|
|
Ready: &wrappers.BoolValue{Value: true},
|
|
},
|
|
ExpectedResponse: true,
|
|
},
|
|
{
|
|
/* When a SP does not return a ready value, a CO MAY treat this as ready.
|
|
We do so because example plugins rely on this behaviour. We may
|
|
re-evaluate this decision in the future. */
|
|
Name: "returns true for ready when the provider returns a nil wrapper",
|
|
ProbeResponse: &csipbv1.ProbeResponse{
|
|
Ready: nil,
|
|
},
|
|
ExpectedResponse: true,
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
ic, client := newTestClient()
|
|
defer client.Close()
|
|
|
|
ic.NextErr = c.ResponseErr
|
|
ic.NextPluginProbe = c.ProbeResponse
|
|
|
|
resp, err := client.PluginProbe(context.TODO())
|
|
if c.ExpectedErr != nil {
|
|
require.Error(t, c.ExpectedErr, err)
|
|
}
|
|
|
|
require.Equal(t, c.ExpectedResponse, resp)
|
|
})
|
|
}
|
|
|
|
}
|
|
|
|
func TestClient_RPC_PluginInfo(t *testing.T) {
|
|
cases := []struct {
|
|
Name string
|
|
ResponseErr error
|
|
InfoResponse *csipbv1.GetPluginInfoResponse
|
|
ExpectedResponse string
|
|
ExpectedErr error
|
|
}{
|
|
{
|
|
Name: "handles underlying grpc errors",
|
|
ResponseErr: fmt.Errorf("some grpc error"),
|
|
ExpectedErr: fmt.Errorf("some grpc error"),
|
|
},
|
|
{
|
|
Name: "returns an error if we receive an empty `name`",
|
|
InfoResponse: &csipbv1.GetPluginInfoResponse{
|
|
Name: "",
|
|
},
|
|
ExpectedErr: fmt.Errorf("PluginGetInfo: plugin returned empty name field"),
|
|
},
|
|
{
|
|
Name: "returns the name when successfully retrieved and not empty",
|
|
InfoResponse: &csipbv1.GetPluginInfoResponse{
|
|
Name: "com.hashicorp.storage",
|
|
},
|
|
ExpectedResponse: "com.hashicorp.storage",
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
ic, client := newTestClient()
|
|
defer client.Close()
|
|
|
|
ic.NextErr = c.ResponseErr
|
|
ic.NextPluginInfo = c.InfoResponse
|
|
|
|
resp, err := client.PluginGetInfo(context.TODO())
|
|
if c.ExpectedErr != nil {
|
|
require.Error(t, c.ExpectedErr, err)
|
|
}
|
|
|
|
require.Equal(t, c.ExpectedResponse, resp)
|
|
})
|
|
}
|
|
|
|
}
|
|
|
|
func TestClient_RPC_PluginGetCapabilities(t *testing.T) {
|
|
cases := []struct {
|
|
Name string
|
|
ResponseErr error
|
|
Response *csipbv1.GetPluginCapabilitiesResponse
|
|
ExpectedResponse *PluginCapabilitySet
|
|
ExpectedErr error
|
|
}{
|
|
{
|
|
Name: "handles underlying grpc errors",
|
|
ResponseErr: fmt.Errorf("some grpc error"),
|
|
ExpectedErr: fmt.Errorf("some grpc error"),
|
|
},
|
|
{
|
|
Name: "HasControllerService is true when it's part of the response",
|
|
Response: &csipbv1.GetPluginCapabilitiesResponse{
|
|
Capabilities: []*csipbv1.PluginCapability{
|
|
{
|
|
Type: &csipbv1.PluginCapability_Service_{
|
|
Service: &csipbv1.PluginCapability_Service{
|
|
Type: csipbv1.PluginCapability_Service_CONTROLLER_SERVICE,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
ExpectedResponse: &PluginCapabilitySet{hasControllerService: true},
|
|
},
|
|
{
|
|
Name: "HasTopologies is true when it's part of the response",
|
|
Response: &csipbv1.GetPluginCapabilitiesResponse{
|
|
Capabilities: []*csipbv1.PluginCapability{
|
|
{
|
|
Type: &csipbv1.PluginCapability_Service_{
|
|
Service: &csipbv1.PluginCapability_Service{
|
|
Type: csipbv1.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
ExpectedResponse: &PluginCapabilitySet{hasTopologies: true},
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.Name, func(t *testing.T) {
|
|
ic, client := newTestClient()
|
|
defer client.Close()
|
|
|
|
ic.NextErr = c.ResponseErr
|
|
ic.NextPluginCapabilities = c.Response
|
|
|
|
resp, err := client.PluginGetCapabilities(context.TODO())
|
|
if c.ExpectedErr != nil {
|
|
require.Error(t, c.ExpectedErr, err)
|
|
}
|
|
|
|
require.Equal(t, c.ExpectedResponse, resp)
|
|
})
|
|
}
|
|
|
|
}
|