new(sdk/plugins): add builtin event stream instance implementations for push and pull models

jasondellaluce commented 2 years ago

What type of PR is this?

/kind feature

Any specific area of the project related to this PR?

/area plugin-sdk

/area tests

What this PR does / why we need it:

This PR introduces two builtin implementations of the source.Instance interface of the sdk/plugins/source package. This interface is responsible of representing opened event sources and of producing event data through the NextBatch() method.

There are two built-in implementations provided, each implemented for its use case:

"Pull model": for use cases in which the event source can be implemented sequentially and the time required to generate a sequence of event is deterministic. This is implemented with a functional design, where the passed-in callback is expected to be non-suspensive and to return quickly.
"Push model": for use cases in with the event source can be suspensive and there is no time guarantee reguarding when an event gets produced. For instance, this applies for all event sources that generate events from webhook events. Given the event-driven nature of this use case, this is implemented by passing event data in the form of byte slices through a channel.

On top of this, in-memory implementations have been added for core SDK interfaces such as sdk.EventWriter and sdk.EventReader. This is now used to implement benchmarks and unit tests for the two push and pull implementations. This was a long due necessity in the SDK, and these new sdk.InMemory* types can be used for a variety of use cases (tests being the most obvious ones).

Which issue(s) this PR fixes:

Special notes for your reviewer:

After starting developing the first plugins, I came to the conclusion that implementing the NextBatch() is not trivial in the general case. There are many things to be aware of:

Return as fast as possible and should try to fill-up the event batch as much as possible
Listen for a timeout and return the batch in its current state once the timeout is expired
Conceive the case in which Close() is called before NextBatch() has returned. This can happen if the plugin framework receives signals such as CTRL+C from the user.
Keep memory allocations to the minimum
Keep returning EOF after returning it the first time

As such, I though that creating an officially-maintained version of it for few use cases would be beneficial for the community. As a matter of fact, most of the plugins under the falcosecurity/plugins repo could be refactored to use these newly-introduced prebuilts.

Obviously, the SDK leaves plugin developers the choice of implementing their own source.Instance type with custom NextBatch() implementations for advanced or specific use cases.

Does this PR introduce a user-facing change?:

new(sdk/plugins): add builtin event stream instance implementations for push and pull models

jasondellaluce commented 2 years ago

Looking forward to gather some feedback on this!

jasondellaluce commented 2 years ago

Here's the report of the benchmarks of these two implementations on my machine. The code is inside sdk/plugins/source/instance_test.go.

Terminology:

Empty: Generating events with no content
Fixed: Generating events with content of fixed size of 1KB
Random: Generating events with content of random size between 1KB and 1MB

The benchmarks generates 1024 events, with a batch size of 128 elements and a timeout of 30 milliseconds.

goos: linux
goarch: amd64
pkg: github.com/falcosecurity/plugin-sdk-go/pkg/sdk/plugins/source
cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
BenchmarkPullEmpty-8           14881         71888 ns/op           0 B/op          0 allocs/op
BenchmarkPushEmpty-8            2679        378497 ns/op           0 B/op          0 allocs/op
BenchmarkPullFixed-8             364       3170573 ns/op     2277739 B/op       5120 allocs/op
BenchmarkPushFixed-8             249       4100726 ns/op     2277947 B/op       5121 allocs/op
BenchmarkPullRandom-8              1    1599520398 ns/op    1931069360 B/op    14858 allocs/op
BenchmarkPushRandom-8              1    1688147566 ns/op    1995180088 B/op    14738 allocs/op
PASS
ok      github.com/falcosecurity/plugin-sdk-go/pkg/sdk/plugins/source   9.950s

Takeaways:

Both implementations require 0 allocations at default
The Pull implementation is more performant on small events
The two implementations have similar performance on events of generic/large size

jasondellaluce commented 2 years ago

The source example has been updated to use the "pull" prebuilt. See: https://github.com/falcosecurity/plugin-sdk-go/pull/50/files#diff-9c5325f5e8159881ad3abf0eab413b61070b1b51276cd878f764c130cf11ac20R84

func (m *MyPlugin) Open(params string) (source.Instance, error) {
    counter := 0
    pull := func(ctx context.Context, evt sdk.EventWriter) error {
        counter++
        if err := gob.NewEncoder(evt.Writer()).Encode(counter); err != nil {
            return err
        }
        evt.SetTimestamp(uint64(time.Now().UnixNano()))
        return nil
    }
    return source.NewPullInstance(pull)
}

poiana commented 2 years ago

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: jasondellaluce, leogr

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files: - ~~[OWNERS](https://github.com/falcosecurity/plugin-sdk-go/blob/main/OWNERS)~~ [jasondellaluce,leogr] Approvers can indicate their approval by writing `/approve` in a comment Approvers can cancel approval by writing `/approve cancel` in a comment

poiana commented 2 years ago

LGTM label has been added.

Git tree hash: e693df47fb194ec31f1b35d0be540fb7e7e16d07

falcosecurity / plugin-sdk-go

new(sdk/plugins): add builtin event stream instance implementations for push and pull models #50