diff --git a/build.zig b/build.zig index ecf4e9b..910bf56 100644 --- a/build.zig +++ b/build.zig @@ -45,7 +45,7 @@ pub fn build(b: *std.Build) !void { const test_step = b.step("test", "Run library tests"); test_step.dependOn(&run_main_tests.step); - var exe = b.addExecutable(.{ + const exe = b.addExecutable(.{ .name = "custom", .root_source_file = .{ .path = "src/sample-main.zig" }, .target = target, @@ -72,6 +72,6 @@ pub fn build(b: *std.Build) !void { /// deploy depends on iam and package /// /// iam and package do not have any dependencies -pub fn lambdaBuildOptions(b: *std.build.Builder, exe: *std.Build.Step.Compile) !void { +pub fn lambdaBuildOptions(b: *std.Build, exe: *std.Build.Step.Compile) !void { try @import("lambdabuild.zig").lambdaBuildOptions(b, exe); } diff --git a/lambdabuild.zig b/lambdabuild.zig index d99b4b3..e9a6ed9 100644 --- a/lambdabuild.zig +++ b/lambdabuild.zig @@ -30,7 +30,7 @@ fn addArgs(allocator: std.mem.Allocator, original: []const u8, args: [][]const u /// deploy depends on iam and package /// /// iam and package do not have any dependencies -pub fn lambdaBuildOptions(b: *std.build.Builder, exe: *std.Build.Step.Compile) !void { +pub fn lambdaBuildOptions(b: *std.Build, exe: *std.Build.Step.Compile) !void { // The rest of this function is currently reliant on the use of Linux // system being used to build the lambda function // @@ -143,7 +143,7 @@ pub fn lambdaBuildOptions(b: *std.build.Builder, exe: *std.Build.Step.Compile) ! } const cmd = try std.fmt.allocPrint(b.allocator, ifstatement, .{ function_name_file, - std.fs.path.dirname(exe.root_src.?.path).?, + std.fs.path.dirname(exe.root_module.root_source_file.?.path).?, function_name_file, function_name, not_found_fmt, diff --git a/src/lambda.zig b/src/lambda.zig index 6d7e262..5cc27f3 100644 --- a/src/lambda.zig +++ b/src/lambda.zig @@ -5,7 +5,6 @@ const HandlerFn = *const fn (std.mem.Allocator, []const u8) anyerror![]const u8; const log = std.log.scoped(.lambda); -var empty_headers: std.http.Headers = undefined; var client: ?std.http.Client = null; const prefix = "http://"; @@ -20,7 +19,7 @@ pub fn deinit() void { /// This function is intended to loop infinitely. If not used in this manner, /// make sure to call the deinit() function pub fn run(allocator: ?std.mem.Allocator, event_handler: HandlerFn) !void { // TODO: remove inferred error set? - const lambda_runtime_uri = std.os.getenv("AWS_LAMBDA_RUNTIME_API") orelse test_lambda_runtime_uri.?; + const lambda_runtime_uri = std.posix.getenv("AWS_LAMBDA_RUNTIME_API") orelse test_lambda_runtime_uri.?; // TODO: If this is null, go into single use command line mode var gpa = std.heap.GeneralPurposeAllocator(.{}){}; @@ -37,8 +36,6 @@ pub fn run(allocator: ?std.mem.Allocator, event_handler: HandlerFn) !void { // T // so we'll do this instead if (client != null) return error.MustDeInitBeforeCallingRunAgain; client = .{ .allocator = alloc }; - empty_headers = std.http.Headers.init(alloc); - defer empty_headers.deinit(); log.info("tid {d} (lambda): Bootstrap initializing with event url: {s}", .{ std.Thread.getCurrentId(), url }); while (lambda_remaining_requests == null or lambda_remaining_requests.? > 0) { @@ -58,7 +55,7 @@ pub fn run(allocator: ?std.mem.Allocator, event_handler: HandlerFn) !void { // T var ev = getEvent(req_allocator, uri) catch |err| { // Well, at this point all we can do is shout at the void log.err("Error fetching event details: {}", .{err}); - std.os.exit(1); + std.posix.exit(1); // continue; }; if (ev == null) continue; // this gets logged in getEvent, and without @@ -79,12 +76,12 @@ pub fn run(allocator: ?std.mem.Allocator, event_handler: HandlerFn) !void { // T const Event = struct { allocator: std.mem.Allocator, - event_data: []u8, - request_id: []u8, + event_data: []const u8, + request_id: []const u8, const Self = @This(); - pub fn init(allocator: std.mem.Allocator, event_data: []u8, request_id: []u8) Self { + pub fn init(allocator: std.mem.Allocator, event_data: []const u8, request_id: []const u8) Self { return .{ .allocator = allocator, .event_data = event_data, @@ -127,45 +124,36 @@ const Event = struct { defer self.allocator.free(content_fmt); log.err("Posting to {s}: Data {s}", .{ err_url, content_fmt }); - var err_headers = std.http.Headers.init(self.allocator); - defer err_headers.deinit(); - err_headers.append( - "Lambda-Runtime-Function-Error-Type", - "HandlerReturned", - ) catch |append_err| { - log.err("Error appending error header to post response for request id {s}: {}", .{ self.request_id, append_err }); - std.os.exit(1); - }; // TODO: There is something up with using a shared client in this way // so we're taking a perf hit in favor of stability. In a practical // sense, without making HTTPS connections (lambda environment is // non-ssl), this shouldn't be a big issue var cl = std.http.Client{ .allocator = self.allocator }; defer cl.deinit(); - var req = try cl.request(.POST, err_uri, empty_headers, .{}); - // var req = try client.?.request(.POST, err_uri, empty_headers, .{}); - // defer req.deinit(); - req.transfer_encoding = .{ .content_length = content_fmt.len }; - req.start() catch |post_err| { // Well, at this point all we can do is shout at the void + const res = cl.fetch(.{ + .method = .POST, + .payload = content_fmt, + .location = .{ .uri = err_uri }, + .extra_headers = &.{ + .{ + .name = "Lambda-Runtime-Function-Error-Type", + .value = "HandlerReturned", + }, + }, + }) catch |post_err| { // Well, at this point all we can do is shout at the void log.err("Error posting response (start) for request id {s}: {}", .{ self.request_id, post_err }); - std.os.exit(1); - }; - try req.writeAll(content_fmt); - try req.finish(); - req.wait() catch |post_err| { // Well, at this point all we can do is shout at the void - log.err("Error posting response (wait) for request id {s}: {}", .{ self.request_id, post_err }); - std.os.exit(1); + std.posix.exit(1); }; // TODO: Determine why this post is not returning - if (req.response.status != .ok) { + if (res.status != .ok) { // Documentation says something about "exit immediately". The // Lambda infrastrucutre restarts, so it's unclear if that's necessary. // It seems as though a continue should be fine, and slightly faster - log.err("Get fail: {} {s}", .{ - @intFromEnum(req.response.status), - req.response.status.phrase() orelse "", + log.err("Post fail: {} {s}", .{ + @intFromEnum(res.status), + res.status.phrase() orelse "", }); - std.os.exit(1); + std.posix.exit(1); } log.err("Error reporting post complete", .{}); } @@ -177,24 +165,20 @@ const Event = struct { .{ prefix, lambda_runtime_uri, postfix, self.request_id }, ); defer self.allocator.free(response_url); - const response_uri = try std.Uri.parse(response_url); - var cl = std.http.Client{ .allocator = self.allocator }; - defer cl.deinit(); - var req = try cl.request(.POST, response_uri, empty_headers, .{}); - // var req = try client.?.request(.POST, response_uri, empty_headers, .{}); - defer req.deinit(); const response_content = try std.fmt.allocPrint( self.allocator, "{{ \"content\": \"{s}\" }}", .{event_response}, ); defer self.allocator.free(response_content); - - req.transfer_encoding = .{ .content_length = response_content.len }; - try req.start(); - try req.writeAll(response_content); - try req.finish(); - try req.wait(); + var cl = std.http.Client{ .allocator = self.allocator }; + defer cl.deinit(); + const res = try cl.fetch(.{ + .method = .POST, + .payload = response_content, + .location = .{ .url = response_url }, + }); + if (res.status != .ok) return error.UnexpectedStatusFromPostResponse; } }; @@ -205,37 +189,33 @@ fn getEvent(allocator: std.mem.Allocator, event_data_uri: std.Uri) !?Event { // non-ssl), this shouldn't be a big issue var cl = std.http.Client{ .allocator = allocator }; defer cl.deinit(); - var req = try cl.request(.GET, event_data_uri, empty_headers, .{}); - // var req = try client.?.request(.GET, event_data_uri, empty_headers, .{}); - // defer req.deinit(); - - try req.start(); - try req.finish(); + var response_bytes = std.ArrayList(u8).init(allocator); + defer response_bytes.deinit(); + var server_header_buffer: [16 * 1024]u8 = undefined; // Lambda freezes the process at this line of code. During warm start, // the process will unfreeze and data will be sent in response to client.get - try req.wait(); - if (req.response.status != .ok) { + var res = try cl.fetch(.{ + .server_header_buffer = &server_header_buffer, + .location = .{ .uri = event_data_uri }, + .response_storage = .{ .dynamic = &response_bytes }, + }); + if (res.status != .ok) { // Documentation says something about "exit immediately". The // Lambda infrastrucutre restarts, so it's unclear if that's necessary. // It seems as though a continue should be fine, and slightly faster // std.os.exit(1); log.err("Lambda server event response returned bad error code: {} {s}", .{ - @intFromEnum(req.response.status), - req.response.status.phrase() orelse "", + @intFromEnum(res.status), + res.status.phrase() orelse "", }); return error.EventResponseNotOkResponse; } var request_id: ?[]const u8 = null; - var content_length: ?usize = null; - for (req.response.headers.list.items) |h| { + var header_it = std.http.HeaderIterator.init(server_header_buffer[0..]); + while (header_it.next()) |h| { if (std.ascii.eqlIgnoreCase(h.name, "Lambda-Runtime-Aws-Request-Id")) request_id = h.value; - if (std.ascii.eqlIgnoreCase(h.name, "Content-Length")) { - content_length = std.fmt.parseUnsigned(usize, h.value, 10) catch null; - if (content_length == null) - log.warn("Error parsing content length value: '{s}'", .{h.value}); - } // TODO: XRay uses an environment variable to do its magic. It's our // responsibility to set this, but no zig-native setenv(3)/putenv(3) // exists. I would kind of rather not link in libc for this, @@ -251,26 +231,12 @@ fn getEvent(allocator: std.mem.Allocator, event_data_uri: std.Uri) !?Event { log.err("Could not find request id: skipping request", .{}); return null; } - if (content_length == null) { - // We can't report back an issue because the runtime error reporting endpoint - // uses request id in its path. So the best we can do is log the error and move - // on here. - log.err("No content length provided for event data", .{}); - return null; - } const req_id = request_id.?; log.debug("got lambda request with id {s}", .{req_id}); - var resp_payload = try std.ArrayList(u8).initCapacity(allocator, content_length.?); - defer resp_payload.deinit(); - try resp_payload.resize(content_length.?); - var response_data = try resp_payload.toOwnedSlice(); - errdefer allocator.free(response_data); - _ = try req.readAll(response_data); - return Event.init( allocator, - response_data, + try response_bytes.toOwnedSlice(), try allocator.dupe(u8, req_id), ); } @@ -299,22 +265,18 @@ fn startServer(allocator: std.mem.Allocator) !std.Thread { } fn threadMain(allocator: std.mem.Allocator) !void { - var server = std.http.Server.init(allocator, .{ .reuse_address = true }); - // defer server.deinit(); - const address = try std.net.Address.parseIp("127.0.0.1", 0); - try server.listen(address); - server_port = server.socket.listen_address.in.getPort(); + var http_server = try address.listen(.{ .reuse_address = true }); + server_port = http_server.listen_address.in.getPort(); test_lambda_runtime_uri = try std.fmt.allocPrint(allocator, "127.0.0.1:{d}", .{server_port.?}); log.debug("server listening at {s}", .{test_lambda_runtime_uri.?}); - defer server.deinit(); defer test_lambda_runtime_uri = null; defer server_port = null; log.info("starting server thread, tid {d}", .{std.Thread.getCurrentId()}); var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); - var aa = arena.allocator(); + const aa = arena.allocator(); // We're in control of all requests/responses, so this flag will tell us // when it's time to shut down while (server_remaining_requests > 0) { @@ -329,7 +291,7 @@ fn threadMain(allocator: std.mem.Allocator) !void { // } // } - processRequest(aa, &server) catch |e| { + processRequest(aa, &http_server) catch |e| { log.err("Unexpected error processing request: {any}", .{e}); if (@errorReturnTrace()) |trace| { std.debug.dumpStackTrace(trace.*); @@ -338,56 +300,54 @@ fn threadMain(allocator: std.mem.Allocator) !void { } } -fn processRequest(allocator: std.mem.Allocator, server: *std.http.Server) !void { +fn processRequest(allocator: std.mem.Allocator, server: *std.net.Server) !void { server_ready = true; errdefer server_ready = false; log.debug( "tid {d} (server): server waiting to accept. requests remaining: {d}", .{ std.Thread.getCurrentId(), server_remaining_requests + 1 }, ); - var res = try server.accept(.{ .allocator = allocator }); + var connection = try server.accept(); + defer connection.stream.close(); server_ready = false; - defer res.deinit(); - defer _ = res.reset(); - try res.wait(); // wait for client to send a complete request head - const errstr = "Internal Server Error\n"; - var errbuf: [errstr.len]u8 = undefined; - @memcpy(&errbuf, errstr); - var response_bytes: []const u8 = errbuf[0..]; + var read_buffer: [1024 * 16]u8 = undefined; + var http_server = std.http.Server.init(connection, &read_buffer); - if (res.request.content_length) |l| - server_request_aka_lambda_response = try res.reader().readAllAlloc(allocator, @as(usize, l)); - - log.debug( - "tid {d} (server): {d} bytes read from request", - .{ std.Thread.getCurrentId(), server_request_aka_lambda_response.len }, - ); - - // try response.headers.append("content-type", "text/plain"); - response_bytes = serve(allocator, &res) catch |e| brk: { - res.status = .internal_server_error; - // TODO: more about this particular request - log.err("Unexpected error from executor processing request: {any}", .{e}); - if (@errorReturnTrace()) |trace| { - std.debug.dumpStackTrace(trace.*); - } - break :brk "Unexpected error generating request to lambda"; - }; - res.transfer_encoding = .{ .content_length = response_bytes.len }; - try res.do(); - _ = try res.writer().writeAll(response_bytes); - try res.finish(); - log.debug( - "tid {d} (server): sent response", - .{std.Thread.getCurrentId()}, - ); + if (http_server.state == .ready) { + var request = http_server.receiveHead() catch |err| switch (err) { + error.HttpConnectionClosing => return, + else => { + std.log.err("closing http connection: {s}", .{@errorName(err)}); + std.log.debug("Error occurred from this request: \n{s}", .{read_buffer[0..http_server.read_buffer_len]}); + return; + }, + }; + server_request_aka_lambda_response = try (try request.reader()).readAllAlloc(allocator, std.math.maxInt(usize)); + var respond_options = std.http.Server.Request.RespondOptions{}; + const response_bytes = serve(allocator, request, &respond_options) catch |e| brk: { + respond_options.status = .internal_server_error; + // TODO: more about this particular request + log.err("Unexpected error from executor processing request: {any}", .{e}); + if (@errorReturnTrace()) |trace| { + std.debug.dumpStackTrace(trace.*); + } + break :brk "Unexpected error generating request to lambda"; + }; + try request.respond(response_bytes, respond_options); + log.debug( + "tid {d} (server): sent response: {s}", + .{ std.Thread.getCurrentId(), response_bytes }, + ); + } } -fn serve(allocator: std.mem.Allocator, res: *std.http.Server.Response) ![]const u8 { +fn serve(allocator: std.mem.Allocator, request: std.http.Server.Request, respond_options: *std.http.Server.Request.RespondOptions) ![]const u8 { _ = allocator; - // try res.headers.append("content-length", try std.fmt.allocPrint(allocator, "{d}", .{server_response.len})); - try res.headers.append("Lambda-Runtime-Aws-Request-Id", "69"); + _ = request; + respond_options.extra_headers = &.{ + .{ .name = "Lambda-Runtime-Aws-Request-Id", .value = "69" }, + }; return server_response; } @@ -406,7 +366,7 @@ fn test_run(allocator: std.mem.Allocator, event_handler: HandlerFn) !std.Thread fn lambda_request(allocator: std.mem.Allocator, request: []const u8, request_count: usize) ![]u8 { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); - var aa = arena.allocator(); + const aa = arena.allocator(); // Setup our server to run, and set the response for the server to the // request. There is a cognitive disconnect here between mental model and // physical model.