lobo/universal-lambda-zig
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

zig 0.12.0: remove all files that should come from dependent packages

Browse Source
This commit is contained in:
Emil Lerch 2024-05-06 12:15:29 -07:00
parent 5f1b1a52be
commit a962e05d82
Signed by: lobo
GPG Key ID: A7B62D657EF764F8
7 changed files with 0 additions and 1248 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
src/CloudflareDeployStep.zig
View File

@ -1,100 +0,0 @@
const std = @import("std");
const cloudflare = @import("cloudflaredeploy.zig");
const CloudflareDeployStep = @This();
pub const base_id: std.Build.Step.Id = .custom;
step: std.Build.Step,
primary_javascript_file: std.Build.LazyPath,
worker_name: []const u8,
options: Options,
pub const Options = struct {
/// if set, the primary file will not be read (and may not exist). This data
/// will be used instead
primary_file_data: ?[]const u8 = null,
/// When set, the Javascript file will be searched/replaced with the target
/// file name for import
wasm_name: ?struct {
search: []const u8,
replace: []const u8,
} = null,
/// When set, the directory specified will be used rather than the current directory
wasm_dir: ?[]const u8 = null,
};
pub fn create(
owner: *std.Build,
worker_name: []const u8,
primary_javascript_file: std.Build.LazyPath,
options: Options,
) *CloudflareDeployStep {
const self = owner.allocator.create(CloudflareDeployStep) catch @panic("OOM");
self.* = CloudflareDeployStep{
.step = std.Build.Step.init(.{
.id = base_id,
.name = owner.fmt("cloudflare deploy {s}", .{primary_javascript_file.getDisplayName()}),
.owner = owner,
.makeFn = make,
}),
.primary_javascript_file = primary_javascript_file,
.worker_name = worker_name,
.options = options,
};
if (options.primary_file_data == null)
primary_javascript_file.addStepDependencies(&self.step);
return self;
}
fn make(step: *std.Build.Step, prog_node: *std.Progress.Node) !void {
_ = prog_node;
const b = step.owner;
const self = @fieldParentPtr(CloudflareDeployStep, "step", step);
var client = std.http.Client{ .allocator = b.allocator };
defer client.deinit();
var proxy_text = std.os.getenv("https_proxy") orelse std.os.getenv("HTTPS_PROXY");
if (proxy_text) |p| {
client.deinit();
const proxy = try std.Uri.parse(p);
client = std.http.Client{
.allocator = b.allocator,
.proxy = .{
.protocol = if (std.ascii.eqlIgnoreCase(proxy.scheme, "http")) .plain else .tls,
.host = proxy.host.?,
.port = proxy.port,
},
};
}
const script = self.options.primary_file_data orelse
try std.fs.cwd().readFileAlloc(b.allocator, self.primary_javascript_file.path, std.math.maxInt(usize));
defer if (self.options.primary_file_data == null) b.allocator.free(script);
var final_script = script;
if (self.options.wasm_name) |n| {
final_script = try std.mem.replaceOwned(u8, b.allocator, script, n.search, n.replace);
if (self.options.primary_file_data == null) b.allocator.free(script);
}
defer if (self.options.wasm_name) |_| b.allocator.free(final_script);
var al = std.ArrayList(u8).init(b.allocator);
defer al.deinit();
try cloudflare.pushWorker(
b.allocator,
&client,
self.worker_name,
self.options.wasm_dir orelse ".",
final_script,
al.writer(),
std.io.getStdErr().writer(),
);
const start = std.mem.lastIndexOf(u8, al.items, "http").?;
step.name = try std.fmt.allocPrint(
b.allocator,
"cloudflare deploy {s} to {s}",
.{ self.primary_javascript_file.getDisplayName(), al.items[start .. al.items.len - 1] },
);
}

26
src/cloudflare_build.zig
View File

@ -1,26 +0,0 @@
const std = @import("std");
const builtin = @import("builtin");
const CloudflareDeployStep = @import("CloudflareDeployStep.zig");
const script = @embedFile("index.js");
pub fn configureBuild(b: *std.build.Builder, cs: *std.Build.Step.Compile, function_name: []const u8) !void {
const wasm_name = try std.fmt.allocPrint(b.allocator, "{s}.wasm", .{cs.name});
const deploy_cmd = CloudflareDeployStep.create(
b,
function_name,
.{ .path = "index.js" },
.{
.primary_file_data = script,
.wasm_name = .{
.search = "custom.wasm",
.replace = wasm_name,
},
.wasm_dir = b.getInstallPath(.bin, "."),
},
);
deploy_cmd.step.dependOn(b.getInstallStep());
const deploy_step = b.step("cloudflare", "Deploy as Cloudflare worker (must be compiled with -Dtarget=wasm32-wasi)");
deploy_step.dependOn(&deploy_cmd.step);
}

377
src/cloudflaredeploy.zig
View File

@ -1,377 +0,0 @@
const std = @import("std");
var x_auth_token: ?[:0]const u8 = undefined;
var x_auth_email: ?[:0]const u8 = undefined;
var x_auth_key: ?[:0]const u8 = undefined;
var initialized = false;
const cf_api_base = "https://api.cloudflare.com/client/v4";
pub fn main() !u8 {
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
defer arena.deinit();
const allocator = arena.allocator();
var client = std.http.Client{ .allocator = allocator };
defer client.deinit();
var proxy_text = std.os.getenv("https_proxy") orelse std.os.getenv("HTTPS_PROXY");
if (proxy_text) |p| {
client.deinit();
const proxy = try std.Uri.parse(p);
client = std.http.Client{
.allocator = allocator,
.proxy = .{
.protocol = if (std.ascii.eqlIgnoreCase(proxy.scheme, "http")) .plain else .tls,
.host = proxy.host.?,
.port = proxy.port,
},
};
}
const stdout_file = std.io.getStdOut().writer();
var bw = std.io.bufferedWriter(stdout_file);
const stdout = bw.writer();
var argIterator = try std.process.argsWithAllocator(allocator);
defer argIterator.deinit();
const exe_name = argIterator.next().?;
var maybe_name = argIterator.next();
if (maybe_name == null) {
try usage(std.io.getStdErr().writer(), exe_name);
return 1;
}
const worker_name = maybe_name.?;
if (std.mem.eql(u8, worker_name, "-h")) {
try usage(stdout, exe_name);
return 0;
}
var maybe_script_name = argIterator.next();
if (maybe_script_name == null) {
try usage(std.io.getStdErr().writer(), exe_name);
return 1;
}
const script = std.fs.cwd().readFileAlloc(allocator, maybe_script_name.?, std.math.maxInt(usize)) catch |err| {
try usage(std.io.getStdErr().writer(), exe_name);
return err;
};
pushWorker(allocator, &client, worker_name, script, ".", stdout, std.io.getStdErr().writer()) catch return 1;
try bw.flush(); // don't forget to flush!
return 0;
}
fn usage(writer: anytype, this: []const u8) !void {
try writer.print("usage: {s} <worker name> <script file>\n", .{this});
}
const Wasm = struct {
allocator: std.mem.Allocator,
name: []const u8,
data: []const u8,
const Self = @This();
pub fn deinit(self: *Self) void {
self.allocator.free(self.name);
self.allocator.free(self.data);
}
};
pub fn pushWorker(
allocator: std.mem.Allocator,
client: *std.http.Client,
worker_name: []const u8,
wasm_dir: []const u8,
script: []const u8,
writer: anytype,
err_writer: anytype,
) !void {
var wasm = try loadWasm(allocator, script, wasm_dir);
defer wasm.deinit();
var accountid = std.os.getenv("CLOUDFLARE_ACCOUNT_ID");
const account_id_free = accountid == null;
if (accountid == null) accountid = try getAccountId(allocator, client);
defer if (account_id_free) allocator.free(accountid.?);
try writer.print("Using Cloudflare account: {s}\n", .{accountid.?});
// Determine if worker exists. This lets us know if we need to enable it later
const worker_exists = try workerExists(allocator, client, accountid.?, worker_name);
try writer.print(
"{s}\n",
.{if (worker_exists) "Worker exists, will not re-enable" else "Worker is new. Will enable after code update"},
);
var worker = Worker{
.account_id = accountid.?,
.name = worker_name,
.wasm = wasm,
.main_module = script,
};
putNewWorker(allocator, client, &worker) catch |err| {
if (worker.errors == null) return err;
try err_writer.print("{d} errors returned from CloudFlare:\n\n", .{worker.errors.?.len});
for (worker.errors.?) |cf_err| {
try err_writer.print("{s}\n", .{cf_err});
allocator.free(cf_err);
}
return error.CloudFlareErrorResponse;
};
const subdomain = try getSubdomain(allocator, client, accountid.?);
defer allocator.free(subdomain);
try writer.print("Worker available at: https://{s}.{s}.workers.dev/\n", .{ worker_name, subdomain });
if (!worker_exists)
try enableWorker(allocator, client, accountid.?, worker_name);
}
fn loadWasm(allocator: std.mem.Allocator, script: []const u8, wasm_dir: []const u8) !Wasm {
// Looking for a string like this: import demoWasm from "demo.wasm"
// JavaScript may or may not have ; characters. We're not doing
// a full JS parsing here, so this may not be the most robust
var inx: usize = 0;
var name: ?[]const u8 = null;
while (true) {
inx = std.mem.indexOf(u8, script[inx..], "import ") orelse if (inx == 0) return error.NoImportFound else break;
inx += "import ".len;
// oh god, we're not doing this: https://262.ecma-international.org/5.1/#sec-7.5
// advance to next token - we don't care what the name is
while (inx < script.len and script[inx] != ' ') inx += 1;
// continue past space(s)
while (inx < script.len and script[inx] == ' ') inx += 1;
// We expect "from " to be next
if (!std.mem.startsWith(u8, script[inx..], "from ")) continue;
inx += "from ".len;
// continue past space(s)
while (inx < script.len and script[inx] == ' ') inx += 1;
// We now expect the name of our file...
if (script[inx] != '"' and script[inx] != '\'') continue; // we're not where we think we are
const quote = script[inx]; // there are two possibilities here
// we don't need to advance inx any more, as we're on the name, and if
// we loop, that's ok
inx += 1; // move off the quote onto the name
const end_quote_inx = std.mem.indexOfScalar(u8, script[inx..], quote);
if (end_quote_inx == null) continue;
const candidate_name = script[inx .. inx + end_quote_inx.?];
if (std.mem.endsWith(u8, candidate_name, ".wasm")) {
if (name != null) // we are importing two wasm files, and we are now lost
return error.MultipleWasmImportsUnsupported;
name = candidate_name;
}
}
if (name == null) return error.NoWasmImportFound;
const nm = try allocator.dupe(u8, name.?);
errdefer allocator.free(nm);
const path = try std.fs.path.join(allocator, &[_][]const u8{ wasm_dir, nm });
defer allocator.free(path);
const data = try std.fs.cwd().readFileAlloc(allocator, path, std.math.maxInt(usize));
return Wasm{
.allocator = allocator,
.name = nm,
.data = data,
};
}
fn getAccountId(allocator: std.mem.Allocator, client: *std.http.Client) ![:0]const u8 {
const url = cf_api_base ++ "/accounts/";
var headers = std.http.Headers.init(allocator);
defer headers.deinit();
try addAuthHeaders(&headers);
var req = try client.request(.GET, try std.Uri.parse(url), headers, .{});
defer req.deinit();
try req.start();
try req.wait();
if (req.response.status != .ok) {
std.debug.print("Status is {}\n", .{req.response.status});
return error.RequestFailed;
}
var json_reader = std.json.reader(allocator, req.reader());
defer json_reader.deinit();
var body = try std.json.parseFromTokenSource(std.json.Value, allocator, &json_reader, .{});
defer body.deinit();
const arr = body.value.object.get("result").?.array.items;
if (arr.len == 0) return error.NoAccounts;
if (arr.len > 1) return error.TooManyAccounts;
return try allocator.dupeZ(u8, arr[0].object.get("id").?.string);
}
fn enableWorker(allocator: std.mem.Allocator, client: *std.http.Client, account_id: []const u8, name: []const u8) !void {
const enable_script = cf_api_base ++ "/accounts/{s}/workers/scripts/{s}/subdomain";
const url = try std.fmt.allocPrint(allocator, enable_script, .{ account_id, name });
defer allocator.free(url);
var headers = std.http.Headers.init(allocator);
defer headers.deinit();
try addAuthHeaders(&headers);
try headers.append("Content-Type", "application/json; charset=UTF-8");
var req = try client.request(.POST, try std.Uri.parse(url), headers, .{});
defer req.deinit();
const request_payload =
\\{ "enabled": true }
;
req.transfer_encoding = .{ .content_length = @as(u64, request_payload.len) };
try req.start();
try req.writeAll(request_payload);
try req.finish();
try req.wait();
if (req.response.status != .ok) {
std.debug.print("Status is {}\n", .{req.response.status});
return error.RequestFailed;
}
}
/// Gets the subdomain for a worker. Caller owns memory
fn getSubdomain(allocator: std.mem.Allocator, client: *std.http.Client, account_id: []const u8) ![]const u8 {
const get_subdomain = cf_api_base ++ "/accounts/{s}/workers/subdomain";
const url = try std.fmt.allocPrint(allocator, get_subdomain, .{account_id});
defer allocator.free(url);
var headers = std.http.Headers.init(allocator);
defer headers.deinit();
try addAuthHeaders(&headers);
var req = try client.request(.GET, try std.Uri.parse(url), headers, .{});
defer req.deinit();
try req.start();
try req.wait();
if (req.response.status != .ok) return error.RequestNotOk;
var json_reader = std.json.reader(allocator, req.reader());
defer json_reader.deinit();
var body = try std.json.parseFromTokenSource(std.json.Value, allocator, &json_reader, .{});
defer body.deinit();
return try allocator.dupe(u8, body.value.object.get("result").?.object.get("subdomain").?.string);
}
const Worker = struct {
account_id: []const u8,
name: []const u8,
main_module: []const u8,
wasm: Wasm,
errors: ?[][]const u8 = null,
};
fn putNewWorker(allocator: std.mem.Allocator, client: *std.http.Client, worker: *Worker) !void {
const put_script = cf_api_base ++ "/accounts/{s}/workers/scripts/{s}?include_subdomain_availability=true&excludeScript=true";
const url = try std.fmt.allocPrint(allocator, put_script, .{ worker.account_id, worker.name });
defer allocator.free(url);
const memfs = @embedFile("dist/memfs.wasm");
const outer_script_shell = @embedFile("script_harness.js");
const script = try std.fmt.allocPrint(allocator, "{s}{s}", .{ outer_script_shell, worker.main_module });
defer allocator.free(script);
const deploy_request =
"------formdata-undici-032998177938\r\n" ++
"Content-Disposition: form-data; name=\"metadata\"\r\n\r\n" ++
"{{\"main_module\":\"index.js\",\"bindings\":[],\"compatibility_date\":\"2023-10-02\",\"compatibility_flags\":[]}}\r\n" ++
"------formdata-undici-032998177938\r\n" ++
"Content-Disposition: form-data; name=\"index.js\"; filename=\"index.js\"\r\n" ++
"Content-Type: application/javascript+module\r\n" ++
"\r\n" ++
"{[script]s}\r\n" ++
"------formdata-undici-032998177938\r\n" ++
"Content-Disposition: form-data; name=\"./{[wasm_name]s}\"; filename=\"./{[wasm_name]s}\"\r\n" ++
"Content-Type: application/wasm\r\n" ++
"\r\n" ++
"{[wasm]s}\r\n" ++
"------formdata-undici-032998177938\r\n" ++
"Content-Disposition: form-data; name=\"./c5f1acc97ad09df861eff9ef567c2186d4e38de3-memfs.wasm\"; filename=\"./c5f1acc97ad09df861eff9ef567c2186d4e38de3-memfs.wasm\"\r\n" ++
"Content-Type: application/wasm\r\n" ++
"\r\n" ++
"{[memfs]s}\r\n" ++
"------formdata-undici-032998177938--";
var headers = std.http.Headers.init(allocator);
defer headers.deinit();
try addAuthHeaders(&headers);
// TODO: fix this
try headers.append("Content-Type", "multipart/form-data; boundary=----formdata-undici-032998177938");
const request_payload = try std.fmt.allocPrint(allocator, deploy_request, .{
.script = script,
.wasm_name = worker.wasm.name,
.wasm = worker.wasm.data,
.memfs = memfs,
});
defer allocator.free(request_payload);
// Get content length. For some reason it's forcing a chunked transfer type without this.
// That's not entirely a bad idea, but for now I want to match what I see
// coming through wrangler
const cl = try std.fmt.allocPrint(allocator, "{d}", .{request_payload.len});
defer allocator.free(cl);
try headers.append("Content-Length", cl);
var req = try client.request(.PUT, try std.Uri.parse(url), headers, .{});
defer req.deinit();
req.transfer_encoding = .{ .content_length = @as(u64, request_payload.len) };
try req.start();
// Workaround for https://github.com/ziglang/zig/issues/15626
const max_bytes: usize = 1 << 14;
var inx: usize = 0;
while (request_payload.len > inx) {
try req.writeAll(request_payload[inx..@min(request_payload.len, inx + max_bytes)]);
inx += max_bytes;
}
try req.finish();
try req.wait();
// std.debug.print("Status is {}\n", .{req.response.status});
// std.debug.print("Url is {s}\n", .{url});
if (req.response.status != .ok) {
std.debug.print("Status is {}\n", .{req.response.status});
if (req.response.status == .bad_request)
worker.errors = getErrors(allocator, &req) catch null;
return error.RequestFailed;
}
}
fn getErrors(allocator: std.mem.Allocator, req: *std.http.Client.Request) !?[][]const u8 {
var json_reader = std.json.reader(allocator, req.reader());
defer json_reader.deinit();
var body = try std.json.parseFromTokenSource(std.json.Value, allocator, &json_reader, .{});
defer body.deinit();
const arr = body.value.object.get("errors").?.array.items;
if (arr.len == 0) return null;
var error_list = try std.ArrayList([]const u8).initCapacity(allocator, arr.len);
defer error_list.deinit();
for (arr) |item| {
error_list.appendAssumeCapacity(item.object.get("message").?.string);
}
return try error_list.toOwnedSlice();
}
fn workerExists(allocator: std.mem.Allocator, client: *std.http.Client, account_id: []const u8, name: []const u8) !bool {
const existence_check = cf_api_base ++ "/accounts/{s}/workers/services/{s}";
const url = try std.fmt.allocPrint(allocator, existence_check, .{ account_id, name });
defer allocator.free(url);
var headers = std.http.Headers.init(allocator);
defer headers.deinit();
try addAuthHeaders(&headers);
var req = try client.request(.GET, try std.Uri.parse(url), headers, .{});
defer req.deinit();
try req.start();
try req.wait();
// std.debug.print("Status is {}\n", .{req.response.status});
// std.debug.print("Url is {s}\n", .{url});
return req.response.status == .ok;
}
threadlocal var auth_buf: [1024]u8 = undefined;
fn addAuthHeaders(headers: *std.http.Headers) !void {
if (!initialized) {
x_auth_email = std.os.getenv("CLOUDFLARE_EMAIL");
x_auth_key = std.os.getenv("CLOUDFLARE_API_KEY");
x_auth_token = std.os.getenv("CLOUDFLARE_API_TOKEN");
initialized = true;
}
if (x_auth_token) |tok| {
var auth = try std.fmt.bufPrint(auth_buf[0..], "Bearer {s}", .{tok});
try headers.append("Authorization", auth);
return;
}
if (x_auth_email) |email| {
if (x_auth_key == null)
return error.MissingCloudflareApiKeyEnvironmentVariable;
try headers.append("X-Auth-Email", email);
try headers.append("X-Auth-Key", x_auth_key.?);
}
return error.NoCloudflareAuthenticationEnvironmentVariablesSet;
}

BIN
src/dist/memfs.wasm vendored
View File

Binary file not shown.

31
src/index.js
View File

@ -1,31 +0,0 @@
import customWasm from "custom.wasm";
export default {
async fetch(request, _env2, ctx) {
const stdout = new TransformStream();
console.log(request);
console.log(_env2);
console.log(ctx);
let env = {};
request.headers.forEach((value, key) => {
env[key] = value;
});
const wasi = new WASI({
args: [
"./custom.wasm",
// In a CLI, the first arg is the name of the exe
"--url=" + request.url,
// this contains the target but is the full url, so we will use a different arg for this
"--method=" + request.method,
'-request="' + JSON.stringify(request) + '"'
],
env,
stdin: request.body,
stdout: stdout.writable
});
const instance = new WebAssembly.Instance(customWasm, {
wasi_snapshot_preview1: wasi.wasiImport
});
ctx.waitUntil(wasi.start(instance));
return new Response(stdout.readable);
}
};

516
src/lambda.zig
View File

@ -1,516 +0,0 @@
const std = @import("std");
const builtin = @import("builtin");
const universal_lambda_interface = @import("universal_lambda_interface");
const HandlerFn = universal_lambda_interface.HandlerFn;
const Context = universal_lambda_interface.Context;
const UniversalLambdaResponse = universal_lambda_interface.Response;
const log = std.log.scoped(.lambda);
var empty_headers: std.http.Headers = undefined;
var client: ?std.http.Client = null;
const prefix = "http://";
const postfix = "/2018-06-01/runtime/invocation";
pub fn deinit() void {
if (client) |*c| c.deinit();
client = null;
}
/// Starts the lambda framework. Handler will be called when an event is processing
/// If an allocator is not provided, an approrpriate allocator will be selected and used
/// 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) !u8 { // TODO: remove inferred error set?
const lambda_runtime_uri = std.os.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(.{}){};
defer _ = gpa.deinit();
const alloc = allocator orelse gpa.allocator();
const url = try std.fmt.allocPrint(alloc, "{s}{s}{s}/next", .{ prefix, lambda_runtime_uri, postfix });
defer alloc.free(url);
const uri = try std.Uri.parse(url);
// TODO: Simply adding this line without even using the client is enough
// to cause seg faults!?
// client = client orelse .{ .allocator = alloc };
// 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) {
if (lambda_remaining_requests) |*r| {
// we're under test
log.debug("lambda remaining requests: {d}", .{r.*});
r.* -= 1;
}
var req_alloc = std.heap.ArenaAllocator.init(alloc);
defer req_alloc.deinit();
const req_allocator = req_alloc.allocator();
// Fundamentally we're doing 3 things:
// 1. Get the next event from Lambda (event data and request id)
// 2. Call our handler to get the response
// 3. Post the response back to Lambda
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);
// continue;
};
if (ev == null) continue; // this gets logged in getEvent, and without
// a request id, we still can't do anything
// reasonable to report back
const event = ev.?;
defer ev.?.deinit();
// Lambda does not have context, just environment variables. API Gateway
// might be configured to pass in lots of context, but this comes through
// event data, not context. In this case, we lose:
//
// request headers
// request method
// request target
var response = UniversalLambdaResponse.init(req_allocator);
defer response.deinit();
const body_writer = std.io.getStdOut();
const event_response = event_handler(req_allocator, event.event_data, &response) catch |err| {
body_writer.writeAll(response.body.items) catch unreachable;
event.reportError(@errorReturnTrace(), err, lambda_runtime_uri) catch unreachable;
continue;
};
// No error during handler. Write anything sent to body to stdout instead
// I'm not totally sure this is the right behavior as it is a little inconsistent
// (flexilib and console both write to the same io stream as the main output)
body_writer.writeAll(response.body.items) catch unreachable;
event.postResponse(lambda_runtime_uri, event_response) catch |err| {
event.reportError(@errorReturnTrace(), err, lambda_runtime_uri) catch unreachable;
continue;
};
}
return 0;
}
const Event = struct {
allocator: std.mem.Allocator,
event_data: []u8,
request_id: []u8,
const Self = @This();
pub fn init(allocator: std.mem.Allocator, event_data: []u8, request_id: []u8) Self {
return .{
.allocator = allocator,
.event_data = event_data,
.request_id = request_id,
};
}
pub fn deinit(self: *Self) void {
self.allocator.free(self.event_data);
self.allocator.free(self.request_id);
}
fn reportError(
self: Self,
return_trace: ?*std.builtin.StackTrace,
err: anytype,
lambda_runtime_uri: []const u8,
) !void {
// If we fail in this function, we're pretty hosed up
if (return_trace) |rt|
log.err("Caught error: {}. Return Trace: {any}", .{ err, rt })
else
log.err("Caught error: {}. No return trace available", .{err});
const err_url = try std.fmt.allocPrint(
self.allocator,
"{s}{s}{s}/{s}/error",
.{ prefix, lambda_runtime_uri, postfix, self.request_id },
);
defer self.allocator.free(err_url);
const err_uri = try std.Uri.parse(err_url);
const content =
\\{{
\\ "errorMessage": "{s}",
\\ "errorType": "HandlerReturnedError",
\\ "stackTrace": [ "{any}" ]
\\}}
;
const content_fmt = if (return_trace) |rt|
try std.fmt.allocPrint(self.allocator, content, .{ @errorName(err), rt })
else
try std.fmt.allocPrint(self.allocator, content, .{ @errorName(err), "no return trace available" });
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
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);
};
// TODO: Determine why this post is not returning
if (req.response.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 "",
});
std.os.exit(1);
}
log.err("Error reporting post complete", .{});
}
fn postResponse(self: Self, lambda_runtime_uri: []const u8, event_response: []const u8) !void {
const response_url = try std.fmt.allocPrint(
self.allocator,
"{s}{s}{s}/{s}/response",
.{ 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();
// Lambda does different things, depending on the runtime. Go 1.x takes
// any return value but escapes double quotes. Custom runtimes can
// do whatever they want. node I believe wraps as a json object. We're
// going to leave the return value up to the handler, and they can
// use a seperate API for normalization so we're explicit.
const response_content = try std.fmt.allocPrint(
self.allocator,
"{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();
}
};
fn getEvent(allocator: std.mem.Allocator, event_data_uri: std.Uri) !?Event {
// 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 = 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();
// 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) {
// 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 "",
});
return error.EventResponseNotOkResponse;
}
var request_id: ?[]const u8 = null;
var content_length: ?usize = null;
for (req.response.headers.list.items) |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,
// so we'll hold for now and think on this
// if (std.mem.indexOf(u8, h.name.value, "Lambda-Runtime-Trace-Id")) |_|
// std.process.
// std.os.setenv("AWS_LAMBDA_RUNTIME_API");
}
if (request_id == 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("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 response_data: []u8 =
if (req.response.transfer_encoding) |_| // the only value here is "chunked"
try req.reader().readAllAlloc(allocator, std.math.maxInt(usize))
else blk: {
// content length
var tmp_data = try allocator.alloc(u8, content_length.?);
errdefer allocator.free(tmp_data);
_ = try req.readAll(tmp_data);
break :blk tmp_data;
};
return Event.init(
allocator,
response_data,
try allocator.dupe(u8, req_id),
);
}
////////////////////////////////////////////////////////////////////////
// All code below this line is for testing
////////////////////////////////////////////////////////////////////////
var server_port: ?u16 = null;
var server_remaining_requests: usize = 0;
var lambda_remaining_requests: ?usize = null;
var server_response: []const u8 = "unset";
var server_request_aka_lambda_response: []u8 = "";
var test_lambda_runtime_uri: ?[]u8 = null;
var server_ready = false;
/// This starts a test server. We're not testing the server itself,
/// so the main tests will start this thing up and create an arena around the
/// whole thing so we can just deallocate everything at once at the end,
/// leaks be damned
fn startServer(allocator: std.mem.Allocator) !std.Thread {
return try std.Thread.spawn(
.{},
threadMain,
.{allocator},
);
}
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();
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();
// 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) {
server_remaining_requests -= 1;
// defer {
// if (!arena.reset(.{ .retain_capacity = {} })) {
// // reallocation failed, arena is degraded
// log.warn("Arena reset failed and is degraded. Resetting arena", .{});
// arena.deinit();
// arena = std.heap.ArenaAllocator.init(allocator);
// aa = arena.allocator();
// }
// }
processRequest(aa, &server) catch |e| {
log.err("Unexpected error processing request: {any}", .{e});
if (@errorReturnTrace()) |trace| {
std.debug.dumpStackTrace(trace.*);
}
};
}
}
fn processRequest(allocator: std.mem.Allocator, server: *std.http.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 });
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..];
if (res.request.content_length) |l|
server_request_aka_lambda_response = try res.reader().readAllAlloc(allocator, @as(usize, @intCast(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()},
);
}
fn serve(allocator: std.mem.Allocator, res: *std.http.Server.Response) ![]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");
return server_response;
}
fn handler(allocator: std.mem.Allocator, event_data: []const u8, context: Context) ![]const u8 {
_ = allocator;
_ = context;
return event_data;
}
fn thread_run(allocator: ?std.mem.Allocator, event_handler: HandlerFn) !void {
_ = try run(allocator, event_handler);
}
fn test_run(allocator: std.mem.Allocator, event_handler: HandlerFn) !std.Thread {
return try std.Thread.spawn(
.{},
thread_run,
.{ allocator, event_handler },
);
}
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();
// 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.
//
// Mental model:
//
// Lambda request -> λ -> Lambda response
//
// Physcial Model:
//
// 1. λ requests instructions from server
// 2. server provides "Lambda request"
// 3. λ posts response back to server
//
// So here we are setting up our server, then our lambda request loop,
// but it all needs to be in seperate threads so we can control startup
// and shut down. Both server and Lambda are set up to watch global variable
// booleans to know when to shut down. This function is designed for a
// single request/response pair only
lambda_remaining_requests = request_count;
server_remaining_requests = lambda_remaining_requests.? * 2; // Lambda functions
// fetch from the server,
// then post back. Always
// 2, no more, no less
server_response = request; // set our instructions to lambda, which in our
// physical model above, is the server response
defer server_response = "unset"; // set it back so we don't get confused later
// when subsequent tests fail
const server_thread = try startServer(aa); // start the server, get it ready
while (!server_ready)
std.time.sleep(100);
log.debug("tid {d} (main): server reports ready", .{std.Thread.getCurrentId()});
// we aren't testing the server,
// so we'll use the arena allocator
defer server_thread.join(); // we'll be shutting everything down before we exit
// Now we need to start the lambda framework, following a siimilar pattern
const lambda_thread = try test_run(allocator, handler); // We want our function under test to report leaks
lambda_thread.join();
return try allocator.dupe(u8, server_request_aka_lambda_response);
}
test "basic request" {
// std.testing.log_level = .debug;
const allocator = std.testing.allocator;
const request =
\\{"foo": "bar", "baz": "qux"}
;
const expected_response =
\\{"foo": "bar", "baz": "qux"}
;
const lambda_response = try lambda_request(allocator, request, 1);
defer deinit();
defer allocator.free(lambda_response);
try std.testing.expectEqualStrings(expected_response, lambda_response);
}
test "several requests do not fail" {
// std.testing.log_level = .debug;
const allocator = std.testing.allocator;
const request =
\\{"foo": "bar", "baz": "qux"}
;
const expected_response =
\\{"foo": "bar", "baz": "qux"}
;
const lambda_response = try lambda_request(allocator, request, 5);
defer deinit();
defer allocator.free(lambda_response);
try std.testing.expectEqualStrings(expected_response, lambda_response);
}

198
src/lambda_build.zig
View File

@ -1,198 +0,0 @@
const std = @import("std");
const builtin = @import("builtin");
fn fileExists(file_name: []const u8) bool {
const file = std.fs.openFileAbsolute(file_name, .{}) catch return false;
defer file.close();
return true;
}
fn addArgs(allocator: std.mem.Allocator, original: []const u8, args: [][]const u8) ![]const u8 {
var rc = original;
for (args) |arg| {
rc = try std.mem.concat(allocator, u8, &.{ rc, " ", arg });
}
return rc;
}
/// lambdaBuildSteps will add four build steps to the build (if compiling
/// the code on a Linux host):
///
/// * awslambda_package: Packages the function for deployment to Lambda
/// (dependencies are the zip executable and a shell)
/// * awslambda_iam: Gets an IAM role for the Lambda function, and creates it if it does not exist
/// (dependencies are the AWS CLI, grep and a shell)
/// * awslambda_deploy: Deploys the lambda function to a live AWS environment
/// (dependencies are the AWS CLI, and a shell)
/// * awslambda_run: Runs the lambda function in a live AWS environment
/// (dependencies are the AWS CLI, and a shell)
///
/// awslambda_run depends on deploy
/// awslambda_deploy depends on iam and package
///
/// iam and package do not have any dependencies
pub fn configureBuild(b: *std.build.Builder, exe: *std.Build.Step.Compile, function_name: []const u8) !void {
// The rest of this function is currently reliant on the use of Linux
// system being used to build the lambda function
//
// It is likely that much of this will work on other Unix-like OSs, but
// we will work this out later
//
// TODO: support other host OSs
if (builtin.os.tag != .linux) return;
// Package step
const package_step = b.step("awslambda_package", "Package the function");
const function_zip = b.getInstallPath(.bin, "function.zip");
// TODO: Avoid use of system-installed zip, maybe using something like
// https://github.com/hdorio/hwzip.zig/blob/master/src/hwzip.zig
const zip = if (std.mem.eql(u8, "bootstrap", exe.out_filename))
try std.fmt.allocPrint(b.allocator,
\\zip -qj9 {s} {s}
, .{
function_zip,
b.getInstallPath(.bin, "bootstrap"),
})
else
// We need to copy stuff around
try std.fmt.allocPrint(b.allocator,
\\cp {s} {s} && \
\\zip -qj9 {s} {s} && \
\\rm {s}
, .{
b.getInstallPath(.bin, exe.out_filename),
b.getInstallPath(.bin, "bootstrap"),
function_zip,
b.getInstallPath(.bin, "bootstrap"),
b.getInstallPath(.bin, "bootstrap"),
});
// std.debug.print("\nzip cmdline: {s}", .{zip});
defer b.allocator.free(zip);
var zip_cmd = b.addSystemCommand(&.{ "/bin/sh", "-c", zip });
zip_cmd.step.dependOn(b.getInstallStep());
package_step.dependOn(&zip_cmd.step);
// Deployment
const deploy_step = b.step("awslambda_deploy", "Deploy the function");
const iam_role_name = b.option(
[]const u8,
"function-role",
"IAM role name for function (will create if it does not exist) [lambda_basic_execution]",
) orelse "lambda_basic_execution";
var iam_role_arn = b.option(
[]const u8,
"function-arn",
"Preexisting IAM role arn for function",
);
const iam_step = b.step("awslambda_iam", "Create/Get IAM role for function");
deploy_step.dependOn(iam_step); // iam_step will either be a noop or all the stuff below
const iam_role_param: []u8 = blk: {
if (iam_role_arn != null)
break :blk try std.fmt.allocPrint(b.allocator, "--role {s}", .{iam_role_arn.?});
if (iam_role_name.len == 0)
@panic("Either function-role or function-arn must be specified. function-arn will allow deployment without creating a role");
// Now we have an iam role name to use, but no iam role arn. Let's go hunting
// Once this is done once, we'll have a file with the arn in "cache"
// The iam arn will reside in an 'iam_role' file in the bin directory
// Build system command to create the role if necessary and get the role arn
const iam_role_file = b.getInstallPath(.bin, "iam_role");
if (!fileExists(iam_role_file)) {
// std.debug.print("file does not exist", .{});
// Our cache file does not exist on disk, so we'll create/get the role
// arn using the AWS CLI and dump to disk here
const ifstatement_fmt =
\\ if aws iam get-role --role-name {s} 2>&1 |grep -q NoSuchEntity; then aws iam create-role --output text --query Role.Arn --role-name {s} --assume-role-policy-document '{{
\\ "Version": "2012-10-17",
\\ "Statement": [
\\ {{
\\ "Sid": "",
\\ "Effect": "Allow",
\\ "Principal": {{
\\ "Service": "lambda.amazonaws.com"
\\ }},
\\ "Action": "sts:AssumeRole"
\\ }}
\\ ]}}' > /dev/null; fi && \
\\ aws iam attach-role-policy --policy-arn arn:aws:iam::aws:policy/AWSLambdaExecute --role-name lambda_basic_execution && \
\\ aws iam get-role --role-name lambda_basic_execution --query Role.Arn --output text > {s}
;
const ifstatement = try std.fmt.allocPrint(
b.allocator,
ifstatement_fmt,
.{ iam_role_name, iam_role_name, iam_role_file },
);
iam_step.dependOn(&b.addSystemCommand(&.{ "/bin/sh", "-c", ifstatement }).step);
}
break :blk try std.fmt.allocPrint(b.allocator, "--role \"$(cat {s})\"", .{iam_role_file});
};
const function_name_file = b.getInstallPath(.bin, function_name);
const ifstatement = "if [ ! -f {s} ] || [ {s} -nt {s} ]; then if aws lambda get-function --function-name {s} 2>&1 |grep -q ResourceNotFoundException; then echo not found > /dev/null; {s}; else echo found > /dev/null; {s}; fi; fi";
// The architectures option was introduced in 2.2.43 released 2021-10-01
// We want to use arm64 here because it is both faster and cheaper for most
// Amazon Linux 2 is the only arm64 supported option
// TODO: This should determine compilation target and use x86_64 if needed
const not_found = "aws lambda create-function --architectures arm64 --runtime provided.al2 --function-name {s} --zip-file fileb://{s} --handler not_applicable {s} && touch {s}";
const not_found_fmt = try std.fmt.allocPrint(b.allocator, not_found, .{ function_name, function_zip, iam_role_param, function_name_file });
defer b.allocator.free(not_found_fmt);
const found = "aws lambda update-function-code --function-name {s} --zip-file fileb://{s} && touch {s}";
const found_fmt = try std.fmt.allocPrint(b.allocator, found, .{ function_name, function_zip, function_name_file });
defer b.allocator.free(found_fmt);
var found_final: []const u8 = undefined;
var not_found_final: []const u8 = undefined;
if (b.args) |args| {
found_final = try addArgs(b.allocator, found_fmt, args);
not_found_final = try addArgs(b.allocator, not_found_fmt, args);
} else {
found_final = found_fmt;
not_found_final = not_found_fmt;
}
const cmd = try std.fmt.allocPrint(b.allocator, ifstatement, .{
function_name_file,
b.getInstallPath(.bin, exe.out_filename),
function_name_file,
function_name,
not_found_fmt,
found_fmt,
});
defer b.allocator.free(cmd);
// std.debug.print("{s}\n", .{cmd});
deploy_step.dependOn(package_step);
deploy_step.dependOn(&b.addSystemCommand(&.{ "/bin/sh", "-c", cmd }).step);
const payload = b.option([]const u8, "payload", "Lambda payload [{\"foo\":\"bar\", \"baz\": \"qux\"}]") orelse
\\ {"foo": "bar", "baz": "qux"}"
;
const run_script =
\\ f=$(mktemp) && \
\\ logs=$(aws lambda invoke \
\\ --cli-binary-format raw-in-base64-out \
\\ --invocation-type RequestResponse \
\\ --function-name {s} \
\\ --payload '{s}' \
\\ --log-type Tail \
\\ --query LogResult \
\\ --output text "$f" |base64 -d) && \
\\ cat "$f" && rm "$f" && \
\\ echo && echo && echo "$logs"
;
const run_script_fmt = try std.fmt.allocPrint(b.allocator, run_script, .{ function_name, payload });
defer b.allocator.free(run_script_fmt);
const run_cmd = b.addSystemCommand(&.{ "/bin/sh", "-c", run_script_fmt });
run_cmd.step.dependOn(deploy_step);
if (b.args) |args| {
run_cmd.addArgs(args);
}
const run_step = b.step("awslambda_run", "Run the app in AWS lambda");
run_step.dependOn(&run_cmd.step);
}