Add first-class support for external destination clusters in ArgoCD with strict error handling

docs/content/reference/cluster-connectivity.md

@@ -7,6 +7,60 @@ When using Nyl as an ArgoCD plugin, to enable the plugin to reach out to the Kub
 `argocd-repo-server` service account with the necessary permissions. See [ArgoCD Plugin](./argocd-plugin.md) for more
 information.
 
+## External destination cluster support
+
+**New in v0.10.6**: Nyl now supports external destination clusters when used as an ArgoCD plugin. This means that
+`lookup()` calls will automatically work against the destination cluster specified in your ArgoCD Application, even
+when it's different from the cluster where ArgoCD itself is running.
+
+### How it works
+
+When Nyl detects that it's running in an ArgoCD environment (via ArgoCD environment variables), it will:
+
+1. **Fetch the Application resource** from the local ArgoCD cluster to determine the destination cluster
+2. **Lookup cluster credentials** from ArgoCD's cluster secrets
+3. **Create a destination cluster API client** using the found credentials
+4. **Use the destination client** for all `lookup()` calls
+
+This process is completely transparent and requires no additional configuration. Your existing ArgoCD Applications will
+automatically gain external cluster lookup support.
+
+### Backwards compatibility
+
+- **Non-ArgoCD usage**: Continues to work unchanged using your kubeconfig or Nyl profiles
+- **Same-cluster deployments**: ArgoCD Applications deploying to the same cluster as ArgoCD continue to work as before
+- **External cluster deployments**: Now work automatically with lookup support
+- **Fallback behavior**: If destination cluster credentials cannot be found, Nyl falls back to using the ArgoCD cluster
+
+### Requirements
+
+For external cluster lookup support to work:
+
+1. The ArgoCD Application must specify a destination cluster
+2. The destination cluster must be registered in ArgoCD with appropriate credentials
+3. The `argocd-repo-server` must have permissions to read Applications and cluster secrets in the ArgoCD namespace
+
+Example ArgoCD Application with external destination:
+
+```yaml
+apiVersion: argoproj.io/v1alpha1
+kind: Application
+metadata:
+  name: my-app
+  namespace: argocd
+spec:
+  project: default
+  source:
+    repoURL: https://github.com/myorg/gitops.git
+    path: manifests
+    plugin:
+      name: nyl-v1
+  destination:
+    # External cluster - lookups will work against this cluster
+    server: https://external-k8s.example.com
+    namespace: production
+```
+
 ## Kubernetes API versions
 
 When Nyl invokes `helm template`, it must pass along a full list of all available API versions in the cluster to
@@ -23,5 +77,8 @@ Nyl provides a `lookup()` function that allows the Helm chart to query the Kuber
 resource to use in the chart. This is an optional feature that your manifests may simply decide not to rely on,
 however it is a powerful feature to pass and transform values from existing resources.
 
+With the new external destination cluster support, lookups will automatically query the correct cluster based on your
+ArgoCD Application configuration.
+
 TODO: Implement security to prevent lookups for resources that the corresponding ArgoCD project has no access to.
 This will require a safe evaluation language instead of Python `eval()`.

src/nyl/commands/template.py

@@ -30,6 +30,7 @@
 from nyl.tools.kubernetes import drop_empty_metadata_labels, populate_namespace_to_resources
 from nyl.tools.logging import lazy_str
 from nyl.tools.types import Resource, ResourceList
+from nyl.tools.argocd import detect_argocd_context, create_destination_client
 
 DEFAULT_NAMESPACE_ANNOTATION = "nyl.io/is-default-namespace"
 
@@ -193,7 +194,30 @@ def template(
     PROVIDER.set(
         ApiClientConfig, ApiClientConfig(in_cluster=in_cluster, profile=profile if connect_with_profile else None)
     )
-    client = PROVIDER.get(ApiClient)
+
+    # Check if running in ArgoCD context and try to create destination cluster client
+    argocd_context = detect_argocd_context()
+    if argocd_context:
+        logger.info("Detected ArgoCD context: app={}/{}, project={}", 
+                   argocd_context.app_namespace, argocd_context.app_name, argocd_context.project_name)
+
+        # Get the local ArgoCD cluster client first
+        argocd_client = PROVIDER.get(ApiClient)
+
+        # Try to create destination cluster client
+        destination_client = create_destination_client(argocd_client, argocd_context)
+        if destination_client:
+            logger.info("Using destination cluster client for lookups")
+            client = destination_client
+        else:
+            raise RuntimeError(
+                f"Failed to create destination cluster client for ArgoCD application "
+                f"{argocd_context.app_namespace}/{argocd_context.app_name}. "
+                f"Please ensure the destination cluster is properly configured in ArgoCD."
+            )
+    else:
+        # Not in ArgoCD context, use normal client
+        client = PROVIDER.get(ApiClient)
 
     project = PROVIDER.get(ProjectConfig)
     if generate_applysets is not None:

src/nyl/tools/argocd.py

@@ -0,0 +1,263 @@
+"""
+ArgoCD integration utilities for Nyl.
+
+This module provides utilities for detecting ArgoCD environment and resolving
+destination cluster credentials for external cluster support.
+"""
+
+import base64
+import json
+import os
+from dataclasses import dataclass
+from typing import Optional
+
+from kubernetes.client.api_client import ApiClient
+from kubernetes.config import new_client_from_config_dict
+from kubernetes.dynamic.client import DynamicClient
+from kubernetes.client.exceptions import ApiException
+from loguru import logger
+
+
+@dataclass
+class ArgoCDContext:
+    """
+    ArgoCD context information extracted from environment variables.
+    """
+    app_name: str
+    app_namespace: str
+    project_name: str
+    revision: Optional[str] = None
+    source_path: Optional[str] = None
+    source_repo_url: Optional[str] = None
+
+
+@dataclass
+class ClusterInfo:
+    """
+    Information about a destination cluster from ArgoCD.
+    """
+    name: Optional[str]
+    server: str
+    config: dict
+
+
+def detect_argocd_context() -> Optional[ArgoCDContext]:
+    """
+    Detect if running in ArgoCD environment by checking for ArgoCD environment variables.
+
+    Returns:
+        ArgoCDContext if running in ArgoCD, None otherwise.
+    """
+    app_name = os.getenv("ARGOCD_APP_NAME")
+    app_namespace = os.getenv("ARGOCD_APP_NAMESPACE") 
+    project_name = os.getenv("ARGOCD_APP_PROJECT_NAME")
+
+    if not all([app_name, app_namespace, project_name]):
+        return None
+
+    return ArgoCDContext(
+        app_name=app_name,
+        app_namespace=app_namespace,
+        project_name=project_name,
+        revision=os.getenv("ARGOCD_APP_REVISION"),
+        source_path=os.getenv("ARGOCD_APP_SOURCE_PATH"),
+        source_repo_url=os.getenv("ARGOCD_APP_SOURCE_REPO_URL")
+    )
+
+
+def get_application_destination(argocd_client: ApiClient, context: ArgoCDContext) -> Optional[ClusterInfo]:
+    """
+    Fetch the ArgoCD Application resource and extract destination cluster information.
+
+    Args:
+        argocd_client: Kubernetes API client for the ArgoCD cluster
+        context: ArgoCD context information
+
+    Returns:
+        ClusterInfo for the destination cluster, or None if not found
+    """
+    try:
+        dynamic_client = DynamicClient(argocd_client)
+
+        # Get the Application resource
+        application_resource = dynamic_client.resources.get(
+            api_version="argoproj.io/v1alpha1", 
+            kind="Application"
+        )
+
+        application = application_resource.get(
+            name=context.app_name,
+            namespace=context.app_namespace
+        )
+
+        destination = application.spec.destination
+        cluster_name = getattr(destination, 'name', None)
+        cluster_server = getattr(destination, 'server', None)
+
+        if not cluster_server:
+            logger.warning("Application {} has no destination server configured", context.app_name)
+            return None
+
+        logger.debug("Found destination cluster: name={}, server={}", cluster_name, cluster_server)
+
+        return ClusterInfo(
+            name=cluster_name,
+            server=cluster_server,
+            config={}  # Will be populated by get_cluster_credentials
+        )
+
+    except ApiException as e:
+        logger.warning("Failed to fetch Application {}/{}: {}", context.app_namespace, context.app_name, e)
+        return None
+    except Exception as e:
+        logger.warning("Error getting application destination: {}", e)
+        return None
+
+
+def get_cluster_credentials(argocd_client: ApiClient, cluster_info: ClusterInfo) -> Optional[dict]:
+    """
+    Fetch cluster credentials from ArgoCD cluster secrets.
+
+    Args:
+        argocd_client: Kubernetes API client for the ArgoCD cluster
+        cluster_info: Information about the destination cluster
+
+    Returns:
+        Kubernetes config dict for the destination cluster, or None if not found
+    """
+    try:
+        dynamic_client = DynamicClient(argocd_client)
+
+        # Get secrets with ArgoCD cluster label
+        secret_resource = dynamic_client.resources.get(api_version="v1", kind="Secret")
+
+        # List all secrets in the ArgoCD namespace with cluster label
+        secrets = secret_resource.get(
+            namespace=os.getenv("ARGOCD_APP_NAMESPACE", "argocd"),
+            label_selector="argocd.argoproj.io/secret-type=cluster"
+        )
+
+        for secret in secrets.items:
+            if not hasattr(secret, 'data') or not secret.data:
+                continue
+
+            # Decode the cluster server URL from the secret
+            try:
+                server_data = secret.data.get('server')
+                if not server_data:
+                    continue
+
+                server_url = base64.b64decode(server_data).decode('utf-8')
+
+                # Match against our target cluster
+                if server_url == cluster_info.server:
+                    logger.debug("Found cluster credentials secret: {}", secret.metadata.name)
+
+                    # Decode cluster configuration
+                    config_data = secret.data.get('config')
+                    if not config_data:
+                        logger.warning("Cluster secret {} has no config data", secret.metadata.name)
+                        continue
+
+                    config_json = base64.b64decode(config_data).decode('utf-8')
+                    config = json.loads(config_json)
+
+                    # Build kubernetes config dict
+                    kube_config = {
+                        'apiVersion': 'v1',
+                        'kind': 'Config',
+                        'clusters': [{
+                            'name': cluster_info.name or 'argocd-cluster',
+                            'cluster': {
+                                'server': cluster_info.server,
+                                'certificate-authority-data': config.get('tlsClientConfig', {}).get('caData'),
+                                'insecure-skip-tls-verify': config.get('tlsClientConfig', {}).get('insecure', False)
+                            }
+                        }],
+                        'users': [{
+                            'name': 'argocd-user',
+                            'user': {}
+                        }],
+                        'contexts': [{
+                            'name': 'argocd-context',
+                            'context': {
+                                'cluster': cluster_info.name or 'argocd-cluster',
+                                'user': 'argocd-user'
+                            }
+                        }],
+                        'current-context': 'argocd-context'
+                    }
+
+                    # Add authentication data
+                    user_config = kube_config['users'][0]['user']
+                    tls_config = config.get('tlsClientConfig', {})
+
+                    if 'certData' in tls_config and 'keyData' in tls_config:
+                        user_config['client-certificate-data'] = tls_config['certData']
+                        user_config['client-key-data'] = tls_config['keyData']
+
+                    if 'bearerToken' in config:
+                        user_config['token'] = config['bearerToken']
+
+                    # Handle AWS IAM authenticator
+                    if 'execProviderConfig' in config:
+                        exec_config = config['execProviderConfig']
+                        user_config['exec'] = {
+                            'apiVersion': exec_config.get('apiVersion', 'client.authentication.k8s.io/v1beta1'),
+                            'command': exec_config.get('command'),
+                            'args': exec_config.get('args', []),
+                            'env': exec_config.get('env', [])
+                        }
+
+                    return kube_config
+
+            except (ValueError, KeyError, TypeError) as e:
+                logger.debug("Failed to decode cluster secret {}: {}", secret.metadata.name, e)
+                continue
+
+        logger.warning("No cluster credentials found for server: {}", cluster_info.server)
+        return None
+
+    except ApiException as e:
+        logger.warning("Failed to list cluster secrets: {}", e)
+        return None
+    except Exception as e:
+        logger.warning("Error getting cluster credentials: {}", e)
+        return None
+
+
+def create_destination_client(argocd_client: ApiClient, context: ArgoCDContext) -> Optional[ApiClient]:
+    """
+    Create a Kubernetes API client for the destination cluster.
+
+    Args:
+        argocd_client: Kubernetes API client for the ArgoCD cluster  
+        context: ArgoCD context information
+
+    Returns:
+        ApiClient for the destination cluster, or None if unable to create
+    """
+    try:
+        # Get destination cluster info from Application
+        cluster_info = get_application_destination(argocd_client, context)
+        if not cluster_info:
+            return None
+
+        # Special case: if destination is the same as ArgoCD cluster, return the ArgoCD client
+        if cluster_info.server in ['https://kubernetes.default.svc', 'https://kubernetes.default.svc.cluster.local']:
+            logger.debug("Destination cluster is the same as ArgoCD cluster, using ArgoCD client")
+            return argocd_client
+
+        # Get credentials for destination cluster
+        kube_config = get_cluster_credentials(argocd_client, cluster_info)
+        if not kube_config:
+            return None
+
+        # Create API client from config
+        client = new_client_from_config_dict(kube_config)
+        logger.info("Successfully created destination cluster client for: {}", cluster_info.server)
+        return client
+
+    except Exception as e:
+        logger.warning("Failed to create destination cluster client: {}", e)
+        return None