use async-client for telemetry

karlseguin · karlseguin · commit 97db8e974431 · 2025-03-03T19:32:18.000+08:00
diff --git a/src/app.zig b/src/app.zig
@@ -1,5 +1,6 @@
 const std = @import("std");
 
+const Loop = @import("jsruntime").Loop;
 const Allocator = std.mem.Allocator;
 const Telemetry = @import("telemetry/telemetry.zig").Telemetry;
 
@@ -8,8 +9,8 @@ const Telemetry = @import("telemetry/telemetry.zig").Telemetry;
 pub const App = struct {
     telemetry: Telemetry,
 
-    pub fn init(allocator: Allocator) !App {
-        const telemetry = Telemetry.init(allocator);
+    pub fn init(allocator: Allocator, loop: *Loop) !App {
+        const telemetry = Telemetry.init(allocator, loop);
         errdefer telemetry.deinit();
 
         return .{
diff --git a/src/main.zig b/src/main.zig
@@ -54,9 +54,6 @@ pub fn main() !void {
         _ = gpa.detectLeaks();
     };
 
-    var app = try @import("app.zig").App.init(alloc);
-    defer app.deinit();
-
     var args_arena = std.heap.ArenaAllocator.init(alloc);
     defer args_arena.deinit();
     const args = try parseArgs(args_arena.allocator());
@@ -68,7 +65,6 @@ pub fn main() !void {
             return std.process.cleanExit();
         },
         .serve => |opts| {
-            app.telemetry.record(.{ .run = .{ .mode = .serve, .version = version } });
             const address = std.net.Address.parseIp4(opts.host, opts.port) catch |err| {
                 log.err("address (host:port) {any}\n", .{err});
                 return args.printUsageAndExit(false);
@@ -77,24 +73,31 @@ pub fn main() !void {
             var loop = try jsruntime.Loop.init(alloc);
             defer loop.deinit();
 
+            var app = try @import("app.zig").App.init(alloc, &loop);
+            defer app.deinit();
+            app.telemetry.record(.{ .run = .{ .mode = .serve, .version = version } });
+
             const timeout = std.time.ns_per_s * @as(u64, opts.timeout);
-            server.run(alloc, address, timeout, &loop) catch |err| {
+            server.run(alloc, address, timeout, &loop, &app) catch |err| {
                 log.err("Server error", .{});
                 return err;
             };
         },
         .fetch => |opts| {
-            app.telemetry.record(.{ .run = .{ .mode = .fetch, .version = version } });
             log.debug("Fetch mode: url {s}, dump {any}", .{ opts.url, opts.dump });
 
-            // vm
-            const vm = jsruntime.VM.init();
-            defer vm.deinit();
-
             // loop
             var loop = try jsruntime.Loop.init(alloc);
             defer loop.deinit();
 
+            var app = try @import("app.zig").App.init(alloc, &loop);
+            defer app.deinit();
+            app.telemetry.record(.{ .run = .{ .mode = .fetch, .version = version } });
+
+            // vm
+            const vm = jsruntime.VM.init();
+            defer vm.deinit();
+
             // browser
             var browser = Browser.init(alloc, &loop);
             defer browser.deinit();
diff --git a/src/server.zig b/src/server.zig
@@ -34,6 +34,7 @@ const CloseError = jsruntime.IO.CloseError;
 const CancelError = jsruntime.IO.CancelOneError;
 const TimeoutError = jsruntime.IO.TimeoutError;
 
+const App = @import("app.zig").App;
 const CDP = @import("cdp/cdp.zig").CDP;
 
 const TimeoutCheck = std.time.ns_per_ms * 100;
@@ -50,6 +51,7 @@ const MAX_MESSAGE_SIZE = 256 * 1024 + 14;
 pub const Client = ClientT(*Server, CDP);
 
 const Server = struct {
+    app: *App,
     allocator: Allocator,
     loop: *jsruntime.Loop,
     current_client_id: usize = 0,
@@ -1028,6 +1030,7 @@ pub fn run(
     address: net.Address,
     timeout: u64,
     loop: *jsruntime.Loop,
+    app: *App,
 ) !void {
     // create socket
     const flags = posix.SOCK.STREAM | posix.SOCK.CLOEXEC | posix.SOCK.NONBLOCK;
@@ -1053,6 +1056,7 @@ pub fn run(
     const json_version_response = try buildJSONVersionResponse(allocator, address);
 
     var server = Server{
+        .app = app,
         .loop = loop,
         .timeout = timeout,
         .listener = listener,
diff --git a/src/telemetry/lightpanda.zig b/src/telemetry/lightpanda.zig
@@ -2,75 +2,176 @@ const std = @import("std");
 const Allocator = std.mem.Allocator;
 const ArenAallocator = std.heap.ArenaAllocator;
 
-const Event = @import("telemetry.zig").Event;
+const Loop = @import("jsruntime").Loop;
+const Client = @import("asyncio").Client;
+
 const log = std.log.scoped(.telemetry);
 
-const URL = "https://lightpanda.io/browser-stats";
+const URL = "https://stats.lightpanda.io";
 
-pub const Lightpanda = struct {
+pub const LightPanda = struct {
     uri: std.Uri,
-    arena: ArenAallocator,
-    client: std.http.Client,
-    headers: [1]std.http.Header,
+    io: Client.IO,
+    client: Client,
+    allocator: Allocator,
+    sending_pool: std.heap.MemoryPool(Sending),
+    client_context_pool: std.heap.MemoryPool(Client.Ctx),
 
-    pub fn init(allocator: Allocator) !Lightpanda {
+    pub fn init(allocator: Allocator, loop: *Loop) !LightPanda {
         return .{
+            .allocator = allocator,
+            .io = Client.IO.init(loop),
             .client = .{ .allocator = allocator },
-            .arena = std.heap.ArenaAllocator.init(allocator),
             .uri = std.Uri.parse(URL) catch unreachable,
-            .headers = [1]std.http.Header{
-                .{ .name = "Content-Type", .value = "application/json" },
-            },
+            .sending_pool = std.heap.MemoryPool(Sending).init(allocator),
+            .client_context_pool = std.heap.MemoryPool(Client.Ctx).init(allocator),
         };
     }
 
-    pub fn deinit(self: *Lightpanda) void {
-        self.arena.deinit();
+    pub fn deinit(self: *LightPanda) void {
         self.client.deinit();
+        self.sending_pool.deinit();
+        self.client_context_pool.deinit();
     }
 
-    pub fn send(self: *Lightpanda, iid: ?[]const u8, eid: []const u8, events: []Event) !void {
-        _ = self;
-        _ = iid;
-        _ = eid;
-        _ = events;
-        // defer _ = self.arena.reset(.{ .retain_capacity = {} });
-        // const body = try std.json.stringifyAlloc(self.arena.allocator(), PlausibleEvent{ .event = event }, .{});
-
-        // var server_headers: [2048]u8 = undefined;
-        // var req = try self.client.open(.POST, self.uri, .{
-        //     .redirect_behavior = .not_allowed,
-        //     .extra_headers = &self.headers,
-        //     .server_header_buffer = &server_headers,
-        // });
-        // req.transfer_encoding = .{ .content_length = body.len };
-        // try req.send();
-
-        // try req.writeAll(body);
-        // try req.finish();
-        // try req.wait();
-
-        // const status = req.response.status;
-        // if (status != .accepted) {
-        //     log.warn("telemetry '{s}' event error: {d}", .{ @tagName(event), @intFromEnum(status) });
-        // } else {
-        //     log.warn("telemetry '{s}' sent", .{@tagName(event)});
-        // }
+    pub fn send(self: *LightPanda, iid: ?[]const u8, eid: []const u8, event: anytype) !void {
+        var arena = std.heap.ArenaAllocator.init(self.allocator);
+        errdefer arena.deinit();
+
+        const resp_header_buffer = try arena.allocator().alloc(u8, 4096);
+        const body = try std.json.stringifyAlloc(arena.allocator(), .{
+            .iid = iid,
+            .eid = eid,
+            .event = event,
+        }, .{});
+
+        const sending = try self.sending_pool.create();
+        errdefer self.sending_pool.destroy(sending);
+
+        sending.* = .{
+            .body = body,
+            .arena = arena,
+            .lightpanda = self,
+            .request = try self.client.create(.POST, self.uri, .{
+                .server_header_buffer = resp_header_buffer,
+            }),
+        };
+        errdefer sending.request.deinit();
+
+        const ctx = try self.client_context_pool.create();
+        errdefer self.client_context_pool.destroy(ctx);
+
+        ctx.* = try Client.Ctx.init(&self.io, &sending.request);
+        ctx.userData = sending;
+
+        try self.client.async_open(
+            .POST,
+            self.uri,
+            .{ .server_header_buffer = resp_header_buffer },
+            ctx,
+            onRequestConnect,
+        );
+    }
+
+    fn handleError(self: *LightPanda, ctx: *Client.Ctx, err: anyerror) anyerror!void {
+        ctx.deinit();
+        self.client_context_pool.destroy(ctx);
+
+        var sending: *Sending = @ptrCast(@alignCast(ctx.userData));
+        sending.deinit();
+        self.sending_pool.destroy(sending);
+        log.info("request failure: {}", .{err});
+    }
+
+    fn onRequestConnect(ctx: *Client.Ctx, res: anyerror!void) anyerror!void {
+        var sending: *Sending = @ptrCast(@alignCast(ctx.userData));
+        res catch |err| return sending.lightpanda.handleError(ctx, err);
+
+        ctx.req.transfer_encoding = .{ .content_length = sending.body.len };
+        return ctx.req.async_send(ctx, onRequestSend) catch |err| {
+            return sending.lightpanda.handleError(ctx, err);
+        };
+    }
+
+    fn onRequestSend(ctx: *Client.Ctx, res: anyerror!void) anyerror!void {
+        var sending: *Sending = @ptrCast(@alignCast(ctx.userData));
+        res catch |err| return sending.lightpanda.handleError(ctx, err);
+
+        return ctx.req.async_writeAll(sending.body, ctx, onRequestWrite) catch |err| {
+            return sending.lightpanda.handleError(ctx, err);
+        };
+    }
+
+    fn onRequestWrite(ctx: *Client.Ctx, res: anyerror!void) anyerror!void {
+        var sending: *Sending = @ptrCast(@alignCast(ctx.userData));
+        res catch |err| return sending.lightpanda.handleError(ctx, err);
+        return ctx.req.async_finish(ctx, onRequestFinish) catch |err| {
+            return sending.lightpanda.handleError(ctx, err);
+        };
+    }
+
+    fn onRequestFinish(ctx: *Client.Ctx, res: anyerror!void) anyerror!void {
+        var sending: *Sending = @ptrCast(@alignCast(ctx.userData));
+        res catch |err| return sending.lightpanda.handleError(ctx, err);
+        return ctx.req.async_wait(ctx, onRequestWait) catch |err| {
+            return sending.lightpanda.handleError(ctx, err);
+        };
+    }
+
+    fn onRequestWait(ctx: *Client.Ctx, res: anyerror!void) anyerror!void {
+        var sending: *Sending = @ptrCast(@alignCast(ctx.userData));
+        res catch |err| return sending.lightpanda.handleError(ctx, err);
+
+        const lightpanda = sending.lightpanda;
+
+        defer {
+            ctx.deinit();
+            lightpanda.client_context_pool.destroy(ctx);
+
+            sending.deinit();
+            lightpanda.sending_pool.destroy(sending);
+        }
+
+        var buffer: [2048]u8 = undefined;
+        const reader = ctx.req.reader();
+        while (true) {
+            const n = reader.read(&buffer) catch 0;
+            if (n == 0) {
+                break;
+            }
+        }
+        if (ctx.req.response.status != .ok) {
+            log.info("invalid response: {d}", .{@intFromEnum(ctx.req.response.status)});
+        }
+    }
+};
+
+const Sending = struct {
+    body: []const u8,
+    request: Client.Request,
+    lightpanda: *LightPanda,
+    arena: std.heap.ArenaAllocator,
+
+    pub fn deinit(self: *Sending) void {
+        self.arena.deinit();
+        self.request.deinit();
     }
 };
 
-// wraps a telemetry event so that we can serialize it to plausible's event endpoint
-// const PlausibleEvent = struct {
-//     event: Event,
+// // wraps a telemetry event so that we can serialize it to plausible's event endpoint
+// const EventWrap = struct {
+//     iid: ?[]const u8,
+//     eid: []const u8,
+//     event: *const Event,
 
-//     pub fn jsonStringify(self: PlausibleEvent, jws: anytype) !void {
+//     pub fn jsonStringify(self: *const EventWrap, jws: anytype) !void {
 //         try jws.beginObject();
-//         try jws.objectField("name");
-//         try jws.write(@tagName(self.event));
-//         try jws.objectField("url");
-//         try jws.write(EVENT_URL);
-//         try jws.objectField("domain");
-//         try jws.write(DOMAIN_KEY);
+//         try jws.objectField("iid");
+//         try jws.write(self.iid);
+//         try jws.objectField("eid");
+//         try jws.write(self.eid);
+//         try jws.objectField("event");
+//         try jws.write(@tagName(self.event.*));
 //         try jws.objectField("props");
 //         switch (self.event) {
 //             inline else => |props| try jws.write(props),
@@ -80,11 +181,15 @@ pub const Lightpanda = struct {
 // };
 
 // const testing = std.testing;
-// test "plausible: json event" {
-//     const json = try std.json.stringifyAlloc(testing.allocator, PlausibleEvent{ .event = .{ .run = .{ .mode = .serve, .version = "over 9000!" } } }, .{});
+// test "telemetry: lightpanda json event" {
+//     const json = try std.json.stringifyAlloc(testing.allocator, EventWrap{
+//         .iid = "1234",
+//         .eid = "abc!",
+//         .event = .{ .run = .{ .mode = .serve, .version = "over 9000!" } }
+//     }, .{});
 //     defer testing.allocator.free(json);
 
 //     try testing.expectEqualStrings(
-//         \\{"name":"run","url":"https://lightpanda.io/browser-stats","domain":"localhost","props":{"version":"over 9000!","mode":"serve"}}
+//         \\{"event":"run","iid""1234","eid":"abc!","props":{"version":"over 9000!","mode":"serve"}}
 //     , json);
 // }
diff --git a/src/telemetry/telemetry.zig b/src/telemetry/telemetry.zig
diff --git a/src/unit_tests.zig b/src/unit_tests.zig
