tools/keep-exercise/keep-exercise.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 // Testing tool for Keep services.
   6 //
   7 // keepexercise helps measure throughput and test reliability under
   8 // various usage patterns.
   9 //
  10 // By default, it reads and writes blocks containing 2^26 NUL
  11 // bytes. This generates network traffic without consuming much disk
  12 // space.
  13 //
  14 // For a more realistic test, enable -vary-request. Warning: this will
  15 // fill your storage volumes with random data if you leave it running,
  16 // which can cost you money or leave you with too little room for
  17 // useful data.
  18 //
  19 package main
  20
  21 import (
  22         "crypto/rand"
  23         "encoding/binary"
  24         "flag"
  25         "fmt"
  26         "io"
  27         "io/ioutil"
  28         "log"
  29         "net/http"
  30         "os"
  31         "time"
  32
  33         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  34         "git.arvados.org/arvados.git/sdk/go/keepclient"
  35 )
  36
  37 var version = "dev"
  38
  39 // Command line config knobs
  40 var (
  41         BlockSize     = flag.Int("block-size", keepclient.BLOCKSIZE, "bytes per read/write op")
  42         ReadThreads   = flag.Int("rthreads", 1, "number of concurrent readers")
  43         WriteThreads  = flag.Int("wthreads", 1, "number of concurrent writers")
  44         VaryRequest   = flag.Bool("vary-request", false, "vary the data for each request: consumes disk space, exercises write behavior")
  45         VaryThread    = flag.Bool("vary-thread", false, "use -wthreads different data blocks")
  46         Replicas      = flag.Int("replicas", 1, "replication level for writing")
  47         StatsInterval = flag.Duration("stats-interval", time.Second, "time interval between IO stats reports, or 0 to disable")
  48         ServiceURL    = flag.String("url", "", "specify scheme://host of a single keep service to exercise (instead of using all advertised services like normal clients)")
  49         ServiceUUID   = flag.String("uuid", "", "specify UUID of a single advertised keep service to exercise")
  50         getVersion    = flag.Bool("version", false, "Print version information and exit.")
  51 )
  52
  53 func main() {
  54         flag.Parse()
  55
  56         // Print version information if requested
  57         if *getVersion {
  58                 fmt.Printf("keep-exercise %s\n", version)
  59                 os.Exit(0)
  60         }
  61
  62         log.Printf("keep-exercise %s started", version)
  63
  64         arv, err := arvadosclient.MakeArvadosClient()
  65         if err != nil {
  66                 log.Fatal(err)
  67         }
  68         kc, err := keepclient.MakeKeepClient(arv)
  69         if err != nil {
  70                 log.Fatal(err)
  71         }
  72         kc.Want_replicas = *Replicas
  73
  74         transport := *(http.DefaultTransport.(*http.Transport))
  75         transport.TLSClientConfig = arvadosclient.MakeTLSConfig(arv.ApiInsecure)
  76         kc.HTTPClient = &http.Client{
  77                 Timeout:   10 * time.Minute,
  78                 Transport: &transport,
  79         }
  80
  81         overrideServices(kc)
  82
  83         nextLocator := make(chan string, *ReadThreads+*WriteThreads)
  84
  85         go countBeans(nextLocator)
  86         for i := 0; i < *WriteThreads; i++ {
  87                 nextBuf := make(chan []byte, 1)
  88                 go makeBufs(nextBuf, i)
  89                 go doWrites(kc, nextBuf, nextLocator)
  90         }
  91         for i := 0; i < *ReadThreads; i++ {
  92                 go doReads(kc, nextLocator)
  93         }
  94         <-make(chan struct{})
  95 }
  96
  97 // Send 1234 to bytesInChan when we receive 1234 bytes from keepstore.
  98 var bytesInChan = make(chan uint64)
  99 var bytesOutChan = make(chan uint64)
 100
 101 // Send struct{}{} to errorsChan when an error happens.
 102 var errorsChan = make(chan struct{})
 103
 104 func countBeans(nextLocator chan string) {
 105         t0 := time.Now()
 106         var tickChan <-chan time.Time
 107         if *StatsInterval > 0 {
 108                 tickChan = time.NewTicker(*StatsInterval).C
 109         }
 110         var bytesIn uint64
 111         var bytesOut uint64
 112         var errors uint64
 113         for {
 114                 select {
 115                 case <-tickChan:
 116                         elapsed := time.Since(t0)
 117                         log.Printf("%v elapsed: read %v bytes (%.1f MiB/s), wrote %v bytes (%.1f MiB/s), errors %d",
 118                                 elapsed,
 119                                 bytesIn, (float64(bytesIn) / elapsed.Seconds() / 1048576),
 120                                 bytesOut, (float64(bytesOut) / elapsed.Seconds() / 1048576),
 121                                 errors,
 122                         )
 123                 case i := <-bytesInChan:
 124                         bytesIn += i
 125                 case o := <-bytesOutChan:
 126                         bytesOut += o
 127                 case <-errorsChan:
 128                         errors++
 129                 }
 130         }
 131 }
 132
 133 func makeBufs(nextBuf chan<- []byte, threadID int) {
 134         buf := make([]byte, *BlockSize)
 135         if *VaryThread {
 136                 binary.PutVarint(buf, int64(threadID))
 137         }
 138         randSize := 524288
 139         if randSize > *BlockSize {
 140                 randSize = *BlockSize
 141         }
 142         for {
 143                 if *VaryRequest {
 144                         rnd := make([]byte, randSize)
 145                         if _, err := io.ReadFull(rand.Reader, rnd); err != nil {
 146                                 log.Fatal(err)
 147                         }
 148                         buf = append(rnd, buf[randSize:]...)
 149                 }
 150                 nextBuf <- buf
 151         }
 152 }
 153
 154 func doWrites(kc *keepclient.KeepClient, nextBuf <-chan []byte, nextLocator chan<- string) {
 155         for buf := range nextBuf {
 156                 locator, _, err := kc.PutB(buf)
 157                 if err != nil {
 158                         log.Print(err)
 159                         errorsChan <- struct{}{}
 160                         continue
 161                 }
 162                 bytesOutChan <- uint64(len(buf))
 163                 for cap(nextLocator) > len(nextLocator)+*WriteThreads {
 164                         // Give the readers something to do, unless
 165                         // they have lots queued up already.
 166                         nextLocator <- locator
 167                 }
 168         }
 169 }
 170
 171 func doReads(kc *keepclient.KeepClient, nextLocator <-chan string) {
 172         for locator := range nextLocator {
 173                 rdr, size, url, err := kc.Get(locator)
 174                 if err != nil {
 175                         log.Print(err)
 176                         errorsChan <- struct{}{}
 177                         continue
 178                 }
 179                 n, err := io.Copy(ioutil.Discard, rdr)
 180                 rdr.Close()
 181                 if n != size || err != nil {
 182                         log.Printf("Got %d bytes (expected %d) from %s: %v", n, size, url, err)
 183                         errorsChan <- struct{}{}
 184                         continue
 185                         // Note we don't count the bytes received in
 186                         // partial/corrupt responses: we are measuring
 187                         // throughput, not resource consumption.
 188                 }
 189                 bytesInChan <- uint64(n)
 190         }
 191 }
 192
 193 func overrideServices(kc *keepclient.KeepClient) {
 194         roots := make(map[string]string)
 195         if *ServiceURL != "" {
 196                 roots["zzzzz-bi6l4-000000000000000"] = *ServiceURL
 197         } else if *ServiceUUID != "" {
 198                 for uuid, url := range kc.GatewayRoots() {
 199                         if uuid == *ServiceUUID {
 200                                 roots[uuid] = url
 201                                 break
 202                         }
 203                 }
 204                 if len(roots) == 0 {
 205                         log.Fatalf("Service %q was not in list advertised by API %+q", *ServiceUUID, kc.GatewayRoots())
 206                 }
 207         } else {
 208                 return
 209         }
 210         kc.SetServiceRoots(roots, roots, roots)
 211 }