cluster_helpers.go 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973
  1. package costmodel
  2. import (
  3. "fmt"
  4. "math"
  5. "strconv"
  6. "time"
  7. coreenv "github.com/opencost/opencost/core/pkg/env"
  8. "github.com/opencost/opencost/pkg/cloud/models"
  9. "github.com/opencost/opencost/pkg/cloud/provider"
  10. "github.com/opencost/opencost/core/pkg/log"
  11. "github.com/opencost/opencost/core/pkg/opencost"
  12. "github.com/opencost/opencost/core/pkg/source"
  13. "github.com/opencost/opencost/core/pkg/util"
  14. )
  15. // mergeTypeMaps takes two maps of (cluster name, node name) -> node type
  16. // and combines them into a single map, preferring the k/v pairs in
  17. // the first map.
  18. func mergeTypeMaps(clusterAndNameToType1, clusterAndNameToType2 map[nodeIdentifierNoProviderID]string) map[nodeIdentifierNoProviderID]string {
  19. merged := map[nodeIdentifierNoProviderID]string{}
  20. for k, v := range clusterAndNameToType2 {
  21. merged[k] = v
  22. }
  23. // This ordering ensures the mappings in the first arg are preferred.
  24. for k, v := range clusterAndNameToType1 {
  25. merged[k] = v
  26. }
  27. return merged
  28. }
  29. func buildCPUCostMap(
  30. resNodeCPUCost []*source.NodeCPUPricePerHrResult,
  31. cp models.Provider,
  32. preemptible map[NodeIdentifier]bool,
  33. ) (map[NodeIdentifier]float64, map[nodeIdentifierNoProviderID]string) {
  34. cpuCostMap := make(map[NodeIdentifier]float64)
  35. clusterAndNameToType := make(map[nodeIdentifierNoProviderID]string)
  36. customPricingEnabled := provider.CustomPricesEnabled(cp)
  37. customPricingConfig, err := cp.GetConfig()
  38. if err != nil {
  39. log.Warnf("ClusterNodes: failed to load custom pricing: %s", err)
  40. }
  41. for _, result := range resNodeCPUCost {
  42. cluster := result.Cluster
  43. if cluster == "" {
  44. cluster = coreenv.GetClusterID()
  45. }
  46. name := result.Node
  47. if name == "" {
  48. log.Warnf("ClusterNodes: CPU cost data missing node")
  49. continue
  50. }
  51. nodeType := result.InstanceType
  52. providerID := result.ProviderID
  53. key := NodeIdentifier{
  54. Cluster: cluster,
  55. Name: name,
  56. ProviderID: provider.ParseID(providerID),
  57. }
  58. keyNon := nodeIdentifierNoProviderID{
  59. Cluster: cluster,
  60. Name: name,
  61. }
  62. var cpuCost float64
  63. // Start with the value from the data source (e.g., collector or Prometheus)
  64. cpuCost = result.Data[0].Value
  65. // If custom pricing is enabled or the data source value is invalid, use custom pricing
  66. if (customPricingEnabled && customPricingConfig != nil) || cpuCost == 0 || math.IsNaN(cpuCost) {
  67. if customPricingConfig != nil {
  68. var customCPUStr string
  69. if spot, ok := preemptible[key]; ok && spot {
  70. customCPUStr = customPricingConfig.SpotCPU
  71. } else {
  72. customCPUStr = customPricingConfig.CPU
  73. }
  74. customCPUCost, err := strconv.ParseFloat(customCPUStr, 64)
  75. if err != nil {
  76. log.Warnf("ClusterNodes: error parsing custom CPU price: %s", customCPUStr)
  77. } else {
  78. // Log the reason for using custom pricing
  79. if cpuCost == 0 {
  80. log.DedupedInfof(10, "ClusterNodes: node %s has invalid CPU cost (0) from data source; falling back to custom pricing: %f", name, customCPUCost)
  81. } else if math.IsNaN(cpuCost) {
  82. log.DedupedInfof(10, "ClusterNodes: node %s has invalid CPU cost (NaN) from data source; falling back to custom pricing: %f", name, customCPUCost)
  83. } else {
  84. log.DedupedInfof(10, "ClusterNodes: node %s using custom pricing: %f", name, customCPUCost)
  85. }
  86. cpuCost = customCPUCost
  87. }
  88. } else {
  89. // custom pricing config is nil, but we needed it because cpuCost was invalid
  90. if cpuCost == 0 || math.IsNaN(cpuCost) {
  91. log.Warnf("ClusterNodes: node %s has invalid CPU cost (0 or NaN), but was unable to fall back to custom pricing because it was nil", name)
  92. }
  93. }
  94. }
  95. clusterAndNameToType[keyNon] = nodeType
  96. cpuCostMap[key] = cpuCost
  97. }
  98. return cpuCostMap, clusterAndNameToType
  99. }
  100. func buildRAMCostMap(
  101. resNodeRAMCost []*source.NodeRAMPricePerGiBHrResult,
  102. cp models.Provider,
  103. preemptible map[NodeIdentifier]bool,
  104. ) (map[NodeIdentifier]float64, map[nodeIdentifierNoProviderID]string) {
  105. ramCostMap := make(map[NodeIdentifier]float64)
  106. clusterAndNameToType := make(map[nodeIdentifierNoProviderID]string)
  107. customPricingEnabled := provider.CustomPricesEnabled(cp)
  108. customPricingConfig, err := cp.GetConfig()
  109. if err != nil {
  110. log.Warnf("ClusterNodes: failed to load custom pricing: %s", err)
  111. }
  112. for _, result := range resNodeRAMCost {
  113. cluster := result.Cluster
  114. if cluster == "" {
  115. cluster = coreenv.GetClusterID()
  116. }
  117. name := result.Node
  118. if name == "" {
  119. log.Warnf("ClusterNodes: RAM cost data missing node")
  120. continue
  121. }
  122. nodeType := result.InstanceType
  123. providerID := result.ProviderID
  124. key := NodeIdentifier{
  125. Cluster: cluster,
  126. Name: name,
  127. ProviderID: provider.ParseID(providerID),
  128. }
  129. keyNon := nodeIdentifierNoProviderID{
  130. Cluster: cluster,
  131. Name: name,
  132. }
  133. var ramCost float64
  134. // Start with the value from the data source (e.g., collector or Prometheus)
  135. ramCost = result.Data[0].Value
  136. // If custom pricing is enabled or the data source value is invalid, use custom pricing
  137. if (customPricingEnabled && customPricingConfig != nil) || ramCost == 0 || math.IsNaN(ramCost) {
  138. if customPricingConfig != nil {
  139. var customRAMStr string
  140. if spot, ok := preemptible[key]; ok && spot {
  141. customRAMStr = customPricingConfig.SpotRAM
  142. } else {
  143. customRAMStr = customPricingConfig.RAM
  144. }
  145. customRAMCost, err := strconv.ParseFloat(customRAMStr, 64)
  146. if err != nil {
  147. log.Warnf("ClusterNodes: error parsing custom RAM price: %s", customRAMStr)
  148. } else {
  149. // Log the reason for using custom pricing
  150. if ramCost == 0 {
  151. log.DedupedInfof(10, "ClusterNodes: node %s has invalid RAM cost (0) from data source; falling back to custom pricing: %f", name, customRAMCost)
  152. } else if math.IsNaN(ramCost) {
  153. log.DedupedInfof(10, "ClusterNodes: node %s has invalid RAM cost (NaN) from data source; falling back to custom pricing: %f", name, customRAMCost)
  154. } else {
  155. log.DedupedInfof(10, "ClusterNodes: node %s using custom pricing: %f", name, customRAMCost)
  156. }
  157. ramCost = customRAMCost
  158. }
  159. } else {
  160. if ramCost == 0 || math.IsNaN(ramCost) {
  161. log.Warnf("ClusterNodes: node %s has invalid RAM cost (0 or NaN), but was unable to fall back to custom pricing because it was nil", name)
  162. }
  163. }
  164. }
  165. clusterAndNameToType[keyNon] = nodeType
  166. // covert to price per byte/hr
  167. ramCostMap[key] = ramCost / 1024.0 / 1024.0 / 1024.0
  168. }
  169. return ramCostMap, clusterAndNameToType
  170. }
  171. func buildGPUCostMap(
  172. resNodeGPUCost []*source.NodeGPUPricePerHrResult,
  173. gpuCountMap map[NodeIdentifier]float64,
  174. cp models.Provider,
  175. preemptible map[NodeIdentifier]bool,
  176. ) (map[NodeIdentifier]float64, map[nodeIdentifierNoProviderID]string) {
  177. gpuCostMap := make(map[NodeIdentifier]float64)
  178. clusterAndNameToType := make(map[nodeIdentifierNoProviderID]string)
  179. customPricingEnabled := provider.CustomPricesEnabled(cp)
  180. customPricingConfig, err := cp.GetConfig()
  181. if err != nil {
  182. log.Warnf("ClusterNodes: failed to load custom pricing: %s", err)
  183. }
  184. for _, result := range resNodeGPUCost {
  185. cluster := result.Cluster
  186. if cluster == "" {
  187. cluster = coreenv.GetClusterID()
  188. }
  189. name := result.Node
  190. if name == "" {
  191. log.Warnf("ClusterNodes: GPU cost data missing node")
  192. continue
  193. }
  194. nodeType := result.InstanceType
  195. providerID := result.ProviderID
  196. key := NodeIdentifier{
  197. Cluster: cluster,
  198. Name: name,
  199. ProviderID: provider.ParseID(providerID),
  200. }
  201. keyNon := nodeIdentifierNoProviderID{
  202. Cluster: cluster,
  203. Name: name,
  204. }
  205. var gpuCost float64
  206. if customPricingEnabled && customPricingConfig != nil {
  207. var customGPUStr string
  208. if spot, ok := preemptible[key]; ok && spot {
  209. customGPUStr = customPricingConfig.SpotGPU
  210. } else {
  211. customGPUStr = customPricingConfig.GPU
  212. }
  213. customGPUCost, err := strconv.ParseFloat(customGPUStr, 64)
  214. if err != nil {
  215. log.Warnf("ClusterNodes: error parsing custom GPU price: %s", customGPUStr)
  216. }
  217. gpuCost = customGPUCost
  218. } else {
  219. gpuCost = result.Data[0].Value
  220. }
  221. clusterAndNameToType[keyNon] = nodeType
  222. // If gpu count is available use it to multiply gpu cost
  223. if value, ok := gpuCountMap[key]; ok {
  224. gpuCostMap[key] = gpuCost * value
  225. } else {
  226. gpuCostMap[key] = 0
  227. }
  228. }
  229. return gpuCostMap, clusterAndNameToType
  230. }
  231. func buildGPUCountMap(resNodeGPUCount []*source.NodeGPUCountResult) map[NodeIdentifier]float64 {
  232. gpuCountMap := make(map[NodeIdentifier]float64)
  233. for _, result := range resNodeGPUCount {
  234. cluster := result.Cluster
  235. if cluster == "" {
  236. cluster = coreenv.GetClusterID()
  237. }
  238. name := result.Node
  239. if name == "" {
  240. log.Warnf("ClusterNodes: GPU count data missing node")
  241. continue
  242. }
  243. gpuCount := result.Data[0].Value
  244. providerID := result.ProviderID
  245. key := NodeIdentifier{
  246. Cluster: cluster,
  247. Name: name,
  248. ProviderID: provider.ParseID(providerID),
  249. }
  250. gpuCountMap[key] = gpuCount
  251. }
  252. return gpuCountMap
  253. }
  254. func buildCPUCoresMap(resNodeCPUCores []*source.NodeCPUCoresCapacityResult) map[nodeIdentifierNoProviderID]float64 {
  255. m := make(map[nodeIdentifierNoProviderID]float64)
  256. for _, result := range resNodeCPUCores {
  257. cluster := result.Cluster
  258. if cluster == "" {
  259. cluster = coreenv.GetClusterID()
  260. }
  261. name := result.Node
  262. if name == "" {
  263. log.Warnf("ClusterNodes: CPU cores data missing node")
  264. continue
  265. }
  266. cpuCores := result.Data[0].Value
  267. key := nodeIdentifierNoProviderID{
  268. Cluster: cluster,
  269. Name: name,
  270. }
  271. m[key] = cpuCores
  272. }
  273. return m
  274. }
  275. func buildRAMBytesMap(resNodeRAMBytes []*source.NodeRAMBytesCapacityResult) map[nodeIdentifierNoProviderID]float64 {
  276. m := make(map[nodeIdentifierNoProviderID]float64)
  277. for _, result := range resNodeRAMBytes {
  278. cluster := result.Cluster
  279. if cluster == "" {
  280. cluster = coreenv.GetClusterID()
  281. }
  282. name := result.Node
  283. if name == "" {
  284. log.Warnf("ClusterNodes: RAM bytes data missing node")
  285. continue
  286. }
  287. ramBytes := result.Data[0].Value
  288. key := nodeIdentifierNoProviderID{
  289. Cluster: cluster,
  290. Name: name,
  291. }
  292. m[key] = ramBytes
  293. }
  294. return m
  295. }
  296. // Mapping of cluster/node=cpu for computing resource efficiency
  297. func buildCPUBreakdownMap(resNodeCPUModeTotal []*source.NodeCPUModeTotalResult) map[nodeIdentifierNoProviderID]*ClusterCostsBreakdown {
  298. cpuBreakdownMap := make(map[nodeIdentifierNoProviderID]*ClusterCostsBreakdown)
  299. // Mapping of cluster/node=cpu for computing resource efficiency
  300. clusterNodeCPUTotal := map[nodeIdentifierNoProviderID]float64{}
  301. // Mapping of cluster/node:mode=cpu for computing resource efficiency
  302. clusterNodeModeCPUTotal := map[nodeIdentifierNoProviderID]map[string]float64{}
  303. // Build intermediate structures for CPU usage by (cluster, node) and by
  304. // (cluster, node, mode) for computing resouce efficiency
  305. for _, result := range resNodeCPUModeTotal {
  306. cluster := result.Cluster
  307. if cluster == "" {
  308. cluster = coreenv.GetClusterID()
  309. }
  310. node := result.Node
  311. if node == "" {
  312. log.DedupedWarningf(5, "ClusterNodes: CPU mode data missing node")
  313. continue
  314. }
  315. mode := result.Mode
  316. if mode == "" {
  317. log.DedupedWarningf(10, "ClusterNodes: unable to read CPU mode data for node %s.", node)
  318. mode = "other"
  319. }
  320. key := nodeIdentifierNoProviderID{
  321. Cluster: cluster,
  322. Name: node,
  323. }
  324. total := result.Data[0].Value
  325. // Increment total
  326. clusterNodeCPUTotal[key] += total
  327. // Increment mode
  328. if _, ok := clusterNodeModeCPUTotal[key]; !ok {
  329. clusterNodeModeCPUTotal[key] = map[string]float64{}
  330. }
  331. clusterNodeModeCPUTotal[key][mode] += total
  332. }
  333. // Compute resource efficiency from intermediate structures
  334. for key, total := range clusterNodeCPUTotal {
  335. if modeTotals, ok := clusterNodeModeCPUTotal[key]; ok {
  336. for mode, subtotal := range modeTotals {
  337. // Compute percentage for the current cluster, node, mode
  338. pct := 0.0
  339. if total > 0 {
  340. pct = subtotal / total
  341. }
  342. if _, ok := cpuBreakdownMap[key]; !ok {
  343. cpuBreakdownMap[key] = &ClusterCostsBreakdown{}
  344. }
  345. switch mode {
  346. case "idle":
  347. cpuBreakdownMap[key].Idle += pct
  348. case "system":
  349. cpuBreakdownMap[key].System += pct
  350. case "user":
  351. cpuBreakdownMap[key].User += pct
  352. default:
  353. cpuBreakdownMap[key].Other += pct
  354. }
  355. }
  356. }
  357. }
  358. return cpuBreakdownMap
  359. }
  360. func buildOverheadMap(capRam, allocRam, capCPU, allocCPU map[nodeIdentifierNoProviderID]float64) map[nodeIdentifierNoProviderID]*NodeOverhead {
  361. m := make(map[nodeIdentifierNoProviderID]*NodeOverhead, len(capRam))
  362. for identifier, ramCapacity := range capRam {
  363. allocatableRam, ok := allocRam[identifier]
  364. if !ok {
  365. log.Warnf("Could not find allocatable ram for node %s", identifier.Name)
  366. continue
  367. }
  368. // ramCapacity == 0 would produce NaN, which is not JSON-representable
  369. // and breaks downstream consumers (e.g. the MCP marshaler).
  370. ramFraction := 0.0
  371. if ramCapacity > 0 {
  372. ramFraction = (ramCapacity - allocatableRam) / ramCapacity
  373. }
  374. m[identifier] = &NodeOverhead{
  375. RamOverheadFraction: ramFraction,
  376. }
  377. }
  378. for identifier, cpuCapacity := range capCPU {
  379. allocatableCPU, ok := allocCPU[identifier]
  380. if !ok {
  381. log.Warnf("Could not find allocatable cpu for node %s", identifier.Name)
  382. continue
  383. }
  384. cpuFraction := 0.0
  385. if cpuCapacity > 0 {
  386. cpuFraction = (cpuCapacity - allocatableCPU) / cpuCapacity
  387. }
  388. if _, found := m[identifier]; found {
  389. m[identifier].CpuOverheadFraction = cpuFraction
  390. } else {
  391. m[identifier] = &NodeOverhead{
  392. CpuOverheadFraction: cpuFraction,
  393. }
  394. }
  395. }
  396. return m
  397. }
  398. func buildRAMUserPctMap(resNodeRAMUserPct []*source.NodeRAMUserPercentResult) map[nodeIdentifierNoProviderID]float64 {
  399. m := make(map[nodeIdentifierNoProviderID]float64)
  400. for _, result := range resNodeRAMUserPct {
  401. cluster := result.Cluster
  402. if cluster == "" {
  403. cluster = coreenv.GetClusterID()
  404. }
  405. name := result.Instance
  406. if name == "" {
  407. log.Warnf("ClusterNodes: RAM user percent missing node")
  408. continue
  409. }
  410. pct := result.Data[0].Value
  411. key := nodeIdentifierNoProviderID{
  412. Cluster: cluster,
  413. Name: name,
  414. }
  415. m[key] = pct
  416. }
  417. return m
  418. }
  419. func buildRAMSystemPctMap(resNodeRAMSystemPct []*source.NodeRAMSystemPercentResult) map[nodeIdentifierNoProviderID]float64 {
  420. m := make(map[nodeIdentifierNoProviderID]float64)
  421. for _, result := range resNodeRAMSystemPct {
  422. cluster := result.Cluster
  423. if cluster == "" {
  424. cluster = coreenv.GetClusterID()
  425. }
  426. name := result.Instance
  427. if name == "" {
  428. log.Warnf("ClusterNodes: RAM system percent missing node")
  429. continue
  430. }
  431. pct := result.Data[0].Value
  432. key := nodeIdentifierNoProviderID{
  433. Cluster: cluster,
  434. Name: name,
  435. }
  436. m[key] = pct
  437. }
  438. return m
  439. }
  440. type activeData struct {
  441. start time.Time
  442. end time.Time
  443. minutes float64
  444. }
  445. // cluster management key gen
  446. func clusterManagementKeyGen(result *source.ClusterManagementDurationResult) (ClusterManagementIdentifier, bool) {
  447. cluster := result.Cluster
  448. if cluster == "" {
  449. cluster = coreenv.GetClusterID()
  450. }
  451. provisionerName := result.Provisioner
  452. return ClusterManagementIdentifier{
  453. Cluster: cluster,
  454. Provisioner: provisionerName,
  455. }, true
  456. }
  457. func clusterManagementValues(result *source.ClusterManagementDurationResult) []*util.Vector {
  458. return result.Data
  459. }
  460. // node key gen
  461. func nodeKeyGen(result *source.NodeActiveMinutesResult) (NodeIdentifier, bool) {
  462. cluster := result.Cluster
  463. if cluster == "" {
  464. cluster = coreenv.GetClusterID()
  465. }
  466. name := result.Node
  467. if name == "" {
  468. log.Warnf("ClusterNodes: active mins missing node")
  469. return NodeIdentifier{}, false
  470. }
  471. providerID := result.ProviderID
  472. return NodeIdentifier{
  473. Cluster: cluster,
  474. Name: name,
  475. ProviderID: provider.ParseID(providerID),
  476. }, true
  477. }
  478. func nodeValues(result *source.NodeActiveMinutesResult) []*util.Vector {
  479. return result.Data
  480. }
  481. func loadBalancerKeyGen(result *source.LBActiveMinutesResult) (LoadBalancerIdentifier, bool) {
  482. cluster := result.Cluster
  483. if cluster == "" {
  484. cluster = coreenv.GetClusterID()
  485. }
  486. namespace := result.Namespace
  487. if namespace == "" {
  488. log.Warnf("ClusterLoadBalancers: LB cost data missing namespace")
  489. return LoadBalancerIdentifier{}, false
  490. }
  491. name := result.Service
  492. if name == "" {
  493. log.Warnf("ClusterLoadBalancers: LB cost data missing service_name")
  494. return LoadBalancerIdentifier{}, false
  495. }
  496. ingressIp := result.IngressIP
  497. if ingressIp == "" {
  498. log.DedupedWarningf(5, "ClusterLoadBalancers: LB cost data missing ingress_ip")
  499. // only update asset cost when an actual IP was returned
  500. return LoadBalancerIdentifier{}, false
  501. }
  502. return LoadBalancerIdentifier{
  503. Cluster: cluster,
  504. Namespace: namespace,
  505. Name: fmt.Sprintf("%s/%s", namespace, name), // TODO: this is kept for backwards-compatibility, but not good,
  506. IngressIP: ingressIp,
  507. }, true
  508. }
  509. func lbValues(result *source.LBActiveMinutesResult) []*util.Vector {
  510. return result.Data
  511. }
  512. func buildActiveDataMap[T comparable, U any](
  513. results []*U,
  514. keyGen func(*U) (T, bool),
  515. valuesFunc func(*U) []*util.Vector,
  516. resolution time.Duration,
  517. window opencost.Window,
  518. ) map[T]activeData {
  519. m := make(map[T]activeData)
  520. for _, result := range results {
  521. key, ok := keyGen(result)
  522. values := valuesFunc(result)
  523. if !ok || len(values) == 0 {
  524. continue
  525. }
  526. s, e := calculateStartAndEnd(values, resolution, window)
  527. mins := e.Sub(s).Minutes()
  528. m[key] = activeData{
  529. start: s,
  530. end: e,
  531. minutes: mins,
  532. }
  533. }
  534. return m
  535. }
  536. // Determine preemptibility with node labels
  537. // node id -> is preemptible?
  538. func buildPreemptibleMap(
  539. resIsSpot []*source.NodeIsSpotResult,
  540. ) map[NodeIdentifier]bool {
  541. m := make(map[NodeIdentifier]bool)
  542. for _, result := range resIsSpot {
  543. cluster := result.Cluster
  544. if cluster == "" {
  545. cluster = coreenv.GetClusterID()
  546. }
  547. name := result.Node
  548. if name == "" {
  549. log.Warnf("ClusterNodes: active mins missing node")
  550. continue
  551. }
  552. providerID := result.ProviderID
  553. key := NodeIdentifier{
  554. Cluster: cluster,
  555. Name: name,
  556. ProviderID: provider.ParseID(providerID),
  557. }
  558. // GCP preemptible label
  559. pre := result.Data[0].Value
  560. // TODO(michaelmdresser): check this condition at merge time?
  561. // if node, ok := nodeMap[key]; pre > 0.0 && ok {
  562. // node.Preemptible = true
  563. // }
  564. m[key] = pre > 0.0
  565. // TODO AWS preemptible
  566. // TODO Azure preemptible
  567. }
  568. return m
  569. }
  570. func buildAssetsPVCMap(resPVCInfo []*source.PVCInfoResult) map[DiskIdentifier]*Disk {
  571. diskMap := map[DiskIdentifier]*Disk{}
  572. for _, result := range resPVCInfo {
  573. cluster := result.Cluster
  574. if cluster == "" {
  575. cluster = coreenv.GetClusterID()
  576. }
  577. volumeName := result.VolumeName
  578. if volumeName == "" {
  579. log.Debugf("ClusterDisks: pv claim data missing volumename")
  580. continue
  581. }
  582. claimName := result.PersistentVolumeClaim
  583. if claimName == "" {
  584. log.Debugf("ClusterDisks: pv claim data missing persistentvolumeclaim")
  585. continue
  586. }
  587. claimNamespace := result.Namespace
  588. if claimNamespace == "" {
  589. log.Debugf("ClusterDisks: pv claim data missing namespace")
  590. continue
  591. }
  592. key := DiskIdentifier{
  593. Cluster: cluster,
  594. Name: volumeName,
  595. }
  596. if _, ok := diskMap[key]; !ok {
  597. diskMap[key] = &Disk{
  598. Cluster: cluster,
  599. Name: volumeName,
  600. Breakdown: &ClusterCostsBreakdown{},
  601. }
  602. }
  603. diskMap[key].VolumeName = volumeName
  604. diskMap[key].ClaimName = claimName
  605. diskMap[key].ClaimNamespace = claimNamespace
  606. }
  607. return diskMap
  608. }
  609. func buildLabelsMap(
  610. resLabels []*source.NodeLabelsResult,
  611. ) map[nodeIdentifierNoProviderID]map[string]string {
  612. m := make(map[nodeIdentifierNoProviderID]map[string]string)
  613. // Copy labels into node
  614. for _, result := range resLabels {
  615. cluster := result.Cluster
  616. if cluster == "" {
  617. cluster = coreenv.GetClusterID()
  618. }
  619. node := result.Node
  620. if node == "" {
  621. log.DedupedWarningf(5, "ClusterNodes: label data missing node")
  622. continue
  623. }
  624. key := nodeIdentifierNoProviderID{
  625. Cluster: cluster,
  626. Name: node,
  627. }
  628. // The QueryResult.GetLabels function needs to be called to sanitize the
  629. // ingested label data. This removes the label_ prefix that prometheus
  630. // adds to emitted labels. It also keeps from ingesting prometheus labels
  631. // that aren't a part of the asset.
  632. if _, ok := m[key]; !ok {
  633. m[key] = map[string]string{}
  634. }
  635. for k, l := range result.Labels {
  636. m[key][k] = l
  637. }
  638. }
  639. return m
  640. }
  641. // checkForKeyAndInitIfMissing inits a key in the provided nodemap if
  642. // it does not exist. Intended to be called ONLY by buildNodeMap
  643. func checkForKeyAndInitIfMissing(
  644. nodeMap map[NodeIdentifier]*Node,
  645. key NodeIdentifier,
  646. clusterAndNameToType map[nodeIdentifierNoProviderID]string,
  647. ) {
  648. if _, ok := nodeMap[key]; !ok {
  649. // default nodeType in case we don't have the mapping
  650. var nodeType string
  651. if t, ok := clusterAndNameToType[nodeIdentifierNoProviderID{
  652. Cluster: key.Cluster,
  653. Name: key.Name,
  654. }]; ok {
  655. nodeType = t
  656. } else {
  657. log.Warnf("ClusterNodes: Type does not exist for node identifier %s", key)
  658. }
  659. nodeMap[key] = &Node{
  660. Cluster: key.Cluster,
  661. Name: key.Name,
  662. NodeType: nodeType,
  663. ProviderID: key.ProviderID,
  664. CPUBreakdown: &ClusterCostsBreakdown{},
  665. RAMBreakdown: &ClusterCostsBreakdown{},
  666. }
  667. }
  668. }
  669. // buildNodeMap creates the main set of node data for ClusterNodes from
  670. // the data maps built from Prometheus queries. Some of the Prometheus
  671. // data has access to the provider_id field and some does not. To get
  672. // around this problem, we use the data that includes provider_id
  673. // to build up the definitive set of nodes and then use the data
  674. // with less-specific identifiers (i.e. without provider_id) to fill
  675. // in the remaining fields.
  676. //
  677. // For example, let's say we have nodes identified like so:
  678. // cluster name/node name/provider_id. For the sake of the example,
  679. // we will also limit data to CPU cost, CPU cores, and preemptibility.
  680. //
  681. // We have CPU cost data that looks like this:
  682. // cluster1/node1/prov_node1_A: $10
  683. // cluster1/node1/prov_node1_B: $8
  684. // cluster1/node2/prov_node2: $15
  685. //
  686. // We have Preemptible data that looks like this:
  687. // cluster1/node1/prov_node1_A: true
  688. // cluster1/node1/prov_node1_B: false
  689. // cluster1/node2/prov_node2_B: false
  690. //
  691. // We have CPU cores data that looks like this:
  692. // cluster1/node1: 4
  693. // cluster1/node2: 6
  694. //
  695. // This function first combines the data that is fully identified,
  696. // creating the following:
  697. // cluster1/node1/prov_node1_A: CPUCost($10), Preemptible(true)
  698. // cluster1/node1/prov_node1_B: CPUCost($8), Preemptible(false)
  699. // cluster1/node2/prov_node2: CPUCost($15), Preemptible(false)
  700. //
  701. // It then uses the less-specific data to extend the specific data,
  702. // making the following:
  703. // cluster1/node1/prov_node1_A: CPUCost($10), Preemptible(true), Cores(4)
  704. // cluster1/node1/prov_node1_B: CPUCost($8), Preemptible(false), Cores(4)
  705. // cluster1/node2/prov_node2: CPUCost($15), Preemptible(false), Cores(6)
  706. //
  707. // In the situation where provider_id doesn't exist for any metrics,
  708. // that is the same as all provider_ids being empty strings. If
  709. // provider_id doesn't exist at all, then we (without having to do
  710. // extra work) easily fall back on identifying nodes only by cluster name
  711. // and node name because the provider_id part of the key will always
  712. // be the empty string.
  713. //
  714. // It is worth nothing that, in this approach, if a node is not present
  715. // in the more specific data but is present in the less-specific data,
  716. // that data is never processed into the final node map. For example,
  717. // let's say the CPU cores map has the following entry:
  718. // cluster1/node8: 6
  719. // But none of the maps with provider_id (CPU cost, RAM cost, etc.)
  720. // have an identifier for cluster1/node8 (regardless of provider_id).
  721. // In this situation, the final node map will not have a cluster1/node8
  722. // entry. This could be fixed by iterating over all of the less specific
  723. // identifiers and, inside that iteration, all of the identifiers in
  724. // the node map, but this would introduce a roughly quadratic time
  725. // complexity.
  726. func buildNodeMap(
  727. cpuCostMap, ramCostMap, gpuCostMap, gpuCountMap map[NodeIdentifier]float64,
  728. cpuCoresMap, ramBytesMap, ramUserPctMap,
  729. ramSystemPctMap map[nodeIdentifierNoProviderID]float64,
  730. cpuBreakdownMap map[nodeIdentifierNoProviderID]*ClusterCostsBreakdown,
  731. activeDataMap map[NodeIdentifier]activeData,
  732. preemptibleMap map[NodeIdentifier]bool,
  733. labelsMap map[nodeIdentifierNoProviderID]map[string]string,
  734. clusterAndNameToType map[nodeIdentifierNoProviderID]string,
  735. overheadMap map[nodeIdentifierNoProviderID]*NodeOverhead,
  736. ) map[NodeIdentifier]*Node {
  737. nodeMap := make(map[NodeIdentifier]*Node)
  738. // Initialize the map with the most-specific data:
  739. for id, cost := range cpuCostMap {
  740. checkForKeyAndInitIfMissing(nodeMap, id, clusterAndNameToType)
  741. nodeMap[id].CPUCost = cost
  742. }
  743. for id, cost := range ramCostMap {
  744. checkForKeyAndInitIfMissing(nodeMap, id, clusterAndNameToType)
  745. nodeMap[id].RAMCost = cost
  746. }
  747. for id, cost := range gpuCostMap {
  748. checkForKeyAndInitIfMissing(nodeMap, id, clusterAndNameToType)
  749. nodeMap[id].GPUCost = cost
  750. }
  751. for id, count := range gpuCountMap {
  752. checkForKeyAndInitIfMissing(nodeMap, id, clusterAndNameToType)
  753. nodeMap[id].GPUCount = count
  754. }
  755. for id, preemptible := range preemptibleMap {
  756. checkForKeyAndInitIfMissing(nodeMap, id, clusterAndNameToType)
  757. nodeMap[id].Preemptible = preemptible
  758. }
  759. for id, activeData := range activeDataMap {
  760. checkForKeyAndInitIfMissing(nodeMap, id, clusterAndNameToType)
  761. nodeMap[id].Start = activeData.start
  762. nodeMap[id].End = activeData.end
  763. nodeMap[id].Minutes = nodeMap[id].End.Sub(nodeMap[id].Start).Minutes()
  764. }
  765. // We now merge in data that doesn't have a provider id by looping over
  766. // all keys already added and inserting data according to their
  767. // cluster name/node name combos.
  768. for id, nodePtr := range nodeMap {
  769. clusterAndNameID := nodeIdentifierNoProviderID{
  770. Cluster: id.Cluster,
  771. Name: id.Name,
  772. }
  773. if cores, ok := cpuCoresMap[clusterAndNameID]; ok {
  774. nodePtr.CPUCores = cores
  775. if v, ok := partialCPUMap[nodePtr.NodeType]; ok {
  776. if cores > 0 {
  777. nodePtr.CPUCores = v
  778. adjustmentFactor := v / cores
  779. nodePtr.CPUCost = nodePtr.CPUCost * adjustmentFactor
  780. }
  781. }
  782. }
  783. if ramBytes, ok := ramBytesMap[clusterAndNameID]; ok {
  784. nodePtr.RAMBytes = ramBytes
  785. }
  786. if ramUserPct, ok := ramUserPctMap[clusterAndNameID]; ok {
  787. nodePtr.RAMBreakdown.User = ramUserPct
  788. }
  789. if ramSystemPct, ok := ramSystemPctMap[clusterAndNameID]; ok {
  790. nodePtr.RAMBreakdown.System = ramSystemPct
  791. }
  792. if cpuBreakdown, ok := cpuBreakdownMap[clusterAndNameID]; ok {
  793. nodePtr.CPUBreakdown = cpuBreakdown
  794. }
  795. if labels, ok := labelsMap[clusterAndNameID]; ok {
  796. nodePtr.Labels = labels
  797. }
  798. if overhead, ok := overheadMap[clusterAndNameID]; ok {
  799. nodePtr.Overhead = overhead
  800. } else {
  801. // we were unable to compute overhead for this node
  802. // assume default case of no overhead
  803. nodePtr.Overhead = &NodeOverhead{}
  804. log.Warnf("unable to compute overhead for node %s - defaulting to no overhead", clusterAndNameID.Name)
  805. }
  806. }
  807. return nodeMap
  808. }