Wasm engine - community feedback thread! <3 🚀

[vuldin]

Is there a way to pass a desired child topic name to the wasm transform engine? In the example , market_activity topic is created by the user and then the wasm engine automatically creates a market_activity._result_ topic. It would be great to have another argument for topic name when deploying the wasm module:

rpk wasm deploy dist/main.js --name json2avro --topicName "avro-market-activity" --description "Transforms JSON to AVRO" --brokers localhost:19092

Another option would be to name the topic after the wasm module name (json2avro above).

[emaxerrno]

cc : @graphcareful @VadimPlh @vsaraswat @dswang

@vuldin this is good idea. What other metadata should we allow from the cli, maybe offset management too?

[vuldin]

cc : @graphcareful @VadimPlh @vsaraswat @dswang

@vuldin this is good idea. What other metadata should we allow from the cli, maybe offset management too?

Ahh yeah, allowing the user to set the child topic's default offset would be great. I can think of scenarios where consumers would only want to get events on the child topic from now (for instance). Two useful options for this on kafka-consumer-groups.sh are --to-datetime and --to-latest.

[graphcareful]

so the output topic name is result in the example code because that is the sole key in the dictionary of the return values from the apply method. The wasm engine itself tacks on the interesting syntax you see there so upon invocation of a list topics command you can easily denote which ones were created by coprocessors, for debugging capability.

https://github.com/vectorizedio/redpanda-examples/blob/main/wasm/js/transform_avro/src/main.js#L59

[vuldin]

So it is customizable to a certain extent! Ok I'll look into this more and make sure I understand how it works. Thanks @graphcareful !

[vuldin]

I'm not sure if this has changed recently, but it looks like it doesn't work as described.

I'm subscribed to topic market_activity:

transform.subscribe([["market_activity", PolicyInjection.Latest]]);

If I call the following:

result.set("avro", transformedRecord);

I would expect to see a new topic market_activity._avro_. But instead I see the following:

> rpk --brokers localhost:19092 topic list                            
NAME                        PARTITIONS  REPLICAS
_schemas                    1           1
coprocessor_internal_topic  1           1
market_activity             1           1
market_activity._result_    1           1

So it seems _result_ is hard-coded outside of the transform function. I'm using the following redpanda docker version:

vectorized/redpanda-nightly:v0.0.0-20220318git2bf8159

[VadimPlh]

Hi!
We use different output names inside our tests. They passed.
Maybe you did not recompile and redeploy new transform function after your changes?

[vuldin]

Hi! We use different output names inside our tests. They passed. Maybe you did not recompile and redeployed new transform function after your changes?

That's exactly what happened... I have now gotten into the habit of running rm -r dist && npm run build.

[graphcareful]

To me this discussion brings up good points about what configuration options should be in the source code vs what should not. A really good idea that I've also been thinking about for some time now.

For example currently, input topics, output topics, offset policies and error policies are all options statically defined in source, which mean that if you wanted to keep the same apply method, but just change metadata, you'd have to write a completely new coprocessor and re-deploy.

However, if everything was defined as arguments via the CLI I thought of one negative which would be, if you coprocessor does something like this:

function apply(topic, batch) {
  if(topic == "market_activity") {
    ...
  }
...
}

This source only makes sense if market_activity is part of a list of inputs. Kind of lending to the idea that at least inputs should be statically defined in source. I know there are stream processors that accept all inputs via cli as metadata and some that don't.

[jrkinley]

@graphcareful what about moving these options to topic metadata (first step towards data policies) and set them via rpk.

[graphcareful]

I mentioned a negative just to play devils advocate and list pros/cons, but I am actually leaning towards this new design. It would also lend nicely to data policies too. Like the same code could run in async or sync mode, no need to push some slightly modified new transform up.

Then I would assume a workflow would go like this:

$ rpk wasm push <code> --xform_name <name>
$ rpk wasm enable --xform_name <name> --mode async --inputs foo,bar --outputs baz

So pushing code pretty much just adds the transform to the internal topic and tacks on identifying metadata such as transform name to the payload. Actually running the enable command is what will start up the coprocessor. This is also similiar to what @VadimPlh has done with data policies. He added a flag which when passed to rpk wasm deploy doesn't actually deploy the coprocessor, so data policies can later look for it in the internal coprocessor topic.

[emaxerrno]

good ideas.

[emaxerrno]

Let's wait to collect more feedback from the community.

[graphcareful]

#1779

[vuldin]

It would be great to have rpk wasm list, which just lists out the transforms.

Right now there is just the "deploy" action, which implies there is just a deploy state. But soon there could be other states: active/inactive (or enabled/disabled). If a transform is active then it is also associated with both an input topic and an output topic, and these details could also be provided with rpk wasm list.

The output could be something like the following:

> rpk wasm list
NAME          STATUS      INPUT_TOPIC      OUTPUT_TOPIC
json2avro     enabled     market_activity  market_activity._avro_
json2proto    enabled     market_activity  market_activity._proto_
transform42   disabled    N/A              N/A

[vuldin]

A minor change would be to allow multiple output topics for a transform. This would be beneficial when you need to do multiple transforms (ie. to both avro and protobuf) for each incoming message. The function passed to transform.processRecord() could look like this:

transform.processRecord((batch) => {
  const result = new Map();
  const avroTransformedBatch = batch.map(({ header, records }) => {
    return {
      header,
      records: records.map(toAvro),
    };
  const protoTransformedBatch = batch.map(({ header, records }) => {
    return {
      header,
      records: records.map(toProto),
    };
  });
  result.set("avro", avroTransformedBatch);
  result.set("proto", protoTransformedBatch);
  // processRecord function returns a Promise
  return Promise.resolve(result);
});

This can be accomplished now with multiple transforms, but then you need to deploy multiple transforms, each transform needs to separately read each message on the same topic, and records need to be batched for both transforms.

[graphcareful]

+1 on rpk wasm list its on our short list of things to do. For allowing multiple output topics, this is actually already supported, and is mentioned in our docs here. That's why the result type of this method is a map. Its so you can define which records are to be produced onto one-or-more output topics.

[vuldin]

See this issue for details on what I'm running into around this discussion.

[vuldin]

The transform object (returned from SimpleTransform) should have the following functions: encode and decode. The API could be something like the following (using avsc Avro library as an example below):

const avro = require("avsc");
const { SimpleTransform } = require("@vectorizedio/wasm-api");

const schema = avro.Type.forSchema({
  name: "market_activity",
  type: "record",
  fields: [
    {name: "Date", type: "string"},
    {name: "CloseLast", type: "string"},
    {name: "Volume", type: "string"},
    {name: "Open", type: "string"},
    {name: "High", type: "string"},
    {name: "Low", type: "string"}
  ]
});

const transform = new SimpleTransform({
  encode: schema.toBuffer,
  decode: schema.fromBuffer,
});

const buffer = transform.encode(someString);
const newString = transform.decode(buffer);

Schema handling libraries are different from each other, but they each serialize/deserialize strings/buffers. One issue with this is that these libraries call these functions different names, and so we need to know implementation details when it comes to tests. In the future, chaining these transforms may be a cool feature (which would also be more difficult without this or a similar change).

[vuldin]

This unfortunately may only be possible by including these libraries (avsc, protobufjs, others) as a dependency. I'm still looking into this.

[graphcareful]

This is very cool stuff, would also be cool to integrate this with our own schema registry. To explain why it works like this today, at the time of development, we decided that the API between redpanda and the wasm-engine would be raw record batches, then the user is free to do whatever they'd like with the contents. Admittedly this adds additional complexity on the user side, but modifying this should be fairly easy, contained within the source of the wasm-engine project.

[vuldin]

After testing asynchronous code within processRecords, I'm fairly sure at this point that the wasm security is blocking things I'm trying to do, such as read a file or perform a fetch request. I think the WASM security needs to allow fetching from at least the container's schema registry port by default.

In the future maybe WASM security could be customizable by the module author, or even better through redpanda.yaml (ie. the container admin) for security reasons. This would allow users to enable fetching from their one registry nodes or from whatever custom resource they need in their transform.

[vuldin]

I don't know where the git repo is for @vectorizedio/wasm-api to look into this further. One possible way around this security blocker is to perform the fetch at build time, and then bundle the result into the wasm module. The same thing could be done with reading files or other storage. This would mean the module wouldn't be responsive to updated schemas (for instance) but that may be fine for most use cases.

[vuldin]

This worked! I switched from webpack to esbuild and wrote a plugin that pulls the schema from registry at build time. I also added a test to make sure the transform function doesn't try to use the network.

Here is the branch: https://github.com/vuldin/redpanda-examples/tree/integrate-registry-into-wasm

It's based off my previous branch associated with this PR: redpanda-data/redpanda-examples#12

Once the above PR gets approved/merged I will create a new PR for the integrate-registry-into-wasm branch.

[graphcareful]

Cool glad you got past that! To add from my end,

The reason you might have encountered strange behavior w.r.t. asynchronous work done in processRecords is because the coprocessor API is single threaded by design. Your coprocessor will consume records from a kafka topic/partition in-order, and it will not move forward (meaning grab more recordBatches and call processRecords again) until the previous value returned from processRecords has been acknowledged as written or consumed by the redpanda/coproc logic.

[vuldin]

The output from const transform = new SimpleTransform() is the following object:

u {
  apply: [Function (anonymous)],
  inputTopics: [ [ 'market_activity', 2 ] ],
  policyError: 0
}

Why is inputTopics an array? I thought transforms can only read from one topic since attempting to join across multiple topics would have to take into account partitions and the broker where it is located.

[vuldin]

This object could be something more like the following:

{
  inputTopic: 'market_activity',
  outputTopic: 'avro',
  transformFunction: [Function: transformFunction], // implemented in main.js, passed to SimpleTransform
  options: { injectionPolicy: 'latest', onError: 'skip' }
}

[graphcareful]

Hi @vuldin ! I was OOO, to answer, the field inputTopics is an array because transform may read from multiple topics. Each coprocessor is virtually sharded and is responsible for a subset of partitions, this is all transparent to the user. There is nothing to join since separate coprocessing units are responsible for their own partitions or slices of the topic.

[vuldin]

Here is an attempt at a simpler API for the SimpleTransform function. I have some comments in the file help explain some things, and also pose some questions I have.

https://github.com/vuldin/redpanda-examples/blob/integrate-registry-into-wasm/wasm/js/transform_avro/src/main.mjs

[awto]

I've made a Kafka Streams based workflow-as-code engine temporal.io style - https://github.com/awto/kafka-workflow, and want to port it to Redpanda wasm engine. It works much faster with Kafka Streams + Redpanda, but I expect it will be even better with the WASM engine, which is necessary for big data processing workflows.

Usual workflow engines use some graphical or declarative text formats (e.g. BPM), encoded in YAML or XML, etc. On the other hand, Workflow-as-code uses plain generic programming language (JavaScript in my implementation, but it can be anything compiled to WASM). This makes workflow definitions significantly easier to develop and maintain.

Unlike Temporal, kafka-workflow is a very tiny Kafka streams process. So all the company infrastructure can be kept if it is already based on some event broker. There are also other benefits.

The tool needs little from the broker, namely either a local state or a join for the local state change log and event stream. The system inherits all the liveness safety properties from the broker.

So do you think it is a fair and achievable use case for the wasm engine? Maybe you may have some API hints for the implementation?

[vuldin]

I looked at your project and see that you are using GraalVM, that's awesome @awto . I need to spend more time with the project to understand everything. As far as porting to the Redpanda wasm transform engine, that would be great. It's still very early with this implementation, so if you do start working with it and see how it could be modified to make it's use easier or more capable then post updates and questions here.

[intech]

@vuldin Hello! Could you implement more flexible settings for the outgoing topic so that it is not a template like source._destination_, but you can specify the full name of the topic?
One of the application patterns for me is the processing of states from one stream and distribution to other existing streams.

[vuldin]

I agree, being able to set the output topic name would be a helpful feature. I think the main reason it was implemented like it is now was to also provide a simple way to track which topics were generated by the Wasm co-processor, but there are other ways to do this that leaves the topic names completely up to the user. This was just the first iteration.

[graphcareful]

Yes, another reason was to explicitly prevent re-producing to input topics.

[intech]

@vuldin Hello!
What do you think about wasmtime?
ScyllaDB recently announced UDF support on WASM via wasmtime. Explained it with integration features with Seastar, which is also based RedPanda.

[vuldin]

Apologies @intech I am just catching up with some messages. I think Wasmtime is awesome, and I think it definitely helps that ScyllaDB announced user-defined functions via this tool. I think the use case is similar, in that we would want to import UDFs that match a definition (use the correct inputs, make available the correct outputs).

I'm really interested in Wasm, and while it hasn't gotten much more additional attention since I've joined that's sure to change in the near future. I expect that will mean some fairly big changes with how Wasm is implemented at the moment for data transforms. I'll look more into how ScyllaDB makes use of Wasmtime.

[intech]

Thank you for your hard work, and I appreciate it!

[mattschumpert]

@rockwotj