mirror of
https://github.com/agent-substrate/substrate.git
synced 2026-11-12 10:35:06 +08:00
router: record target IP in request lo (#41)
Before <img width="2560" height="447" alt="2026-05-21_08-00-39" src="https://github.com/user-attachments/assets/3242ec19-9783-4727-80b3-24136ff5a3fc" /> After <img width="2600" height="807" alt="2026-05-21_08-00-57" src="https://github.com/user-attachments/assets/c6892106-a9de-428a-b41d-6fdcbc7c39bd" /> > It's a good idea to open an issue first for discussion. - [x] Tests pass - [x] Appropriate changes to documentation are included in the PR
This commit is contained in:
@@ -83,7 +83,7 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer
|
||||
switch reqType := req.Request.(type) {
|
||||
case *extprocv3.ProcessingRequest_RequestHeaders:
|
||||
start := time.Now()
|
||||
hResponse, rqm, err := s.handleRequestHeaders(stream.Context(), reqType.RequestHeaders)
|
||||
hResponse, rqm, target, err := s.handleRequestHeaders(stream.Context(), reqType.RequestHeaders)
|
||||
if err != nil {
|
||||
slog.ErrorContext(stream.Context(), "Error during ext_proc RequestHeaders processing", slog.String("err", err.Error()))
|
||||
var reqErr *reqError
|
||||
@@ -95,7 +95,7 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer
|
||||
s.recorder.AddRouterRequest(start, time.Since(start), "Error", "-", rqm)
|
||||
} else {
|
||||
resp.Response = &extprocv3.ProcessingResponse_RequestHeaders{RequestHeaders: hResponse}
|
||||
s.recorder.AddRouterRequest(start, time.Since(start), "Route ok", "-", rqm)
|
||||
s.recorder.AddRouterRequest(start, time.Since(start), "Route ok", target, rqm)
|
||||
}
|
||||
|
||||
default:
|
||||
@@ -118,14 +118,14 @@ func (s *ExtProcServer) Process(stream extprocv3.ExternalProcessor_ProcessServer
|
||||
func (s *ExtProcServer) handleRequestHeaders(
|
||||
ctx context.Context,
|
||||
reqHeaders *extprocv3.HttpHeaders,
|
||||
) (*extprocv3.HeadersResponse, *requestMetadata, error) {
|
||||
) (*extprocv3.HeadersResponse, *requestMetadata, string, error) {
|
||||
metadata := newRequestMetadata(reqHeaders.Headers.GetHeaders())
|
||||
slog.InfoContext(ctx, "Request", slog.String("metadata", metadata.String()))
|
||||
|
||||
actorID, err := parseActorID(metadata.host)
|
||||
if err != nil {
|
||||
// Host is invalid, respond with 404.
|
||||
return nil, metadata, notFoundErr
|
||||
return nil, metadata, "", notFoundErr
|
||||
}
|
||||
|
||||
slog.InfoContext(ctx, "ResumeActor", slog.String("actorID", actorID))
|
||||
@@ -137,12 +137,12 @@ func (s *ExtProcServer) handleRequestHeaders(
|
||||
slog.Any("err", err))
|
||||
|
||||
if err != nil {
|
||||
return nil, metadata, fmt.Errorf("error resuming actor %s: %w", actorID, err)
|
||||
return nil, metadata, "", fmt.Errorf("error resuming actor %s: %w", actorID, err)
|
||||
}
|
||||
|
||||
workerIP := actor.GetAteomPodIp()
|
||||
if ip := net.ParseIP(workerIP); ip == nil {
|
||||
return nil, metadata, fmt.Errorf("actor %q did not have a valid IP %q", actorID, workerIP)
|
||||
return nil, metadata, "", fmt.Errorf("actor %q did not have a valid IP %q", actorID, workerIP)
|
||||
}
|
||||
|
||||
// TODO(bowei) -- handle more than port 80 on the actor.
|
||||
@@ -158,5 +158,5 @@ func (s *ExtProcServer) handleRequestHeaders(
|
||||
Response: &extprocv3.CommonResponse{
|
||||
HeaderMutation: mutation,
|
||||
},
|
||||
}, metadata, nil
|
||||
}, metadata, targetAddr, nil
|
||||
}
|
||||
|
||||
@@ -112,7 +112,7 @@ func TestExtProcHeadersEvaluation(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
res, metadata, err := s.handleRequestHeaders(context.Background(), reqHeaders)
|
||||
res, metadata, target, err := s.handleRequestHeaders(context.Background(), reqHeaders)
|
||||
if tc.expectErr {
|
||||
if err == nil {
|
||||
t.Fatalf("expected error but got nil")
|
||||
@@ -129,6 +129,9 @@ func TestExtProcHeadersEvaluation(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("ext_proc processing error: %v", err)
|
||||
}
|
||||
if target != tc.expectedTarget {
|
||||
t.Errorf("expected target %q, got %q", tc.expectedTarget, target)
|
||||
}
|
||||
|
||||
mutation := res.Response.GetHeaderMutation()
|
||||
if len(mutation.GetSetHeaders()) != 1 {
|
||||
|
||||
Reference in New Issue
Block a user