server.go 45 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335
  1. package mcp
  2. import (
  3. "context"
  4. "crypto/rand"
  5. "encoding/hex"
  6. "fmt"
  7. "math"
  8. "reflect"
  9. "strings"
  10. "sync"
  11. "time"
  12. "github.com/go-playground/validator/v10"
  13. "github.com/opencost/opencost/core/pkg/filter"
  14. "github.com/opencost/opencost/core/pkg/filter/allocation"
  15. cloudcostfilter "github.com/opencost/opencost/core/pkg/filter/cloudcost"
  16. "github.com/opencost/opencost/core/pkg/log"
  17. "github.com/opencost/opencost/core/pkg/opencost"
  18. models "github.com/opencost/opencost/pkg/cloud/models"
  19. "github.com/opencost/opencost/pkg/cloudcost"
  20. "github.com/opencost/opencost/pkg/costmodel"
  21. "github.com/opencost/opencost/pkg/env"
  22. )
  23. // QueryType defines the type of query to be executed.
  24. type QueryType string
  25. const (
  26. AllocationQueryType QueryType = "allocation"
  27. AssetQueryType QueryType = "asset"
  28. CloudCostQueryType QueryType = "cloudcost"
  29. EfficiencyQueryType QueryType = "efficiency"
  30. )
  31. // Efficiency calculation constants
  32. const (
  33. efficiencyBufferMultiplier = 1.2 // 20% headroom for stability
  34. efficiencyMinCPU = 0.001 // minimum CPU cores
  35. efficiencyMinRAM = 1024 * 1024 // 1 MB minimum RAM
  36. )
  37. // MCPRequest represents a single turn in a conversation with the OpenCost MCP server.
  38. type MCPRequest struct {
  39. SessionID string `json:"sessionId"`
  40. Query *OpenCostQueryRequest `json:"query"`
  41. }
  42. // MCPResponse is the response from the OpenCost MCP server for a single turn.
  43. type MCPResponse struct {
  44. Data interface{} `json:"data"`
  45. QueryInfo QueryMetadata `json:"queryInfo"`
  46. }
  47. // QueryMetadata contains metadata about the query execution.
  48. type QueryMetadata struct {
  49. QueryID string `json:"queryId"`
  50. Timestamp time.Time `json:"timestamp"`
  51. ProcessingTime time.Duration `json:"processingTime"`
  52. }
  53. // OpenCostQueryRequest provides a unified interface for all OpenCost query types.
  54. type OpenCostQueryRequest struct {
  55. QueryType QueryType `json:"queryType" validate:"required,oneof=allocation asset cloudcost efficiency"`
  56. Window string `json:"window" validate:"required"`
  57. AllocationParams *AllocationQuery `json:"allocationParams,omitempty"`
  58. AssetParams *AssetQuery `json:"assetParams,omitempty"`
  59. CloudCostParams *CloudCostQuery `json:"cloudCostParams,omitempty"`
  60. EfficiencyParams *EfficiencyQuery `json:"efficiencyParams,omitempty"`
  61. }
  62. // AllocationQuery contains the parameters for an allocation query.
  63. type AllocationQuery struct {
  64. Step time.Duration `json:"step,omitempty"`
  65. Accumulate bool `json:"accumulate,omitempty"`
  66. ShareIdle bool `json:"shareIdle,omitempty"`
  67. Aggregate string `json:"aggregate,omitempty"`
  68. IncludeIdle bool `json:"includeIdle,omitempty"`
  69. IdleByNode bool `json:"idleByNode,omitempty"`
  70. IncludeProportionalAssetResourceCosts bool `json:"includeProportionalAssetResourceCosts,omitempty"`
  71. IncludeAggregatedMetadata bool `json:"includeAggregatedMetadata,omitempty"`
  72. ShareLB bool `json:"sharelb,omitempty"`
  73. Filter string `json:"filter,omitempty"` // Filter expression for allocations (e.g., "cluster:production", "namespace:kube-system")
  74. }
  75. // AssetQuery contains the parameters for an asset query.
  76. type AssetQuery struct {
  77. // Currently no specific parameters needed for asset queries as it only takes window as parameter
  78. }
  79. // CloudCostQuery contains the parameters for a cloud cost query.
  80. type CloudCostQuery struct {
  81. Aggregate string `json:"aggregate,omitempty"` // Comma-separated list of aggregation properties
  82. Accumulate string `json:"accumulate,omitempty"` // e.g., "week", "day", "month"
  83. Filter string `json:"filter,omitempty"` // Filter expression for cloud costs
  84. Provider string `json:"provider,omitempty"` // Cloud provider filter (aws, gcp, azure, etc.)
  85. Service string `json:"service,omitempty"` // Service filter (ec2, s3, compute, etc.)
  86. Category string `json:"category,omitempty"` // Category filter (compute, storage, network, etc.)
  87. Region string `json:"region,omitempty"` // Region filter
  88. // Additional explicit fields for filtering
  89. AccountID string `json:"accountID,omitempty"` // Alias of Account; maps to accountID
  90. InvoiceEntityID string `json:"invoiceEntityID,omitempty"` // Invoice entity ID filter
  91. ProviderID string `json:"providerID,omitempty"` // Cloud provider resource ID filter
  92. Labels map[string]string `json:"labels,omitempty"` // Label filters (key->value)
  93. }
  94. // EfficiencyQuery contains the parameters for an efficiency query.
  95. type EfficiencyQuery struct {
  96. Step time.Duration `json:"step,omitempty"` // Query step size; controls peak memory by batching large windows (default: auto-scaled based on window)
  97. Aggregate string `json:"aggregate,omitempty"` // Aggregation properties (e.g., "pod", "namespace", "controller")
  98. Filter string `json:"filter,omitempty"` // Filter expression for allocations (same as AllocationQuery)
  99. EfficiencyBufferMultiplier *float64 `json:"efficiencyBufferMultiplier,omitempty"` // Buffer multiplier for recommendations (default: 1.2 for 20% headroom)
  100. }
  101. // AllocationResponse represents the allocation data returned to the AI agent.
  102. type AllocationResponse struct {
  103. // The allocation data, as a map of allocation sets.
  104. Allocations map[string]*AllocationSet `json:"allocations"`
  105. }
  106. // AllocationSet represents a set of allocation data.
  107. type AllocationSet struct {
  108. // The name of the allocation set.
  109. Name string `json:"name"`
  110. Properties map[string]string `json:"properties"`
  111. Allocations []*Allocation `json:"allocations"`
  112. }
  113. // TotalCost calculates the total cost of all allocations in the set.
  114. func (as *AllocationSet) TotalCost() float64 {
  115. var total float64
  116. for _, alloc := range as.Allocations {
  117. total += alloc.TotalCost
  118. }
  119. return total
  120. }
  121. // Allocation represents a single allocation data point.
  122. type Allocation struct {
  123. Name string `json:"name"` // Allocation key (namespace, cluster, etc.)
  124. CPUCost float64 `json:"cpuCost"` // Cost of CPU usage
  125. GPUCost float64 `json:"gpuCost"` // Cost of GPU usage
  126. RAMCost float64 `json:"ramCost"` // Cost of memory usage
  127. PVCost float64 `json:"pvCost"` // Cost of persistent volumes
  128. NetworkCost float64 `json:"networkCost"` // Cost of network usage
  129. SharedCost float64 `json:"sharedCost"` // Shared/unallocated costs assigned here
  130. ExternalCost float64 `json:"externalCost"` // External costs (cloud services, etc.)
  131. TotalCost float64 `json:"totalCost"` // Sum of all costs above
  132. CPUCoreHours float64 `json:"cpuCoreHours"` // Usage metrics: CPU core-hours
  133. RAMByteHours float64 `json:"ramByteHours"` // Usage metrics: RAM byte-hours
  134. GPUHours float64 `json:"gpuHours"` // Usage metrics: GPU-hours
  135. PVByteHours float64 `json:"pvByteHours"` // Usage metrics: PV byte-hours
  136. Start time.Time `json:"start"` // Start timestamp for this allocation
  137. End time.Time `json:"end"` // End timestamp for this allocation
  138. }
  139. // AssetResponse represents the asset data returned to the AI agent.
  140. type AssetResponse struct {
  141. // The asset data, as a map of asset sets.
  142. Assets map[string]*AssetSet `json:"assets"`
  143. }
  144. // AssetSet represents a set of asset data.
  145. type AssetSet struct {
  146. // The name of the asset set.
  147. Name string `json:"name"`
  148. // The asset data for the set.
  149. Assets []*Asset `json:"assets"`
  150. }
  151. // Asset represents a single asset data point.
  152. type Asset struct {
  153. Type string `json:"type"`
  154. Properties AssetProperties `json:"properties"`
  155. Labels map[string]string `json:"labels,omitempty"`
  156. Start time.Time `json:"start"`
  157. End time.Time `json:"end"`
  158. Minutes float64 `json:"minutes"`
  159. Adjustment float64 `json:"adjustment"`
  160. TotalCost float64 `json:"totalCost"`
  161. // Disk-specific fields
  162. ByteHours float64 `json:"byteHours,omitempty"`
  163. ByteHoursUsed *float64 `json:"byteHoursUsed,omitempty"`
  164. ByteUsageMax *float64 `json:"byteUsageMax,omitempty"`
  165. StorageClass string `json:"storageClass,omitempty"`
  166. VolumeName string `json:"volumeName,omitempty"`
  167. ClaimName string `json:"claimName,omitempty"`
  168. ClaimNamespace string `json:"claimNamespace,omitempty"`
  169. Local float64 `json:"local,omitempty"`
  170. // Node-specific fields
  171. NodeType string `json:"nodeType,omitempty"`
  172. CPUCoreHours float64 `json:"cpuCoreHours,omitempty"`
  173. RAMByteHours float64 `json:"ramByteHours,omitempty"`
  174. GPUHours float64 `json:"gpuHours,omitempty"`
  175. GPUCount float64 `json:"gpuCount,omitempty"`
  176. CPUCost float64 `json:"cpuCost,omitempty"`
  177. GPUCost float64 `json:"gpuCost,omitempty"`
  178. RAMCost float64 `json:"ramCost,omitempty"`
  179. Discount float64 `json:"discount,omitempty"`
  180. Preemptible float64 `json:"preemptible,omitempty"`
  181. // Breakdown fields (can be used for different types)
  182. Breakdown *AssetBreakdown `json:"breakdown,omitempty"`
  183. CPUBreakdown *AssetBreakdown `json:"cpuBreakdown,omitempty"`
  184. RAMBreakdown *AssetBreakdown `json:"ramBreakdown,omitempty"`
  185. // Overhead (Node-specific)
  186. Overhead *NodeOverhead `json:"overhead,omitempty"`
  187. // LoadBalancer-specific fields
  188. Private bool `json:"private,omitempty"`
  189. Ip string `json:"ip,omitempty"`
  190. // Cloud-specific fields
  191. Credit float64 `json:"credit,omitempty"`
  192. }
  193. // NodeOverhead represents node overhead information
  194. type NodeOverhead struct {
  195. RamOverheadFraction float64 `json:"ramOverheadFraction"`
  196. CpuOverheadFraction float64 `json:"cpuOverheadFraction"`
  197. OverheadCostFraction float64 `json:"overheadCostFraction"`
  198. }
  199. type AssetProperties struct {
  200. Category string `json:"category,omitempty"`
  201. Provider string `json:"provider,omitempty"`
  202. Account string `json:"account,omitempty"`
  203. Project string `json:"project,omitempty"`
  204. Service string `json:"service,omitempty"`
  205. Cluster string `json:"cluster,omitempty"`
  206. Name string `json:"name,omitempty"`
  207. ProviderID string `json:"providerID,omitempty"`
  208. }
  209. type AssetBreakdown struct {
  210. Idle float64 `json:"idle"`
  211. Other float64 `json:"other"`
  212. System float64 `json:"system"`
  213. User float64 `json:"user"`
  214. }
  215. // CloudCostResponse represents the cloud cost data returned to the AI agent.
  216. type CloudCostResponse struct {
  217. // The cloud cost data, as a map of cloud cost sets.
  218. CloudCosts map[string]*CloudCostSet `json:"cloudCosts"`
  219. // Summary information
  220. Summary *CloudCostSummary `json:"summary,omitempty"`
  221. }
  222. // CloudCostSummary provides summary information about cloud costs
  223. type CloudCostSummary struct {
  224. TotalNetCost float64 `json:"totalNetCost"`
  225. TotalAmortizedCost float64 `json:"totalAmortizedCost"`
  226. TotalInvoicedCost float64 `json:"totalInvoicedCost"`
  227. KubernetesPercent float64 `json:"kubernetesPercent"`
  228. ProviderBreakdown map[string]float64 `json:"providerBreakdown,omitempty"`
  229. ServiceBreakdown map[string]float64 `json:"serviceBreakdown,omitempty"`
  230. RegionBreakdown map[string]float64 `json:"regionBreakdown,omitempty"`
  231. }
  232. // CloudCostSet represents a set of cloud cost data.
  233. type CloudCostSet struct {
  234. // The name of the cloud cost set.
  235. Name string `json:"name"`
  236. // The cloud cost data for the set.
  237. CloudCosts []*CloudCost `json:"cloudCosts"`
  238. // Aggregation information
  239. AggregationProperties []string `json:"aggregationProperties,omitempty"`
  240. // Time window
  241. Window *TimeWindow `json:"window,omitempty"`
  242. }
  243. // TimeWindow represents a time range
  244. type TimeWindow struct {
  245. Start time.Time `json:"start"`
  246. End time.Time `json:"end"`
  247. }
  248. // CloudCostProperties defines the properties of a cloud cost item.
  249. type CloudCostProperties struct {
  250. ProviderID string `json:"providerID,omitempty"`
  251. Provider string `json:"provider,omitempty"`
  252. AccountID string `json:"accountID,omitempty"`
  253. AccountName string `json:"accountName,omitempty"`
  254. InvoiceEntityID string `json:"invoiceEntityID,omitempty"`
  255. InvoiceEntityName string `json:"invoiceEntityName,omitempty"`
  256. RegionID string `json:"regionID,omitempty"`
  257. AvailabilityZone string `json:"availabilityZone,omitempty"`
  258. Service string `json:"service,omitempty"`
  259. Category string `json:"category,omitempty"`
  260. Labels map[string]string `json:"labels,omitempty"`
  261. }
  262. // CloudCost represents a single cloud cost data point.
  263. type CloudCost struct {
  264. Properties CloudCostProperties `json:"properties"`
  265. Window TimeWindow `json:"window"`
  266. ListCost CostMetric `json:"listCost"`
  267. NetCost CostMetric `json:"netCost"`
  268. AmortizedNetCost CostMetric `json:"amortizedNetCost"`
  269. InvoicedCost CostMetric `json:"invoicedCost"`
  270. AmortizedCost CostMetric `json:"amortizedCost"`
  271. }
  272. // CostMetric represents a cost value with Kubernetes percentage
  273. type CostMetric struct {
  274. Cost float64 `json:"cost"`
  275. KubernetesPercent float64 `json:"kubernetesPercent"`
  276. }
  277. // EfficiencyResponse represents the efficiency data returned to the AI agent.
  278. type EfficiencyResponse struct {
  279. Efficiencies []*EfficiencyMetric `json:"efficiencies"`
  280. }
  281. // EfficiencyMetric represents efficiency data for a single pod/workload.
  282. type EfficiencyMetric struct {
  283. Name string `json:"name"` // Pod/namespace/controller name based on aggregation
  284. // Current state
  285. CPUEfficiency float64 `json:"cpuEfficiency"` // Usage / Request ratio (0-1+)
  286. MemoryEfficiency float64 `json:"memoryEfficiency"` // Usage / Request ratio (0-1+)
  287. // Current requests and usage
  288. CPUCoresRequested float64 `json:"cpuCoresRequested"`
  289. CPUCoresUsed float64 `json:"cpuCoresUsed"`
  290. RAMBytesRequested float64 `json:"ramBytesRequested"`
  291. RAMBytesUsed float64 `json:"ramBytesUsed"`
  292. // Recommendations (based on actual usage with buffer)
  293. RecommendedCPURequest float64 `json:"recommendedCpuRequest"` // Recommended CPU cores
  294. RecommendedRAMRequest float64 `json:"recommendedRamRequest"` // Recommended RAM bytes
  295. // Resulting efficiency after applying recommendations
  296. ResultingCPUEfficiency float64 `json:"resultingCpuEfficiency"`
  297. ResultingMemoryEfficiency float64 `json:"resultingMemoryEfficiency"`
  298. // Cost analysis
  299. CurrentTotalCost float64 `json:"currentTotalCost"` // Current total cost
  300. RecommendedCost float64 `json:"recommendedCost"` // Estimated cost with recommendations
  301. CostSavings float64 `json:"costSavings"` // Potential savings
  302. CostSavingsPercent float64 `json:"costSavingsPercent"` // Savings as percentage
  303. // Buffer multiplier used for recommendations
  304. EfficiencyBufferMultiplier float64 `json:"efficiencyBufferMultiplier"` // Buffer multiplier applied (e.g., 1.2 for 20% headroom)
  305. // Time window
  306. Start time.Time `json:"start"`
  307. End time.Time `json:"end"`
  308. }
  309. // MCPServer holds the dependencies for the MCP API server.
  310. type MCPServer struct {
  311. costModel *costmodel.CostModel
  312. provider models.Provider
  313. cloudQuerier cloudcost.Querier
  314. }
  315. // NewMCPServer creates a new MCP Server.
  316. func NewMCPServer(costModel *costmodel.CostModel, provider models.Provider, cloudQuerier cloudcost.Querier) *MCPServer {
  317. return &MCPServer{
  318. costModel: costModel,
  319. provider: provider,
  320. cloudQuerier: cloudQuerier,
  321. }
  322. }
  323. // ProcessMCPRequest processes an MCP request and returns an MCP response.
  324. // It accepts a context for proper timeout handling and cancellation.
  325. func (s *MCPServer) ProcessMCPRequest(ctx context.Context, request *MCPRequest) (*MCPResponse, error) {
  326. // 1. Validate Request
  327. if err := validate.Struct(request); err != nil {
  328. return nil, fmt.Errorf("validation failed: %w", err)
  329. }
  330. // 2. Query Dispatching
  331. var data interface{}
  332. var err error
  333. queryStart := time.Now()
  334. switch request.Query.QueryType {
  335. case AllocationQueryType:
  336. data, err = s.QueryAllocations(request.Query)
  337. case AssetQueryType:
  338. data, err = s.QueryAssets(request.Query)
  339. case CloudCostQueryType:
  340. data, err = s.QueryCloudCosts(ctx, request.Query)
  341. case EfficiencyQueryType:
  342. data, err = s.QueryEfficiency(request.Query)
  343. default:
  344. return nil, fmt.Errorf("unsupported query type: %s", request.Query.QueryType)
  345. }
  346. if err != nil {
  347. // Handle error appropriately, maybe return a JSON-RPC error response
  348. return nil, err
  349. }
  350. // The MCP SDK marshals tool output with encoding/json, which errors on
  351. // non-finite floats. Upstream cost calculations can yield NaN or Inf (e.g.
  352. // a 0/0 breakdown or overhead fraction), so scrub them before they reach the
  353. // SDK and fail the whole tool call.
  354. data = sanitizeNonFiniteFloats(data)
  355. processingTime := time.Since(queryStart)
  356. // 3. Construct Final Response
  357. mcpResponse := &MCPResponse{
  358. Data: data,
  359. QueryInfo: QueryMetadata{
  360. QueryID: generateQueryID(),
  361. Timestamp: time.Now(),
  362. ProcessingTime: processingTime,
  363. },
  364. }
  365. return mcpResponse, nil
  366. }
  367. // sanitizeNonFiniteFloats returns v with every non-finite float (NaN, +Inf,
  368. // -Inf) replaced by 0 so the value can be marshaled by encoding/json, which the
  369. // MCP SDK uses and which rejects non-finite floats. Callers must use the return
  370. // value, since value-type inputs are sanitized on a copy. It is best-effort:
  371. // any reflection panic is recovered and the original value returned unchanged.
  372. func sanitizeNonFiniteFloats(v any) (out any) {
  373. out = v
  374. defer func() {
  375. if r := recover(); r != nil {
  376. log.Warnf("mcp: sanitizeNonFiniteFloats recovered: %v", r)
  377. out = v
  378. }
  379. }()
  380. if v == nil {
  381. return nil
  382. }
  383. // Work on an addressable copy so value-type inputs are sanitized too, not
  384. // only pointers. For a pointer input this copies the pointer and mutates the
  385. // pointed-to value in place; for a value input it yields a sanitized copy.
  386. box := reflect.New(reflect.TypeOf(v))
  387. box.Elem().Set(reflect.ValueOf(v))
  388. sanitizeFloatsValue(box.Elem())
  389. return box.Elem().Interface()
  390. }
  391. func sanitizeFloatsValue(v reflect.Value) {
  392. switch v.Kind() {
  393. case reflect.Ptr, reflect.Interface:
  394. if !v.IsNil() {
  395. sanitizeFloatsValue(v.Elem())
  396. }
  397. case reflect.Struct:
  398. for i := 0; i < v.NumField(); i++ {
  399. sanitizeFloatsValue(v.Field(i))
  400. }
  401. case reflect.Slice, reflect.Array:
  402. for i := 0; i < v.Len(); i++ {
  403. sanitizeFloatsValue(v.Index(i))
  404. }
  405. case reflect.Map:
  406. for _, key := range v.MapKeys() {
  407. elem := v.MapIndex(key)
  408. // Map elements aren't addressable. Pointer/interface/slice/map
  409. // values are mutated in place by recursing; value-type elements
  410. // (e.g. a float or struct) must be rebuilt and reassigned.
  411. switch elem.Kind() {
  412. case reflect.Ptr, reflect.Interface, reflect.Slice, reflect.Map:
  413. sanitizeFloatsValue(elem)
  414. default:
  415. tmp := reflect.New(elem.Type()).Elem()
  416. tmp.Set(elem)
  417. sanitizeFloatsValue(tmp)
  418. v.SetMapIndex(key, tmp)
  419. }
  420. }
  421. case reflect.Float32, reflect.Float64:
  422. if v.CanSet() && (math.IsNaN(v.Float()) || math.IsInf(v.Float(), 0)) {
  423. v.SetFloat(0)
  424. }
  425. }
  426. }
  427. // validate is the singleton validator instance.
  428. var validate = validator.New()
  429. func generateQueryID() string {
  430. bytes := make([]byte, 8) // 16 hex characters
  431. if _, err := rand.Read(bytes); err != nil {
  432. // Fallback to timestamp-based ID if crypto/rand fails
  433. return fmt.Sprintf("query-%d", time.Now().UnixNano())
  434. }
  435. return fmt.Sprintf("query-%s", hex.EncodeToString(bytes))
  436. }
  437. func (s *MCPServer) QueryAllocations(query *OpenCostQueryRequest) (*AllocationResponse, error) {
  438. // 1. Parse Window
  439. window, err := opencost.ParseWindowWithOffset(query.Window, 0) // 0 offset for UTC
  440. if err != nil {
  441. return nil, fmt.Errorf("failed to parse window '%s': %w", query.Window, err)
  442. }
  443. // 2. Set default parameters
  444. var step time.Duration
  445. var aggregateBy []string
  446. var includeIdle, idleByNode, includeProportionalAssetResourceCosts, includeAggregatedMetadata, sharedLoadBalancer, shareIdle bool
  447. var accumulateBy opencost.AccumulateOption
  448. var filterString string
  449. // 3. Parse allocation parameters if provided
  450. if query.AllocationParams != nil {
  451. // Set step duration (default to window duration if not specified)
  452. if query.AllocationParams.Step > 0 {
  453. step = query.AllocationParams.Step
  454. } else {
  455. step = window.Duration()
  456. }
  457. // Parse aggregation properties
  458. if query.AllocationParams.Aggregate != "" {
  459. aggregateBy = strings.Split(query.AllocationParams.Aggregate, ",")
  460. }
  461. // Set boolean parameters
  462. includeIdle = query.AllocationParams.IncludeIdle
  463. idleByNode = query.AllocationParams.IdleByNode
  464. includeProportionalAssetResourceCosts = query.AllocationParams.IncludeProportionalAssetResourceCosts
  465. includeAggregatedMetadata = query.AllocationParams.IncludeAggregatedMetadata
  466. sharedLoadBalancer = query.AllocationParams.ShareLB
  467. shareIdle = query.AllocationParams.ShareIdle
  468. // Set filter string
  469. filterString = query.AllocationParams.Filter
  470. // Validate filter string if provided
  471. if filterString != "" {
  472. parser := allocation.NewAllocationFilterParser()
  473. _, err := parser.Parse(filterString)
  474. if err != nil {
  475. return nil, fmt.Errorf("invalid allocation filter '%s': %w", filterString, err)
  476. }
  477. }
  478. // Set accumulation option
  479. if query.AllocationParams.Accumulate {
  480. accumulateBy = opencost.AccumulateOptionAll
  481. } else {
  482. accumulateBy = opencost.AccumulateOptionNone
  483. }
  484. } else {
  485. // Default values when no parameters provided
  486. step = window.Duration()
  487. accumulateBy = opencost.AccumulateOptionNone
  488. filterString = ""
  489. }
  490. // 4. Call the existing QueryAllocation function with all parameters
  491. asr, err := s.costModel.QueryAllocation(
  492. window,
  493. step,
  494. aggregateBy,
  495. includeIdle,
  496. idleByNode,
  497. includeProportionalAssetResourceCosts,
  498. includeAggregatedMetadata,
  499. sharedLoadBalancer,
  500. accumulateBy,
  501. shareIdle,
  502. filterString,
  503. )
  504. if err != nil {
  505. return nil, fmt.Errorf("failed to query allocations: %w", err)
  506. }
  507. // 5. Handle the AllocationSetRange result
  508. if asr == nil || len(asr.Allocations) == 0 {
  509. return &AllocationResponse{
  510. Allocations: make(map[string]*AllocationSet),
  511. }, nil
  512. }
  513. // 6. Transform the result to MCP format
  514. // If we have multiple sets, we'll combine them or return the first one
  515. // For now, let's return the first allocation set
  516. firstSet := asr.Allocations[0]
  517. return transformAllocationSet(firstSet), nil
  518. }
  519. // transformAllocationSet converts an opencost.AllocationSet into the MCP's AllocationResponse format.
  520. func transformAllocationSet(allocSet *opencost.AllocationSet) *AllocationResponse {
  521. if allocSet == nil {
  522. return &AllocationResponse{Allocations: make(map[string]*AllocationSet)}
  523. }
  524. mcpAllocations := make(map[string]*AllocationSet)
  525. // Create a single set for all allocations
  526. mcpSet := &AllocationSet{
  527. Name: "allocations",
  528. Allocations: []*Allocation{},
  529. }
  530. // Convert each allocation
  531. for _, alloc := range allocSet.Allocations {
  532. if alloc == nil {
  533. continue
  534. }
  535. mcpAlloc := &Allocation{
  536. Name: alloc.Name,
  537. CPUCost: alloc.CPUCost,
  538. GPUCost: alloc.GPUCost,
  539. RAMCost: alloc.RAMCost,
  540. PVCost: alloc.PVCost(), // Call the method
  541. NetworkCost: alloc.NetworkCost,
  542. SharedCost: alloc.SharedCost,
  543. ExternalCost: alloc.ExternalCost,
  544. TotalCost: alloc.TotalCost(),
  545. CPUCoreHours: alloc.CPUCoreHours,
  546. RAMByteHours: alloc.RAMByteHours,
  547. GPUHours: alloc.GPUHours,
  548. PVByteHours: alloc.PVBytes(), // Use the method directly
  549. Start: alloc.Start,
  550. End: alloc.End,
  551. }
  552. mcpSet.Allocations = append(mcpSet.Allocations, mcpAlloc)
  553. }
  554. mcpAllocations["allocations"] = mcpSet
  555. return &AllocationResponse{
  556. Allocations: mcpAllocations,
  557. }
  558. }
  559. func (s *MCPServer) QueryAssets(query *OpenCostQueryRequest) (*AssetResponse, error) {
  560. // 1. Parse Window
  561. window, err := opencost.ParseWindowWithOffset(query.Window, 0) // 0 offset for UTC
  562. if err != nil {
  563. return nil, fmt.Errorf("failed to parse window '%s': %w", query.Window, err)
  564. }
  565. // 2. Set Query Options
  566. start := *window.Start()
  567. end := *window.End()
  568. // 3. Call CostModel to get the asset set
  569. assetSet, err := s.costModel.ComputeAssets(start, end)
  570. if err != nil {
  571. return nil, fmt.Errorf("failed to compute assets: %w", err)
  572. }
  573. // 4. Transform Response for the MCP API
  574. return transformAssetSet(assetSet), nil
  575. }
  576. // transformAssetSet converts a opencost.AssetSet into the MCP's AssetResponse format.
  577. func transformAssetSet(assetSet *opencost.AssetSet) *AssetResponse {
  578. if assetSet == nil {
  579. return &AssetResponse{Assets: make(map[string]*AssetSet)}
  580. }
  581. mcpAssets := make(map[string]*AssetSet)
  582. // Create a single set for all assets
  583. mcpSet := &AssetSet{
  584. Name: "assets",
  585. Assets: []*Asset{},
  586. }
  587. for _, asset := range assetSet.Assets {
  588. if asset == nil {
  589. continue
  590. }
  591. properties := asset.GetProperties()
  592. labels := asset.GetLabels()
  593. mcpAsset := &Asset{
  594. Type: asset.Type().String(),
  595. Properties: AssetProperties{
  596. Category: properties.Category,
  597. Provider: properties.Provider,
  598. Account: properties.Account,
  599. Project: properties.Project,
  600. Service: properties.Service,
  601. Cluster: properties.Cluster,
  602. Name: properties.Name,
  603. ProviderID: properties.ProviderID,
  604. },
  605. Labels: labels,
  606. Start: asset.GetStart(),
  607. End: asset.GetEnd(),
  608. Minutes: asset.Minutes(),
  609. Adjustment: asset.GetAdjustment(),
  610. TotalCost: asset.TotalCost(),
  611. }
  612. // Handle type-specific fields
  613. switch a := asset.(type) {
  614. case *opencost.Disk:
  615. mcpAsset.ByteHours = a.ByteHours
  616. mcpAsset.ByteHoursUsed = a.ByteHoursUsed
  617. mcpAsset.ByteUsageMax = a.ByteUsageMax
  618. mcpAsset.StorageClass = a.StorageClass
  619. mcpAsset.VolumeName = a.VolumeName
  620. mcpAsset.ClaimName = a.ClaimName
  621. mcpAsset.ClaimNamespace = a.ClaimNamespace
  622. mcpAsset.Local = a.Local
  623. if a.Breakdown != nil {
  624. mcpAsset.Breakdown = &AssetBreakdown{
  625. Idle: a.Breakdown.Idle,
  626. Other: a.Breakdown.Other,
  627. System: a.Breakdown.System,
  628. User: a.Breakdown.User,
  629. }
  630. }
  631. case *opencost.Node:
  632. mcpAsset.NodeType = a.NodeType
  633. mcpAsset.CPUCoreHours = a.CPUCoreHours
  634. mcpAsset.RAMByteHours = a.RAMByteHours
  635. mcpAsset.GPUHours = a.GPUHours
  636. mcpAsset.GPUCount = a.GPUCount
  637. mcpAsset.CPUCost = a.CPUCost
  638. mcpAsset.GPUCost = a.GPUCost
  639. mcpAsset.RAMCost = a.RAMCost
  640. mcpAsset.Discount = a.Discount
  641. mcpAsset.Preemptible = a.Preemptible
  642. if a.CPUBreakdown != nil {
  643. mcpAsset.CPUBreakdown = &AssetBreakdown{
  644. Idle: a.CPUBreakdown.Idle,
  645. Other: a.CPUBreakdown.Other,
  646. System: a.CPUBreakdown.System,
  647. User: a.CPUBreakdown.User,
  648. }
  649. }
  650. if a.RAMBreakdown != nil {
  651. mcpAsset.RAMBreakdown = &AssetBreakdown{
  652. Idle: a.RAMBreakdown.Idle,
  653. Other: a.RAMBreakdown.Other,
  654. System: a.RAMBreakdown.System,
  655. User: a.RAMBreakdown.User,
  656. }
  657. }
  658. if a.Overhead != nil {
  659. mcpAsset.Overhead = &NodeOverhead{
  660. RamOverheadFraction: a.Overhead.RamOverheadFraction,
  661. CpuOverheadFraction: a.Overhead.CpuOverheadFraction,
  662. OverheadCostFraction: a.Overhead.OverheadCostFraction,
  663. }
  664. }
  665. case *opencost.LoadBalancer:
  666. mcpAsset.Private = a.Private
  667. mcpAsset.Ip = a.Ip
  668. case *opencost.Network:
  669. // Network assets have no specific fields beyond the base asset structure
  670. // All relevant data is in Properties, Labels, Cost, etc.
  671. case *opencost.Cloud:
  672. mcpAsset.Credit = a.Credit
  673. case *opencost.ClusterManagement:
  674. // ClusterManagement assets have no specific fields beyond the base asset structure
  675. // All relevant data is in Properties, Labels, Cost, etc.
  676. }
  677. mcpSet.Assets = append(mcpSet.Assets, mcpAsset)
  678. }
  679. mcpAssets["assets"] = mcpSet
  680. return &AssetResponse{
  681. Assets: mcpAssets,
  682. }
  683. }
  684. // QueryCloudCosts translates an MCP query into a CloudCost repository query and transforms the result.
  685. // The ctx parameter is used for timeout and cancellation handling of the cloud cost query.
  686. func (s *MCPServer) QueryCloudCosts(ctx context.Context, query *OpenCostQueryRequest) (*CloudCostResponse, error) {
  687. // 1. Check if cloud cost querier is available
  688. if s.cloudQuerier == nil {
  689. return nil, fmt.Errorf("cloud cost querier not configured - check cloud-integration.json file")
  690. }
  691. // 2. Parse Window
  692. window, err := opencost.ParseWindowWithOffset(query.Window, 0) // 0 offset for UTC
  693. if err != nil {
  694. return nil, fmt.Errorf("failed to parse window '%s': %w", query.Window, err)
  695. }
  696. // 3. Build query request
  697. request := cloudcost.QueryRequest{
  698. Start: *window.Start(),
  699. End: *window.End(),
  700. Filter: nil, // Will be set from CloudCostParams if provided
  701. }
  702. // 4. Apply filtering and aggregation from CloudCostParams
  703. if query.CloudCostParams != nil {
  704. request = s.buildCloudCostQueryRequest(request, query.CloudCostParams)
  705. }
  706. // 5. Create a timeout context for the query with configured timeout
  707. queryTimeout := env.GetMCPQueryTimeout()
  708. queryCtx, cancel := context.WithTimeout(ctx, queryTimeout)
  709. defer cancel()
  710. // 6. Query the repository (this handles multiple cloud providers automatically)
  711. ccsr, err := s.cloudQuerier.Query(queryCtx, request)
  712. if err != nil {
  713. return nil, fmt.Errorf("failed to query cloud costs: %w", err)
  714. }
  715. // 7. Transform Response
  716. return transformCloudCostSetRange(ccsr), nil
  717. }
  718. // buildCloudCostQueryRequest builds a QueryRequest from CloudCostParams
  719. func (s *MCPServer) buildCloudCostQueryRequest(request cloudcost.QueryRequest, params *CloudCostQuery) cloudcost.QueryRequest {
  720. // Set aggregation
  721. if params.Aggregate != "" {
  722. aggregateBy := strings.Split(params.Aggregate, ",")
  723. request.AggregateBy = aggregateBy
  724. }
  725. // Set accumulation
  726. if params.Accumulate != "" {
  727. request.Accumulate = opencost.ParseAccumulate(params.Accumulate)
  728. }
  729. // Build filter from individual parameters or filter string
  730. var filter filter.Filter
  731. var err error
  732. if params.Filter != "" {
  733. // Parse the filter string directly
  734. parser := cloudcostfilter.NewCloudCostFilterParser()
  735. filter, err = parser.Parse(params.Filter)
  736. if err != nil {
  737. // Log error but continue without filter rather than failing the entire request
  738. log.Warnf("failed to parse filter string '%s': %v", params.Filter, err)
  739. }
  740. } else {
  741. // Build filter from individual parameters
  742. filter = s.buildFilterFromParams(params)
  743. }
  744. request.Filter = filter
  745. return request
  746. }
  747. // buildFilterFromParams creates a filter from individual CloudCostQuery parameters
  748. func (s *MCPServer) buildFilterFromParams(params *CloudCostQuery) filter.Filter {
  749. var filterParts []string
  750. // Add provider filter
  751. if params.Provider != "" {
  752. filterParts = append(filterParts, fmt.Sprintf(`provider:"%s"`, params.Provider))
  753. }
  754. // Add providerID filter
  755. if params.ProviderID != "" {
  756. filterParts = append(filterParts, fmt.Sprintf(`providerID:"%s"`, params.ProviderID))
  757. }
  758. // Add service filter
  759. if params.Service != "" {
  760. filterParts = append(filterParts, fmt.Sprintf(`service:"%s"`, params.Service))
  761. }
  762. // Add category filter
  763. if params.Category != "" {
  764. filterParts = append(filterParts, fmt.Sprintf(`category:"%s"`, params.Category))
  765. }
  766. // Region is intentionally not supported here
  767. // Add account filter (maps to accountID)
  768. if params.AccountID != "" {
  769. filterParts = append(filterParts, fmt.Sprintf(`accountID:"%s"`, params.AccountID))
  770. }
  771. // Add invoiceEntityID filter
  772. if params.InvoiceEntityID != "" {
  773. filterParts = append(filterParts, fmt.Sprintf(`invoiceEntityID:"%s"`, params.InvoiceEntityID))
  774. }
  775. // Add label filters (label[key]:"value")
  776. if len(params.Labels) > 0 {
  777. for k, v := range params.Labels {
  778. if k == "" {
  779. continue
  780. }
  781. filterParts = append(filterParts, fmt.Sprintf(`label[%s]:"%s"`, k, v))
  782. }
  783. }
  784. // If no filters specified, return nil
  785. if len(filterParts) == 0 {
  786. return nil
  787. }
  788. // Combine all filter parts with AND logic (parser expects 'and')
  789. filterString := strings.Join(filterParts, " and ")
  790. // Parse the combined filter string
  791. parser := cloudcostfilter.NewCloudCostFilterParser()
  792. filter, err := parser.Parse(filterString)
  793. if err != nil {
  794. // Log error but return nil rather than failing
  795. log.Warnf("failed to parse combined filter '%s': %v", filterString, err)
  796. return nil
  797. }
  798. return filter
  799. }
  800. // transformCloudCostSetRange converts a opencost.CloudCostSetRange into the MCP's CloudCostResponse format.
  801. func transformCloudCostSetRange(ccsr *opencost.CloudCostSetRange) *CloudCostResponse {
  802. if ccsr == nil || len(ccsr.CloudCostSets) == 0 {
  803. return &CloudCostResponse{
  804. CloudCosts: make(map[string]*CloudCostSet),
  805. Summary: &CloudCostSummary{
  806. TotalNetCost: 0,
  807. },
  808. }
  809. }
  810. mcpCloudCosts := make(map[string]*CloudCostSet)
  811. var totalNetCost, totalAmortizedCost, totalInvoicedCost float64
  812. providerBreakdown := make(map[string]float64)
  813. serviceBreakdown := make(map[string]float64)
  814. regionBreakdown := make(map[string]float64)
  815. // Process each cloud cost set in the range
  816. for i, ccSet := range ccsr.CloudCostSets {
  817. if ccSet == nil {
  818. log.Warnf("transformCloudCostSetRange: skipping nil CloudCostSet at index %d", i)
  819. continue
  820. }
  821. // Check for nil Window or nil Start/End pointers before dereferencing
  822. if ccSet.Window.Start() == nil || ccSet.Window.End() == nil {
  823. log.Warnf("transformCloudCostSetRange: skipping CloudCostSet at index %d with invalid window (start=%v, end=%v)", i, ccSet.Window.Start(), ccSet.Window.End())
  824. continue
  825. }
  826. setName := fmt.Sprintf("cloudcosts_%d", i)
  827. mcpSet := &CloudCostSet{
  828. Name: setName,
  829. CloudCosts: []*CloudCost{},
  830. AggregationProperties: ccSet.AggregationProperties,
  831. Window: &TimeWindow{
  832. Start: *ccSet.Window.Start(),
  833. End: *ccSet.Window.End(),
  834. },
  835. }
  836. // Convert each cloud cost item
  837. for _, item := range ccSet.CloudCosts {
  838. if item == nil {
  839. log.Warnf("transformCloudCostSetRange: skipping nil CloudCost item in set %s", setName)
  840. continue
  841. }
  842. // Check for nil Window or nil Start/End pointers on the item
  843. if item.Window.Start() == nil || item.Window.End() == nil {
  844. log.Warnf("transformCloudCostSetRange: skipping CloudCost item with invalid window (start=%v, end=%v) in set %s", item.Window.Start(), item.Window.End(), setName)
  845. continue
  846. }
  847. mcpCC := &CloudCost{
  848. Properties: CloudCostProperties{
  849. ProviderID: item.Properties.ProviderID,
  850. Provider: item.Properties.Provider,
  851. AccountID: item.Properties.AccountID,
  852. AccountName: item.Properties.AccountName,
  853. InvoiceEntityID: item.Properties.InvoiceEntityID,
  854. InvoiceEntityName: item.Properties.InvoiceEntityName,
  855. RegionID: item.Properties.RegionID,
  856. AvailabilityZone: item.Properties.AvailabilityZone,
  857. Service: item.Properties.Service,
  858. Category: item.Properties.Category,
  859. Labels: item.Properties.Labels,
  860. },
  861. Window: TimeWindow{
  862. Start: *item.Window.Start(),
  863. End: *item.Window.End(),
  864. },
  865. ListCost: CostMetric{
  866. Cost: item.ListCost.Cost,
  867. KubernetesPercent: item.ListCost.KubernetesPercent,
  868. },
  869. NetCost: CostMetric{
  870. Cost: item.NetCost.Cost,
  871. KubernetesPercent: item.NetCost.KubernetesPercent,
  872. },
  873. AmortizedNetCost: CostMetric{
  874. Cost: item.AmortizedNetCost.Cost,
  875. KubernetesPercent: item.AmortizedNetCost.KubernetesPercent,
  876. },
  877. InvoicedCost: CostMetric{
  878. Cost: item.InvoicedCost.Cost,
  879. KubernetesPercent: item.InvoicedCost.KubernetesPercent,
  880. },
  881. AmortizedCost: CostMetric{
  882. Cost: item.AmortizedCost.Cost,
  883. KubernetesPercent: item.AmortizedCost.KubernetesPercent,
  884. },
  885. }
  886. mcpSet.CloudCosts = append(mcpSet.CloudCosts, mcpCC)
  887. // Update summary totals
  888. totalNetCost += item.NetCost.Cost
  889. totalAmortizedCost += item.AmortizedNetCost.Cost
  890. totalInvoicedCost += item.InvoicedCost.Cost
  891. // Update breakdowns
  892. providerBreakdown[item.Properties.Provider] += item.NetCost.Cost
  893. serviceBreakdown[item.Properties.Service] += item.NetCost.Cost
  894. regionBreakdown[item.Properties.RegionID] += item.NetCost.Cost
  895. }
  896. mcpCloudCosts[setName] = mcpSet
  897. }
  898. // Calculate cost-weighted average Kubernetes percentage (by NetCost)
  899. var avgKubernetesPercent float64
  900. var numerator, denominator float64
  901. for _, ccSet := range ccsr.CloudCostSets {
  902. if ccSet == nil {
  903. log.Warnf("transformCloudCostSetRange: skipping nil CloudCostSet in Kubernetes percent calculation")
  904. continue
  905. }
  906. // Skip sets with invalid windows (consistent with first loop)
  907. if ccSet.Window.Start() == nil || ccSet.Window.End() == nil {
  908. log.Warnf("transformCloudCostSetRange: skipping CloudCostSet with invalid window (start=%v, end=%v) in Kubernetes percent calculation", ccSet.Window.Start(), ccSet.Window.End())
  909. continue
  910. }
  911. for _, item := range ccSet.CloudCosts {
  912. if item == nil {
  913. log.Warnf("transformCloudCostSetRange: skipping nil CloudCost item in Kubernetes percent calculation")
  914. continue
  915. }
  916. // Skip items with invalid windows (consistent with first loop)
  917. if item.Window.Start() == nil || item.Window.End() == nil {
  918. log.Warnf("transformCloudCostSetRange: skipping CloudCost item with invalid window (start=%v, end=%v) in Kubernetes percent calculation", item.Window.Start(), item.Window.End())
  919. continue
  920. }
  921. cost := item.NetCost.Cost
  922. percent := item.NetCost.KubernetesPercent
  923. if cost <= 0 {
  924. continue
  925. }
  926. numerator += cost * percent
  927. denominator += cost
  928. }
  929. }
  930. if denominator > 0 {
  931. avgKubernetesPercent = numerator / denominator
  932. }
  933. summary := &CloudCostSummary{
  934. TotalNetCost: totalNetCost,
  935. TotalAmortizedCost: totalAmortizedCost,
  936. TotalInvoicedCost: totalInvoicedCost,
  937. KubernetesPercent: avgKubernetesPercent,
  938. ProviderBreakdown: providerBreakdown,
  939. ServiceBreakdown: serviceBreakdown,
  940. RegionBreakdown: regionBreakdown,
  941. }
  942. return &CloudCostResponse{
  943. CloudCosts: mcpCloudCosts,
  944. Summary: summary,
  945. }
  946. }
  947. // defaultEfficiencyStep returns a step duration that keeps peak memory
  948. // bounded for large query windows. When the caller does not specify a step,
  949. // this provides a safe default that avoids loading the entire window into
  950. // memory at once.
  951. func defaultEfficiencyStep(windowDuration time.Duration) time.Duration {
  952. switch {
  953. case windowDuration >= 30*24*time.Hour:
  954. return 24 * time.Hour
  955. case windowDuration >= 7*24*time.Hour:
  956. return 6 * time.Hour
  957. case windowDuration >= 24*time.Hour:
  958. return time.Hour
  959. default:
  960. return windowDuration
  961. }
  962. }
  963. // QueryEfficiency queries allocation data and computes efficiency metrics with recommendations.
  964. func (s *MCPServer) QueryEfficiency(query *OpenCostQueryRequest) (*EfficiencyResponse, error) {
  965. // 1. Parse Window
  966. window, err := opencost.ParseWindowWithOffset(query.Window, 0)
  967. if err != nil {
  968. return nil, fmt.Errorf("failed to parse window '%s': %w", query.Window, err)
  969. }
  970. // 2. Set default parameters
  971. var aggregateBy []string
  972. var filterString string
  973. var bufferMultiplier float64 = efficiencyBufferMultiplier // Default to 1.2 (20% headroom)
  974. // 3. Parse efficiency parameters if provided
  975. if query.EfficiencyParams != nil {
  976. // Parse aggregation properties (default to pod if not specified)
  977. if query.EfficiencyParams.Aggregate != "" {
  978. aggregateBy = strings.Split(query.EfficiencyParams.Aggregate, ",")
  979. } else {
  980. aggregateBy = []string{"pod"}
  981. }
  982. // Set filter string
  983. filterString = query.EfficiencyParams.Filter
  984. // Validate filter string if provided
  985. if filterString != "" {
  986. parser := allocation.NewAllocationFilterParser()
  987. _, err := parser.Parse(filterString)
  988. if err != nil {
  989. return nil, fmt.Errorf("invalid allocation filter '%s': %w", filterString, err)
  990. }
  991. }
  992. // Set buffer multiplier if provided, otherwise use default
  993. if query.EfficiencyParams.EfficiencyBufferMultiplier != nil {
  994. bufferMultiplier = *query.EfficiencyParams.EfficiencyBufferMultiplier
  995. }
  996. } else {
  997. // Default to pod-level aggregation
  998. aggregateBy = []string{"pod"}
  999. filterString = ""
  1000. }
  1001. // 4. Determine query step size.
  1002. // A smaller step reduces peak memory by breaking large windows into batches.
  1003. // Results are accumulated so the output is functionally equivalent regardless
  1004. // of step, though minor floating-point differences are possible because
  1005. // per-step cost calculations (which use max(request, usage)) are summed
  1006. // rather than computed in a single pass.
  1007. var step time.Duration
  1008. if query.EfficiencyParams != nil && query.EfficiencyParams.Step > 0 {
  1009. step = query.EfficiencyParams.Step
  1010. } else {
  1011. step = defaultEfficiencyStep(window.Duration())
  1012. }
  1013. if step > window.Duration() {
  1014. step = window.Duration()
  1015. }
  1016. if step <= 0 {
  1017. return nil, fmt.Errorf("invalid query: window has zero or negative duration")
  1018. }
  1019. accumulateBy := opencost.AccumulateOptionNone
  1020. if step < window.Duration() {
  1021. accumulateBy = opencost.AccumulateOptionAll
  1022. }
  1023. asr, err := s.costModel.QueryAllocation(
  1024. window,
  1025. step,
  1026. aggregateBy,
  1027. false, // includeIdle
  1028. false, // idleByNode
  1029. false, // includeProportionalAssetResourceCosts
  1030. false, // includeAggregatedMetadata
  1031. false, // sharedLoadBalancer
  1032. accumulateBy,
  1033. false, // shareIdle
  1034. filterString,
  1035. )
  1036. if err != nil {
  1037. return nil, fmt.Errorf("failed to query allocations: %w", err)
  1038. }
  1039. // 5. Handle empty results
  1040. if asr == nil || len(asr.Allocations) == 0 {
  1041. return &EfficiencyResponse{
  1042. Efficiencies: []*EfficiencyMetric{},
  1043. }, nil
  1044. }
  1045. // 6. Compute efficiency metrics from allocations using concurrent processing
  1046. var (
  1047. mu sync.Mutex
  1048. wg sync.WaitGroup
  1049. efficiencies = make([]*EfficiencyMetric, 0)
  1050. )
  1051. // Process each allocation set (typically one per time window) concurrently
  1052. for _, allocSet := range asr.Allocations {
  1053. if allocSet == nil {
  1054. continue
  1055. }
  1056. // Process this allocation set in a goroutine
  1057. wg.Add(1)
  1058. go func(allocSet *opencost.AllocationSet) {
  1059. defer wg.Done()
  1060. // Compute metrics for all allocations in this set
  1061. localMetrics := make([]*EfficiencyMetric, 0, len(allocSet.Allocations))
  1062. for _, alloc := range allocSet.Allocations {
  1063. if metric := computeEfficiencyMetric(alloc, bufferMultiplier); metric != nil {
  1064. localMetrics = append(localMetrics, metric)
  1065. }
  1066. }
  1067. // Append results to shared slice (thread-safe)
  1068. if len(localMetrics) > 0 {
  1069. mu.Lock()
  1070. efficiencies = append(efficiencies, localMetrics...)
  1071. mu.Unlock()
  1072. }
  1073. }(allocSet)
  1074. }
  1075. // Wait for all goroutines to complete
  1076. wg.Wait()
  1077. return &EfficiencyResponse{
  1078. Efficiencies: efficiencies,
  1079. }, nil
  1080. }
  1081. // safeDiv performs division and returns 0 if denominator is 0.
  1082. func safeDiv(numerator, denominator float64) float64 {
  1083. if denominator == 0 {
  1084. return 0
  1085. }
  1086. return numerator / denominator
  1087. }
  1088. // computeEfficiencyMetric calculates efficiency metrics for a single allocation.
  1089. func computeEfficiencyMetric(alloc *opencost.Allocation, bufferMultiplier float64) *EfficiencyMetric {
  1090. if alloc == nil {
  1091. return nil
  1092. }
  1093. // Calculate time duration in hours
  1094. hours := alloc.Minutes() / 60.0
  1095. if hours <= 0 {
  1096. return nil
  1097. }
  1098. // Get current usage (average over the period)
  1099. cpuCoresUsed := alloc.CPUCoreHours / hours
  1100. ramBytesUsed := alloc.RAMByteHours / hours
  1101. // Get requested amounts
  1102. cpuCoresRequested := alloc.CPUCoreRequestAverage
  1103. ramBytesRequested := alloc.RAMBytesRequestAverage
  1104. // Calculate current efficiency (will be 0 if no requests are set)
  1105. cpuEfficiency := safeDiv(cpuCoresUsed, cpuCoresRequested)
  1106. memoryEfficiency := safeDiv(ramBytesUsed, ramBytesRequested)
  1107. // Calculate recommendations with buffer for headroom
  1108. recommendedCPU := cpuCoresUsed * bufferMultiplier
  1109. recommendedRAM := ramBytesUsed * bufferMultiplier
  1110. // Ensure recommendations meet minimum thresholds
  1111. if recommendedCPU < efficiencyMinCPU {
  1112. recommendedCPU = efficiencyMinCPU
  1113. }
  1114. if recommendedRAM < efficiencyMinRAM {
  1115. recommendedRAM = efficiencyMinRAM
  1116. }
  1117. // Calculate resulting efficiency after applying recommendations
  1118. resultingCPUEff := safeDiv(cpuCoresUsed, recommendedCPU)
  1119. resultingMemEff := safeDiv(ramBytesUsed, recommendedRAM)
  1120. // Calculate cost per unit based on REQUESTED amounts (not used amounts)
  1121. // This gives us the cost per core-hour or byte-hour that the cluster charges
  1122. cpuCostPerCoreHour := safeDiv(alloc.CPUCost, cpuCoresRequested*hours)
  1123. ramCostPerByteHour := safeDiv(alloc.RAMCost, ramBytesRequested*hours)
  1124. // Current total cost
  1125. currentTotalCost := alloc.TotalCost()
  1126. // Estimate recommended cost based on recommended requests
  1127. recommendedCPUCost := recommendedCPU * hours * cpuCostPerCoreHour
  1128. recommendedRAMCost := recommendedRAM * hours * ramCostPerByteHour
  1129. // Keep other costs the same (PV, network, shared, external, GPU)
  1130. otherCosts := alloc.PVCost() + alloc.NetworkCost + alloc.SharedCost + alloc.ExternalCost + alloc.GPUCost
  1131. recommendedTotalCost := recommendedCPUCost + recommendedRAMCost + otherCosts
  1132. // Clamp recommended cost to avoid rounding issues making it higher than current
  1133. if recommendedTotalCost > currentTotalCost && (recommendedTotalCost-currentTotalCost) < 0.0001 {
  1134. recommendedTotalCost = currentTotalCost
  1135. }
  1136. // Calculate savings
  1137. costSavings := currentTotalCost - recommendedTotalCost
  1138. costSavingsPercent := safeDiv(costSavings, currentTotalCost) * 100
  1139. return &EfficiencyMetric{
  1140. Name: alloc.Name,
  1141. CPUEfficiency: cpuEfficiency,
  1142. MemoryEfficiency: memoryEfficiency,
  1143. CPUCoresRequested: cpuCoresRequested,
  1144. CPUCoresUsed: cpuCoresUsed,
  1145. RAMBytesRequested: ramBytesRequested,
  1146. RAMBytesUsed: ramBytesUsed,
  1147. RecommendedCPURequest: recommendedCPU,
  1148. RecommendedRAMRequest: recommendedRAM,
  1149. ResultingCPUEfficiency: resultingCPUEff,
  1150. ResultingMemoryEfficiency: resultingMemEff,
  1151. CurrentTotalCost: currentTotalCost,
  1152. RecommendedCost: recommendedTotalCost,
  1153. CostSavings: costSavings,
  1154. CostSavingsPercent: costSavingsPercent,
  1155. EfficiencyBufferMultiplier: bufferMultiplier,
  1156. Start: alloc.Start,
  1157. End: alloc.End,
  1158. }
  1159. }