mirror of
https://github.com/andwati/zitadel.git
synced 2026-11-05 22:02:15 +00:00
fix(login): http 2 memory leak's (#11830)
# Which Problems Are Solved All http 2 based connect-es clients caused memory leaks in our testing using the login v2 application as the host. # How the Problems Are Solved Switch to connect-rpc based http 1.1 transport for the login v2 # Additional Changes Improved the otel interceptor implementation and tests. # Additional Context - connect-rpc is only enabled for v2 API's so we have to make sure to not use any v1 apis in the login but this should be the case anyways. - Discussion #10562 --------- Co-authored-by: Max Peintner <[email protected]>
This commit is contained in:
Ramon
Max Peintner
@@ -1,7 +1,11 @@
|
||||
import { Code, ConnectError, createRouterTransport } from "@connectrpc/connect";
|
||||
import { SpanKind, SpanStatusCode } from "@opentelemetry/api";
|
||||
import { resourceFromAttributes } from "@opentelemetry/resources";
|
||||
import { InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base";
|
||||
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
|
||||
import { createClientFor } from "@zitadel/client";
|
||||
import { SessionService } from "@zitadel/proto/zitadel/session/v2/session_service_pb";
|
||||
import { UserService } from "@zitadel/proto/zitadel/user/v2/user_service_pb";
|
||||
import { afterAll, beforeEach, describe, expect, it } from "vitest";
|
||||
import { otelGrpcInterceptor } from "./otel";
|
||||
|
||||
@@ -18,31 +22,37 @@ describe("otelGrpcInterceptor", () => {
|
||||
beforeEach(() => exporter.reset());
|
||||
|
||||
it("injects traceparent header into gRPC requests", async () => {
|
||||
const capturedHeaders: Record<string, string> = {};
|
||||
let capturedHeaders: Headers | undefined;
|
||||
|
||||
await otelGrpcInterceptor(async () => ({ success: true }))({
|
||||
service: { typeName: "zitadel.session.v2.SessionService" },
|
||||
method: { name: "CreateSession" },
|
||||
url: "https://api.zitadel.example.com/zitadel.session.v2/CreateSession",
|
||||
header: {
|
||||
set: (key: string, value: string) => {
|
||||
capturedHeaders[key.toLowerCase()] = value;
|
||||
},
|
||||
const mockTransport = createRouterTransport(() => {}, {
|
||||
transport: {
|
||||
interceptors: [
|
||||
otelGrpcInterceptor,
|
||||
(next) => (req) => {
|
||||
capturedHeaders = req.header;
|
||||
return next(req);
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect(capturedHeaders["traceparent"]).toMatch(/^[0-9a-f]{2}-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$/);
|
||||
const client = createClientFor(SessionService)(mockTransport);
|
||||
await expect(client.createSession({})).rejects.toThrow();
|
||||
expect(capturedHeaders!.get("traceparent")!).toMatch(/^[0-9a-f]{2}-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$/);
|
||||
});
|
||||
|
||||
it("creates client spans with RPC attributes", async () => {
|
||||
await otelGrpcInterceptor(async () => ({ user: { id: "123" } }))({
|
||||
service: { typeName: "zitadel.user.v2.UserService" },
|
||||
method: { name: "GetUserByID" },
|
||||
url: "https://api.zitadel.example.com/zitadel.user.v2/GetUserByID",
|
||||
header: { set: () => {} },
|
||||
const mockTransport = createRouterTransport(() => {}, {
|
||||
transport: {
|
||||
interceptors: [otelGrpcInterceptor],
|
||||
},
|
||||
});
|
||||
|
||||
const client = createClientFor(UserService)(mockTransport);
|
||||
await expect(client.getUserByID({})).rejects.toThrow();
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0]).toMatchObject({
|
||||
name: "zitadel.user.v2.UserService/GetUserByID",
|
||||
@@ -51,151 +61,125 @@ describe("otelGrpcInterceptor", () => {
|
||||
"rpc.system": "grpc",
|
||||
"rpc.service": "zitadel.user.v2.UserService",
|
||||
"rpc.method": "GetUserByID",
|
||||
"server.address": "api.zitadel.example.com",
|
||||
"server.address": "in-memory",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("records error spans with gRPC status codes", async () => {
|
||||
const error = Object.assign(new Error("Session not found"), { code: 5 });
|
||||
const mockTransport = createRouterTransport(
|
||||
({ service }) => {
|
||||
service(SessionService, {
|
||||
deleteSession: () => {
|
||||
throw new ConnectError("Session not found", Code.NotFound);
|
||||
},
|
||||
});
|
||||
},
|
||||
{
|
||||
transport: {
|
||||
interceptors: [otelGrpcInterceptor],
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
await expect(
|
||||
otelGrpcInterceptor(async () => {
|
||||
throw error;
|
||||
})({
|
||||
service: { typeName: "zitadel.session.v2.SessionService" },
|
||||
method: { name: "DeleteSession" },
|
||||
url: "https://api.zitadel.example.com",
|
||||
header: { set: () => {} },
|
||||
}),
|
||||
).rejects.toThrow("Session not found");
|
||||
const client = createClientFor(SessionService)(mockTransport);
|
||||
await expect(client.deleteSession({})).rejects.toThrow();
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0]).toMatchObject({
|
||||
name: "zitadel.session.v2.SessionService/DeleteSession",
|
||||
status: { code: SpanStatusCode.ERROR, message: "Session not found" },
|
||||
status: {
|
||||
code: SpanStatusCode.ERROR,
|
||||
// https://github.com/connectrpc/connect-es/blob/beb1864a45786f01c8ecb76c93a08ac14483885a/packages/connect/src/connect-error.ts#L32
|
||||
// code prefix gets added by connect
|
||||
message: "[not_found] Session not found",
|
||||
},
|
||||
attributes: { "rpc.grpc.status_code": 5 },
|
||||
});
|
||||
expect(spans[0].events).toHaveLength(1);
|
||||
expect(spans[0].events[0].name).toBe("exception");
|
||||
});
|
||||
|
||||
it("handles missing service/method gracefully", async () => {
|
||||
await otelGrpcInterceptor(async () => ({ ok: true }))({
|
||||
url: "https://api.zitadel.example.com",
|
||||
header: { set: () => {} },
|
||||
});
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0]).toMatchObject({
|
||||
name: "unknown/unknown",
|
||||
attributes: { "rpc.service": "unknown", "rpc.method": "unknown" },
|
||||
});
|
||||
});
|
||||
|
||||
it("returns the response from the next handler", async () => {
|
||||
const expectedResponse = { data: "test-data", items: [1, 2, 3] };
|
||||
const expectedResponse = {
|
||||
details: {},
|
||||
sessionId: "test-session-id",
|
||||
sessionToken: "test-token",
|
||||
};
|
||||
|
||||
const result = await otelGrpcInterceptor(async () => expectedResponse)({
|
||||
service: { typeName: "TestService" },
|
||||
method: { name: "TestMethod" },
|
||||
url: "https://test.example.com",
|
||||
header: { set: () => {} },
|
||||
});
|
||||
const mockTransport = createRouterTransport(
|
||||
({ service }) => {
|
||||
service(SessionService, {
|
||||
createSession: () => expectedResponse,
|
||||
});
|
||||
},
|
||||
{
|
||||
transport: {
|
||||
interceptors: [otelGrpcInterceptor],
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
expect(result).toEqual(expectedResponse);
|
||||
const client = createClientFor(SessionService)(mockTransport);
|
||||
const result = await client.createSession({});
|
||||
expect(result).toMatchObject(expectedResponse);
|
||||
});
|
||||
|
||||
it("records error spans without gRPC status code", async () => {
|
||||
const error = new Error("Network failure");
|
||||
it("handles in-memory transport URL gracefully", async () => {
|
||||
const mockTransport = createRouterTransport(() => {}, {
|
||||
transport: {
|
||||
interceptors: [otelGrpcInterceptor],
|
||||
},
|
||||
});
|
||||
|
||||
await expect(
|
||||
otelGrpcInterceptor(async () => {
|
||||
throw error;
|
||||
})({
|
||||
service: { typeName: "zitadel.user.v2.UserService" },
|
||||
method: { name: "ListUsers" },
|
||||
url: "https://api.zitadel.example.com",
|
||||
header: { set: () => {} },
|
||||
}),
|
||||
).rejects.toThrow("Network failure");
|
||||
const client = createClientFor(SessionService)(mockTransport);
|
||||
await expect(client.createSession({})).rejects.toThrow();
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0].attributes["server.address"]).toBeDefined();
|
||||
});
|
||||
|
||||
it("sets correct service and method names from protobuf definitions", async () => {
|
||||
const mockTransport = createRouterTransport(() => {}, {
|
||||
transport: {
|
||||
interceptors: [otelGrpcInterceptor],
|
||||
},
|
||||
});
|
||||
|
||||
const client = createClientFor(SessionService)(mockTransport);
|
||||
await expect(client.deleteSession({})).rejects.toThrow();
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0]).toMatchObject({
|
||||
name: "zitadel.user.v2.UserService/ListUsers",
|
||||
status: { code: SpanStatusCode.ERROR, message: "Network failure" },
|
||||
});
|
||||
expect(spans[0].attributes).not.toHaveProperty("rpc.grpc.status_code");
|
||||
});
|
||||
|
||||
it("handles malformed URL gracefully", async () => {
|
||||
await otelGrpcInterceptor(async () => ({ ok: true }))({
|
||||
service: { typeName: "TestService" },
|
||||
method: { name: "TestMethod" },
|
||||
url: "not-a-valid-url",
|
||||
header: { set: () => {} },
|
||||
});
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0].attributes["server.address"]).toBe("unknown");
|
||||
});
|
||||
|
||||
it("handles missing URL gracefully", async () => {
|
||||
await otelGrpcInterceptor(async () => ({ ok: true }))({
|
||||
service: { typeName: "TestService" },
|
||||
method: { name: "TestMethod" },
|
||||
header: { set: () => {} },
|
||||
});
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0].attributes["server.address"]).toBe("unknown");
|
||||
});
|
||||
|
||||
it("handles empty string URL gracefully", async () => {
|
||||
await otelGrpcInterceptor(async () => ({ ok: true }))({
|
||||
service: { typeName: "TestService" },
|
||||
method: { name: "TestMethod" },
|
||||
url: "",
|
||||
header: { set: () => {} },
|
||||
});
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0].attributes["server.address"]).toBe("unknown");
|
||||
});
|
||||
|
||||
it("uses empty strings as-is for service/method names", async () => {
|
||||
await otelGrpcInterceptor(async () => ({ ok: true }))({
|
||||
service: { typeName: "" },
|
||||
method: { name: "" },
|
||||
url: "https://api.example.com",
|
||||
header: { set: () => {} },
|
||||
});
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
expect(spans[0]).toMatchObject({
|
||||
name: "/",
|
||||
attributes: { "rpc.service": "", "rpc.method": "" },
|
||||
name: "zitadel.session.v2.SessionService/DeleteSession",
|
||||
attributes: {
|
||||
"rpc.service": "zitadel.session.v2.SessionService",
|
||||
"rpc.method": "DeleteSession",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("handles non-Error exceptions", async () => {
|
||||
await expect(
|
||||
otelGrpcInterceptor(async () => {
|
||||
throw "string error";
|
||||
})({
|
||||
service: { typeName: "TestService" },
|
||||
method: { name: "TestMethod" },
|
||||
url: "https://api.example.com",
|
||||
header: { set: () => {} },
|
||||
}),
|
||||
).rejects.toBe("string error");
|
||||
const mockTransport = createRouterTransport(
|
||||
({ service }) => {
|
||||
service(SessionService, {
|
||||
createSession: () => {
|
||||
throw "string error";
|
||||
},
|
||||
});
|
||||
},
|
||||
{
|
||||
transport: {
|
||||
interceptors: [otelGrpcInterceptor],
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
const client = createClientFor(SessionService)(mockTransport);
|
||||
await expect(client.createSession({})).rejects.toThrow();
|
||||
|
||||
const spans = exporter.getFinishedSpans();
|
||||
expect(spans).toHaveLength(1);
|
||||
|
||||
@@ -19,8 +19,8 @@
|
||||
* ```
|
||||
*/
|
||||
|
||||
import { ConnectError, Interceptor } from "@connectrpc/connect";
|
||||
import { context, propagation, SpanKind, SpanStatusCode, trace } from "@opentelemetry/api";
|
||||
import { GrpcError, Interceptor } from "./types";
|
||||
|
||||
const TRACER_NAME = "zitadel-login-grpc" as const;
|
||||
|
||||
@@ -28,10 +28,7 @@ const TRACER_NAME = "zitadel-login-grpc" as const;
|
||||
* Extracts the hostname from a URL string. Returns "unknown" if the URL is
|
||||
* undefined or cannot be parsed.
|
||||
*/
|
||||
function parseHostname(url: string | undefined): string {
|
||||
if (!url) {
|
||||
return "unknown";
|
||||
}
|
||||
function parseHostname(url: string): string {
|
||||
try {
|
||||
return new URL(url).hostname;
|
||||
} catch {
|
||||
@@ -44,8 +41,8 @@ function parseHostname(url: string | undefined): string {
|
||||
* client span, injects trace context headers for distributed tracing, and
|
||||
* records any errors that occur during the call.
|
||||
*/
|
||||
export function otelGrpcInterceptor(next: Parameters<Interceptor>[0]): ReturnType<Interceptor> {
|
||||
return async function tracedCall(req) {
|
||||
export const otelGrpcInterceptor: Interceptor = (next) =>
|
||||
async function tracedCall(req) {
|
||||
const serviceName: string = req.service?.typeName ?? "unknown";
|
||||
const methodName: string = req.method?.name ?? "unknown";
|
||||
|
||||
@@ -70,12 +67,12 @@ export function otelGrpcInterceptor(next: Parameters<Interceptor>[0]): ReturnTyp
|
||||
const response = await next(req);
|
||||
span.setStatus({ code: SpanStatusCode.OK });
|
||||
return response;
|
||||
} catch (err: unknown) {
|
||||
} catch (err) {
|
||||
const exception = err instanceof Error ? err : new Error(String(err));
|
||||
span.recordException(exception);
|
||||
const grpcError = err as GrpcError;
|
||||
if (grpcError.code !== undefined) {
|
||||
span.setAttribute("rpc.grpc.status_code", grpcError.code);
|
||||
|
||||
if (exception instanceof ConnectError) {
|
||||
span.setAttribute("rpc.grpc.status_code", exception.code);
|
||||
}
|
||||
span.setStatus({
|
||||
code: SpanStatusCode.ERROR,
|
||||
@@ -88,4 +85,3 @@ export function otelGrpcInterceptor(next: Parameters<Interceptor>[0]): ReturnTyp
|
||||
},
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
export interface GrpcRequestHeader {
|
||||
readonly set: (key: string, value: string) => void;
|
||||
}
|
||||
|
||||
export interface GrpcRequest {
|
||||
readonly service?: { readonly typeName: string };
|
||||
readonly method?: { readonly name: string };
|
||||
readonly url?: string;
|
||||
readonly header: GrpcRequestHeader;
|
||||
}
|
||||
|
||||
export interface GrpcError extends Error {
|
||||
readonly code?: number;
|
||||
}
|
||||
|
||||
export type Interceptor<T = unknown> = (next: (req: GrpcRequest) => Promise<T>) => (req: GrpcRequest) => Promise<T>;
|
||||
@@ -529,7 +529,6 @@ export async function sendLoginname(command: SendLoginnameCommand) {
|
||||
|
||||
// user not found, check if IDPs are available when local auth is not allowed
|
||||
if (!effectiveLoginSettings?.allowLocalAuthentication) {
|
||||
console.log("redirecting to IDP (register allowed, password not allowed)");
|
||||
logger.debug("redirecting to IDP (register allowed, password not allowed)");
|
||||
const resp = await redirectUserToIDP(undefined, discoveredOrganization);
|
||||
if (resp) {
|
||||
|
||||
@@ -2,7 +2,6 @@ import crypto from "crypto";
|
||||
import { cookies } from "next/headers";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { getFingerprintIdCookie } from "./fingerprint";
|
||||
import { shouldEnforceMFA } from "./verify-helper";
|
||||
|
||||
import { AuthenticationMethodType } from "@zitadel/proto/zitadel/user/v2/user_service_pb";
|
||||
import {
|
||||
@@ -11,6 +10,7 @@ import {
|
||||
checkMFAFactors,
|
||||
checkPasswordChangeRequired,
|
||||
checkUserVerification,
|
||||
shouldEnforceMFA,
|
||||
} from "./verify-helper";
|
||||
|
||||
// Mock function to create timestamps - following the same pattern as session.test.ts
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createConnectTransport } from "@connectrpc/connect-node";
|
||||
import { Client, create, Duration } from "@zitadel/client";
|
||||
import { createServerTransport as libCreateServerTransport } from "@zitadel/client/node";
|
||||
import { makeReqCtx } from "@zitadel/client/v2";
|
||||
import { IdentityProviderService } from "@zitadel/proto/zitadel/idp/v2/idp_service_pb";
|
||||
import { OrganizationSchema, TextQueryMethod } from "@zitadel/proto/zitadel/object/v2/object_pb";
|
||||
@@ -32,9 +32,10 @@ import {
|
||||
import { getTranslations } from "next-intl/server";
|
||||
import { unstable_cacheLife as cacheLife } from "next/cache";
|
||||
import { getUserAgent } from "./fingerprint";
|
||||
import { otelGrpcInterceptor } from "./grpc/interceptors/otel";
|
||||
import { createLogger } from "./logger";
|
||||
|
||||
import { otelGrpcInterceptor } from "@/lib/grpc/interceptors/otel";
|
||||
import { Interceptor } from "@connectrpc/connect";
|
||||
import { createServiceForHost } from "./service";
|
||||
|
||||
const logger = createLogger("zitadel");
|
||||
@@ -1243,46 +1244,46 @@ export type WithServiceConfig<T = {}> = T & {
|
||||
};
|
||||
|
||||
export function createServerTransport(token: string, serviceConfig: ServiceConfig) {
|
||||
// Build interceptors list - OTEL interceptor is always included for trace propagation
|
||||
const interceptors: any[] = [otelGrpcInterceptor];
|
||||
const authorizationInterceptor: Interceptor = (next) => (req) => {
|
||||
if (!req.header.get("Authorization")) {
|
||||
req.header.set("Authorization", `Bearer ${token}`);
|
||||
}
|
||||
return next(req);
|
||||
};
|
||||
|
||||
// Add custom headers interceptor if needed
|
||||
if (process.env.CUSTOM_REQUEST_HEADERS || serviceConfig.instanceHost || serviceConfig.publicHost) {
|
||||
interceptors.push((next: any) => {
|
||||
return (req: any) => {
|
||||
// Apply headers from serviceConfig
|
||||
if (serviceConfig.instanceHost) {
|
||||
req.header.set("x-zitadel-instance-host", serviceConfig.instanceHost);
|
||||
}
|
||||
if (serviceConfig.publicHost) {
|
||||
req.header.set("x-zitadel-public-host", serviceConfig.publicHost);
|
||||
const headerInterceptor: Interceptor = (next) => (req) => {
|
||||
// Apply headers from serviceConfig
|
||||
if (serviceConfig.instanceHost) {
|
||||
req.header.set("x-zitadel-instance-host", serviceConfig.instanceHost);
|
||||
}
|
||||
if (serviceConfig.publicHost) {
|
||||
req.header.set("x-zitadel-public-host", serviceConfig.publicHost);
|
||||
}
|
||||
|
||||
// Apply headers from CUSTOM_REQUEST_HEADERS environment variable
|
||||
if (process.env.CUSTOM_REQUEST_HEADERS) {
|
||||
process.env.CUSTOM_REQUEST_HEADERS.split(",").forEach((header) => {
|
||||
const kv = header.indexOf(":");
|
||||
if (kv > 0) {
|
||||
const key = header.slice(0, kv).trim();
|
||||
const value = header.slice(kv + 1).trim();
|
||||
if (value) {
|
||||
req.header.set(key, value);
|
||||
} else {
|
||||
req.header.delete(key);
|
||||
}
|
||||
} else {
|
||||
logger.warn("Skipping malformed header", { header });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Apply headers from CUSTOM_REQUEST_HEADERS environment variable
|
||||
if (process.env.CUSTOM_REQUEST_HEADERS) {
|
||||
process.env.CUSTOM_REQUEST_HEADERS.split(",").forEach((header) => {
|
||||
const kv = header.indexOf(":");
|
||||
if (kv > 0) {
|
||||
const key = header.slice(0, kv).trim();
|
||||
const value = header.slice(kv + 1).trim();
|
||||
if (value) {
|
||||
req.header.set(key, value);
|
||||
} else {
|
||||
req.header.delete(key);
|
||||
}
|
||||
} else {
|
||||
logger.warn("Skipping malformed header", { header });
|
||||
}
|
||||
});
|
||||
}
|
||||
return next(req);
|
||||
};
|
||||
|
||||
return next(req);
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
return libCreateServerTransport(token, {
|
||||
return createConnectTransport({
|
||||
httpVersion: "1.1",
|
||||
baseUrl: serviceConfig.baseUrl,
|
||||
interceptors,
|
||||
interceptors: [otelGrpcInterceptor, authorizationInterceptor, headerInterceptor],
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user