awsprovider.go 42 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445
  1. package cloud
  2. import (
  3. "bytes"
  4. "compress/gzip"
  5. "encoding/csv"
  6. "encoding/json"
  7. "fmt"
  8. "io"
  9. "io/ioutil"
  10. "net/http"
  11. "net/url"
  12. "os"
  13. "regexp"
  14. "strconv"
  15. "strings"
  16. "sync"
  17. "time"
  18. "k8s.io/klog"
  19. "github.com/kubecost/cost-model/clustercache"
  20. "github.com/aws/aws-sdk-go/aws"
  21. "github.com/aws/aws-sdk-go/aws/awserr"
  22. "github.com/aws/aws-sdk-go/aws/session"
  23. "github.com/aws/aws-sdk-go/service/athena"
  24. "github.com/aws/aws-sdk-go/service/ec2"
  25. "github.com/aws/aws-sdk-go/service/s3"
  26. "github.com/aws/aws-sdk-go/service/s3/s3manager"
  27. "github.com/jszwec/csvutil"
  28. v1 "k8s.io/api/core/v1"
  29. )
  30. const awsAccessKeyIDEnvVar = "AWS_ACCESS_KEY_ID"
  31. const awsAccessKeySecretEnvVar = "AWS_SECRET_ACCESS_KEY"
  32. const supportedSpotFeedVersion = "1"
  33. const SpotInfoUpdateType = "spotinfo"
  34. const AthenaInfoUpdateType = "athenainfo"
  35. // AWS represents an Amazon Provider
  36. type AWS struct {
  37. Pricing map[string]*AWSProductTerms
  38. SpotPricingByInstanceID map[string]*spotInfo
  39. ValidPricingKeys map[string]bool
  40. Clientset clustercache.ClusterCache
  41. BaseCPUPrice string
  42. BaseRAMPrice string
  43. BaseGPUPrice string
  44. BaseSpotCPUPrice string
  45. BaseSpotRAMPrice string
  46. SpotLabelName string
  47. SpotLabelValue string
  48. ServiceKeyName string
  49. ServiceKeySecret string
  50. SpotDataRegion string
  51. SpotDataBucket string
  52. SpotDataPrefix string
  53. ProjectID string
  54. DownloadPricingDataLock sync.RWMutex
  55. *CustomProvider
  56. }
  57. // AWSPricing maps a k8s node to an AWS Pricing "product"
  58. type AWSPricing struct {
  59. Products map[string]*AWSProduct `json:"products"`
  60. Terms AWSPricingTerms `json:"terms"`
  61. }
  62. // AWSProduct represents a purchased SKU
  63. type AWSProduct struct {
  64. Sku string `json:"sku"`
  65. Attributes AWSProductAttributes `json:"attributes"`
  66. }
  67. // AWSProductAttributes represents metadata about the product used to map to a node.
  68. type AWSProductAttributes struct {
  69. Location string `json:"location"`
  70. InstanceType string `json:"instanceType"`
  71. Memory string `json:"memory"`
  72. Storage string `json:"storage"`
  73. VCpu string `json:"vcpu"`
  74. UsageType string `json:"usagetype"`
  75. OperatingSystem string `json:"operatingSystem"`
  76. PreInstalledSw string `json:"preInstalledSw"`
  77. InstanceFamily string `json:"instanceFamily"`
  78. GPU string `json:"gpu"` // GPU represents the number of GPU on the instance
  79. }
  80. // AWSPricingTerms are how you pay for the node: OnDemand, Reserved, or (TODO) Spot
  81. type AWSPricingTerms struct {
  82. OnDemand map[string]map[string]*AWSOfferTerm `json:"OnDemand"`
  83. Reserved map[string]map[string]*AWSOfferTerm `json:"Reserved"`
  84. }
  85. // AWSOfferTerm is a sku extension used to pay for the node.
  86. type AWSOfferTerm struct {
  87. Sku string `json:"sku"`
  88. PriceDimensions map[string]*AWSRateCode `json:"priceDimensions"`
  89. }
  90. // AWSRateCode encodes data about the price of a product
  91. type AWSRateCode struct {
  92. Unit string `json:"unit"`
  93. PricePerUnit AWSCurrencyCode `json:"pricePerUnit"`
  94. }
  95. // AWSCurrencyCode is the localized currency. (TODO: support non-USD)
  96. type AWSCurrencyCode struct {
  97. USD string `json:"USD"`
  98. }
  99. // AWSProductTerms represents the full terms of the product
  100. type AWSProductTerms struct {
  101. Sku string `json:"sku"`
  102. OnDemand *AWSOfferTerm `json:"OnDemand"`
  103. Reserved *AWSOfferTerm `json:"Reserved"`
  104. Memory string `json:"memory"`
  105. Storage string `json:"storage"`
  106. VCpu string `json:"vcpu"`
  107. GPU string `json:"gpu"` // GPU represents the number of GPU on the instance
  108. PV *PV `json:"pv"`
  109. }
  110. // ClusterIdEnvVar is the environment variable in which one can manually set the ClusterId
  111. const ClusterIdEnvVar = "AWS_CLUSTER_ID"
  112. // OnDemandRateCode is appended to an node sku
  113. const OnDemandRateCode = ".JRTCKXETXF"
  114. // ReservedRateCode is appended to a node sku
  115. const ReservedRateCode = ".38NPMPTW36"
  116. // HourlyRateCode is appended to a node sku
  117. const HourlyRateCode = ".6YS6EN2CT7"
  118. // volTypes are used to map between AWS UsageTypes and
  119. // EBS volume types, as they would appear in K8s storage class
  120. // name and the EC2 API.
  121. var volTypes = map[string]string{
  122. "EBS:VolumeUsage.gp2": "gp2",
  123. "EBS:VolumeUsage": "standard",
  124. "EBS:VolumeUsage.sc1": "sc1",
  125. "EBS:VolumeP-IOPS.piops": "io1",
  126. "EBS:VolumeUsage.st1": "st1",
  127. "EBS:VolumeUsage.piops": "io1",
  128. "gp2": "EBS:VolumeUsage.gp2",
  129. "standard": "EBS:VolumeUsage",
  130. "sc1": "EBS:VolumeUsage.sc1",
  131. "io1": "EBS:VolumeUsage.piops",
  132. "st1": "EBS:VolumeUsage.st1",
  133. }
  134. // locationToRegion maps AWS region names (As they come from Billing)
  135. // to actual region identifiers
  136. var locationToRegion = map[string]string{
  137. "US East (Ohio)": "us-east-2",
  138. "US East (N. Virginia)": "us-east-1",
  139. "US West (N. California)": "us-west-1",
  140. "US West (Oregon)": "us-west-2",
  141. "Asia Pacific (Hong Kong)": "ap-east-1",
  142. "Asia Pacific (Mumbai)": "ap-south-1",
  143. "Asia Pacific (Osaka-Local)": "ap-northeast-3",
  144. "Asia Pacific (Seoul)": "ap-northeast-2",
  145. "Asia Pacific (Singapore)": "ap-southeast-1",
  146. "Asia Pacific (Sydney)": "ap-southeast-2",
  147. "Asia Pacific (Tokyo)": "ap-northeast-1",
  148. "Canada (Central)": "ca-central-1",
  149. "China (Beijing)": "cn-north-1",
  150. "China (Ningxia)": "cn-northwest-1",
  151. "EU (Frankfurt)": "eu-central-1",
  152. "EU (Ireland)": "eu-west-1",
  153. "EU (London)": "eu-west-2",
  154. "EU (Paris)": "eu-west-3",
  155. "EU (Stockholm)": "eu-north-1",
  156. "South America (Sao Paulo)": "sa-east-1",
  157. "AWS GovCloud (US-East)": "us-gov-east-1",
  158. "AWS GovCloud (US)": "us-gov-west-1",
  159. }
  160. var regionToBillingRegionCode = map[string]string{
  161. "us-east-2": "USE2",
  162. "us-east-1": "",
  163. "us-west-1": "USW1",
  164. "us-west-2": "USW2",
  165. "ap-east-1": "APE1",
  166. "ap-south-1": "APS3",
  167. "ap-northeast-3": "APN3",
  168. "ap-northeast-2": "APN2",
  169. "ap-southeast-1": "APS1",
  170. "ap-southeast-2": "APS2",
  171. "ap-northeast-1": "APN1",
  172. "ca-central-1": "CAN1",
  173. "cn-north-1": "",
  174. "cn-northwest-1": "",
  175. "eu-central-1": "EUC1",
  176. "eu-west-1": "EU",
  177. "eu-west-2": "EUW2",
  178. "eu-west-3": "EUW3",
  179. "eu-north-1": "EUN1",
  180. "sa-east-1": "SAE1",
  181. "us-gov-east-1": "UGE1",
  182. "us-gov-west-1": "UGW1",
  183. }
  184. func (aws *AWS) GetLocalStorageQuery(offset string) (string, error) {
  185. return "", nil
  186. }
  187. // KubeAttrConversion maps the k8s labels for region to an aws region
  188. func (aws *AWS) KubeAttrConversion(location, instanceType, operatingSystem string) string {
  189. operatingSystem = strings.ToLower(operatingSystem)
  190. region := locationToRegion[location]
  191. return region + "," + instanceType + "," + operatingSystem
  192. }
  193. type AwsSpotFeedInfo struct {
  194. BucketName string `json:"bucketName"`
  195. Prefix string `json:"prefix"`
  196. Region string `json:"region"`
  197. AccountID string `json:"projectID"`
  198. ServiceKeyName string `json:"serviceKeyName"`
  199. ServiceKeySecret string `json:"serviceKeySecret"`
  200. SpotLabel string `json:"spotLabel"`
  201. SpotLabelValue string `json:"spotLabelValue"`
  202. }
  203. type AwsAthenaInfo struct {
  204. AthenaBucketName string `json:"athenaBucketName"`
  205. AthenaRegion string `json:"athenaRegion"`
  206. AthenaDatabase string `json:"athenaDatabase"`
  207. AthenaTable string `json:"athenaTable"`
  208. ServiceKeyName string `json:"serviceKeyName"`
  209. ServiceKeySecret string `json:"serviceKeySecret"`
  210. AccountID string `json:"projectID"`
  211. }
  212. func (aws *AWS) GetManagementPlatform() (string, error) {
  213. nodes := aws.Clientset.GetAllNodes()
  214. if len(nodes) > 0 {
  215. n := nodes[0]
  216. version := n.Status.NodeInfo.KubeletVersion
  217. if strings.Contains(version, "eks") {
  218. return "eks", nil
  219. }
  220. if _, ok := n.Labels["kops.k8s.io/instancegroup"]; ok {
  221. return "kops", nil
  222. }
  223. }
  224. return "", nil
  225. }
  226. func (aws *AWS) GetConfig() (*CustomPricing, error) {
  227. c, err := GetDefaultPricingData("aws.json")
  228. if c.Discount == "" {
  229. c.Discount = "0%"
  230. }
  231. if c.NegotiatedDiscount == "" {
  232. c.NegotiatedDiscount = "0%"
  233. }
  234. if err != nil {
  235. return nil, err
  236. }
  237. return c, nil
  238. }
  239. func (aws *AWS) UpdateConfig(r io.Reader, updateType string) (*CustomPricing, error) {
  240. c, err := GetDefaultPricingData("aws.json")
  241. if err != nil {
  242. return nil, err
  243. }
  244. if updateType == SpotInfoUpdateType {
  245. a := AwsSpotFeedInfo{}
  246. err := json.NewDecoder(r).Decode(&a)
  247. if err != nil {
  248. return nil, err
  249. }
  250. if err != nil {
  251. return nil, err
  252. }
  253. c.ServiceKeyName = a.ServiceKeyName
  254. c.ServiceKeySecret = a.ServiceKeySecret
  255. c.SpotDataPrefix = a.Prefix
  256. c.SpotDataBucket = a.BucketName
  257. c.ProjectID = a.AccountID
  258. c.SpotDataRegion = a.Region
  259. c.SpotLabel = a.SpotLabel
  260. c.SpotLabelValue = a.SpotLabelValue
  261. } else if updateType == AthenaInfoUpdateType {
  262. a := AwsAthenaInfo{}
  263. err := json.NewDecoder(r).Decode(&a)
  264. if err != nil {
  265. return nil, err
  266. }
  267. c.AthenaBucketName = a.AthenaBucketName
  268. c.AthenaRegion = a.AthenaRegion
  269. c.AthenaDatabase = a.AthenaDatabase
  270. c.AthenaTable = a.AthenaTable
  271. c.ServiceKeyName = a.ServiceKeyName
  272. c.ServiceKeySecret = a.ServiceKeySecret
  273. c.ProjectID = a.AccountID
  274. } else {
  275. a := make(map[string]string)
  276. err = json.NewDecoder(r).Decode(&a)
  277. if err != nil {
  278. return nil, err
  279. }
  280. for k, v := range a {
  281. kUpper := strings.Title(k) // Just so we consistently supply / receive the same values, uppercase the first letter.
  282. err := SetCustomPricingField(c, kUpper, v)
  283. if err != nil {
  284. return nil, err
  285. }
  286. }
  287. }
  288. cj, err := json.Marshal(c)
  289. if err != nil {
  290. return nil, err
  291. }
  292. path := os.Getenv("CONFIG_PATH")
  293. if path == "" {
  294. path = "/models/"
  295. }
  296. path += "aws.json"
  297. remoteEnabled := os.Getenv(remoteEnabled)
  298. if remoteEnabled == "true" {
  299. err = UpdateClusterMeta(os.Getenv(clusterIDKey), c.ClusterName)
  300. if err != nil {
  301. return nil, err
  302. }
  303. }
  304. err = ioutil.WriteFile(path, cj, 0644)
  305. if err != nil {
  306. return nil, err
  307. }
  308. return c, nil
  309. }
  310. type awsKey struct {
  311. SpotLabelName string
  312. SpotLabelValue string
  313. Labels map[string]string
  314. ProviderID string
  315. }
  316. func (k *awsKey) GPUType() string {
  317. return ""
  318. }
  319. func (k *awsKey) ID() string {
  320. provIdRx := regexp.MustCompile("aws:///([^/]+)/([^/]+)") // It's of the form aws:///us-east-2a/i-0fea4fd46592d050b and we want i-0fea4fd46592d050b, if it exists
  321. for matchNum, group := range provIdRx.FindStringSubmatch(k.ProviderID) {
  322. if matchNum == 2 {
  323. return group
  324. }
  325. }
  326. klog.V(3).Infof("Could not find instance ID in \"%s\"", k.ProviderID)
  327. return ""
  328. }
  329. func (k *awsKey) Features() string {
  330. instanceType := k.Labels[v1.LabelInstanceType]
  331. var operatingSystem string
  332. operatingSystem, ok := k.Labels[v1.LabelOSStable]
  333. if !ok {
  334. operatingSystem = k.Labels["beta.kubernetes.io/os"]
  335. }
  336. region := k.Labels[v1.LabelZoneRegion]
  337. key := region + "," + instanceType + "," + operatingSystem
  338. usageType := "preemptible"
  339. spotKey := key + "," + usageType
  340. if l, ok := k.Labels["lifecycle"]; ok && l == "EC2Spot" {
  341. return spotKey
  342. }
  343. if l, ok := k.Labels[k.SpotLabelName]; ok && l == k.SpotLabelValue {
  344. return spotKey
  345. }
  346. return key
  347. }
  348. func (aws *AWS) PVPricing(pvk PVKey) (*PV, error) {
  349. pricing, ok := aws.Pricing[pvk.Features()]
  350. if !ok {
  351. klog.V(4).Infof("Persistent Volume pricing not found for %s: %s", pvk.GetStorageClass(), pvk.Features())
  352. return &PV{}, nil
  353. }
  354. return pricing.PV, nil
  355. }
  356. type awsPVKey struct {
  357. Labels map[string]string
  358. StorageClassParameters map[string]string
  359. StorageClassName string
  360. Name string
  361. }
  362. func (aws *AWS) GetPVKey(pv *v1.PersistentVolume, parameters map[string]string) PVKey {
  363. return &awsPVKey{
  364. Labels: pv.Labels,
  365. StorageClassName: pv.Spec.StorageClassName,
  366. StorageClassParameters: parameters,
  367. Name: pv.Name,
  368. }
  369. }
  370. func (key *awsPVKey) GetStorageClass() string {
  371. return key.StorageClassName
  372. }
  373. func (key *awsPVKey) Features() string {
  374. storageClass := key.StorageClassParameters["type"]
  375. if storageClass == "standard" {
  376. storageClass = "gp2"
  377. }
  378. // Storage class names are generally EBS volume types (gp2)
  379. // Keys in Pricing are based on UsageTypes (EBS:VolumeType.gp2)
  380. // Converts between the 2
  381. region := key.Labels[v1.LabelZoneRegion]
  382. //if region == "" {
  383. // region = "us-east-1"
  384. //}
  385. class, ok := volTypes[storageClass]
  386. if !ok {
  387. klog.V(4).Infof("No voltype mapping for %s's storageClass: %s", key.Name, storageClass)
  388. }
  389. return region + "," + class
  390. }
  391. // GetKey maps node labels to information needed to retrieve pricing data
  392. func (aws *AWS) GetKey(labels map[string]string) Key {
  393. return &awsKey{
  394. SpotLabelName: aws.SpotLabelName,
  395. SpotLabelValue: aws.SpotLabelValue,
  396. Labels: labels,
  397. ProviderID: labels["providerID"],
  398. }
  399. }
  400. func (aws *AWS) isPreemptible(key string) bool {
  401. s := strings.Split(key, ",")
  402. if len(s) == 4 && s[3] == "preemptible" {
  403. return true
  404. }
  405. return false
  406. }
  407. // DownloadPricingData fetches data from the AWS Pricing API
  408. func (aws *AWS) DownloadPricingData() error {
  409. aws.DownloadPricingDataLock.Lock()
  410. defer aws.DownloadPricingDataLock.Unlock()
  411. c, err := GetDefaultPricingData("aws.json")
  412. if err != nil {
  413. klog.V(1).Infof("Error downloading default pricing data: %s", err.Error())
  414. }
  415. aws.BaseCPUPrice = c.CPU
  416. aws.BaseRAMPrice = c.RAM
  417. aws.BaseGPUPrice = c.GPU
  418. aws.BaseSpotCPUPrice = c.SpotCPU
  419. aws.BaseSpotRAMPrice = c.SpotRAM
  420. aws.SpotLabelName = c.SpotLabel
  421. aws.SpotLabelValue = c.SpotLabelValue
  422. aws.SpotDataBucket = c.SpotDataBucket
  423. aws.SpotDataPrefix = c.SpotDataPrefix
  424. aws.ProjectID = c.ProjectID
  425. aws.SpotDataRegion = c.SpotDataRegion
  426. aws.ServiceKeyName = c.ServiceKeyName
  427. aws.ServiceKeySecret = c.ServiceKeySecret
  428. if len(aws.SpotDataBucket) != 0 && len(aws.ProjectID) == 0 {
  429. klog.V(1).Infof("using SpotDataBucket \"%s\" without ProjectID will not end well", aws.SpotDataBucket)
  430. }
  431. nodeList := aws.Clientset.GetAllNodes()
  432. inputkeys := make(map[string]bool)
  433. for _, n := range nodeList {
  434. labels := n.GetObjectMeta().GetLabels()
  435. key := aws.GetKey(labels)
  436. inputkeys[key.Features()] = true
  437. }
  438. pvList := aws.Clientset.GetAllPersistentVolumes()
  439. storageClasses := aws.Clientset.GetAllStorageClasses()
  440. storageClassMap := make(map[string]map[string]string)
  441. for _, storageClass := range storageClasses {
  442. params := storageClass.Parameters
  443. storageClassMap[storageClass.ObjectMeta.Name] = params
  444. if storageClass.GetAnnotations()["storageclass.kubernetes.io/is-default-class"] == "true" || storageClass.GetAnnotations()["storageclass.beta.kubernetes.io/is-default-class"] == "true" {
  445. storageClassMap["default"] = params
  446. storageClassMap[""] = params
  447. }
  448. }
  449. pvkeys := make(map[string]PVKey)
  450. for _, pv := range pvList {
  451. params, ok := storageClassMap[pv.Spec.StorageClassName]
  452. if !ok {
  453. klog.V(2).Infof("Unable to find params for storageClassName %s, falling back to default pricing", pv.Spec.StorageClassName)
  454. continue
  455. }
  456. key := aws.GetPVKey(pv, params)
  457. pvkeys[key.Features()] = key
  458. }
  459. aws.Pricing = make(map[string]*AWSProductTerms)
  460. aws.ValidPricingKeys = make(map[string]bool)
  461. skusToKeys := make(map[string]string)
  462. pricingURL := "https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonEC2/current/index.json"
  463. klog.V(2).Infof("starting download of \"%s\", which is quite large ...", pricingURL)
  464. resp, err := http.Get(pricingURL)
  465. if err != nil {
  466. klog.V(2).Infof("Bogus fetch of \"%s\": %v", pricingURL, err)
  467. return err
  468. }
  469. klog.V(2).Infof("Finished downloading \"%s\"", pricingURL)
  470. dec := json.NewDecoder(resp.Body)
  471. for {
  472. t, err := dec.Token()
  473. if err == io.EOF {
  474. klog.V(2).Infof("done loading \"%s\"\n", pricingURL)
  475. break
  476. }
  477. if t == "products" {
  478. _, err := dec.Token() // this should parse the opening "{""
  479. if err != nil {
  480. return err
  481. }
  482. for dec.More() {
  483. _, err := dec.Token() // the sku token
  484. if err != nil {
  485. return err
  486. }
  487. product := &AWSProduct{}
  488. err = dec.Decode(&product)
  489. if err != nil {
  490. klog.V(1).Infof("Error parsing response from \"%s\": %v", pricingURL, err.Error())
  491. break
  492. }
  493. if product.Attributes.PreInstalledSw == "NA" &&
  494. (strings.HasPrefix(product.Attributes.UsageType, "BoxUsage") || strings.Contains(product.Attributes.UsageType, "-BoxUsage")) {
  495. key := aws.KubeAttrConversion(product.Attributes.Location, product.Attributes.InstanceType, product.Attributes.OperatingSystem)
  496. spotKey := key + ",preemptible"
  497. if inputkeys[key] || inputkeys[spotKey] { // Just grab the sku even if spot, and change the price later.
  498. productTerms := &AWSProductTerms{
  499. Sku: product.Sku,
  500. Memory: product.Attributes.Memory,
  501. Storage: product.Attributes.Storage,
  502. VCpu: product.Attributes.VCpu,
  503. GPU: product.Attributes.GPU,
  504. }
  505. aws.Pricing[key] = productTerms
  506. aws.Pricing[spotKey] = productTerms
  507. skusToKeys[product.Sku] = key
  508. }
  509. aws.ValidPricingKeys[key] = true
  510. aws.ValidPricingKeys[spotKey] = true
  511. } else if strings.Contains(product.Attributes.UsageType, "EBS:Volume") {
  512. // UsageTypes may be prefixed with a region code - we're removing this when using
  513. // volTypes to keep lookups generic
  514. usageTypeRegx := regexp.MustCompile(".*(-|^)(EBS.+)")
  515. usageTypeMatch := usageTypeRegx.FindStringSubmatch(product.Attributes.UsageType)
  516. usageTypeNoRegion := usageTypeMatch[len(usageTypeMatch)-1]
  517. key := locationToRegion[product.Attributes.Location] + "," + usageTypeNoRegion
  518. spotKey := key + ",preemptible"
  519. pv := &PV{
  520. Class: volTypes[usageTypeNoRegion],
  521. Region: locationToRegion[product.Attributes.Location],
  522. }
  523. productTerms := &AWSProductTerms{
  524. Sku: product.Sku,
  525. PV: pv,
  526. }
  527. aws.Pricing[key] = productTerms
  528. aws.Pricing[spotKey] = productTerms
  529. skusToKeys[product.Sku] = key
  530. aws.ValidPricingKeys[key] = true
  531. aws.ValidPricingKeys[spotKey] = true
  532. }
  533. }
  534. }
  535. if t == "terms" {
  536. _, err := dec.Token() // this should parse the opening "{""
  537. if err != nil {
  538. return err
  539. }
  540. termType, err := dec.Token()
  541. if err != nil {
  542. return err
  543. }
  544. if termType == "OnDemand" {
  545. _, err := dec.Token()
  546. if err != nil { // again, should parse an opening "{"
  547. return err
  548. }
  549. for dec.More() {
  550. sku, err := dec.Token()
  551. if err != nil {
  552. return err
  553. }
  554. _, err = dec.Token() // another opening "{"
  555. if err != nil {
  556. return err
  557. }
  558. skuOnDemand, err := dec.Token()
  559. if err != nil {
  560. return err
  561. }
  562. offerTerm := &AWSOfferTerm{}
  563. err = dec.Decode(&offerTerm)
  564. if err != nil {
  565. klog.V(1).Infof("Error decoding AWS Offer Term: " + err.Error())
  566. }
  567. if sku.(string)+OnDemandRateCode == skuOnDemand {
  568. key, ok := skusToKeys[sku.(string)]
  569. spotKey := key + ",preemptible"
  570. if ok {
  571. aws.Pricing[key].OnDemand = offerTerm
  572. aws.Pricing[spotKey].OnDemand = offerTerm
  573. if strings.Contains(key, "EBS:VolumeP-IOPS.piops") {
  574. // If the specific UsageType is the per IO cost used on io1 volumes
  575. // we need to add the per IO cost to the io1 PV cost
  576. cost := offerTerm.PriceDimensions[sku.(string)+OnDemandRateCode+HourlyRateCode].PricePerUnit.USD
  577. // Add the per IO cost to the PV object for the io1 volume type
  578. aws.Pricing[key].PV.CostPerIO = cost
  579. } else if strings.Contains(key, "EBS:Volume") {
  580. // If volume, we need to get hourly cost and add it to the PV object
  581. cost := offerTerm.PriceDimensions[sku.(string)+OnDemandRateCode+HourlyRateCode].PricePerUnit.USD
  582. costFloat, _ := strconv.ParseFloat(cost, 64)
  583. hourlyPrice := costFloat / 730
  584. aws.Pricing[key].PV.Cost = strconv.FormatFloat(hourlyPrice, 'f', -1, 64)
  585. }
  586. }
  587. }
  588. _, err = dec.Token()
  589. if err != nil {
  590. return err
  591. }
  592. }
  593. _, err = dec.Token()
  594. if err != nil {
  595. return err
  596. }
  597. }
  598. }
  599. }
  600. sp, err := parseSpotData(aws.SpotDataBucket, aws.SpotDataPrefix, aws.ProjectID, aws.SpotDataRegion, aws.ServiceKeyName, aws.ServiceKeySecret)
  601. if err != nil {
  602. klog.V(1).Infof("Skipping AWS spot data download: %s", err.Error())
  603. } else {
  604. aws.SpotPricingByInstanceID = sp
  605. }
  606. return nil
  607. }
  608. // Stubbed NetworkPricing for AWS. Pull directly from aws.json for now
  609. func (c *AWS) NetworkPricing() (*Network, error) {
  610. cpricing, err := GetDefaultPricingData("aws.json")
  611. if err != nil {
  612. return nil, err
  613. }
  614. znec, err := strconv.ParseFloat(cpricing.ZoneNetworkEgress, 64)
  615. if err != nil {
  616. return nil, err
  617. }
  618. rnec, err := strconv.ParseFloat(cpricing.RegionNetworkEgress, 64)
  619. if err != nil {
  620. return nil, err
  621. }
  622. inec, err := strconv.ParseFloat(cpricing.InternetNetworkEgress, 64)
  623. if err != nil {
  624. return nil, err
  625. }
  626. return &Network{
  627. ZoneNetworkEgressCost: znec,
  628. RegionNetworkEgressCost: rnec,
  629. InternetNetworkEgressCost: inec,
  630. }, nil
  631. }
  632. // AllNodePricing returns all the billing data fetched.
  633. func (aws *AWS) AllNodePricing() (interface{}, error) {
  634. aws.DownloadPricingDataLock.RLock()
  635. defer aws.DownloadPricingDataLock.RUnlock()
  636. return aws.Pricing, nil
  637. }
  638. func (aws *AWS) createNode(terms *AWSProductTerms, usageType string, k Key) (*Node, error) {
  639. key := k.Features()
  640. if aws.isPreemptible(key) {
  641. if spotInfo, ok := aws.SpotPricingByInstanceID[k.ID()]; ok { // try and match directly to an ID for pricing. We'll still need the features
  642. var spotcost string
  643. arr := strings.Split(spotInfo.Charge, " ")
  644. if len(arr) == 2 {
  645. spotcost = arr[0]
  646. } else {
  647. klog.V(2).Infof("Spot data for node %s is missing", k.ID())
  648. }
  649. return &Node{
  650. Cost: spotcost,
  651. VCPU: terms.VCpu,
  652. RAM: terms.Memory,
  653. GPU: terms.GPU,
  654. Storage: terms.Storage,
  655. BaseCPUPrice: aws.BaseCPUPrice,
  656. BaseRAMPrice: aws.BaseRAMPrice,
  657. BaseGPUPrice: aws.BaseGPUPrice,
  658. UsageType: usageType,
  659. }, nil
  660. }
  661. return &Node{
  662. VCPU: terms.VCpu,
  663. VCPUCost: aws.BaseSpotCPUPrice,
  664. RAM: terms.Memory,
  665. GPU: terms.GPU,
  666. RAMCost: aws.BaseSpotRAMPrice,
  667. Storage: terms.Storage,
  668. BaseCPUPrice: aws.BaseCPUPrice,
  669. BaseRAMPrice: aws.BaseRAMPrice,
  670. BaseGPUPrice: aws.BaseGPUPrice,
  671. UsageType: usageType,
  672. }, nil
  673. }
  674. c, ok := terms.OnDemand.PriceDimensions[terms.Sku+OnDemandRateCode+HourlyRateCode]
  675. if !ok {
  676. return nil, fmt.Errorf("Could not fetch data for \"%s\"", k.ID())
  677. }
  678. cost := c.PricePerUnit.USD
  679. return &Node{
  680. Cost: cost,
  681. VCPU: terms.VCpu,
  682. RAM: terms.Memory,
  683. GPU: terms.GPU,
  684. Storage: terms.Storage,
  685. BaseCPUPrice: aws.BaseCPUPrice,
  686. BaseRAMPrice: aws.BaseRAMPrice,
  687. BaseGPUPrice: aws.BaseGPUPrice,
  688. UsageType: usageType,
  689. }, nil
  690. }
  691. // NodePricing takes in a key from GetKey and returns a Node object for use in building the cost model.
  692. func (aws *AWS) NodePricing(k Key) (*Node, error) {
  693. aws.DownloadPricingDataLock.RLock()
  694. defer aws.DownloadPricingDataLock.RUnlock()
  695. key := k.Features()
  696. usageType := "ondemand"
  697. if aws.isPreemptible(key) {
  698. usageType = "preemptible"
  699. }
  700. terms, ok := aws.Pricing[key]
  701. if ok {
  702. return aws.createNode(terms, usageType, k)
  703. } else if _, ok := aws.ValidPricingKeys[key]; ok {
  704. aws.DownloadPricingDataLock.RUnlock()
  705. err := aws.DownloadPricingData()
  706. aws.DownloadPricingDataLock.RLock()
  707. if err != nil {
  708. return &Node{
  709. Cost: aws.BaseCPUPrice,
  710. BaseCPUPrice: aws.BaseCPUPrice,
  711. BaseRAMPrice: aws.BaseRAMPrice,
  712. BaseGPUPrice: aws.BaseGPUPrice,
  713. UsageType: usageType,
  714. UsesBaseCPUPrice: true,
  715. }, err
  716. }
  717. terms, termsOk := aws.Pricing[key]
  718. if !termsOk {
  719. return &Node{
  720. Cost: aws.BaseCPUPrice,
  721. BaseCPUPrice: aws.BaseCPUPrice,
  722. BaseRAMPrice: aws.BaseRAMPrice,
  723. BaseGPUPrice: aws.BaseGPUPrice,
  724. UsageType: usageType,
  725. UsesBaseCPUPrice: true,
  726. }, fmt.Errorf("Unable to find any Pricing data for \"%s\"", key)
  727. }
  728. return aws.createNode(terms, usageType, k)
  729. } else { // Fall back to base pricing if we can't find the key.
  730. klog.V(1).Infof("Invalid Pricing Key \"%s\"", key)
  731. return &Node{
  732. Cost: aws.BaseCPUPrice,
  733. BaseCPUPrice: aws.BaseCPUPrice,
  734. BaseRAMPrice: aws.BaseRAMPrice,
  735. BaseGPUPrice: aws.BaseGPUPrice,
  736. UsageType: usageType,
  737. UsesBaseCPUPrice: true,
  738. }, nil
  739. }
  740. }
  741. // ClusterInfo returns an object that represents the cluster. TODO: actually return the name of the cluster. Blocked on cluster federation.
  742. func (awsProvider *AWS) ClusterInfo() (map[string]string, error) {
  743. defaultClusterName := "AWS Cluster #1"
  744. c, err := awsProvider.GetConfig()
  745. if err != nil {
  746. return nil, err
  747. }
  748. remote := os.Getenv(remoteEnabled)
  749. remoteEnabled := false
  750. if os.Getenv(remote) == "true" {
  751. remoteEnabled = true
  752. }
  753. if c.ClusterName != "" {
  754. m := make(map[string]string)
  755. m["name"] = c.ClusterName
  756. m["provider"] = "AWS"
  757. m["id"] = os.Getenv(clusterIDKey)
  758. m["remoteReadEnabled"] = strconv.FormatBool(remoteEnabled)
  759. return m, nil
  760. }
  761. makeStructure := func(clusterName string) (map[string]string, error) {
  762. klog.V(2).Infof("Returning \"%s\" as ClusterName", clusterName)
  763. m := make(map[string]string)
  764. m["name"] = clusterName
  765. m["provider"] = "AWS"
  766. m["id"] = os.Getenv(clusterIDKey)
  767. m["remoteReadEnabled"] = strconv.FormatBool(remoteEnabled)
  768. return m, nil
  769. }
  770. maybeClusterId := os.Getenv(ClusterIdEnvVar)
  771. if len(maybeClusterId) != 0 {
  772. return makeStructure(maybeClusterId)
  773. }
  774. provIdRx := regexp.MustCompile("aws:///([^/]+)/([^/]+)")
  775. clusterIdRx := regexp.MustCompile("^kubernetes\\.io/cluster/([^/]+)")
  776. nodeList := awsProvider.Clientset.GetAllNodes()
  777. for _, n := range nodeList {
  778. region := ""
  779. instanceId := ""
  780. providerId := n.Spec.ProviderID
  781. for matchNum, group := range provIdRx.FindStringSubmatch(providerId) {
  782. if matchNum == 1 {
  783. region = group
  784. } else if matchNum == 2 {
  785. instanceId = group
  786. }
  787. }
  788. if len(instanceId) == 0 {
  789. klog.V(2).Infof("Unable to decode Node.ProviderID \"%s\", skipping it", providerId)
  790. continue
  791. }
  792. c := &aws.Config{
  793. Region: aws.String(region),
  794. }
  795. s := session.Must(session.NewSession(c))
  796. ec2Svc := ec2.New(s)
  797. di, diErr := ec2Svc.DescribeInstances(&ec2.DescribeInstancesInput{
  798. InstanceIds: []*string{
  799. aws.String(instanceId),
  800. },
  801. })
  802. if diErr != nil {
  803. // maybe log this?
  804. continue
  805. }
  806. if len(di.Reservations) != 1 {
  807. klog.V(2).Infof("Expected 1 Reservation back from DescribeInstances(%s), received %d", instanceId, len(di.Reservations))
  808. continue
  809. }
  810. res := di.Reservations[0]
  811. if len(res.Instances) != 1 {
  812. klog.V(2).Infof("Expected 1 Instance back from DescribeInstances(%s), received %d", instanceId, len(res.Instances))
  813. continue
  814. }
  815. inst := res.Instances[0]
  816. for _, tag := range inst.Tags {
  817. tagKey := *tag.Key
  818. for matchNum, group := range clusterIdRx.FindStringSubmatch(tagKey) {
  819. if matchNum != 1 {
  820. continue
  821. }
  822. return makeStructure(group)
  823. }
  824. }
  825. }
  826. klog.V(2).Infof("Unable to sniff out cluster ID, perhaps set $%s to force one", ClusterIdEnvVar)
  827. return makeStructure(defaultClusterName)
  828. }
  829. // AddServiceKey adds an AWS service key, useful for pulling down out-of-cluster costs. Optional-- the container this runs in can be directly authorized.
  830. func (*AWS) AddServiceKey(formValues url.Values) error {
  831. keyID := formValues.Get("access_key_ID")
  832. key := formValues.Get("secret_access_key")
  833. m := make(map[string]string)
  834. m["access_key_ID"] = keyID
  835. m["secret_access_key"] = key
  836. result, err := json.Marshal(m)
  837. if err != nil {
  838. return err
  839. }
  840. return ioutil.WriteFile("/var/configs/key.json", result, 0644)
  841. }
  842. // 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.
  843. func (*AWS) GetDisks() ([]byte, error) {
  844. jsonFile, err := os.Open("/var/configs/key.json")
  845. if err == nil {
  846. byteValue, _ := ioutil.ReadAll(jsonFile)
  847. var result map[string]string
  848. err := json.Unmarshal([]byte(byteValue), &result)
  849. if err != nil {
  850. return nil, err
  851. }
  852. err = os.Setenv(awsAccessKeyIDEnvVar, result["access_key_ID"])
  853. if err != nil {
  854. return nil, err
  855. }
  856. err = os.Setenv(awsAccessKeySecretEnvVar, result["secret_access_key"])
  857. if err != nil {
  858. return nil, err
  859. }
  860. } else if os.IsNotExist(err) {
  861. klog.V(2).Infof("Using Default Credentials")
  862. } else {
  863. return nil, err
  864. }
  865. defer jsonFile.Close()
  866. clusterConfig, err := os.Open("/var/configs/cluster.json")
  867. if err != nil {
  868. return nil, err
  869. }
  870. defer clusterConfig.Close()
  871. b, err := ioutil.ReadAll(clusterConfig)
  872. if err != nil {
  873. return nil, err
  874. }
  875. var clusterConf map[string]string
  876. err = json.Unmarshal([]byte(b), &clusterConf)
  877. if err != nil {
  878. return nil, err
  879. }
  880. region := aws.String(clusterConf["region"])
  881. c := &aws.Config{
  882. Region: region,
  883. }
  884. s := session.Must(session.NewSession(c))
  885. ec2Svc := ec2.New(s)
  886. input := &ec2.DescribeVolumesInput{}
  887. volumeResult, err := ec2Svc.DescribeVolumes(input)
  888. if err != nil {
  889. if aerr, ok := err.(awserr.Error); ok {
  890. switch aerr.Code() {
  891. default:
  892. return nil, aerr
  893. }
  894. } else {
  895. return nil, err
  896. }
  897. }
  898. return json.Marshal(volumeResult)
  899. }
  900. // ConvertToGlueColumnFormat takes a string and runs through various regex
  901. // and string replacement statements to convert it to a format compatible
  902. // with AWS Glue and Athena column names.
  903. // Following guidance from AWS provided here ('Column Names' section):
  904. // https://docs.aws.amazon.com/awsaccountbilling/latest/aboutv2/run-athena-sql.html
  905. // It returns a string containing the column name in proper column name format and length.
  906. func ConvertToGlueColumnFormat(column_name string) string {
  907. klog.V(5).Infof("Converting string \"%s\" to proper AWS Glue column name.", column_name)
  908. // An underscore is added in front of uppercase letters
  909. capital_underscore := regexp.MustCompile(`[A-Z]`)
  910. final := capital_underscore.ReplaceAllString(column_name, `_$0`)
  911. // Any non-alphanumeric characters are replaced with an underscore
  912. no_space_punc := regexp.MustCompile(`[\s]{1,}|[^A-Za-z0-9]`)
  913. final = no_space_punc.ReplaceAllString(final, "_")
  914. // Duplicate underscores are removed
  915. no_dup_underscore := regexp.MustCompile(`_{2,}`)
  916. final = no_dup_underscore.ReplaceAllString(final, "_")
  917. // Any leading and trailing underscores are removed
  918. no_front_end_underscore := regexp.MustCompile(`(^\_|\_$)`)
  919. final = no_front_end_underscore.ReplaceAllString(final, "")
  920. // Uppercase to lowercase
  921. final = strings.ToLower(final)
  922. // Longer column name than expected - remove _ left to right
  923. allowed_col_len := 128
  924. undersc_to_remove := len(final) - allowed_col_len
  925. if undersc_to_remove > 0 {
  926. final = strings.Replace(final, "_", "", undersc_to_remove)
  927. }
  928. // If removing all of the underscores still didn't
  929. // make the column name < 128 characters, trim it!
  930. if len(final) > allowed_col_len {
  931. final = final[:allowed_col_len]
  932. }
  933. klog.V(5).Infof("Column name being returned: \"%s\". Length: \"%d\".", final, len(final))
  934. return final
  935. }
  936. // ExternalAllocations represents tagged assets outside the scope of kubernetes.
  937. // "start" and "end" are dates of the format YYYY-MM-DD
  938. // "aggregator" is the tag used to determine how to allocate those assets, ie namespace, pod, etc.
  939. func (a *AWS) ExternalAllocations(start string, end string, aggregator string) ([]*OutOfClusterAllocation, error) {
  940. customPricing, err := a.GetConfig()
  941. if err != nil {
  942. return nil, err
  943. }
  944. aggregator_column_name := "resource_tags_user_" + aggregator
  945. aggregator_column_name = ConvertToGlueColumnFormat(aggregator_column_name)
  946. query := fmt.Sprintf(`SELECT
  947. CAST(line_item_usage_start_date AS DATE) as start_date,
  948. %s,
  949. line_item_product_code,
  950. SUM(line_item_blended_cost) as blended_cost
  951. FROM %s as cost_data
  952. WHERE line_item_usage_start_date BETWEEN date '%s' AND date '%s'
  953. GROUP BY 1,2,3`, aggregator_column_name, customPricing.AthenaTable, start, end)
  954. if customPricing.ServiceKeyName != "" {
  955. err = os.Setenv(awsAccessKeyIDEnvVar, customPricing.ServiceKeyName)
  956. if err != nil {
  957. return nil, err
  958. }
  959. err = os.Setenv(awsAccessKeySecretEnvVar, customPricing.ServiceKeySecret)
  960. if err != nil {
  961. return nil, err
  962. }
  963. }
  964. region := aws.String(customPricing.AthenaRegion)
  965. resultsBucket := customPricing.AthenaBucketName
  966. database := customPricing.AthenaDatabase
  967. c := &aws.Config{
  968. Region: region,
  969. }
  970. s := session.Must(session.NewSession(c))
  971. svc := athena.New(s)
  972. var e athena.StartQueryExecutionInput
  973. var r athena.ResultConfiguration
  974. r.SetOutputLocation(resultsBucket)
  975. e.SetResultConfiguration(&r)
  976. e.SetQueryString(query)
  977. var q athena.QueryExecutionContext
  978. q.SetDatabase(database)
  979. e.SetQueryExecutionContext(&q)
  980. res, err := svc.StartQueryExecution(&e)
  981. if err != nil {
  982. return nil, err
  983. }
  984. klog.V(2).Infof("StartQueryExecution result:")
  985. klog.V(2).Infof(res.GoString())
  986. var qri athena.GetQueryExecutionInput
  987. qri.SetQueryExecutionId(*res.QueryExecutionId)
  988. var qrop *athena.GetQueryExecutionOutput
  989. duration := time.Duration(2) * time.Second // Pause for 2 seconds
  990. for {
  991. qrop, err = svc.GetQueryExecution(&qri)
  992. if err != nil {
  993. return nil, err
  994. }
  995. if *qrop.QueryExecution.Status.State != "RUNNING" {
  996. break
  997. }
  998. time.Sleep(duration)
  999. }
  1000. var oocAllocs []*OutOfClusterAllocation
  1001. if *qrop.QueryExecution.Status.State == "SUCCEEDED" {
  1002. var ip athena.GetQueryResultsInput
  1003. ip.SetQueryExecutionId(*res.QueryExecutionId)
  1004. op, err := svc.GetQueryResults(&ip)
  1005. if err != nil {
  1006. return nil, err
  1007. }
  1008. if len(op.ResultSet.Rows) > 1 {
  1009. for _, r := range op.ResultSet.Rows[1:(len(op.ResultSet.Rows) - 1)] {
  1010. cost, err := strconv.ParseFloat(*r.Data[3].VarCharValue, 64)
  1011. if err != nil {
  1012. return nil, err
  1013. }
  1014. ooc := &OutOfClusterAllocation{
  1015. Aggregator: aggregator,
  1016. Environment: *r.Data[1].VarCharValue,
  1017. Service: *r.Data[2].VarCharValue,
  1018. Cost: cost,
  1019. }
  1020. oocAllocs = append(oocAllocs, ooc)
  1021. }
  1022. } else {
  1023. klog.V(1).Infof("No results available for %s at database %s between %s and %s", aggregator_column_name, customPricing.AthenaTable, start, end)
  1024. }
  1025. }
  1026. return oocAllocs, nil // TODO: transform the QuerySQL lines into the new OutOfClusterAllocation Struct
  1027. }
  1028. // QuerySQL can query a properly configured Athena database.
  1029. // Used to fetch billing data.
  1030. // Requires a json config in /var/configs with key region, output, and database.
  1031. func (a *AWS) QuerySQL(query string) ([]byte, error) {
  1032. customPricing, err := a.GetConfig()
  1033. if err != nil {
  1034. return nil, err
  1035. }
  1036. if customPricing.ServiceKeyName != "" {
  1037. err = os.Setenv(awsAccessKeyIDEnvVar, customPricing.ServiceKeyName)
  1038. if err != nil {
  1039. return nil, err
  1040. }
  1041. err = os.Setenv(awsAccessKeySecretEnvVar, customPricing.ServiceKeySecret)
  1042. if err != nil {
  1043. return nil, err
  1044. }
  1045. }
  1046. athenaConfigs, err := os.Open("/var/configs/athena.json")
  1047. if err != nil {
  1048. return nil, err
  1049. }
  1050. defer athenaConfigs.Close()
  1051. b, err := ioutil.ReadAll(athenaConfigs)
  1052. if err != nil {
  1053. return nil, err
  1054. }
  1055. var athenaConf map[string]string
  1056. json.Unmarshal([]byte(b), &athenaConf)
  1057. region := aws.String(customPricing.AthenaRegion)
  1058. resultsBucket := customPricing.AthenaBucketName
  1059. database := customPricing.AthenaDatabase
  1060. c := &aws.Config{
  1061. Region: region,
  1062. }
  1063. s := session.Must(session.NewSession(c))
  1064. svc := athena.New(s)
  1065. var e athena.StartQueryExecutionInput
  1066. var r athena.ResultConfiguration
  1067. r.SetOutputLocation(resultsBucket)
  1068. e.SetResultConfiguration(&r)
  1069. e.SetQueryString(query)
  1070. var q athena.QueryExecutionContext
  1071. q.SetDatabase(database)
  1072. e.SetQueryExecutionContext(&q)
  1073. res, err := svc.StartQueryExecution(&e)
  1074. if err != nil {
  1075. return nil, err
  1076. }
  1077. klog.V(2).Infof("StartQueryExecution result:")
  1078. klog.V(2).Infof(res.GoString())
  1079. var qri athena.GetQueryExecutionInput
  1080. qri.SetQueryExecutionId(*res.QueryExecutionId)
  1081. var qrop *athena.GetQueryExecutionOutput
  1082. duration := time.Duration(2) * time.Second // Pause for 2 seconds
  1083. for {
  1084. qrop, err = svc.GetQueryExecution(&qri)
  1085. if err != nil {
  1086. return nil, err
  1087. }
  1088. if *qrop.QueryExecution.Status.State != "RUNNING" {
  1089. break
  1090. }
  1091. time.Sleep(duration)
  1092. }
  1093. if *qrop.QueryExecution.Status.State == "SUCCEEDED" {
  1094. var ip athena.GetQueryResultsInput
  1095. ip.SetQueryExecutionId(*res.QueryExecutionId)
  1096. op, err := svc.GetQueryResults(&ip)
  1097. if err != nil {
  1098. return nil, err
  1099. }
  1100. b, err := json.Marshal(op.ResultSet)
  1101. if err != nil {
  1102. return nil, err
  1103. }
  1104. return b, nil
  1105. }
  1106. return nil, fmt.Errorf("Error getting query results : %s", *qrop.QueryExecution.Status.State)
  1107. }
  1108. type spotInfo struct {
  1109. Timestamp string `csv:"Timestamp"`
  1110. UsageType string `csv:"UsageType"`
  1111. Operation string `csv:"Operation"`
  1112. InstanceID string `csv:"InstanceID"`
  1113. MyBidID string `csv:"MyBidID"`
  1114. MyMaxPrice string `csv:"MyMaxPrice"`
  1115. MarketPrice string `csv:"MarketPrice"`
  1116. Charge string `csv:"Charge"`
  1117. Version string `csv:"Version"`
  1118. }
  1119. type fnames []*string
  1120. func (f fnames) Len() int {
  1121. return len(f)
  1122. }
  1123. func (f fnames) Swap(i, j int) {
  1124. f[i], f[j] = f[j], f[i]
  1125. }
  1126. func (f fnames) Less(i, j int) bool {
  1127. key1 := strings.Split(*f[i], ".")
  1128. key2 := strings.Split(*f[j], ".")
  1129. t1, err := time.Parse("2006-01-02-15", key1[1])
  1130. if err != nil {
  1131. klog.V(1).Info("Unable to parse timestamp" + key1[1])
  1132. return false
  1133. }
  1134. t2, err := time.Parse("2006-01-02-15", key2[1])
  1135. if err != nil {
  1136. klog.V(1).Info("Unable to parse timestamp" + key2[1])
  1137. return false
  1138. }
  1139. return t1.Before(t2)
  1140. }
  1141. func parseSpotData(bucket string, prefix string, projectID string, region string, accessKeyID string, accessKeySecret string) (map[string]*spotInfo, error) {
  1142. if accessKeyID != "" && accessKeySecret != "" { // credentials may exist on the actual AWS node-- if so, use those. If not, override with the service key
  1143. err := os.Setenv(awsAccessKeyIDEnvVar, accessKeyID)
  1144. if err != nil {
  1145. return nil, err
  1146. }
  1147. err = os.Setenv(awsAccessKeySecretEnvVar, accessKeySecret)
  1148. if err != nil {
  1149. return nil, err
  1150. }
  1151. }
  1152. s3Prefix := projectID
  1153. if len(prefix) != 0 {
  1154. s3Prefix = prefix + "/" + s3Prefix
  1155. }
  1156. c := aws.NewConfig().WithRegion(region)
  1157. s := session.Must(session.NewSession(c))
  1158. s3Svc := s3.New(s)
  1159. downloader := s3manager.NewDownloaderWithClient(s3Svc)
  1160. tNow := time.Now()
  1161. tOneDayAgo := tNow.Add(time.Duration(-24) * time.Hour) // Also get files from one day ago to avoid boundary conditions
  1162. ls := &s3.ListObjectsInput{
  1163. Bucket: aws.String(bucket),
  1164. Prefix: aws.String(s3Prefix + "." + tOneDayAgo.Format("2006-01-02")),
  1165. }
  1166. ls2 := &s3.ListObjectsInput{
  1167. Bucket: aws.String(bucket),
  1168. Prefix: aws.String(s3Prefix + "." + tNow.Format("2006-01-02")),
  1169. }
  1170. lso, err := s3Svc.ListObjects(ls)
  1171. if err != nil {
  1172. return nil, err
  1173. }
  1174. lsoLen := len(lso.Contents)
  1175. klog.V(2).Infof("Found %d spot data files from yesterday", lsoLen)
  1176. if lsoLen == 0 {
  1177. klog.V(5).Infof("ListObjects \"s3://%s/%s\" produced no keys", *ls.Bucket, *ls.Prefix)
  1178. }
  1179. lso2, err := s3Svc.ListObjects(ls2)
  1180. if err != nil {
  1181. return nil, err
  1182. }
  1183. lso2Len := len(lso2.Contents)
  1184. klog.V(2).Infof("Found %d spot data files from today", lso2Len)
  1185. if lso2Len == 0 {
  1186. klog.V(5).Infof("ListObjects \"s3://%s/%s\" produced no keys", *ls2.Bucket, *ls2.Prefix)
  1187. }
  1188. var keys []*string
  1189. for _, obj := range lso.Contents {
  1190. keys = append(keys, obj.Key)
  1191. }
  1192. for _, obj := range lso2.Contents {
  1193. keys = append(keys, obj.Key)
  1194. }
  1195. versionRx := regexp.MustCompile("^#Version: (\\d+)\\.\\d+$")
  1196. header, err := csvutil.Header(spotInfo{}, "csv")
  1197. if err != nil {
  1198. return nil, err
  1199. }
  1200. fieldsPerRecord := len(header)
  1201. spots := make(map[string]*spotInfo)
  1202. for _, key := range keys {
  1203. getObj := &s3.GetObjectInput{
  1204. Bucket: aws.String(bucket),
  1205. Key: key,
  1206. }
  1207. buf := aws.NewWriteAtBuffer([]byte{})
  1208. _, err := downloader.Download(buf, getObj)
  1209. if err != nil {
  1210. return nil, err
  1211. }
  1212. r := bytes.NewReader(buf.Bytes())
  1213. gr, err := gzip.NewReader(r)
  1214. if err != nil {
  1215. return nil, err
  1216. }
  1217. csvReader := csv.NewReader(gr)
  1218. csvReader.Comma = '\t'
  1219. csvReader.FieldsPerRecord = fieldsPerRecord
  1220. dec, err := csvutil.NewDecoder(csvReader, header...)
  1221. if err != nil {
  1222. return nil, err
  1223. }
  1224. var foundVersion string
  1225. for {
  1226. spot := spotInfo{}
  1227. err := dec.Decode(&spot)
  1228. csvParseErr, isCsvParseErr := err.(*csv.ParseError)
  1229. if err == io.EOF {
  1230. break
  1231. } else if err == csvutil.ErrFieldCount || (isCsvParseErr && csvParseErr.Err == csv.ErrFieldCount) {
  1232. rec := dec.Record()
  1233. // the first two "Record()" will be the comment lines
  1234. // and they show up as len() == 1
  1235. // the first of which is "#Version"
  1236. // the second of which is "#Fields: "
  1237. if len(rec) != 1 {
  1238. klog.V(2).Infof("Expected %d spot info fields but received %d: %s", fieldsPerRecord, len(rec), rec)
  1239. continue
  1240. }
  1241. if len(foundVersion) == 0 {
  1242. spotFeedVersion := rec[0]
  1243. klog.V(4).Infof("Spot feed version is \"%s\"", spotFeedVersion)
  1244. matches := versionRx.FindStringSubmatch(spotFeedVersion)
  1245. if matches != nil {
  1246. foundVersion = matches[1]
  1247. if foundVersion != supportedSpotFeedVersion {
  1248. klog.V(2).Infof("Unsupported spot info feed version: wanted \"%s\" got \"%s\"", supportedSpotFeedVersion, foundVersion)
  1249. break
  1250. }
  1251. }
  1252. continue
  1253. } else if strings.Index(rec[0], "#") == 0 {
  1254. continue
  1255. } else {
  1256. klog.V(3).Infof("skipping non-TSV line: %s", rec)
  1257. continue
  1258. }
  1259. } else if err != nil {
  1260. klog.V(2).Infof("Error during spot info decode: %+v", err)
  1261. continue
  1262. }
  1263. klog.V(4).Infof("Found spot info %+v", spot)
  1264. spots[spot.InstanceID] = &spot
  1265. }
  1266. gr.Close()
  1267. }
  1268. return spots, nil
  1269. }
  1270. func (aws *AWS) ApplyReservedInstancePricing(nodes map[string]*Node) {
  1271. }
  1272. /*
  1273. func (aws *AWS) getReservedInstances() ([]interface{}, error) {
  1274. customPricing, err := a.GetConfig()
  1275. if err != nil {
  1276. return nil, err
  1277. }
  1278. if customPricing.ServiceKeyName != "" {
  1279. err = os.Setenv(awsAccessKeyIDEnvVar, customPricing.ServiceKeyName)
  1280. if err != nil {
  1281. return nil, err
  1282. }
  1283. err = os.Setenv(awsAccessKeySecretEnvVar, customPricing.ServiceKeySecret)
  1284. if err != nil {
  1285. return nil, err
  1286. }
  1287. }
  1288. athenaConfigs, err := os.Open("/var/configs/athena.json")
  1289. if err != nil {
  1290. return nil, err
  1291. }
  1292. defer athenaConfigs.Close()
  1293. b, err := ioutil.ReadAll(athenaConfigs)
  1294. if err != nil {
  1295. return nil, err
  1296. }
  1297. var athenaConf map[string]string
  1298. json.Unmarshal([]byte(b), &athenaConf)
  1299. region := aws.String(customPricing.AthenaRegion)
  1300. c := &aws.Config{
  1301. Region: region,
  1302. }
  1303. s := session.Must(session.NewSession(c))
  1304. svc := ec2.New(s)
  1305. svc.DescribeReservedInstances()
  1306. }
  1307. */