mirrors/ceph-csi
1
0
Fork 0
mirror of https://github.com/ceph/ceph-csi.git synced 2025-04-11 18:13:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

journal: change omap set func to handle multiple key-value pairs

Browse Source 
For any function that sets more than one key on a single oid setting
them as a batch will be more efficient.

Signed-off-by: John Mulligan <jmulligan@redhat.com>
This commit is contained in:
John Mulligan 2020-05-21 16:36:50 -04:00 committed by mergify[bot]
parent a8409eccc8
commit c14cbee132
2 changed files with 27 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
internal/journal/omap.go
View File

@ -105,10 +105,10 @@ func removeMapKeys(
return err return err
} }
func setOneOMapKey( func setOMapKeys(
ctx context.Context, ctx context.Context,
conn *Connection, conn *Connection,
poolName, namespace, oMapName, oMapKey, keyValue string) error { poolName, namespace, oid string, pairs map[string]string) error {
// fetch and configure the rados ioctx // fetch and configure the rados ioctx
ioctx, err := conn.conn.GetIoctx(poolName) ioctx, err := conn.conn.GetIoctx(poolName)
if err != nil { if err != nil {
@ -120,20 +120,21 @@ func setOneOMapKey(
ioctx.SetNamespace(namespace) ioctx.SetNamespace(namespace)
} }
pairs := map[string][]byte{ bpairs := make(map[string][]byte, len(pairs))
oMapKey: []byte(keyValue), for k, v := range pairs {
bpairs[k] = []byte(v)
} }
err = ioctx.SetOmap(oMapName, pairs) err = ioctx.SetOmap(oid, bpairs)
if err != nil { if err != nil {
klog.Errorf( klog.Errorf(
util.Log(ctx, "failed setting omap key (pool=%q, namespace=%q, name=%q, key=%q, value=%q): %v"), util.Log(ctx, "failed setting omap keys (pool=%q, namespace=%q, name=%q, pairs=%+v): %v"),
poolName, namespace, oMapName, oMapKey, keyValue, err) poolName, namespace, oid, pairs, err)
} else { return err
klog.Infof(
util.Log(ctx, "XXX set omap key (pool=%q, namespace=%q, name=%q, key=%q, value=%q): %v"),
poolName, namespace, oMapName, oMapKey, keyValue, err)
} }
return err klog.V(4).Infof(
util.Log(ctx, "set omap keys (pool=%q, namespace=%q, name=%q): %+v)"),
poolName, namespace, oid, pairs)
return nil
} }
func omapPoolError(poolName string, err error) error { func omapPoolError(poolName string, err error) error {

41
internal/journal/voljournal.go
View File

@ -517,8 +517,8 @@ func (conn *Connection) ReserveName(ctx context.Context,
nameKeyVal = volUUID nameKeyVal = volUUID
} }
err = setOneOMapKey(ctx, conn, journalPool, cj.namespace, cj.csiDirectory, err = setOMapKeys(ctx, conn, journalPool, cj.namespace, cj.csiDirectory,
cj.csiNameKeyPrefix+reqName, nameKeyVal) map[string]string{cj.csiNameKeyPrefix + reqName: nameKeyVal})
if err != nil { if err != nil {
return "", "", err return "", "", err
} }
@ -532,30 +532,21 @@ func (conn *Connection) ReserveName(ctx context.Context,
} }
}() }()
oid := cj.cephUUIDDirectoryPrefix + volUUID
omapValues := map[string]string{}
// NOTE: UUID directory is stored on the same pool as the image, helps determine image attributes // NOTE: UUID directory is stored on the same pool as the image, helps determine image attributes
// and also CSI journal pool, when only the VolumeID is passed in (e.g DeleteVolume/DeleteSnapshot, // and also CSI journal pool, when only the VolumeID is passed in (e.g DeleteVolume/DeleteSnapshot,
// VolID during CreateSnapshot). // VolID during CreateSnapshot).
// Update UUID directory to store CSI request name // Update UUID directory to store CSI request name
err = setOneOMapKey(ctx, conn, imagePool, cj.namespace, cj.cephUUIDDirectoryPrefix+volUUID, omapValues[cj.csiNameKey] = reqName
cj.csiNameKey, reqName)
if err != nil {
return "", "", err
}
// Update UUID directory to store image name // Update UUID directory to store image name
err = setOneOMapKey(ctx, conn, imagePool, cj.namespace, cj.cephUUIDDirectoryPrefix+volUUID, omapValues[cj.csiImageKey] = imageName
cj.csiImageKey, imageName)
if err != nil {
return "", "", err
}
// Update UUID directory to store encryption values // Update UUID directory to store encryption values
if kmsConf != "" { if kmsConf != "" {
err = setOneOMapKey(ctx, conn, imagePool, cj.namespace, cj.cephUUIDDirectoryPrefix+volUUID, omapValues[cj.encryptKMSKey] = kmsConf
cj.encryptKMSKey, kmsConf)
if err != nil {
return "", "", err
}
} }
if journalPool != imagePool && journalPoolID != util.InvalidPoolID { if journalPool != imagePool && journalPoolID != util.InvalidPoolID {
@ -564,22 +555,18 @@ func (conn *Connection) ReserveName(ctx context.Context,
journalPoolIDStr := hex.EncodeToString(buf64) journalPoolIDStr := hex.EncodeToString(buf64)
// Update UUID directory to store CSI journal pool name (prefer ID instead of name to be pool rename proof) // Update UUID directory to store CSI journal pool name (prefer ID instead of name to be pool rename proof)
err = setOneOMapKey(ctx, conn, imagePool, cj.namespace, cj.cephUUIDDirectoryPrefix+volUUID, omapValues[cj.csiJournalPool] = journalPoolIDStr
cj.csiJournalPool, journalPoolIDStr)
if err != nil {
return "", "", err
}
} }
if snapSource { if snapSource {
// Update UUID directory to store source volume UUID in case of snapshots // Update UUID directory to store source volume UUID in case of snapshots
err = setOneOMapKey(ctx, conn, imagePool, cj.namespace, cj.cephUUIDDirectoryPrefix+volUUID, omapValues[cj.cephSnapSourceKey] = parentName
cj.cephSnapSourceKey, parentName)
if err != nil {
return "", "", err
}
} }
err = setOMapKeys(ctx, conn, journalPool, cj.namespace, oid, omapValues)
if err != nil {
return "", "", err
}
return volUUID, imageName, nil return volUUID, imageName, nil
} }