Dekaf materialization endpoint support

[jshearer]

Description:

This adds support for the server side of Dekaf's support for materialization endpoints. At a high level, Dekaf is just another way to get data out of Flow. We already have a well fleshed out concept for these things: materializations. So back in #1665 we introduced support for a new materialization "endpoint type": dekaf. This lives alongside local and connector as the third kind of materialization, and is configured like so:

    endpoint:
      dekaf:
        variant: some-dekaf-variant
        config:
          token: "foo"
          ...other configuration for the behavior of your Dekaf task... 

The second part of this work is for Dekaf the server to support this mode of operation. Briefly, it needs to:

• Support authentication and authorization using the control plane. This means that conceptually, Dekaf is authenticating you based on the task name and token specified in the endpoint config, and authorizing you based on the access granted to that task
  • Specifically, this means that you now use the task name for your username, and the token specified in the endpoint config when connecting to Dekaf.
• Act more like a regular materialization. This means...
  • Support for field selection. Materializations let you specify which field(s) you want included in your destination using field selection. This fundamentally looks like a document transformation, where some fields may be removed and some fields may be projected from their original home to a new location in the output.
  • Support for task logs. Just like captures, materialization log output is captured and presented to users for status monitoring and debugging. As Dekaf is multi-tenant at its core, presenting these logs requires identifying which log messages are associated with which task, and then capturing and writing them to the corresponding logs journal.
  • [Fast-Follow] Support for task stats. In order to monitor the status of your running tasks, as well as aggregate usage information for billing, all tasks need to periodically emit information about how much work they've been doing. While Dekaf was in beta, all usage was free so this was less of a priority, but now that there's a task to associate stats with, implementing stats will likely be one of the last things to do before going GA.

I still have a couple of things on my list before this is fully wrapped up:

• Implement emitting stats
• Implement CI task for Dekaf integration tests
• Implement routing SessionAuthentication::Task sessions to a new/different backing store for migration purposes
• Figure out how to make shard statuses show green in the UI for Dekaf tasks

---

This change is 

[jshearer]

Alright, I could probably do another few rounds of self-review here, but I think it's at a good enough spot to mark this as ready for review. Let me know what y'all think!

[jgraettinger]

Sending comments from a partial review. I haven't finished.

[jgraettinger]

Okay, I've finished a pass.

As a meta comment, for big features like this please either:
a) Open multiple smaller PRs which can be reviewed independently, or

b) Maintain ever-green commits which are re-written for the reader: they should tell more bite-size stories about changes being made, stand on their own, and be feasible to review (or re-review) independently. This is the workflow @psFried and I use. In essence each commit is a mini-PR -- it requires that you regularly rebase/squash/fixup as you develop.

[jgraettinger]

As a general comment, I'd like to see this kind of extraction and swizzling pulled out into pure functions over input data structures, returning a new output data structure (or error, if validation fails). I'll point to the various Task structs in the runtime crate as an example of this.

This makes them test-able without postgrest in the loop. Recently, it became possible to build catalog specs in Rust as part of tests, which would let you exercise your logic with "real" tasks. (An example: https://github.com/estuary/flow/blob/master/crates/activate/src/lib.rs#L858-L875)

Big picture, the goal is to roughly separate routines into "glue" (connecting services, moving data, calling things) which are easy to inspect because they obviously succeed or obviously fail, and "business logic" (extract / validate / swizzle / build), which are easy to test if they're pure functions.

[jshearer]

Okay, I think this is ready for another look. I changed around the field selection to be closer to your suggested pattern of building up a list of extractors and their associated avro schemas. The realization that avro documents can be encoded sequentially allowed me to avoid building up the full HeapNode before encoding, which should reduce memory usage a good bit.

FWIW though, I ended up generating a (avro::Schema, Vec<doc::Extractor>) in build_field_extractors(), mainly because I need the full schema in Collection in order to save it to the database and get a content-addressed ID. I then extract out the field schemas and zip them with the extractors in Read. I suppose I could instead avoid building the full schema in build_field_extractors and just build it when dealing with schema registry stuff... this feels like six of one, half dozen of the other IMO.

Recently, it became possible to build catalog specs in Rust as part of tests, which would let you exercise your logic with "real" tasks.

I'll take a look at this and come up with a good way to unit test build_field_extractors(). As it stands now, are you still wanting more separation between "glue" and "business logic"? If so, could you try to articulate where to draw that imaginary line? Usually when I think about this sort of thing I've been thinking about what logic can be written in a more reusable/more broadly applicable way (i.e what belongs in an "externally" exposed interface vs what should remain internal, etc), but it sounds like you're talking more about distinguishing between things that can fail vs things that can't?

For example, I extracted the log forwarding stuff into its own module which roughly exposes an interface for how to use it (with the rough edges getting refined over time, as the things it needs to do becomes more clear), so that's all you should have to think about when you want to forward logs -- this should probably eventually be moved to its own crate, especially if anything else wants to use it as well later on. But this doesn't sound to me like the same kind of separation you're asking for.

As far as your meta-comments, like I mentioned over Slack, I certainly have been opening PRs along the way, though granted I could have been doing even more of that.

And as far as evergreen commits, that is indeed always my goal, and with this round of changes I've done some rewriting and condensing of the commits to make them more self-contained. I do have a question about that though: when you're at the point of pushing your work up to a PR, do you actually just squash everything into one set of changes, and then chunk it out from there into separate "mini-PR" commits? Because I've been trying to work with the evergreen commits the whole time, and if I need to make more-than-trivial changes, I find myself spending more time than I'd like on figuring out exactly which chunks of changes to send to which commit. Thoughts?