PWX-35577: correlation tracing for Snapshot function (#2407)

Signed-off-by: Adam Krpan <[email protected]>

api/client/volume/client.go

@@ -198,11 +198,7 @@ func (v *volumeClient) Delete(ctx context.Context, volumeID string) error {
 // Snap specified volume. IO to the underlying volume should be quiesced before
 // calling this function.
 // Errors ErrEnoEnt may be returned
-func (v *volumeClient) Snapshot(volumeID string,
-	readonly bool,
-	locator *api.VolumeLocator,
-	noRetry bool,
-) (string, error) {
+func (v *volumeClient) Snapshot(ctx context.Context, volumeID string, readonly bool, locator *api.VolumeLocator, noRetry bool) (string, error) {
 	response := &api.SnapCreateResponse{}
 	request := &api.SnapCreateRequest{
 		Id:       volumeID,

api/server/sdk/volume_ops.go

@@ -109,7 +109,7 @@ func (s *VolumeServer) create(
 
 	// Check if the volume has already been created or is in process of creation
 	volName := locator.GetName()
-	v, err := util.VolumeFromName(ctx, s.driver(ctx), volName)
+	v, err := util.VolumeFromName(correlation.TODO(), s.driver(ctx), volName)
 	// If the volume is still there but it is being delete, then wait until it is removed
 	if err == nil && v.GetState() == api.VolumeState_VOLUME_STATE_DELETED {
 		if err = s.waitForVolumeRemoved(ctx, volName); err != nil {
@@ -171,7 +171,7 @@ func (s *VolumeServer) create(
 		}
 
 		// Create a snapshot from the parent
-		id, err = s.driver(ctx).Snapshot(parent.GetId(), false, &api.VolumeLocator{
+		id, err = s.driver(ctx).Snapshot(ctx, parent.GetId(), false, &api.VolumeLocator{
 			Name: volName,
 		}, false)
 		if err != nil {

api/server/sdk/volume_ops_test.go

@@ -313,7 +313,7 @@ func TestSdkVolumeClone(t *testing.T) {
 
 		s.MockDriver().
 			EXPECT().
-			Snapshot(parentid, false, &api.VolumeLocator{Name: name}, false).
+			Snapshot(gomock.Any(), parentid, false, &api.VolumeLocator{Name: name}, false).
 			Return(id, nil).
 			Times(1),
 
@@ -1087,7 +1087,7 @@ func TestSdkCloneOwnership(t *testing.T) {
 
 		mv.
 			EXPECT().
-			Snapshot(parentid, false, &api.VolumeLocator{Name: name}, false).
+			Snapshot(gomock.Any(), parentid, false, &api.VolumeLocator{Name: name}, false).
 			Return(id, nil).
 			Times(1),
 
@@ -1134,7 +1134,7 @@ func TestSdkCloneOwnership(t *testing.T) {
 
 		mv.
 			EXPECT().
-			Snapshot(parentid, false, &api.VolumeLocator{Name: name}, false).
+			Snapshot(gomock.Any(), parentid, false, &api.VolumeLocator{Name: name}, false).
 			Return(id, nil).
 			Times(1),
 
@@ -1205,7 +1205,7 @@ func TestSdkCloneOwnership(t *testing.T) {
 
 		mv.
 			EXPECT().
-			Snapshot(parentid, false, &api.VolumeLocator{Name: name}, false).
+			Snapshot(gomock.Any(), parentid, false, &api.VolumeLocator{Name: name}, false).
 			Return(id, nil).
 			Times(1),
 
@@ -1261,7 +1261,7 @@ func TestSdkCloneOwnership(t *testing.T) {
 
 		mv.
 			EXPECT().
-			Snapshot(parentid, false, &api.VolumeLocator{Name: name}, false).
+			Snapshot(gomock.Any(), parentid, false, &api.VolumeLocator{Name: name}, false).
 			Return(id, nil).
 			Times(1),
 

api/server/sdk/volume_snapshot.go

@@ -47,7 +47,7 @@ func (s *VolumeServer) SnapshotCreate(
 	}
 
 	readonly := true
-	snapshotID, err := s.driver(ctx).Snapshot(req.GetVolumeId(), readonly, &api.VolumeLocator{
+	snapshotID, err := s.driver(ctx).Snapshot(ctx, req.GetVolumeId(), readonly, &api.VolumeLocator{
 		Name:         req.GetName(),
 		VolumeLabels: req.GetLabels(),
 	}, false)

api/server/sdk/volume_snapshot_test.go

@@ -78,7 +78,7 @@ func TestSdkVolumeSnapshotCreate(t *testing.T) {
 		Times(1)
 	s.MockDriver().
 		EXPECT().
-		Snapshot(req.GetVolumeId(), true, &api.VolumeLocator{
+		Snapshot(gomock.Any(), req.GetVolumeId(), true, &api.VolumeLocator{
 			Name: snapName,
 		}, false).
 		Return(snapid, nil).

api/server/volume.go

@@ -1486,35 +1486,6 @@ func (vd *volAPI) volumeusage(w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(capacityInfo)
 }
 
-func (vd *volAPI) volumeBytesUsedByNode(w http.ResponseWriter, r *http.Request) {
-	var err error
-
-	method := "volumeBytesUsedByNode"
-	var req api.SdkVolumeBytesUsedRequest
-	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
-		vd.sendError(vd.name, method, w, err.Error(), http.StatusBadRequest)
-		return
-	}
-	d, err := vd.getVolDriver(r)
-	if err != nil {
-		notFound(w, r)
-		return
-	}
-
-	volUtilInfo, err := d.VolumeBytesUsedByNode(req.NodeId, req.Ids)
-	if err != nil {
-		var e error
-		if err != nil {
-			e = fmt.Errorf("Failed to get volumeBytesUsedByNode: %s", err.Error())
-		}
-		vd.sendError(vd.name, method, w, e.Error(), http.StatusInternalServerError)
-		return
-	}
-	var result api.SdkVolumeBytesUsedResponse
-	result.VolUtilInfo = volUtilInfo
-	json.NewEncoder(w).Encode(&result)
-}
-
 // swagger:operation GET /osd-volumes/quiesce/{id} volume quiesceVolume
 //
 // Quiesce volume with specified id.

api/server/volume_test.go

@@ -471,7 +471,7 @@ func TestVolumeSnapshotCreateSuccess(t *testing.T) {
 		Readonly: true,
 	}
 
-	_, err = driverclient.Snapshot(id, req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
+	_, err = driverclient.Snapshot(context.TODO(), id, req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
 	assert.Nil(t, err)
 
 	_, err = volumes.Delete(ctx, &api.SdkVolumeDeleteRequest{
@@ -529,7 +529,7 @@ func TestVolumeSnapshotCreateFailed(t *testing.T) {
 		Readonly: true,
 	}
 
-	res, _ := driverclient.Snapshot("doesnotexist", req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
+	res, _ := driverclient.Snapshot(context.TODO(), "doesnotexist", req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
 	assert.Equal(t, "", res)
 
 	_, err = volumes.Delete(ctx, &api.SdkVolumeDeleteRequest{
@@ -696,15 +696,15 @@ func TestVolumeSnapshotList(t *testing.T) {
 		Readonly: true,
 	}
 
-	_, err = driverclient.Snapshot(id, req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
+	_, err = driverclient.Snapshot(context.TODO(), id, req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
 	assert.Nil(t, err)
 
 	res, err := driverclient.SnapEnumerate([]string{id}, nil)
 	assert.Nil(t, err)
 	assert.NotNil(t, res)
 	assert.Len(t, res, 1)
 
-	_, err = driverclient.Snapshot(id, req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
+	_, err = driverclient.Snapshot(context.TODO(), id, req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
 	assert.Nil(t, err)
 
 	res, err = driverclient.SnapEnumerate([]string{id}, nil)
@@ -1650,7 +1650,7 @@ func TestVolumeRestoreSuccess(t *testing.T) {
 		Readonly: true,
 	}
 
-	res, err := driverclient.Snapshot(req2.GetId(), req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
+	res, err := driverclient.Snapshot(context.TODO(), req2.GetId(), req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
 	assert.Nil(t, err)
 
 	// create client
@@ -1711,7 +1711,7 @@ func TestVolumeRestoreFailed(t *testing.T) {
 		Readonly: true,
 	}
 
-	_, err = driverclient.Snapshot(req2.GetId(), req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
+	_, err = driverclient.Snapshot(context.TODO(), req2.GetId(), req2.GetReadonly(), req2.GetLocator(), req2.GetNoRetry())
 	assert.Nil(t, err)
 
 	// create client

cli/volumes.go

@@ -304,7 +304,7 @@ func (v *volDriver) snapCreate(cliContext *cli.Context) {
 	}
 	readonly := cliContext.Bool("readonly")
 	noRetry := cliContext.Bool("noretry")
-	id, err := v.volDriver.Snapshot(volumeID, readonly, locator, noRetry)
+	id, err := v.volDriver.Snapshot(context.TODO(), volumeID, readonly, locator, noRetry)
 	if err != nil {
 		cmdError(cliContext, fn, err)
 		return

csi/controller.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/libopenstorage/openstorage/api"
+	"github.com/libopenstorage/openstorage/pkg/correlation"
 	"github.com/libopenstorage/openstorage/pkg/grpcutil"
 	"github.com/libopenstorage/openstorage/pkg/units"
 	"github.com/libopenstorage/openstorage/pkg/util"
@@ -894,7 +895,7 @@ func (s *OsdCsiServer) CreateSnapshot(
 	volumes := api.NewOpenStorageVolumeClient(conn)
 
 	// Check if the snapshot with this name already exists
-	v, err := util.VolumeFromNameSdk(ctx, volumes, req.GetName())
+	v, err := util.VolumeFromNameSdk(correlation.TODO(), volumes, req.GetName())
 	if err == nil {
 		// Verify the parent is the same
 		if req.GetSourceVolumeId() != v.GetSource().GetParent() {

csi/controller_test.go

@@ -1324,7 +1324,7 @@ func TestControllerCreateVolumeBadSnapshot(t *testing.T) {
 		// Return an error from snapshot
 		s.MockDriver().
 			EXPECT().
-			Snapshot(parent, false, &api.VolumeLocator{Name: name}, false).
+			Snapshot(gomock.Any(), parent, false, &api.VolumeLocator{Name: name}, false).
 			Return("", fmt.Errorf("snapshoterr")).
 			Times(1),
 	)
@@ -1929,7 +1929,7 @@ func TestControllerCreateVolumeFromSnapshot(t *testing.T) {
 		// create
 		s.MockDriver().
 			EXPECT().
-			Snapshot(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+			Snapshot(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 			Return(snapID, nil).
 			Times(1),
 		s.MockDriver().
@@ -2036,7 +2036,7 @@ func TestControllerCreateVolumeSnapshotThroughParameters(t *testing.T) {
 		// create snap
 		s.MockDriver().
 			EXPECT().
-			Snapshot(mockParentID, false, &api.VolumeLocator{
+			Snapshot(gomock.Any(), mockParentID, false, &api.VolumeLocator{
 				Name: name,
 			},
 				false).
@@ -2925,7 +2925,7 @@ func TestControllerCreateSnapshot(t *testing.T) {
 		// snapshot
 		s.MockDriver().
 			EXPECT().
-			Snapshot(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+			Snapshot(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 			Return(snapId, nil).
 			Times(1),
 

csi/v0.3/controller.go

@@ -22,6 +22,7 @@ import (
 	"github.com/portworx/kvdb"
 
 	"github.com/libopenstorage/openstorage/api"
+	"github.com/libopenstorage/openstorage/pkg/correlation"
 	"github.com/libopenstorage/openstorage/pkg/util"
 
 	"github.com/golang/protobuf/ptypes"
@@ -108,7 +109,6 @@ func (s *OsdCsiServer) ControllerUnpublishVolume(
 // Note: The method used here to return errors is still not part of the spec.
 // See: https://github.com/container-storage-interface/spec/pull/115
 // Discussion:  https://groups.google.com/forum/#!topic/kubernetes-sig-storage-wg-csi/TpTrNFbRa1I
-//
 func (s *OsdCsiServer) ValidateVolumeCapabilities(
 	ctx context.Context,
 	req *csi.ValidateVolumeCapabilitiesRequest,
@@ -133,7 +133,7 @@ func (s *OsdCsiServer) ValidateVolumeCapabilities(
 		attributes)
 
 	// Check ID is valid with the specified volume capabilities
-	volumes, err := s.driver.Inspect([]string{id})
+	volumes, err := s.driver.Inspect(ctx, []string{id})
 	if err != nil || len(volumes) == 0 {
 		return nil, status.Error(codes.NotFound, "ID not found")
 	}
@@ -343,7 +343,7 @@ func (s *OsdCsiServer) CreateVolume(
 	}
 
 	// Check if the volume has already been created or is in process of creation
-	v, err := util.VolumeFromName(s.driver, req.GetName())
+	v, err := util.VolumeFromName(correlation.TODO(), s.driver, req.GetName())
 	if err == nil {
 		// Check the requested arguments match that of the existing volume
 		if spec.Size != v.GetSpec().GetSize() {
@@ -378,15 +378,15 @@ func (s *OsdCsiServer) CreateVolume(
 	var id string
 	if source != nil && len(source.GetParent()) != 0 {
 		// Get parent volume information
-		parent, err := util.VolumeFromName(s.driver, source.Parent)
+		parent, err := util.VolumeFromName(correlation.TODO(), s.driver, source.Parent)
 		if err != nil {
 			e := fmt.Sprintf("unable to get parent volume information: %s\n", err.Error())
 			logrus.Errorln(e)
 			return nil, status.Error(codes.InvalidArgument, e)
 		}
 
 		// Create a snapshot from the parent
-		id, err = s.driver.Snapshot(parent.GetId(), false, &api.VolumeLocator{
+		id, err = s.driver.Snapshot(ctx, parent.GetId(), false, &api.VolumeLocator{
 			Name: req.GetName(),
 		},
 			false)
@@ -415,7 +415,7 @@ func (s *OsdCsiServer) CreateVolume(
 	}
 
 	// id must have been set
-	v, err = util.VolumeFromName(s.driver, id)
+	v, err = util.VolumeFromName(correlation.TODO(), s.driver, id)
 	if err != nil {
 		e := fmt.Sprintf("Unable to find newly created volume: %s", err.Error())
 		logrus.Errorln(e)
@@ -440,7 +440,7 @@ func (s *OsdCsiServer) DeleteVolume(
 	}
 
 	// If the volume is not found, then we can return OK
-	volumes, err := s.driver.Inspect([]string{req.GetVolumeId()})
+	volumes, err := s.driver.Inspect(ctx, []string{req.GetVolumeId()})
 	if (err == nil && len(volumes) == 0) ||
 		(err != nil && err == kvdb.ErrNotFound) {
 		return &csi.DeleteVolumeResponse{}, nil
@@ -492,7 +492,7 @@ func (s *OsdCsiServer) CreateSnapshot(
 	}
 
 	// Check if the snapshot with this name already exists
-	v, err := util.VolumeFromName(s.driver, req.GetName())
+	v, err := util.VolumeFromName(correlation.TODO(), s.driver, req.GetName())
 	if err == nil {
 		// Verify the parent is the same
 		if req.GetSourceVolumeId() != v.GetSource().GetParent() {
@@ -526,7 +526,7 @@ func (s *OsdCsiServer) CreateSnapshot(
 
 	// Create snapshot
 	readonly := true
-	snapshotID, err := s.driver.Snapshot(req.GetSourceVolumeId(), readonly, &api.VolumeLocator{
+	snapshotID, err := s.driver.Snapshot(ctx, req.GetSourceVolumeId(), readonly, &api.VolumeLocator{
 		Name:         req.GetName(),
 		VolumeLabels: locator.GetVolumeLabels(),
 	}, false)
@@ -537,7 +537,7 @@ func (s *OsdCsiServer) CreateSnapshot(
 		return nil, status.Errorf(codes.Internal, "Failed to create snapshot: %v", err)
 	}
 
-	snapInfo, err := util.VolumeFromName(s.driver, snapshotID)
+	snapInfo, err := util.VolumeFromName(correlation.TODO(), s.driver, snapshotID)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Failed to get information about the snapshot: %v", err)
 	}
@@ -571,7 +571,7 @@ func (s *OsdCsiServer) DeleteSnapshot(
 	}
 
 	// If the snapshot is not found, then we can return OK
-	volumes, err := s.driver.Inspect([]string{req.GetSnapshotId()})
+	volumes, err := s.driver.Inspect(ctx, []string{req.GetSnapshotId()})
 	if (err == nil && len(volumes) == 0) ||
 		(err != nil && err == kvdb.ErrNotFound) {
 		return &csi.DeleteSnapshotResponse{}, nil

csi/v0.3/node.go

@@ -21,6 +21,7 @@ import (
 	"os"
 
 	"github.com/libopenstorage/openstorage/api"
+	"github.com/libopenstorage/openstorage/pkg/correlation"
 	"github.com/libopenstorage/openstorage/pkg/options"
 	"github.com/libopenstorage/openstorage/pkg/util"
 
@@ -71,7 +72,6 @@ func (s *OsdCsiServer) NodeGetId(
 // target path on the node.
 //
 // TODO: Support READ ONLY Mounts
-//
 func (s *OsdCsiServer) NodePublishVolume(
 	ctx context.Context,
 	req *csi.NodePublishVolumeRequest,
@@ -91,7 +91,7 @@ func (s *OsdCsiServer) NodePublishVolume(
 	}
 
 	// Get volume information
-	v, err := util.VolumeFromName(s.driver, req.GetVolumeId())
+	v, err := util.VolumeFromName(correlation.TODO(), s.driver, req.GetVolumeId())
 	if err != nil {
 		return nil, status.Errorf(codes.NotFound, "Volume id %s not found: %s",
 			req.GetVolumeId(),
@@ -197,7 +197,7 @@ func (s *OsdCsiServer) NodeUnpublishVolume(
 	}
 
 	// Get volume information
-	_, err := util.VolumeFromName(s.driver, req.GetVolumeId())
+	_, err := util.VolumeFromName(correlation.TODO(), s.driver, req.GetVolumeId())
 	if err != nil {
 		return nil, status.Errorf(codes.NotFound, "Volume id %s not found: %s",
 			req.GetVolumeId(),