a3s-gateway 0.2.5

A3S Gateway - AI-native API gateway with reverse proxy, routing, and agent orchestration
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

//! gRPC protocol handler

use crate::entrypoint::protocol::{ProtocolContext, ResponseBody};
use crate::proxy::grpc::GrpcProxy;
use bytes::Bytes;
use http::Response;
use std::sync::Arc;

pub async fn handle_grpc_dispatch(
    ctx: ProtocolContext,
    grpc_proxy: Arc<GrpcProxy>,
) -> Response<ResponseBody> {
    let backend = ctx.backend.clone();
    let state = ctx.state.clone();
    let route = ctx.route.clone();
    let req_parts = ctx.req_parts;
    let body_bytes = ctx.body_bytes;
    let pipeline = ctx.pipeline;
    let remote_addr = ctx.remote_addr;
    let entrypoint = ctx.entrypoint;
    let access_tracker = ctx.access_tracker;
    let method_str = ctx.method_str;
    let path = ctx.path;
    let host = ctx.host;
    let request_start = ctx.request_start;

    match grpc_proxy
        .forward(
            &backend,
            &req_parts.method,
            &req_parts.uri,
            &req_parts.headers,
            body_bytes,
        )
        .await
    {
        Ok(grpc_resp) => {
            let status_code = grpc_resp.http_status.as_u16();
            let _ = access_tracker.build_entry(
                remote_addr.ip().to_string(),
                method_str,
                path,
                host,
                status_code,
                grpc_resp.body.len() as u64,
                Some(backend.url.clone()),
                Some(route.router_name.clone()),
                Some(entrypoint),
                req_parts
                    .headers
                    .get("user-agent")
                    .and_then(|v| v.to_str().ok())
                    .map(|s| s.to_string()),
            );

            if let Some(phc) = state.passive_health.get(&route.service_name) {
                if phc.is_error_status(status_code) {
                    phc.record_error(&backend, status_code);
                } else {
                    phc.record_success(&backend);
                }
            }

            let mut resp_builder = http::Response::builder().status(grpc_resp.http_status.as_u16());
            for (key, value) in grpc_resp.headers.iter() {
                resp_builder = resp_builder.header(key, value);
            }
            let (mut resp_parts, _) = resp_builder.body(()).unwrap().into_parts();

            if let Err(e) = pipeline.process_response(&mut resp_parts).await {
                tracing::warn!(error = %e, "Response middleware error (gRPC)");
            }

            let mut builder = http::Response::builder().status(resp_parts.status);
            for (key, value) in resp_parts.headers.iter() {
                builder = builder.header(key, value);
            }

            let body_len = grpc_resp.body.len() as u64;
            state.metrics.record_request(status_code, body_len);
            state.metrics.record_router_latency(
                &route.router_name,
                request_start.elapsed().as_micros() as u64,
            );
            if status_code >= 400 {
                state.metrics.record_router_error(&route.router_name);
                state.metrics.record_service_error(&route.service_name);
            }

            builder
                .body(crate::entrypoint::protocol::full_body(grpc_resp.body))
                .unwrap()
        }
        Err(e) => {
            tracing::error!(error = %e, backend = backend.url, "gRPC proxy error");
            let _ = access_tracker.build_entry(
                remote_addr.ip().to_string(),
                method_str,
                path,
                host,
                502,
                0,
                Some(backend.url.clone()),
                Some(route.router_name.clone()),
                Some(entrypoint),
                req_parts
                    .headers
                    .get("user-agent")
                    .and_then(|v| v.to_str().ok())
                    .map(|s| s.to_string()),
            );
            if let Some(phc) = state.passive_health.get(&route.service_name) {
                phc.record_error(&backend, 502);
            }

            state.metrics.record_request(502, 0);
            state.metrics.record_router_latency(
                &route.router_name,
                request_start.elapsed().as_micros() as u64,
            );
            state.metrics.record_router_error(&route.router_name);
            state.metrics.record_service_error(&route.service_name);

            let (mut err_parts, _) = http::Response::builder()
                .status(502)
                .body(())
                .unwrap()
                .into_parts();
            if let Err(mw_err) = pipeline.process_response(&mut err_parts).await {
                tracing::warn!(error = %mw_err, "Response middleware error on gRPC 502");
            }
            let mut builder = http::Response::builder().status(502);
            for (key, value) in err_parts.headers.iter() {
                builder = builder.header(key, value);
            }
            builder
                .body(crate::entrypoint::protocol::full_body(Bytes::from(
                    format!(r#"{{"error":"{}"}}"#, e),
                )))
                .unwrap()
        }
    }
}