Alicloud Network Alb

interface:
  display_name: "Alibaba Cloud Network ALB"
  short_description: "ALB instance, listener, server group, health check, and troubleshooting workflows"
  default_prompt: "Use $alicloud-network-alb to complete this ALB task on Alibaba Cloud."

#!/usr/bin/env python3
"""Add backend servers to an ALB server group.

Supports ECS instances, ENI, ECI, and IP-based servers.

Examples:
    # Add a single ECS server
    python add_servers.py --region cn-hangzhou --sg-id sgp-xxx \\
        --server ecs:i-xxx:8080

    # Add multiple servers with weight
    python add_servers.py --region cn-hangzhou --sg-id sgp-xxx \\
        --server ecs:i-xxx:8080:100 \\
        --server ecs:i-yyy:8080:50

    # Add ENI server
    python add_servers.py --region cn-hangzhou --sg-id sgp-xxx \\
        --server eni:eni-xxx:8080

    # Add IP-based server (for Ip-type server group)
    python add_servers.py --region cn-hangzhou --sg-id sgp-xxx \\
        --server ip:10.0.1.100:8080

    # Dry run
    python add_servers.py --region cn-hangzhou --sg-id sgp-xxx \\
        --server ecs:i-xxx:8080 --dry-run
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any

from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_alb20200616 import models as alb_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Alb20200616Client:
    """Create ALB client with credentials from environment."""
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"alb.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not ak or not sk:
        raise RuntimeError("ALICLOUD_ACCESS_KEY_ID and ALICLOUD_ACCESS_KEY_SECRET must be set")
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token
    return Alb20200616Client(config)


SERVER_TYPE_MAP = {
    "ecs": "Ecs",
    "eni": "Eni",
    "eci": "Eci",
    "ip": "Ip",
}


def parse_server(spec: str) -> dict[str, Any]:
    """Parse server spec: type:id:port[:weight[:description]].

    For IP type: ip:address:port[:weight[:description]]
    """
    parts = spec.split(":")
    if len(parts) < 3:
        raise ValueError(
            f"Invalid server spec '{spec}'. "
            "Format: type:id:port[:weight[:description]]  "
            "e.g. ecs:i-xxx:8080 or ecs:i-xxx:8080:100:web-server"
        )

    raw_type = parts[0].lower()
    server_type = SERVER_TYPE_MAP.get(raw_type)
    if not server_type:
        raise ValueError(
            f"Unknown server type '{raw_type}'. Supported: {', '.join(SERVER_TYPE_MAP.keys())}"
        )

    server: dict[str, Any] = {"server_type": server_type, "port": int(parts[2])}

    if server_type == "Ip":
        server["server_ip"] = parts[1]
    else:
        server["server_id"] = parts[1]

    if len(parts) >= 4 and parts[3]:
        server["weight"] = int(parts[3])
    else:
        server["weight"] = 100

    if len(parts) >= 5 and parts[4]:
        server["description"] = parts[4]

    return server


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Add backend servers to ALB server group",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Server spec format: type:id:port[:weight[:description]]
  type:   ecs | eni | eci | ip
  id:     instance/ENI/ECI ID, or IP address for 'ip' type
  port:   backend port
  weight: 0-100, default 100
  description: optional

Examples:
  %(prog)s --region cn-hangzhou --sg-id sgp-xxx --server ecs:i-xxx:8080
  %(prog)s --region cn-hangzhou --sg-id sgp-xxx --server ecs:i-xxx:8080:100:web
  %(prog)s --region cn-hangzhou --sg-id sgp-xxx --server ip:10.0.1.100:8080:50
        """,
    )
    parser.add_argument("--region", required=True, help="Region ID")
    parser.add_argument("--sg-id", required=True, dest="server_group_id", help="Server group ID")
    parser.add_argument(
        "--server", required=True, action="append", dest="servers",
        help="Server spec: type:id:port[:weight[:description]]",
    )
    parser.add_argument("--dry-run", action="store_true", help="Print request without executing")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--output", help="Write output to file")

    args = parser.parse_args()

    try:
        parsed_servers = [parse_server(s) for s in args.servers]
    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1

    if args.dry_run:
        output = json.dumps({
            "server_group_id": args.server_group_id,
            "servers": parsed_servers,
        }, indent=2, ensure_ascii=False)
        if args.output:
            Path(args.output).write_text(output + "\n", encoding="utf-8")
            print(f"Dry-run request written to {args.output}")
        else:
            print("Dry-run mode - would add servers:")
            print(output)
        return 0

    client = create_client(args.region)

    sdk_servers = []
    for s in parsed_servers:
        srv = alb_models.AddServersToServerGroupRequestServers(
            server_type=s["server_type"],
            port=s["port"],
            weight=s.get("weight", 100),
        )
        if "server_id" in s:
            srv.server_id = s["server_id"]
        if "server_ip" in s:
            srv.server_ip = s["server_ip"]
        if "description" in s:
            srv.description = s["description"]
        sdk_servers.append(srv)

    try:
        resp = client.add_servers_to_server_group(alb_models.AddServersToServerGroupRequest(
            server_group_id=args.server_group_id,
            servers=sdk_servers,
        ))
        result = {"job_id": resp.body.job_id, "request_id": resp.body.request_id}
    except Exception as e:
        print(f"Error adding servers: {e}", file=sys.stderr)
        return 1

    if args.json:
        output = json.dumps(result, indent=2, ensure_ascii=False)
    else:
        output = f"Servers added successfully ({len(parsed_servers)} server(s)):\n  Job ID: {result['job_id']}"

    if args.output:
        Path(args.output).write_text(output + "\n", encoding="utf-8")
        print(f"Output written to {args.output}")
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Check ALB listener health status and report unhealthy backends."""

from __future__ import annotations

import argparse
import json
import os

from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_alb20200616 import models as alb_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Alb20200616Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"alb.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not ak or not sk:
        raise RuntimeError("ALICLOUD_ACCESS_KEY_ID and ALICLOUD_ACCESS_KEY_SECRET must be set")
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token
    return Alb20200616Client(config)


def list_all_listeners(client: Alb20200616Client, lb_id: str) -> list:
    """Fetch all listeners for a given ALB instance."""
    listeners: list = []
    next_token: str | None = None
    while True:
        req = alb_models.ListListenersRequest(
            load_balancer_ids=[lb_id],
            max_results=100,
            next_token=next_token,
        )
        resp = client.list_listeners(req)
        listeners.extend(resp.body.listeners or [])
        next_token = resp.body.next_token
        if not next_token:
            break
    return listeners


def _get_all_health_status(client: Alb20200616Client, listener_id: str):
    """Fetch all health status pages for a listener."""
    all_status = []
    all_rule_status = []
    next_token: str | None = None
    while True:
        req = alb_models.GetListenerHealthStatusRequest(
            listener_id=listener_id,
            include_rule=True,
            max_results=30,
            next_token=next_token,
        )
        resp = client.get_listener_health_status(req)
        all_status.extend(resp.body.listener_health_status or [])
        all_rule_status.extend(resp.body.rule_health_status or [])
        next_token = resp.body.next_token
        if not next_token:
            break
    return all_status, all_rule_status


def _format_unhealthy_servers(non_normal, indent: str = "    ") -> list[str]:
    """Format unhealthy server entries."""
    lines: list[str] = []
    for srv in non_normal:
        server_ip = srv.server_ip or "?"
        port = srv.port or "?"
        status = srv.status or "?"
        reason_str = ""
        if srv.reason:
            reason_code = srv.reason.reason_code or ""
            actual = srv.reason.actual_response or ""
            expected = srv.reason.expected_response or ""
            reason_str = reason_code
            if actual or expected:
                reason_str += f" (expected:{expected}, actual:{actual})"
        line = f"{indent}- {server_ip}:{port} [{status}]"
        if reason_str:
            line += f" {reason_str}"
        lines.append(line)
    return lines


def check_listener_health(client: Alb20200616Client, listener_id: str) -> list[str]:
    """Check health status for a single listener, return formatted lines."""
    lines: list[str] = []

    health_status, rule_health_status = _get_all_health_status(client, listener_id)

    if not health_status:
        lines.append("  (No health status data available)")
        return lines

    for lhs in health_status:
        for sg_info in lhs.server_group_infos or []:
            sg_id = sg_info.server_group_id or "?"

            # health_check_enabled is a str: "on" means enabled
            hc_enabled = sg_info.health_check_enabled
            if hc_enabled is not None and hc_enabled != "on":
                lines.append(f"  ServerGroup: {sg_id}")
                lines.append("    ⏸️  Health check disabled")
                continue

            non_normal = sg_info.non_normal_servers or []
            if not non_normal:
                lines.append(f"  ServerGroup: {sg_id}")
                lines.append("    ✅ All backends healthy")
            else:
                lines.append(f"  ServerGroup: {sg_id}")
                lines.append(f"    ⚠️  Unhealthy backends detected ({len(non_normal)}):")
                lines.extend(_format_unhealthy_servers(non_normal))

    # Rule-level health status
    if rule_health_status:
        lines.append("")
        lines.append("  Rule-level health status:")
        for rhs in rule_health_status:
            rule_id = rhs.rule_id or "?"
            for sg_info in rhs.server_group_infos or []:
                sg_id = sg_info.server_group_id or "?"
                hc_enabled = sg_info.health_check_enabled
                if hc_enabled is not None and hc_enabled != "on":
                    lines.append(f"    Rule {rule_id} → ServerGroup {sg_id}: health check disabled")
                    continue
                non_normal = sg_info.non_normal_servers or []
                if not non_normal:
                    lines.append(f"    Rule {rule_id} → ServerGroup {sg_id}: all healthy")
                else:
                    lines.append(f"    Rule {rule_id} → ServerGroup {sg_id}: {len(non_normal)} unhealthy")
                    lines.extend(_format_unhealthy_servers(non_normal, indent="      "))

    return lines


def main() -> int:
    parser = argparse.ArgumentParser(description="Check ALB listener health status")
    parser.add_argument("--region", required=True, help="Region ID, e.g. cn-hangzhou")
    parser.add_argument("--lb-id", required=True, help="LoadBalancer ID, e.g. alb-xxx")
    parser.add_argument("--listener-id", help="Specific listener ID. If omitted, checks all listeners.")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    client = create_client(args.region)

    # Determine which listeners to check
    if args.listener_id:
        listener_ids = [args.listener_id]
    else:
        listeners = list_all_listeners(client, args.lb_id)
        listener_ids = [ls.listener_id for ls in listeners if ls.listener_id]

    if args.json:
        # JSON mode: collect raw health status data for all listeners
        result = []
        for lid in listener_ids:
            health_status, rule_health_status = _get_all_health_status(client, lid)
            entry = {
                "listener_id": lid,
                "listener_health_status": [s.to_map() for s in health_status],
                "rule_health_status": [r.to_map() for r in rule_health_status],
            }
            result.append(entry)
        output = json.dumps(result, indent=2, ensure_ascii=False, default=str)
    else:
        lines: list[str] = []
        if args.listener_id:
            lines.append(f"Listener: {args.listener_id}")
            lines.extend(check_listener_health(client, args.listener_id))
        else:
            if not listener_ids:
                lines.append(f"No listeners found for ALB {args.lb_id}")
            else:
                lines.append(f"Health check for ALB: {args.lb_id} ({len(listener_ids)} listener(s))")
                lines.append("")
                for ls in listeners:
                    lid = ls.listener_id or "?"
                    proto = ls.listener_protocol or "?"
                    port = ls.listener_port or "?"
                    lstatus = ls.listener_status or "?"
                    lines.append(f"Listener: {lid} ({proto}:{port}) [{lstatus}]")
                    lines.extend(check_listener_health(client, lid))
                    lines.append("")
        output = "\n".join(lines).rstrip()
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output + "\n")
        print(f"Output written to {args.output}")
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Create ALB listener with support for HTTP/HTTPS/QUIC and various default actions."""

from __future__ import annotations

import argparse
import json
import os
from typing import Any

from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_alb20200616 import models as alb_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Alb20200616Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"alb.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not ak or not sk:
        raise RuntimeError("ALICLOUD_ACCESS_KEY_ID and ALICLOUD_ACCESS_KEY_SECRET must be set")
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token
    return Alb20200616Client(config)


def build_default_action(args: argparse.Namespace) -> alb_models.CreateListenerRequestDefaultActions:
    """Build default action from CLI arguments."""
    action_type = args.action_type

    if action_type == "ForwardGroup":
        if not args.forward_server_groups:
            raise ValueError("--forward-server-groups required for ForwardGroup action")
        tuples = [
            alb_models.CreateListenerRequestDefaultActionsForwardGroupConfigServerGroupTuples(
                server_group_id=sg_id,
            )
            for sg_id in args.forward_server_groups
        ]
        return alb_models.CreateListenerRequestDefaultActions(
            type="ForwardGroup",
            forward_group_config=alb_models.CreateListenerRequestDefaultActionsForwardGroupConfig(
                server_group_tuples=tuples,
            ),
        )

    elif action_type == "Redirect":
        if not args.redirect_protocol and not args.redirect_port:
            raise ValueError("--redirect-protocol or --redirect-port required for Redirect action")
        return alb_models.CreateListenerRequestDefaultActions(
            type="Redirect",
            redirect_config=alb_models.CreateListenerRequestDefaultActionsRedirectConfig(
                protocol=args.redirect_protocol,
                port=str(args.redirect_port) if args.redirect_port else None,
                http_redirect_code=str(args.redirect_code) if args.redirect_code else "301",
            ),
        )

    elif action_type == "FixedResponse":
        if not args.fixed_response_code:
            raise ValueError("--fixed-response-code required for FixedResponse action")
        return alb_models.CreateListenerRequestDefaultActions(
            type="FixedResponse",
            fixed_response_config=alb_models.CreateListenerRequestDefaultActionsFixedResponseConfig(
                http_code=str(args.fixed_response_code),
                content=args.fixed_response_content or "",
                content_type=args.fixed_response_content_type or "text/plain",
            ),
        )

    else:
        raise ValueError(f"Unsupported action type: {action_type}")


def build_certificates(args: argparse.Namespace) -> list | None:
    """Build certificate list for HTTPS/QUIC listeners."""
    if args.protocol not in ("HTTPS", "QUIC"):
        return None
    if not args.certificate_ids:
        raise ValueError(f"--certificate-ids required for {args.protocol} listener")
    return [
        alb_models.CreateListenerRequestCertificates(certificate_id=cert_id)
        for cert_id in args.certificate_ids
    ]


def create_listener(client: Alb20200616Client, args: argparse.Namespace) -> dict[str, Any]:
    """Create listener and return result."""
    action = build_default_action(args)
    certificates = build_certificates(args)

    req = alb_models.CreateListenerRequest(
        load_balancer_id=args.load_balancer_id,
        listener_protocol=args.protocol,
        listener_port=args.port,
        default_actions=[action],
        listener_description=args.description,
    )

    if certificates:
        req.certificates = certificates

    if args.security_policy_id:
        req.security_policy_id = args.security_policy_id

    if args.http2_enabled is not None and args.protocol == "HTTPS":
        req.http_2enabled = args.http2_enabled

    if args.idle_timeout:
        req.idle_timeout = args.idle_timeout

    if args.request_timeout:
        req.request_timeout = args.request_timeout

    resp = client.create_listener(req)
    return {
        "listener_id": resp.body.listener_id,
        "job_id": resp.body.job_id,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Create ALB listener",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # HTTP listener forwarding to server group
  python create_listener.py --region cn-hangzhou --lb-id alb-xxx \\
    --protocol HTTP --port 80 --action-type ForwardGroup \\
    --forward-server-groups sgp-xxx

  # HTTPS listener with certificate
  python create_listener.py --region cn-hangzhou --lb-id alb-xxx \\
    --protocol HTTPS --port 443 --action-type ForwardGroup \\
    --forward-server-groups sgp-xxx --certificate-ids cert-xxx

  # HTTP to HTTPS redirect
  python create_listener.py --region cn-hangzhou --lb-id alb-xxx \\
    --protocol HTTP --port 80 --action-type Redirect \\
    --redirect-protocol HTTPS --redirect-port 443

  # QUIC listener
  python create_listener.py --region cn-hangzhou --lb-id alb-xxx \\
    --protocol QUIC --port 443 --forward-server-groups sgp-xxx \\
    --certificate-ids cert-xxx
        """,
    )

    # Required
    parser.add_argument("--region", required=True, help="Region ID, e.g. cn-hangzhou")
    parser.add_argument("--lb-id", required=True, dest="load_balancer_id", help="Load balancer ID")
    parser.add_argument("--protocol", required=True, choices=["HTTP", "HTTPS", "QUIC"], help="Listener protocol")
    parser.add_argument("--port", required=True, type=int, help="Listener port")

    # Action type
    parser.add_argument("--action-type", required=True, choices=["ForwardGroup", "Redirect", "FixedResponse"], help="Default action type")

    # ForwardGroup action options
    parser.add_argument("--forward-server-groups", nargs="+", help="Server group IDs for forwarding")

    # Redirect action options
    parser.add_argument("--redirect-protocol", choices=["HTTP", "HTTPS"], help="Target protocol for redirect")
    parser.add_argument("--redirect-port", type=int, help="Target port for redirect")
    parser.add_argument("--redirect-code", type=int, default=301, help="HTTP redirect code (default: 301)")

    # FixedResponse action options
    parser.add_argument("--fixed-response-code", type=int, help="HTTP status code for fixed response")
    parser.add_argument("--fixed-response-content", help="Response body content")
    parser.add_argument("--fixed-response-content-type", default="text/plain", help="Content type (default: text/plain)")

    # HTTPS/QUIC options
    parser.add_argument("--certificate-ids", nargs="+", help="Certificate IDs for HTTPS/QUIC")
    parser.add_argument("--security-policy-id", help="TLS security policy ID")
    parser.add_argument("--http2-enabled", type=lambda x: x.lower() == "true", help="Enable HTTP/2 (HTTPS only)")

    # General options
    parser.add_argument("--description", help="Listener description")
    parser.add_argument("--idle-timeout", type=int, help="Idle timeout in seconds")
    parser.add_argument("--request-timeout", type=int, help="Request timeout in seconds")

    # Output
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--dry-run", action="store_true", help="Print request without executing")

    args = parser.parse_args()

    try:
        action = build_default_action(args)
        certificates = build_certificates(args)

        if args.dry_run:
            print("Dry run - would create listener with:")
            print(f"  LoadBalancerId: {args.load_balancer_id}")
            print(f"  Protocol: {args.protocol}")
            print(f"  Port: {args.port}")
            print(f"  Action: {args.action_type}")
            if certificates:
                print(f"  Certificates: {args.certificate_ids}")
            return 0

        client = create_client(args.region)
        result = create_listener(client, args)

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print(f"Listener created: {result['listener_id']}")
            print(f"Job ID: {result['job_id']}")
            print(f"Note: Listener creation is async. Use get_listener_attribute.py to check status.")

        return 0

    except ValueError as e:
        print(f"Error: {e}")
        return 1
    except Exception as e:
        print(f"API Error: {e}")
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Create an ALB (Application Load Balancer) instance.

Examples:
    # Internet-facing ALB in two zones
    python create_load_balancer.py --region cn-hangzhou --name my-alb \\
        --vpc-id vpc-xxx --address-type Internet \\
        --zone cn-hangzhou-h:vsw-aaa --zone cn-hangzhou-i:vsw-bbb

    # Internal ALB
    python create_load_balancer.py --region cn-hangzhou --name my-alb \\
        --vpc-id vpc-xxx --address-type Intranet \\
        --zone cn-hangzhou-h:vsw-aaa --zone cn-hangzhou-i:vsw-bbb

    # Standard edition with deletion protection
    python create_load_balancer.py --region cn-hangzhou --name my-alb \\
        --vpc-id vpc-xxx --address-type Internet --edition Standard \\
        --deletion-protection \\
        --zone cn-hangzhou-h:vsw-aaa --zone cn-hangzhou-i:vsw-bbb

    # Dry run
    python create_load_balancer.py --region cn-hangzhou --name my-alb \\
        --vpc-id vpc-xxx --address-type Internet \\
        --zone cn-hangzhou-h:vsw-aaa --zone cn-hangzhou-i:vsw-bbb \\
        --dry-run
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any

from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_alb20200616 import models as alb_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Alb20200616Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"alb.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not ak or not sk:
        raise RuntimeError("ALICLOUD_ACCESS_KEY_ID and ALICLOUD_ACCESS_KEY_SECRET must be set")
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token
    return Alb20200616Client(config)


def parse_zone(spec: str) -> dict[str, str]:
    """Parse zone spec: zone_id:vswitch_id[:eip_allocation_id]."""
    parts = spec.split(":")
    if len(parts) < 2:
        raise ValueError(
            f"Invalid zone spec '{spec}'. "
            "Format: zone_id:vswitch_id  e.g. cn-hangzhou-h:vsw-xxx"
        )
    result: dict[str, str] = {"zone_id": parts[0], "v_switch_id": parts[1]}
    if len(parts) >= 3 and parts[2]:
        result["allocation_id"] = parts[2]
    return result


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Create an ALB instance",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Zone spec format: zone_id:vswitch_id[:eip_allocation_id]

Examples:
  %(prog)s --region cn-hangzhou --name my-alb \\
      --vpc-id vpc-xxx --address-type Internet \\
      --zone cn-hangzhou-h:vsw-aaa --zone cn-hangzhou-i:vsw-bbb
        """,
    )
    parser.add_argument("--region", required=True, help="Region ID")
    parser.add_argument("--name", required=True, help="Load balancer name")
    parser.add_argument("--vpc-id", required=True, help="VPC ID")
    parser.add_argument(
        "--address-type", required=True, choices=["Internet", "Intranet"],
        help="Network type",
    )
    parser.add_argument(
        "--zone", required=True, action="append", dest="zones",
        help="Zone mapping: zone_id:vswitch_id[:eip_allocation_id] (at least 2 zones)",
    )
    parser.add_argument(
        "--edition", default="Standard",
        choices=["Basic", "Standard", "StandardWithWaf"],
        help="ALB edition (default: Standard)",
    )
    parser.add_argument(
        "--address-ip-version", default="IPv4",
        choices=["IPv4", "DualStack"],
        help="IP version (default: IPv4)",
    )
    parser.add_argument(
        "--pay-type", default="PostPay",
        choices=["PostPay"],
        help="Billing type (default: PostPay)",
    )
    parser.add_argument("--deletion-protection", action="store_true", help="Enable deletion protection")
    parser.add_argument("--resource-group-id", help="Resource group ID")
    parser.add_argument("--dry-run", action="store_true", help="Print request without executing")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--output", help="Write output to file")

    args = parser.parse_args()

    try:
        parsed_zones = [parse_zone(z) for z in args.zones]
    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1

    if len(parsed_zones) < 2:
        print("Error: at least 2 zones are required for ALB.", file=sys.stderr)
        return 1

    if args.dry_run:
        request: dict[str, Any] = {
            "load_balancer_name": args.name,
            "vpc_id": args.vpc_id,
            "address_type": args.address_type,
            "load_balancer_edition": args.edition,
            "address_ip_version": args.address_ip_version,
            "pay_type": args.pay_type,
            "deletion_protection_enabled": args.deletion_protection,
            "zone_mappings": parsed_zones,
        }
        output = json.dumps(request, indent=2, ensure_ascii=False)
        if args.output:
            Path(args.output).write_text(output + "\n", encoding="utf-8")
            print(f"Dry-run request written to {args.output}")
        else:
            print("Dry-run mode - would create ALB:")
            print(output)
        return 0

    client = create_client(args.region)

    zone_mappings = [
        alb_models.CreateLoadBalancerRequestZoneMappings(
            zone_id=z["zone_id"],
            v_switch_id=z["v_switch_id"],
            **({"allocation_id": z["allocation_id"]} if "allocation_id" in z else {}),
        )
        for z in parsed_zones
    ]

    req = alb_models.CreateLoadBalancerRequest(
        load_balancer_name=args.name,
        vpc_id=args.vpc_id,
        address_type=args.address_type,
        load_balancer_edition=args.edition,
        address_ip_version=args.address_ip_version,
        load_balancer_billing_config=alb_models.CreateLoadBalancerRequestLoadBalancerBillingConfig(
            pay_type=args.pay_type,
        ),
        zone_mappings=zone_mappings,
        deletion_protection_enabled=args.deletion_protection,
    )

    if args.resource_group_id:
        req.resource_group_id = args.resource_group_id

    try:
        resp = client.create_load_balancer(req)
        result = {
            "load_balancer_id": resp.body.load_balancer_id,
            "request_id": resp.body.request_id,
        }
    except Exception as e:
        print(f"Error creating ALB: {e}", file=sys.stderr)
        return 1

    if args.json:
        output = json.dumps(result, indent=2, ensure_ascii=False)
    else:
        output = (
            f"ALB created successfully:\n"
            f"  Load Balancer ID: {result['load_balancer_id']}\n"
            f"  Note: ALB creation is async. Use get_instance_status.py to check status."
        )

    if args.output:
        Path(args.output).write_text(output + "\n", encoding="utf-8")
        print(f"Output written to {args.output}")
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Create ALB forwarding rule with flexible conditions and actions.

Supports common patterns like blocking HTTP methods, host-based routing,
path-based routing, and fixed responses.

Examples:
    # Block DELETE method with 405 response
    python create_rule.py --region cn-hangzhou --listener-id lsn-xxx \\
        --name "block-delete" --priority 10 \\
        --condition-method DELETE \\
        --action-fixed-response 405 "Method Not Allowed"

    # Host-based routing to server group
    python create_rule.py --region cn-hangzhou --listener-id lsn-xxx \\
        --name "api-route" --priority 20 \\
        --condition-host "api.example.com" \\
        --action-forward-to sgp-xxx

    # Path-based routing with multiple conditions
    python create_rule.py --region cn-hangzhou --listener-id lsn-xxx \\
        --name "api-v1-route" --priority 30 \\
        --condition-host "api.example.com" \\
        --condition-path "/v1/*" \\
        --action-forward-to sgp-xxx

    # HTTP to HTTPS redirect
    python create_rule.py --region cn-hangzhou --listener-id lsn-xxx \\
        --name "force-https" --priority 5 \\
        --action-redirect https 443

    # Dry run - validate without creating
    python create_rule.py --region cn-hangzhou --listener-id lsn-xxx \\
        --name "test" --priority 100 \\
        --condition-method POST \\
        --action-fixed-response 200 "OK" \\
        --dry-run
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any


def create_client(region_id: str):
    """Create ALB client with credentials from environment."""
    from alibabacloud_alb20200616.client import Client as Alb20200616Client
    from alibabacloud_alb20200616 import models as alb_models
    from alibabacloud_tea_openapi import models as open_api_models

    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"alb.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not ak or not sk:
        raise RuntimeError("ALICLOUD_ACCESS_KEY_ID and ALICLOUD_ACCESS_KEY_SECRET must be set")
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token
    return Alb20200616Client(config)


def parse_conditions(args: argparse.Namespace) -> list[dict]:
    """Parse condition arguments into ALB condition structures."""
    conditions: list[dict] = []

    if args.condition_method:
        conditions.append({
            "type": "Method",
            "method_config": {"values": args.condition_method.split(",")},
        })

    if args.condition_host:
        conditions.append({
            "type": "Host",
            "host_config": {"values": args.condition_host.split(",")},
        })

    if args.condition_path:
        conditions.append({
            "type": "Path",
            "path_config": {"values": args.condition_path.split(",")},
        })

    if args.condition_header:
        for header in args.condition_header:
            # Format: "X-Api-Key: value1,value2" or "X-Api-Key=value1,value2"
            if ":" in header:
                key, values = header.split(":", 1)
            elif "=" in header:
                key, values = header.split("=", 1)
            else:
                key, values = header, ""
            conditions.append({
                "type": "Header",
                "header_config": {
                    "key": key.strip(),
                    "values": [v.strip() for v in values.split(",") if v.strip()],
                },
            })

    if args.condition_query:
        for query in args.condition_query:
            # Format: "version=1.0,2.0"
            if "=" in query:
                key, values = query.split("=", 1)
            else:
                key, values = query, ""
            conditions.append({
                "type": "QueryString",
                "query_string_config": {
                    "values": [{"key": key.strip(), "value": v.strip()}
                               for v in values.split(",") if v.strip()],
                },
            })

    if args.condition_cookie:
        for cookie in args.condition_cookie:
            # Format: "session=abc,def"
            if "=" in cookie:
                key, values = cookie.split("=", 1)
            else:
                key, values = cookie, ""
            conditions.append({
                "type": "Cookie",
                "cookie_config": {
                    "values": [{"key": key.strip(), "value": v.strip()}
                               for v in values.split(",") if v.strip()],
                },
            })

    if args.condition_source_ip:
        conditions.append({
            "type": "SourceIp",
            "source_ip_config": {"values": args.condition_source_ip.split(",")},
        })

    return conditions


def parse_actions(args: argparse.Namespace) -> list[dict]:
    """Parse action arguments into ALB action structures."""
    actions: list[dict] = []
    order = 1

    if args.action_forward_to:
        actions.append({
            "type": "ForwardGroup",
            "order": order,
            "forward_group_config": {
                "server_group_tuples": [
                    {"server_group_id": sg_id.strip()}
                    for sg_id in args.action_forward_to.split(",")
                ],
            },
        })
        order += 1

    if args.action_redirect:
        # Format: "https 443" or "https" or "443"
        parts = args.action_redirect.split()
        protocol = None
        port = None
        for part in parts:
            if part.lower() in ("http", "https"):
                protocol = part.upper()
            elif part.isdigit():
                port = part
        actions.append({
            "type": "Redirect",
            "order": order,
            "redirect_config": {
                **({"protocol": protocol} if protocol else {}),
                **({"port": port} if port else {}),
                "http_redirect_code": args.action_redirect_code,
            },
        })
        order += 1

    if args.action_fixed_response:
        code, *content_parts = args.action_fixed_response.split(None, 1)
        content = content_parts[0] if content_parts else ""
        actions.append({
            "type": "FixedResponse",
            "order": order,
            "fixed_response_config": {
                "http_code": code,
                "content": content,
                **({"content_type": args.action_fixed_response_type} if args.action_fixed_response_type else {}),
            },
        })
        order += 1

    if args.action_rewrite:
        # Format: "host=/new-host" or "path=/new-path" or both
        config: dict[str, str] = {}
        for part in args.action_rewrite.split():
            if "=" in part:
                key, value = part.split("=", 1)
                if key == "host":
                    config["host"] = value
                elif key == "path":
                    config["path"] = value
        if config:
            actions.append({
                "type": "Rewrite",
                "order": order,
                "rewrite_config": config,
            })
            order += 1

    if args.action_insert_header:
        for header in args.action_insert_header:
            # Format: "X-Custom-Header: value" or "X-Custom-Header=value"
            if ":" in header:
                key, value = header.split(":", 1)
            elif "=" in header:
                key, value = header.split("=", 1)
            else:
                key, value = header, ""
            actions.append({
                "type": "InsertHeader",
                "order": order,
                "insert_header_config": {
                    "key": key.strip(),
                    "value": value.strip(),
                },
            })
            order += 1

    return actions


def build_request(args: argparse.Namespace) -> dict[str, Any]:
    """Build the CreateRule request structure."""
    conditions = parse_conditions(args)
    actions = parse_actions(args)

    if not conditions and not args.allow_no_conditions:
        raise ValueError(
            "No conditions specified. Use --condition-* options or --allow-no-conditions "
            "to create a catch-all rule."
        )

    if not actions:
        raise ValueError("No actions specified. Use --action-* options.")

    request: dict[str, Any] = {
        "listener_id": args.listener_id,
        "rule_name": args.name,
        "priority": args.priority,
        "rule_conditions": conditions,
        "rule_actions": actions,
    }

    return request


def execute_create(client, request: dict[str, Any]) -> dict[str, Any]:
    """Execute the CreateRule API call."""
    from alibabacloud_alb20200616 import models as alb_models

    # Convert dict to model objects
    conditions = []
    for c in request["rule_conditions"]:
        cond = alb_models.CreateRuleRequestRuleConditions(type=c["type"])
        if "host_config" in c:
            cond.host_config = alb_models.CreateRuleRequestRuleConditionsHostConfig(**c["host_config"])
        elif "path_config" in c:
            cond.path_config = alb_models.CreateRuleRequestRuleConditionsPathConfig(**c["path_config"])
        elif "method_config" in c:
            cond.method_config = alb_models.CreateRuleRequestRuleConditionsMethodConfig(**c["method_config"])
        elif "header_config" in c:
            cond.header_config = alb_models.CreateRuleRequestRuleConditionsHeaderConfig(**c["header_config"])
        elif "query_string_config" in c:
            values = [alb_models.CreateRuleRequestRuleConditionsQueryStringConfigValues(**v)
                      for v in c["query_string_config"]["values"]]
            cond.query_string_config = alb_models.CreateRuleRequestRuleConditionsQueryStringConfig(values=values)
        elif "cookie_config" in c:
            values = [alb_models.CreateRuleRequestRuleConditionsCookieConfigValues(**v)
                      for v in c["cookie_config"]["values"]]
            cond.cookie_config = alb_models.CreateRuleRequestRuleConditionsCookieConfig(values=values)
        elif "source_ip_config" in c:
            cond.source_ip_config = alb_models.CreateRuleRequestRuleConditionsSourceIpConfig(**c["source_ip_config"])
        conditions.append(cond)

    actions = []
    for a in request["rule_actions"]:
        act = alb_models.CreateRuleRequestRuleActions(type=a["type"], order=a["order"])
        if "forward_group_config" in a:
            tuples = [alb_models.CreateRuleRequestRuleActionsForwardGroupConfigServerGroupTuples(**t)
                      for t in a["forward_group_config"]["server_group_tuples"]]
            act.forward_group_config = alb_models.CreateRuleRequestRuleActionsForwardGroupConfig(
                server_group_tuples=tuples
            )
        elif "redirect_config" in a:
            act.redirect_config = alb_models.CreateRuleRequestRuleActionsRedirectConfig(**a["redirect_config"])
        elif "fixed_response_config" in a:
            act.fixed_response_config = alb_models.CreateRuleRequestRuleActionsFixedResponseConfig(
                **a["fixed_response_config"]
            )
        elif "rewrite_config" in a:
            act.rewrite_config = alb_models.CreateRuleRequestRuleActionsRewriteConfig(**a["rewrite_config"])
        elif "insert_header_config" in a:
            act.insert_header_config = alb_models.CreateRuleRequestRuleActionsInsertHeaderConfig(
                **a["insert_header_config"]
            )
        actions.append(act)

    req = alb_models.CreateRuleRequest(
        listener_id=request["listener_id"],
        rule_name=request["rule_name"],
        priority=request["priority"],
        rule_conditions=conditions,
        rule_actions=actions,
    )

    resp = client.create_rule(req)
    return {
        "rule_id": resp.body.rule_id,
        "job_id": resp.body.job_id,
        "request_id": resp.body.request_id,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Create ALB forwarding rule",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Block DELETE method
  %(prog)s --region cn-hangzhou --listener-id lsn-xxx \\
      --name "block-delete" --priority 10 \\
      --condition-method DELETE \\
      --action-fixed-response 405 "Method Not Allowed"

  # Route by host
  %(prog)s --region cn-hangzhou --listener-id lsn-xxx \\
      --name "api-route" --priority 20 \\
      --condition-host "api.example.com" \\
      --action-forward-to sgp-xxx

  # HTTP to HTTPS redirect
  %(prog)s --region cn-hangzhou --listener-id lsn-xxx \\
      --name "force-https" --priority 5 \\
      --action-redirect https 443
        """,
    )
    parser.add_argument("--region", required=True, help="Region ID, e.g. cn-hangzhou")
    parser.add_argument("--listener-id", required=True, help="Listener ID (lsn-xxx)")
    parser.add_argument("--name", required=True, help="Rule name")
    parser.add_argument("--priority", type=int, required=True, help="Rule priority (1-10000, lower = higher priority)")

    # Condition options
    cond_group = parser.add_argument_group("Conditions (at least one required)")
    cond_group.add_argument("--condition-method", help="HTTP method(s), comma-separated: GET,POST,DELETE")
    cond_group.add_argument("--condition-host", help="Host(s), comma-separated: api.example.com,*.example.com")
    cond_group.add_argument("--condition-path", help="Path(s), comma-separated: /api/*,/v1/*")
    cond_group.add_argument("--condition-header", action="append", help="Header condition: 'X-Api-Key: value1,value2'")
    cond_group.add_argument("--condition-query", action="append", help="Query string condition: 'version=1.0,2.0'")
    cond_group.add_argument("--condition-cookie", action="append", help="Cookie condition: 'session=abc,def'")
    cond_group.add_argument("--condition-source-ip", help="Source IP(s), comma-separated: 10.0.0.0/8,192.168.1.1")
    cond_group.add_argument("--allow-no-conditions", action="store_true", help="Allow creating a catch-all rule")

    # Action options
    act_group = parser.add_argument_group("Actions (at least one required)")
    act_group.add_argument("--action-forward-to", help="Forward to server group ID(s), comma-separated")
    act_group.add_argument("--action-redirect", help="Redirect: 'https 443' or 'https' or '443'")
    act_group.add_argument("--action-redirect-code", default="301", help="Redirect HTTP code (default: 301)")
    act_group.add_argument("--action-fixed-response", help="Fixed response: '404 Not Found'")
    act_group.add_argument("--action-fixed-response-type", help="Content-Type for fixed response")
    act_group.add_argument("--action-rewrite", help="Rewrite: 'host=newhost' or 'path=/newpath' or 'host=newhost path=/new'")
    act_group.add_argument("--action-insert-header", action="append", help="Insert header: 'X-Custom: value'")

    # Other options
    parser.add_argument("--dry-run", action="store_true", help="Validate and print request without creating")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--output", help="Write output to file")

    args = parser.parse_args()

    try:
        request = build_request(args)
    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1

    if args.dry_run:
        output = json.dumps(request, indent=2, ensure_ascii=False)
        if args.output:
            Path(args.output).write_text(output + "\n", encoding="utf-8")
            print(f"Dry-run request written to {args.output}")
        else:
            print("Dry-run mode - request structure:")
            print(output)
        return 0

    # Execute the create
    client = create_client(args.region)
    try:
        result = execute_create(client, request)
    except Exception as e:
        print(f"Error creating rule: {e}", file=sys.stderr)
        return 1

    # Output
    if args.json:
        output = json.dumps(result, indent=2, ensure_ascii=False)
    else:
        output = f"Rule created successfully:\n  Rule ID: {result['rule_id']}\n  Job ID: {result['job_id']}"

    if args.output:
        Path(args.output).write_text(output + "\n", encoding="utf-8")
        print(f"Output written to {args.output}")
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Create ALB server group with health check and sticky session configuration."""

from __future__ import annotations

import argparse
import json
import os
from typing import Any

from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_alb20200616 import models as alb_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Alb20200616Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"alb.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not ak or not sk:
        raise RuntimeError("ALICLOUD_ACCESS_KEY_ID and ALICLOUD_ACCESS_KEY_SECRET must be set")
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token
    return Alb20200616Client(config)


def build_health_check_config(args: argparse.Namespace) -> alb_models.CreateServerGroupRequestHealthCheckConfig | None:
    """Build health check config from CLI arguments."""
    if args.health_check_disabled:
        return alb_models.CreateServerGroupRequestHealthCheckConfig(
            health_check_enabled=False,
        )

    return alb_models.CreateServerGroupRequestHealthCheckConfig(
        health_check_enabled=True,
        health_check_protocol=args.health_check_protocol or "HTTP",
        health_check_method=args.health_check_method or "HEAD",
        health_check_path=args.health_check_path or "/",
        health_check_host=args.health_check_host,
        health_check_http_version=args.health_check_http_version or "HTTP1.1",
        health_check_interval=args.health_check_interval or 5,
        health_check_timeout=args.health_check_timeout or 3,
        healthy_threshold=args.healthy_threshold or 3,
        unhealthy_threshold=args.unhealthy_threshold or 3,
        health_check_codes=args.health_check_codes or ["http_2xx"],
    )


def build_sticky_session_config(args: argparse.Namespace) -> alb_models.CreateServerGroupRequestStickySessionConfig:
    """Build sticky session config from CLI arguments.

    ALB API requires StickySessionConfig to be present even when disabled.
    """
    if not args.sticky_session_enabled:
        return alb_models.CreateServerGroupRequestStickySessionConfig(
            sticky_session_enabled=False,
        )

    return alb_models.CreateServerGroupRequestStickySessionConfig(
        sticky_session_enabled=True,
        sticky_session_type=args.sticky_session_type or "Server",
        cookie=args.sticky_session_cookie,
        cookie_timeout=args.sticky_session_timeout,
    )


def create_server_group(client: Alb20200616Client, args: argparse.Namespace) -> dict[str, Any]:
    """Create server group and return result."""
    health_check_config = build_health_check_config(args)
    sticky_session_config = build_sticky_session_config(args)

    req = alb_models.CreateServerGroupRequest(
        server_group_name=args.name,
        vpc_id=args.vpc_id,
        protocol=args.protocol or "HTTP",
        scheduler=args.scheduler or "Wrr",
        server_group_type=args.server_group_type or "Instance",
    )

    if health_check_config:
        req.health_check_config = health_check_config

    # ALB API requires StickySessionConfig to always be present
    req.sticky_session_config = sticky_session_config

    if args.connection_drain_timeout is not None:
        req.connection_drain_config = alb_models.CreateServerGroupRequestConnectionDrainConfig(
            connection_drain_enabled=args.connection_drain_timeout > 0,
            connection_drain_timeout=max(0, args.connection_drain_timeout),
        )

    resp = client.create_server_group(req)
    return {
        "server_group_id": resp.body.server_group_id,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Create ALB server group",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Basic HTTP server group
  python create_server_group.py --region cn-hangzhou --name my-sg \\
    --vpc-id vpc-xxx --protocol HTTP

  # With health check customization
  python create_server_group.py --region cn-hangzhou --name my-sg \\
    --vpc-id vpc-xxx --protocol HTTP \\
    --health-check-path /health --health-check-interval 10

  # With sticky sessions
  python create_server_group.py --region cn-hangzhou --name my-sg \\
    --vpc-id vpc-xxx --protocol HTTP \\
    --sticky-session-enabled --sticky-session-type Server --sticky-session-cookie SERVERID

  # Disable health check (not recommended for production)
  python create_server_group.py --region cn-hangzhou --name my-sg \\
    --vpc-id vpc-xxx --health-check-disabled
        """,
    )

    # Required
    parser.add_argument("--region", required=True, help="Region ID, e.g. cn-hangzhou")
    parser.add_argument("--name", required=True, help="Server group name")
    parser.add_argument("--vpc-id", required=True, help="VPC ID")

    # Basic config
    parser.add_argument("--protocol", choices=["HTTP", "HTTPS", "GRPC"], default="HTTP", help="Protocol (default: HTTP)")
    parser.add_argument("--scheduler", choices=["Wrr", "Wlc", "Sch", "Uch"], default="Wrr", help="Scheduling algorithm (default: Wrr)")
    parser.add_argument("--server-group-type", choices=["Instance", "Ip"], default="Instance", help="Server group type (default: Instance)")

    # Health check options
    parser.add_argument("--health-check-disabled", action="store_true", help="Disable health check")
    parser.add_argument("--health-check-protocol", choices=["HTTP", "HTTPS", "TCP"], default="HTTP", help="Health check protocol")
    parser.add_argument("--health-check-method", choices=["HEAD", "GET"], default="HEAD", help="Health check method")
    parser.add_argument("--health-check-path", default="/", help="Health check path (default: /)")
    parser.add_argument("--health-check-host", help="Health check host header")
    parser.add_argument("--health-check-http-version", choices=["HTTP1.0", "HTTP1.1"], default="HTTP1.1", help="Health check HTTP version (default: HTTP1.1)")
    parser.add_argument("--health-check-interval", type=int, default=5, help="Check interval in seconds (default: 5)")
    parser.add_argument("--health-check-timeout", type=int, default=3, help="Check timeout in seconds (default: 3)")
    parser.add_argument("--healthy-threshold", type=int, default=3, help="Healthy threshold (default: 3)")
    parser.add_argument("--unhealthy-threshold", type=int, default=3, help="Unhealthy threshold (default: 3)")
    parser.add_argument("--health-check-codes", nargs="+", default=["http_2xx"], help="Expected status codes (default: http_2xx)")

    # Sticky session options
    parser.add_argument("--sticky-session-enabled", action="store_true", help="Enable sticky sessions")
    parser.add_argument("--sticky-session-type", choices=["Server", "Insert"], help="Sticky session type")
    parser.add_argument("--sticky-session-cookie", help="Cookie name for Server type")
    parser.add_argument("--sticky-session-timeout", type=int, help="Cookie timeout for Insert type (seconds)")

    # Connection draining
    parser.add_argument("--connection-drain-timeout", type=int, default=300, help="Connection drain timeout in seconds (default: 300, 0 to disable)")

    # Output
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--dry-run", action="store_true", help="Print request without executing")

    args = parser.parse_args()

    try:
        if args.dry_run:
            print("Dry run - would create server group with:")
            print(f"  Name: {args.name}")
            print(f"  VPC: {args.vpc_id}")
            print(f"  Protocol: {args.protocol}")
            print(f"  Scheduler: {args.scheduler}")
            print(f"  Health check: {'disabled' if args.health_check_disabled else 'enabled'}")
            if args.sticky_session_enabled:
                print(f"  Sticky session: enabled (type={args.sticky_session_type})")
            return 0

        client = create_client(args.region)
        result = create_server_group(client, args)

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print(f"Server group created: {result['server_group_id']}")

        return 0

    except Exception as e:
        print(f"Error: {e}")
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Delete an ALB listener.

WARNING: This is a destructive operation. The listener and its associated
forwarding rules will be deleted.

Examples:
    python delete_listener.py --region cn-hangzhou --listener-id lsn-xxx
    python delete_listener.py --region cn-hangzhou --listener-id lsn-xxx --json
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_alb20200616 import models as alb_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> Alb20200616Client:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"alb.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not ak or not sk:
        raise RuntimeError("ALICLOUD_ACCESS_KEY_ID and ALICLOUD_ACCESS_KEY_SECRET must be set")
    config.access_key_id = ak
    config.access_key_secret = sk
    if token:
        config.security_token = token
    return Alb20200616Client(config)


def main() -> int:
    parser = argparse.ArgumentParser(description="Delete an ALB listener")
    parser.add_argument("--region", required=True, help="Region ID")
    parser.add_argument("--listener-id", required=True, help="Listener ID (lsn-xxx)")
    parser.add_argument("--yes", action="store_true", help="Skip confirmation prompt")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--output", help="Write output to file")

    args = parser.parse_args()

    if not args.yes:
        confirm = input(f"Delete listener {args.listener_id}? This will also delete associated rules. [y/N] ")
        if confirm.lower() not in ("y", "yes"):
            print("Cancelled.")
            return 0

    client = create_client(args.region)
    try:
        resp = client.delete_listener(alb_models.DeleteListenerRequest(
            listener_id=args.listener_id,
        ))
        result = {"job_id": resp.body.job_id, "request_id": resp.body.request_id}
    except Exception as e:
        print(f"Error deleting listener: {e}", file=sys.stderr)
        return 1

    if args.json:
        output = json.dumps(result, indent=2, ensure_ascii=False)
    else:
        output = f"Listener {args.listener_id} deleted.\n  Job ID: {result['job_id']}"

    if args.output:
        Path(args.output).write_text(output + "\n", encoding="utf-8")
        print(f"Output written to {args.output}")
    else:
        print(output)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())