diff --git a/heartbeat/powervs-move-ip.in b/heartbeat/powervs-move-ip.in index f5d30ec5b..d55979e52 100755 --- a/heartbeat/powervs-move-ip.in +++ b/heartbeat/powervs-move-ip.in @@ -1,1035 +1,1035 @@ #!@PYTHON@ -tt # ------------------------------------------------------------------------ # Description: Resource agent for moving an overlay IP address between # virtual server instances in different PowerVS workspaces. # # Authors: Edmund Haefele # Walter Orb # # Copyright (c) 2025 International Business Machines, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # ------------------------------------------------------------------------ import fcntl import ipaddress import json import os import socket import subprocess import sys import textwrap import time from pathlib import Path from urllib.parse import urlparse import requests import requests.adapters import urllib3.util # Constants OCF_FUNCTIONS_DIR = os.environ.get( "OCF_FUNCTIONS_DIR", "%s/lib/heartbeat" % os.environ.get("OCF_ROOT") ) RESOURCE_OPTIONS = ( "ip", "api_key", "api_type", "region", "route_host_map", "use_token_cache", "monitor_api", "device", "proxy", ) IP_CMD = "/usr/sbin/ip" REQUESTS_TIMEOUT = 5 # Timeout for requests calls HTTP_MAX_RETRIES = 3 # Maximum number of retries for HTTP requests HTTP_BACKOFF_FACTOR = 0.3 # Sleep (factor * (2^number of previous retries)) secs HTTP_STATUS_FORCE_RETRIES = (500, 502, 503, 504) # HTTP status codes to retry on HTTP_RETRY_ALLOWED_METHODS = frozenset({"GET", "POST", "PUT", "DELETE"}) CIDR_NETMASK = "32" sys.path.append(OCF_FUNCTIONS_DIR) try: import ocf except ImportError: sys.stderr.write("ImportError: ocf module import failed.") sys.exit(5) class OCFExitError(Exception): """Exception class for OCF (Open Cluster Framework) exit errors.""" def __init__(self, message, exit_code): ocf.ocf_exit_reason(message) sys.exit(exit_code) class CmdError(OCFExitError): """Exception class for errors when running system commands.""" def __init__(self, message, exit_code): super().__init__(f"[CmdError] {message}", exit_code) def os_cmd(cmd_args, is_json=False, timeout=10): """Run a system command and optionally parse JSON output.""" ocf.logger.debug(f"[os_cmd]: args: {cmd_args}") try: result = subprocess.run( cmd_args, capture_output=True, text=True, check=True, timeout=timeout, env={"LANG": "C"}, ) if is_json: try: return json.loads(result.stdout) except json.JSONDecodeError as e: raise CmdError(f"os_cmd: JSON parsing failed: {e}", ocf.OCF_ERR_GENERIC) return result.returncode except subprocess.CalledProcessError as e: raise CmdError( f"os_cmd: command failed: {e.stderr}", ocf.OCF_ERR_GENERIC, ) except subprocess.TimeoutExpired: raise CmdError("os_cmd: command timed out", ocf.OCF_ERR_GENERIC) def ip_cmd(*args, is_json=False): """Generic wrapper for the ip command.""" return os_cmd([IP_CMD] + list(args), is_json=is_json) def ip_address_show(): """Show IP addresses in JSON format.""" return ip_cmd("-json", "address", "show", is_json=True) def ip_address_add(cidr, device, label=None): """Add an IP address to a device.""" cmd = ["address", "add", cidr, "dev", device] if label: cmd += ["label", label] return ip_cmd(*cmd) def ip_address_delete(cidr, device): """Delete an IP address from a device.""" return ip_cmd("address", "delete", cidr, "dev", device) def ip_find_device(ip): """Find the device associated with a given IP address.""" for iface in ip_address_show(): addresses = [a["local"] for a in iface["addr_info"]] if ip in addresses and "UP" in iface["flags"]: return iface["ifname"] return None def ip_check_device(device): """Verify that a device with the specified interface name (device) exists.""" for iface in ip_address_show(): if iface["ifname"] == device and "UP" in iface["flags"]: return True return False def ip_alias_add(ip, device): """Add an IP alias to the given device.""" ip_cidr = f"{ip}/{CIDR_NETMASK}" ocf.logger.debug( f"[ip_alias_add]: adding IP alias '{ip_cidr}' to interface '{device}'" ) _ = ip_address_add(ip_cidr, device) def ip_alias_remove(ip): """Find the device with the given IP alias and remove the alias.""" device = ip_find_device(ip) if device: ip_cidr = f"{ip}/{CIDR_NETMASK}" ocf.logger.debug( f"[ip_alias_remove]: removing IP alias '{ip_cidr}' from interface '{device}'" ) _ = ip_address_delete(ip_cidr, device) def create_session_with_retries(): """Create a request session with a retry strategy.""" retry_strategy = urllib3.util.Retry( total=HTTP_MAX_RETRIES, status_forcelist=HTTP_STATUS_FORCE_RETRIES, allowed_methods=HTTP_RETRY_ALLOWED_METHODS, backoff_factor=HTTP_BACKOFF_FACTOR, raise_on_status=False, ) adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy) session = requests.Session() session.mount("https://", adapter) return session class PowerCloudTokenManagerError(OCFExitError): """Exception class for errors in the PowerCloudTokenManager.""" def __init__(self, message, exit_code): super().__init__(f"[PowerCloudTokenManagerError] {message}", exit_code) class PowerCloudTokenManager: """Request and cache IBM Cloud tokens.""" _DEFAULT_RESOURCE_INSTANCE = "powervs-move-ip" _TOKEN_REFRESH_BUFFER = 900 # 15 minutes def __init__( self, api_type="", api_key="", proxy="", use_cache=False, ): self._auth_url = ( "https://private.iam.cloud.ibm.com/identity/token" if api_type == "private" else "https://iam.cloud.ibm.com/identity/token" ) self._api_key = self._load_api_key(api_key) self._proxy = proxy self._session = create_session_with_retries() self._cache_file = None if use_cache: resource_instance = os.environ.get( "OCF_RESOURCE_INSTANCE", self._DEFAULT_RESOURCE_INSTANCE ) self._cache_file = Path( f"/var/run/resource-agents/{resource_instance}-token.json" ) self._cache_file.parent.mkdir(parents=True, exist_ok=True) if not self._cache_file.exists(): self._cache_file.touch() os.chmod(self._cache_file, 0o600) def _load_api_key(self, api_key): """Load API key from string or file.""" if not api_key: raise PowerCloudTokenManagerError( "_load_api_key: API key is missing", ocf.OCF_ERR_CONFIGURED, ) # API key in string if not api_key.startswith("@"): return api_key # API key in file api_key_path = Path(api_key[1:]) if not api_key_path.is_file(): raise PowerCloudTokenManagerError( f"_load_api_key: API key file not found: '{api_key_path}'", ocf.OCF_ERR_ARGS, ) try: content = api_key_path.read_text().strip() api_key_field = json.loads(content).get("apikey", "") except json.JSONDecodeError: # data is text, return as is api_key_field = content if not api_key_field: raise PowerCloudTokenManagerError( f"_load_api_key: invalid API key in file '{api_key_path}'", ocf.OCF_ERR_ARGS, ) return api_key_field def _request_new_token(self): """Request a new access token.""" headers = { "content-type": "application/x-www-form-urlencoded", "accept": "application/json", } data = { "grant_type": "urn:ibm:params:oauth:grant-type:apikey", "apikey": f"{self._api_key}", } current_time = time.time() try: response = self._session.post( self._auth_url, headers=headers, data=data, proxies=self._proxy, timeout=REQUESTS_TIMEOUT, ) response.raise_for_status() token_data = response.json() return ( token_data["access_token"], current_time + token_data["expires_in"], current_time, ) except requests.RequestException as e: ocf.logger.warning( f"[PowerCloudTokenManager] _request_new_token: failed to request token: '{e}'" ) return None def _read_cache(self): """Read token cache.""" try: with self._cache_file.open("r") as f: fcntl.flock(f, fcntl.LOCK_EX) try: return json.load(f) finally: fcntl.flock(f, fcntl.LOCK_UN) except (json.JSONDecodeError, FileNotFoundError, PermissionError): ocf.logger.warning( "[PowerCloudTokenManager] _read_cache: failed to read token cache read due to missing file or malformed JSON." ) return {} def _write_cache(self, token, expiration, refreshed_at): """Write token cache.""" try: with self._cache_file.open("w") as f: fcntl.flock(f, fcntl.LOCK_EX) try: json.dump( { "token": token, "expiration": expiration, "refreshed_at": refreshed_at, }, f, ) finally: fcntl.flock(f, fcntl.LOCK_UN) except Exception as e: raise PowerCloudTokenManagerError( f"_write_cache: failed to write token cache file: '{e}'", ocf.OCF_ERR_GENERIC, ) def _is_token_expired(self, expiration): """Check if token is expired or near expiry.""" return time.time() + self._TOKEN_REFRESH_BUFFER >= expiration def get_token(self): """Get a valid access token, using cache if enabled.""" if not self._cache_file: result = self._request_new_token() if result: token, _, _ = result return token raise PowerCloudTokenManagerError( "get_token: token request failed and no cache available", ocf.OCF_ERR_GENERIC, ) cache = self._read_cache() token = cache.get("token") expiration = cache.get("expiration", 0) if not token or self._is_token_expired(expiration): result = self._request_new_token() if result: token, expiration, refreshed_at = result refresh_time = time.ctime(refreshed_at) ocf.logger.debug( f"[PowerCloudTokenManager] get_token: refreshed token at '{refresh_time}'" ) self._write_cache(token, expiration, refreshed_at) else: ocf.logger.error( "[PowerCloudTokenManager] get_token: failed to refresh token" ) if token and time.time() < expiration: ocf.logger.warning( "[PowerCloudTokenManager] get_token: using cached token as fallback" ) else: raise PowerCloudTokenManagerError( "get_token: no valid token available", ocf.OCF_ERR_GENERIC, ) return token class PowerCloudAPIError(OCFExitError): """Exception class for errors in PowerCloudAPI.""" def __init__(self, message, exit_code): super().__init__(f"[PowerCloudAPIError] {message}", exit_code) class PowerCloudAPI: """Offers a convenient method for sending requests to the IBM Power Cloud API.""" _ALLOWED_API_TYPES = {"public", "private"} def __init__( self, api_key="", api_type="", region="", crn="", proxy="", use_cache=False, ): """Initialize class variables, including the IBM Power Cloud API endpoint URL and HTTP header, and get an API token.""" self._crn = crn self._proxy = self._get_proxy(proxy) self._api_url = self._get_api_url(region, api_type) token_manager = PowerCloudTokenManager( api_type=api_type, api_key=api_key, proxy=self._proxy, use_cache=use_cache ) self._token = token_manager.get_token() self._header = self._get_header() self._session = create_session_with_retries() def _get_proxy(self, proxy): """Validate a proxy URL and test TCP connectivity. Returns a proxy dict if reachable.""" if not proxy: return None parsed_url = urlparse(proxy) is_valid_url = ( parsed_url.hostname and parsed_url.port and parsed_url.scheme in ("http", "https") ) if not is_valid_url: raise PowerCloudAPIError( f"_get_proxy: invalid proxy URL '{proxy}'", ocf.OCF_ERR_CONFIGURED, ) try: with socket.create_connection( (parsed_url.hostname, parsed_url.port), timeout=REQUESTS_TIMEOUT ): return {"https": proxy} except OSError as e: raise PowerCloudAPIError( f"_get_proxy: cannot connect to proxy '{proxy}': {e}", ocf.OCF_ERR_ARGS, ) def _get_api_url(self, region, api_type): """Generate and return the API URL for a given region and API type.""" if not region: raise PowerCloudAPIError( "_get_api_url: missing region parameter", ocf.OCF_ERR_CONFIGURED, ) api_type = str(api_type).lower() if api_type not in self._ALLOWED_API_TYPES: raise PowerCloudAPIError( f"_get_api_url: invalid api_type: '{api_type}', must be one of {self._ALLOWED_API_TYPES} ", ocf.OCF_ERR_CONFIGURED, ) if api_type == "public" and not self._proxy: raise PowerCloudAPIError( "_get_api_url: api_type 'public' requires a proxy", ocf.OCF_ERR_CONFIGURED, ) subdomain = "private." if api_type == "private" else "" return f"https://{subdomain}{region}.power-iaas.cloud.ibm.com" def _get_header(self): """Construct request header.""" return { "Authorization": f"Bearer {self._token}", "CRN": self._crn, "Content-Type": "application/json", } def send_api_request(self, method, resource, **kwargs): """Perform an HTTP API call to the specified resource using the given method""" url = f"{self._api_url}{resource}" method = method.upper() ocf.logger.debug(f"[PowerCloudAPI] send_api_request: '{method}' '{resource}'") try: response = self._session.request( method, url, headers=self._header, proxies=self._proxy, timeout=REQUESTS_TIMEOUT, **kwargs, ) response.raise_for_status() return response.json() except requests.RequestException as e: raise PowerCloudAPIError( f"send_api_request: request error occured: '{method}' - '{url}' - '{e}'", ocf.OCF_ERR_GENERIC, ) class PowerCloudRouteError(OCFExitError): """Exception class for errors encountered while managing PowerVS network routes.""" def __init__(self, message, exit_code): super().__init__(f"[PowerCloudRouteError] {message}", exit_code) class PowerCloudRoute(PowerCloudAPI): """Provides methods for managing network routes in Power Virtual Server.""" _CRN_PREFIX_INDEX = 0 _CRN_TYPE_INDEX = 8 _CRN_ROUTE_ID_INDEX = 9 _CRN_EXPECTED_LENGTH = 10 def __init__( self, ip="", api_key="", api_type="", region="", route_host_map="", device="", proxy="", monitor_api="", use_token_cache="", is_remote_route=False, ): """Initialize PowerCloudRoute instance.""" self._is_remote_route = is_remote_route self.ip = self._get_ip_info(ip) self.crn, self.route_id = self._parse_route_map(route_host_map) use_cache = str(use_token_cache).lower() == "true" super().__init__( api_key=api_key, api_type=api_type, region=region, crn=self.crn, proxy=proxy, use_cache=use_cache, ) self.route_info = self._get_route_info() self.route_name = self.route_info["name"] self.device = self._get_device_name(device) def _get_ip_info(self, ip): """Validate the given IP address and return its standard form.""" try: return str(ipaddress.ip_address(ip)) except ValueError: raise PowerCloudRouteError( f"_get_ip_info: invalid IP address '{ip}'", ocf.OCF_ERR_CONFIGURED, ) def _parse_route_crn(self, route_crn): """Parses a PowerVS route CRN and extract its base CRN and route ID.""" crn_parts = route_crn.split(":") if ( len(crn_parts) != self._CRN_EXPECTED_LENGTH or crn_parts[self._CRN_PREFIX_INDEX] != "crn" or crn_parts[self._CRN_TYPE_INDEX] != "route" ): raise PowerCloudAPIError( f"_parse_route_crn: invalid CRN format for network-route: '{route_crn}'", ocf.OCF_ERR_CONFIGURED, ) workspace_crn = ":".join(crn_parts[: self._CRN_TYPE_INDEX]) + "::" route_id = crn_parts[self._CRN_ROUTE_ID_INDEX] return workspace_crn, route_id def _parse_route_map(self, route_host_map): """Validate the route host map and extract the associated CRN and route ID.""" try: route_map = dict(item.split(":", 1) for item in route_host_map.split(";")) except ValueError: raise PowerCloudRouteError( f"_parse_route_map: invalid route_host_map format: '{route_host_map}'", ocf.OCF_ERR_CONFIGURED, ) hostname = os.uname().nodename # set nodename to local hostname or get hostname of remote host from route_map nodename = ( hostname if not self._is_remote_route else next((h for h in route_map if h != hostname), None) ) if not nodename or nodename not in route_map: raise PowerCloudRouteError( f"_parse_route_map: hostname '{nodename}' not found in route_host_map '{route_host_map}'", ocf.OCF_ERR_CONFIGURED, ) return self._parse_route_crn(route_map[nodename]) def _get_route_info(self): """Retrieve and validate attributes of a PowerVS network route.""" resource = f"/v1/routes/{self.route_id}" route_info = self.send_api_request("GET", resource) zone = "remote" if self._is_remote_route else "local" ocf.logger.debug( f"[PowerCloudRoute] _get_route_info: {zone} route info: '{route_info}'" ) if self.ip != route_info["destination"]: raise PowerCloudRouteError( f"_get_route_info: IP '{self.ip}' does not match the route destination address '{route_info['destination']}'", ocf.OCF_ERR_CONFIGURED, ) if route_info["advertise"] != "enable": raise PowerCloudRouteError( f"_get_route_info: route '{route_info['name']}' advertise flag must be set to enable", ocf.OCF_ERR_CONFIGURED, ) return route_info def _get_device_name(self, name): """Verify the existence of a network interface with the specified name.""" if self._is_remote_route: return "" if name: if ip_check_device(name): return name raise PowerCloudRouteError( f"_get_device_name: network interface '{name}' does not exist or is down", ocf.OCF_ERR_CONFIGURED, ) next_hop = self.route_info["nextHop"] interface_name = ip_find_device(next_hop) if interface_name: return interface_name raise PowerCloudRouteError( f"_get_device_name: network interface with next hop '{next_hop}' does not exist or is down", ocf.OCF_ERR_CONFIGURED, ) def _set_route_enabled(self, enabled: bool): """Enable or disable the PowerVS network route.""" resource = f"/v1/routes/{self.route_id}" data = json.dumps({"enabled": enabled}) state = "enabled" if enabled else "disabled" response = self.send_api_request("PUT", resource, data=data) ocf.logger.debug( f"[PowerCloudRoute] _set_route_enabled: successfully {state} route '{self.route_name}', response: '{response}'" ) def is_enabled(self): """Check whether the PowerVS network route is currently enabled.""" return self.route_info["state"] == "deployed" def enable(self): """Enable the PowerVS network route.""" if not self.is_enabled(): self._set_route_enabled(True) def disable(self): """Disable the PowerVS network route.""" if self.is_enabled(): self._set_route_enabled(False) def create_route_instance(options, is_remote_route=False, catch_exception=False): """Instantiate a PowerCloudRoute object and handle errors. Returns: - PowerCloudRoute: The initialized route object if successful. - None: If an error occurs and catch_exception is True. Raises: - PowerCloudRouteError: If instantiation fails and catch_exception is False. """ # Filter only the valid resource agent options from options dictionary. resource_options = {k: options.get(k, "") for k in RESOURCE_OPTIONS} try: return PowerCloudRoute(**resource_options, is_remote_route=is_remote_route) except Exception as e: zone = "remote" if is_remote_route else "local" ocf.logger.error( f"[create_route_instance]: failed to instantiate {zone} route: '{e}'" ) if catch_exception: return None raise def start_action( ip="", api_key="", api_type="", region="", route_host_map="", use_token_cache="", monitor_api="", device="", proxy="", ): """Assign the service IP. This function performs the following actions: - Adds the specified IP address as an alias to the given network interface or the interface matching the route's next hop. - Disables the remote network route. - Enables the network route associated with the provided route host map. """ resource_options = locals() ocf.logger.info("[start_action]: enabling overlay IP") ocf.logger.debug(f"[start_action]: options: '{resource_options}'") remote_route = create_route_instance(resource_options, is_remote_route=True) # Disable remote route ocf.logger.debug( f"[start_action]: disabling remote route '{remote_route.route_name}'" ) remote_route.disable() local_route = create_route_instance(resource_options) # Add IP alias ip_alias_add(ip, local_route.device) # Enable local route ocf.logger.debug(f"[start_action]: enabling local route '{local_route.route_name}'") local_route.enable() monitor_result = monitor_action(**resource_options) if monitor_result != ocf.OCF_SUCCESS: raise PowerCloudRouteError( f"start_action: failed to enable local route '{local_route.route_name}'", monitor_result, ) ocf.logger.info( f"[start_action]: successfully added IP alias '{ip}' and enabled local route '{local_route.route_name}'" ) return ocf.OCF_SUCCESS def stop_action( ip="", api_key="", api_type="", region="", route_host_map="", use_token_cache="", monitor_api="", device="", proxy="", ): """Remove the service IP. This function performs the following actions: - Disables the network route associated with the provided route host map. - Removes the IP alias from the network interface. """ resource_options = locals() ocf.logger.info("[stop_action]: disabling overlay IP") ocf.logger.debug(f"[stop_action]: options: '{resource_options}'") try: remote_route = create_route_instance(resource_options, is_remote_route=True) ocf.logger.debug( f"[stop_action]: disabling remote route '{remote_route.route_name}'" ) remote_route.disable() local_route = create_route_instance(resource_options) ocf.logger.debug( f"[stop_action]: disabling local route '{local_route.route_name}'" ) local_route.disable() finally: # Remove IP alias ip_alias_remove(ip) monitor_result = monitor_action(**resource_options) if monitor_result != ocf.OCF_NOT_RUNNING: raise PowerCloudRouteError( f"stop_action: failed to disable local route '{local_route.route_name}'", monitor_result, ) ocf.logger.info( f"[stop_action]: successfully removed IP alias '{ip}' and disabled local route '{local_route.route_name}'" ) return ocf.OCF_SUCCESS def monitor_action( ip="", api_key="", api_type="", region="", route_host_map="", use_token_cache="", monitor_api="", device="", proxy="", ): """Monitor the service IP. Checks the status of the assigned service IP address. """ resource_options = locals() is_probe = ocf.is_probe() use_extended_monitor = ocf.OCF_ACTION == "start" or ( str(monitor_api).lower() == "true" and not is_probe ) ocf.logger.debug( f"[monitor_action]: options: '{resource_options}', is_probe: '{is_probe}'" ) interface_name = ip_find_device(ip) if not use_extended_monitor: if interface_name: ocf.logger.debug( f"[monitor_action]: IP alias '{ip}' is active'" ) return ocf.OCF_SUCCESS else: ocf.logger.debug( f"[monitor_action]: IP alias '{ip}' is not active" ) return ocf.OCF_NOT_RUNNING remote_route = create_route_instance( resource_options, is_remote_route=True, catch_exception=True ) if remote_route is None: ocf.logger.error("[monitor_action]: failed to instantiate remote route") return ocf.OCF_ERR_GENERIC elif remote_route.is_enabled(): ocf.logger.error( f"[monitor_action]: remote route '{remote_route.route_name}' is enabled" ) return ocf.OCF_ERR_GENERIC local_route = create_route_instance( resource_options, is_remote_route=False, catch_exception=True ) if local_route is None: ocf.logger.error("[monitor_action]: failed to instantiate local route") return ocf.OCF_ERR_GENERIC if interface_name: if local_route.is_enabled(): ocf.logger.debug( f"[monitor_action]: IP alias '{ip}' is active, local route '{local_route.route_name}' is enabled" ) return ocf.OCF_SUCCESS else: ocf.logger.error( f"[monitor_action]: local route '{local_route.route_name}' is not enabled" ) return ocf.OCF_ERR_GENERIC else: if local_route.is_enabled(): ocf.logger.error( f"[monitor_action]: local route '{local_route.route_name}' is enabled, but IP alias is not configured" ) return ocf.OCF_ERR_GENERIC else: ocf.logger.debug( f"[monitor_action]: IP alias '{ip}' is not active and local route '{local_route.route_name}' is disabled" ) return ocf.OCF_NOT_RUNNING def validate_all_action( ip="", api_key="", api_type="", region="", route_host_map="", use_token_cache="", monitor_api="", device="", proxy="", ): """Validate resource agent parameters. Verifies the provided resource agent options by attempting to instantiate route objects for both local and remote routes. """ resource_options = locals() ocf.logger.info("[validate_all_action]: validate local and remote routes") _ = create_route_instance(resource_options) _ = create_route_instance(resource_options, is_remote_route=True) return ocf.OCF_SUCCESS def main(): """Instantiate the resource agent.""" agent_description = textwrap.dedent("""\ Resource Agent to move an IP address from one Power Virtual Server instance to another. Prerequisites: 1. Red Hat Enterprise Linux 9.4 or higher 2. Two-node cluster - Distributed across two PowerVS workspaces in separate data centers within the same region. 3. IBM Cloud API Key: - Create a service API key with privileges for both workspaces. - Save the key in a file and copy it to both cluster nodes using the same path and filename. - Reference the key file path in the resource definition. For detailed guidance on high availability for SAP applications on PowerVS, visit: https://cloud.ibm.com/docs/sap?topic=sap-ha-overview. """) agent = ocf.Agent( "powervs-move-ip", shortdesc="Manages Power Virtual Server overlay IP routes.", longdesc=agent_description, - version=0.90, + version=1.00, ) agent.add_parameter( "ip", shortdesc="IP address", longdesc=( "The virtual IP address is the destination address of a network route." ), content_type="string", required=True, ) agent.add_parameter( "api_key", shortdesc="API Key or @API_KEY_FILE_PATH", longdesc=( "API Key or @API_KEY_FILE_PATH for IBM Cloud access. " "The API key content or the path of an API key file that is indicated by the @ symbol." ), content_type="string", required=True, ) agent.add_parameter( "api_type", shortdesc="API type", longdesc="Connect to Power Virtual Server regional endpoints over a public or private network (public|private).", content_type="string", default="private", required=True, ) agent.add_parameter( "region", shortdesc="Power Virtual Server region", longdesc=( "Region that represents the geographic area where the instance is located. " "The region is used to identify the Cloud API endpoint." ), content_type="string", required=True, ) agent.add_parameter( "route_host_map", shortdesc="Mapping of hostnames to IBM Cloud route CRNs", longdesc=( "Map the hostname of the Power Virtual Server instance to the route CRN of the overlay IP route. " "Separate hostname and route CRN with a colon ':', separate different hostname and route CRN pairs with a semicolon ';'. " "Example: hostname1:route-crn-of-instance1;hostname2:route-crn-of-instance2" ), content_type="string", required=True, ) agent.add_parameter( "use_token_cache", shortdesc="Enable API token cache", longdesc="Enable caching of the API access token in a local file to reduce authentication overhead. ", content_type="string", default="True", required=False, ) agent.add_parameter( "monitor_api", shortdesc="Enhanced API Monitoring", longdesc="Enable enhanced monitoring by using Power Cloud API calls to verify route configuration correctness. ", content_type="string", default="False", required=False, ) agent.add_parameter( "device", shortdesc="Network adapter for the overlay IP address", longdesc=( "Network adapter for the overlay IP address. " "The adapter must have the same name on all Power Virtual Server instances. " "If the `device` parameter is not specified, the IP alias is assigned to the interface whose configured IP address matches the route's next hop address. " ), content_type="string", default="", required=False, ) agent.add_parameter( "proxy", shortdesc="Proxy", longdesc=( "Proxy server used to access IBM Cloud API endpoints. " "The value must be a valid URL in the format 'http[s]://hostname:port'. " ), content_type="string", default="", required=False, ) agent.add_action("start", timeout=60, handler=start_action) agent.add_action("stop", timeout=60, handler=stop_action) agent.add_action( "monitor", depth=0, timeout=60, interval=60, handler=monitor_action ) agent.add_action("validate-all", timeout=60, handler=validate_all_action) agent.run() if __name__ == "__main__": main()