package physical import ( "fmt" "os" "sort" "strconv" "strings" "time" log "github.com/mgutz/logxi/v1" "github.com/armon/go-metrics" "github.com/hashicorp/errwrap" "github.com/hashicorp/go-cleanhttp" "github.com/ncw/swift" ) // SwiftBackend is a physical backend that stores data // within an OpenStack Swift container. type SwiftBackend struct { container string client *swift.Connection logger log.Logger permitPool *PermitPool } // newSwiftBackend constructs a Swift backend using a pre-existing // container. Credentials can be provided to the backend, sourced // from the environment. func newSwiftBackend(conf map[string]string, logger log.Logger) (Backend, error) { username := os.Getenv("OS_USERNAME") if username == "" { username = conf["username"] if username == "" { return nil, fmt.Errorf("missing username") } } password := os.Getenv("OS_PASSWORD") if password == "" { password = conf["password"] if password == "" { return nil, fmt.Errorf("missing password") } } authUrl := os.Getenv("OS_AUTH_URL") if authUrl == "" { authUrl = conf["auth_url"] if authUrl == "" { return nil, fmt.Errorf("missing auth_url") } } container := os.Getenv("OS_CONTAINER") if container == "" { container = conf["container"] if container == "" { return nil, fmt.Errorf("missing container") } } tenant := os.Getenv("OS_TENANT_NAME") if tenant == "" { tenant = conf["tenant"] } c := swift.Connection{ UserName: username, ApiKey: password, AuthUrl: authUrl, Tenant: tenant, Transport: cleanhttp.DefaultPooledTransport(), } err := c.Authenticate() if err != nil { return nil, err } _, _, err = c.Container(container) if err != nil { return nil, fmt.Errorf("Unable to access container '%s': %v", container, err) } maxParStr, ok := conf["max_parallel"] var maxParInt int if ok { maxParInt, err = strconv.Atoi(maxParStr) if err != nil { return nil, errwrap.Wrapf("failed parsing max_parallel parameter: {{err}}", err) } if logger.IsDebug() { logger.Debug("swift: max_parallel set", "max_parallel", maxParInt) } } s := &SwiftBackend{ client: &c, container: container, logger: logger, permitPool: NewPermitPool(maxParInt), } return s, nil } // Put is used to insert or update an entry func (s *SwiftBackend) Put(entry *Entry) error { defer metrics.MeasureSince([]string{"swift", "put"}, time.Now()) s.permitPool.Acquire() defer s.permitPool.Release() err := s.client.ObjectPutBytes(s.container, entry.Key, entry.Value, "") if err != nil { return err } return nil } // Get is used to fetch an entry func (s *SwiftBackend) Get(key string) (*Entry, error) { defer metrics.MeasureSince([]string{"swift", "get"}, time.Now()) s.permitPool.Acquire() defer s.permitPool.Release() //Do a list of names with the key first since eventual consistency means //it might be deleted, but a node might return a read of bytes which fails //the physical test list, err := s.client.ObjectNames(s.container, &swift.ObjectsOpts{Prefix: key}) if err != nil { return nil, err } if 0 == len(list) { return nil, nil } data, err := s.client.ObjectGetBytes(s.container, key) if err == swift.ObjectNotFound { return nil, nil } if err != nil { return nil, err } ent := &Entry{ Key: key, Value: data, } return ent, nil } // Delete is used to permanently delete an entry func (s *SwiftBackend) Delete(key string) error { defer metrics.MeasureSince([]string{"swift", "delete"}, time.Now()) s.permitPool.Acquire() defer s.permitPool.Release() err := s.client.ObjectDelete(s.container, key) if err != nil && err != swift.ObjectNotFound { return err } return nil } // List is used to list all the keys under a given // prefix, up to the next prefix. func (s *SwiftBackend) List(prefix string) ([]string, error) { defer metrics.MeasureSince([]string{"swift", "list"}, time.Now()) s.permitPool.Acquire() defer s.permitPool.Release() list, err := s.client.ObjectNamesAll(s.container, &swift.ObjectsOpts{Prefix: prefix}) if nil != err { return nil, err } keys := []string{} for _, key := range list { key := strings.TrimPrefix(key, prefix) if i := strings.Index(key, "/"); i == -1 { // Add objects only from the current 'folder' keys = append(keys, key) } else if i != -1 { // Add truncated 'folder' paths keys = appendIfMissing(keys, key[:i+1]) } } sort.Strings(keys) return keys, nil }