awsprovider.go 63 KB

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