Add cluster ping to publish

If the IP address for a cluster is not reachable, it takes
a very long time for the mount to time out, holding up K8S
operations and preventing quick deletion of incorrect volumes.

This change adds ping operation to ensure that the node is
able to communicate with the expected cluster before attempting
the mount operation. The result is cached for 30 seconds to
prevent overwhelming the API if many volumes are being
published in a short period of time.

Author: dabradley

pkg/azurelustre/azurelustre.go

@@ -41,6 +41,7 @@ import (
 	csicommon "sigs.k8s.io/azurelustre-csi-driver/pkg/csi-common"
 	"sigs.k8s.io/azurelustre-csi-driver/pkg/util"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/configloader"
+	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
 	azureconfig "sigs.k8s.io/cloud-provider-azure/pkg/provider/config"
 )
@@ -151,6 +152,8 @@ type Driver struct {
 	resourceGroup      string
 	location           string
 	dynamicProvisioner DynamicProvisionerInterface
+	commandRunner      util.CommandRunnerInterface
+	pingCache          azcache.Resource
 
 	removeNotReadyTaint bool
 	kubeClient          kubernetes.Interface
@@ -170,6 +173,7 @@ func NewDriver(options *DriverOptions) *Driver {
 		enableAzureLustreMockDynProv: options.EnableAzureLustreMockDynProv,
 		workingMountDir:              options.WorkingMountDir,
 		removeNotReadyTaint:          options.RemoveNotReadyTaint,
+		commandRunner:                &util.DefaultCommandRunner{},
 	}
 	d.Name = options.DriverName
 	d.Version = driverVersion
@@ -263,6 +267,10 @@ func NewDriver(options *DriverOptions) *Driver {
 		}
 	}
 
+	if d.pingCache, err = azcache.NewTimedCache(20*time.Second, d.pingCluster, false); err != nil {
+		klog.Fatalf("%v", err)
+	}
+
 	return &d
 }
 

pkg/azurelustre/azurelustre_test.go

@@ -39,6 +39,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	kubefake "k8s.io/client-go/kubernetes/fake"
+	"sigs.k8s.io/azurelustre-csi-driver/pkg/util"
+	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
 )
 
@@ -56,6 +58,7 @@ const (
 	clusterRequestFailureName = "testShouldFail"
 	driverDefaultLocation     = "defaultFakeLocation"
 	emptyZonesLocation        = "emptyZonesLocation"
+	fakeCacheTTL              = 1 * time.Second
 )
 
 func NewFakeDriver() *Driver {
@@ -72,10 +75,39 @@ func NewFakeDriver() *Driver {
 	driver.location = driverDefaultLocation
 	driver.resourceGroup = "defaultFakeResourceGroup"
 	driver.dynamicProvisioner = &FakeDynamicProvisioner{}
+	driver.commandRunner = NewFakeCommandRunner(false)
+
+	getter := func(_ context.Context, _ string) (interface{}, error) {
+		return true, nil
+	}
+
+	driver.pingCache, _ = azcache.NewTimedCache(fakeCacheTTL, getter, false) //nolint:errcheck // TimedCache never returns error here
+	driver.pingCache.Set("key", nil)
 
 	return driver
 }
 
+type FakeCommandRunner struct {
+	util.CommandRunnerInterface
+	CalledCommands []string
+	cmdShouldFail  bool
+}
+
+func NewFakeCommandRunner(shouldFail bool) *FakeCommandRunner {
+	return &FakeCommandRunner{
+		CalledCommands: []string{},
+		cmdShouldFail:  shouldFail,
+	}
+}
+
+func (f *FakeCommandRunner) RunWithTimeout(_ context.Context, _ time.Duration, cmd string, args ...string) (string, error) {
+	f.CalledCommands = append(f.CalledCommands, fmt.Sprintf("%s %s", cmd, strings.Join(args, " ")))
+	if f.cmdShouldFail {
+		return "", errors.New("error occurred calling command: " + cmd)
+	}
+	return "fake command output", nil
+}
+
 type FakeDynamicProvisioner struct {
 	DynamicProvisionerInterface
 	Filesystems   []*AmlFilesystemProperties

pkg/azurelustre/nodeserver.go

@@ -31,12 +31,41 @@ import (
 	"k8s.io/kubernetes/pkg/volume"
 	mount "k8s.io/mount-utils"
 	volumehelper "sigs.k8s.io/azurelustre-csi-driver/pkg/util"
+	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	"sigs.k8s.io/cloud-provider-azure/pkg/metrics"
 )
 
+// pingCluster used as the getter function for pingCache. Pings the MGS IP address to see if it's reachable.
+// Allows the ping to timeout after 5 seconds to avoid long delays
+func (d *Driver) pingCluster(ctx context.Context, mgsIPAddress string) (interface{}, error) {
+	pingAddr := fmt.Sprintf("%s@tcp", mgsIPAddress)
+	_, err := d.commandRunner.RunWithTimeout(ctx, 5*time.Second, "lnetctl", "ping", pingAddr)
+	if err != nil {
+		klog.Warningf("lnetctl ping to %s failed with error: %v", pingAddr, err)
+		return false, nil
+	}
+	return true, nil
+}
+
+// getCachedClusterPing checks whether the MGS IP address is reachable by looking it up in the pingCache.
+// If not found or expired, it will ping the MGS IP address and update the cache.
+func (d *Driver) getCachedClusterPing(ctx context.Context, mgsIPAddress string) error {
+	cacheData, err := d.pingCache.Get(ctx, mgsIPAddress, azcache.CacheReadTypeDefault)
+	if err != nil {
+		return status.Errorf(codes.Internal, "error getting pingCache value for MGS IP %q: %v", mgsIPAddress, err)
+	}
+	if pinged, ok := cacheData.(bool); !ok {
+		return status.Error(codes.Internal, "type assertion failed for pingCache value")
+	} else if !pinged {
+		return status.Errorf(codes.InvalidArgument,
+			"MGS IP address %q didn't respond to lnetctl ping: Is the cluster IP address correct and the network configured correctly?", mgsIPAddress)
+	}
+	return nil
+}
+
 // NodePublishVolume mount the volume from staging to target path
 func (d *Driver) NodePublishVolume(
-	_ context.Context,
+	ctx context.Context,
 	req *csi.NodePublishVolumeRequest,
 ) (*csi.NodePublishVolumeResponse, error) {
 	mc := metrics.NewMetricContext(azureLustreCSIDriverName,
@@ -64,13 +93,13 @@ func (d *Driver) NodePublishVolume(
 			"Target path not provided")
 	}
 
-	context := req.GetVolumeContext()
-	if context == nil {
+	volumeContext := req.GetVolumeContext()
+	if volumeContext == nil {
 		return nil, status.Error(codes.InvalidArgument,
 			"Volume context must be provided")
 	}
 
-	vol, err := getVolume(volumeID, context)
+	vol, err := getVolume(volumeID, volumeContext)
 	if err != nil {
 		return nil, err
 	}
@@ -88,12 +117,20 @@ func (d *Driver) NodePublishVolume(
 		mc.ObserveOperationWithResult(isOperationSucceeded)
 	}()
 
+	if !d.enableAzureLustreMockMount {
+		klog.V(2).Infof("NodePublishVolume: ensuring MGS IP address %s responds to ping before attempting mount", vol.mgsIPAddress)
+		if err := d.getCachedClusterPing(ctx, vol.mgsIPAddress); err != nil {
+			return nil, err
+		}
+		klog.V(2).Infof("NodePublishVolume: ping to MGS IP address %s successful", vol.mgsIPAddress)
+	}
+
 	source := getSourceString(vol.mgsIPAddress, vol.azureLustreName)
 
 	mountOptions, readOnly := getMountOptions(req, userMountFlags)
 
 	if len(vol.subDir) > 0 && !d.enableAzureLustreMockMount {
-		interpolatedSubDir := interpolateSubDirVariables(context, vol)
+		interpolatedSubDir := interpolateSubDirVariables(volumeContext, vol)
 
 		if isSubpath := ensureStrictSubpath(interpolatedSubDir); !isSubpath {
 			return nil, status.Error(

pkg/azurelustre/nodeserver_test.go

@@ -521,11 +521,25 @@ func TestNodePublishVolume(t *testing.T) {
 				d.volumeLocks.Release(lockKey)
 			},
 		},
+		{
+			desc: "Error failed ping to MGS",
+			setup: func(d *Driver) {
+				d.pingCache.Set("1.1.1.1", false)
+			},
+			req: csi.NodePublishVolumeRequest{
+				VolumeCapability: &csi.VolumeCapability{AccessMode: &volumeCap},
+				VolumeId:         "vol_1#lustrefs#1.1.1.1#",
+				TargetPath:       targetTest,
+				VolumeContext:    map[string]string{"mgs-ip-address": "1.1.1.1", "fs-name": "lustrefs"},
+			},
+			expectedErr:          status.Error(codes.InvalidArgument, "MGS IP address \"1.1.1.1\" didn't respond to lnetctl ping: Is the cluster IP address correct and the network configured correctly?"),
+			expectedMountpoints:  nil,
+			expectedMountActions: []mount.FakeAction{},
+		},
 	}
 
-	d := NewFakeDriver()
-
 	for i := range tests {
+		d := NewFakeDriver()
 		test := &tests[i]
 
 		fakeMounter := &fakeMounter{}

pkg/util/util.go

@@ -17,10 +17,13 @@ limitations under the License.
 package util
 
 import (
+	"context"
 	"fmt"
 	"os"
+	"os/exec"
 	"strings"
 	"sync"
+	"time"
 )
 
 const (
@@ -87,6 +90,26 @@ func MakeDir(pathname string) error {
 	return nil
 }
 
+type CommandRunnerInterface interface {
+	RunWithTimeout(ctx context.Context, timeout time.Duration, cmd string, args ...string) (string, error)
+}
+
+type DefaultCommandRunner struct {
+	CommandRunnerInterface
+}
+
+func (r *DefaultCommandRunner) RunWithTimeout(ctx context.Context, timeout time.Duration, cmd string, args ...string) (string, error) {
+	cmdTimeout, cmdCancel := context.WithTimeout(ctx, timeout)
+	defer cmdCancel()
+
+	command := exec.CommandContext(cmdTimeout, cmd, args...)
+	output, err := command.CombinedOutput()
+	if err != nil {
+		return string(output), err
+	}
+	return string(output), nil
+}
+
 // LockMap used to lock on entries
 type LockMap struct {
 	sync.Mutex

pkg/util/util_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package util
 
 import (
+	"context"
 	"errors"
 	"os"
 	"reflect"
@@ -42,6 +43,30 @@ func TestRoundUpGiB(t *testing.T) {
 	}
 }
 
+func TestCommandRunnerSuccess(t *testing.T) {
+	runner := &DefaultCommandRunner{}
+
+	output, err := runner.RunWithTimeout(context.Background(), 1*time.Second, "echo", "hello")
+	require.NoError(t, err)
+	require.Equal(t, "hello\n", output)
+}
+
+func TestCommandRunnerTimeout(t *testing.T) {
+	runner := &DefaultCommandRunner{}
+	output, err := runner.RunWithTimeout(context.Background(), 1*time.Second, "sleep", "10")
+	require.ErrorContains(t, err, "killed")
+	require.Empty(t, output, "Expected no output on timeout")
+}
+
+func TestCommandRunnerError(t *testing.T) {
+	runner := &DefaultCommandRunner{}
+
+	nonexistentPath := "./non-existent-path"
+	output, err := runner.RunWithTimeout(context.Background(), 1*time.Second, "ls", nonexistentPath)
+	require.Error(t, err)
+	require.Contains(t, output, nonexistentPath)
+}
+
 func TestSimpleLockEntry(t *testing.T) {
 	testLockMap := NewLockMap()