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