yomorun/yomo

🦖 Stateful Serverless Framework for Geo-distributed Edge AI Infra. with function calling support, write once, run on any model.

yomo's Issues

`/docs/sink` for how to write a `yomo-sink`

Streaming-serverless arch branch for community feedbacks

Feedbacks from the community, developers consider writing Serverless functions rather than implementing & maintaining microservices on the edge. Consider build a version on feat/sl branch for this.

Simplify QUIC client & server

In order to decrease the code complexity, we need to remove the customized QUIC interfaces in core/quic, and directly use quic-go as QUIC client & server.
See skeleton project for reference.

high cpu

Summary (概要)

After running yomo dev for 2 minutes, the cpu is 100%

One QUIC stream between zipper and sfn

Currently, zipper and sfn creates individual QUIC streams for each data, we need to replace it with one persistent QUIC stream.

The stream.Write() should also add lock to prevent concurrent writing in the same QUIC stream.

Separate CLI and Framework

By getting rid of Go plugins, is friendly to developers:

• never need clone whole project to $GOPATH
• Windows developers can use yomo dev xxx.go

Replace logger from skeleton project

It's inconvenient to use the current logger, we need to replace it with the one from skeleton project.

Adding a description of contributing

Add contributing file about yomo project

failed to sufficiently increase receive buffer size when `yomo run` on Manjaro

Summary

Run yomo dev on Manjaro, failed to sufficiently increase receive buffer size error.

Steps To Reproduce

~/.../yomo/example >>> ../bin/yomo dev app.go
2020/12/21 13:01:37 Building the Serverless Function File...
2020/12/21 13:01:42 failed to sufficiently increase receive buffer size (was: 208 kiB, wanted: 2048 kiB, got: 416 kiB). See https://github.com/lucas-clemente/quic-go/wiki/UDP-Receive-Buffer-Size for details.
2020/12/21 13:01:42 QUIC Server listens on 0.0.0.0:4242
^C

Environments

~/.../yomo/example >>> cat /etc/lsb-release
DISTRIB_ID=ManjaroLinux
DISTRIB_RELEASE=20.2
DISTRIB_CODENAME=Nibia
DISTRIB_DESCRIPTION="Manjaro Linux"

Solution

Run sudo sysctl -w net.core.rmem_max=2500000 to solve this problem followed by quic-go wiki

Count the amount of `data frame` in zipper

Count the amount of data frame when zipper receiving the data frame from source, and record it in a local file.

`/docs` show the index page of docs

update codec to v1.0.0

• optimising performance
• sticky packet handling

Support `yomo version` to get current yomo version

`/docs/tutorial` for `source->flow->sink(faunaDB)`

Use 0-RTT in QUIC client

on 170ca40

Remove Rx in zipper

Currently, zipper is using Rx to dispatch the data from source to stream-functions, we need to remove Rx in zipper and use a clean way to dispatch the data.

Refactor dev server for `yomo dev`, support yomo version matching

When developers write Rx code, logic can be implemented easily and quickly, when they run yomo dev, they get test data streams directly, especially for user onboarding. after v2.0, we consider provide a way for custom data types.

use data frame between zipper and sfn

Currently, zipper and sfn are communicating with raw bytes, we need to replace it with Data Frame.

RxStream is better than raw QUICStream when writing streaming-processing codes.

Streaming is not like HTTP's Request/Response, by introducing ReactiveX, developers could write concise codes.

Rewrite Official Website

Consider rewrite YoMo official website by Nextra.

• i18n support, English and Chinese at first

Allow to set the quic-go debug level in CLI

In order to debug the QUIC transport information, we need to set QUIC_GO_LOG_LEVEL environment variable in CLI.

The stream should be immutable, `stream function` can only append new data to stream

• The stream from yomo-zipper should be immutable, the stream functions shouldn't change the raw stream.
  In case the upstream function needs to generate new data for downstream, it can only append the new data to the stream.
• Combine flow and sink as stream function, it will be executed one by one by zipper.

Print `UDP Receive Buffer Size` in CLI

When running the CLI, we need to print UDP Receive Buffer Size in the current OS and validate if it doesn't fit the recommended size.

CLI can lower down the difficulty when writing YoMo applications

also from the community, related to #58 , developers can save more time when they can yomo run app.go

Integrate steam-zipper to CLI

also from the community, related to #58. Developers can chain-up their yomo-flows by yomo run workflow.yaml

The absence of `yomo-function` configuration section can be accepted by `yomo-server`

As YoMo Server can be started by the -m parameter for loading a remote configuration, a YoMo Server w/o functions section should work.

When functions section present, only listed yomo-functions can be connected to.

[BUG] plugin different version of package

Summary (概要)

when I use cmd "yomo dev", I got some problem

Steps To Reproduce (复现的步骤)

1.mkdir demo
2.open my IDE goland to open my demo project
3.go mod tidy
4.yomo dev

Additional Details & Logs (附加的信息和日志)

• Yomo version (版本)1.0.0-3gf175e41 (2021-03-10)
• Yomo error logs (错误日志)Load handle from .so file failure with error: plugin.Open("sl"): plugin was built with a different version of package github.com/yomorun/yomo/pkg/yy3
• Yomo configuration (配置) I got the issue when I enter the cmd "yomo dev",so there`s probably is yomo init config
• Operating system (操作系统)MacOS Catalina v10.15.7

[BUG] CopyN 修改参数配置可引起消息阻塞

Summary (概要)

流处理使用io.CopyN，当设置较少的拷贝字节数时，会出现消息阻塞和消息切断的现象。

Steps To Reproduce (复现的步骤)

1. Prj/Yomo >> run.go >> RunDev >> 修改io.CopyN参数值:
   io.CopyN(pluginStream.Writer, deStream1.Reader, 64)

2. Prj/yomo-echo-plugin >> go run echo.go, 出现消息阻塞和消息切断：
   name:yomo!-new
   name:yom

3. 设置的值越小则情况越严重，甚至一开始就出现阻塞

4. 后续测试: 默认1024的值也会出现消息阻塞和消息切断，应该是处理消息量达到一定程度就会出现该现象。

Additional Details & Logs (附加的信息和日志)

• Yomo version (版本): v0.1.0
• Yomo error logs (错误日志): 无
• Yomo configuration (配置): io.CopyN(pluginStream.Writer, deStream1.Reader, 64)
• Operating system (操作系统) macOS

[BUG] Multiple streams between zipper and flow

In some cases, when the connection was interrupted between yomo-zipper and yomo-flow, a new stream will be created. And currently yomo-zipper keeps multiple streams for all yomo-flow, yomo-sink and yomo-source.
It causes a performance issue when there are too many streams between yomo-zipper and yomo-flow/sink.

But actually, yomo-zipper should only support multiple streams for yomo-source, and one stream for yomo-flow/ yomo-sink.

`/docs/source` for how to write a `yomo-source`

Rename source/flow/sink

It is difficult and confused for users to understand. the source/flow/sink concepts in YoMo.
It's better to simplify the naming of source/flow/sink/zipper.

Connectors

Connectors provides the common and reusable services for input/output data in YoMo.
The users can import the YoMo SDK to implement the input/output connector and add the main() function. Then you can also run the connectors by Go cli, f.e. go run main.go.

Input Connector

Input connector sends the input data from the external system (f.e. IoT Sensors) to yomo.

• Rename source to input connector.

Output Connector

Output connector sends the output data from yomo to an external system (f.e. database).

• Rename sink to output connector.

Stream Functions

Stream Functions are the streaming Serverless functions in YoMo, it contains a Handler function which the users can write the business logic there.
The users can use yomo cli to run the function locally or deploy it to YCloud.

• Rename flow to stream function.

YoMo Server

YoMo Server is an orchestrator which receives the data from sources (input connectors) and routes the data to Stream Functions one by one. The result of real-time stream computing will be sent to Output Connectors.

• Rename zipper to yomo server.

Remove Rx in sfn

There are two scenarios in sfn:

1. All data are independent, the users will handle the raw data immediately when the sfn receives data from zipper. In this case, we don't need to use Rx.
2. The data is sequential, the users might need to handle a collection of data (f.e. the average value in a time window). In this case, it is suitable to use Rx.

We need to remove Rx in sfn and use a different way to support both raw bytes and rx stream.

zip 3 or more streams

I thought it would be straightforward, but I'm facing some difficulties...

I have a data source app that sends 5 packets encoded with 5 different keys:

func generateAndSendData(stream quic.Stream) {
    keys := []byte{0x10, 0x11, 0x13, 0x7E, 0x7F}

    for i, key := range keys {
        time.Sleep(100 * time.Millisecond)
        
        codec := y3.NewCodec(key)

        sendingBuf, _ := codec.Marshal(int64(i))

        _, err := stream.Write(sendingBuf)
        if err != nil {
            log.Printf("Couldn't send buffer with i=%v", i)
        } else {
            fmt.Print(".")
        }
    }
}

And the flow app that has 5 streams that are subscribed to each individual key and then zipped together:

var zipper = func(_ context.Context, a interface{}, b interface{}) (interface{}, error) {
    accumulator, ok := a.([]interface{})
    if !ok {
        fmt.Printf("No accumulator: %v + %v\n", a, b)
        return []interface{}{a, b}, nil
    }

    fmt.Printf("With accumulator: %v + %v\n", accumulator, b)
    accumulator = append(accumulator, b)
    return accumulator, nil
}

var convert = func(v []byte) (interface{}, error) {
    fmt.Printf("Got: %v\n", v)
	return y3.ToInt64(v)
}

// Handler will handle data in Rx way
func Handler(rxstream rx.RxStream) rx.RxStream {
	streamA:= rxstream.Subscribe(0x10).OnObserve(convert)
	streamB:= rxstream.Subscribe(0x11).OnObserve(convert)
	streamC:= rxstream.Subscribe(0x13).OnObserve(convert)
	streamD:= rxstream.Subscribe(0x7E).OnObserve(convert)
	streamE:= rxstream.Subscribe(0x7F).OnObserve(convert)

    return streamA.
        ZipFromIterable(streamB, zipper).
        ZipFromIterable(streamC, zipper).
        ZipFromIterable(streamD, zipper).
        ZipFromIterable(streamE, zipper).
        StdOut().
	Encode(0x11)
}

The problem, is that only the first two packets sent by the source are received on the flow. The order of zipping doesn't matter, the order of sending by the source matters only, for what I can see...

What can I do? (Also, I'm zipping sequentially, because there's no Merge operator... Is it really missing?)

Originally posted by @kuredoro in #158

Support basic JSON codec support

on 9a46852: for WebSocket / Crawler scenario, applications always running on JSON codec, implement basic support and give examples for users get there easily.

`/docs/rx` for `rxstream` methods supported

update codec to v1.1.0

• Solve the problem of base type encoding without wrapper node

Could install YoMo by curl

As many CLI can do like this, developers love copy & past 😄

`/docs/flow` for how to write a streaming serverless application

cpu 100% after ran quick-start 3~5 mins later

Summary (概要)

quick start demo make cpu 100%

Steps To Reproduce (复现的步骤)

1. write quick-start project, use yomo.Run
2. QUIC_GO_LOG_LEVEL=debug go run echo.go
3. wait 3~5 mins
4. cpu goes to 100%

Additional Details & Logs (附加的信息和日志)

• Yomo version (版本)
  v0.1.1
• Yomo error logs (错误日志)
  no error
• Yomo configuration (配置)
• Operating system (操作系统)
  macOS Big Sur 11.0 Beta (20A4300b)

Simplify workflow.yaml

We will rename source/flow/sink in #173, and it's also not needed to place sinks in workflow.yaml.

• Remove sinks
• Rename flows to functions
• Connect output connectors to yomo and receive the output data in parallel.

name: Service
host: localhost
port: 9000
functions:
  - name: func1
  - name: func2

Y3 Codec Draft-02

Connectors

The input/output connectors in YoMo.

replace Codec with yomo-codec-golang

• using yomo-codec-golang
• RunDev support OutputFormatter: HexString/PacketPrinter/EchoData/ThermometerData

YoMo-Zipper communication

For edge-mesh feature, YoMo-Zipper will be geo-distributed and closer to end users, all YoMo-Zippers need to communicate with each other.

Each YoMo-Zipper is both a Sender and a Receiver:

• Sender: send the output stream to other yomp-zippers in edge-mesh network.
• Receiver: receive the input stream from other yomp-zippers.

Introduce frames to YoMo

• Frames will be the minimal element transferred between services built on YoMo, frames built on Y3 codec for fast decoding
• Introduce HandshakeFrame
• Introduce PayloadFrame with MetaFrame and DataFrame
• Introduce TransactionID to MetaFrame, improve traceability

Reusable Server-Sent Events connector for frontend

about Server-Sent Events: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events

[BUG]无法成功执行执行demo中echo.go

Summary (概要)

无法成功执行执行demo中echo.go

Steps To Reproduce (复现的步骤)

1.在$HOME/go/yomo中执行go mod init yomo
2.执行go get -u github.com/yomorun/yomo
3.按文档创建echo.go
4.执行go run echo.go

Additional Details & Logs (附加的信息和日志)

• Yomo version: latest
• Yomo error logs:

command-line-arguments

./echo.go:16:14: cannot use &EchoPlugin literal (type *EchoPlugin) as type plugin.YomoObjectPlugin in argument to yomo.RunDev:
*EchoPlugin does not implement plugin.YomoObjectPlugin (missing Mold method)

• Operating system: Debian 10

Add yomo to homebrew formula

Developers can install yomo by brew install yomo

Update documentation

Update the documentations on yomo.run to fit the new changes in #173 #174

Log the connected/disconnected information in zipper

When source, sfn, and upstream zipper are connected/disconnected to zipper, we need to log them in Info level.