portforward

import os import time import requests from portforward import Portforward from kubernetes import config # --- Configuration for the quickstart example --- # Ensure you have a valid KubeConfig file (~/.kube/config) and a running Kubernetes cluster. # Replace 'nginx-pod' and 'default' with actual pod name and namespace in your cluster. # Set K8S_NAMESPACE, K8S_POD_NAME, K8S_POD_TARGET_PORT, K8S_LOCAL_PORT environment variables # or modify the defaults below to match your setup. namespace = os.environ.get("K8S_NAMESPACE", "default") pod_name = os.environ.get("K8S_POD_NAME", "nginx-pod") # Example: 'nginx-xxxxxx-xxxxx' target_port = int(os.environ.get("K8S_POD_TARGET_PORT", "80")) local_port = int(os.environ.get("K8S_LOCAL_PORT", "8080")) print(f"--- Kubernetes Port-Forward Quickstart ---") try: # Load Kubernetes configuration config.load_kube_config() print(f"Attempting to forward port {target_port} from pod '{pod_name}' in namespace '{namespace}' to local port {local_port}...") # Establish the port-forward connection using a context manager with Portforward.from_pod( namespace=namespace, pod_name=pod_name, target_port=target_port, local_port=local_port ) as pf: print(f"Successfully established port-forward: http://localhost:{pf.local_port}") print("You can now access your pod/service via this local address.") # Example: Make a request to the forwarded port print("\nAttempting a sample HTTP GET request via the forwarded port...") try: # This will only succeed if the pod exposes an HTTP service on the target_port response = requests.get(f"http://localhost:{pf.local_port}", timeout=5) print(f"Request successful! Status code: {response.status_code}") print(f"Response (first 100 chars): {response.text[:100]}...") except requests.exceptions.ConnectionError: print("Could not connect to the forwarded port. Is the target service running and exposing HTTP?") print("Please ensure the pod name, namespace, and target port are correct.") except Exception as e: print(f"An unexpected error occurred during the sample request: {e}") print("\nKeeping port-forward active for 15 seconds. Press Ctrl+C to stop sooner.") time.sleep(15) # Keep the port forward open for a duration print("Port-forward session ending.") except config.ConfigException as e: print(f"Error loading Kubernetes configuration: {e}") print("Please ensure your KubeConfig file is valid and accessible (e.g., at ~/.kube/config) or your environment is configured for in-cluster access.") except Exception as e: print(f"Failed to establish port-forward: {e}") print("Common causes: pod not found, incorrect namespace/pod_name, target_port not exposed, RBAC issues, or local port already in use.")