From ea687215d9e639009ac1c066de9b78a1778594fa Mon Sep 17 00:00:00 2001 From: shivatantarapelly-fl Date: Mon, 12 May 2025 13:08:09 +0530 Subject: [PATCH 1/6] changes related to directional cost calculation --- app/configuration_validation.py | 6 ++++- app/price_calculator.py | 45 ++++++++++++++++++--------------- 2 files changed, 29 insertions(+), 22 deletions(-) diff --git a/app/configuration_validation.py b/app/configuration_validation.py index de2171f..5bb0ded 100644 --- a/app/configuration_validation.py +++ b/app/configuration_validation.py @@ -327,6 +327,8 @@ def generate_all_combinations(data): "embedding_model": configuration["embedding"]["model"], "retrieval_service": configuration["retrieval"]["service"], "retrieval_model": configuration["retrieval"]["model"], + "retrieval_model_input_token_cost": configuration["retrieval"].get("input_token_cost", 0), + "retrieval_model_output_token_cost": configuration["retrieval"].get("output_token_cost", 0), "eval_service": configuration["evaluation"]["service"], "eval_embedding_model": configuration["evaluation"]["embedding_model"], "eval_retrieval_model": configuration["evaluation"]["retrieval_model"], @@ -360,7 +362,9 @@ def generate_all_combinations(data): #Calculate the inferencing price - doesn't include OpenSearch pricing if configuration.get('gateway_enabled', False): - configuration["inferencing_cost_estimate"] += 0 + configuration["inferencing_cost_estimate"] += estimate_retrieval_model_bedrock_price(bedrock_price_df, configuration, avg_prompt_length, num_prompts, + input_price=configuration.get("retrieval_model_input_token_cost"), + output_price=configuration.get("retrieval_model_output_token_cost")) else: if configuration["retrieval_service"] == "bedrock": inferencing_price = estimate_retrieval_model_bedrock_price(bedrock_price_df, configuration, avg_prompt_length, num_prompts) diff --git a/app/price_calculator.py b/app/price_calculator.py index a71173b..c9c896d 100644 --- a/app/price_calculator.py +++ b/app/price_calculator.py @@ -48,11 +48,8 @@ def estimate_embedding_model_bedrock_price(file_path, configuration, effective_k def estimate_retrieval_model_bedrock_price(file_path, configuration, avg_prompt_length, - num_prompts): + num_prompts, input_token_cost=None, output_token_cost=None): try: - if configuration.get('gateway_enabled', False): - return 0 - df = file_path.copy() # return df except Exception as e: @@ -72,25 +69,31 @@ def estimate_retrieval_model_bedrock_price(file_path, configuration, avg_prompt_ gen_model = configuration["retrieval_model"] n_shot_prompts = configuration["n_shot_prompts"] k = configuration["knn_num"] - - gen_model_price = df[(df['model'] == gen_model) & (df['Region'] == region)]['input_price'] - if gen_model_price.empty: - logger.warning("Returning price as Zero, as model is not present in Sheet") - return 0 + if input_token_cost != None and output_token_cost != None: + gen_model_input_price = float(input_token_cost) + gen_model_output_price = float(output_token_cost) else: - gen_model_price = float(gen_model_price.values[0]) # this price is in millions of tokens - gen_model_out_price = df[(df['model'] == gen_model) & (df['Region'] == region)]['output_price'] - gen_model_out_price = float(gen_model_out_price.values[0]) # this price is in millions of tokens - context_len = k * chunk_size - prompt_len = (n_shot_prompts + 1) * avg_prompt_length - total_input_tokens = (context_len + prompt_len) * num_prompts - retrieval_input_price = gen_model_price * float(total_input_tokens) / 1000000 - total_output_tokens = avg_prompt_length - retrieval_output_price = gen_model_out_price * float(total_output_tokens) / 1000000 - retrieval_price = retrieval_input_price + retrieval_output_price - return retrieval_price - + gen_model_price_series = df[(df['model'] == gen_model) & (df['Region'] == region)]['input_price'] + if gen_model_price_series.empty: + logger.warning("Returning price as Zero, as input price for model is not present in Sheet") + return 0 + else: + gen_model_input_price = float(gen_model_price_series.values[0]) + gen_model_out_price_series = df[(df['model'] == gen_model) & (df['Region'] == region)]['output_price'] + if gen_model_out_price_series.empty: + logger.warning("Returning price as Zero, as output price for model is not present in Sheet") + return 0 + else: + gen_model_output_price = float(gen_model_out_price_series.values[0]) + context_len = k * chunk_size + prompt_len = (n_shot_prompts + 1) * avg_prompt_length + total_input_tokens = (context_len + prompt_len) * num_prompts + retrieval_input_price = gen_model_input_price * float(total_input_tokens) / 1000000 + total_output_tokens = avg_prompt_length + retrieval_output_price = gen_model_output_price * float(total_output_tokens) / 1000000 + retrieval_price = retrieval_input_price + retrieval_output_price + return retrieval_price def estimate_fargate_price(total_time, vpc=8, mem=16): #Fargate pricing From 4bdd99420d77f58e3d7621737c03a943796f5bdc Mon Sep 17 00:00:00 2001 From: shivatantarapelly-fl Date: Mon, 12 May 2025 16:20:01 +0530 Subject: [PATCH 2/6] actual cost calculation changes --- lambda_handlers/cost_handler/pricing.py | 167 ++++++++++++------------ 1 file changed, 87 insertions(+), 80 deletions(-) diff --git a/lambda_handlers/cost_handler/pricing.py b/lambda_handlers/cost_handler/pricing.py index 51690b2..f824c7e 100644 --- a/lambda_handlers/cost_handler/pricing.py +++ b/lambda_handlers/cost_handler/pricing.py @@ -27,14 +27,6 @@ def compute_actual_price_breakdown( is_gateway_enabled = configuration.get("config", {}).get("gateway_enabled", False) - if is_gateway_enabled: - overall_metadata = {'total_cost': 0.0, 'total_time': total_time, 'order': ['total_time', 'total_cost']} - indexing_metadata = {'runtime': indexing_time, 'model': '', 'service': '', 'sagemaker_cost': 0.0, 'ecs_cost': 0.0, 'total_cost': 0.0, 'order': ['model', 'service', 'knowledge_base_tokens', 'bedrock_cost', 'runtime', 'sagemaker_cost', 'ecs_cost', 'opensearch_cost', 'total_cost']} - retriever_metadata = {'runtime': retrieval_time, 'ecs_cost': 0.0, 'total_cost': 0.0, 'order': ['no_of_questions', 'rerank_model', 'reranker_queries', 'reranking_cost', 'runtime', 'ecs_cost', 'opensearch_cost', 'total_cost']} - inferencer_metadata = {'model': configuration.get("config", {}).get("retrieval_model", ""), 'service': '', 'runtime': retrieval_time, 'sagemaker_cost': 0.0, 'sagemaker_embedding_cost': 0.0, 'total_cost': 0.0, 'order': ['model', 'service', 'no_of_questions', 'input_tokens', 'output_tokens', 'query_embed_tokens', 'input_tokens_cost', 'output_tokens_cost', 'query_embed_tokens_cost', 'average_latency', 'runtime', 'sagemaker_embedding_cost', 'sagemaker_cost', 'total_cost']} - evaluator_metadata = {'runtime': eval_time, 'ecs_cost': 0.0, 'total_cost': 0.0, 'order': ['runtime', 'ecs_cost', 'opensearch_cost', 'sagemaker_embedding_cost', 'sagemaker_inferencer_cost', 'total_cost']} - - return overall_metadata, indexing_metadata, retriever_metadata, inferencer_metadata, evaluator_metadata is_input_valid, input_missing = validate_params( configuration=configuration @@ -57,6 +49,8 @@ def compute_actual_price_breakdown( retrieval_model = configuration.get("config", {}).get("retrieval_model", "") embedding_service = configuration.get("config", {}).get("embedding_service", "") retrieval_service = configuration.get("config", {}).get("retrieval_service", "") + retrieval_model_input_token_cost = configuration.get("config", {}).get("retrieval_model_input_token_cost", 0) + retrieval_model_output_token_cost = configuration.get("config", {}).get("retrieval_model_output_token_cost", 0) bedrock_knowledge_base = configuration.get("config", {}).get("bedrock_knowledge_base", False) rerank_model_id = configuration.get("config", {}).get("rerank_model_id", None) is_opensearch = configuration.get("config", {}).get("is_opensearch", True) @@ -88,92 +82,105 @@ def compute_actual_price_breakdown( embedding_model_price = 0 # Calculating indexing, inferencing and evaluation costs for bedrock/Sagemaker - if not bedrock_knowledge_base: - indexing_metadata['runtime'] = indexing_time - indexing_metadata['model'] = embedding_model - indexing_metadata['service'] = embedding_service - if embedding_service == "bedrock" : - embedding_model_price = df[(df["model"] == embedding_model) & (df["Region"] == aws_region)]["input_price"] - if embedding_model_price.empty: - logger.error(f"No embedding model {embedding_model} price found.") - return None - embedding_model_price = float(embedding_model_price.values[0]) # Price per 1000 tokens - indexing_cost = (embedding_model_price * float(index_embed_tokens)) / THOUSAND - indexing_metadata['knowledge_base_tokens'] = index_embed_tokens - indexing_metadata['bedrock_cost'] = indexing_cost - else: - indexing_cost = sagemaker_cost(indexing_time) - indexing_metadata['sagemaker_cost'] = indexing_cost + if is_gateway_enabled: + indexing_metadata = {'runtime': indexing_time, 'model': '', 'service': '', 'sagemaker_cost': 0.0, 'ecs_cost': 0.0, 'total_cost': 0.0} + else: + if not bedrock_knowledge_base: + indexing_metadata['runtime'] = indexing_time + indexing_metadata['model'] = embedding_model + indexing_metadata['service'] = embedding_service + if embedding_service == "bedrock" : + embedding_model_price = df[(df["model"] == embedding_model) & (df["Region"] == aws_region)]["input_price"] + if embedding_model_price.empty: + logger.error(f"No embedding model {embedding_model} price found.") + return None + embedding_model_price = float(embedding_model_price.values[0]) # Price per 1000 tokens + indexing_cost = (embedding_model_price * float(index_embed_tokens)) / THOUSAND + indexing_metadata['knowledge_base_tokens'] = index_embed_tokens + indexing_metadata['bedrock_cost'] = indexing_cost + else: + indexing_cost = sagemaker_cost(indexing_time) + indexing_metadata['sagemaker_cost'] = indexing_cost retriever_metadata['runtime'] = retrieval_time inferencer_metadata['model'] = retrieval_model inferencer_metadata['service'] = retrieval_service - if retrieval_service == "bedrock" : - inferencer_metadata['input_tokens'] = input_tokens - inferencer_metadata['output_tokens'] = output_tokens - question_details = calculate_experiment_question_details(experiment_question_metrics_items) - retriever_metadata['no_of_questions'] = question_details["total_questions"] - - inferencer_metadata['no_of_questions'] = question_details["total_questions"] - inferencer_metadata['inference_time'] = question_details["overall_inferencer_time"] - inferencer_metadata['average_latency'] = question_details["average_inferencer_time"] - - reranking_cost = 0 - - retrieval_model_input_price = df[ - (df["model"] == retrieval_model) & (df["Region"] == aws_region) - ]["input_price"] - retrieval_model_output_price = df[ - (df["model"] == retrieval_model) & (df["Region"] == aws_region) - ]["output_price"] - - if retrieval_model_input_price.empty: - logger.error(f"No retrieval model {retrieval_model} input price found.") - return None - - if retrieval_model_output_price.empty: - logger.error(f"No retrieval model {retrieval_model} output price found.") - return None - - retrieval_model_input_price = float(retrieval_model_input_price.values[0]) # Price per million tokens - retrieval_model_output_price = float(retrieval_model_output_price.values[0]) # Price per million tokens - # Calculate costs - + if is_gateway_enabled: + inferencer_metadata = {'model': configuration.get("config", {}).get("retrieval_model", ""), 'service': '', 'runtime': retrieval_time, 'sagemaker_cost': 0.0, 'sagemaker_embedding_cost': 0.0, 'total_cost': 0.0} + retrieval_model_input_price = float(retrieval_model_input_token_cost) + retrieval_model_output_price = float(retrieval_model_output_token_cost) retrieval_model_input_actual_cost = (retrieval_model_input_price * float(input_tokens)) / MILLION retrieval_model_output_actual_cost = (retrieval_model_output_price * float(output_tokens)) / MILLION inferencer_metadata['input_tokens_cost'] = retrieval_model_input_actual_cost inferencer_metadata['output_tokens_cost'] = retrieval_model_output_actual_cost - if (not bedrock_knowledge_base) and embedding_service == "bedrock": - query_embedding_cost = (embedding_model_price * float(query_embed_tokens)) / THOUSAND - inferencer_metadata['query_embed_tokens'] = query_embed_tokens - inferencer_metadata['query_embed_tokens_cost'] = query_embedding_cost - if rerank_model_id and rerank_model_id != "none" : - retriever_metadata['rerank_model'] = rerank_model_id - retriever_metadata['reranker_queries'] = question_details["reranker_queries"] - reranker_model_price = df[(df["model"] == rerank_model_id) & (df["Region"] == aws_region)]["input_price"] - if reranker_model_price.empty: - logger.error(f"No reranker model {rerank_model_id} price found.") - return None - reranker_model_price = float(reranker_model_price.values[0]) # Price per 1000 queries - reranking_cost = (reranker_model_price * float(question_details['reranker_queries'])) / THOUSAND - retriever_metadata['reranking_cost'] = reranking_cost - retrieval_cost += reranking_cost inferencing_cost = retrieval_model_input_actual_cost + retrieval_model_output_actual_cost + query_embedding_cost else: - inferencer_metadata['runtime'] = retrieval_time - inferencing_cost = sagemaker_cost(retrieval_time) - inferencer_metadata['sagemaker_cost'] = inferencing_cost - if not bedrock_knowledge_base: - if embedding_service == "bedrock": + if retrieval_service == "bedrock" : + inferencer_metadata['input_tokens'] = input_tokens + inferencer_metadata['output_tokens'] = output_tokens + question_details = calculate_experiment_question_details(experiment_question_metrics_items) + retriever_metadata['no_of_questions'] = question_details["total_questions"] + + inferencer_metadata['no_of_questions'] = question_details["total_questions"] + inferencer_metadata['inference_time'] = question_details["overall_inferencer_time"] + inferencer_metadata['average_latency'] = question_details["average_inferencer_time"] + + reranking_cost = 0 + + retrieval_model_input_price = df[ + (df["model"] == retrieval_model) & (df["Region"] == aws_region) + ]["input_price"] + retrieval_model_output_price = df[ + (df["model"] == retrieval_model) & (df["Region"] == aws_region) + ]["output_price"] + + if retrieval_model_input_price.empty: + logger.error(f"No retrieval model {retrieval_model} input price found.") + return None + + if retrieval_model_output_price.empty: + logger.error(f"No retrieval model {retrieval_model} output price found.") + return None + + retrieval_model_input_price = float(retrieval_model_input_price.values[0]) # Price per million tokens + retrieval_model_output_price = float(retrieval_model_output_price.values[0]) # Price per million tokens + # Calculate costs + + retrieval_model_input_actual_cost = (retrieval_model_input_price * float(input_tokens)) / MILLION + retrieval_model_output_actual_cost = (retrieval_model_output_price * float(output_tokens)) / MILLION + inferencer_metadata['input_tokens_cost'] = retrieval_model_input_actual_cost + inferencer_metadata['output_tokens_cost'] = retrieval_model_output_actual_cost + if (not bedrock_knowledge_base) and embedding_service == "bedrock": query_embedding_cost = (embedding_model_price * float(query_embed_tokens)) / THOUSAND - inferencing_cost += query_embedding_cost inferencer_metadata['query_embed_tokens'] = query_embed_tokens inferencer_metadata['query_embed_tokens_cost'] = query_embedding_cost - else: - embedding_sagemaker_cost = sagemaker_cost(retrieval_time) - inferencing_cost += embedding_sagemaker_cost - inferencer_metadata['sagemaker_embedding_cost'] = embedding_sagemaker_cost + if rerank_model_id and rerank_model_id != "none" : + retriever_metadata['rerank_model'] = rerank_model_id + retriever_metadata['reranker_queries'] = question_details["reranker_queries"] + reranker_model_price = df[(df["model"] == rerank_model_id) & (df["Region"] == aws_region)]["input_price"] + if reranker_model_price.empty: + logger.error(f"No reranker model {rerank_model_id} price found.") + return None + reranker_model_price = float(reranker_model_price.values[0]) # Price per 1000 queries + reranking_cost = (reranker_model_price * float(question_details['reranker_queries'])) / THOUSAND + retriever_metadata['reranking_cost'] = reranking_cost + retrieval_cost += reranking_cost + inferencing_cost = retrieval_model_input_actual_cost + retrieval_model_output_actual_cost + query_embedding_cost + else: + inferencer_metadata['runtime'] = retrieval_time + inferencing_cost = sagemaker_cost(retrieval_time) + inferencer_metadata['sagemaker_cost'] = inferencing_cost + if not bedrock_knowledge_base: + if embedding_service == "bedrock": + query_embedding_cost = (embedding_model_price * float(query_embed_tokens)) / THOUSAND + inferencing_cost += query_embedding_cost + inferencer_metadata['query_embed_tokens'] = query_embed_tokens + inferencer_metadata['query_embed_tokens_cost'] = query_embedding_cost + else: + embedding_sagemaker_cost = sagemaker_cost(retrieval_time) + inferencing_cost += embedding_sagemaker_cost + inferencer_metadata['sagemaker_embedding_cost'] = embedding_sagemaker_cost inferencer_metadata['total_cost'] = inferencing_cost # Eval costs doesn't include ragas at the moment From 1a04f8ea705a97eb47adaaf375d4f1b3e7a37dc1 Mon Sep 17 00:00:00 2001 From: shivatantarapelly-fl Date: Thu, 15 May 2025 14:55:11 +0530 Subject: [PATCH 3/6] argument key fix as per usage --- app/configuration_validation.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/app/configuration_validation.py b/app/configuration_validation.py index 5bb0ded..ddd1ce0 100644 --- a/app/configuration_validation.py +++ b/app/configuration_validation.py @@ -2,6 +2,7 @@ import os import shutil import logging +import traceback from util.s3util import S3Util from util.bedrock_utils import KnowledgeBaseUtils from config.config import get_config @@ -314,7 +315,6 @@ def generate_all_combinations(data): if experiment_count >= MAX_VALID_EXPERIMENTS: break # Exit the loop when the limit is reached - configuration = { **combination } @@ -363,8 +363,8 @@ def generate_all_combinations(data): #Calculate the inferencing price - doesn't include OpenSearch pricing if configuration.get('gateway_enabled', False): configuration["inferencing_cost_estimate"] += estimate_retrieval_model_bedrock_price(bedrock_price_df, configuration, avg_prompt_length, num_prompts, - input_price=configuration.get("retrieval_model_input_token_cost"), - output_price=configuration.get("retrieval_model_output_token_cost")) + input_token_cost=configuration.get("retrieval_model_input_token_cost"), + output_token_cost=configuration.get("retrieval_model_output_token_cost")) else: if configuration["retrieval_service"] == "bedrock": inferencing_price = estimate_retrieval_model_bedrock_price(bedrock_price_df, configuration, avg_prompt_length, num_prompts) @@ -396,7 +396,6 @@ def generate_all_combinations(data): configuration["directional_pricing"] = configuration["indexing_cost_estimate"] + configuration["retrieval_cost_estimate"] + configuration["inferencing_cost_estimate"] + configuration["eval_cost_estimate"] configuration["directional_pricing"] +=configuration["directional_pricing"]*0.05 #extra configuration["directional_pricing"] = round(configuration["directional_pricing"],2) - return valid_configurations def generate_all_combinations_in_background(execution_id: str, execution_config_data): From 03d5b3981383cae9ad939919c1f7bbf880e9c585 Mon Sep 17 00:00:00 2001 From: shivatantarapelly-fl Date: Mon, 19 May 2025 12:12:17 +0530 Subject: [PATCH 4/6] changes for moving cost calculation to docker and multi db support --- app/configuration_validation.py | 1 - cost_compute_handler/fargate/__init__.py | 0 .../fargate/base_task_processor.py | 41 ++ .../fargate/cost_compute_processor.py | 197 +++++++++ .../fargate/handler/__init__.py | 0 .../fargate/handler/cost_compute/Dockerfile | 28 ++ .../fargate/handler/cost_compute/__init__.py | 0 .../fargate_cost_compute_handler.py | 41 ++ cost_compute_handler/fargate/pricing.py | 414 ++++++++++++++++++ 9 files changed, 721 insertions(+), 1 deletion(-) create mode 100644 cost_compute_handler/fargate/__init__.py create mode 100644 cost_compute_handler/fargate/base_task_processor.py create mode 100644 cost_compute_handler/fargate/cost_compute_processor.py create mode 100644 cost_compute_handler/fargate/handler/__init__.py create mode 100644 cost_compute_handler/fargate/handler/cost_compute/Dockerfile create mode 100644 cost_compute_handler/fargate/handler/cost_compute/__init__.py create mode 100644 cost_compute_handler/fargate/handler/cost_compute/fargate_cost_compute_handler.py create mode 100644 cost_compute_handler/fargate/pricing.py diff --git a/app/configuration_validation.py b/app/configuration_validation.py index ddd1ce0..d33659c 100644 --- a/app/configuration_validation.py +++ b/app/configuration_validation.py @@ -2,7 +2,6 @@ import os import shutil import logging -import traceback from util.s3util import S3Util from util.bedrock_utils import KnowledgeBaseUtils from config.config import get_config diff --git a/cost_compute_handler/fargate/__init__.py b/cost_compute_handler/fargate/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/cost_compute_handler/fargate/base_task_processor.py b/cost_compute_handler/fargate/base_task_processor.py new file mode 100644 index 0000000..ee52e46 --- /dev/null +++ b/cost_compute_handler/fargate/base_task_processor.py @@ -0,0 +1,41 @@ +import json +from abc import ABC, abstractmethod +from flotorch_core.logger.global_logger import get_logger + +logger = get_logger() + +class BaseFargateTaskProcessor(ABC): + """ + Abstract base class for Fargate task processors. + """ + + def __init__(self, input_data: dict): + """ + Initializes the task processor with input data. + Args: + input_data (dict): The input data for the task. + """ + self.input_data = input_data + + @abstractmethod + def process(self): + """ + Abstract method to be implemented by subclasses for processing tasks. + """ + raise NotImplementedError("Subclasses must implement the process method.") + + def send_task_success(self, output: dict): + """ + Sends task success signal. + Args: + output (dict): The output data to send. + """ + pass + + def send_task_failure(self, error_message: str): + """ + Sends task failure signal. + Args: + error_message (str): The error message to send. + """ + pass \ No newline at end of file diff --git a/cost_compute_handler/fargate/cost_compute_processor.py b/cost_compute_handler/fargate/cost_compute_processor.py new file mode 100644 index 0000000..aac744c --- /dev/null +++ b/cost_compute_handler/fargate/cost_compute_processor.py @@ -0,0 +1,197 @@ +import json +import math +import os +from fargate.base_task_processor import BaseFargateTaskProcessor +from decimal import Decimal +from .pricing import compute_actual_price_breakdown, calculate_experiment_duration +from flotorch_core.config.config import Config +from flotorch_core.config.env_config_provider import EnvConfigProvider +from flotorch_core.storage.db.postgresdb import PostgresDB +from flotorch_core.storage.db.dynamodb import DynamoDB +from flotorch_core.logger.global_logger import get_logger + + +MILLION = 1_000_000 +THOUSAND = 1_000 +SECONDS_IN_MINUTE = 60 +MINUTES_IN_HOUR = 60 +HOURS_IN_DAY = 24 +DAYS_IN_MONTH = 30 + +logger = get_logger() +env_config_provider = EnvConfigProvider() +config = Config(env_config_provider) +db_type = config.get_db_type() + +def fetch_data_from_db(table_name, key, value, index_name=None): + """ + Fetch items with the specified key and value from DynamoDB or postgres based on db type. + """ + try: + if db_type == "DYNAMODB": + db_client = DynamoDB(table_name=table_name, region_name=config.get_region()) + if index_name: + items = db_client.read(keys={index_name: value}) + else: + items = db_client.read(keys={key: value}) + + elif db_type == "POSTGRESDB": + db_client = PostgresDB(dbname=config.get_postgres_db(), user=config.get_postgres_user(), password=config.get_postgres_password, table_name=table_name, host=config.get_postgres_host(), port=config.get_postgres_port()) + items = db_client.read(key={key: value}) + return items + except Exception as e: + logger.error(f"Error fetching data from DB: {e}") + raise + + +def validate_event(event): + """ + Validate the input event to ensure required fields are present. + """ + required_fields = ["experiment_id"] + for field in required_fields: + if field not in event: + raise ValueError(f"Missing required field: {field}") + + if not isinstance(event["experiment_id"], str): + raise ValueError("'experiment_id' must be a string") + + +class RetrieverProcessor(BaseFargateTaskProcessor): + """ + Processor for retriever tasks in Fargate. + """ + + def process(self): + logger.info("Starting retriever process.") + try: + logger.info(f"Experiment Configuration received: {self.input_data}") + + # Validate input event + validate_event(self.input_data) + + experiment_id = self.input_data["experiment_id"] + experiment_table = config.get_experiment_table_name() + experiment_question_metrics_table = config.get_experiment_question_metrics_table() + experiment_question_metrics_index = os.getenv("experiment_question_metrics_index") + + if not experiment_table: + raise EnvironmentError("Environment variable 'experiment_table' is not set") + + if not experiment_question_metrics_table: + raise EnvironmentError("Environment variable 'experiment_question_metrics_table' is not set") + + # Initialize variables + total_query_embed_tokens = 0 + total_answer_input_tokens = 0 + total_answer_output_tokens = 0 + + experiment_items = fetch_data_from_db(experiment_table, 'id', experiment_id) + experiment_question_metrics_items = fetch_data_from_db(experiment_question_metrics_table, 'experiment_id', experiment_id, experiment_question_metrics_index) + total_duration = 0 + indexing_time = 0 + retrieval_time = 0 + eval_time = 0 + total_index_embed_tokens = 0 + + if experiment_items: + experiment = experiment_items[0] + indexing_time, retrieval_time, eval_time = calculate_experiment_duration(experiment) + indexing_time_in_min = math.ceil(indexing_time / SECONDS_IN_MINUTE) + retrieval_time_in_min = math.ceil(retrieval_time / SECONDS_IN_MINUTE) + eval_time_in_min = math.ceil(eval_time / SECONDS_IN_MINUTE) + total_duration = indexing_time + retrieval_time + eval_time + total_duration_in_min = indexing_time_in_min + retrieval_time_in_min + eval_time_in_min + logger.info(f"Experiment {experiment_id} Total Time (in minutes): {total_duration_in_min} Indexing Time: {indexing_time_in_min}, Retrieval: {retrieval_time_in_min}, Evaluation: {eval_time_in_min}") + + total_index_embed_tokens = experiment.get("index_embed_tokens", 0) + total_query_embed_tokens = experiment.get("retrieval_query_embed_tokens", 0) + total_answer_input_tokens = experiment.get("retrieval_input_tokens", 1) + total_answer_output_tokens = experiment.get("retrieval_output_tokens", 1) + + overall_metadata, indexing_metadata, retriever_metadata, inferencer_metadata, eval_metadata = compute_actual_price_breakdown( + experiment, + input_tokens=total_answer_input_tokens, + output_tokens=total_answer_output_tokens, + index_embed_tokens=total_index_embed_tokens, + query_embed_tokens=total_query_embed_tokens, + total_time=total_duration, + indexing_time=indexing_time, + retrieval_time=retrieval_time, + eval_time=eval_time, + experiment_question_metrics_items=experiment_question_metrics_items + ) + + total_cost = overall_metadata['total_cost'] + indexing_cost = indexing_metadata['total_cost'] + retriever_cost = retriever_metadata['total_cost'] + inferencer_cost = inferencer_metadata['total_cost'] + eval_cost = eval_metadata['total_cost'] + logger.info(f"Experiment {experiment_id} Actual Cost (in $): {total_cost}, Indexing: {indexing_cost}, Retrieval: {retriever_cost}, Inferencing: {inferencer_cost}, Evaluation : {eval_cost}") + + # Update DynamoDB with the new cost + if total_cost is None: + logger.error(f"Experiment {experiment_id} Actual Cost is None") + total_cost = 0 + + try: + key={"id": experiment_id} + data = { + "cost": str(total_cost), + "indexing_time": str(indexing_time), + "retrieval_time": str(retrieval_time), + "eval_time": str(eval_time), + "total_time": str(total_duration), + "indexing_cost": str(indexing_cost), + "retrieval_cost": str(retriever_cost), + "inferencing_cost": str(inferencer_cost), + "eval_cost": str(eval_cost), + "indexing_metadata": convert_floats_to_decimal(indexing_metadata), + "retriever_metadata": convert_floats_to_decimal(retriever_metadata), + "inferencer_metadata": convert_floats_to_decimal(inferencer_metadata), + "eval_metadata": convert_floats_to_decimal(eval_metadata), + "overall_metadata": convert_floats_to_decimal(overall_metadata) + } + if db_type == "DYNAMODB": + db_client = DynamoDB(table_name=experiment_table, region_name=config.get_region()) + elif db_type == "POSTGRESDB": + db_client = PostgresDB(dbname=config.get_postgres_db(), user=config.get_postgres_user(), password=config.get_postgres_password, table_name=experiment_table, host=config.get_postgres_host(), port=config.get_postgres_port()) + + db_client.update(key, data) + + except Exception as e: + logger.error(f"Error updating DB: {e}") + raise + + return { + "statusCode": 200, + "body": json.dumps( + { + "total_cost": total_cost, + "dynamodb_update_count": len(experiment_items), + } + ), + } + + except ValueError as ve: + logger.error(f"Validation error: {ve}") + return {"statusCode": 400, "body": json.dumps({"error": str(ve)})} + except EnvironmentError as ee: + logger.error(f"Environment error: {ee}") + return {"statusCode": 500, "body": json.dumps({"error": str(ee)})} + except Exception as e: + logger.error(f"Unhandled error: {e}") + return { + "statusCode": 500, + "body": json.dumps({"error": "Internal server error"}), + } + +def convert_floats_to_decimal(obj): + if isinstance(obj, float): + return Decimal(str(obj)) # Convert float to string first to prevent precision loss + elif isinstance(obj, dict): + return {k: convert_floats_to_decimal(v) for k, v in obj.items()} + elif isinstance(obj, list): + return [convert_floats_to_decimal(i) for i in obj] + else: + return obj \ No newline at end of file diff --git a/cost_compute_handler/fargate/handler/__init__.py b/cost_compute_handler/fargate/handler/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/cost_compute_handler/fargate/handler/cost_compute/Dockerfile b/cost_compute_handler/fargate/handler/cost_compute/Dockerfile new file mode 100644 index 0000000..3cfb5ff --- /dev/null +++ b/cost_compute_handler/fargate/handler/cost_compute/Dockerfile @@ -0,0 +1,28 @@ +# Use an official Python runtime as a parent image +FROM python:3.9-slim-buster + +# Set the working directory in the container +WORKDIR /app + +# Install FloTorch-core +RUN pip install FloTorch-core + +# Copy the application code +COPY fargate/ fargate/ +COPY fargate/handler/cost_compute/fargate_cost_compute_handler.py . + +# Set environment variables (these can be overridden at runtime) +ENV experiment_table=your_experiment_table_name +ENV experiment_question_metrics_table=your_experiment_question_metrics_table_name +ENV experiment_question_metrics_index=your_experiment_question_metrics_index_name +ENV bedrock_limit_csv=your_bedrock_limit_csv_file +ENV s3_bucket=your_s3_bucket_name +ENV AWS_REGION=your_aws_region +ENV AWS_ACCESS_KEY_ID=your_aws_access_key_id +ENV AWS_SECRET_ACCESS_KEY=your_aws_secret_access_key +ENV AWS_SESSION_TOKEN=your_aws_session_token +ENV AWS_DEFAULT_REGION=your_aws_default_region +ENV db_type=type_of_your_database + +# Command to run the main application +CMD ["python", "fargate_cost_compute_handler.py"] \ No newline at end of file diff --git a/cost_compute_handler/fargate/handler/cost_compute/__init__.py b/cost_compute_handler/fargate/handler/cost_compute/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/cost_compute_handler/fargate/handler/cost_compute/fargate_cost_compute_handler.py b/cost_compute_handler/fargate/handler/cost_compute/fargate_cost_compute_handler.py new file mode 100644 index 0000000..1113864 --- /dev/null +++ b/cost_compute_handler/fargate/handler/cost_compute/fargate_cost_compute_handler.py @@ -0,0 +1,41 @@ +import os +import json +from fargate.cost_compute_processor import RetrieverProcessor +from flotorch_core.logger.global_logger import get_logger +from flotorch_core.config.config import Config +from flotorch_core.config.env_config_provider import EnvConfigProvider + +logger = get_logger() + +# Initialize configuration provider and config +env_config_provider = EnvConfigProvider() +config = Config(env_config_provider) + + +def get_environment_data(): + """ + Fetches task token and input data from environment variables. + Returns: + tuple: Task token (str) and input data (dict). + """ + input_data = {"experiment_id":os.getenv("experiment_id")} + return input_data + + +def main(): + """ + Main entry point for the Fargate retriever handler. + """ + try: + input_data = get_environment_data() + + # Initialize and process the RetrieverProcessor + fargate_processor = RetrieverProcessor(input_data) + fargate_processor.process() + except Exception as e: + logger.error(f"Error processing experiment cost calculation: {str(e)}") + raise + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/cost_compute_handler/fargate/pricing.py b/cost_compute_handler/fargate/pricing.py new file mode 100644 index 0000000..1760f9f --- /dev/null +++ b/cost_compute_handler/fargate/pricing.py @@ -0,0 +1,414 @@ +from datetime import datetime, timezone +import logging +import math +import os +import boto3 +import io +import csv +import pandas as pd +from typing import Optional +from botocore.exceptions import ClientError + +MILLION = 1_000_000 +THOUSAND = 1_000 +SECONDS_IN_MINUTE = 60 +MINUTES_IN_HOUR = 60 +HOURS_IN_DAY = 24 +DAYS_IN_MONTH = 30 + +s3 = boto3.client("s3") +S3_BUCKET = os.getenv("s3_bucket") +BEDROCK_CSV_PATH = os.getenv("bedrock_limit_csv") +logger = logging.getLogger() +logging.basicConfig(level=logging.INFO) + +def parse_datetime(datetime_str): + if not datetime_str: + return None + try: + dt = datetime.strptime(datetime_str, '%Y-%m-%dT%H:%M:%S.%fZ') + except ValueError: + dt = datetime.strptime(datetime_str, '%Y-%m-%dT%H:%M:%S%z') + + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def calculate_experiment_duration(experiment): + """Calculate various durations (total, indexing, retrieval, evaluation) from the experiment dictionary.""" + try: + def calculate_difference(start_key, end_key): + if experiment.get(start_key) and experiment.get(end_key): + start = parse_datetime(experiment[start_key]) + end = parse_datetime(experiment[end_key]) + return (end - start).total_seconds() + return 0 + + # total_duration = calculate_difference("start_datetime", "end_datetime") + indexing_duration = calculate_difference("indexing_start", "indexing_end") + retrieval_duration = calculate_difference("retrieval_start", "retrieval_end") + eval_duration = calculate_difference("eval_start", "eval_end") + + return ( + # math.ceil(total_duration / SECONDS_IN_MINUTE), + indexing_duration, + retrieval_duration, + eval_duration + # math.ceil(indexing_duration / SECONDS_IN_MINUTE), + # math.ceil(retrieval_duration / SECONDS_IN_MINUTE), + # math.ceil(eval_duration / SECONDS_IN_MINUTE), + ) + except Exception as e: + logger.error(f"Error occurred during duration computation: {e}") + return 0, 0, 0 + +def sagemaker_cost(time, number_of_instances = 1): + instance_cost_per_hour = 1.210 #per hour ml.g5.2xlarge per model + overall_cost = instance_cost_per_hour * number_of_instances * ((time / SECONDS_IN_MINUTE) / MINUTES_IN_HOUR) + + return overall_cost + +def opensearch_cost(time): + number_of_instances = 3 + + instance_cost_per_hour = 0.711 # r7g.2xlarge.search + instance_total_cost = (instance_cost_per_hour * number_of_instances * (time / SECONDS_IN_MINUTE) / MINUTES_IN_HOUR) + + ebs_volume_size = 10 # 2 GB + ebs_volume_price_per_month = .122 + ebs_total_cost = ebs_volume_price_per_month * ebs_volume_size * number_of_instances * ((time / SECONDS_IN_MINUTE) / MINUTES_IN_HOUR) / (HOURS_IN_DAY * DAYS_IN_MONTH) # 3 instances for 10GB each + + iops_cost_per_month = .008 + iops_per_instance = 16000 # instances per hour for 16000 IOPS (3000 free) + free_iops = 3000 + costing_iops_per_instance = iops_per_instance - free_iops + iops_total_cost = iops_cost_per_month * costing_iops_per_instance * number_of_instances * ((time / SECONDS_IN_MINUTE) / MINUTES_IN_HOUR) / (HOURS_IN_DAY * DAYS_IN_MONTH) # 3 instances for 16000 iops each + + overall_cost = instance_total_cost + ebs_total_cost + iops_total_cost + + return overall_cost + +def ecs_cost(time): + # 8 vCPU, 16 GB Memory + vCPU = 8 + memory = 16 + fargate_cpu_cost_per_vcpu = 0.04048 + fargate_memory_cost_per_gb = 0.004445 + + fargate_cpu_total_cost = fargate_cpu_cost_per_vcpu * vCPU + fargate_memory_total_cost = fargate_memory_cost_per_gb * memory + + overall_cost = (fargate_cpu_total_cost + fargate_memory_total_cost) * ((time / SECONDS_IN_MINUTE) / MINUTES_IN_HOUR) + return overall_cost + +def calculate_experiment_question_details(experiment_question_metrics_items): + total_questions = len(experiment_question_metrics_items) + overall_inferencer_time = 0 + average_inferencer_time = 0 + reranker_queries = 0 + for question in experiment_question_metrics_items: + answer_metadata = question.get("answer_metadata", None) + if answer_metadata: + latency = answer_metadata.get("latencyMs", 0) + inputTokens = answer_metadata.get('inputTokens', 0) + overall_inferencer_time += (latency / THOUSAND) + if math.ceil(inputTokens / 500) >= 100: + reranker_queries += (math.ceil(inputTokens / 500) / 100) + else: + reranker_queries += 1 + average_inferencer_time = overall_inferencer_time / total_questions + return { + "total_questions": total_questions, + "overall_inferencer_time": overall_inferencer_time, + "average_inferencer_time": average_inferencer_time, + "reranker_queries": reranker_queries + } + + +def compute_actual_price_breakdown( + configuration, input_tokens, output_tokens, index_embed_tokens, query_embed_tokens, total_time, indexing_time, retrieval_time, eval_time, experiment_question_metrics_items +): + """Compute the actual price based on the given configuration and token/time inputs.""" + + + is_gateway_enabled = configuration.get("config", {}).get("gateway_enabled", False) + + is_input_valid, input_missing = validate_params( + configuration=configuration + ) + + if not is_input_valid: + logger.error(f"Missing required parameters: {', '.join(input_missing)}.") + return None + + try: + # Read pricing data from S3 + df = read_csv_from_s3(BEDROCK_CSV_PATH, S3_BUCKET) + except Exception as e: + logger.error(f"Error reading the CSV file from S3: {e}") + return None + + try: + aws_region = configuration.get("config", {}).get("region", "") + embedding_model = configuration.get("config", {}).get("embedding_model", "") + retrieval_model = configuration.get("config", {}).get("retrieval_model", "") + embedding_service = configuration.get("config", {}).get("embedding_service", "") + retrieval_service = configuration.get("config", {}).get("retrieval_service", "") + retrieval_model_input_token_cost = configuration.get("config", {}).get("retrieval_model_input_token_cost", 0) + retrieval_model_output_token_cost = configuration.get("config", {}).get("retrieval_model_output_token_cost", 0) + bedrock_knowledge_base = configuration.get("config", {}).get("bedrock_knowledge_base", False) + rerank_model_id = configuration.get("config", {}).get("rerank_model_id", None) + is_opensearch = configuration.get("config", {}).get("is_opensearch", True) + validation_params = { + "aws_region": aws_region, + "retrieval_model": retrieval_model, + } + if not is_gateway_enabled: + validation_params["retrieval_service"] = retrieval_service + is_config_valid, config_missing = validate_params(**validation_params) + if not is_config_valid: + logger.error( + f"Configuration is missing required fields: {', '.join(config_missing)}." + ) + return None + + indexing_cost = 0 + retrieval_cost = 0 + inferencing_cost = 0 + eval_cost = 0 + total_cost = 0 + + overall_metadata = {} + indexing_metadata = {} + retriever_metadata = {} + inferencer_metadata = {} + evaluator_metadata = {} + + query_embedding_cost = 0 + embedding_model_price = 0 + + # Calculating indexing, inferencing and evaluation costs for bedrock/Sagemaker + if is_gateway_enabled: + indexing_metadata = {'runtime': indexing_time, 'model': '', 'service': '', 'sagemaker_cost': 0.0, 'ecs_cost': 0.0, 'total_cost': 0.0} + else: + if not bedrock_knowledge_base: + indexing_metadata['runtime'] = indexing_time + indexing_metadata['model'] = embedding_model + indexing_metadata['service'] = embedding_service + if embedding_service == "bedrock" : + embedding_model_price = df[(df["model"] == embedding_model) & (df["Region"] == aws_region)]["input_price"] + if embedding_model_price.empty: + logger.error(f"No embedding model {embedding_model} price found.") + return None + embedding_model_price = float(embedding_model_price.values[0]) # Price per 1000 tokens + indexing_cost = (embedding_model_price * float(index_embed_tokens)) / THOUSAND + indexing_metadata['knowledge_base_tokens'] = index_embed_tokens + indexing_metadata['bedrock_cost'] = indexing_cost + else: + indexing_cost = sagemaker_cost(indexing_time) + indexing_metadata['sagemaker_cost'] = indexing_cost + + retriever_metadata['runtime'] = retrieval_time + + inferencer_metadata['model'] = retrieval_model + inferencer_metadata['service'] = retrieval_service + if is_gateway_enabled: + inferencer_metadata = {'model': configuration.get("config", {}).get("retrieval_model", ""), 'service': '', 'runtime': retrieval_time, 'sagemaker_cost': 0.0, 'sagemaker_embedding_cost': 0.0, 'total_cost': 0.0} + retrieval_model_input_price = float(retrieval_model_input_token_cost) + retrieval_model_output_price = float(retrieval_model_output_token_cost) + retrieval_model_input_actual_cost = (retrieval_model_input_price * float(input_tokens)) / MILLION + retrieval_model_output_actual_cost = (retrieval_model_output_price * float(output_tokens)) / MILLION + inferencer_metadata['input_tokens_cost'] = retrieval_model_input_actual_cost + inferencer_metadata['output_tokens_cost'] = retrieval_model_output_actual_cost + inferencing_cost = retrieval_model_input_actual_cost + retrieval_model_output_actual_cost + query_embedding_cost + else: + if retrieval_service == "bedrock" : + inferencer_metadata['input_tokens'] = input_tokens + inferencer_metadata['output_tokens'] = output_tokens + question_details = calculate_experiment_question_details(experiment_question_metrics_items) + retriever_metadata['no_of_questions'] = question_details["total_questions"] + + inferencer_metadata['no_of_questions'] = question_details["total_questions"] + inferencer_metadata['inference_time'] = question_details["overall_inferencer_time"] + inferencer_metadata['average_latency'] = question_details["average_inferencer_time"] + + reranking_cost = 0 + + retrieval_model_input_price = df[ + (df["model"] == retrieval_model) & (df["Region"] == aws_region) + ]["input_price"] + retrieval_model_output_price = df[ + (df["model"] == retrieval_model) & (df["Region"] == aws_region) + ]["output_price"] + + if retrieval_model_input_price.empty: + logger.error(f"No retrieval model {retrieval_model} input price found.") + return None + + if retrieval_model_output_price.empty: + logger.error(f"No retrieval model {retrieval_model} output price found.") + return None + + retrieval_model_input_price = float(retrieval_model_input_price.values[0]) # Price per million tokens + retrieval_model_output_price = float(retrieval_model_output_price.values[0]) # Price per million tokens + # Calculate costs + + retrieval_model_input_actual_cost = (retrieval_model_input_price * float(input_tokens)) / MILLION + retrieval_model_output_actual_cost = (retrieval_model_output_price * float(output_tokens)) / MILLION + inferencer_metadata['input_tokens_cost'] = retrieval_model_input_actual_cost + inferencer_metadata['output_tokens_cost'] = retrieval_model_output_actual_cost + if (not bedrock_knowledge_base) and embedding_service == "bedrock": + query_embedding_cost = (embedding_model_price * float(query_embed_tokens)) / THOUSAND + inferencer_metadata['query_embed_tokens'] = query_embed_tokens + inferencer_metadata['query_embed_tokens_cost'] = query_embedding_cost + if rerank_model_id and rerank_model_id != "none" : + retriever_metadata['rerank_model'] = rerank_model_id + retriever_metadata['reranker_queries'] = question_details["reranker_queries"] + reranker_model_price = df[(df["model"] == rerank_model_id) & (df["Region"] == aws_region)]["input_price"] + if reranker_model_price.empty: + logger.error(f"No reranker model {rerank_model_id} price found.") + return None + reranker_model_price = float(reranker_model_price.values[0]) # Price per 1000 queries + reranking_cost = (reranker_model_price * float(question_details['reranker_queries'])) / THOUSAND + retriever_metadata['reranking_cost'] = reranking_cost + retrieval_cost += reranking_cost + inferencing_cost = retrieval_model_input_actual_cost + retrieval_model_output_actual_cost + query_embedding_cost + else: + inferencer_metadata['runtime'] = retrieval_time + inferencing_cost = sagemaker_cost(retrieval_time) + inferencer_metadata['sagemaker_cost'] = inferencing_cost + if not bedrock_knowledge_base: + if embedding_service == "bedrock": + query_embedding_cost = (embedding_model_price * float(query_embed_tokens)) / THOUSAND + inferencing_cost += query_embedding_cost + inferencer_metadata['query_embed_tokens'] = query_embed_tokens + inferencer_metadata['query_embed_tokens_cost'] = query_embedding_cost + else: + embedding_sagemaker_cost = sagemaker_cost(retrieval_time) + inferencing_cost += embedding_sagemaker_cost + inferencer_metadata['sagemaker_embedding_cost'] = embedding_sagemaker_cost + + inferencer_metadata['total_cost'] = inferencing_cost + # Eval costs doesn't include ragas at the moment + # Only adding sagemaker endpoint costs considering it is still running + evaluator_metadata['runtime'] = eval_time + if embedding_service == "sagemaker": + eval_cost += sagemaker_cost(eval_time) + evaluator_metadata['sagemaker_embedding_cost'] = eval_cost + if retrieval_service == "sagemaker": + eval_cost += sagemaker_cost(eval_time) + evaluator_metadata['sagemaker_inferencer_cost'] = eval_cost + + #Calculating fargate container costs + indexing_ecs_cost = ecs_cost(indexing_time) + retriever_ecs_cost = ecs_cost(retrieval_time) + eval_ecs_cost = ecs_cost(eval_time) + + if not bedrock_knowledge_base: + indexing_cost += indexing_ecs_cost + indexing_metadata['ecs_cost'] = indexing_ecs_cost + + retriever_metadata['ecs_cost'] = retriever_ecs_cost + evaluator_metadata['ecs_cost'] = eval_ecs_cost + + retrieval_cost += retriever_ecs_cost + eval_cost += eval_ecs_cost + + + # Adding opensearch provisioned costs + if not bedrock_knowledge_base and is_opensearch: + indexing_os_cost = opensearch_cost(indexing_time) + retriever_os_cost = opensearch_cost(retrieval_time) + eval_os_cost = opensearch_cost(eval_time) + + indexing_metadata['opensearch_cost'] = indexing_os_cost + retriever_metadata['opensearch_cost'] = retriever_os_cost + evaluator_metadata['opensearch_cost'] = eval_os_cost + + indexing_cost += indexing_os_cost + retrieval_cost += retriever_os_cost + eval_cost += eval_os_cost + + indexing_metadata['total_cost'] = indexing_cost + retriever_metadata['total_cost'] = retrieval_cost + evaluator_metadata['total_cost'] = eval_cost + + total_cost = indexing_cost + retrieval_cost + inferencing_cost + eval_cost + overall_metadata['total_cost'] = total_cost + overall_metadata['total_time'] = total_time + overall_metadata['order'] = ['total_time', 'total_cost'] + indexing_metadata['order'] = ['model', 'service', 'knowledge_base_tokens', 'bedrock_cost', 'runtime', 'sagemaker_cost', 'ecs_cost', 'opensearch_cost', 'total_cost'] + retriever_metadata['order'] = ['no_of_questions', 'rerank_model', 'reranker_queries', 'reranking_cost', 'runtime', 'ecs_cost', 'opensearch_cost', 'total_cost'] + inferencer_metadata['order'] = ['model', 'service', 'no_of_questions', 'input_tokens', 'output_tokens', 'query_embed_tokens', 'input_tokens_cost', 'output_tokens_cost', 'query_embed_tokens_cost', 'average_latency', 'runtime', 'sagemaker_embedding_cost', 'sagemaker_cost', 'total_cost'] + evaluator_metadata['order'] = ['runtime', 'ecs_cost', 'opensearch_cost', 'sagemaker_embedding_cost', 'sagemaker_inferencer_cost', 'total_cost'] + return overall_metadata, indexing_metadata, retriever_metadata, inferencer_metadata, evaluator_metadata + + except Exception as e: + logger.error(f"Error during price computation: {e}") + return None + +def validate_params(**kwargs): + """ + Validates the given parameters and identifies missing ones. + + Args: + **kwargs: Key-value pairs of parameter names and their values. + + Returns: + tuple: A boolean indicating if there are missing parameters and a list of missing parameter names. + """ + missing_params = [param for param, value in kwargs.items() if not value] + return not missing_params, missing_params + +def read_csv_from_s3(object_key: str, bucket_name: str, as_dataframe: bool = True) -> Optional[object]: + """ + Read CSV data from S3 and convert it to a list of dictionaries or a pandas DataFrame. + Args: + object_key (str): The key (path) of the S3 object. + bucket_name (str): The name of the S3 bucket. + as_dataframe (bool): If True, return a pandas DataFrame, otherwise return a list of dictionaries. + Returns: + Optional[object]: List of dictionaries or pandas DataFrame containing the CSV data if successful, None otherwise. + Raises: + ClientError: If there's an error accessing S3. + csv.Error: If the content cannot be parsed as CSV. + """ + logger = logging.getLogger() + logger.setLevel(logging.INFO) + + # Initialize the S3 client + s3_client = boto3.client('s3') + + try: + logger.info(f"Reading file from S3: Bucket={bucket_name}, Key={object_key}") + + # Get the object from S3 + response = s3_client.get_object(Bucket=bucket_name, Key=object_key) + + # Read the data + file_content = response['Body'].read().decode('utf-8') + + if as_dataframe: + # Parse content into a pandas DataFrame + csv_data = pd.read_csv(io.StringIO(file_content), float_precision="round_trip") + else: + # Parse content into a list of dictionaries + csv_reader = csv.DictReader(file_content.splitlines()) + csv_data = [row for row in csv_reader] + + return csv_data + + except ClientError as e: + error_code = e.response.get('Error', {}).get('Code', 'Unknown') + error_message = e.response.get('Error', {}).get('Message', 'Unknown error') + logger.error(f"Failed to read from S3: {error_code} - {error_message}", exc_info=True) + raise + + except csv.Error as e: + logger.error(f"Failed to parse CSV content from {bucket_name}/{object_key}", exc_info=True) + raise + + except Exception as e: + logger.error(f"Unexpected error reading from S3: {str(e)}", exc_info=True) + raise \ No newline at end of file From 6ad351103a68f1895bcce3471d0b9bd054033aa2 Mon Sep 17 00:00:00 2001 From: shivatantarapelly-fl Date: Mon, 19 May 2025 12:21:57 +0530 Subject: [PATCH 5/6] missed function calls added --- cost_compute_handler/fargate/cost_compute_processor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cost_compute_handler/fargate/cost_compute_processor.py b/cost_compute_handler/fargate/cost_compute_processor.py index aac744c..d2b5935 100644 --- a/cost_compute_handler/fargate/cost_compute_processor.py +++ b/cost_compute_handler/fargate/cost_compute_processor.py @@ -36,7 +36,7 @@ def fetch_data_from_db(table_name, key, value, index_name=None): items = db_client.read(keys={key: value}) elif db_type == "POSTGRESDB": - db_client = PostgresDB(dbname=config.get_postgres_db(), user=config.get_postgres_user(), password=config.get_postgres_password, table_name=table_name, host=config.get_postgres_host(), port=config.get_postgres_port()) + db_client = PostgresDB(dbname=config.get_postgres_db(), user=config.get_postgres_user(), password=config.get_postgres_password(), table_name=table_name, host=config.get_postgres_host(), port=config.get_postgres_port()) items = db_client.read(key={key: value}) return items except Exception as e: @@ -155,7 +155,7 @@ def process(self): if db_type == "DYNAMODB": db_client = DynamoDB(table_name=experiment_table, region_name=config.get_region()) elif db_type == "POSTGRESDB": - db_client = PostgresDB(dbname=config.get_postgres_db(), user=config.get_postgres_user(), password=config.get_postgres_password, table_name=experiment_table, host=config.get_postgres_host(), port=config.get_postgres_port()) + db_client = PostgresDB(dbname=config.get_postgres_db(), user=config.get_postgres_user(), password=config.get_postgres_password(), table_name=experiment_table, host=config.get_postgres_host(), port=config.get_postgres_port()) db_client.update(key, data) From ed75a913cecce8ff40ac562086b0ef4e27ae909c Mon Sep 17 00:00:00 2001 From: shivatantarapelly-fl Date: Mon, 19 May 2025 13:32:12 +0530 Subject: [PATCH 6/6] added task success and failure methods --- .../fargate/base_task_processor.py | 11 +++++----- .../fargate/cost_compute_processor.py | 20 ++++--------------- 2 files changed, 10 insertions(+), 21 deletions(-) diff --git a/cost_compute_handler/fargate/base_task_processor.py b/cost_compute_handler/fargate/base_task_processor.py index ee52e46..6daaf69 100644 --- a/cost_compute_handler/fargate/base_task_processor.py +++ b/cost_compute_handler/fargate/base_task_processor.py @@ -1,5 +1,6 @@ import json from abc import ABC, abstractmethod +import sys from flotorch_core.logger.global_logger import get_logger logger = get_logger() @@ -24,13 +25,12 @@ def process(self): """ raise NotImplementedError("Subclasses must implement the process method.") - def send_task_success(self, output: dict): + def send_task_success(self): """ Sends task success signal. - Args: - output (dict): The output data to send. """ - pass + print(json.dumps({"status": "success", "output": "Retrieval completed successfully"})) + sys.exit(0) def send_task_failure(self, error_message: str): """ @@ -38,4 +38,5 @@ def send_task_failure(self, error_message: str): Args: error_message (str): The error message to send. """ - pass \ No newline at end of file + print(json.dumps({"status": "failure", "output": "Retrieval Failed", "error": error_message})) + sys.exit(1) \ No newline at end of file diff --git a/cost_compute_handler/fargate/cost_compute_processor.py b/cost_compute_handler/fargate/cost_compute_processor.py index d2b5935..3412165 100644 --- a/cost_compute_handler/fargate/cost_compute_processor.py +++ b/cost_compute_handler/fargate/cost_compute_processor.py @@ -162,29 +162,17 @@ def process(self): except Exception as e: logger.error(f"Error updating DB: {e}") raise - - return { - "statusCode": 200, - "body": json.dumps( - { - "total_cost": total_cost, - "dynamodb_update_count": len(experiment_items), - } - ), - } + self.send_task_success() except ValueError as ve: logger.error(f"Validation error: {ve}") - return {"statusCode": 400, "body": json.dumps({"error": str(ve)})} + self.send_task_failure(str(ve)) except EnvironmentError as ee: logger.error(f"Environment error: {ee}") - return {"statusCode": 500, "body": json.dumps({"error": str(ee)})} + self.send_task_failure(str(ee)) except Exception as e: logger.error(f"Unhandled error: {e}") - return { - "statusCode": 500, - "body": json.dumps({"error": "Internal server error"}), - } + self.send_task_failure(str(e)) def convert_floats_to_decimal(obj): if isinstance(obj, float):