plasma
Attention: This repository was transfererd from openfresh.
plasma is event push middleware by using gRPC stream.
Description
Plasma is middleware for sending event specialized for a stream. Plasma provides EventSource and gRPC Stream from the same endpoint.
Installation
This middleware requires Redis.
From Source
$ git clone git://github.com/openfresh/plasma.git $GOPATH/src/github.com/openfresh/plasma
$ cd $GOPATH/src/github.com/openfresh/plasma
$ make deps
$ make build
The binary is generated under the bin/
directory.
Using docker
You can also use the Docker image.
$ docker run -p 8080:8080 -p 50051:50051 -p 9999:9999 openfresh/plasma
Using docker-compose
You can use docker-compose for easy use without preparing Redis.
$ git clone git://github.com/openfresh/plasma.git $GOPATH/src/github.com/openfresh/plasma
$ cd $GOPATH/src/github.com/openfresh/plasma
$ docker-compose up -d
Usage Subscriber
Server Sent Events
Your can use SSE if you request with Accept:text-stream
header.
You request events that you want to subscribe to this endpoint. You can specify multiple events separated by commas.
The query name can be set with the EventQuery
environment variable๏ผ(default value of EventQuery is eventType
)๏ผ
Here is a simple example using [Yaffle / EventSource] (https://github.com/Yaffle/EventSource).
var source = new EventSource('//localhost:8080/?eventType=program:1234:views,program:1234:poll,program:1234:annotation');
source.addEventListener("open", function(e) {
console.log("open");
});
source.addEventListener("error", function(e) {
console.log("error");
});
source.addEventListener("message", function(e) {
console.log("message event: ", e.data);
});
The JSON schema of data returned from Plasma is as follows.
{
"$schema": "http://json-schema.org/draft-04/schema#",
"id": "/events",
"properties": {
"data": {
"id": "/events/data",
"type": "string"
},
"meta": {
"id": "/events/meta",
"properties": {
"type": {
"id": "/events/meta/type",
"type": "string"
}
},
"type": "object"
}
},
"type": "object"
}
If the DEBUG
environment variable is enabled, you can access the debug endpoint.
GET /debug
You can publish events to Redis from this endpoint. You need to enter valid JSON in EventData form.
gRPC Stream
You can subscribe to events using gRPC Stream.
The ProtocolBuffer file is here .
The following is a simple Go sample.
func main() {
conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
if err != nil {
log.Fatal(err)
}
defer conn.Close()
client := proto.NewStreamServiceClient(conn)
ctx := context.Background()
req := proto.Request{
Events: []*proto.EventType{
eventType("program:1234:poll"),
eventType("program:1234:views"),
},
}
ss, err := client.Events(ctx)
if err != nil {
log.Fatal(err)
}
// subscribe event
if err := ss.Send(&req); err != nil {
log.Fatal(err)
}
for {
resp, err := ss.Recv()
if err != nil {
log.Println(err)
continue
}
if resp == nil {
log.Println("payload is nil")
continue
}
fmt.Printf("Meta: %s\tData: %s\n", resp.EventType.Type, resp.Data)
}
}
unsubscribe
Events
request is stream. If you unsubscribe event, set empty event data.
req := proto.Request{
// empty events
Events: []*proto.EventType{},
}
ss, err := client.Events(ctx)
if err != nil {
log.Fatal(err)
}
// unsubscribe event
if err := ss.Send(&req); err != nil {
log.Fatal(err)
}
Usage Publisher
You publish events to the channel that Plasma subscribes according to the following JSON Schema.
{
"$schema": "http://json-schema.org/draft-04/schema#",
"id": "/events",
"properties": {
"data": {
"id": "/events/data",
"type": "string"
},
"meta": {
"id": "/events/meta",
"properties": {
"type": {
"id": "/events/meta/type",
"type": "string"
}
},
"type": "object"
}
},
"type": "object"
}
openfresh/plasma-go is a library that wraps publish an event to Redis.
HealthCheck
GET /hc
You can do a health check. Check the status of Redis. If there is a problem it returns 500, and if there is no problem it returns 200.
Metrics
GET /metrics/go
You can get golang metrics from this endpoint.
The following golang metrics can be taken.
name | type | desc |
---|---|---|
go_version | string | version of go |
go_os | string | os |
go_arch | string | arch |
cpu_num | int | number of cpus |
goroutine_num | int | number of goroutines |
gomaxprocs | int | number of operating system threads |
cgo_call_num | int64 | number of cgo calls |
memory_alloc | uint64 | bytes of allocated heap objects |
memory_total_alloc | uint64 | cumulative bytes allocated for heap objects |
memory_sys | uint64 | total bytes of memory |
memory_lookups | uint64 | number of pointer |
memory_mallocs | uint64 | cumulative count of heap objects allocated |
memory_frees | uint64 | cumulative count of heap objects freed |
memory_stack | uint64 | bytes in stack spans |
heap_alloc | uint64 | bytes of allocated heap objects |
heap_sys | uint64 | bytes of heap memory obtained |
heap_idle | uint64 | bytes in idle spans |
heap_inuse | uint64 | bytes in in-use spans |
heap_released | uint64 | bytes of physical memory returned to the OS |
heap_objects | uint64 | number of allocated heap objects |
gc_next | uint64 | target heap size of the next GC cycle |
gc_last | uint64 | time the last garbage collection finished, as, nanoseconds since 1970 |
gc_num | uint32 | number of completed GC cycles |
gc_per_second | float64 | GC per second |
gc_pause_per_second | float64 | GC pause per second |
gc_pause | []float64 | GC pause |
GET /metrics/plasma
You can get plasma metrics from this endpoint.
The following golang metrics can be taken.
name | type | desc |
---|---|---|
connections | int64 | number of connected all clients |
connections_sse | int64 | number of connected SSE sclients |
connections_grpc | int64 | number of connected gRPC sclients |
Config
name | type | desc | default | note |
---|---|---|---|---|
PLASMA_PORT | string | http(https) port number | 8080 | |
PLASMA_GRPC_PORT | string | gRPC port number | 50051 | |
PLASMA_METRICS_PORT | string | metrics port number | 9999 | |
PLASMA_PPROF_HOST | string | pprof host | 0.0.0.0 | |
PLASMA_PPROF_PORT | string | pprof port number | 6060 | |
PLASMA_DEBUG | bool | debug mode | false | |
PLASMA_ORIGIN | string | set to Access-Controll-Allow-Origin | ||
PLASMA_SSE_RETRY | int | reconnect to the source milliseconds after each connection is closed | 2000 | |
PLASMA_SSE_EVENTQUERY | string | use as a querystring in SSE | eventType | ex) /?eventType=program |
PLASMA_SUBSCRIBER_TYPE | string | subscriber type | mock | support "mock" and "redis" |
PLASMA_SUBSCRIBER_REDIS_ADDR | string | Redis address including port number | localhost:6379 | |
PLASMA_SUBSCRIBER_REDIS_PASSWORD | string | Redis password | ||
PLASMA_SUBSCRIBER_REDIS_DB | int | Redis DB | 0 | |
PLASMA_SUBSCRIBER_REDIS_CHANNELS | string | channels of Redis to subscribe (multiple specifications possible) | ||
PLASMA_SUBSCRIBER_REDIS_OVER_MAX_RETRY_BEHAVIOR | string | Behavior of plasma when the number of retries connecting to Redis exceeds the maximum | "die" or "alive" | |
PLASMA_SUBSCRIBER_REDIS_TIMEOUT | time.Duration | timeout for receive message from Redis | 1s | |
PLASMA_SUBSCRIBER_REDIS_RETRY_INTERVAL | time.Duration | interval for retry to receive message from Redis | 5s | |
PLASMA_ERROR_LOG_OUT | string | log file path | stdout, stderr, filepath | |
PLASMA_ERROR_LOG_LEVEL | string | log output level | panic,fatal,error,warn,info,debug | |
PLASMA_ACCESS_LOG_OUT | string | log file path | stdout, stderr, filepath | |
PLASMA_ACCESS_LOG_LEVEL | string | log output level | panic,fatal,error,warn,info,debug | |
PLASMA_TLS_CERT_FILE | string | cert file path | TLS is enabled only when you set both PLASMA_TLS_CERT_FILE and PLASMA_TLS_KEY_FILE | |
PLASMA_TLS_KEY_FILE | string | key file path | ||
PLASMA_METRICS_TYPE | string | metrics type | support "log" or "syslog". if this value is empty, metrics will be disabled | |
PLASMA_METRICS_INTERVAL | time.Duration | interval for update metrics | 10s | |
PLASMA_METRICS_LOG_OUT | string | log file path | stdout | |
PLASMA_METRICS_LOG_PREFIX | string | log prefix | metrics | |
PLASMA_METRICS_LOG_FLAG | int | define which text to prefix to each log entry generated by the Logger | log.Lmicroseconds | https://golang.org/pkg/log/#pkg-constants |
PLASMA_METRICS_LOG_INTERVAL | time.Duration | interval for send to logger | 1m | |
PLASMA_METRICS_SYSLOG_TAG | string | tag for syslog | plasma | |
PLASMA_METRICS_SYSLOG_INTERVAL | time.Duration | interval for send to syslog | 1m | |
PLASMA_METRICS_SYSLOG_SEVERITY | int | syslog serverity | 0 | https://golang.org/pkg/log/syslog/#Priority |
PLASMA_METRICS_SYSLOG_FACILITY | int | syslog facility | 0 | https://golang.org/pkg/log/syslog/#Priority |
PLASMA_METRICS_SYSLOG_NETWORDK | string | network for syslog | ||
PLASMA_METRICS_SYSLOG_ADDR | string | address for syslog |
License
See LICENSE.
Copyright ยฉ CyberAgent, Inc. All Rights Reserved.