projectriff-archive / function-sidecar Goto Github PK

View Code? Open in Web Editor NEW

3.0 3.0 6.0 21.52 MB

riff function sidecar written in Go

License: Apache License 2.0

Go 97.44% Makefile 2.56%

function-sidecar's People

Contributors

Stargazers

Watchers

Forkers

jamieklassen markfisher freynca ericbottard imikushin samaitra

function-sidecar's Issues

investigate / spike streaming http dispatcher

consider

http2
gRPC
websocket
others?

gRPC everywhere

umbrella issue to track support for gRPC across invokers

Add easy delve setup for running inside k8s

messages sent from KafkaTemplate are missing embedded headers

the FunctionController sends messages to the "function-replicas" topic from within FunctionDeployer like so:

public <T> void publish(String topic, Object event) {
	try {
		byte[] payload = this.mapper.writeValueAsString(event).getBytes(StandardCharsets.UTF_8.name());
		Message<byte[]> message = MessageBuilder.withPayload(payload)
				.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
				.build();
		byte[] bytes = EmbeddedHeaderUtils.embedHeaders(new MessageValues(message)); 
		this.kafkaTemplate.send(topic, bytes);
	}
	catch (Exception e) {
		logger.warn("failed to publish event", e);;
	}
}

But when the message is deserialized in the function-sidecar no headers are present:

[redis-writer-1800636090-t9rzc sidecar] 2017/12/02 01:26:00 >>> Message{{"square":1}, map[]}
[redis-writer-1800636090-t9rzc sidecar] 2017/12/02 01:26:00 Wrapper received Message{{"square":1}, map[]}

sidecar should wait for function container to be available

I'm seeing this error in the logs when the sidecar tries to dispatch messages to the function container too soon:

[greeter-1890857377-xnl8k sidecar] 2017/11/11 14:23:36 Error invoking http://localhost:8080: Post http://localhost:8080: dial tcp [::1]:8080: getsockopt: connection refused

This is easy to reproduce with the java-function-invoker.

We should either find a way to be notified about the endpoint availability (potentially by the Function Container itself?) or at least have a retry with an aggressive loop (we want to start dispatching as soon as possible).

Change gRPC dispatcher to use byte[]+headers

See projectriff-archive/function-proto#5

Sidecar panics if function invoker not available

Not sure if this is an issue, but it's easy to reproduce: just launch a sidecar and a function invoker on localhost, and then kill the invoker. The sidecar panics and never recovers. Seems like there might be a similar failure mode even in a healthy pod in k8s.

Implement streaming support

http dispatcher should not block

While working on projectriff/node-function-invoker#9 it appears that the http dispatcher is blocking, and not allowing concurrent invocations to happen in one instance.

This prevents async http functions from operating concurrently, forcing riff to scale out when async http functions are idle, waiting for io, rather than sending more requests to the same instance.

terminating container continues to accept input events

when the pod enters the terminating state we should be shutting down the consumer then completing in-flight processing, but after a kubectl delete function ... command continuing to send events to the function's input topic keeps it active as a consumer even while the pod state shows "terminating".

Better encapsulate dispatcher.Message

Also, change type of headers to map[string][]string ie MultiMap<String, String>

http dispatcher error processing events without a correlation id

Found this when trying to run S1P demo using redis-writer with http.
In this case the upstream java vote-stream-processor is emitting events every 2s.
Those events don't have a correlation id.

The sidecar eventually panics with the following error

default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 2018/02/19 20:48:45 Wrapper received Message{{"_list":"demo:votes-windows","_time":1519073323372}, map[timestamp:[1519073325372]]}
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 2018/02/19 20:48:47 Wrapper received Message{{"_list":"demo:votes-log","_time":1519073325367}, map[timestamp:[1519073327367]]}
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 2018/02/19 20:48:47 Wrapper received Message{{"_list":"demo:votes-windows","_time":1519073325372}, map[timestamp:[1519073327373]]}
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 2018/02/19 20:48:47 Error invoking http://localhost:8080: Post http://localhost:8080: net/http: request canceled (Client.Timeout exceeded while awaiting headers)
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 2018/02/19 20:48:47 Error calling synch dispatcher Post http://localhost:8080: net/http: request canceled (Client.Timeout exceeded while awaiting headers)
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 2018/02/19 20:48:47 Wrapper about to forward <nil>
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 2018/02/19 20:48:47 <<< %!s(<nil>)
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: panic: runtime error: invalid memory address or nil pointer dereference
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: [signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0x83ba6a]
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: goroutine 31 [running]:
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: github.com/projectriff/function-sidecar/vendor/github.com/projectriff/message-transport/pkg/transport/kafka.encodeMessage(0x0, 0x0, 0xc4204be240, 0xc4205da000, 0x100, 0x23, 0xc63120)
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 	/go/src/github.com/projectriff/function-sidecar/vendor/github.com/projectriff/message-transport/pkg/transport/kafka/wireformat.go:79 +0x3a
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: github.com/projectriff/function-sidecar/vendor/github.com/projectriff/message-transport/pkg/transport/kafka.toKafka(0x0, 0x0, 0xc420282fd0, 0xf, 0x0)
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 	/go/src/github.com/projectriff/function-sidecar/vendor/github.com/projectriff/message-transport/pkg/transport/kafka/wireformat.go:36 +0x39
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: github.com/projectriff/function-sidecar/vendor/github.com/projectriff/message-transport/pkg/transport/kafka.(*producer).Send(0xc420144960, 0x7ffc71886789, 0x5, 0x0, 0x0, 0x0, 0x0)
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 	/go/src/github.com/projectriff/function-sidecar/vendor/github.com/projectriff/message-transport/pkg/transport/kafka/producer.go:35 +0x39
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: github.com/projectriff/function-sidecar/pkg/carrier.Run.func2(0xc19400, 0xc420166540, 0x7ffc71886789, 0x5, 0xc19540, 0xc420144960)
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 	/go/src/github.com/projectriff/function-sidecar/pkg/carrier/carrier.go:48 +0x14b
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: created by github.com/projectriff/function-sidecar/pkg/carrier.Run
default/redis-writer-56fb49c4d9-qdbc2[sidecar]: 	/go/src/github.com/projectriff/function-sidecar/pkg/carrier/carrier.go:41 +0xbb

switching the redis-writer function to use grpc instead of http fixes the issue.

default/redis-writer-6ff59f6fcc-t6xzr[sidecar]: 2018/02/19 20:52:06 Sidecar for function 'redis-writer' (function-replicas->dummy) using grpc dispatcher starting
default/redis-writer-6ff59f6fcc-t6xzr[sidecar]: 2018/02/19 20:52:06 Rebalanced: &{Type:rebalance start Claimed:map[] Released:map[] Current:map[]}
default/redis-writer-6ff59f6fcc-t6xzr[sidecar]: 2018/02/19 20:52:06 Rebalanced: &{Type:rebalance OK Claimed:map[function-replicas:[0]] Released:map[] Current:map[function-replicas:[0]]}
default/redis-writer-6ff59f6fcc-t6xzr[main]: Node started in 117ms
default/redis-writer-6ff59f6fcc-t6xzr[main]: gRPC loaded in 97ms
default/redis-writer-6ff59f6fcc-t6xzr[main]: HTTP loaded in 68ms
default/redis-writer-6ff59f6fcc-t6xzr[main]: gRPC running on localhost:10382
default/redis-writer-6ff59f6fcc-t6xzr[main]: HTTP running on http://localhost:8080
default/redis-writer-6ff59f6fcc-t6xzr[main]: Function invoker started in 287ms
default/redis-writer-6ff59f6fcc-t6xzr[main]: RIFF 1: Result: <Buffer 7b 22 5f 6c 69 73 74 22 3a 22 64 65 6d 6f 3a 76 6f 74 65 73 2d 6c 6f 67 22 2c 22 5f 74 69 6d 65 22 3a 31 35 31 39 30 37 33 35 32 33 33 36 37 7d>
default/redis-writer-6ff59f6fcc-t6xzr[main]: RIFF 1: Result: <Buffer 7b 22 5f 6c 69 73 74 22 3a 22 64 65 6d 6f 3a 76 6f 74 65 73 2d 77 69 6e 64 6f 77 73 22 2c 22 5f 74 69 6d 65 22 3a 31 35 31 39 30 37 33 35 32 33 33 37 ... >
default/redis-writer-6ff59f6fcc-t6xzr[sidecar]: 2018/02/19 20:52:07 <<< Message{{"_list":"demo:votes-log","_time":1519073523367}, map[Content-Type:[text/plain]]}
default/redis-writer-6ff59f6fcc-t6xzr[sidecar]: 2018/02/19 20:52:07 <<< Message{{"_list":"demo:votes-windows","_time":1519073523373}, map[Content-Type:[text/plain]]}
default/redis-writer-6ff59f6fcc-t6xzr[main]: RIFF 1: Result: <Buffer 7b 22 5f 6c 69 73 74 22 3a 22 64 65 6d 6f 3a 76 6f 74 65 73 2d 6c 6f 67 22 2c 22 5f 74 69 6d 65 22 3a 31 35 31 39 30 37 33 35 32 35 33 36 37 7d>
default/redis-writer-6ff59f6fcc-t6xzr[sidecar]: 2018/02/19 20:52:07 <<< Message{{"_list":"demo:votes-log","_time":1519073525367}, map[Content-Type:[text/plain]]}

Retry limit for function errors?

When the sidecar sends a gRPC stream it fails with a continuous stream of logs (apparently infinite):

Error receiving message from function: rpc error: code = Unknown desc =
Error receiving message from function: rpc error: code = Unknown desc =
Error receiving message from function: rpc error: code = Unknown desc =
...

The function invoker is failing (but only logging it once, strangely), but the sidecar goes on spitting out empty failure logs forever, and really it has no hope of ever succeeding.

Change dispatcher.Message.Payload to []byte

This is what is used in practice and is not exposed to the outside world

Implement zipkin integration

Propagate Content-Type when using http.

Dispatcher contract should have access to payload + headers, whatever the way they were encoded

Change the way the sidecar gets its config

Currently inherited from Spring Cloud Stream

document invoker contract

With gRPC in all the invokers, document how it works and find the gaps/differences across languages.

error handling
content type negotiation
header propagation

Allow health probes

Multi stage docker build

Correctly handle the no-output case

Forward all headers from sidecar to invoker and back

In the same way that we filter headers from the incoming message that we send to the dispatcher, we should limit headers on the way back (Content-Type seems like the only legitimate candidate for now)

This avoids exposing internal details (correlationId) and/or wrong values (eg http-gw replying with Served-By stating Express (node invoker))

publish event metrics

this is a prerequisite for https://github.com/projectriff/function-controller/issues/29 and projectriff/riff#220

this will allow a listener (such as the function-controller) to keep track of producer and consumer throughput

should also publish consumer lifecycle events (when the actual event consumer is activated/deactivated) so that function-controller scaling can be more accurate

2018/02/07 12:55:13 Error receiving message from function: rpc error: code = Unavailable desc = transport is closing

What's strange is that I manage to see between zero and seven instances of this message for one invocation of make test. The tests still pass.