OCI MCP Server

""" Tools for managing OCI Load Balancer resources. """ import logging from typing import Dict, List, Any, Optional import oci logger = logging.getLogger(__name__) def list_load_balancers(load_balancer_client: oci.load_balancer.LoadBalancerClient, compartment_id: str) -> List[Dict[str, Any]]: """ List all load balancers in a compartment. Args: load_balancer_client: OCI LoadBalancer client compartment_id: OCID of the compartment Returns: List of load balancers with their details """ try: load_balancers_response = oci.pagination.list_call_get_all_results( load_balancer_client.list_load_balancers, compartment_id ) load_balancers = [] for lb in load_balancers_response.data: load_balancers.append({ "id": lb.id, "display_name": lb.display_name, "compartment_id": lb.compartment_id, "lifecycle_state": lb.lifecycle_state, "time_created": str(lb.time_created), "shape_name": lb.shape_name, "is_private": lb.is_private, "ip_addresses": [ { "ip_address": ip.ip_address, "is_public": ip.is_public, } for ip in lb.ip_addresses ] if lb.ip_addresses else [], "subnet_ids": lb.subnet_ids, "network_security_group_ids": lb.network_security_group_ids, }) logger.info(f"Found {len(load_balancers)} load balancers in compartment {compartment_id}") return load_balancers except Exception as e: logger.exception(f"Error listing load balancers: {e}") raise def get_load_balancer(load_balancer_client: oci.load_balancer.LoadBalancerClient, load_balancer_id: str) -> Dict[str, Any]: """ Get details of a specific load balancer. Args: load_balancer_client: OCI LoadBalancer client load_balancer_id: OCID of the load balancer Returns: Details of the load balancer """ try: lb = load_balancer_client.get_load_balancer(load_balancer_id).data # Format backend sets backend_sets = {} if lb.backend_sets: for name, backend_set in lb.backend_sets.items(): backend_sets[name] = { "name": name, "policy": backend_set.policy, "backends": [ { "ip_address": backend.ip_address, "port": backend.port, "weight": backend.weight, "drain": backend.drain, "backup": backend.backup, "offline": backend.offline, } for backend in backend_set.backends ] if backend_set.backends else [], "health_checker": { "protocol": backend_set.health_checker.protocol, "url_path": backend_set.health_checker.url_path, "port": backend_set.health_checker.port, "return_code": backend_set.health_checker.return_code, "retries": backend_set.health_checker.retries, "timeout_in_millis": backend_set.health_checker.timeout_in_millis, "interval_in_millis": backend_set.health_checker.interval_in_millis, "response_body_regex": backend_set.health_checker.response_body_regex, } if backend_set.health_checker else None, } # Format listeners listeners = {} if lb.listeners: for name, listener in lb.listeners.items(): listeners[name] = { "name": name, "default_backend_set_name": listener.default_backend_set_name, "port": listener.port, "protocol": listener.protocol, "hostname_names": listener.hostname_names, "path_route_set_name": listener.path_route_set_name, "ssl_configuration": { "certificate_name": listener.ssl_configuration.certificate_name if listener.ssl_configuration else None, "verify_peer_certificate": listener.ssl_configuration.verify_peer_certificate if listener.ssl_configuration else None, "verify_depth": listener.ssl_configuration.verify_depth if listener.ssl_configuration else None, } if listener.ssl_configuration else None, "connection_configuration": { "idle_timeout": listener.connection_configuration.idle_timeout if listener.connection_configuration else None, "backend_tcp_proxy_protocol_version": listener.connection_configuration.backend_tcp_proxy_protocol_version if listener.connection_configuration else None, } if listener.connection_configuration else None, } lb_details = { "id": lb.id, "display_name": lb.display_name, "compartment_id": lb.compartment_id, "lifecycle_state": lb.lifecycle_state, "time_created": str(lb.time_created), "shape_name": lb.shape_name, "is_private": lb.is_private, "ip_addresses": [ { "ip_address": ip.ip_address, "is_public": ip.is_public, } for ip in lb.ip_addresses ] if lb.ip_addresses else [], "subnet_ids": lb.subnet_ids, "network_security_group_ids": lb.network_security_group_ids, "backend_sets": backend_sets, "listeners": listeners, "certificates": dict(lb.certificates) if lb.certificates else {}, "path_route_sets": dict(lb.path_route_sets) if lb.path_route_sets else {}, "hostnames": dict(lb.hostnames) if lb.hostnames else {}, } logger.info(f"Retrieved details for load balancer {load_balancer_id}") return lb_details except Exception as e: logger.exception(f"Error getting load balancer details: {e}") raise def list_network_load_balancers(network_load_balancer_client: oci.network_load_balancer.NetworkLoadBalancerClient, compartment_id: str) -> List[Dict[str, Any]]: """ List all network load balancers in a compartment. Args: network_load_balancer_client: OCI NetworkLoadBalancer client compartment_id: OCID of the compartment Returns: List of network load balancers with their details """ try: nlbs_response = oci.pagination.list_call_get_all_results( network_load_balancer_client.list_network_load_balancers, compartment_id ) nlbs = [] for nlb in nlbs_response.data: nlbs.append({ "id": nlb.id, "display_name": nlb.display_name, "compartment_id": nlb.compartment_id, "lifecycle_state": nlb.lifecycle_state, "time_created": str(nlb.time_created), "is_private": nlb.is_private, "ip_addresses": [ { "ip_address": ip.ip_address, "is_public": ip.is_public, "ip_version": ip.ip_version, } for ip in nlb.ip_addresses ] if nlb.ip_addresses else [], "subnet_id": nlb.subnet_id, "network_security_group_ids": nlb.network_security_group_ids, "is_preserve_source_destination": nlb.is_preserve_source_destination, }) logger.info(f"Found {len(nlbs)} network load balancers in compartment {compartment_id}") return nlbs except Exception as e: logger.exception(f"Error listing network load balancers: {e}") raise def get_network_load_balancer(network_load_balancer_client: oci.network_load_balancer.NetworkLoadBalancerClient, network_load_balancer_id: str) -> Dict[str, Any]: """ Get details of a specific network load balancer. Args: network_load_balancer_client: OCI NetworkLoadBalancer client network_load_balancer_id: OCID of the network load balancer Returns: Details of the network load balancer """ try: nlb = network_load_balancer_client.get_network_load_balancer(network_load_balancer_id).data # Format backend sets backend_sets = {} if nlb.backend_sets: for name, backend_set in nlb.backend_sets.items(): backend_sets[name] = { "name": name, "policy": backend_set.policy, "is_preserve_source": backend_set.is_preserve_source, "backends": [ { "ip_address": backend.ip_address, "port": backend.port, "weight": backend.weight, "target_id": backend.target_id, "is_drain": backend.is_drain, "is_backup": backend.is_backup, "is_offline": backend.is_offline, } for backend in backend_set.backends ] if backend_set.backends else [], "health_checker": { "protocol": backend_set.health_checker.protocol, "port": backend_set.health_checker.port, "url_path": backend_set.health_checker.url_path, "return_code": backend_set.health_checker.return_code, "retries": backend_set.health_checker.retries, "timeout_in_millis": backend_set.health_checker.timeout_in_millis, "interval_in_millis": backend_set.health_checker.interval_in_millis, "request_data": backend_set.health_checker.request_data, "response_data": backend_set.health_checker.response_data, } if backend_set.health_checker else None, } # Format listeners listeners = {} if nlb.listeners: for name, listener in nlb.listeners.items(): listeners[name] = { "name": name, "default_backend_set_name": listener.default_backend_set_name, "port": listener.port, "protocol": listener.protocol, "ip_version": listener.ip_version, } nlb_details = { "id": nlb.id, "display_name": nlb.display_name, "compartment_id": nlb.compartment_id, "lifecycle_state": nlb.lifecycle_state, "time_created": str(nlb.time_created), "is_private": nlb.is_private, "ip_addresses": [ { "ip_address": ip.ip_address, "is_public": ip.is_public, "ip_version": ip.ip_version, } for ip in nlb.ip_addresses ] if nlb.ip_addresses else [], "subnet_id": nlb.subnet_id, "network_security_group_ids": nlb.network_security_group_ids, "is_preserve_source_destination": nlb.is_preserve_source_destination, "backend_sets": backend_sets, "listeners": listeners, } logger.info(f"Retrieved details for network load balancer {network_load_balancer_id}") return nlb_details except Exception as e: logger.exception(f"Error getting network load balancer details: {e}") raise