sdk/go/keepclient/discover.go

   1 package keepclient
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "log"
   7         "os"
   8         "os/signal"
   9         "reflect"
  10         "strings"
  11         "syscall"
  12         "time"
  13 )
  14
  15 // DiscoverKeepServers gets list of available keep services from the
  16 // API server.
  17 //
  18 // If a list of services is provided in the arvadosclient (e.g., from
  19 // an environment variable or local config), that list is used
  20 // instead.
  21 func (this *KeepClient) DiscoverKeepServers() error {
  22         if this.Arvados.KeepServiceURIs != nil {
  23                 this.foundNonDiskSvc = true
  24                 this.replicasPerService = 0
  25                 this.setClientSettingsNonDisk()
  26                 roots := make(map[string]string)
  27                 for i, uri := range this.Arvados.KeepServiceURIs {
  28                         roots[fmt.Sprintf("00000-bi6l4-%015d", i)] = uri
  29                 }
  30                 this.SetServiceRoots(roots, roots, roots)
  31                 return nil
  32         }
  33
  34         // ArvadosClient did not provide a services list. Ask API
  35         // server for a list of accessible services.
  36         var list svcList
  37         err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &list)
  38         if err != nil {
  39                 return err
  40         }
  41         return this.loadKeepServers(list)
  42 }
  43
  44 // LoadKeepServicesFromJSON gets list of available keep services from given JSON
  45 func (this *KeepClient) LoadKeepServicesFromJSON(services string) error {
  46         var list svcList
  47
  48         // Load keep services from given json
  49         dec := json.NewDecoder(strings.NewReader(services))
  50         if err := dec.Decode(&list); err != nil {
  51                 return err
  52         }
  53
  54         return this.loadKeepServers(list)
  55 }
  56
  57 // RefreshServices calls DiscoverKeepServers to refresh the keep
  58 // service list on SIGHUP; when the given interval has elapsed since
  59 // the last refresh; and (if the last refresh failed) the given
  60 // errInterval has elapsed.
  61 func (kc *KeepClient) RefreshServices(interval, errInterval time.Duration) {
  62         var previousRoots = []map[string]string{}
  63
  64         timer := time.NewTimer(interval)
  65         gotHUP := make(chan os.Signal, 1)
  66         signal.Notify(gotHUP, syscall.SIGHUP)
  67
  68         for {
  69                 select {
  70                 case <-gotHUP:
  71                 case <-timer.C:
  72                 }
  73                 timer.Reset(interval)
  74
  75                 if err := kc.DiscoverKeepServers(); err != nil {
  76                         log.Printf("WARNING: Error retrieving services list: %v (retrying in %v)", err, errInterval)
  77                         timer.Reset(errInterval)
  78                         continue
  79                 }
  80                 newRoots := []map[string]string{kc.LocalRoots(), kc.GatewayRoots()}
  81
  82                 if !reflect.DeepEqual(previousRoots, newRoots) {
  83                         DebugPrintf("DEBUG: Updated services list: locals %v gateways %v", newRoots[0], newRoots[1])
  84                         previousRoots = newRoots
  85                 }
  86
  87                 if len(newRoots[0]) == 0 {
  88                         log.Printf("WARNING: No local services (retrying in %v)", errInterval)
  89                         timer.Reset(errInterval)
  90                 }
  91         }
  92 }
  93
  94 // loadKeepServers
  95 func (this *KeepClient) loadKeepServers(list svcList) error {
  96         listed := make(map[string]bool)
  97         localRoots := make(map[string]string)
  98         gatewayRoots := make(map[string]string)
  99         writableLocalRoots := make(map[string]string)
 100
 101         // replicasPerService is 1 for disks; unknown or unlimited otherwise
 102         this.replicasPerService = 1
 103
 104         for _, service := range list.Items {
 105                 scheme := "http"
 106                 if service.SSL {
 107                         scheme = "https"
 108                 }
 109                 url := fmt.Sprintf("%s://%s:%d", scheme, service.Hostname, service.Port)
 110
 111                 // Skip duplicates
 112                 if listed[url] {
 113                         continue
 114                 }
 115                 listed[url] = true
 116
 117                 localRoots[service.Uuid] = url
 118                 if service.ReadOnly == false {
 119                         writableLocalRoots[service.Uuid] = url
 120                         if service.SvcType != "disk" {
 121                                 this.replicasPerService = 0
 122                         }
 123                 }
 124
 125                 if service.SvcType != "disk" {
 126                         this.foundNonDiskSvc = true
 127                 }
 128
 129                 // Gateway services are only used when specified by
 130                 // UUID, so there's nothing to gain by filtering them
 131                 // by service type. Including all accessible services
 132                 // (gateway and otherwise) merely accommodates more
 133                 // service configurations.
 134                 gatewayRoots[service.Uuid] = url
 135         }
 136
 137         if this.foundNonDiskSvc {
 138                 this.setClientSettingsNonDisk()
 139         } else {
 140                 this.setClientSettingsDisk()
 141         }
 142
 143         this.SetServiceRoots(localRoots, writableLocalRoots, gatewayRoots)
 144         return nil
 145 }