awsprovider.go 41 KB

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