awsprovider.go 63 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100
  1. package cloud
  2. import (
  3. "bytes"
  4. "compress/gzip"
  5. "context"
  6. "encoding/csv"
  7. "fmt"
  8. "io"
  9. "io/ioutil"
  10. "net/http"
  11. "regexp"
  12. "strconv"
  13. "strings"
  14. "sync"
  15. "time"
  16. "k8s.io/klog"
  17. "github.com/kubecost/cost-model/pkg/clustercache"
  18. "github.com/kubecost/cost-model/pkg/env"
  19. "github.com/kubecost/cost-model/pkg/errors"
  20. "github.com/kubecost/cost-model/pkg/log"
  21. "github.com/kubecost/cost-model/pkg/util"
  22. "github.com/kubecost/cost-model/pkg/util/fileutil"
  23. "github.com/kubecost/cost-model/pkg/util/json"
  24. awsSDK "github.com/aws/aws-sdk-go-v2/aws"
  25. "github.com/aws/aws-sdk-go-v2/config"
  26. "github.com/aws/aws-sdk-go-v2/credentials/stscreds"
  27. "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
  28. "github.com/aws/aws-sdk-go-v2/service/athena"
  29. athenaTypes "github.com/aws/aws-sdk-go-v2/service/athena/types"
  30. "github.com/aws/aws-sdk-go-v2/service/ec2"
  31. ec2Types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
  32. "github.com/aws/aws-sdk-go-v2/service/s3"
  33. "github.com/aws/aws-sdk-go-v2/service/sts"
  34. "github.com/jszwec/csvutil"
  35. v1 "k8s.io/api/core/v1"
  36. )
  37. const supportedSpotFeedVersion = "1"
  38. const SpotInfoUpdateType = "spotinfo"
  39. const AthenaInfoUpdateType = "athenainfo"
  40. const PreemptibleType = "preemptible"
  41. const APIPricingSource = "Public API"
  42. const SpotPricingSource = "Spot Data Feed"
  43. const ReservedInstancePricingSource = "Savings Plan, Reserved Instance, and Out-Of-Cluster"
  44. func (aws *AWS) PricingSourceStatus() map[string]*PricingSource {
  45. sources := make(map[string]*PricingSource)
  46. sps := &PricingSource{
  47. Name: SpotPricingSource,
  48. Enabled: true,
  49. }
  50. if !aws.SpotRefreshEnabled() {
  51. sps.Available = false
  52. sps.Error = "Spot instances not set up"
  53. sps.Enabled = false
  54. } else {
  55. sps.Error = ""
  56. if aws.SpotPricingError != nil {
  57. sps.Error = aws.SpotPricingError.Error()
  58. }
  59. if sps.Error != "" {
  60. sps.Available = false
  61. } else if len(aws.SpotPricingByInstanceID) > 0 {
  62. sps.Available = true
  63. } else {
  64. sps.Error = "No spot instances detected"
  65. }
  66. }
  67. sources[SpotPricingSource] = sps
  68. rps := &PricingSource{
  69. Name: ReservedInstancePricingSource,
  70. Enabled: true,
  71. }
  72. rps.Error = ""
  73. if aws.RIPricingError != nil {
  74. rps.Error = aws.RIPricingError.Error()
  75. }
  76. if rps.Error != "" {
  77. rps.Available = false
  78. } else {
  79. rps.Available = true
  80. }
  81. sources[ReservedInstancePricingSource] = rps
  82. return sources
  83. }
  84. // How often spot data is refreshed
  85. const SpotRefreshDuration = 15 * time.Minute
  86. const defaultConfigPath = "/var/configs/"
  87. var awsRegions = []string{
  88. "us-east-2",
  89. "us-east-1",
  90. "us-west-1",
  91. "us-west-2",
  92. "ap-east-1",
  93. "ap-south-1",
  94. "ap-northeast-3",
  95. "ap-northeast-2",
  96. "ap-southeast-1",
  97. "ap-southeast-2",
  98. "ap-northeast-1",
  99. "ap-southeast-3",
  100. "ca-central-1",
  101. "cn-north-1",
  102. "cn-northwest-1",
  103. "eu-central-1",
  104. "eu-west-1",
  105. "eu-west-2",
  106. "eu-west-3",
  107. "eu-north-1",
  108. "eu-south-1",
  109. "me-south-1",
  110. "sa-east-1",
  111. "af-south-1",
  112. "us-gov-east-1",
  113. "us-gov-west-1",
  114. }
  115. // AWS represents an Amazon Provider
  116. type AWS struct {
  117. Pricing map[string]*AWSProductTerms
  118. SpotPricingByInstanceID map[string]*spotInfo
  119. SpotPricingUpdatedAt *time.Time
  120. SpotRefreshRunning bool
  121. SpotPricingLock sync.RWMutex
  122. SpotPricingError error
  123. RIPricingByInstanceID map[string]*RIData
  124. RIPricingError error
  125. RIDataRunning bool
  126. RIDataLock sync.RWMutex
  127. SavingsPlanDataByInstanceID map[string]*SavingsPlanData
  128. SavingsPlanDataRunning bool
  129. SavingsPlanDataLock sync.RWMutex
  130. ValidPricingKeys map[string]bool
  131. Clientset clustercache.ClusterCache
  132. BaseCPUPrice string
  133. BaseRAMPrice string
  134. BaseGPUPrice string
  135. BaseSpotCPUPrice string
  136. BaseSpotRAMPrice string
  137. BaseSpotGPUPrice string
  138. SpotLabelName string
  139. SpotLabelValue string
  140. SpotDataRegion string
  141. SpotDataBucket string
  142. SpotDataPrefix string
  143. ProjectID string
  144. DownloadPricingDataLock sync.RWMutex
  145. Config *ProviderConfig
  146. serviceAccountChecks *ServiceAccountChecks
  147. clusterManagementPrice float64
  148. clusterAccountId string
  149. clusterRegion string
  150. clusterProvisioner string
  151. *CustomProvider
  152. }
  153. // AWSAccessKey holds AWS credentials and fulfils the awsV2.CredentialsProvider interface
  154. type AWSAccessKey struct {
  155. AccessKeyID string `json:"aws_access_key_id"`
  156. SecretAccessKey string `json:"aws_secret_access_key"`
  157. }
  158. // Retrieve returns a set of awsV2 credentials using the AWSAccessKey's key and secret.
  159. // This fulfils the awsV2.CredentialsProvider interface contract.
  160. func (accessKey AWSAccessKey) Retrieve(ctx context.Context) (awsSDK.Credentials, error) {
  161. return awsSDK.Credentials{
  162. AccessKeyID: accessKey.AccessKeyID,
  163. SecretAccessKey: accessKey.SecretAccessKey,
  164. }, nil
  165. }
  166. // CreateConfig creates an AWS SDK V2 Config for the credentials that it contains for the provided region
  167. func (accessKey AWSAccessKey) CreateConfig(region string) (awsSDK.Config, error) {
  168. var cfg awsSDK.Config
  169. var err error
  170. // If accessKey values have not been provided, attempt to load cfg from service key annotations
  171. if accessKey.AccessKeyID == "" && accessKey.SecretAccessKey == "" {
  172. cfg, err = config.LoadDefaultConfig(context.TODO(), config.WithRegion(region))
  173. if err != nil {
  174. return cfg, fmt.Errorf("failed to initialize AWS SDK config for region from annotation %s: %s", region, err)
  175. }
  176. } else {
  177. // The AWS SDK v2 requires an object fulfilling the CredentialsProvider interface, which cloud.AWSAccessKey does
  178. cfg, err = config.LoadDefaultConfig(context.TODO(), config.WithCredentialsProvider(accessKey), config.WithRegion(region))
  179. if err != nil {
  180. return cfg, fmt.Errorf("failed to initialize AWS SDK config for region %s: %s", region, err)
  181. }
  182. }
  183. return cfg, nil
  184. }
  185. // AWSPricing maps a k8s node to an AWS Pricing "product"
  186. type AWSPricing struct {
  187. Products map[string]*AWSProduct `json:"products"`
  188. Terms AWSPricingTerms `json:"terms"`
  189. }
  190. // AWSProduct represents a purchased SKU
  191. type AWSProduct struct {
  192. Sku string `json:"sku"`
  193. Attributes AWSProductAttributes `json:"attributes"`
  194. }
  195. // AWSProductAttributes represents metadata about the product used to map to a node.
  196. type AWSProductAttributes struct {
  197. Location string `json:"location"`
  198. InstanceType string `json:"instanceType"`
  199. Memory string `json:"memory"`
  200. Storage string `json:"storage"`
  201. VCpu string `json:"vcpu"`
  202. UsageType string `json:"usagetype"`
  203. OperatingSystem string `json:"operatingSystem"`
  204. PreInstalledSw string `json:"preInstalledSw"`
  205. InstanceFamily string `json:"instanceFamily"`
  206. CapacityStatus string `json:"capacitystatus"`
  207. GPU string `json:"gpu"` // GPU represents the number of GPU on the instance
  208. }
  209. // AWSPricingTerms are how you pay for the node: OnDemand, Reserved, or (TODO) Spot
  210. type AWSPricingTerms struct {
  211. OnDemand map[string]map[string]*AWSOfferTerm `json:"OnDemand"`
  212. Reserved map[string]map[string]*AWSOfferTerm `json:"Reserved"`
  213. }
  214. // AWSOfferTerm is a sku extension used to pay for the node.
  215. type AWSOfferTerm struct {
  216. Sku string `json:"sku"`
  217. PriceDimensions map[string]*AWSRateCode `json:"priceDimensions"`
  218. }
  219. func (ot *AWSOfferTerm) String() string {
  220. var strs []string
  221. for k, rc := range ot.PriceDimensions {
  222. strs = append(strs, fmt.Sprintf("%s:%s", k, rc.String()))
  223. }
  224. return fmt.Sprintf("%s:%s", ot.Sku, strings.Join(strs, ","))
  225. }
  226. // AWSRateCode encodes data about the price of a product
  227. type AWSRateCode struct {
  228. Unit string `json:"unit"`
  229. PricePerUnit AWSCurrencyCode `json:"pricePerUnit"`
  230. }
  231. func (rc *AWSRateCode) String() string {
  232. return fmt.Sprintf("{unit: %s, pricePerUnit: %v", rc.Unit, rc.PricePerUnit)
  233. }
  234. // AWSCurrencyCode is the localized currency. (TODO: support non-USD)
  235. type AWSCurrencyCode struct {
  236. USD string `json:"USD,omitempty"`
  237. CNY string `json:"CNY,omitempty"`
  238. }
  239. // AWSProductTerms represents the full terms of the product
  240. type AWSProductTerms struct {
  241. Sku string `json:"sku"`
  242. OnDemand *AWSOfferTerm `json:"OnDemand"`
  243. Reserved *AWSOfferTerm `json:"Reserved"`
  244. Memory string `json:"memory"`
  245. Storage string `json:"storage"`
  246. VCpu string `json:"vcpu"`
  247. GPU string `json:"gpu"` // GPU represents the number of GPU on the instance
  248. PV *PV `json:"pv"`
  249. }
  250. // ClusterIdEnvVar is the environment variable in which one can manually set the ClusterId
  251. const ClusterIdEnvVar = "AWS_CLUSTER_ID"
  252. // OnDemandRateCode is appended to an node sku
  253. const OnDemandRateCode = ".JRTCKXETXF"
  254. const OnDemandRateCodeCn = ".99YE2YK9UR"
  255. // ReservedRateCode is appended to a node sku
  256. const ReservedRateCode = ".38NPMPTW36"
  257. // HourlyRateCode is appended to a node sku
  258. const HourlyRateCode = ".6YS6EN2CT7"
  259. const HourlyRateCodeCn = ".Q7UJUT2CE6"
  260. // volTypes are used to map between AWS UsageTypes and
  261. // EBS volume types, as they would appear in K8s storage class
  262. // name and the EC2 API.
  263. var volTypes = map[string]string{
  264. "EBS:VolumeUsage.gp2": "gp2",
  265. "EBS:VolumeUsage": "standard",
  266. "EBS:VolumeUsage.sc1": "sc1",
  267. "EBS:VolumeP-IOPS.piops": "io1",
  268. "EBS:VolumeUsage.st1": "st1",
  269. "EBS:VolumeUsage.piops": "io1",
  270. "gp2": "EBS:VolumeUsage.gp2",
  271. "standard": "EBS:VolumeUsage",
  272. "sc1": "EBS:VolumeUsage.sc1",
  273. "io1": "EBS:VolumeUsage.piops",
  274. "st1": "EBS:VolumeUsage.st1",
  275. }
  276. // locationToRegion maps AWS region names (As they come from Billing)
  277. // to actual region identifiers
  278. var locationToRegion = map[string]string{
  279. "US East (Ohio)": "us-east-2",
  280. "US East (N. Virginia)": "us-east-1",
  281. "US West (N. California)": "us-west-1",
  282. "US West (Oregon)": "us-west-2",
  283. "Asia Pacific (Hong Kong)": "ap-east-1",
  284. "Asia Pacific (Mumbai)": "ap-south-1",
  285. "Asia Pacific (Osaka)": "ap-northeast-3",
  286. "Asia Pacific (Seoul)": "ap-northeast-2",
  287. "Asia Pacific (Singapore)": "ap-southeast-1",
  288. "Asia Pacific (Sydney)": "ap-southeast-2",
  289. "Asia Pacific (Tokyo)": "ap-northeast-1",
  290. "Asia Pacific (Jakarta)": "ap-southeast-3",
  291. "Canada (Central)": "ca-central-1",
  292. "China (Beijing)": "cn-north-1",
  293. "China (Ningxia)": "cn-northwest-1",
  294. "EU (Frankfurt)": "eu-central-1",
  295. "EU (Ireland)": "eu-west-1",
  296. "EU (London)": "eu-west-2",
  297. "EU (Paris)": "eu-west-3",
  298. "EU (Stockholm)": "eu-north-1",
  299. "EU (Milan)": "eu-south-1",
  300. "South America (Sao Paulo)": "sa-east-1",
  301. "Africa (Cape Town)": "af-south-1",
  302. "AWS GovCloud (US-East)": "us-gov-east-1",
  303. "AWS GovCloud (US-West)": "us-gov-west-1",
  304. }
  305. var regionToBillingRegionCode = map[string]string{
  306. "us-east-2": "USE2",
  307. "us-east-1": "",
  308. "us-west-1": "USW1",
  309. "us-west-2": "USW2",
  310. "ap-east-1": "APE1",
  311. "ap-south-1": "APS3",
  312. "ap-northeast-3": "APN3",
  313. "ap-northeast-2": "APN2",
  314. "ap-southeast-1": "APS1",
  315. "ap-southeast-2": "APS2",
  316. "ap-northeast-1": "APN1",
  317. "ap-southeast-3": "APS4",
  318. "ca-central-1": "CAN1",
  319. "cn-north-1": "",
  320. "cn-northwest-1": "",
  321. "eu-central-1": "EUC1",
  322. "eu-west-1": "EU",
  323. "eu-west-2": "EUW2",
  324. "eu-west-3": "EUW3",
  325. "eu-north-1": "EUN1",
  326. "eu-south-1": "EUS1",
  327. "sa-east-1": "SAE1",
  328. "af-south-1": "AFS1",
  329. "us-gov-east-1": "UGE1",
  330. "us-gov-west-1": "UGW1",
  331. }
  332. var loadedAWSSecret bool = false
  333. var awsSecret *AWSAccessKey = nil
  334. func (aws *AWS) GetLocalStorageQuery(window, offset time.Duration, rate bool, used bool) string {
  335. return ""
  336. }
  337. // KubeAttrConversion maps the k8s labels for region to an aws region
  338. func (aws *AWS) KubeAttrConversion(location, instanceType, operatingSystem string) string {
  339. operatingSystem = strings.ToLower(operatingSystem)
  340. region := locationToRegion[location]
  341. return region + "," + instanceType + "," + operatingSystem
  342. }
  343. // AwsSpotFeedInfo contains configuration for spot feed integration
  344. type AwsSpotFeedInfo struct {
  345. BucketName string `json:"bucketName"`
  346. Prefix string `json:"prefix"`
  347. Region string `json:"region"`
  348. AccountID string `json:"projectID"`
  349. ServiceKeyName string `json:"serviceKeyName"`
  350. ServiceKeySecret string `json:"serviceKeySecret"`
  351. SpotLabel string `json:"spotLabel"`
  352. SpotLabelValue string `json:"spotLabelValue"`
  353. }
  354. // AwsAthenaInfo contains configuration for CUR integration
  355. type AwsAthenaInfo struct {
  356. AthenaBucketName string `json:"athenaBucketName"`
  357. AthenaRegion string `json:"athenaRegion"`
  358. AthenaDatabase string `json:"athenaDatabase"`
  359. AthenaTable string `json:"athenaTable"`
  360. ServiceKeyName string `json:"serviceKeyName"`
  361. ServiceKeySecret string `json:"serviceKeySecret"`
  362. AccountID string `json:"projectID"`
  363. MasterPayerARN string `json:"masterPayerARN"`
  364. }
  365. // IsEmpty returns true if all fields in config are empty, false if not.
  366. func (aai *AwsAthenaInfo) IsEmpty() bool {
  367. return aai.AthenaBucketName == "" &&
  368. aai.AthenaRegion == "" &&
  369. aai.AthenaDatabase == "" &&
  370. aai.AthenaTable == "" &&
  371. aai.ServiceKeyName == "" &&
  372. aai.ServiceKeySecret == "" &&
  373. aai.AccountID == "" &&
  374. aai.MasterPayerARN == ""
  375. }
  376. // CreateConfig creates an AWS SDK V2 Config for the credentials that it contains
  377. func (aai *AwsAthenaInfo) CreateConfig() (awsSDK.Config, error) {
  378. keyProvider := AWSAccessKey{AccessKeyID: aai.ServiceKeyName, SecretAccessKey: aai.ServiceKeySecret}
  379. cfg, err := keyProvider.CreateConfig(aai.AthenaRegion)
  380. if err != nil {
  381. return cfg, err
  382. }
  383. if aai.MasterPayerARN != "" {
  384. // Create the credentials from AssumeRoleProvider to assume the role
  385. // referenced by the roleARN.
  386. stsSvc := sts.NewFromConfig(cfg)
  387. creds := stscreds.NewAssumeRoleProvider(stsSvc, aai.MasterPayerARN)
  388. cfg.Credentials = awsSDK.NewCredentialsCache(creds)
  389. }
  390. return cfg, nil
  391. }
  392. func (aws *AWS) GetManagementPlatform() (string, error) {
  393. nodes := aws.Clientset.GetAllNodes()
  394. if len(nodes) > 0 {
  395. n := nodes[0]
  396. version := n.Status.NodeInfo.KubeletVersion
  397. if strings.Contains(version, "eks") {
  398. return "eks", nil
  399. }
  400. if _, ok := n.Labels["kops.k8s.io/instancegroup"]; ok {
  401. return "kops", nil
  402. }
  403. }
  404. return "", nil
  405. }
  406. func (aws *AWS) GetConfig() (*CustomPricing, error) {
  407. c, err := aws.Config.GetCustomPricingData()
  408. if err != nil {
  409. return nil, err
  410. }
  411. if c.Discount == "" {
  412. c.Discount = "0%"
  413. }
  414. if c.NegotiatedDiscount == "" {
  415. c.NegotiatedDiscount = "0%"
  416. }
  417. if c.ShareTenancyCosts == "" {
  418. c.ShareTenancyCosts = defaultShareTenancyCost
  419. }
  420. return c, nil
  421. }
  422. // GetAWSAccessKey generate an AWSAccessKey object from the config
  423. func (aws *AWS) GetAWSAccessKey() (*AWSAccessKey, error) {
  424. config, err := aws.GetConfig()
  425. if err != nil {
  426. return nil, fmt.Errorf("could not retrieve AwsAthenaInfo %s", err)
  427. }
  428. err = aws.ConfigureAuthWith(config)
  429. if err != nil {
  430. return nil, fmt.Errorf("error configuring Cloud Provider %s", err)
  431. }
  432. //Look for service key values in env if not present in config
  433. if config.ServiceKeyName == "" {
  434. config.ServiceKeyName = env.GetAWSAccessKeyID()
  435. }
  436. if config.ServiceKeySecret == "" {
  437. config.ServiceKeySecret = env.GetAWSAccessKeySecret()
  438. }
  439. if config.ServiceKeyName == "" && config.ServiceKeySecret == "" {
  440. log.DedupedInfof(1, "missing service key values for AWS cloud integration attempting to use service account integration")
  441. }
  442. return &AWSAccessKey{AccessKeyID: config.ServiceKeyName, SecretAccessKey: config.ServiceKeySecret}, nil
  443. }
  444. // GetAWSAthenaInfo generate an AWSAthenaInfo object from the config
  445. func (aws *AWS) GetAWSAthenaInfo() (*AwsAthenaInfo, error) {
  446. config, err := aws.GetConfig()
  447. if err != nil {
  448. return nil, fmt.Errorf("could not retrieve AwsAthenaInfo %s", err)
  449. }
  450. aak, err := aws.GetAWSAccessKey()
  451. if err != nil {
  452. return nil, err
  453. }
  454. return &AwsAthenaInfo{
  455. AthenaBucketName: config.AthenaBucketName,
  456. AthenaRegion: config.AthenaRegion,
  457. AthenaDatabase: config.AthenaDatabase,
  458. AthenaTable: config.AthenaTable,
  459. ServiceKeyName: aak.AccessKeyID,
  460. ServiceKeySecret: aak.SecretAccessKey,
  461. AccountID: config.AthenaProjectID,
  462. MasterPayerARN: config.MasterPayerARN,
  463. }, nil
  464. }
  465. func (aws *AWS) UpdateConfigFromConfigMap(cm map[string]string) (*CustomPricing, error) {
  466. return aws.Config.UpdateFromMap(cm)
  467. }
  468. func (aws *AWS) UpdateConfig(r io.Reader, updateType string) (*CustomPricing, error) {
  469. return aws.Config.Update(func(c *CustomPricing) error {
  470. if updateType == SpotInfoUpdateType {
  471. asfi := AwsSpotFeedInfo{}
  472. err := json.NewDecoder(r).Decode(&asfi)
  473. if err != nil {
  474. return err
  475. }
  476. c.ServiceKeyName = asfi.ServiceKeyName
  477. if asfi.ServiceKeySecret != "" {
  478. c.ServiceKeySecret = asfi.ServiceKeySecret
  479. }
  480. c.SpotDataPrefix = asfi.Prefix
  481. c.SpotDataBucket = asfi.BucketName
  482. c.ProjectID = asfi.AccountID
  483. c.SpotDataRegion = asfi.Region
  484. c.SpotLabel = asfi.SpotLabel
  485. c.SpotLabelValue = asfi.SpotLabelValue
  486. } else if updateType == AthenaInfoUpdateType {
  487. aai := AwsAthenaInfo{}
  488. err := json.NewDecoder(r).Decode(&aai)
  489. if err != nil {
  490. return err
  491. }
  492. c.AthenaBucketName = aai.AthenaBucketName
  493. c.AthenaRegion = aai.AthenaRegion
  494. c.AthenaDatabase = aai.AthenaDatabase
  495. c.AthenaTable = aai.AthenaTable
  496. c.ServiceKeyName = aai.ServiceKeyName
  497. if aai.ServiceKeySecret != "" {
  498. c.ServiceKeySecret = aai.ServiceKeySecret
  499. }
  500. if aai.MasterPayerARN != "" {
  501. c.MasterPayerARN = aai.MasterPayerARN
  502. }
  503. c.AthenaProjectID = aai.AccountID
  504. } else {
  505. a := make(map[string]interface{})
  506. err := json.NewDecoder(r).Decode(&a)
  507. if err != nil {
  508. return err
  509. }
  510. for k, v := range a {
  511. kUpper := strings.Title(k) // Just so we consistently supply / receive the same values, uppercase the first letter.
  512. vstr, ok := v.(string)
  513. if ok {
  514. err := SetCustomPricingField(c, kUpper, vstr)
  515. if err != nil {
  516. return err
  517. }
  518. } else {
  519. return fmt.Errorf("type error while updating config for %s", kUpper)
  520. }
  521. }
  522. }
  523. if env.IsRemoteEnabled() {
  524. err := UpdateClusterMeta(env.GetClusterID(), c.ClusterName)
  525. if err != nil {
  526. return err
  527. }
  528. }
  529. return nil
  530. })
  531. }
  532. type awsKey struct {
  533. SpotLabelName string
  534. SpotLabelValue string
  535. Labels map[string]string
  536. ProviderID string
  537. }
  538. func (k *awsKey) GPUType() string {
  539. return ""
  540. }
  541. func (k *awsKey) ID() string {
  542. provIdRx := regexp.MustCompile("aws:///([^/]+)/([^/]+)") // It's of the form aws:///us-east-2a/i-0fea4fd46592d050b and we want i-0fea4fd46592d050b, if it exists
  543. for matchNum, group := range provIdRx.FindStringSubmatch(k.ProviderID) {
  544. if matchNum == 2 {
  545. return group
  546. }
  547. }
  548. klog.V(3).Infof("Could not find instance ID in \"%s\"", k.ProviderID)
  549. return ""
  550. }
  551. func (k *awsKey) Features() string {
  552. instanceType, _ := util.GetInstanceType(k.Labels)
  553. operatingSystem, _ := util.GetOperatingSystem(k.Labels)
  554. region, _ := util.GetRegion(k.Labels)
  555. key := region + "," + instanceType + "," + operatingSystem
  556. usageType := PreemptibleType
  557. spotKey := key + "," + usageType
  558. if l, ok := k.Labels["lifecycle"]; ok && l == "EC2Spot" {
  559. return spotKey
  560. }
  561. if l, ok := k.Labels[k.SpotLabelName]; ok && l == k.SpotLabelValue {
  562. return spotKey
  563. }
  564. return key
  565. }
  566. func (aws *AWS) PVPricing(pvk PVKey) (*PV, error) {
  567. pricing, ok := aws.Pricing[pvk.Features()]
  568. if !ok {
  569. klog.V(4).Infof("Persistent Volume pricing not found for %s: %s", pvk.GetStorageClass(), pvk.Features())
  570. return &PV{}, nil
  571. }
  572. return pricing.PV, nil
  573. }
  574. type awsPVKey struct {
  575. Labels map[string]string
  576. StorageClassParameters map[string]string
  577. StorageClassName string
  578. Name string
  579. DefaultRegion string
  580. ProviderID string
  581. }
  582. func (aws *AWS) GetPVKey(pv *v1.PersistentVolume, parameters map[string]string, defaultRegion string) PVKey {
  583. providerID := ""
  584. if pv.Spec.AWSElasticBlockStore != nil {
  585. providerID = pv.Spec.AWSElasticBlockStore.VolumeID
  586. } else if pv.Spec.CSI != nil {
  587. providerID = pv.Spec.CSI.VolumeHandle
  588. }
  589. return &awsPVKey{
  590. Labels: pv.Labels,
  591. StorageClassName: pv.Spec.StorageClassName,
  592. StorageClassParameters: parameters,
  593. Name: pv.Name,
  594. DefaultRegion: defaultRegion,
  595. ProviderID: providerID,
  596. }
  597. }
  598. func (key *awsPVKey) ID() string {
  599. return key.ProviderID
  600. }
  601. func (key *awsPVKey) GetStorageClass() string {
  602. return key.StorageClassName
  603. }
  604. func (key *awsPVKey) Features() string {
  605. storageClass := key.StorageClassParameters["type"]
  606. if storageClass == "standard" {
  607. storageClass = "gp2"
  608. }
  609. // Storage class names are generally EBS volume types (gp2)
  610. // Keys in Pricing are based on UsageTypes (EBS:VolumeType.gp2)
  611. // Converts between the 2
  612. region, ok := util.GetRegion(key.Labels)
  613. if !ok {
  614. region = key.DefaultRegion
  615. }
  616. class, ok := volTypes[storageClass]
  617. if !ok {
  618. klog.V(4).Infof("No voltype mapping for %s's storageClass: %s", key.Name, storageClass)
  619. }
  620. return region + "," + class
  621. }
  622. // GetKey maps node labels to information needed to retrieve pricing data
  623. func (aws *AWS) GetKey(labels map[string]string, n *v1.Node) Key {
  624. return &awsKey{
  625. SpotLabelName: aws.SpotLabelName,
  626. SpotLabelValue: aws.SpotLabelValue,
  627. Labels: labels,
  628. ProviderID: labels["providerID"],
  629. }
  630. }
  631. func (aws *AWS) isPreemptible(key string) bool {
  632. s := strings.Split(key, ",")
  633. if len(s) == 4 && s[3] == PreemptibleType {
  634. return true
  635. }
  636. return false
  637. }
  638. func (aws *AWS) ClusterManagementPricing() (string, float64, error) {
  639. return aws.clusterProvisioner, aws.clusterManagementPrice, nil
  640. }
  641. // Use the pricing data from the current region. Fall back to using all region data if needed.
  642. func (aws *AWS) getRegionPricing(nodeList []*v1.Node) (*http.Response, string, error) {
  643. pricingURL := "https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonEC2/current/"
  644. region := ""
  645. multiregion := false
  646. for _, n := range nodeList {
  647. labels := n.GetLabels()
  648. currentNodeRegion := ""
  649. if r, ok := util.GetRegion(labels); ok {
  650. currentNodeRegion = r
  651. // Switch to Chinese endpoint for regions with the Chinese prefix
  652. if strings.HasPrefix(currentNodeRegion, "cn-") {
  653. pricingURL = "https://pricing.cn-north-1.amazonaws.com.cn/offers/v1.0/cn/AmazonEC2/current/"
  654. }
  655. } else {
  656. multiregion = true // We weren't able to detect the node's region, so pull all data.
  657. break
  658. }
  659. if region == "" { // We haven't set a region yet
  660. region = currentNodeRegion
  661. } else if region != "" && currentNodeRegion != region { // If two nodes have different regions here, we'll need to fetch all pricing data.
  662. multiregion = true
  663. break
  664. }
  665. }
  666. // Chinese multiregion endpoint only contains data for Chinese regions and Chinese regions are excluded from other endpoint
  667. if region != "" && !multiregion {
  668. pricingURL += region + "/"
  669. }
  670. pricingURL += "index.json"
  671. klog.V(2).Infof("starting download of \"%s\", which is quite large ...", pricingURL)
  672. resp, err := http.Get(pricingURL)
  673. if err != nil {
  674. klog.V(2).Infof("Bogus fetch of \"%s\": %v", pricingURL, err)
  675. return nil, pricingURL, err
  676. }
  677. return resp, pricingURL, err
  678. }
  679. // SpotRefreshEnabled determines whether the required configs to run the spot feed query have been set up
  680. func (aws *AWS) SpotRefreshEnabled() bool {
  681. // Need a valid value for at least one of these fields to consider spot pricing as enabled
  682. return len(aws.SpotDataBucket) != 0 || len(aws.SpotDataRegion) != 0 || len(aws.ProjectID) != 0
  683. }
  684. // DownloadPricingData fetches data from the AWS Pricing API
  685. func (aws *AWS) DownloadPricingData() error {
  686. aws.DownloadPricingDataLock.Lock()
  687. defer aws.DownloadPricingDataLock.Unlock()
  688. c, err := aws.Config.GetCustomPricingData()
  689. if err != nil {
  690. klog.V(1).Infof("Error downloading default pricing data: %s", err.Error())
  691. }
  692. aws.BaseCPUPrice = c.CPU
  693. aws.BaseRAMPrice = c.RAM
  694. aws.BaseGPUPrice = c.GPU
  695. aws.BaseSpotCPUPrice = c.SpotCPU
  696. aws.BaseSpotRAMPrice = c.SpotRAM
  697. aws.BaseSpotGPUPrice = c.SpotGPU
  698. aws.SpotLabelName = c.SpotLabel
  699. aws.SpotLabelValue = c.SpotLabelValue
  700. aws.SpotDataBucket = c.SpotDataBucket
  701. aws.SpotDataPrefix = c.SpotDataPrefix
  702. aws.ProjectID = c.ProjectID
  703. aws.SpotDataRegion = c.SpotDataRegion
  704. aws.ConfigureAuthWith(c) // load aws authentication from configuration or secret
  705. if len(aws.SpotDataBucket) != 0 && len(aws.ProjectID) == 0 {
  706. klog.V(1).Infof("using SpotDataBucket \"%s\" without ProjectID will not end well", aws.SpotDataBucket)
  707. }
  708. nodeList := aws.Clientset.GetAllNodes()
  709. inputkeys := make(map[string]bool)
  710. for _, n := range nodeList {
  711. if _, ok := n.Labels["eks.amazonaws.com/nodegroup"]; ok {
  712. aws.clusterManagementPrice = 0.10
  713. aws.clusterProvisioner = "EKS"
  714. } else if _, ok := n.Labels["kops.k8s.io/instancegroup"]; ok {
  715. aws.clusterProvisioner = "KOPS"
  716. }
  717. labels := n.GetObjectMeta().GetLabels()
  718. key := aws.GetKey(labels, n)
  719. inputkeys[key.Features()] = true
  720. }
  721. pvList := aws.Clientset.GetAllPersistentVolumes()
  722. storageClasses := aws.Clientset.GetAllStorageClasses()
  723. storageClassMap := make(map[string]map[string]string)
  724. for _, storageClass := range storageClasses {
  725. params := storageClass.Parameters
  726. storageClassMap[storageClass.ObjectMeta.Name] = params
  727. if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  728. storageClassMap["default"] = params
  729. storageClassMap[""] = params
  730. }
  731. }
  732. pvkeys := make(map[string]PVKey)
  733. for _, pv := range pvList {
  734. params, ok := storageClassMap[pv.Spec.StorageClassName]
  735. if !ok {
  736. klog.V(2).Infof("Unable to find params for storageClassName %s, falling back to default pricing", pv.Spec.StorageClassName)
  737. continue
  738. }
  739. key := aws.GetPVKey(pv, params, "")
  740. pvkeys[key.Features()] = key
  741. }
  742. // RIDataRunning establishes the existance of the goroutine. Since it's possible we
  743. // run multiple downloads, we don't want to create multiple go routines if one already exists
  744. if !aws.RIDataRunning {
  745. err = aws.GetReservationDataFromAthena() // Block until one run has completed.
  746. if err != nil {
  747. klog.V(1).Infof("Failed to lookup reserved instance data: %s", err.Error())
  748. } else { // If we make one successful run, check on new reservation data every hour
  749. go func() {
  750. defer errors.HandlePanic()
  751. aws.RIDataRunning = true
  752. for {
  753. klog.Infof("Reserved Instance watcher running... next update in 1h")
  754. time.Sleep(time.Hour)
  755. err := aws.GetReservationDataFromAthena()
  756. if err != nil {
  757. klog.Infof("Error updating RI data: %s", err.Error())
  758. }
  759. }
  760. }()
  761. }
  762. }
  763. if !aws.SavingsPlanDataRunning {
  764. err = aws.GetSavingsPlanDataFromAthena()
  765. if err != nil {
  766. klog.V(1).Infof("Failed to lookup savings plan data: %s", err.Error())
  767. } else {
  768. go func() {
  769. defer errors.HandlePanic()
  770. aws.SavingsPlanDataRunning = true
  771. for {
  772. klog.Infof("Savings Plan watcher running... next update in 1h")
  773. time.Sleep(time.Hour)
  774. err := aws.GetSavingsPlanDataFromAthena()
  775. if err != nil {
  776. klog.Infof("Error updating Savings Plan data: %s", err.Error())
  777. }
  778. }
  779. }()
  780. }
  781. }
  782. aws.Pricing = make(map[string]*AWSProductTerms)
  783. aws.ValidPricingKeys = make(map[string]bool)
  784. skusToKeys := make(map[string]string)
  785. resp, pricingURL, err := aws.getRegionPricing(nodeList)
  786. if err != nil {
  787. return err
  788. }
  789. dec := json.NewDecoder(resp.Body)
  790. for {
  791. t, err := dec.Token()
  792. if err == io.EOF {
  793. klog.V(2).Infof("done loading \"%s\"\n", pricingURL)
  794. break
  795. } else if err != nil {
  796. klog.V(2).Infof("error parsing response json %v", resp.Body)
  797. break
  798. }
  799. if t == "products" {
  800. _, err := dec.Token() // this should parse the opening "{""
  801. if err != nil {
  802. return err
  803. }
  804. for dec.More() {
  805. _, err := dec.Token() // the sku token
  806. if err != nil {
  807. return err
  808. }
  809. product := &AWSProduct{}
  810. err = dec.Decode(&product)
  811. if err != nil {
  812. klog.V(1).Infof("Error parsing response from \"%s\": %v", pricingURL, err.Error())
  813. break
  814. }
  815. if product.Attributes.PreInstalledSw == "NA" &&
  816. (strings.HasPrefix(product.Attributes.UsageType, "BoxUsage") || strings.Contains(product.Attributes.UsageType, "-BoxUsage")) &&
  817. product.Attributes.CapacityStatus == "Used" {
  818. key := aws.KubeAttrConversion(product.Attributes.Location, product.Attributes.InstanceType, product.Attributes.OperatingSystem)
  819. spotKey := key + ",preemptible"
  820. if inputkeys[key] || inputkeys[spotKey] { // Just grab the sku even if spot, and change the price later.
  821. productTerms := &AWSProductTerms{
  822. Sku: product.Sku,
  823. Memory: product.Attributes.Memory,
  824. Storage: product.Attributes.Storage,
  825. VCpu: product.Attributes.VCpu,
  826. GPU: product.Attributes.GPU,
  827. }
  828. aws.Pricing[key] = productTerms
  829. aws.Pricing[spotKey] = productTerms
  830. skusToKeys[product.Sku] = key
  831. }
  832. aws.ValidPricingKeys[key] = true
  833. aws.ValidPricingKeys[spotKey] = true
  834. } else if strings.Contains(product.Attributes.UsageType, "EBS:Volume") {
  835. // UsageTypes may be prefixed with a region code - we're removing this when using
  836. // volTypes to keep lookups generic
  837. usageTypeRegx := regexp.MustCompile(".*(-|^)(EBS.+)")
  838. usageTypeMatch := usageTypeRegx.FindStringSubmatch(product.Attributes.UsageType)
  839. usageTypeNoRegion := usageTypeMatch[len(usageTypeMatch)-1]
  840. key := locationToRegion[product.Attributes.Location] + "," + usageTypeNoRegion
  841. spotKey := key + ",preemptible"
  842. pv := &PV{
  843. Class: volTypes[usageTypeNoRegion],
  844. Region: locationToRegion[product.Attributes.Location],
  845. }
  846. productTerms := &AWSProductTerms{
  847. Sku: product.Sku,
  848. PV: pv,
  849. }
  850. aws.Pricing[key] = productTerms
  851. aws.Pricing[spotKey] = productTerms
  852. skusToKeys[product.Sku] = key
  853. aws.ValidPricingKeys[key] = true
  854. aws.ValidPricingKeys[spotKey] = true
  855. }
  856. }
  857. }
  858. if t == "terms" {
  859. _, err := dec.Token() // this should parse the opening "{""
  860. if err != nil {
  861. return err
  862. }
  863. termType, err := dec.Token()
  864. if err != nil {
  865. return err
  866. }
  867. if termType == "OnDemand" {
  868. _, err := dec.Token()
  869. if err != nil { // again, should parse an opening "{"
  870. return err
  871. }
  872. for dec.More() {
  873. sku, err := dec.Token()
  874. if err != nil {
  875. return err
  876. }
  877. _, err = dec.Token() // another opening "{"
  878. if err != nil {
  879. return err
  880. }
  881. skuOnDemand, err := dec.Token()
  882. if err != nil {
  883. return err
  884. }
  885. offerTerm := &AWSOfferTerm{}
  886. err = dec.Decode(&offerTerm)
  887. if err != nil {
  888. klog.V(1).Infof("Error decoding AWS Offer Term: " + err.Error())
  889. }
  890. key, ok := skusToKeys[sku.(string)]
  891. spotKey := key + ",preemptible"
  892. if ok {
  893. aws.Pricing[key].OnDemand = offerTerm
  894. aws.Pricing[spotKey].OnDemand = offerTerm
  895. var cost string
  896. if sku.(string)+OnDemandRateCode == skuOnDemand {
  897. cost = offerTerm.PriceDimensions[sku.(string)+OnDemandRateCode+HourlyRateCode].PricePerUnit.USD
  898. } else if sku.(string)+OnDemandRateCodeCn == skuOnDemand {
  899. cost = offerTerm.PriceDimensions[sku.(string)+OnDemandRateCodeCn+HourlyRateCodeCn].PricePerUnit.CNY
  900. }
  901. if strings.Contains(key, "EBS:VolumeP-IOPS.piops") {
  902. // If the specific UsageType is the per IO cost used on io1 volumes
  903. // we need to add the per IO cost to the io1 PV cost
  904. // Add the per IO cost to the PV object for the io1 volume type
  905. aws.Pricing[key].PV.CostPerIO = cost
  906. } else if strings.Contains(key, "EBS:Volume") {
  907. // If volume, we need to get hourly cost and add it to the PV object
  908. costFloat, _ := strconv.ParseFloat(cost, 64)
  909. hourlyPrice := costFloat / 730
  910. aws.Pricing[key].PV.Cost = strconv.FormatFloat(hourlyPrice, 'f', -1, 64)
  911. }
  912. }
  913. _, err = dec.Token()
  914. if err != nil {
  915. return err
  916. }
  917. }
  918. _, err = dec.Token()
  919. if err != nil {
  920. return err
  921. }
  922. }
  923. }
  924. }
  925. klog.V(2).Infof("Finished downloading \"%s\"", pricingURL)
  926. if !aws.SpotRefreshEnabled() {
  927. return nil
  928. }
  929. // Always run spot pricing refresh when performing download
  930. aws.refreshSpotPricing(true)
  931. // Only start a single refresh goroutine
  932. if !aws.SpotRefreshRunning {
  933. aws.SpotRefreshRunning = true
  934. go func() {
  935. defer errors.HandlePanic()
  936. for {
  937. klog.Infof("Spot Pricing Refresh scheduled in %.2f minutes.", SpotRefreshDuration.Minutes())
  938. time.Sleep(SpotRefreshDuration)
  939. // Reoccurring refresh checks update times
  940. aws.refreshSpotPricing(false)
  941. }
  942. }()
  943. }
  944. return nil
  945. }
  946. func (aws *AWS) refreshSpotPricing(force bool) {
  947. aws.SpotPricingLock.Lock()
  948. defer aws.SpotPricingLock.Unlock()
  949. now := time.Now().UTC()
  950. updateTime := now.Add(-SpotRefreshDuration)
  951. // Return if there was an update time set and an hour hasn't elapsed
  952. if !force && aws.SpotPricingUpdatedAt != nil && aws.SpotPricingUpdatedAt.After(updateTime) {
  953. return
  954. }
  955. sp, err := aws.parseSpotData(aws.SpotDataBucket, aws.SpotDataPrefix, aws.ProjectID, aws.SpotDataRegion)
  956. if err != nil {
  957. klog.V(1).Infof("Skipping AWS spot data download: %s", err.Error())
  958. aws.SpotPricingError = err
  959. return
  960. }
  961. aws.SpotPricingError = nil
  962. // update time last updated
  963. aws.SpotPricingUpdatedAt = &now
  964. aws.SpotPricingByInstanceID = sp
  965. }
  966. // Stubbed NetworkPricing for AWS. Pull directly from aws.json for now
  967. func (aws *AWS) NetworkPricing() (*Network, error) {
  968. cpricing, err := aws.Config.GetCustomPricingData()
  969. if err != nil {
  970. return nil, err
  971. }
  972. znec, err := strconv.ParseFloat(cpricing.ZoneNetworkEgress, 64)
  973. if err != nil {
  974. return nil, err
  975. }
  976. rnec, err := strconv.ParseFloat(cpricing.RegionNetworkEgress, 64)
  977. if err != nil {
  978. return nil, err
  979. }
  980. inec, err := strconv.ParseFloat(cpricing.InternetNetworkEgress, 64)
  981. if err != nil {
  982. return nil, err
  983. }
  984. return &Network{
  985. ZoneNetworkEgressCost: znec,
  986. RegionNetworkEgressCost: rnec,
  987. InternetNetworkEgressCost: inec,
  988. }, nil
  989. }
  990. func (aws *AWS) LoadBalancerPricing() (*LoadBalancer, error) {
  991. fffrc := 0.025
  992. afrc := 0.010
  993. lbidc := 0.008
  994. numForwardingRules := 1.0
  995. dataIngressGB := 0.0
  996. var totalCost float64
  997. if numForwardingRules < 5 {
  998. totalCost = fffrc*numForwardingRules + lbidc*dataIngressGB
  999. } else {
  1000. totalCost = fffrc*5 + afrc*(numForwardingRules-5) + lbidc*dataIngressGB
  1001. }
  1002. return &LoadBalancer{
  1003. Cost: totalCost,
  1004. }, nil
  1005. }
  1006. // AllNodePricing returns all the billing data fetched.
  1007. func (aws *AWS) AllNodePricing() (interface{}, error) {
  1008. aws.DownloadPricingDataLock.RLock()
  1009. defer aws.DownloadPricingDataLock.RUnlock()
  1010. return aws.Pricing, nil
  1011. }
  1012. func (aws *AWS) spotPricing(instanceID string) (*spotInfo, bool) {
  1013. aws.SpotPricingLock.RLock()
  1014. defer aws.SpotPricingLock.RUnlock()
  1015. info, ok := aws.SpotPricingByInstanceID[instanceID]
  1016. return info, ok
  1017. }
  1018. func (aws *AWS) reservedInstancePricing(instanceID string) (*RIData, bool) {
  1019. aws.RIDataLock.RLock()
  1020. defer aws.RIDataLock.RUnlock()
  1021. data, ok := aws.RIPricingByInstanceID[instanceID]
  1022. return data, ok
  1023. }
  1024. func (aws *AWS) savingsPlanPricing(instanceID string) (*SavingsPlanData, bool) {
  1025. aws.SavingsPlanDataLock.RLock()
  1026. defer aws.SavingsPlanDataLock.RUnlock()
  1027. data, ok := aws.SavingsPlanDataByInstanceID[instanceID]
  1028. return data, ok
  1029. }
  1030. func (aws *AWS) createNode(terms *AWSProductTerms, usageType string, k Key) (*Node, error) {
  1031. key := k.Features()
  1032. if spotInfo, ok := aws.spotPricing(k.ID()); ok {
  1033. var spotcost string
  1034. log.DedupedInfof(5, "Looking up spot data from feed for node %s", k.ID())
  1035. arr := strings.Split(spotInfo.Charge, " ")
  1036. if len(arr) == 2 {
  1037. spotcost = arr[0]
  1038. } else {
  1039. klog.V(2).Infof("Spot data for node %s is missing", k.ID())
  1040. }
  1041. return &Node{
  1042. Cost: spotcost,
  1043. VCPU: terms.VCpu,
  1044. RAM: terms.Memory,
  1045. GPU: terms.GPU,
  1046. Storage: terms.Storage,
  1047. BaseCPUPrice: aws.BaseCPUPrice,
  1048. BaseRAMPrice: aws.BaseRAMPrice,
  1049. BaseGPUPrice: aws.BaseGPUPrice,
  1050. UsageType: PreemptibleType,
  1051. }, nil
  1052. } else if aws.isPreemptible(key) { // Preemptible but we don't have any data in the pricing report.
  1053. log.DedupedWarningf(5, "Node %s marked preemptible but we have no data in spot feed", k.ID())
  1054. return &Node{
  1055. VCPU: terms.VCpu,
  1056. VCPUCost: aws.BaseSpotCPUPrice,
  1057. RAM: terms.Memory,
  1058. GPU: terms.GPU,
  1059. Storage: terms.Storage,
  1060. BaseCPUPrice: aws.BaseCPUPrice,
  1061. BaseRAMPrice: aws.BaseRAMPrice,
  1062. BaseGPUPrice: aws.BaseGPUPrice,
  1063. UsageType: PreemptibleType,
  1064. }, nil
  1065. } else if sp, ok := aws.savingsPlanPricing(k.ID()); ok {
  1066. strCost := fmt.Sprintf("%f", sp.EffectiveCost)
  1067. return &Node{
  1068. Cost: strCost,
  1069. VCPU: terms.VCpu,
  1070. RAM: terms.Memory,
  1071. GPU: terms.GPU,
  1072. Storage: terms.Storage,
  1073. BaseCPUPrice: aws.BaseCPUPrice,
  1074. BaseRAMPrice: aws.BaseRAMPrice,
  1075. BaseGPUPrice: aws.BaseGPUPrice,
  1076. UsageType: usageType,
  1077. }, nil
  1078. } else if ri, ok := aws.reservedInstancePricing(k.ID()); ok {
  1079. strCost := fmt.Sprintf("%f", ri.EffectiveCost)
  1080. return &Node{
  1081. Cost: strCost,
  1082. VCPU: terms.VCpu,
  1083. RAM: terms.Memory,
  1084. GPU: terms.GPU,
  1085. Storage: terms.Storage,
  1086. BaseCPUPrice: aws.BaseCPUPrice,
  1087. BaseRAMPrice: aws.BaseRAMPrice,
  1088. BaseGPUPrice: aws.BaseGPUPrice,
  1089. UsageType: usageType,
  1090. }, nil
  1091. }
  1092. var cost string
  1093. c, ok := terms.OnDemand.PriceDimensions[terms.Sku+OnDemandRateCode+HourlyRateCode]
  1094. if ok {
  1095. cost = c.PricePerUnit.USD
  1096. } else {
  1097. // Check for Chinese pricing before throwing error
  1098. c, ok = terms.OnDemand.PriceDimensions[terms.Sku+OnDemandRateCodeCn+HourlyRateCodeCn]
  1099. if ok {
  1100. cost = c.PricePerUnit.CNY
  1101. } else {
  1102. return nil, fmt.Errorf("Could not fetch data for \"%s\"", k.ID())
  1103. }
  1104. }
  1105. return &Node{
  1106. Cost: cost,
  1107. VCPU: terms.VCpu,
  1108. RAM: terms.Memory,
  1109. GPU: terms.GPU,
  1110. Storage: terms.Storage,
  1111. BaseCPUPrice: aws.BaseCPUPrice,
  1112. BaseRAMPrice: aws.BaseRAMPrice,
  1113. BaseGPUPrice: aws.BaseGPUPrice,
  1114. UsageType: usageType,
  1115. }, nil
  1116. }
  1117. // NodePricing takes in a key from GetKey and returns a Node object for use in building the cost model.
  1118. func (aws *AWS) NodePricing(k Key) (*Node, error) {
  1119. aws.DownloadPricingDataLock.RLock()
  1120. defer aws.DownloadPricingDataLock.RUnlock()
  1121. key := k.Features()
  1122. usageType := "ondemand"
  1123. if aws.isPreemptible(key) {
  1124. usageType = PreemptibleType
  1125. }
  1126. terms, ok := aws.Pricing[key]
  1127. if ok {
  1128. return aws.createNode(terms, usageType, k)
  1129. } else if _, ok := aws.ValidPricingKeys[key]; ok {
  1130. aws.DownloadPricingDataLock.RUnlock()
  1131. err := aws.DownloadPricingData()
  1132. aws.DownloadPricingDataLock.RLock()
  1133. if err != nil {
  1134. return &Node{
  1135. Cost: aws.BaseCPUPrice,
  1136. BaseCPUPrice: aws.BaseCPUPrice,
  1137. BaseRAMPrice: aws.BaseRAMPrice,
  1138. BaseGPUPrice: aws.BaseGPUPrice,
  1139. UsageType: usageType,
  1140. UsesBaseCPUPrice: true,
  1141. }, err
  1142. }
  1143. terms, termsOk := aws.Pricing[key]
  1144. if !termsOk {
  1145. return &Node{
  1146. Cost: aws.BaseCPUPrice,
  1147. BaseCPUPrice: aws.BaseCPUPrice,
  1148. BaseRAMPrice: aws.BaseRAMPrice,
  1149. BaseGPUPrice: aws.BaseGPUPrice,
  1150. UsageType: usageType,
  1151. UsesBaseCPUPrice: true,
  1152. }, fmt.Errorf("Unable to find any Pricing data for \"%s\"", key)
  1153. }
  1154. return aws.createNode(terms, usageType, k)
  1155. } else { // Fall back to base pricing if we can't find the key. Base pricing is handled at the costmodel level.
  1156. return nil, fmt.Errorf("Invalid Pricing Key \"%s\"", key)
  1157. }
  1158. }
  1159. // ClusterInfo returns an object that represents the cluster. TODO: actually return the name of the cluster. Blocked on cluster federation.
  1160. func (awsProvider *AWS) ClusterInfo() (map[string]string, error) {
  1161. defaultClusterName := "AWS Cluster #1"
  1162. c, err := awsProvider.GetConfig()
  1163. if err != nil {
  1164. return nil, err
  1165. }
  1166. remoteEnabled := env.IsRemoteEnabled()
  1167. makeStructure := func(clusterName string) (map[string]string, error) {
  1168. m := make(map[string]string)
  1169. m["name"] = clusterName
  1170. m["provider"] = "AWS"
  1171. m["account"] = c.AthenaProjectID // this value requires configuration but is unavailable else where
  1172. m["region"] = awsProvider.clusterRegion
  1173. m["id"] = env.GetClusterID()
  1174. m["remoteReadEnabled"] = strconv.FormatBool(remoteEnabled)
  1175. m["provisioner"] = awsProvider.clusterProvisioner
  1176. return m, nil
  1177. }
  1178. if c.ClusterName != "" {
  1179. return makeStructure(c.ClusterName)
  1180. }
  1181. maybeClusterId := env.GetAWSClusterID()
  1182. if len(maybeClusterId) != 0 {
  1183. klog.V(2).Infof("Returning \"%s\" as ClusterName", maybeClusterId)
  1184. return makeStructure(maybeClusterId)
  1185. }
  1186. klog.V(2).Infof("Unable to sniff out cluster ID, perhaps set $%s to force one", env.AWSClusterIDEnvVar)
  1187. return makeStructure(defaultClusterName)
  1188. }
  1189. // updates the authentication to the latest values (via config or secret)
  1190. func (aws *AWS) ConfigureAuth() error {
  1191. c, err := aws.Config.GetCustomPricingData()
  1192. if err != nil {
  1193. klog.V(1).Infof("Error downloading default pricing data: %s", err.Error())
  1194. }
  1195. return aws.ConfigureAuthWith(c)
  1196. }
  1197. // updates the authentication to the latest values (via config or secret)
  1198. func (aws *AWS) ConfigureAuthWith(config *CustomPricing) error {
  1199. accessKeyID, accessKeySecret := aws.getAWSAuth(false, config)
  1200. if accessKeyID != "" && accessKeySecret != "" { // credentials may exist on the actual AWS node-- if so, use those. If not, override with the service key
  1201. err := env.Set(env.AWSAccessKeyIDEnvVar, accessKeyID)
  1202. if err != nil {
  1203. return err
  1204. }
  1205. err = env.Set(env.AWSAccessKeySecretEnvVar, accessKeySecret)
  1206. if err != nil {
  1207. return err
  1208. }
  1209. }
  1210. return nil
  1211. }
  1212. // Gets the aws key id and secret
  1213. func (aws *AWS) getAWSAuth(forceReload bool, cp *CustomPricing) (string, string) {
  1214. // 1. Check config values first (set from frontend UI)
  1215. if cp.ServiceKeyName != "" && cp.ServiceKeySecret != "" {
  1216. aws.serviceAccountChecks.set("hasKey", &ServiceAccountCheck{
  1217. Message: "AWS ServiceKey exists",
  1218. Status: true,
  1219. })
  1220. return cp.ServiceKeyName, cp.ServiceKeySecret
  1221. }
  1222. // 2. Check for secret
  1223. s, _ := aws.loadAWSAuthSecret(forceReload)
  1224. if s != nil && s.AccessKeyID != "" && s.SecretAccessKey != "" {
  1225. aws.serviceAccountChecks.set("hasKey", &ServiceAccountCheck{
  1226. Message: "AWS ServiceKey exists",
  1227. Status: true,
  1228. })
  1229. return s.AccessKeyID, s.SecretAccessKey
  1230. }
  1231. // 3. Fall back to env vars
  1232. if env.GetAWSAccessKeyID() == "" || env.GetAWSAccessKeyID() == "" {
  1233. aws.serviceAccountChecks.set("hasKey", &ServiceAccountCheck{
  1234. Message: "AWS ServiceKey exists",
  1235. Status: false,
  1236. })
  1237. } else {
  1238. aws.serviceAccountChecks.set("hasKey", &ServiceAccountCheck{
  1239. Message: "AWS ServiceKey exists",
  1240. Status: true,
  1241. })
  1242. }
  1243. return env.GetAWSAccessKeyID(), env.GetAWSAccessKeySecret()
  1244. }
  1245. // Load once and cache the result (even on failure). This is an install time secret, so
  1246. // we don't expect the secret to change. If it does, however, we can force reload using
  1247. // the input parameter.
  1248. func (aws *AWS) loadAWSAuthSecret(force bool) (*AWSAccessKey, error) {
  1249. if !force && loadedAWSSecret {
  1250. return awsSecret, nil
  1251. }
  1252. loadedAWSSecret = true
  1253. exists, err := fileutil.FileExists(authSecretPath)
  1254. if !exists || err != nil {
  1255. return nil, fmt.Errorf("Failed to locate service account file: %s", authSecretPath)
  1256. }
  1257. result, err := ioutil.ReadFile(authSecretPath)
  1258. if err != nil {
  1259. return nil, err
  1260. }
  1261. var ak AWSAccessKey
  1262. err = json.Unmarshal(result, &ak)
  1263. if err != nil {
  1264. return nil, err
  1265. }
  1266. awsSecret = &ak
  1267. return awsSecret, nil
  1268. }
  1269. func (aws *AWS) getAddressesForRegion(ctx context.Context, region string) (*ec2.DescribeAddressesOutput, error) {
  1270. aak, err := aws.GetAWSAccessKey()
  1271. if err != nil {
  1272. return nil, err
  1273. }
  1274. cfg, err := aak.CreateConfig(region)
  1275. if err != nil {
  1276. return nil, err
  1277. }
  1278. cli := ec2.NewFromConfig(cfg)
  1279. return cli.DescribeAddresses(ctx, &ec2.DescribeAddressesInput{})
  1280. }
  1281. // GetAddresses retrieves EC2 addresses
  1282. func (aws *AWS) GetAddresses() ([]byte, error) {
  1283. aws.ConfigureAuth() // load authentication data into env vars
  1284. addressCh := make(chan *ec2.DescribeAddressesOutput, len(awsRegions))
  1285. errorCh := make(chan error, len(awsRegions))
  1286. var wg sync.WaitGroup
  1287. wg.Add(len(awsRegions))
  1288. // Get volumes from each AWS region
  1289. for _, r := range awsRegions {
  1290. // Fetch IP address response and send results and errors to their
  1291. // respective channels
  1292. go func(region string) {
  1293. defer wg.Done()
  1294. defer errors.HandlePanic()
  1295. // Query for first page of volume results
  1296. resp, err := aws.getAddressesForRegion(context.TODO(), region)
  1297. if err != nil {
  1298. errorCh <- err
  1299. return
  1300. }
  1301. addressCh <- resp
  1302. }(r)
  1303. }
  1304. // Close the result channels after everything has been sent
  1305. go func() {
  1306. defer errors.HandlePanic()
  1307. wg.Wait()
  1308. close(errorCh)
  1309. close(addressCh)
  1310. }()
  1311. var addresses []*ec2Types.Address
  1312. for adds := range addressCh {
  1313. for _, add := range adds.Addresses {
  1314. a := add // duplicate to avoid pointer to iterator
  1315. addresses = append(addresses, &a)
  1316. }
  1317. }
  1318. var errs []error
  1319. for err := range errorCh {
  1320. log.DedupedWarningf(5, "unable to get addresses: %s", err)
  1321. errs = append(errs, err)
  1322. }
  1323. // Return error if no addresses are returned
  1324. if len(errs) > 0 && len(addresses) == 0 {
  1325. return nil, fmt.Errorf("%d error(s) retrieving addresses: %v", len(errs), errs)
  1326. }
  1327. // Format the response this way to match the JSON-encoded formatting of a single response
  1328. // from DescribeAddresss, so that consumers can always expect AWS disk responses to have
  1329. // a "Addresss" key at the top level.
  1330. return json.Marshal(map[string][]*ec2Types.Address{
  1331. "Addresses": addresses,
  1332. })
  1333. }
  1334. func (aws *AWS) getDisksForRegion(ctx context.Context, region string, maxResults int32, nextToken *string) (*ec2.DescribeVolumesOutput, error) {
  1335. aak, err := aws.GetAWSAccessKey()
  1336. if err != nil {
  1337. return nil, err
  1338. }
  1339. cfg, err := aak.CreateConfig(region)
  1340. if err != nil {
  1341. return nil, err
  1342. }
  1343. cli := ec2.NewFromConfig(cfg)
  1344. return cli.DescribeVolumes(ctx, &ec2.DescribeVolumesInput{
  1345. MaxResults: &maxResults,
  1346. NextToken: nextToken,
  1347. })
  1348. }
  1349. // GetDisks returns the AWS disks backing PVs. Useful because sometimes k8s will not clean up PVs correctly. Requires a json config in /var/configs with key region.
  1350. func (aws *AWS) GetDisks() ([]byte, error) {
  1351. aws.ConfigureAuth() // load authentication data into env vars
  1352. volumeCh := make(chan *ec2.DescribeVolumesOutput, len(awsRegions))
  1353. errorCh := make(chan error, len(awsRegions))
  1354. var wg sync.WaitGroup
  1355. wg.Add(len(awsRegions))
  1356. // Get volumes from each AWS region
  1357. for _, r := range awsRegions {
  1358. // Fetch volume response and send results and errors to their
  1359. // respective channels
  1360. go func(region string) {
  1361. defer wg.Done()
  1362. defer errors.HandlePanic()
  1363. // Query for first page of volume results
  1364. resp, err := aws.getDisksForRegion(context.TODO(), region, 1000, nil)
  1365. if err != nil {
  1366. errorCh <- err
  1367. return
  1368. }
  1369. volumeCh <- resp
  1370. // A NextToken indicates more pages of results. Keep querying
  1371. // until all pages are retrieved.
  1372. for resp.NextToken != nil {
  1373. resp, err = aws.getDisksForRegion(context.TODO(), region, 100, resp.NextToken)
  1374. if err != nil {
  1375. errorCh <- err
  1376. return
  1377. }
  1378. volumeCh <- resp
  1379. }
  1380. }(r)
  1381. }
  1382. // Close the result channels after everything has been sent
  1383. go func() {
  1384. defer errors.HandlePanic()
  1385. wg.Wait()
  1386. close(errorCh)
  1387. close(volumeCh)
  1388. }()
  1389. var volumes []*ec2Types.Volume
  1390. for vols := range volumeCh {
  1391. for _, vol := range vols.Volumes {
  1392. v := vol // duplicate to avoid pointer to iterator
  1393. volumes = append(volumes, &v)
  1394. }
  1395. }
  1396. var errs []error
  1397. for err := range errorCh {
  1398. log.DedupedWarningf(5, "unable to get disks: %s", err)
  1399. errs = append(errs, err)
  1400. }
  1401. // Return error if no volumes are returned
  1402. if len(errs) > 0 && len(volumes) == 0 {
  1403. return nil, fmt.Errorf("%d error(s) retrieving volumes: %v", len(errs), errs)
  1404. }
  1405. // Format the response this way to match the JSON-encoded formatting of a single response
  1406. // from DescribeVolumes, so that consumers can always expect AWS disk responses to have
  1407. // a "Volumes" key at the top level.
  1408. return json.Marshal(map[string][]*ec2Types.Volume{
  1409. "Volumes": volumes,
  1410. })
  1411. }
  1412. // QueryAthenaPaginated executes athena query and processes results.
  1413. func (aws *AWS) QueryAthenaPaginated(ctx context.Context, query string, fn func(*athena.GetQueryResultsOutput) bool) error {
  1414. awsAthenaInfo, err := aws.GetAWSAthenaInfo()
  1415. if err != nil {
  1416. return err
  1417. }
  1418. if awsAthenaInfo.AthenaDatabase == "" || awsAthenaInfo.AthenaTable == "" || awsAthenaInfo.AthenaRegion == "" ||
  1419. awsAthenaInfo.AthenaBucketName == "" || awsAthenaInfo.AccountID == "" {
  1420. return fmt.Errorf("QueryAthenaPaginated: athena configuration incomplete")
  1421. }
  1422. queryExecutionCtx := &athenaTypes.QueryExecutionContext{
  1423. Database: awsSDK.String(awsAthenaInfo.AthenaDatabase),
  1424. }
  1425. resultConfiguration := &athenaTypes.ResultConfiguration{
  1426. OutputLocation: awsSDK.String(awsAthenaInfo.AthenaBucketName),
  1427. }
  1428. startQueryExecutionInput := &athena.StartQueryExecutionInput{
  1429. QueryString: awsSDK.String(query),
  1430. QueryExecutionContext: queryExecutionCtx,
  1431. ResultConfiguration: resultConfiguration,
  1432. }
  1433. // Create Athena Client
  1434. cfg, err := awsAthenaInfo.CreateConfig()
  1435. if err != nil {
  1436. log.Errorf("Could not retrieve Athena Configuration: %s", err.Error())
  1437. }
  1438. cli := athena.NewFromConfig(cfg)
  1439. // Query Athena
  1440. startQueryExecutionOutput, err := cli.StartQueryExecution(ctx, startQueryExecutionInput)
  1441. if err != nil {
  1442. return fmt.Errorf("QueryAthenaPaginated: start query error: %s", err.Error())
  1443. }
  1444. err = waitForQueryToComplete(ctx, cli, startQueryExecutionOutput.QueryExecutionId)
  1445. if err != nil {
  1446. return fmt.Errorf("QueryAthenaPaginated: query execution error: %s", err.Error())
  1447. }
  1448. queryResultsInput := &athena.GetQueryResultsInput{
  1449. QueryExecutionId: startQueryExecutionOutput.QueryExecutionId,
  1450. }
  1451. getQueryResultsPaginator := athena.NewGetQueryResultsPaginator(cli, queryResultsInput)
  1452. for getQueryResultsPaginator.HasMorePages() {
  1453. pg, err := getQueryResultsPaginator.NextPage(ctx)
  1454. if err != nil {
  1455. log.Errorf("QueryAthenaPaginated: NextPage error: %s", err.Error())
  1456. continue
  1457. }
  1458. fn(pg)
  1459. }
  1460. return nil
  1461. }
  1462. func waitForQueryToComplete(ctx context.Context, client *athena.Client, queryExecutionID *string) error {
  1463. inp := &athena.GetQueryExecutionInput{
  1464. QueryExecutionId: queryExecutionID,
  1465. }
  1466. isQueryStillRunning := true
  1467. for isQueryStillRunning {
  1468. qe, err := client.GetQueryExecution(ctx, inp)
  1469. if err != nil {
  1470. return err
  1471. }
  1472. if qe.QueryExecution.Status.State == "SUCCEEDED" {
  1473. isQueryStillRunning = false
  1474. continue
  1475. }
  1476. if qe.QueryExecution.Status.State != "RUNNING" && qe.QueryExecution.Status.State != "QUEUED" {
  1477. return fmt.Errorf("no query results available for query %s", *queryExecutionID)
  1478. }
  1479. time.Sleep(2 * time.Second)
  1480. }
  1481. return nil
  1482. }
  1483. type SavingsPlanData struct {
  1484. ResourceID string
  1485. EffectiveCost float64
  1486. SavingsPlanARN string
  1487. MostRecentDate string
  1488. }
  1489. func (aws *AWS) GetSavingsPlanDataFromAthena() error {
  1490. cfg, err := aws.GetConfig()
  1491. if err != nil {
  1492. aws.RIPricingError = err
  1493. return err
  1494. }
  1495. if cfg.AthenaBucketName == "" {
  1496. err = fmt.Errorf("No Athena Bucket configured")
  1497. aws.RIPricingError = err
  1498. return err
  1499. }
  1500. if aws.SavingsPlanDataByInstanceID == nil {
  1501. aws.SavingsPlanDataByInstanceID = make(map[string]*SavingsPlanData)
  1502. }
  1503. tNow := time.Now()
  1504. tOneDayAgo := tNow.Add(time.Duration(-25) * time.Hour) // Also get files from one day ago to avoid boundary conditions
  1505. start := tOneDayAgo.Format("2006-01-02")
  1506. end := tNow.Format("2006-01-02")
  1507. // Use Savings Plan Effective Rate as an estimation for cost, assuming the 1h most recent period got a fully loaded savings plan.
  1508. //
  1509. q := `SELECT
  1510. line_item_usage_start_date,
  1511. savings_plan_savings_plan_a_r_n,
  1512. line_item_resource_id,
  1513. savings_plan_savings_plan_rate
  1514. FROM %s as cost_data
  1515. WHERE line_item_usage_start_date BETWEEN date '%s' AND date '%s'
  1516. AND line_item_line_item_type = 'SavingsPlanCoveredUsage' ORDER BY
  1517. line_item_usage_start_date DESC`
  1518. page := 0
  1519. processResults := func(op *athena.GetQueryResultsOutput) bool {
  1520. if op == nil {
  1521. log.Errorf("GetSavingsPlanDataFromAthena: Athena page is nil")
  1522. return false
  1523. } else if op.ResultSet == nil {
  1524. log.Errorf("GetSavingsPlanDataFromAthena: Athena page.ResultSet is nil")
  1525. return false
  1526. }
  1527. aws.SavingsPlanDataLock.Lock()
  1528. aws.SavingsPlanDataByInstanceID = make(map[string]*SavingsPlanData) // Clean out the old data and only report a savingsplan price if its in the most recent run.
  1529. mostRecentDate := ""
  1530. iter := op.ResultSet.Rows
  1531. if page == 0 && len(iter) > 0 {
  1532. iter = op.ResultSet.Rows[1:len(op.ResultSet.Rows)]
  1533. }
  1534. page++
  1535. for _, r := range iter {
  1536. d := *r.Data[0].VarCharValue
  1537. if mostRecentDate == "" {
  1538. mostRecentDate = d
  1539. } else if mostRecentDate != d { // Get all most recent assignments
  1540. break
  1541. }
  1542. cost, err := strconv.ParseFloat(*r.Data[3].VarCharValue, 64)
  1543. if err != nil {
  1544. klog.Infof("Error converting `%s` from float ", *r.Data[3].VarCharValue)
  1545. }
  1546. r := &SavingsPlanData{
  1547. ResourceID: *r.Data[2].VarCharValue,
  1548. EffectiveCost: cost,
  1549. SavingsPlanARN: *r.Data[1].VarCharValue,
  1550. MostRecentDate: d,
  1551. }
  1552. aws.SavingsPlanDataByInstanceID[r.ResourceID] = r
  1553. }
  1554. klog.V(1).Infof("Found %d savings plan applied instances", len(aws.SavingsPlanDataByInstanceID))
  1555. for k, r := range aws.SavingsPlanDataByInstanceID {
  1556. log.DedupedInfof(5, "Savings Plan Instance Data found for node %s : %f at time %s", k, r.EffectiveCost, r.MostRecentDate)
  1557. }
  1558. aws.SavingsPlanDataLock.Unlock()
  1559. return true
  1560. }
  1561. query := fmt.Sprintf(q, cfg.AthenaTable, start, end)
  1562. klog.V(3).Infof("Running Query: %s", query)
  1563. err = aws.QueryAthenaPaginated(context.TODO(), query, processResults)
  1564. if err != nil {
  1565. aws.RIPricingError = err
  1566. return fmt.Errorf("Error fetching Savings Plan Data: %s", err)
  1567. }
  1568. return nil
  1569. }
  1570. type RIData struct {
  1571. ResourceID string
  1572. EffectiveCost float64
  1573. ReservationARN string
  1574. MostRecentDate string
  1575. }
  1576. func (aws *AWS) GetReservationDataFromAthena() error {
  1577. cfg, err := aws.GetConfig()
  1578. if err != nil {
  1579. aws.RIPricingError = err
  1580. return err
  1581. }
  1582. if cfg.AthenaBucketName == "" {
  1583. err = fmt.Errorf("No Athena Bucket configured")
  1584. aws.RIPricingError = err
  1585. return err
  1586. }
  1587. // Query for all column names in advance in order to validate configured
  1588. // label columns
  1589. columns, _ := aws.fetchColumns()
  1590. if !columns["reservation_reservation_a_r_n"] || !columns["reservation_effective_cost"] {
  1591. err = fmt.Errorf("no reservation data available in Athena")
  1592. aws.RIPricingError = err
  1593. return err
  1594. }
  1595. if aws.RIPricingByInstanceID == nil {
  1596. aws.RIPricingByInstanceID = make(map[string]*RIData)
  1597. }
  1598. tNow := time.Now()
  1599. tOneDayAgo := tNow.Add(time.Duration(-25) * time.Hour) // Also get files from one day ago to avoid boundary conditions
  1600. start := tOneDayAgo.Format("2006-01-02")
  1601. end := tNow.Format("2006-01-02")
  1602. q := `SELECT
  1603. line_item_usage_start_date,
  1604. reservation_reservation_a_r_n,
  1605. line_item_resource_id,
  1606. reservation_effective_cost
  1607. FROM %s as cost_data
  1608. WHERE line_item_usage_start_date BETWEEN date '%s' AND date '%s'
  1609. AND reservation_reservation_a_r_n <> '' ORDER BY
  1610. line_item_usage_start_date DESC`
  1611. page := 0
  1612. processResults := func(op *athena.GetQueryResultsOutput) bool {
  1613. if op == nil {
  1614. log.Errorf("GetReservationDataFromAthena: Athena page is nil")
  1615. return false
  1616. } else if op.ResultSet == nil {
  1617. log.Errorf("GetReservationDataFromAthena: Athena page.ResultSet is nil")
  1618. return false
  1619. }
  1620. aws.RIDataLock.Lock()
  1621. aws.RIPricingByInstanceID = make(map[string]*RIData) // Clean out the old data and only report a RI price if its in the most recent run.
  1622. mostRecentDate := ""
  1623. iter := op.ResultSet.Rows
  1624. if page == 0 && len(iter) > 0 {
  1625. iter = op.ResultSet.Rows[1:len(op.ResultSet.Rows)]
  1626. }
  1627. page++
  1628. for _, r := range iter {
  1629. d := *r.Data[0].VarCharValue
  1630. if mostRecentDate == "" {
  1631. mostRecentDate = d
  1632. } else if mostRecentDate != d { // Get all most recent assignments
  1633. break
  1634. }
  1635. cost, err := strconv.ParseFloat(*r.Data[3].VarCharValue, 64)
  1636. if err != nil {
  1637. klog.Infof("Error converting `%s` from float ", *r.Data[3].VarCharValue)
  1638. }
  1639. r := &RIData{
  1640. ResourceID: *r.Data[2].VarCharValue,
  1641. EffectiveCost: cost,
  1642. ReservationARN: *r.Data[1].VarCharValue,
  1643. MostRecentDate: d,
  1644. }
  1645. aws.RIPricingByInstanceID[r.ResourceID] = r
  1646. }
  1647. klog.V(1).Infof("Found %d reserved instances", len(aws.RIPricingByInstanceID))
  1648. for k, r := range aws.RIPricingByInstanceID {
  1649. log.DedupedInfof(5, "Reserved Instance Data found for node %s : %f at time %s", k, r.EffectiveCost, r.MostRecentDate)
  1650. }
  1651. aws.RIDataLock.Unlock()
  1652. return true
  1653. }
  1654. query := fmt.Sprintf(q, cfg.AthenaTable, start, end)
  1655. klog.V(3).Infof("Running Query: %s", query)
  1656. err = aws.QueryAthenaPaginated(context.TODO(), query, processResults)
  1657. if err != nil {
  1658. aws.RIPricingError = err
  1659. return fmt.Errorf("Error fetching Reserved Instance Data: %s", err)
  1660. }
  1661. aws.RIPricingError = nil
  1662. return nil
  1663. }
  1664. // fetchColumns returns a list of the names of all columns in the configured
  1665. // Athena tables
  1666. func (aws *AWS) fetchColumns() (map[string]bool, error) {
  1667. columnSet := map[string]bool{}
  1668. awsAthenaInfo, err := aws.GetAWSAthenaInfo()
  1669. if err != nil {
  1670. return nil, err
  1671. }
  1672. // This Query is supported by Athena tables and views
  1673. q := `SELECT column_name FROM information_schema.columns WHERE table_schema = '%s' AND table_name = '%s'`
  1674. query := fmt.Sprintf(q, awsAthenaInfo.AthenaDatabase, awsAthenaInfo.AthenaTable)
  1675. pageNum := 0
  1676. athenaErr := aws.QueryAthenaPaginated(context.TODO(), query, func(page *athena.GetQueryResultsOutput) bool {
  1677. if page == nil {
  1678. log.Errorf("fetchColumns: Athena page is nil")
  1679. return false
  1680. } else if page.ResultSet == nil {
  1681. log.Errorf("fetchColumns: Athena page.ResultSet is nil")
  1682. return false
  1683. }
  1684. // remove header row 'column_name'
  1685. rows := page.ResultSet.Rows[1:]
  1686. for _, row := range rows {
  1687. columnSet[*row.Data[0].VarCharValue] = true
  1688. }
  1689. pageNum++
  1690. return true
  1691. })
  1692. if athenaErr != nil {
  1693. return columnSet, athenaErr
  1694. }
  1695. if len(columnSet) == 0 {
  1696. log.Infof("No columns retrieved from Athena")
  1697. }
  1698. return columnSet, nil
  1699. }
  1700. type spotInfo struct {
  1701. Timestamp string `csv:"Timestamp"`
  1702. UsageType string `csv:"UsageType"`
  1703. Operation string `csv:"Operation"`
  1704. InstanceID string `csv:"InstanceID"`
  1705. MyBidID string `csv:"MyBidID"`
  1706. MyMaxPrice string `csv:"MyMaxPrice"`
  1707. MarketPrice string `csv:"MarketPrice"`
  1708. Charge string `csv:"Charge"`
  1709. Version string `csv:"Version"`
  1710. }
  1711. func (aws *AWS) parseSpotData(bucket string, prefix string, projectID string, region string) (map[string]*spotInfo, error) {
  1712. aws.ConfigureAuth() // configure aws api authentication by setting env vars
  1713. s3Prefix := projectID
  1714. if len(prefix) != 0 {
  1715. s3Prefix = prefix + "/" + s3Prefix
  1716. }
  1717. aak, err := aws.GetAWSAccessKey()
  1718. if err != nil {
  1719. return nil, err
  1720. }
  1721. cfg, err := aak.CreateConfig(region)
  1722. if err != nil {
  1723. return nil, err
  1724. }
  1725. cli := s3.NewFromConfig(cfg)
  1726. downloader := manager.NewDownloader(cli)
  1727. tNow := time.Now()
  1728. tOneDayAgo := tNow.Add(time.Duration(-24) * time.Hour) // Also get files from one day ago to avoid boundary conditions
  1729. ls := &s3.ListObjectsInput{
  1730. Bucket: awsSDK.String(bucket),
  1731. Prefix: awsSDK.String(s3Prefix + "." + tOneDayAgo.Format("2006-01-02")),
  1732. }
  1733. ls2 := &s3.ListObjectsInput{
  1734. Bucket: awsSDK.String(bucket),
  1735. Prefix: awsSDK.String(s3Prefix + "." + tNow.Format("2006-01-02")),
  1736. }
  1737. lso, err := cli.ListObjects(context.TODO(), ls)
  1738. if err != nil {
  1739. aws.serviceAccountChecks.set("bucketList", &ServiceAccountCheck{
  1740. Message: "Bucket List Permissions Available",
  1741. Status: false,
  1742. AdditionalInfo: err.Error(),
  1743. })
  1744. return nil, err
  1745. } else {
  1746. aws.serviceAccountChecks.set("bucketList", &ServiceAccountCheck{
  1747. Message: "Bucket List Permissions Available",
  1748. Status: true,
  1749. })
  1750. }
  1751. lsoLen := len(lso.Contents)
  1752. klog.V(2).Infof("Found %d spot data files from yesterday", lsoLen)
  1753. if lsoLen == 0 {
  1754. klog.V(5).Infof("ListObjects \"s3://%s/%s\" produced no keys", *ls.Bucket, *ls.Prefix)
  1755. }
  1756. lso2, err := cli.ListObjects(context.TODO(), ls2)
  1757. if err != nil {
  1758. return nil, err
  1759. }
  1760. lso2Len := len(lso2.Contents)
  1761. klog.V(2).Infof("Found %d spot data files from today", lso2Len)
  1762. if lso2Len == 0 {
  1763. klog.V(5).Infof("ListObjects \"s3://%s/%s\" produced no keys", *ls2.Bucket, *ls2.Prefix)
  1764. }
  1765. // TODO: Worth it to use LastModifiedDate to determine if we should reparse the spot data?
  1766. var keys []*string
  1767. for _, obj := range lso.Contents {
  1768. keys = append(keys, obj.Key)
  1769. }
  1770. for _, obj := range lso2.Contents {
  1771. keys = append(keys, obj.Key)
  1772. }
  1773. versionRx := regexp.MustCompile("^#Version: (\\d+)\\.\\d+$")
  1774. header, err := csvutil.Header(spotInfo{}, "csv")
  1775. if err != nil {
  1776. return nil, err
  1777. }
  1778. fieldsPerRecord := len(header)
  1779. spots := make(map[string]*spotInfo)
  1780. for _, key := range keys {
  1781. getObj := &s3.GetObjectInput{
  1782. Bucket: awsSDK.String(bucket),
  1783. Key: key,
  1784. }
  1785. buf := manager.NewWriteAtBuffer([]byte{})
  1786. _, err := downloader.Download(context.TODO(), buf, getObj)
  1787. if err != nil {
  1788. aws.serviceAccountChecks.set("objectList", &ServiceAccountCheck{
  1789. Message: "Object Get Permissions Available",
  1790. Status: false,
  1791. AdditionalInfo: err.Error(),
  1792. })
  1793. return nil, err
  1794. } else {
  1795. aws.serviceAccountChecks.set("objectList", &ServiceAccountCheck{
  1796. Message: "Object Get Permissions Available",
  1797. Status: true,
  1798. })
  1799. }
  1800. r := bytes.NewReader(buf.Bytes())
  1801. gr, err := gzip.NewReader(r)
  1802. if err != nil {
  1803. return nil, err
  1804. }
  1805. csvReader := csv.NewReader(gr)
  1806. csvReader.Comma = '\t'
  1807. csvReader.FieldsPerRecord = fieldsPerRecord
  1808. dec, err := csvutil.NewDecoder(csvReader, header...)
  1809. if err != nil {
  1810. return nil, err
  1811. }
  1812. var foundVersion string
  1813. for {
  1814. spot := spotInfo{}
  1815. err := dec.Decode(&spot)
  1816. csvParseErr, isCsvParseErr := err.(*csv.ParseError)
  1817. if err == io.EOF {
  1818. break
  1819. } else if err == csvutil.ErrFieldCount || (isCsvParseErr && csvParseErr.Err == csv.ErrFieldCount) {
  1820. rec := dec.Record()
  1821. // the first two "Record()" will be the comment lines
  1822. // and they show up as len() == 1
  1823. // the first of which is "#Version"
  1824. // the second of which is "#Fields: "
  1825. if len(rec) != 1 {
  1826. klog.V(2).Infof("Expected %d spot info fields but received %d: %s", fieldsPerRecord, len(rec), rec)
  1827. continue
  1828. }
  1829. if len(foundVersion) == 0 {
  1830. spotFeedVersion := rec[0]
  1831. klog.V(4).Infof("Spot feed version is \"%s\"", spotFeedVersion)
  1832. matches := versionRx.FindStringSubmatch(spotFeedVersion)
  1833. if matches != nil {
  1834. foundVersion = matches[1]
  1835. if foundVersion != supportedSpotFeedVersion {
  1836. klog.V(2).Infof("Unsupported spot info feed version: wanted \"%s\" got \"%s\"", supportedSpotFeedVersion, foundVersion)
  1837. break
  1838. }
  1839. }
  1840. continue
  1841. } else if strings.Index(rec[0], "#") == 0 {
  1842. continue
  1843. } else {
  1844. klog.V(3).Infof("skipping non-TSV line: %s", rec)
  1845. continue
  1846. }
  1847. } else if err != nil {
  1848. klog.V(2).Infof("Error during spot info decode: %+v", err)
  1849. continue
  1850. }
  1851. log.DedupedInfof(5, "Found spot info for: %s", spot.InstanceID)
  1852. spots[spot.InstanceID] = &spot
  1853. }
  1854. gr.Close()
  1855. }
  1856. return spots, nil
  1857. }
  1858. // ApplyReservedInstancePricing TODO
  1859. func (aws *AWS) ApplyReservedInstancePricing(nodes map[string]*Node) {
  1860. }
  1861. func (aws *AWS) ServiceAccountStatus() *ServiceAccountStatus {
  1862. return aws.serviceAccountChecks.getStatus()
  1863. }
  1864. func (aws *AWS) CombinedDiscountForNode(instanceType string, isPreemptible bool, defaultDiscount, negotiatedDiscount float64) float64 {
  1865. return 1.0 - ((1.0 - defaultDiscount) * (1.0 - negotiatedDiscount))
  1866. }
  1867. // Regions returns a predefined list of AWS regions
  1868. func (aws *AWS) Regions() []string {
  1869. return awsRegions
  1870. }