Skip to content

Commit

Permalink
replication: set replicationSource
Browse files Browse the repository at this point in the history 
When setting the volumeID set the replication
Source of the volumeID as well as in csi spec
volumeID is deprecated and replicationSource
is the right one going forward.

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>
  • Loading branch information
@Madhu-1
Madhu-1 committed Jun 19, 2024
1 parent de027d6 commit 566ba1c
Showing 1 changed file with 108 additions and 44 deletions.
152 changes: 108 additions & 44 deletions internal/sidecar/service/volumereplication.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,14 +63,24 @@ func (rs *ReplicationServer) EnableVolumeReplication(
klog.Errorf("Failed to get secret %s in namespace %s: %v", req.GetSecretName(), req.GetSecretNamespace(), err)
return nil, status.Error(codes.Internal, err.Error())
}
repReq := &csiReplication.EnableVolumeReplicationRequest{
ReplicationId: req.ReplicationId,
Parameters: req.Parameters,
Secrets: data,
}
if req.VolumeId != "" {
// setting repReq.VolumeId for backward compatibility for volume replication of a given volume
repReq.VolumeId = req.VolumeId // nolint:staticcheck
repReq.ReplicationSource = &csiReplication.ReplicationSource{
Type: &csiReplication.ReplicationSource_Volume{
Volume: &csiReplication.ReplicationSource_VolumeSource{
VolumeId: req.VolumeId,
},
},
}
}
_, err = rs.controllerClient.EnableVolumeReplication(ctx, repReq)

_, err = rs.controllerClient.EnableVolumeReplication(ctx,
&csiReplication.EnableVolumeReplicationRequest{
VolumeId: req.VolumeId,
ReplicationId: req.ReplicationId,
Parameters: req.Parameters,
Secrets: data,
})
if err != nil {
klog.Errorf("Failed to enable volume replication: %v", err)
return nil, err
Expand All @@ -91,13 +101,24 @@ func (rs *ReplicationServer) DisableVolumeReplication(
return nil, status.Error(codes.Internal, err.Error())
}

_, err = rs.controllerClient.DisableVolumeReplication(ctx,
&csiReplication.DisableVolumeReplicationRequest{
VolumeId: req.VolumeId,
ReplicationId: req.ReplicationId,
Parameters: req.Parameters,
Secrets: data,
})
repReq := &csiReplication.DisableVolumeReplicationRequest{
ReplicationId: req.ReplicationId,
Parameters: req.Parameters,
Secrets: data,
}
if req.VolumeId != "" {
// setting repReq.VolumeId for backward compatibility for volume replication of a given volume
repReq.VolumeId = req.VolumeId // nolint:staticcheck
repReq.ReplicationSource = &csiReplication.ReplicationSource{
Type: &csiReplication.ReplicationSource_Volume{
Volume: &csiReplication.ReplicationSource_VolumeSource{
VolumeId: req.VolumeId,
},
},
}
}

_, err = rs.controllerClient.DisableVolumeReplication(ctx, repReq)
if err != nil {
klog.Errorf("Failed to disable volume replication: %v", err)
return nil, err
Expand All @@ -118,14 +139,25 @@ func (rs *ReplicationServer) PromoteVolume(
return nil, status.Error(codes.Internal, err.Error())
}

_, err = rs.controllerClient.PromoteVolume(ctx,
&csiReplication.PromoteVolumeRequest{
VolumeId: req.VolumeId,
ReplicationId: req.ReplicationId,
Force: req.Force,
Parameters: req.Parameters,
Secrets: data,
})
repReq := &csiReplication.PromoteVolumeRequest{
ReplicationId: req.ReplicationId,
Parameters: req.Parameters,
Force: req.Force,
Secrets: data,
}
if req.VolumeId != "" {
// setting repReq.VolumeId for backward compatibility for volume replication of a given volume
repReq.VolumeId = req.VolumeId // nolint:staticcheck
repReq.ReplicationSource = &csiReplication.ReplicationSource{
Type: &csiReplication.ReplicationSource_Volume{
Volume: &csiReplication.ReplicationSource_VolumeSource{
VolumeId: req.VolumeId,
},
},
}
}

_, err = rs.controllerClient.PromoteVolume(ctx, repReq)
if err != nil {
klog.Errorf("Failed to promote volume: %v", err)
return nil, err
Expand All @@ -146,14 +178,24 @@ func (rs *ReplicationServer) DemoteVolume(
return nil, status.Error(codes.Internal, err.Error())
}

_, err = rs.controllerClient.DemoteVolume(ctx,
&csiReplication.DemoteVolumeRequest{
VolumeId: req.VolumeId,
ReplicationId: req.ReplicationId,
Force: req.Force,
Parameters: req.Parameters,
Secrets: data,
})
repReq := &csiReplication.DemoteVolumeRequest{
ReplicationId: req.ReplicationId,
Parameters: req.Parameters,
Force: req.Force,
Secrets: data,
}
if req.VolumeId != "" {
// setting repReq.VolumeId for backward compatibility for volume replication of a given volume
repReq.VolumeId = req.VolumeId // nolint:staticcheck
repReq.ReplicationSource = &csiReplication.ReplicationSource{
Type: &csiReplication.ReplicationSource_Volume{
Volume: &csiReplication.ReplicationSource_VolumeSource{
VolumeId: req.VolumeId,
},
},
}
}
_, err = rs.controllerClient.DemoteVolume(ctx, repReq)
if err != nil {
klog.Errorf("Failed to demote volume: %v", err)
return nil, err
Expand All @@ -174,14 +216,25 @@ func (rs *ReplicationServer) ResyncVolume(
return nil, status.Error(codes.Internal, err.Error())
}

resp, err := rs.controllerClient.ResyncVolume(ctx,
&csiReplication.ResyncVolumeRequest{
VolumeId: req.VolumeId,
ReplicationId: req.ReplicationId,
Force: req.Force,
Parameters: req.Parameters,
Secrets: data,
})
repReq := &csiReplication.ResyncVolumeRequest{
ReplicationId: req.ReplicationId,
Parameters: req.Parameters,
Force: req.Force,
Secrets: data,
}
if req.VolumeId != "" {
// setting repReq.VolumeId for backward compatibility for volume replication of a given volume
repReq.VolumeId = req.VolumeId // nolint:staticcheck
repReq.ReplicationSource = &csiReplication.ReplicationSource{
Type: &csiReplication.ReplicationSource_Volume{
Volume: &csiReplication.ReplicationSource_VolumeSource{
VolumeId: req.VolumeId,
},
},
}
}

resp, err := rs.controllerClient.ResyncVolume(ctx, repReq)
if err != nil {
klog.Errorf("Failed to resync volume: %v", err)
return nil, err
Expand All @@ -204,12 +257,23 @@ func (rs *ReplicationServer) GetVolumeReplicationInfo(
return nil, status.Error(codes.Internal, err.Error())
}

resp, err := rs.controllerClient.GetVolumeReplicationInfo(ctx,
&csiReplication.GetVolumeReplicationInfoRequest{
VolumeId: req.VolumeId,
Secrets: data,
ReplicationId: req.ReplicationId,
})
repReq := &csiReplication.GetVolumeReplicationInfoRequest{
ReplicationId: req.ReplicationId,
Secrets: data,
}
if req.VolumeId != "" {
// setting repReq.VolumeId for backward compatibility for volume replication of a given volume
repReq.VolumeId = req.VolumeId // nolint:staticcheck
repReq.ReplicationSource = &csiReplication.ReplicationSource{
Type: &csiReplication.ReplicationSource_Volume{
Volume: &csiReplication.ReplicationSource_VolumeSource{
VolumeId: req.VolumeId,
},
},
}
}

resp, err := rs.controllerClient.GetVolumeReplicationInfo(ctx, repReq)
if err != nil {
klog.Errorf("Failed to get volume replication info: %v", err)
return nil, err
Expand Down

0 comments on commit 566ba1c

Please sign in to comment.