feature/pricing-aws

prisma/migrations/20230909140952_cost_estimate/migration.sql

@@ -0,0 +1,6 @@
+-- AlterTable
+ALTER TABLE "Task" ADD COLUMN     "costEstimate" DOUBLE PRECISION,
+ADD COLUMN     "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;
+
+-- CreateIndex
+CREATE INDEX "Task_updatedAt_idx" ON "Task"("updatedAt");

prisma/schema.prisma

@@ -114,7 +114,10 @@ model Task {
   data       Json
   workflowId String
   workflow   Workflow @relation(fields: [workflowId], references: [id])
+  costEstimate       Float?
+  updatedAt DateTime @default(now()) @updatedAt
 
   @@unique([workflowId, taskId])
   @@index([taskId, workflowId])
+  @@index([updatedAt])
 }

src/app/api/cost/route.ts

@@ -0,0 +1,22 @@
+import { NextResponse } from "next/server"
+import { EstimateComputeCost } from "@/services"
+
+type TPriceRequest = {
+	executor: string
+	machineType: string
+	cloudZone: string
+	priceModel: string
+	duration: number
+}
+export async function POST(request: Request) {
+	const r: TPriceRequest = await request.json()
+
+	let costEstimate: number | null = null
+	const pricePerCPUHour = await EstimateComputeCost(r.executor, r.duration, r.machineType, r.cloudZone, r.priceModel)
+	if (pricePerCPUHour) {
+		costEstimate = (pricePerCPUHour * 16 * 25000000) / (3600 * 1000)
+	}
+
+	console.error({ costEstimate, pricePerCPUHour })
+	return NextResponse.json(costEstimate)
+}

src/app/api/runs/[id]/route.ts

@@ -12,7 +12,7 @@ export async function GET(request: Request, { params }: any) {
 			progress: workflow?.progress,
 		})
 	} catch (e: any) {
-		console.log(e)
+		console.error(e)
 		return NextResponse.json({ error: e }, { status: 500 })
 	}
 }

src/app/api/trace/[id]/progress/route.ts

@@ -1,5 +1,5 @@
 import { NextResponse } from "next/server"
-import { prisma } from "@/services/prisma/prisma"
+import { prisma, EstimateComputeCost } from "@/services"
 
 export async function PUT(request: Request, { params }: any) {
 	const id = params.id
@@ -8,17 +8,28 @@ export async function PUT(request: Request, { params }: any) {
 	try {
 		// update tasks
 		for (const task of requestJson.tasks) {
+			/* Cost estimation */
+			let costEstimate = await EstimateComputeCost(
+				task.executor,
+				task.duration,
+				task.machineType,
+				task.cloudZone,
+				task.priceModel
+			)
+
 			await prisma.task.upsert({
 				where: {
 					workflowId_taskId: { workflowId: id, taskId: task.taskId },
 				},
 				update: {
 					data: task,
+					costEstimate: costEstimate,
 				},
 				create: {
 					workflowId: id,
 					taskId: task.taskId,
 					data: task,
+					costEstimate: costEstimate,
 				},
 			})
 		}

src/jsonTypes.ts

@@ -155,6 +155,8 @@ declare global {
 			cloudZone?: string
 			machineType?: string
 			priceModel?: string
+			costEstimate?: number
+			updatedAt: Date
 		}
 	}
 }

src/services/index.ts

@@ -1 +1,2 @@
 export * from "./prisma/prisma"
+export * from "./pricing"

src/services/pricing/aws/aws.ts

@@ -0,0 +1,125 @@
+import { PricingClient, GetProductsCommand } from "@aws-sdk/client-pricing"
+import { EC2Client, DescribeSpotPriceHistoryCommand } from "@aws-sdk/client-ec2"
+
+/**
+ *
+ * @param {string} instanceType - AWS instance type (e.g. m5.large)
+ * @param {string} regionCode - region code (e.g. eu-west-1)
+ * @param {string} marketOption - "spot" or "ondemand"
+ * @returns
+ */
+export const PricePerHourAWS = (instanceType: string, regionCode: string, marketOption: string) => {
+	if (marketOption === "standard") {
+		return PricerPerHourOnDemand(instanceType, regionCode)
+	} else if (marketOption === "spot") {
+		return PricePerHourSpot(instanceType, regionCode)
+	} else {
+		console.log("unsupported pricing option")
+		return null
+	}
+}
+/**
+ *
+ * @param {string} availabilityZone Converts availability zone to region
+ * @returns
+ */
+const azToRegion = (availabilityZone: string) => {
+	const lastChar = availabilityZone.charAt(availabilityZone.length - 1)
+
+	if (!isNaN(Number(lastChar))) {
+		return availabilityZone
+	}
+
+	return availabilityZone.slice(0, -1)
+}
+
+const PricePerHourSpot = async (instanceType: string, availabilityZone: string) => {
+	const region = azToRegion(availabilityZone)
+
+	const ec2Client = new EC2Client({ region: region })
+
+	const input = {
+		ProductDescriptions: ["Linux/UNIX (Amazon VPC)"],
+		InstanceTypes: [instanceType],
+		MaxResults: Number(10),
+		Filters: [
+			{
+				Name: "availability-zone",
+				Values: [availabilityZone],
+			},
+		],
+	}
+
+	const command = new DescribeSpotPriceHistoryCommand(input)
+	const response = await ec2Client.send(command)
+
+	if (!response.SpotPriceHistory) {
+		return null
+	}
+
+	if (response.SpotPriceHistory.length === 0) {
+		return null
+	}
+
+	const prices = response.SpotPriceHistory.map((item) => Number(item.SpotPrice))
+	const highestPriceUSD = Math.max(0, ...prices)
+	return highestPriceUSD
+}
+
+const pricingClient = new PricingClient({ region: "us-east-1" })
+
+const PricerPerHourOnDemand = async (instanceType: string, regionCode: string) => {
+	const region = azToRegion(regionCode)
+
+	const input = {
+		ServiceCode: "AmazonEC2",
+		Filters: [
+			{
+				Type: "TERM_MATCH",
+				Field: "instanceType",
+				Value: instanceType,
+			},
+			{
+				Type: "TERM_MATCH",
+				Field: "operatingSystem",
+				Value: "Linux",
+			},
+			{
+				Type: "TERM_MATCH",
+				Field: "regionCode",
+				Value: region,
+			},
+			{
+				Type: "TERM_MATCH",
+				Field: "marketOption",
+				Value: "OnDemand",
+			},
+		],
+		MaxResults: 10,
+	}
+
+	const command = new GetProductsCommand(input)
+	const response = await pricingClient.send(command)
+
+	if (!response.PriceList) {
+		return null
+	}
+
+	if (response.PriceList.length === 0) {
+		return null
+	}
+
+	const productList = response.PriceList.map((item) => JSON.parse(item as string))
+	const prices = productList.map((data) => {
+		try {
+			return data.terms.OnDemand[Object.keys(data.terms.OnDemand)[0]].priceDimensions[
+				Object.keys(data.terms.OnDemand[Object.keys(data.terms.OnDemand)[0]].priceDimensions)[0]
+			].pricePerUnit.USD
+		} catch (error) {
+			return 0
+		}
+	})
+
+	const highestPriceUSD = Math.max(0, ...prices)
+	return highestPriceUSD
+}

src/services/pricing/aws/index.ts

@@ -0,0 +1 @@
+export { PricePerHourAWS } from "./aws"

src/services/pricing/index.ts

@@ -0,0 +1 @@
+export { EstimateComputeCost } from "./pricing"

src/services/pricing/pricing.ts

@@ -0,0 +1,64 @@
+import { PricePerHourAWS } from "./aws"
+
+const fallbackPricePerCPUHour = 0.1
+
+/**
+ *
+ * @param {string} executor Nextflow executor
+ * @param {string} instanceType The instance type to get the price for
+ * @param {string} regionCode AWS region (eu-west-1, us-east-1, etc.)
+ * @param {string} marketOption OnDemand or Spot
+ * @returns
+ */
+export const EstimateComputeCost = async (
+	executor: string,
+	durationMs?: number,
+	instanceType?: string,
+	regionCode?: string,
+	marketOption?: string
+) => {
+	if (!durationMs) {
+		return 0
+	}
+
+	if (executor === "awsbatch") {
+		if (!instanceType || !regionCode || !marketOption) {
+			return 0
+		}
+
+		const pricePerHour = await PricePerHourAWS(instanceType, regionCode, marketOption)
+
+		if (!pricePerHour) {
+			return null
+		}
+
+		return (pricePerHour * durationMs) / (3600 * 1000)
+	}
+
+	if (
+		executor === "azurebatch" ||
+		executor === "bridge" ||
+		executor === "flux" ||
+		executor === "google-batch" ||
+		executor === "google-lifesciences" ||
+		executor === "hyperqueue" ||
+		executor === "condor" ||
+		executor === "ignite" ||
+		executor === "lsf" ||
+		executor === "moab" ||
+		executor === "nqsii" ||
+		executor === "oar" ||
+		executor === "pbs" ||
+		executor === "pbspro" ||
+		executor === "sge" ||
+		executor === "slurm" ||
+		executor === "k8s"
+	) {
+		console.log(
+			`Pricing for this executor is not implemented. Using fallback price of \$${fallbackPricePerCPUHour}/cpu/h`
+		)
+		return (fallbackPricePerCPUHour * durationMs) / (3600 * 1000)
+	}
+
+	return (fallbackPricePerCPUHour * durationMs) / (3600 * 1000)
+}