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