vitalif
/
ceph-gobench
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

SnakeCase refactor

master
Alexey Kostin 2019-03-04 12:42:43 +03:00
parent ad14e37433
commit 2493a2391b
5 changed files with 38 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
ceph-gobench.go
View File

@ -16,7 +16,7 @@ import (
"time" "time"
) )
func bench(cephconn *Cephconnection, osddevice Device, buff *[]byte, startbuff *[]byte, params *Params, func bench(cephconn *cephconnection, osddevice Device, buff *[]byte, startbuff *[]byte, params *params,
wg *sync.WaitGroup, result chan string) { wg *sync.WaitGroup, result chan string) {
defer wg.Done() defer wg.Done()
threadresult := make(chan []time.Duration, params.threadsCount) threadresult := make(chan []time.Duration, params.threadsCount)
@ -30,7 +30,7 @@ func bench(cephconn *Cephconnection, osddevice Device, buff *[]byte, startbuff *
// calculate object for each thread // calculate object for each thread
for suffix := 0; len(objectnames) < int(params.threadsCount)*16; suffix++ { for suffix := 0; len(objectnames) < int(params.threadsCount)*16; suffix++ {
name := "bench_" + strconv.Itoa(suffix) name := "bench_" + strconv.Itoa(suffix)
if osddevice.ID == GetObjActingPrimary(cephconn, *params, name) { if osddevice.ID == getObjActingPrimary(cephconn, *params, name) {
objectnames = append(objectnames, name) objectnames = append(objectnames, name)
if err := cephconn.ioctx.WriteFull(name, *startbuff); err != nil { if err := cephconn.ioctx.WriteFull(name, *startbuff); err != nil {
log.Printf("Can't write object: %v, osd: %v", name, osddevice.Name) log.Printf("Can't write object: %v, osd: %v", name, osddevice.Name)
@ -41,7 +41,7 @@ func bench(cephconn *Cephconnection, osddevice Device, buff *[]byte, startbuff *
} }
} }
for i := 0; i < int(params.threadsCount); i++ { for i := 0; i < int(params.threadsCount); i++ {
go BenchThread(cephconn, osddevice, params, buff, threadresult, objectnames[i*16:i*16+16]) go benchthread(cephconn, osddevice, params, buff, threadresult, objectnames[i*16:i*16+16])
} }
for i := uint64(0); i < params.threadsCount; i++ { for i := uint64(0); i < params.threadsCount; i++ {
for _, lat := range <-threadresult { for _, lat := range <-threadresult {
@ -174,7 +174,7 @@ func bench(cephconn *Cephconnection, osddevice Device, buff *[]byte, startbuff *
result <- buffer.String() result <- buffer.String()
} }
func BenchThread(cephconn *Cephconnection, osddevice Device, params *Params, buff *[]byte, func benchthread(cephconn *cephconnection, osddevice Device, params *params, buff *[]byte,
result chan []time.Duration, objnames []string) { result chan []time.Duration, objnames []string) {
var latencies []time.Duration var latencies []time.Duration
starttime := time.Now() starttime := time.Now()
@ -198,7 +198,7 @@ func BenchThread(cephconn *Cephconnection, osddevice Device, params *Params, buf
} }
func main() { func main() {
params := Route() params := route()
if params.cpuprofile != "" { if params.cpuprofile != "" {
f, err := os.Create(params.cpuprofile) f, err := os.Create(params.cpuprofile)
if err != nil { if err != nil {
@ -229,7 +229,7 @@ func main() {
time.Sleep(time.Millisecond * 100) time.Sleep(time.Millisecond * 100)
startbuff := make([]byte, 4096) startbuff := make([]byte, 4096)
osddevices := GetOsds(cephconn, params) osddevices := getOsds(cephconn, params)
buff := make([]byte, params.blocksize) buff := make([]byte, params.blocksize)
rand.Read(buff) rand.Read(buff)

4
cephconnection.go
View File

@ -6,8 +6,8 @@ import (
"os" "os"
) )
func connectioninit(params Params) *Cephconnection { func connectioninit(params params) *cephconnection {
cephconn := &Cephconnection{} cephconn := &cephconnection{}
var err error var err error
if _, err := os.Stat(params.config); os.IsNotExist(err) { if _, err := os.Stat(params.config); os.IsNotExist(err) {
log.Fatalf("Congif file not exists. Error: %v\n", err) log.Fatalf("Congif file not exists. Error: %v\n", err)

4
flags.go
View File

@ -7,8 +7,8 @@ import (
"time" "time"
) )
func Route() Params { func route() params {
params := Params{} params := params{}
gnuflag.DurationVar(&params.duration, "duration", 30*time.Second, gnuflag.DurationVar(&params.duration, "duration", 30*time.Second,
"Time limit for each test in seconds") "Time limit for each test in seconds")
gnuflag.DurationVar(&params.duration, "d", 30*time.Second, gnuflag.DurationVar(&params.duration, "d", 30*time.Second,

52
getosd.go
View File

@ -6,7 +6,7 @@ import (
"strings" "strings"
) )
func MakeMonQuery(cephconn *Cephconnection, query map[string]string) []byte { func makeMonQuery(cephconn *cephconnection, query map[string]string) []byte {
monjson, err := json.Marshal(query) monjson, err := json.Marshal(query)
if err != nil { if err != nil {
log.Fatalf("Can't marshal json mon query. Error: %v", err) log.Fatalf("Can't marshal json mon query. Error: %v", err)
@ -19,8 +19,8 @@ func MakeMonQuery(cephconn *Cephconnection, query map[string]string) []byte {
return monrawanswer return monrawanswer
} }
func GetPoolSize(cephconn *Cephconnection, params Params) Poolinfo { func getPoolSize(cephconn *cephconnection, params params) Poolinfo {
monrawanswer := MakeMonQuery(cephconn, map[string]string{"prefix": "osd pool get", "pool": params.pool, monrawanswer := makeMonQuery(cephconn, map[string]string{"prefix": "osd pool get", "pool": params.pool,
"format": "json", "var": "size"}) "format": "json", "var": "size"})
monanswer := Poolinfo{} monanswer := Poolinfo{}
if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil { if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil {
@ -30,8 +30,8 @@ func GetPoolSize(cephconn *Cephconnection, params Params) Poolinfo {
} }
func GetPgByPool(cephconn *Cephconnection, params Params) []PlacementGroup { func getPgByPool(cephconn *cephconnection, params params) []PlacementGroup {
monrawanswer := MakeMonQuery(cephconn, map[string]string{"prefix": "pg ls-by-pool", "poolstr": params.pool, monrawanswer := makeMonQuery(cephconn, map[string]string{"prefix": "pg ls-by-pool", "poolstr": params.pool,
"format": "json"}) "format": "json"})
var monanswer []PlacementGroup var monanswer []PlacementGroup
if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil { if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil {
@ -40,8 +40,8 @@ func GetPgByPool(cephconn *Cephconnection, params Params) []PlacementGroup {
return monanswer return monanswer
} }
func GetOsdCrushDump(cephconn *Cephconnection) OsdCrushDump { func getOsdCrushDump(cephconn *cephconnection) OsdCrushDump {
monrawanswer := MakeMonQuery(cephconn, map[string]string{"prefix": "osd crush dump", "format": "json"}) monrawanswer := makeMonQuery(cephconn, map[string]string{"prefix": "osd crush dump", "format": "json"})
var monanswer OsdCrushDump var monanswer OsdCrushDump
if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil { if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil {
log.Fatalf("Can't parse monitor answer. Error: %v", err) log.Fatalf("Can't parse monitor answer. Error: %v", err)
@ -49,8 +49,8 @@ func GetOsdCrushDump(cephconn *Cephconnection) OsdCrushDump {
return monanswer return monanswer
} }
func GetOsdDump(cephconn *Cephconnection) OsdDump { func getOsdDump(cephconn *cephconnection) OsdDump {
monrawanswer := MakeMonQuery(cephconn, map[string]string{"prefix": "osd dump", "format": "json"}) monrawanswer := makeMonQuery(cephconn, map[string]string{"prefix": "osd dump", "format": "json"})
var monanswer OsdDump var monanswer OsdDump
if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil { if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil {
log.Fatalf("Can't parse monitor answer. Error: %v", err) log.Fatalf("Can't parse monitor answer. Error: %v", err)
@ -58,8 +58,8 @@ func GetOsdDump(cephconn *Cephconnection) OsdDump {
return monanswer return monanswer
} }
func GetOsdMetadata(cephconn *Cephconnection) []OsdMetadata { func getOsdMetadata(cephconn *cephconnection) []OsdMetadata {
monrawanswer := MakeMonQuery(cephconn, map[string]string{"prefix": "osd metadata", "format": "json"}) monrawanswer := makeMonQuery(cephconn, map[string]string{"prefix": "osd metadata", "format": "json"})
var monanswer []OsdMetadata var monanswer []OsdMetadata
if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil { if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil {
log.Fatalf("Can't parse monitor answer. Error: %v", err) log.Fatalf("Can't parse monitor answer. Error: %v", err)
@ -67,8 +67,8 @@ func GetOsdMetadata(cephconn *Cephconnection) []OsdMetadata {
return monanswer return monanswer
} }
func GetObjActingPrimary(cephconn *Cephconnection, params Params, objname string) int64 { func getObjActingPrimary(cephconn *cephconnection, params params, objname string) int64 {
monrawanswer := MakeMonQuery(cephconn, map[string]string{"prefix": "osd map", "pool": params.pool, monrawanswer := makeMonQuery(cephconn, map[string]string{"prefix": "osd map", "pool": params.pool,
"object": objname, "format": "json"}) "object": objname, "format": "json"})
var monanswer OsdMap var monanswer OsdMap
if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil { if err := json.Unmarshal([]byte(monrawanswer), &monanswer); err != nil {
@ -77,7 +77,7 @@ func GetObjActingPrimary(cephconn *Cephconnection, params Params, objname string
return monanswer.UpPrimary return monanswer.UpPrimary
} }
func GetCrushHostBuckets(buckets []Bucket, itemid int64) []Bucket { func getCrushHostBuckets(buckets []Bucket, itemid int64) []Bucket {
var rootbuckets []Bucket var rootbuckets []Bucket
for _, bucket := range buckets { for _, bucket := range buckets {
if bucket.ID == itemid { if bucket.ID == itemid {
@ -85,7 +85,7 @@ func GetCrushHostBuckets(buckets []Bucket, itemid int64) []Bucket {
rootbuckets = append(rootbuckets, bucket) rootbuckets = append(rootbuckets, bucket)
} else { } else {
for _, item := range bucket.Items { for _, item := range bucket.Items {
result := GetCrushHostBuckets(buckets, item.ID) result := getCrushHostBuckets(buckets, item.ID)
for _, it := range result { for _, it := range result {
rootbuckets = append(rootbuckets, it) rootbuckets = append(rootbuckets, it)
} }
@ -96,7 +96,7 @@ func GetCrushHostBuckets(buckets []Bucket, itemid int64) []Bucket {
return rootbuckets return rootbuckets
} }
func GetOsdForLocations(params Params, osdcrushdump OsdCrushDump, osddump OsdDump, poolinfo Poolinfo, osdsmetadata []OsdMetadata) []Device { func getOsdForLocations(params params, osdcrushdump OsdCrushDump, osddump OsdDump, poolinfo Poolinfo, osdsmetadata []OsdMetadata) []Device {
var crushrule, rootid int64 var crushrule, rootid int64
var crushrulename string var crushrulename string
for _, pool := range osddump.Pools { for _, pool := range osddump.Pools {
@ -120,7 +120,7 @@ func GetOsdForLocations(params Params, osdcrushdump OsdCrushDump, osddump OsdDum
} }
var osddevices []Device var osddevices []Device
bucketitems := GetCrushHostBuckets(osdcrushdump.Buckets, rootid) bucketitems := getCrushHostBuckets(osdcrushdump.Buckets, rootid)
if params.define != "" { if params.define != "" {
if strings.HasPrefix(params.define, "osd.") { if strings.HasPrefix(params.define, "osd.") {
for _, hostbucket := range bucketitems { for _, hostbucket := range bucketitems {
@ -185,7 +185,7 @@ func GetOsdForLocations(params Params, osdcrushdump OsdCrushDump, osddump OsdDum
return osddevices return osddevices
} }
func ContainsPg(pgs []PlacementGroup, i int64) bool { func containsPg(pgs []PlacementGroup, i int64) bool {
for _, pg := range pgs { for _, pg := range pgs {
if i == pg.ActingPrimary { if i == pg.ActingPrimary {
return true return true
@ -194,19 +194,19 @@ func ContainsPg(pgs []PlacementGroup, i int64) bool {
return false return false
} }
func GetOsds(cephconn *Cephconnection, params Params) []Device { func getOsds(cephconn *cephconnection, params params) []Device {
poolinfo := GetPoolSize(cephconn, params) poolinfo := getPoolSize(cephconn, params)
if poolinfo.Size != 1 { if poolinfo.Size != 1 {
log.Fatalf("Pool size must be 1. Current size for pool %v is %v. Don't forget that it must be useless pool (not production). Do:\n # ceph osd pool set %v min_size 1\n # ceph osd pool set %v size 1", log.Fatalf("Pool size must be 1. Current size for pool %v is %v. Don't forget that it must be useless pool (not production). Do:\n # ceph osd pool set %v min_size 1\n # ceph osd pool set %v size 1",
poolinfo.Pool, poolinfo.Size, poolinfo.Pool, poolinfo.Pool) poolinfo.Pool, poolinfo.Size, poolinfo.Pool, poolinfo.Pool)
} }
placementGroups := GetPgByPool(cephconn, params) placementGroups := getPgByPool(cephconn, params)
crushosddump := GetOsdCrushDump(cephconn) crushosddump := getOsdCrushDump(cephconn)
osddump := GetOsdDump(cephconn) osddump := getOsdDump(cephconn)
osdsmetadata := GetOsdMetadata(cephconn) osdsmetadata := getOsdMetadata(cephconn)
osddevices := GetOsdForLocations(params, crushosddump, osddump, poolinfo, osdsmetadata) osddevices := getOsdForLocations(params, crushosddump, osddump, poolinfo, osdsmetadata)
for _, device := range osddevices { for _, device := range osddevices {
if exist := ContainsPg(placementGroups, device.ID); exist == false { if exist := containsPg(placementGroups, device.ID); exist == false {
log.Fatalln("Not enough pg for test. Some osd haven't placement group at all. Increase pg_num and pgp_num") log.Fatalln("Not enough pg for test. Some osd haven't placement group at all. Increase pg_num and pgp_num")
} }
} }

4
types.go
View File

@ -5,7 +5,7 @@ import (
"time" "time"
) )
type Params struct { type params struct {
duration time.Duration duration time.Duration
threadsCount uint64 threadsCount uint64
blocksize, objectsize uint64 blocksize, objectsize uint64
@ -13,7 +13,7 @@ type Params struct {
bs, os, cluster, user, keyring, config, pool, define, cpuprofile, memprofile string bs, os, cluster, user, keyring, config, pool, define, cpuprofile, memprofile string
} }
type Cephconnection struct { type cephconnection struct {
conn *rados.Conn conn *rados.Conn
ioctx *rados.IOContext ioctx *rados.IOContext
} }