feat: add internal listener and upstream transport support

examples/internal-listener-demo/README.md

@@ -0,0 +1,83 @@
+# Internal Listener and Upstream Transport Demo
+
+This demo demonstrates Orion's internal listener and upstream transport functionality for service mesh communication.
+
+## Quick Test
+
+```bash
+./test_internal_config.sh
+```
+
+## Configuration
+
+### Internal Listener
+
+```yaml
+listeners:
+  - name: internal_mesh_listener
+    address:
+      internal:
+        buffer_size_kb: 1024
+    filter_chains:
+      - name: internal_proxy_chain
+        terminal_filter:
+          tcp_proxy:
+            cluster: internal_backend_cluster
+```
+
+### Internal Endpoints
+
+```yaml
+clusters:
+  - name: internal_service_cluster
+    type: STATIC
+    load_assignment:
+      endpoints:
+        - lb_endpoints:
+            - endpoint:
+                address:
+                  internal:
+                    server_listener_name: internal_mesh_listener
+                    endpoint_id: service_a_endpoint_1
+```
+
+### Internal Upstream Transport
+
+```yaml
+transport_socket:
+  internal_upstream:
+    passthrough_metadata:
+      - kind: HOST
+        name: envoy.filters.listener.original_dst
+    transport_socket:
+      raw_buffer: {}
+```
+
+### Bootstrap Extensions
+
+```yaml
+bootstrap_extensions:
+  - internal_listener:
+      buffer_size_kb: 2048
+```
+
+## Usage
+
+```bash
+# Start Orion
+../../target/debug/orion -c orion-config.yaml
+
+# Test endpoints
+curl http://localhost:10000/
+curl http://localhost:10000/service-a
+
+# Monitor
+curl http://localhost:9901/stats
+```
+
+## Features
+
+- Internal listeners for service mesh communication
+- Internal endpoints with server_listener_name references
+- Metadata passthrough via internal upstream transport
+- Global bootstrap extensions configuration

examples/internal-listener-demo/internal_client.rs

@@ -0,0 +1,139 @@
+// Copyright 2025 The kmesh Authors
+//
+//
+// 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.
+//
+//
+
+use std::env;
+use std::io::{Read, Write};
+use std::net::{TcpStream, ToSocketAddrs};
+use std::time::Duration;
+
+fn main() {
+    let args: Vec<String> = env::args().collect();
+
+    if args.len() < 2 {
+        println!("Usage: {} <test_type>", args[0]);
+        println!("Test types: gateway, admin, health");
+        return;
+    }
+
+    match args[1].as_str() {
+        "gateway" => test_gateway(),
+        "admin" => test_admin(),
+        "health" => test_health(),
+        _ => println!("Invalid test type. Use: gateway, admin, health"),
+    }
+}
+
+fn test_gateway() {
+    println!("Testing External Gateway");
+
+    let endpoints = vec![
+        ("localhost:10000", "/", "Main Gateway"),
+        ("localhost:10000", "/service-a", "Service A"),
+    ];
+
+    for (address, path, description) in endpoints {
+        println!("Testing {}: http://{}{}", description, address, path);
+        match test_http_endpoint(address, path) {
+            Ok(response) => println!("  ✅ {}", response.trim()),
+            Err(e) => println!("  ❌ {}", e),
+        }
+    }
+}
+
+fn test_admin() {
+    println!("Testing Admin Interface");
+
+    let endpoints = vec![
+        ("localhost:9901", "/stats", "Statistics"),
+        ("localhost:9901", "/listeners", "Listeners"),
+        ("localhost:9901", "/clusters", "Clusters"),
+    ];
+
+    for (address, path, description) in endpoints {
+        println!("Testing {}: http://{}{}", description, address, path);
+        match test_http_endpoint(address, path) {
+            Ok(response) => {
+                println!("  ✅ {} lines received", response.lines().count());
+                for line in response.lines().take(2) {
+                    if !line.trim().is_empty() {
+                        println!("  {}", line.trim());
+                    }
+                }
+            }
+            Err(e) => println!("  ❌ {}", e),
+        }
+    }
+}
+
+fn test_health() {
+    println!("Testing Health Metrics");
+
+    match test_http_endpoint("localhost:9901", "/stats") {
+        Ok(response) => {
+            let internal_stats: Vec<&str> = response
+                .lines()
+                .filter(|line| line.contains("internal") || line.contains("listener"))
+                .take(5)
+                .collect();
+
+            if internal_stats.is_empty() {
+                println!("  ⚠️  No internal listener stats found");
+            } else {
+                for stat in internal_stats {
+                    println!("  📊 {}", stat.trim());
+                }
+            }
+        }
+        Err(e) => println!("  ❌ {}", e),
+    }
+}
+
+fn test_http_endpoint(address: &str, path: &str) -> Result<String, String> {
+    let addr = address.to_socket_addrs()
+        .map_err(|e| format!("Failed to resolve {}: {}", address, e))?
+        .next()
+        .ok_or_else(|| format!("No addresses found for {}", address))?;
+
+    let mut stream = TcpStream::connect_timeout(&addr, Duration::from_secs(5))
+        .map_err(|e| format!("Connection failed: {}", e))?;
+
+    let request = format!(
+        "GET {} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n\r\n",
+        path, address.split(':').next().unwrap_or("localhost")
+    );
+
+    stream.write_all(request.as_bytes())
+        .map_err(|e| format!("Failed to send request: {}", e))?;
+
+    let mut response = String::new();
+    stream.read_to_string(&mut response)
+        .map_err(|e| format!("Failed to read response: {}", e))?;
+
+    if let Some(body_start) = response.find("\r\n\r\n") {
+        let (headers, body) = response.split_at(body_start + 4);
+
+        if headers.contains("200 OK") {
+            Ok(body.to_string())
+        } else if headers.contains("404") {
+            Err("Endpoint not found (404)".to_string())
+        } else {
+            Err(format!("HTTP error: {}", headers.lines().next().unwrap_or("Unknown")))
+        }
+    } else {
+        Err("Invalid HTTP response".to_string())
+    }
+}

examples/internal-listener-demo/orion-config.yaml

@@ -0,0 +1,74 @@
+runtime:
+  num_cpus: 1
+  num_runtimes: 1
+
+logging:
+  log_level: "debug"
+
+bootstrap_extensions:
+  - internal_listener:
+      buffer_size_kb: 2048
+
+admin:
+  address: "127.0.0.1:9901"
+
+static_resources:
+  listeners:
+    - name: "external_gateway_listener"
+      address: "0.0.0.0:10000"
+      filter_chains:
+        - name: "gateway_filter_chain"
+          terminal_filter:
+            http_connection_manager:
+              route_config:
+                name: "gateway_route"
+                virtual_hosts:
+                  - name: "services"
+                    domains: ["*"]
+                    routes:
+                      - match:
+                          prefix: "/service-a"
+                        route:
+                          cluster: "internal_service_a_cluster"
+                      - match:
+                          prefix: "/"
+                        direct_response:
+                          status: 200
+                          body: "Internal Listener Demo Active"
+
+    - name: "internal_mesh_listener"
+      address:
+        internal:
+          buffer_size_kb: 1024
+      filter_chains:
+        - name: "internal_proxy_chain"
+          terminal_filter:
+            tcp_proxy:
+              cluster: "internal_backend_cluster"
+
+  clusters:
+    - name: "internal_service_a_cluster"
+      type: STATIC
+      transport_socket:
+        internal_upstream:
+          passthrough_metadata:
+            - kind: HOST
+              name: "envoy.filters.listener.original_dst"
+          transport_socket:
+            raw_buffer: {}
+      load_assignment:
+        endpoints:
+          - lb_endpoints:
+              - endpoint:
+                  address:
+                    internal:
+                      server_listener_name: "internal_mesh_listener"
+                      endpoint_id: "service_a_endpoint_1"
+
+    - name: "internal_backend_cluster"
+      type: STATIC
+      load_assignment:
+        endpoints:
+          - lb_endpoints:
+              - endpoint:
+                  address: "127.0.0.1:8080"