19320: Comment re future use of spot attr in priceKey.
[arvados.git] / lib / cloud / ec2 / ec2.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package ec2
6
7 import (
8         "crypto/md5"
9         "crypto/rsa"
10         "crypto/sha1"
11         "crypto/x509"
12         "encoding/base64"
13         "encoding/json"
14         "fmt"
15         "math/big"
16         "strconv"
17         "sync"
18         "sync/atomic"
19         "time"
20
21         "git.arvados.org/arvados.git/lib/cloud"
22         "git.arvados.org/arvados.git/sdk/go/arvados"
23         "github.com/aws/aws-sdk-go/aws"
24         "github.com/aws/aws-sdk-go/aws/awserr"
25         "github.com/aws/aws-sdk-go/aws/credentials"
26         "github.com/aws/aws-sdk-go/aws/credentials/ec2rolecreds"
27         "github.com/aws/aws-sdk-go/aws/ec2metadata"
28         "github.com/aws/aws-sdk-go/aws/request"
29         "github.com/aws/aws-sdk-go/aws/session"
30         "github.com/aws/aws-sdk-go/service/ec2"
31         "github.com/sirupsen/logrus"
32         "golang.org/x/crypto/ssh"
33 )
34
35 // Driver is the ec2 implementation of the cloud.Driver interface.
36 var Driver = cloud.DriverFunc(newEC2InstanceSet)
37
38 const (
39         throttleDelayMin = time.Second
40         throttleDelayMax = time.Minute
41 )
42
43 type ec2InstanceSetConfig struct {
44         AccessKeyID             string
45         SecretAccessKey         string
46         Region                  string
47         SecurityGroupIDs        arvados.StringSet
48         SubnetID                string
49         AdminUsername           string
50         EBSVolumeType           string
51         EBSPrice                float64
52         IAMInstanceProfile      string
53         SpotPriceUpdateInterval arvados.Duration
54 }
55
56 type ec2Interface interface {
57         DescribeKeyPairs(input *ec2.DescribeKeyPairsInput) (*ec2.DescribeKeyPairsOutput, error)
58         ImportKeyPair(input *ec2.ImportKeyPairInput) (*ec2.ImportKeyPairOutput, error)
59         RunInstances(input *ec2.RunInstancesInput) (*ec2.Reservation, error)
60         DescribeInstances(input *ec2.DescribeInstancesInput) (*ec2.DescribeInstancesOutput, error)
61         DescribeInstanceStatusPages(input *ec2.DescribeInstanceStatusInput, fn func(*ec2.DescribeInstanceStatusOutput, bool) bool) error
62         DescribeSpotPriceHistoryPages(input *ec2.DescribeSpotPriceHistoryInput, fn func(*ec2.DescribeSpotPriceHistoryOutput, bool) bool) error
63         CreateTags(input *ec2.CreateTagsInput) (*ec2.CreateTagsOutput, error)
64         TerminateInstances(input *ec2.TerminateInstancesInput) (*ec2.TerminateInstancesOutput, error)
65 }
66
67 type ec2InstanceSet struct {
68         ec2config              ec2InstanceSetConfig
69         instanceSetID          cloud.InstanceSetID
70         logger                 logrus.FieldLogger
71         client                 ec2Interface
72         keysMtx                sync.Mutex
73         keys                   map[string]string
74         throttleDelayCreate    atomic.Value
75         throttleDelayInstances atomic.Value
76
77         prices        map[priceKey][]cloud.InstancePrice
78         pricesLock    sync.Mutex
79         pricesUpdated map[priceKey]time.Time
80 }
81
82 func newEC2InstanceSet(config json.RawMessage, instanceSetID cloud.InstanceSetID, _ cloud.SharedResourceTags, logger logrus.FieldLogger) (prv cloud.InstanceSet, err error) {
83         instanceSet := &ec2InstanceSet{
84                 instanceSetID: instanceSetID,
85                 logger:        logger,
86         }
87         err = json.Unmarshal(config, &instanceSet.ec2config)
88         if err != nil {
89                 return nil, err
90         }
91
92         sess, err := session.NewSession()
93         if err != nil {
94                 return nil, err
95         }
96         // First try any static credentials, fall back to an IAM instance profile/role
97         creds := credentials.NewChainCredentials(
98                 []credentials.Provider{
99                         &credentials.StaticProvider{Value: credentials.Value{AccessKeyID: instanceSet.ec2config.AccessKeyID, SecretAccessKey: instanceSet.ec2config.SecretAccessKey}},
100                         &ec2rolecreds.EC2RoleProvider{Client: ec2metadata.New(sess)},
101                 })
102
103         awsConfig := aws.NewConfig().WithCredentials(creds).WithRegion(instanceSet.ec2config.Region)
104         instanceSet.client = ec2.New(session.Must(session.NewSession(awsConfig)))
105         instanceSet.keys = make(map[string]string)
106         if instanceSet.ec2config.EBSVolumeType == "" {
107                 instanceSet.ec2config.EBSVolumeType = "gp2"
108         }
109         return instanceSet, nil
110 }
111
112 func awsKeyFingerprint(pk ssh.PublicKey) (md5fp string, sha1fp string, err error) {
113         // AWS key fingerprints don't use the usual key fingerprint
114         // you get from ssh-keygen or ssh.FingerprintLegacyMD5()
115         // (you can get that from md5.Sum(pk.Marshal())
116         //
117         // AWS uses the md5 or sha1 of the PKIX DER encoding of the
118         // public key, so calculate those fingerprints here.
119         var rsaPub struct {
120                 Name string
121                 E    *big.Int
122                 N    *big.Int
123         }
124         if err := ssh.Unmarshal(pk.Marshal(), &rsaPub); err != nil {
125                 return "", "", fmt.Errorf("agent: Unmarshal failed to parse public key: %v", err)
126         }
127         rsaPk := rsa.PublicKey{
128                 E: int(rsaPub.E.Int64()),
129                 N: rsaPub.N,
130         }
131         pkix, _ := x509.MarshalPKIXPublicKey(&rsaPk)
132         md5pkix := md5.Sum([]byte(pkix))
133         sha1pkix := sha1.Sum([]byte(pkix))
134         md5fp = ""
135         sha1fp = ""
136         for i := 0; i < len(md5pkix); i++ {
137                 md5fp += fmt.Sprintf(":%02x", md5pkix[i])
138         }
139         for i := 0; i < len(sha1pkix); i++ {
140                 sha1fp += fmt.Sprintf(":%02x", sha1pkix[i])
141         }
142         return md5fp[1:], sha1fp[1:], nil
143 }
144
145 func (instanceSet *ec2InstanceSet) Create(
146         instanceType arvados.InstanceType,
147         imageID cloud.ImageID,
148         newTags cloud.InstanceTags,
149         initCommand cloud.InitCommand,
150         publicKey ssh.PublicKey) (cloud.Instance, error) {
151
152         md5keyFingerprint, sha1keyFingerprint, err := awsKeyFingerprint(publicKey)
153         if err != nil {
154                 return nil, fmt.Errorf("Could not make key fingerprint: %v", err)
155         }
156         instanceSet.keysMtx.Lock()
157         var keyname string
158         var ok bool
159         if keyname, ok = instanceSet.keys[md5keyFingerprint]; !ok {
160                 keyout, err := instanceSet.client.DescribeKeyPairs(&ec2.DescribeKeyPairsInput{
161                         Filters: []*ec2.Filter{{
162                                 Name:   aws.String("fingerprint"),
163                                 Values: []*string{&md5keyFingerprint, &sha1keyFingerprint},
164                         }},
165                 })
166                 if err != nil {
167                         return nil, fmt.Errorf("Could not search for keypair: %v", err)
168                 }
169
170                 if len(keyout.KeyPairs) > 0 {
171                         keyname = *(keyout.KeyPairs[0].KeyName)
172                 } else {
173                         keyname = "arvados-dispatch-keypair-" + md5keyFingerprint
174                         _, err := instanceSet.client.ImportKeyPair(&ec2.ImportKeyPairInput{
175                                 KeyName:           &keyname,
176                                 PublicKeyMaterial: ssh.MarshalAuthorizedKey(publicKey),
177                         })
178                         if err != nil {
179                                 return nil, fmt.Errorf("Could not import keypair: %v", err)
180                         }
181                 }
182                 instanceSet.keys[md5keyFingerprint] = keyname
183         }
184         instanceSet.keysMtx.Unlock()
185
186         ec2tags := []*ec2.Tag{}
187         for k, v := range newTags {
188                 ec2tags = append(ec2tags, &ec2.Tag{
189                         Key:   aws.String(k),
190                         Value: aws.String(v),
191                 })
192         }
193
194         var groups []string
195         for sg := range instanceSet.ec2config.SecurityGroupIDs {
196                 groups = append(groups, sg)
197         }
198
199         rii := ec2.RunInstancesInput{
200                 ImageId:      aws.String(string(imageID)),
201                 InstanceType: &instanceType.ProviderType,
202                 MaxCount:     aws.Int64(1),
203                 MinCount:     aws.Int64(1),
204                 KeyName:      &keyname,
205
206                 NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{
207                         {
208                                 AssociatePublicIpAddress: aws.Bool(false),
209                                 DeleteOnTermination:      aws.Bool(true),
210                                 DeviceIndex:              aws.Int64(0),
211                                 Groups:                   aws.StringSlice(groups),
212                                 SubnetId:                 &instanceSet.ec2config.SubnetID,
213                         }},
214                 DisableApiTermination:             aws.Bool(false),
215                 InstanceInitiatedShutdownBehavior: aws.String("terminate"),
216                 TagSpecifications: []*ec2.TagSpecification{
217                         {
218                                 ResourceType: aws.String("instance"),
219                                 Tags:         ec2tags,
220                         }},
221                 UserData: aws.String(base64.StdEncoding.EncodeToString([]byte("#!/bin/sh\n" + initCommand + "\n"))),
222         }
223
224         if instanceType.AddedScratch > 0 {
225                 rii.BlockDeviceMappings = []*ec2.BlockDeviceMapping{{
226                         DeviceName: aws.String("/dev/xvdt"),
227                         Ebs: &ec2.EbsBlockDevice{
228                                 DeleteOnTermination: aws.Bool(true),
229                                 VolumeSize:          aws.Int64((int64(instanceType.AddedScratch) + (1<<30 - 1)) >> 30),
230                                 VolumeType:          &instanceSet.ec2config.EBSVolumeType,
231                         }}}
232         }
233
234         if instanceType.Preemptible {
235                 rii.InstanceMarketOptions = &ec2.InstanceMarketOptionsRequest{
236                         MarketType: aws.String("spot"),
237                         SpotOptions: &ec2.SpotMarketOptions{
238                                 InstanceInterruptionBehavior: aws.String("terminate"),
239                                 MaxPrice:                     aws.String(fmt.Sprintf("%v", instanceType.Price)),
240                         }}
241         }
242
243         if instanceSet.ec2config.IAMInstanceProfile != "" {
244                 rii.IamInstanceProfile = &ec2.IamInstanceProfileSpecification{
245                         Name: aws.String(instanceSet.ec2config.IAMInstanceProfile),
246                 }
247         }
248
249         rsv, err := instanceSet.client.RunInstances(&rii)
250         err = wrapError(err, &instanceSet.throttleDelayCreate)
251         if err != nil {
252                 return nil, err
253         }
254         return &ec2Instance{
255                 provider: instanceSet,
256                 instance: rsv.Instances[0],
257         }, nil
258 }
259
260 func (instanceSet *ec2InstanceSet) Instances(tags cloud.InstanceTags) (instances []cloud.Instance, err error) {
261         var filters []*ec2.Filter
262         for k, v := range tags {
263                 filters = append(filters, &ec2.Filter{
264                         Name:   aws.String("tag:" + k),
265                         Values: []*string{aws.String(v)},
266                 })
267         }
268         needAZs := false
269         dii := &ec2.DescribeInstancesInput{Filters: filters}
270         for {
271                 dio, err := instanceSet.client.DescribeInstances(dii)
272                 err = wrapError(err, &instanceSet.throttleDelayInstances)
273                 if err != nil {
274                         return nil, err
275                 }
276
277                 for _, rsv := range dio.Reservations {
278                         for _, inst := range rsv.Instances {
279                                 if *inst.State.Name != "shutting-down" && *inst.State.Name != "terminated" {
280                                         instances = append(instances, &ec2Instance{
281                                                 provider: instanceSet,
282                                                 instance: inst,
283                                         })
284                                         if aws.StringValue(inst.InstanceLifecycle) == "spot" {
285                                                 needAZs = true
286                                         }
287                                 }
288                         }
289                 }
290                 if dio.NextToken == nil {
291                         break
292                 }
293                 dii.NextToken = dio.NextToken
294         }
295         if needAZs && instanceSet.ec2config.SpotPriceUpdateInterval > 0 {
296                 az := map[string]string{}
297                 err := instanceSet.client.DescribeInstanceStatusPages(&ec2.DescribeInstanceStatusInput{
298                         IncludeAllInstances: aws.Bool(true),
299                 }, func(page *ec2.DescribeInstanceStatusOutput, lastPage bool) bool {
300                         for _, ent := range page.InstanceStatuses {
301                                 az[*ent.InstanceId] = *ent.AvailabilityZone
302                         }
303                         return true
304                 })
305                 if err != nil {
306                         instanceSet.logger.Warnf("error getting instance statuses: %s", err)
307                 }
308                 for _, inst := range instances {
309                         inst := inst.(*ec2Instance)
310                         inst.availabilityZone = az[*inst.instance.InstanceId]
311                 }
312                 instanceSet.updateSpotPrices(instances)
313         }
314         return instances, err
315 }
316
317 type priceKey struct {
318         instanceType     string
319         spot             bool
320         availabilityZone string
321 }
322
323 // Refresh recent spot instance pricing data for the given instances,
324 // unless we already have recent pricing data for all relevant types.
325 func (instanceSet *ec2InstanceSet) updateSpotPrices(instances []cloud.Instance) {
326         if len(instances) == 0 {
327                 return
328         }
329
330         instanceSet.pricesLock.Lock()
331         defer instanceSet.pricesLock.Unlock()
332         if instanceSet.prices == nil {
333                 instanceSet.prices = map[priceKey][]cloud.InstancePrice{}
334                 instanceSet.pricesUpdated = map[priceKey]time.Time{}
335         }
336
337         updateTime := time.Now()
338         staleTime := updateTime.Add(-instanceSet.ec2config.SpotPriceUpdateInterval.Duration())
339         needUpdate := false
340         var typeFilterValues []*string
341         for _, inst := range instances {
342                 ec2inst := inst.(*ec2Instance).instance
343                 if aws.StringValue(ec2inst.InstanceLifecycle) == "spot" {
344                         pk := priceKey{
345                                 instanceType:     *ec2inst.InstanceType,
346                                 spot:             true,
347                                 availabilityZone: inst.(*ec2Instance).availabilityZone,
348                         }
349                         if instanceSet.pricesUpdated[pk].Before(staleTime) {
350                                 needUpdate = true
351                         }
352                         typeFilterValues = append(typeFilterValues, ec2inst.InstanceType)
353                 }
354         }
355         if !needUpdate {
356                 return
357         }
358         // Get 3x update interval worth of pricing data. (Ideally the
359         // AWS API would tell us "we have shown you all of the price
360         // changes up to time T", but it doesn't, so we'll just ask
361         // for 3 intervals worth of data on each update, de-duplicate
362         // the data points, and not worry too much about occasionally
363         // missing some data points when our lookups fail twice in a
364         // row.
365         dsphi := &ec2.DescribeSpotPriceHistoryInput{
366                 StartTime: aws.Time(updateTime.Add(-3 * instanceSet.ec2config.SpotPriceUpdateInterval.Duration())),
367                 Filters: []*ec2.Filter{
368                         &ec2.Filter{Name: aws.String("instance-type"), Values: typeFilterValues},
369                         &ec2.Filter{Name: aws.String("product-description"), Values: []*string{aws.String("Linux/UNIX")}},
370                 },
371         }
372         err := instanceSet.client.DescribeSpotPriceHistoryPages(dsphi, func(page *ec2.DescribeSpotPriceHistoryOutput, lastPage bool) bool {
373                 for _, ent := range page.SpotPriceHistory {
374                         if ent.InstanceType == nil || ent.SpotPrice == nil || ent.Timestamp == nil {
375                                 // bogus record?
376                                 continue
377                         }
378                         price, err := strconv.ParseFloat(*ent.SpotPrice, 64)
379                         if err != nil {
380                                 // bogus record?
381                                 continue
382                         }
383                         pk := priceKey{
384                                 instanceType:     *ent.InstanceType,
385                                 spot:             true,
386                                 availabilityZone: *ent.AvailabilityZone,
387                         }
388                         instanceSet.prices[pk] = append(instanceSet.prices[pk], cloud.InstancePrice{
389                                 StartTime: *ent.Timestamp,
390                                 Price:     price,
391                         })
392                         instanceSet.pricesUpdated[pk] = updateTime
393                 }
394                 return true
395         })
396         if err != nil {
397                 instanceSet.logger.Warnf("error retrieving spot instance prices: %s", err)
398         }
399
400         expiredTime := updateTime.Add(-64 * instanceSet.ec2config.SpotPriceUpdateInterval.Duration())
401         for pk, last := range instanceSet.pricesUpdated {
402                 if last.Before(expiredTime) {
403                         delete(instanceSet.pricesUpdated, pk)
404                         delete(instanceSet.prices, pk)
405                 }
406         }
407         for pk, prices := range instanceSet.prices {
408                 instanceSet.prices[pk] = cloud.NormalizePriceHistory(prices)
409         }
410 }
411
412 func (instanceSet *ec2InstanceSet) Stop() {
413 }
414
415 type ec2Instance struct {
416         provider         *ec2InstanceSet
417         instance         *ec2.Instance
418         availabilityZone string // sometimes available for spot instances
419 }
420
421 func (inst *ec2Instance) ID() cloud.InstanceID {
422         return cloud.InstanceID(*inst.instance.InstanceId)
423 }
424
425 func (inst *ec2Instance) String() string {
426         return *inst.instance.InstanceId
427 }
428
429 func (inst *ec2Instance) ProviderType() string {
430         return *inst.instance.InstanceType
431 }
432
433 func (inst *ec2Instance) SetTags(newTags cloud.InstanceTags) error {
434         var ec2tags []*ec2.Tag
435         for k, v := range newTags {
436                 ec2tags = append(ec2tags, &ec2.Tag{
437                         Key:   aws.String(k),
438                         Value: aws.String(v),
439                 })
440         }
441
442         _, err := inst.provider.client.CreateTags(&ec2.CreateTagsInput{
443                 Resources: []*string{inst.instance.InstanceId},
444                 Tags:      ec2tags,
445         })
446
447         return err
448 }
449
450 func (inst *ec2Instance) Tags() cloud.InstanceTags {
451         tags := make(map[string]string)
452
453         for _, t := range inst.instance.Tags {
454                 tags[*t.Key] = *t.Value
455         }
456
457         return tags
458 }
459
460 func (inst *ec2Instance) Destroy() error {
461         _, err := inst.provider.client.TerminateInstances(&ec2.TerminateInstancesInput{
462                 InstanceIds: []*string{inst.instance.InstanceId},
463         })
464         return err
465 }
466
467 func (inst *ec2Instance) Address() string {
468         if inst.instance.PrivateIpAddress != nil {
469                 return *inst.instance.PrivateIpAddress
470         }
471         return ""
472 }
473
474 func (inst *ec2Instance) RemoteUser() string {
475         return inst.provider.ec2config.AdminUsername
476 }
477
478 func (inst *ec2Instance) VerifyHostKey(ssh.PublicKey, *ssh.Client) error {
479         return cloud.ErrNotImplemented
480 }
481
482 // PriceHistory returns the price history for this specific instance.
483 //
484 // AWS documentation is elusive about whether the hourly cost of a
485 // given spot instance changes as the current spot price changes for
486 // the corresponding instance type and availability zone. Our
487 // implementation assumes the answer is yes, based on the following
488 // hints.
489 //
490 // https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/spot-requests.html
491 // says: "After your Spot Instance is running, if the Spot price rises
492 // above your maximum price, Amazon EC2 interrupts your Spot
493 // Instance." (This doesn't address what happens when the spot price
494 // rises *without* exceeding your maximum price.)
495 //
496 // https://docs.aws.amazon.com/whitepapers/latest/cost-optimization-leveraging-ec2-spot-instances/how-spot-instances-work.html
497 // says: "You pay the Spot price that's in effect, billed to the
498 // nearest second." (But it's not explicitly stated whether "the price
499 // in effect" changes over time for a given instance.)
500 //
501 // The same page also says, in a discussion about the effect of
502 // specifying a maximum price: "Note that you never pay more than the
503 // Spot price that is in effect when your Spot Instance is running."
504 // (The use of the phrase "is running", as opposed to "was launched",
505 // hints that pricing is dynamic.)
506 func (inst *ec2Instance) PriceHistory(instType arvados.InstanceType) []cloud.InstancePrice {
507         inst.provider.pricesLock.Lock()
508         defer inst.provider.pricesLock.Unlock()
509         // Note updateSpotPrices currently populates
510         // inst.provider.prices only for spot instances, so if
511         // spot==false here, we will return no data.
512         pk := priceKey{
513                 instanceType:     *inst.instance.InstanceType,
514                 spot:             aws.StringValue(inst.instance.InstanceLifecycle) == "spot",
515                 availabilityZone: inst.availabilityZone,
516         }
517         var prices []cloud.InstancePrice
518         for _, price := range inst.provider.prices[pk] {
519                 // ceil(added scratch space in GiB)
520                 gib := (instType.AddedScratch + 1<<30 - 1) >> 30
521                 monthly := inst.provider.ec2config.EBSPrice * float64(gib)
522                 hourly := monthly / 30 / 24
523                 price.Price += hourly
524                 prices = append(prices, price)
525         }
526         return prices
527 }
528
529 type rateLimitError struct {
530         error
531         earliestRetry time.Time
532 }
533
534 func (err rateLimitError) EarliestRetry() time.Time {
535         return err.earliestRetry
536 }
537
538 var isCodeCapacity = map[string]bool{
539         "InsufficientInstanceCapacity": true,
540         "VcpuLimitExceeded":            true,
541         "MaxSpotInstanceCountExceeded": true,
542 }
543
544 // isErrorCapacity returns whether the error is to be throttled based on its code.
545 // Returns false if error is nil.
546 func isErrorCapacity(err error) bool {
547         if aerr, ok := err.(awserr.Error); ok && aerr != nil {
548                 if _, ok := isCodeCapacity[aerr.Code()]; ok {
549                         return true
550                 }
551         }
552         return false
553 }
554
555 type ec2QuotaError struct {
556         error
557 }
558
559 func (er *ec2QuotaError) IsQuotaError() bool {
560         return true
561 }
562
563 func wrapError(err error, throttleValue *atomic.Value) error {
564         if request.IsErrorThrottle(err) {
565                 // Back off exponentially until an upstream call
566                 // either succeeds or returns a non-throttle error.
567                 d, _ := throttleValue.Load().(time.Duration)
568                 d = d*3/2 + time.Second
569                 if d < throttleDelayMin {
570                         d = throttleDelayMin
571                 } else if d > throttleDelayMax {
572                         d = throttleDelayMax
573                 }
574                 throttleValue.Store(d)
575                 return rateLimitError{error: err, earliestRetry: time.Now().Add(d)}
576         } else if isErrorCapacity(err) {
577                 return &ec2QuotaError{err}
578         } else if err != nil {
579                 throttleValue.Store(time.Duration(0))
580                 return err
581         }
582         throttleValue.Store(time.Duration(0))
583         return nil
584 }