<commit_before>package storage

import (
	"fmt"
	"strconv"
	"strings"
	"time"

	gofig "github.com/akutz/gofig/types"
	"github.com/akutz/goof"

	"github.com/digitalocean/godo"

	"github.com/thecodeteam/rexray/libstorage/api/context"
	"github.com/thecodeteam/rexray/libstorage/api/registry"
	"github.com/thecodeteam/rexray/libstorage/api/types"
	apiUtils "github.com/thecodeteam/rexray/libstorage/api/utils"

	do "github.com/thecodeteam/rexray/libstorage/drivers/storage/dobs"
)

const (
	minSizeGiB = 1
)

type driver struct {
	name           string
	config         gofig.Config
	client         *godo.Client
	maxAttempts    int
	statusDelay    int64
	statusTimeout  time.Duration
	defaultRegion  *string
	convUnderscore bool
}

func init() {
	registry.RegisterStorageDriver(do.Name, newDriver)
}

func newDriver() types.StorageDriver {
	return &driver{name: do.Name}
}

func (d *driver) Name() string {
	return do.Name
}

func (d *driver) Init(
	ctx types.Context,
	config gofig.Config) error {

	d.config = config
	token := d.config.GetString(do.ConfigToken)
	d.maxAttempts = d.config.GetInt(do.ConfigStatusMaxAttempts)

	statusDelayStr := d.config.GetString(do.ConfigStatusInitDelay)
	statusDelay, err := time.ParseDuration(statusDelayStr)
	if err != nil {
		return err
	}
	d.statusDelay = statusDelay.Nanoseconds()

	statusTimeoutStr := d.config.GetString(do.ConfigStatusTimeout)
	d.statusTimeout, err = time.ParseDuration(statusTimeoutStr)
	if err != nil {
		return err
	}

	fields := map[string]interface{}{
		"maxStatusAttempts": d.maxAttempts,
		"statusDelay": fmt.Sprintf(
			"%v", time.Duration(d.statusDelay)*time.Nanosecond),
		"statusTimeout": d.statusTimeout,
	}

	if region := d.config.GetString(do.ConfigRegion); region != "" {
		d.defaultRegion = &region
		fields["region"] = region
	}

	if token == "" {
		fields["token"] = ""
	} else {
		fields["token"] = "******"
	}

	client, err := Client(token)
	if err != nil {
		return err
	}
	d.client = client

	d.convUnderscore = d.config.GetBool(do.ConfigConvertUnderscores)

	ctx.WithFields(fields).Info("storage driver initialized")

	return nil
}

func (d *driver) Type(ctx types.Context) (types.StorageType, error) {
	return types.Block, nil
}

// DigitalOcean volumes are are found using device-by-id, ex:
// /dev/disk/by-id/scsi-0DO_Volume_volume-nyc1-01 See
// https://www.digitalocean.com/community/tutorials/how-to-use-block-storage-on-digitalocean#preparing-volumes-for-use-in-linux
func (d *driver) NextDeviceInfo(
	ctx types.Context) (*types.NextDeviceInfo, error) {

	return nil, nil
}

func (d *driver) InstanceInspect(
	ctx types.Context,
	opts types.Store) (*types.Instance, error) {

	iid := context.MustInstanceID(ctx)
	return &types.Instance{
		InstanceID:   iid,
		Region:       iid.Fields[do.InstanceIDFieldRegion],
		Name:         iid.Fields[do.InstanceIDFieldName],
		ProviderName: iid.Driver,
	}, nil
}

func (d *driver) Volumes(
	ctx types.Context,
	opts *types.VolumesOpts) ([]*types.Volume, error) {

	doVolumes, _, err := d.client.Storage.ListVolumes(ctx, nil)
	if err != nil {
		return nil, err
	}

	var volumes []*types.Volume
	for _, vol := range doVolumes {
		volume, err := d.toTypesVolume(ctx, &vol, opts.Attachments)
		if err != nil {
			return nil, goof.New("error converting to types.Volume")
		}
		volumes = append(volumes, volume)
	}

	return volumes, nil
}

func (d *driver) VolumeInspect(
	ctx types.Context,
	volumeID string,
	opts *types.VolumeInspectOpts) (*types.Volume, error) {

	doVolume, _, err := d.client.Storage.GetVolume(ctx, volumeID)
	if err != nil {
		return nil, err
	}

	volume, err := d.toTypesVolume(ctx, doVolume, opts.Attachments)
	if err != nil {
		return nil, goof.New("error converting to types.Volume")
	}
	return volume, nil
}

func (d *driver) VolumeInspectByName(
	ctx types.Context,
	volumeName string,
	opts *types.VolumeInspectOpts) (*types.Volume, error) {

	region := d.mustRegion(ctx)
	if region == nil || *region == "" {
		return nil, goof.New("No region provided or configured")
	}

	volumeName = d.convUnderscores(volumeName)

	doVolumes, _, err := d.client.Storage.ListVolumes(
		ctx,<commit_msg>dobs: add paging logic for regions with many volumes<commit_after>package storage

import (
	"fmt"
	"strconv"
	"strings"
	"time"

	gofig "github.com/akutz/gofig/types"
	"github.com/akutz/goof"

	"github.com/digitalocean/godo"

	"github.com/thecodeteam/rexray/libstorage/api/context"
	"github.com/thecodeteam/rexray/libstorage/api/registry"
	"github.com/thecodeteam/rexray/libstorage/api/types"
	apiUtils "github.com/thecodeteam/rexray/libstorage/api/utils"

	do "github.com/thecodeteam/rexray/libstorage/drivers/storage/dobs"
)

const (
	minSizeGiB = 1
)

type driver struct {
	name           string
	config         gofig.Config
	client         *godo.Client
	maxAttempts    int
	statusDelay    int64
	statusTimeout  time.Duration
	defaultRegion  *string
	convUnderscore bool
}

func init() {
	registry.RegisterStorageDriver(do.Name, newDriver)
}

func newDriver() types.StorageDriver {
	return &driver{name: do.Name}
}

func (d *driver) Name() string {
	return do.Name
}

func (d *driver) Init(
	ctx types.Context,
	config gofig.Config) error {

	d.config = config
	token := d.config.GetString(do.ConfigToken)
	d.maxAttempts = d.config.GetInt(do.ConfigStatusMaxAttempts)

	statusDelayStr := d.config.GetString(do.ConfigStatusInitDelay)
	statusDelay, err := time.ParseDuration(statusDelayStr)
	if err != nil {
		return err
	}
	d.statusDelay = statusDelay.Nanoseconds()

	statusTimeoutStr := d.config.GetString(do.ConfigStatusTimeout)
	d.statusTimeout, err = time.ParseDuration(statusTimeoutStr)
	if err != nil {
		return err
	}

	fields := map[string]interface{}{
		"maxStatusAttempts": d.maxAttempts,
		"statusDelay": fmt.Sprintf(
			"%v", time.Duration(d.statusDelay)*time.Nanosecond),
		"statusTimeout": d.statusTimeout,
	}

	if region := d.config.GetString(do.ConfigRegion); region != "" {
		d.defaultRegion = &region
		fields["region"] = region
	}

	if token == "" {
		fields["token"] = ""
	} else {
		fields["token"] = "******"
	}

	client, err := Client(token)
	if err != nil {
		return err
	}
	d.client = client

	d.convUnderscore = d.config.GetBool(do.ConfigConvertUnderscores)

	ctx.WithFields(fields).Info("storage driver initialized")

	return nil
}

func (d *driver) Type(ctx types.Context) (types.StorageType, error) {
	return types.Block, nil
}

// DigitalOcean volumes are are found using device-by-id, ex:
// /dev/disk/by-id/scsi-0DO_Volume_volume-nyc1-01 See
// https://www.digitalocean.com/community/tutorials/how-to-use-block-storage-on-digitalocean#preparing-volumes-for-use-in-linux
func (d *driver) NextDeviceInfo(
	ctx types.Context) (*types.NextDeviceInfo, error) {

	return nil, nil
}

func (d *driver) InstanceInspect(
	ctx types.Context,
	opts types.Store) (*types.Instance, error) {

	iid := context.MustInstanceID(ctx)
	return &types.Instance{
		InstanceID:   iid,
		Region:       iid.Fields[do.InstanceIDFieldRegion],
		Name:         iid.Fields[do.InstanceIDFieldName],
		ProviderName: iid.Driver,
	}, nil
}

func (d *driver) Volumes(
	ctx types.Context,
	opts *types.VolumesOpts) ([]*types.Volume, error) {

	region := d.mustRegion(ctx)
	if region == nil || *region == "" {
		return nil, goof.New("No region provided or configured")
	}

	listOpts := &godo.ListVolumeParams{
		ListOptions: &godo.ListOptions{PerPage: 200},
		Region:      *region,
	}

	var volumes []*types.Volume
	for {
		doVolumes, resp, err := d.client.Storage.ListVolumes(ctx, listOpts)
		if err != nil {
			return nil, err
		}

		for _, vol := range doVolumes {
			volume, err := d.toTypesVolume(ctx, &vol, opts.Attachments)
			if err != nil {
				return nil, goof.New("error converting to types.Volume")
			}
			volumes = append(volumes, volume)
		}

		if resp.Links == nil || resp.Links.IsLastPage() {
			break
		}

		page, err := resp.Links.CurrentPage()
		if err != nil {
			return nil, err
		}

		listOpts.ListOptions.Page = page + 1
	}

	return volumes, nil
}

func (d *driver) VolumeInspect(
	ctx types.Context,
	volumeID string,
	opts *types.VolumeInspectOpts) (*types.Volume, error) {

	doVolume, _, err := d.client.Storage.GetVolume(ctx, volumeID)
	if err != nil {
		return nil, err
	}

	volume, err := d.toTypesVolume(ctx, doVolume, opts.Attachments)
	if err != nil {
		return nil, goof.New("error converting to types.Volume")
	}
	return volume, nil
}

func (d *driver) VolumeInspectByName(
	ctx types.Context,
	volumeName string,
	opts *types.VolumeInspectOpts) (*types.Volume, error) {
	region := d.mustRegion(ctx)
	if region == nil || *region == "" {
		return nil, goof.New("No region provided or configured")
	}

	volumeName = d.convUnderscores(volumeName)

	doVolumes, _, err := d.client.Storage.ListVolumes(
		ctx,
{'fraction_non_alphanumeric': 0.13619254020759666, 'fraction_numerical': 0.002053153872476332, 'mean_word_length': 4.068208092485549, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2887883', 'n_tokens_mistral': 3280, 'n_tokens_neox': 3108, 'n_words': 658}