github.com/Azure/go-autorest/autorest/to v0.4.0
github.com/arvados/cgofuse v1.2.0-arvados1
github.com/aws/aws-sdk-go v1.44.174
- github.com/aws/aws-sdk-go-v2 v0.23.0
+ github.com/aws/aws-sdk-go-v2 v1.26.1
github.com/bmatcuk/doublestar/v4 v4.6.1
github.com/bradleypeabody/godap v0.0.0-20170216002349-c249933bc092
github.com/coreos/go-oidc/v3 v3.5.0
github.com/Microsoft/go-winio v0.5.2 // indirect
github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7 // indirect
github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239 // indirect
+ github.com/aws/aws-sdk-go-v2/config v1.27.11 // indirect
+ github.com/aws/aws-sdk-go-v2/credentials v1.17.11 // indirect
+ github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.5 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.5 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 // indirect
+ github.com/aws/aws-sdk-go-v2/service/ec2 v1.160.0 // indirect
+ github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.2 // indirect
+ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7 // indirect
+ github.com/aws/aws-sdk-go-v2/service/sso v1.20.5 // indirect
+ github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4 // indirect
+ github.com/aws/aws-sdk-go-v2/service/sts v1.28.6 // indirect
+ github.com/aws/smithy-go v1.20.2 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/bgentry/speakeasy v0.1.0 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/aws/aws-sdk-go v1.44.174/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
github.com/aws/aws-sdk-go-v2 v0.23.0 h1:+E1q1LLSfHSDn/DzOtdJOX+pLZE2HiNV2yO5AjZINwM=
github.com/aws/aws-sdk-go-v2 v0.23.0/go.mod h1:2LhT7UgHOXK3UXONKI5OMgIyoQL6zTAw/jwIeX6yqzw=
+github.com/aws/aws-sdk-go-v2 v1.26.1 h1:5554eUqIYVWpU0YmeeYZ0wU64H2VLBs8TlhRB2L+EkA=
+github.com/aws/aws-sdk-go-v2 v1.26.1/go.mod h1:ffIFB97e2yNsv4aTSGkqtHnppsIJzw7G7BReUZ3jCXM=
+github.com/aws/aws-sdk-go-v2/config v1.27.11 h1:f47rANd2LQEYHda2ddSCKYId18/8BhSRM4BULGmfgNA=
+github.com/aws/aws-sdk-go-v2/config v1.27.11/go.mod h1:SMsV78RIOYdve1vf36z8LmnszlRWkwMQtomCAI0/mIE=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.11 h1:YuIB1dJNf1Re822rriUOTxopaHHvIq0l/pX3fwO+Tzs=
+github.com/aws/aws-sdk-go-v2/credentials v1.17.11/go.mod h1:AQtFPsDH9bI2O+71anW6EKL+NcD7LG3dpKGMV4SShgo=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1 h1:FVJ0r5XTHSmIHJV6KuDmdYhEpvlHpiSd38RQWhut5J4=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1/go.mod h1:zusuAeqezXzAB24LGuzuekqMAEgWkVYukBec3kr3jUg=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.5 h1:aw39xVGeRWlWx9EzGVnhOR4yOjQDHPQ6o6NmBlscyQg=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.5/go.mod h1:FSaRudD0dXiMPK2UjknVwwTYyZMRsHv3TtkabsZih5I=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.5 h1:PG1F3OD1szkuQPzDw3CIQsRIrtTlUC3lP84taWzHlq0=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.5/go.mod h1:jU1li6RFryMz+so64PpKtudI+QzbKoIEivqdf6LNpOc=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 h1:hT8rVHwugYE2lEfdFE0QWVo81lF7jMrYJVDWI+f+VxU=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0/go.mod h1:8tu/lYfQfFe6IGnaOdrpVgEL2IrrDOf6/m9RQum4NkY=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.160.0 h1:ooy0OFbrdSwgk32OFGPnvBwry5ySYCKkgTEbQ2hejs8=
+github.com/aws/aws-sdk-go-v2/service/ec2 v1.160.0/go.mod h1:xejKuuRDjz6z5OqyeLsz01MlOqqW7CqpAB4PabNvpu8=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.2 h1:Ji0DY1xUsUr3I8cHps0G+XM3WWU16lP6yG8qu1GAZAs=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.2/go.mod h1:5CsjAbs3NlGQyZNFACh+zztPDI7fU6eW9QsxjfnuBKg=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7 h1:ogRAwT1/gxJBcSWDMZlgyFUM962F51A5CRhDLbxLdmo=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7/go.mod h1:YCsIZhXfRPLFFCl5xxY+1T9RKzOKjCut+28JSX2DnAk=
+github.com/aws/aws-sdk-go-v2/service/sso v1.20.5 h1:vN8hEbpRnL7+Hopy9dzmRle1xmDc7o8tmY0klsr175w=
+github.com/aws/aws-sdk-go-v2/service/sso v1.20.5/go.mod h1:qGzynb/msuZIE8I75DVRCUXw3o3ZyBmUvMwQ2t/BrGM=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4 h1:Jux+gDDyi1Lruk+KHF91tK2KCuY61kzoCpvtvJJBtOE=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4/go.mod h1:mUYPBhaF2lGiukDEjJX2BLRRKTmoUSitGDUgM4tRxak=
+github.com/aws/aws-sdk-go-v2/service/sts v1.28.6 h1:cwIxeBttqPN3qkaAjcEcsh8NYr8n2HZPkcKgPAi1phU=
+github.com/aws/aws-sdk-go-v2/service/sts v1.28.6/go.mod h1:FZf1/nKNEkHdGGJP/cI2MoIMquumuRK6ol3QQJNDxmw=
+github.com/aws/smithy-go v1.20.2 h1:tbp628ireGtzcHDDmLT/6ADHidqnwgF57XOXZe6tp4Q=
+github.com/aws/smithy-go v1.20.2/go.mod h1:krry+ya/rV9RDcV/Q16kpu6ypI4K2czasz0NC3qS14E=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bgentry/speakeasy v0.1.0 h1:ByYyxL9InA1OWqxJqqp2A5pYHUrCiAL6K3J+LKSsQkY=
package ec2
import (
+ "context"
"crypto/md5"
"crypto/rsa"
"crypto/sha1"
"crypto/x509"
"encoding/base64"
"encoding/json"
+ "errors"
"fmt"
"math/big"
+ "os"
"regexp"
"strconv"
"strings"
"git.arvados.org/arvados.git/lib/cloud"
"git.arvados.org/arvados.git/sdk/go/arvados"
- "github.com/aws/aws-sdk-go/aws"
- "github.com/aws/aws-sdk-go/aws/awserr"
- "github.com/aws/aws-sdk-go/aws/credentials"
- "github.com/aws/aws-sdk-go/aws/credentials/ec2rolecreds"
- "github.com/aws/aws-sdk-go/aws/ec2metadata"
- "github.com/aws/aws-sdk-go/aws/request"
- "github.com/aws/aws-sdk-go/aws/session"
- "github.com/aws/aws-sdk-go/service/ec2"
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/aws/retry"
+ awsconfig "github.com/aws/aws-sdk-go-v2/config"
+ "github.com/aws/aws-sdk-go-v2/service/ec2"
+ "github.com/aws/aws-sdk-go-v2/service/ec2/types"
+ "github.com/aws/smithy-go"
"github.com/prometheus/client_golang/prometheus"
"github.com/sirupsen/logrus"
"golang.org/x/crypto/ssh"
SecurityGroupIDs arvados.StringSet
SubnetID sliceOrSingleString
AdminUsername string
- EBSVolumeType string
+ EBSVolumeType types.VolumeType
EBSPrice float64
IAMInstanceProfile string
SpotPriceUpdateInterval arvados.Duration
}
type ec2Interface interface {
- DescribeKeyPairs(input *ec2.DescribeKeyPairsInput) (*ec2.DescribeKeyPairsOutput, error)
- ImportKeyPair(input *ec2.ImportKeyPairInput) (*ec2.ImportKeyPairOutput, error)
- RunInstances(input *ec2.RunInstancesInput) (*ec2.Reservation, error)
- DescribeInstances(input *ec2.DescribeInstancesInput) (*ec2.DescribeInstancesOutput, error)
- DescribeInstanceStatusPages(input *ec2.DescribeInstanceStatusInput, fn func(*ec2.DescribeInstanceStatusOutput, bool) bool) error
- DescribeSpotPriceHistoryPages(input *ec2.DescribeSpotPriceHistoryInput, fn func(*ec2.DescribeSpotPriceHistoryOutput, bool) bool) error
- CreateTags(input *ec2.CreateTagsInput) (*ec2.CreateTagsOutput, error)
- TerminateInstances(input *ec2.TerminateInstancesInput) (*ec2.TerminateInstancesOutput, error)
+ DescribeKeyPairs(context.Context, *ec2.DescribeKeyPairsInput, ...func(*ec2.Options)) (*ec2.DescribeKeyPairsOutput, error)
+ ImportKeyPair(context.Context, *ec2.ImportKeyPairInput, ...func(*ec2.Options)) (*ec2.ImportKeyPairOutput, error)
+ RunInstances(context.Context, *ec2.RunInstancesInput, ...func(*ec2.Options)) (*ec2.RunInstancesOutput, error)
+ DescribeInstances(context.Context, *ec2.DescribeInstancesInput, ...func(*ec2.Options)) (*ec2.DescribeInstancesOutput, error)
+ DescribeInstanceStatus(context.Context, *ec2.DescribeInstanceStatusInput, ...func(*ec2.Options)) (*ec2.DescribeInstanceStatusOutput, error)
+ DescribeSpotPriceHistory(context.Context, *ec2.DescribeSpotPriceHistoryInput, ...func(*ec2.Options)) (*ec2.DescribeSpotPriceHistoryOutput, error)
+ CreateTags(context.Context, *ec2.CreateTagsInput, ...func(*ec2.Options)) (*ec2.CreateTagsOutput, error)
+ TerminateInstances(context.Context, *ec2.TerminateInstancesInput, ...func(*ec2.Options)) (*ec2.TerminateInstancesOutput, error)
}
type ec2InstanceSet struct {
return nil, err
}
- sess, err := session.NewSession()
+ if len(instanceSet.ec2config.AccessKeyID)+len(instanceSet.ec2config.SecretAccessKey) > 0 {
+ // AWS SDK will use credentials in environment vars if
+ // present.
+ os.Setenv("AWS_ACCESS_KEY_ID", instanceSet.ec2config.AccessKeyID)
+ os.Setenv("AWS_SECRET_ACCESS_KEY", instanceSet.ec2config.SecretAccessKey)
+ } else {
+ os.Unsetenv("AWS_ACCESS_KEY_ID")
+ os.Unsetenv("AWS_SECRET_ACCESS_KEY")
+ }
+ awsConfig, err := awsconfig.LoadDefaultConfig(context.TODO(),
+ awsconfig.WithRegion(instanceSet.ec2config.Region))
if err != nil {
return nil, err
}
- // First try any static credentials, fall back to an IAM instance profile/role
- creds := credentials.NewChainCredentials(
- []credentials.Provider{
- &credentials.StaticProvider{Value: credentials.Value{AccessKeyID: instanceSet.ec2config.AccessKeyID, SecretAccessKey: instanceSet.ec2config.SecretAccessKey}},
- &ec2rolecreds.EC2RoleProvider{Client: ec2metadata.New(sess)},
- })
- awsConfig := aws.NewConfig().WithCredentials(creds).WithRegion(instanceSet.ec2config.Region)
- instanceSet.client = ec2.New(session.Must(session.NewSession(awsConfig)))
+ instanceSet.client = ec2.NewFromConfig(awsConfig)
instanceSet.keys = make(map[string]string)
if instanceSet.ec2config.EBSVolumeType == "" {
instanceSet.ec2config.EBSVolumeType = "gp2"
initCommand cloud.InitCommand,
publicKey ssh.PublicKey) (cloud.Instance, error) {
- ec2tags := []*ec2.Tag{}
+ ec2tags := []types.Tag{}
for k, v := range newTags {
- ec2tags = append(ec2tags, &ec2.Tag{
+ ec2tags = append(ec2tags, types.Tag{
Key: aws.String(k),
Value: aws.String(v),
})
rii := ec2.RunInstancesInput{
ImageId: aws.String(string(imageID)),
- InstanceType: &instanceType.ProviderType,
- MaxCount: aws.Int64(1),
- MinCount: aws.Int64(1),
-
- NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{
- {
- AssociatePublicIpAddress: aws.Bool(false),
- DeleteOnTermination: aws.Bool(true),
- DeviceIndex: aws.Int64(0),
- Groups: aws.StringSlice(groups),
- }},
+ InstanceType: types.InstanceType(instanceType.ProviderType),
+ MaxCount: aws.Int32(1),
+ MinCount: aws.Int32(1),
+
+ NetworkInterfaces: []types.InstanceNetworkInterfaceSpecification{{
+ AssociatePublicIpAddress: aws.Bool(false),
+ DeleteOnTermination: aws.Bool(true),
+ DeviceIndex: aws.Int32(0),
+ Groups: groups,
+ }},
DisableApiTermination: aws.Bool(false),
- InstanceInitiatedShutdownBehavior: aws.String("terminate"),
- TagSpecifications: []*ec2.TagSpecification{
+ InstanceInitiatedShutdownBehavior: types.ShutdownBehaviorTerminate,
+ TagSpecifications: []types.TagSpecification{
{
- ResourceType: aws.String("instance"),
+ ResourceType: types.ResourceTypeInstance,
Tags: ec2tags,
}},
- MetadataOptions: &ec2.InstanceMetadataOptionsRequest{
+ MetadataOptions: &types.InstanceMetadataOptionsRequest{
// Require IMDSv2, as described at
// https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-IMDS-new-instances.html
- HttpEndpoint: aws.String(ec2.InstanceMetadataEndpointStateEnabled),
- HttpTokens: aws.String(ec2.HttpTokensStateRequired),
+ HttpEndpoint: types.InstanceMetadataEndpointStateEnabled,
+ HttpTokens: types.HttpTokensStateRequired,
},
UserData: aws.String(base64.StdEncoding.EncodeToString([]byte("#!/bin/sh\n" + initCommand + "\n"))),
}
}
if instanceType.AddedScratch > 0 {
- rii.BlockDeviceMappings = []*ec2.BlockDeviceMapping{{
+ rii.BlockDeviceMappings = []types.BlockDeviceMapping{{
DeviceName: aws.String("/dev/xvdt"),
- Ebs: &ec2.EbsBlockDevice{
+ Ebs: &types.EbsBlockDevice{
DeleteOnTermination: aws.Bool(true),
- VolumeSize: aws.Int64((int64(instanceType.AddedScratch) + (1<<30 - 1)) >> 30),
- VolumeType: &instanceSet.ec2config.EBSVolumeType,
+ VolumeSize: aws.Int32(int32((int64(instanceType.AddedScratch) + (1<<30 - 1)) >> 30)),
+ VolumeType: instanceSet.ec2config.EBSVolumeType,
}}}
}
if instanceType.Preemptible {
- rii.InstanceMarketOptions = &ec2.InstanceMarketOptionsRequest{
- MarketType: aws.String("spot"),
- SpotOptions: &ec2.SpotMarketOptions{
- InstanceInterruptionBehavior: aws.String("terminate"),
+ rii.InstanceMarketOptions = &types.InstanceMarketOptionsRequest{
+ MarketType: types.MarketTypeSpot,
+ SpotOptions: &types.SpotMarketOptions{
+ InstanceInterruptionBehavior: types.InstanceInterruptionBehaviorTerminate,
MaxPrice: aws.String(fmt.Sprintf("%v", instanceType.Price)),
}}
}
if instanceSet.ec2config.IAMInstanceProfile != "" {
- rii.IamInstanceProfile = &ec2.IamInstanceProfileSpecification{
+ rii.IamInstanceProfile = &types.IamInstanceProfileSpecification{
Name: aws.String(instanceSet.ec2config.IAMInstanceProfile),
}
}
- var rsv *ec2.Reservation
+ var rsv *ec2.RunInstancesOutput
var errToReturn error
subnets := instanceSet.ec2config.SubnetID
currentSubnetIDIndex := int(atomic.LoadInt32(&instanceSet.currentSubnetIDIndex))
rii.NetworkInterfaces[0].SubnetId = aws.String(trySubnet)
}
var err error
- rsv, err = instanceSet.client.RunInstances(&rii)
+ rsv, err = instanceSet.client.RunInstances(context.TODO(), &rii)
instanceSet.mInstanceStarts.WithLabelValues(trySubnet, boolLabelValue[err == nil]).Add(1)
if !isErrorCapacity(errToReturn) || isErrorCapacity(err) {
// We want to return the last capacity error,
if keyname, ok := instanceSet.keys[md5keyFingerprint]; ok {
return keyname, nil
}
- keyout, err := instanceSet.client.DescribeKeyPairs(&ec2.DescribeKeyPairsInput{
- Filters: []*ec2.Filter{{
+ keyout, err := instanceSet.client.DescribeKeyPairs(context.TODO(), &ec2.DescribeKeyPairsInput{
+ Filters: []types.Filter{{
Name: aws.String("fingerprint"),
- Values: []*string{&md5keyFingerprint, &sha1keyFingerprint},
+ Values: []string{md5keyFingerprint, sha1keyFingerprint},
}},
})
if err != nil {
return *(keyout.KeyPairs[0].KeyName), nil
}
keyname := "arvados-dispatch-keypair-" + md5keyFingerprint
- _, err = instanceSet.client.ImportKeyPair(&ec2.ImportKeyPairInput{
+ _, err = instanceSet.client.ImportKeyPair(context.TODO(), &ec2.ImportKeyPairInput{
KeyName: &keyname,
PublicKeyMaterial: ssh.MarshalAuthorizedKey(publicKey),
})
}
func (instanceSet *ec2InstanceSet) Instances(tags cloud.InstanceTags) (instances []cloud.Instance, err error) {
- var filters []*ec2.Filter
+ var filters []types.Filter
for k, v := range tags {
- filters = append(filters, &ec2.Filter{
+ filters = append(filters, types.Filter{
Name: aws.String("tag:" + k),
- Values: []*string{aws.String(v)},
+ Values: []string{v},
})
}
needAZs := false
dii := &ec2.DescribeInstancesInput{Filters: filters}
for {
- dio, err := instanceSet.client.DescribeInstances(dii)
+ dio, err := instanceSet.client.DescribeInstances(context.TODO(), dii)
err = wrapError(err, &instanceSet.throttleDelayInstances)
if err != nil {
return nil, err
for _, rsv := range dio.Reservations {
for _, inst := range rsv.Instances {
- if *inst.State.Name != "shutting-down" && *inst.State.Name != "terminated" {
+ switch inst.State.Name {
+ case types.InstanceStateNameShuttingDown:
+ case types.InstanceStateNameTerminated:
+ default:
instances = append(instances, &ec2Instance{
provider: instanceSet,
instance: inst,
})
- if aws.StringValue(inst.InstanceLifecycle) == "spot" {
+ if inst.InstanceLifecycle == types.InstanceLifecycleTypeSpot {
needAZs = true
}
}
}
if needAZs && instanceSet.ec2config.SpotPriceUpdateInterval > 0 {
az := map[string]string{}
- err := instanceSet.client.DescribeInstanceStatusPages(&ec2.DescribeInstanceStatusInput{
- IncludeAllInstances: aws.Bool(true),
- }, func(page *ec2.DescribeInstanceStatusOutput, lastPage bool) bool {
+ disi := &ec2.DescribeInstanceStatusInput{IncludeAllInstances: aws.Bool(true)}
+ for {
+ page, err := instanceSet.client.DescribeInstanceStatus(context.TODO(), disi)
+ if err != nil {
+ instanceSet.logger.Warnf("error getting instance statuses: %s", err)
+ break
+ }
for _, ent := range page.InstanceStatuses {
az[*ent.InstanceId] = *ent.AvailabilityZone
}
- return true
- })
- if err != nil {
- instanceSet.logger.Warnf("error getting instance statuses: %s", err)
+ if page.NextToken == nil {
+ break
+ }
+ disi.NextToken = page.NextToken
}
for _, inst := range instances {
inst := inst.(*ec2Instance)
updateTime := time.Now()
staleTime := updateTime.Add(-instanceSet.ec2config.SpotPriceUpdateInterval.Duration())
needUpdate := false
- allTypes := map[string]bool{}
+ allTypes := map[types.InstanceType]bool{}
for _, inst := range instances {
ec2inst := inst.(*ec2Instance).instance
- if aws.StringValue(ec2inst.InstanceLifecycle) == "spot" {
+ if ec2inst.InstanceLifecycle == types.InstanceLifecycleTypeSpot {
pk := priceKey{
- instanceType: *ec2inst.InstanceType,
+ instanceType: string(ec2inst.InstanceType),
spot: true,
availabilityZone: inst.(*ec2Instance).availabilityZone,
}
if instanceSet.pricesUpdated[pk].Before(staleTime) {
needUpdate = true
}
- allTypes[*ec2inst.InstanceType] = true
+ allTypes[ec2inst.InstanceType] = true
}
}
if !needUpdate {
return
}
- var typeFilterValues []*string
+ var typeFilterValues []string
for instanceType := range allTypes {
- typeFilterValues = append(typeFilterValues, aws.String(instanceType))
+ typeFilterValues = append(typeFilterValues, string(instanceType))
}
// Get 3x update interval worth of pricing data. (Ideally the
// AWS API would tell us "we have shown you all of the price
// row.
dsphi := &ec2.DescribeSpotPriceHistoryInput{
StartTime: aws.Time(updateTime.Add(-3 * instanceSet.ec2config.SpotPriceUpdateInterval.Duration())),
- Filters: []*ec2.Filter{
- &ec2.Filter{Name: aws.String("instance-type"), Values: typeFilterValues},
- &ec2.Filter{Name: aws.String("product-description"), Values: []*string{aws.String("Linux/UNIX")}},
+ Filters: []types.Filter{
+ types.Filter{Name: aws.String("instance-type"), Values: typeFilterValues},
+ types.Filter{Name: aws.String("product-description"), Values: []string{"Linux/UNIX"}},
},
}
- err := instanceSet.client.DescribeSpotPriceHistoryPages(dsphi, func(page *ec2.DescribeSpotPriceHistoryOutput, lastPage bool) bool {
+ for {
+ page, err := instanceSet.client.DescribeSpotPriceHistory(context.TODO(), dsphi)
+ if err != nil {
+ instanceSet.logger.Warnf("error retrieving spot instance prices: %s", err)
+ break
+ }
for _, ent := range page.SpotPriceHistory {
- if ent.InstanceType == nil || ent.SpotPrice == nil || ent.Timestamp == nil {
+ if ent.InstanceType == "" || ent.SpotPrice == nil || ent.Timestamp == nil {
// bogus record?
continue
}
continue
}
pk := priceKey{
- instanceType: *ent.InstanceType,
+ instanceType: string(ent.InstanceType),
spot: true,
availabilityZone: *ent.AvailabilityZone,
}
})
instanceSet.pricesUpdated[pk] = updateTime
}
- return true
- })
- if err != nil {
- instanceSet.logger.Warnf("error retrieving spot instance prices: %s", err)
+ if page.NextToken == nil {
+ break
+ }
+ dsphi.NextToken = page.NextToken
}
expiredTime := updateTime.Add(-64 * instanceSet.ec2config.SpotPriceUpdateInterval.Duration())
type ec2Instance struct {
provider *ec2InstanceSet
- instance *ec2.Instance
+ instance types.Instance
availabilityZone string // sometimes available for spot instances
}
}
func (inst *ec2Instance) ProviderType() string {
- return *inst.instance.InstanceType
+ return string(inst.instance.InstanceType)
}
func (inst *ec2Instance) SetTags(newTags cloud.InstanceTags) error {
- var ec2tags []*ec2.Tag
+ var ec2tags []types.Tag
for k, v := range newTags {
- ec2tags = append(ec2tags, &ec2.Tag{
+ ec2tags = append(ec2tags, types.Tag{
Key: aws.String(k),
Value: aws.String(v),
})
}
- _, err := inst.provider.client.CreateTags(&ec2.CreateTagsInput{
- Resources: []*string{inst.instance.InstanceId},
+ _, err := inst.provider.client.CreateTags(context.TODO(), &ec2.CreateTagsInput{
+ Resources: []string{*inst.instance.InstanceId},
Tags: ec2tags,
})
}
func (inst *ec2Instance) Destroy() error {
- _, err := inst.provider.client.TerminateInstances(&ec2.TerminateInstancesInput{
- InstanceIds: []*string{inst.instance.InstanceId},
+ _, err := inst.provider.client.TerminateInstances(context.TODO(), &ec2.TerminateInstancesInput{
+ InstanceIds: []string{*inst.instance.InstanceId},
})
return err
}
// inst.provider.prices only for spot instances, so if
// spot==false here, we will return no data.
pk := priceKey{
- instanceType: *inst.instance.InstanceType,
- spot: aws.StringValue(inst.instance.InstanceLifecycle) == "spot",
+ instanceType: string(inst.instance.InstanceType),
+ spot: inst.instance.InstanceLifecycle == types.InstanceLifecycleTypeSpot,
availabilityZone: inst.availabilityZone,
}
var prices []cloud.InstancePrice
//
// Returns false if error is nil.
func isErrorQuota(err error) bool {
- if aerr, ok := err.(awserr.Error); ok && aerr != nil {
- if _, ok := isCodeQuota[aerr.Code()]; ok {
+ var aerr smithy.APIError
+ if errors.As(err, &aerr) {
+ if _, ok := isCodeQuota[aerr.ErrorCode()]; ok {
return true
}
}
// isErrorSubnetSpecific returns true if the problem encountered by
// RunInstances might be avoided by trying a different subnet.
func isErrorSubnetSpecific(err error) bool {
- aerr, ok := err.(awserr.Error)
- if !ok {
+ var aerr smithy.APIError
+ if !errors.As(err, &aerr) {
return false
}
- code := aerr.Code()
+ code := aerr.ErrorCode()
return strings.Contains(code, "Subnet") ||
code == "InsufficientInstanceCapacity" ||
code == "InsufficientVolumeCapacity" ||
// we look for substrings in code/message instead of
// only using specific codes here.
(strings.Contains(code, "InvalidParameter") &&
- reSubnetSpecificInvalidParameterMessage.MatchString(aerr.Message()))
+ reSubnetSpecificInvalidParameterMessage.MatchString(aerr.ErrorMessage()))
}
// isErrorCapacity returns true if the error indicates lack of
// type -- i.e., retrying with a different instance type might
// succeed.
func isErrorCapacity(err error) bool {
- aerr, ok := err.(awserr.Error)
- if !ok {
+ var aerr smithy.APIError
+ if !errors.As(err, &aerr) {
return false
}
- code := aerr.Code()
+ code := aerr.ErrorCode()
return code == "InsufficientInstanceCapacity" ||
- (code == "Unsupported" && strings.Contains(aerr.Message(), "requested instance type"))
+ (code == "Unsupported" && strings.Contains(aerr.ErrorMessage(), "requested instance type"))
}
type ec2QuotaError struct {
return true
}
+func isThrottleError(err error) bool {
+ var aerr smithy.APIError
+ if !errors.As(err, &aerr) {
+ return false
+ }
+ _, is := retry.DefaultThrottleErrorCodes[aerr.ErrorCode()]
+ return is
+}
+
func wrapError(err error, throttleValue *atomic.Value) error {
- if request.IsErrorThrottle(err) {
+ if isThrottleError(err) {
// Back off exponentially until an upstream call
// either succeeds or returns a non-throttle error.
d, _ := throttleValue.Load().(time.Duration)
package ec2
import (
+ "context"
"encoding/json"
"errors"
"flag"
- "fmt"
"sync/atomic"
"testing"
"time"
"git.arvados.org/arvados.git/sdk/go/arvadostest"
"git.arvados.org/arvados.git/sdk/go/config"
"git.arvados.org/arvados.git/sdk/go/ctxlog"
- "github.com/aws/aws-sdk-go/aws"
- "github.com/aws/aws-sdk-go/aws/awserr"
- "github.com/aws/aws-sdk-go/service/ec2"
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/service/ec2"
+ "github.com/aws/aws-sdk-go-v2/service/ec2/types"
+ "github.com/aws/smithy-go"
"github.com/ghodss/yaml"
"github.com/prometheus/client_golang/prometheus"
"github.com/sirupsen/logrus"
subnetErrorOnRunInstances map[string]error
}
-func (e *ec2stub) ImportKeyPair(input *ec2.ImportKeyPairInput) (*ec2.ImportKeyPairOutput, error) {
+func (e *ec2stub) ImportKeyPair(ctx context.Context, input *ec2.ImportKeyPairInput, _ ...func(*ec2.Options)) (*ec2.ImportKeyPairOutput, error) {
e.importKeyPairCalls = append(e.importKeyPairCalls, input)
return nil, nil
}
-func (e *ec2stub) DescribeKeyPairs(input *ec2.DescribeKeyPairsInput) (*ec2.DescribeKeyPairsOutput, error) {
+func (e *ec2stub) DescribeKeyPairs(ctx context.Context, input *ec2.DescribeKeyPairsInput, _ ...func(*ec2.Options)) (*ec2.DescribeKeyPairsOutput, error) {
e.describeKeyPairsCalls = append(e.describeKeyPairsCalls, input)
return &ec2.DescribeKeyPairsOutput{}, nil
}
-func (e *ec2stub) RunInstances(input *ec2.RunInstancesInput) (*ec2.Reservation, error) {
+func (e *ec2stub) RunInstances(ctx context.Context, input *ec2.RunInstancesInput, _ ...func(*ec2.Options)) (*ec2.RunInstancesOutput, error) {
e.runInstancesCalls = append(e.runInstancesCalls, input)
if len(input.NetworkInterfaces) > 0 && input.NetworkInterfaces[0].SubnetId != nil {
err := e.subnetErrorOnRunInstances[*input.NetworkInterfaces[0].SubnetId]
return nil, err
}
}
- return &ec2.Reservation{Instances: []*ec2.Instance{{
+ return &ec2.RunInstancesOutput{Instances: []types.Instance{{
InstanceId: aws.String("i-123"),
- InstanceType: aws.String("t2.micro"),
+ InstanceType: types.InstanceTypeT2Micro,
Tags: input.TagSpecifications[0].Tags,
}}}, nil
}
-func (e *ec2stub) DescribeInstances(input *ec2.DescribeInstancesInput) (*ec2.DescribeInstancesOutput, error) {
+func (e *ec2stub) DescribeInstances(ctx context.Context, input *ec2.DescribeInstancesInput, _ ...func(*ec2.Options)) (*ec2.DescribeInstancesOutput, error) {
return &ec2.DescribeInstancesOutput{
- Reservations: []*ec2.Reservation{{
- Instances: []*ec2.Instance{{
+ Reservations: []types.Reservation{{
+ Instances: []types.Instance{{
InstanceId: aws.String("i-123"),
- InstanceLifecycle: aws.String("spot"),
- InstanceType: aws.String("t2.micro"),
+ InstanceLifecycle: types.InstanceLifecycleTypeSpot,
+ InstanceType: types.InstanceTypeT2Micro,
PrivateIpAddress: aws.String("10.1.2.3"),
- State: &ec2.InstanceState{Name: aws.String("running"), Code: aws.Int64(16)},
+ State: &types.InstanceState{Name: types.InstanceStateNameRunning, Code: aws.Int32(16)},
}, {
InstanceId: aws.String("i-124"),
- InstanceLifecycle: aws.String("spot"),
- InstanceType: aws.String("t2.micro"),
+ InstanceLifecycle: types.InstanceLifecycleTypeSpot,
+ InstanceType: types.InstanceTypeT2Micro,
PrivateIpAddress: aws.String("10.1.2.4"),
- State: &ec2.InstanceState{Name: aws.String("running"), Code: aws.Int64(16)},
+ State: &types.InstanceState{Name: types.InstanceStateNameRunning, Code: aws.Int32(16)},
}},
}},
}, nil
}
-func (e *ec2stub) DescribeInstanceStatusPages(input *ec2.DescribeInstanceStatusInput, fn func(*ec2.DescribeInstanceStatusOutput, bool) bool) error {
- fn(&ec2.DescribeInstanceStatusOutput{
- InstanceStatuses: []*ec2.InstanceStatus{{
+func (e *ec2stub) DescribeInstanceStatus(ctx context.Context, input *ec2.DescribeInstanceStatusInput, _ ...func(*ec2.Options)) (*ec2.DescribeInstanceStatusOutput, error) {
+ return &ec2.DescribeInstanceStatusOutput{
+ InstanceStatuses: []types.InstanceStatus{{
InstanceId: aws.String("i-123"),
AvailabilityZone: aws.String("aa-east-1a"),
}, {
InstanceId: aws.String("i-124"),
AvailabilityZone: aws.String("aa-east-1a"),
}},
- }, true)
- return nil
+ }, nil
}
-func (e *ec2stub) DescribeSpotPriceHistoryPages(input *ec2.DescribeSpotPriceHistoryInput, fn func(*ec2.DescribeSpotPriceHistoryOutput, bool) bool) error {
- if !fn(&ec2.DescribeSpotPriceHistoryOutput{
- SpotPriceHistory: []*ec2.SpotPrice{
- &ec2.SpotPrice{
- InstanceType: aws.String("t2.micro"),
- AvailabilityZone: aws.String("aa-east-1a"),
- SpotPrice: aws.String("0.005"),
- Timestamp: aws.Time(e.reftime.Add(-9 * time.Minute)),
+func (e *ec2stub) DescribeSpotPriceHistory(ctx context.Context, input *ec2.DescribeSpotPriceHistoryInput, _ ...func(*ec2.Options)) (*ec2.DescribeSpotPriceHistoryOutput, error) {
+ if input.NextToken == nil {
+ return &ec2.DescribeSpotPriceHistoryOutput{
+ SpotPriceHistory: []types.SpotPrice{
+ types.SpotPrice{
+ InstanceType: types.InstanceTypeT2Micro,
+ AvailabilityZone: aws.String("aa-east-1a"),
+ SpotPrice: aws.String("0.005"),
+ Timestamp: aws.Time(e.reftime.Add(-9 * time.Minute)),
+ },
+ types.SpotPrice{
+ InstanceType: types.InstanceTypeT2Micro,
+ AvailabilityZone: aws.String("aa-east-1a"),
+ SpotPrice: aws.String("0.015"),
+ Timestamp: aws.Time(e.reftime.Add(-5 * time.Minute)),
+ },
},
- &ec2.SpotPrice{
- InstanceType: aws.String("t2.micro"),
- AvailabilityZone: aws.String("aa-east-1a"),
- SpotPrice: aws.String("0.015"),
- Timestamp: aws.Time(e.reftime.Add(-5 * time.Minute)),
+ NextToken: aws.String("stubnexttoken"),
+ }, nil
+ } else {
+ return &ec2.DescribeSpotPriceHistoryOutput{
+ SpotPriceHistory: []types.SpotPrice{
+ types.SpotPrice{
+ InstanceType: types.InstanceTypeT2Micro,
+ AvailabilityZone: aws.String("aa-east-1a"),
+ SpotPrice: aws.String("0.01"),
+ Timestamp: aws.Time(e.reftime.Add(-2 * time.Minute)),
+ },
},
- },
- }, false) {
- return nil
+ }, nil
}
- fn(&ec2.DescribeSpotPriceHistoryOutput{
- SpotPriceHistory: []*ec2.SpotPrice{
- &ec2.SpotPrice{
- InstanceType: aws.String("t2.micro"),
- AvailabilityZone: aws.String("aa-east-1a"),
- SpotPrice: aws.String("0.01"),
- Timestamp: aws.Time(e.reftime.Add(-2 * time.Minute)),
- },
- },
- }, true)
- return nil
}
-func (e *ec2stub) CreateTags(input *ec2.CreateTagsInput) (*ec2.CreateTagsOutput, error) {
+func (e *ec2stub) CreateTags(ctx context.Context, input *ec2.CreateTagsInput, _ ...func(*ec2.Options)) (*ec2.CreateTagsOutput, error) {
return nil, nil
}
-func (e *ec2stub) TerminateInstances(input *ec2.TerminateInstancesInput) (*ec2.TerminateInstancesOutput, error) {
+func (e *ec2stub) TerminateInstances(ctx context.Context, input *ec2.TerminateInstancesInput, _ ...func(*ec2.Options)) (*ec2.TerminateInstancesOutput, error) {
return nil, nil
}
-type ec2stubError struct {
- code string
- message string
-}
-
-func (err *ec2stubError) Code() string { return err.code }
-func (err *ec2stubError) Message() string { return err.message }
-func (err *ec2stubError) Error() string { return fmt.Sprintf("%s: %s", err.code, err.message) }
-func (err *ec2stubError) OrigErr() error { return errors.New("stub OrigErr") }
+type ec2stubError = smithy.GenericAPIError
-// Ensure ec2stubError satisfies the aws.Error interface
-var _ = awserr.Error(&ec2stubError{})
+// Ensure ec2stubError satisfies the smithy.APIError interface
+var _ = smithy.APIError(&ec2stubError{})
func GetInstanceSet(c *check.C, conf string) (*ec2InstanceSet, cloud.ImageID, arvados.Cluster, *prometheus.Registry) {
reg := prometheus.NewRegistry()
runcalls := ap.client.(*ec2stub).runInstancesCalls
if c.Check(runcalls, check.HasLen, 1) {
- c.Check(runcalls[0].MetadataOptions.HttpEndpoint, check.DeepEquals, aws.String("enabled"))
- c.Check(runcalls[0].MetadataOptions.HttpTokens, check.DeepEquals, aws.String("required"))
+ c.Check(runcalls[0].MetadataOptions.HttpEndpoint, check.DeepEquals, types.InstanceMetadataEndpointStateEnabled)
+ c.Check(runcalls[0].MetadataOptions.HttpTokens, check.DeepEquals, types.HttpTokensStateRequired)
}
}
}
ap, img, cluster, reg := GetInstanceSet(c, `{"SubnetID":["subnet-full","subnet-good"]}`)
ap.client.(*ec2stub).subnetErrorOnRunInstances = map[string]error{
"subnet-full": &ec2stubError{
- code: "InsufficientFreeAddressesInSubnet",
- message: "subnet is full",
+ Code: "InsufficientFreeAddressesInSubnet",
+ Message: "subnet is full",
},
}
inst, err := ap.Create(cluster.InstanceTypes["tiny"], img, nil, "", nil)
c.Check(isErrorSubnetSpecific(errors.New("misc error")), check.Equals, false)
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "InsufficientInstanceCapacity",
+ Code: "InsufficientInstanceCapacity",
}), check.Equals, true)
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "InsufficientVolumeCapacity",
+ Code: "InsufficientVolumeCapacity",
}), check.Equals, true)
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "InsufficientFreeAddressesInSubnet",
- message: "Not enough free addresses in subnet subnet-abcdefg\n\tstatus code: 400, request id: abcdef01-2345-6789-abcd-ef0123456789",
+ Code: "InsufficientFreeAddressesInSubnet",
+ Message: "Not enough free addresses in subnet subnet-abcdefg\n\tstatus code: 400, request id: abcdef01-2345-6789-abcd-ef0123456789",
}), check.Equals, true)
// #21603: (Sometimes?) EC2 returns code InvalidParameterValue
// even though the code "InsufficientFreeAddressesInSubnet"
// seems like it must be meant for exactly this error.
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "InvalidParameterValue",
- message: "Not enough free addresses in subnet subnet-abcdefg\n\tstatus code: 400, request id: abcdef01-2345-6789-abcd-ef0123456789",
+ Code: "InvalidParameterValue",
+ Message: "Not enough free addresses in subnet subnet-abcdefg\n\tstatus code: 400, request id: abcdef01-2345-6789-abcd-ef0123456789",
}), check.Equals, true)
// Similarly, AWS docs
// (https://repost.aws/knowledge-center/vpc-insufficient-ip-errors)
// suggest the following code/message combinations also exist.
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "Client.InvalidParameterValue",
- message: "There aren't sufficient free Ipv4 addresses or prefixes",
+ Code: "Client.InvalidParameterValue",
+ Message: "There aren't sufficient free Ipv4 addresses or prefixes",
}), check.Equals, true)
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "InvalidParameterValue",
- message: "There aren't sufficient free Ipv4 addresses or prefixes",
+ Code: "InvalidParameterValue",
+ Message: "There aren't sufficient free Ipv4 addresses or prefixes",
}), check.Equals, true)
// Meanwhile, other AWS docs
// (https://docs.aws.amazon.com/AWSEC2/latest/APIReference/errors-overview.html)
// suggest Client.InvalidParameterValue is not a real code but
// ClientInvalidParameterValue is.
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "ClientInvalidParameterValue",
- message: "There aren't sufficient free Ipv4 addresses or prefixes",
+ Code: "ClientInvalidParameterValue",
+ Message: "There aren't sufficient free Ipv4 addresses or prefixes",
}), check.Equals, true)
c.Check(isErrorSubnetSpecific(&ec2stubError{
- code: "InvalidParameterValue",
- message: "Some other invalid parameter error",
+ Code: "InvalidParameterValue",
+ Message: "Some other invalid parameter error",
}), check.Equals, false)
}
ap, img, cluster, reg := GetInstanceSet(c, `{"SubnetID":["subnet-full","subnet-broken"]}`)
ap.client.(*ec2stub).subnetErrorOnRunInstances = map[string]error{
"subnet-full": &ec2stubError{
- code: "InsufficientFreeAddressesInSubnet",
- message: "subnet is full",
+ Code: "InsufficientFreeAddressesInSubnet",
+ Message: "subnet is full",
},
"subnet-broken": &ec2stubError{
- code: "InvalidSubnetId.NotFound",
- message: "bogus subnet id",
+ Code: "InvalidSubnetId.NotFound",
+ Message: "bogus subnet id",
},
}
_, err := ap.Create(cluster.InstanceTypes["tiny"], img, nil, "", nil)
ap, img, cluster, reg := GetInstanceSet(c, `{"SubnetID":["subnet-full","subnet-broken"]}`)
ap.client.(*ec2stub).subnetErrorOnRunInstances = map[string]error{
"subnet-full": &ec2stubError{
- code: "InsufficientFreeAddressesInSubnet",
- message: "subnet is full",
+ Code: "InsufficientFreeAddressesInSubnet",
+ Message: "subnet is full",
},
"subnet-broken": &ec2stubError{
- code: "InsufficientInstanceCapacity",
- message: "insufficient capacity",
+ Code: "InsufficientInstanceCapacity",
+ Message: "insufficient capacity",
},
}
for i := 0; i < 3; i++ {
running := 0
for _, inst := range instances {
ec2i := inst.(*ec2Instance).instance
- if *ec2i.InstanceLifecycle == "spot" && *ec2i.State.Code&16 != 0 {
+ if ec2i.InstanceLifecycle == types.InstanceLifecycleTypeSpot && *ec2i.State.Code&16 != 0 {
running++
}
}
}
func (*EC2InstanceSetSuite) TestWrapError(c *check.C) {
- retryError := awserr.New("Throttling", "", nil)
+ retryError := &ec2stubError{Code: "Throttling"}
wrapped := wrapError(retryError, &atomic.Value{})
_, ok := wrapped.(cloud.RateLimitError)
c.Check(ok, check.Equals, true)
- quotaError := awserr.New("InstanceLimitExceeded", "", nil)
+ quotaError := &ec2stubError{Code: "InstanceLimitExceeded"}
wrapped = wrapError(quotaError, nil)
_, ok = wrapped.(cloud.QuotaError)
c.Check(ok, check.Equals, true)
{"InsufficientInstanceCapacity", ""},
{"Unsupported", "Your requested instance type (t3.micro) is not supported in your requested Availability Zone (us-east-1e). Please retry your request by not specifying an Availability Zone or choosing us-east-1a, us-east-1b, us-east-1c, us-east-1d, us-east-1f."},
} {
- capacityError := awserr.New(trial.code, trial.msg, nil)
+ capacityError := &ec2stubError{Code: trial.code, Message: trial.msg}
wrapped = wrapError(capacityError, nil)
caperr, ok := wrapped.(cloud.CapacityError)
c.Check(ok, check.Equals, true)
projects / arvados.git / commitdiff