ViduraCID ToolsAthenaAWS Cost

CID / Cost Intelligence Tools

The CID (Cost Intelligence Dashboard) tools provide AWS cost analysis via Amazon Athena. They query pre-built CID views in the cid_cur Athena database derived from the AWS Cost and Usage Report (CUR).

Tool Listing

check_aws_connection()

Verifies AWS credentials and Athena connectivity. Runs a lightweight test query and returns connection status.

list_cid_views()

Discovers all available CID views in the configured Athena database. Returns view names, schemas, and descriptions from the query registry.

query_cid_view(view_name, limit)

Queries a specific CID view directly. View names are sanitized to prevent SQL injection (removes quotes, semicolons). Limit capped at 1000 rows.

Security:SQL injection prevention: sanitizes view_name (removes quotes, semicolons). Row limit capped at 1000.

describe_cid_view(view_name)

Returns the schema (columns and data types) for a specific CID view using DESCRIBE TABLE. Helps the agent understand available fields before querying.

run_athena_sql(sql_query)

Executes custom SQL against Athena. Restricted to SELECT, SHOW, and DESCRIBE statements only. Prevents destructive operations.

Security:Query type enforcement: only SELECT, SHOW, DESCRIBE allowed. Other statement types return an error.

get_cost_analysis_help()

Returns a help string listing all available CID commands and example queries. Used by the agent to orient new users.

Athena Integration

The AthenaUtils class manages Athena query execution with polling:

AthenaUtils.execute_query()

class AthenaUtils:
    def execute_query(
        self,
        sql_query: str,
        database: str,      # e.g., "cid_cur"
        workgroup: str,     # e.g., "primary"
        output_location: str  # S3 bucket for results
    ) -> str:  # Returns QueryExecutionId

        response = self.client.start_query_execution(
            QueryString=sql_query,
            QueryExecutionContext={'Database': database},
            WorkGroup=workgroup,
            ResultConfiguration={'OutputLocation': output_location}
        )
        query_id = response['QueryExecutionId']

        # Poll until SUCCEEDED or FAILED
        while True:
            status = self.client.get_query_execution(
                QueryExecutionId=query_id
            )['QueryExecution']['Status']['State']

            if status in ('SUCCEEDED', 'FAILED', 'CANCELLED'):
                break
            time.sleep(1)  # Poll every second

        if status != 'SUCCEEDED':
            raise AthenaQueryException(f"Query {status}")

        return query_id

Query Registry

The CID module includes a registry of 40+ pre-built Athena queries organized by category:

Category	Query ID	Description
CID	cid.summary_view	Cost summary by service, account, purchase option
CID	cid.ec2_running_cost	EC2 costs by instance type and purchase option
CID	cid.ri_sp_mapping	Reserved Instances & Savings Plans mapping
CID	cid.s3_view	S3/Glacier storage costs by storage class
Compute Optimizer	co.ec2_instance_options	EC2 rightsizing recommendations
Compute Optimizer	co.rds_instance_options	RDS instance rightsizing
Compute Optimizer	co.ebs_volume_options	EBS IOPS/throughput recommendations
Compute Optimizer	co.lambda_options	Lambda memory sizing
Compute Optimizer	co.idle_options	Identify idle/underutilized resources
CUDOS	cudos.hourly_view	Hourly cost/usage data (last 30 days)
CUDOS	cudos.resource_view	Resource-level costs with tags
KPI	kpi.kpi_ebs_snap	EBS snapshot costs and age analysis
KPI	kpi.kpi_ebs_storage	EBS storage with GP2→GP3 migration opportunities
KPI	kpi.kpi_s3_storage	S3 storage with intelligent tiering opportunities
KPI	kpi.kpi_tracker	Comprehensive KPI dashboard aggregation

Security Model

Query Restrictions

The run_athena_sql tool enforces read-only access. The allowed statement prefixes are checked before execution:

python

ALLOWED_STATEMENTS = ('SELECT', 'SHOW', 'DESCRIBE')

def run_athena_sql(sql_query: str) -> dict:
    stripped = sql_query.strip().upper()
    if not any(stripped.startswith(stmt) for stmt in ALLOWED_STATEMENTS):
        return {
            "error": f"Only {ALLOWED_STATEMENTS} statements are allowed",
            "provided": sql_query[:50]
        }
    # Proceed with execution...

View Name Sanitization

def sanitize_view_name(view_name: str) -> str:
    # Remove potentially dangerous characters
    # Prevents SQL injection via view name parameter
    return re.sub(r"['";]", "", view_name).strip()