sour-is/ev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: opentelemetry for tracing and metrics

Browse Source
This commit is contained in:
Jon Lundy 2022-08-13 13:34:13 -06:00
parent 868d41de9f
commit 72d07276cd
Signed by untrusted user who does not match committer: xuu
GPG Key ID: C63E6D61F3035024
15 changed files with 1245 additions and 1655 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
api/gql_ev/resolver.go
View File

@ -5,24 +5,47 @@ import (
"fmt"
"time"
"github.com/sour-is/ev/internal/logz"
"github.com/sour-is/ev/pkg/es"
"github.com/sour-is/ev/pkg/msgbus"
"go.opentelemetry.io/otel/metric/instrument/syncint64"
"go.uber.org/multierr"
)
// This file will not be regenerated automatically.
//
// It serves as dependency injection for your app, add any dependencies you require here.
type Resolver struct {
es *es.EventStore
Mresolver_posts syncint64.Counter
Mresolver_post_added syncint64.Counter
Mresolver_post_added_event syncint64.Counter
}
func New(es *es.EventStore) *Resolver {
return &Resolver{es}
func New(es *es.EventStore) (*Resolver, error) {
m := logz.Meter(context.Background())
var errs error
Mresolver_posts, err := m.SyncInt64().Counter("resolver_posts")
errs = multierr.Append(errs, err)
Mresolver_post_added, err := m.SyncInt64().Counter("resolver_post_added")
errs = multierr.Append(errs, err)
Mresolver_post_added_event, err := m.SyncInt64().Counter("resolver_post_added")
errs = multierr.Append(errs, err)
return &Resolver{
es,
Mresolver_posts,
Mresolver_post_added,
Mresolver_post_added_event,
}, errs
}
// Posts is the resolver for the events field.
func (r *Resolver) Posts(ctx context.Context, streamID string, paging *PageInput) (*Connection, error) {
r.Mresolver_posts.Add(ctx, 1)
lis, err := r.es.Read(ctx, streamID, paging.GetIdx(0), paging.GetCount(30))
if err != nil {
return nil, err
@ -66,6 +89,8 @@ func (r *Resolver) Posts(ctx context.Context, streamID string, paging *PageInput
}
func (r *Resolver) PostAdded(ctx context.Context, streamID string, after int64) (<-chan *PostEvent, error) {
r.Mresolver_post_added.Add(ctx, 1)
es := r.es.EventStream()
if es == nil {
return nil, fmt.Errorf("EventStore does not implement streaming")
@ -90,6 +115,8 @@ func (r *Resolver) PostAdded(ctx context.Context, streamID string, after int64)
if err != nil {
break
}
r.Mresolver_post_added_event.Add(ctx, int64(len(events)))
for _, e := range events {
m := e.EventMeta()
if p, ok := e.(*msgbus.PostEvent); ok {

22
go.mod
View File

@ -6,18 +6,18 @@ require (
github.com/99designs/gqlgen v0.17.13
github.com/go-logr/stdr v1.2.2
github.com/gorilla/websocket v1.5.0
github.com/logzio/go-metrics-sdk v1.0.0
github.com/logzio/logzio-go v1.0.6
github.com/ravilushqa/otelgqlgen v0.9.0
github.com/rs/cors v1.8.2
github.com/tidwall/wal v1.1.7
github.com/vektah/gqlparser/v2 v2.4.7
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.30.0
go.opentelemetry.io/contrib/instrumentation/runtime v0.30.0
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.34.0
go.opentelemetry.io/contrib/instrumentation/runtime v0.34.0
go.opentelemetry.io/otel v1.9.0
go.opentelemetry.io/otel/metric v0.27.0
go.opentelemetry.io/otel/exporters/prometheus v0.31.0
go.opentelemetry.io/otel/metric v0.31.0
go.opentelemetry.io/otel/sdk v1.9.0
go.opentelemetry.io/otel/sdk/metric v0.27.0
go.opentelemetry.io/otel/sdk/metric v0.31.0
go.opentelemetry.io/otel/trace v1.9.0
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
)
@ -25,25 +25,31 @@ require (
require (
github.com/agnivade/levenshtein v1.1.1 // indirect
github.com/beeker1121/goque v2.1.0+incompatible // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cenkalti/backoff/v4 v4.1.3 // indirect
github.com/cespare/xxhash/v2 v2.1.2 // indirect
github.com/felixge/httpsnoop v1.0.3 // indirect
github.com/go-logr/logr v1.2.3 // indirect
github.com/go-ole/go-ole v1.2.6 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/protobuf v1.5.2 // indirect
github.com/golang/snappy v0.0.4 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0 // indirect
github.com/hashicorp/golang-lru v0.5.4 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
github.com/onsi/ginkgo v1.14.0 // indirect
github.com/onsi/gomega v1.10.3 // indirect
github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
github.com/prometheus/prometheus v1.8.2-0.20210928085443-fafb309d4027 // indirect
github.com/prometheus/client_golang v1.12.2 // indirect
github.com/prometheus/client_model v0.2.0 // indirect
github.com/prometheus/common v0.32.1 // indirect
github.com/prometheus/procfs v0.7.3 // indirect
github.com/shirou/gopsutil/v3 v3.22.3 // indirect
github.com/syndtr/goleveldb v1.0.0 // indirect
github.com/yusufpapurcu/wmi v1.2.2 // indirect
go.opentelemetry.io/contrib v1.9.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.9.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.9.0 // indirect
go.opentelemetry.io/otel/internal/metric v0.27.0 // indirect
go.opentelemetry.io/proto/otlp v0.18.0 // indirect
go.uber.org/atomic v1.9.0 // indirect
golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f // indirect

1382
go.sum
View File

File diff suppressed because it is too large Load Diff

28
internal/logz/init.go Normal file
View File

@ -0,0 +1,28 @@
package logz
import (
"context"
"log"
"go.uber.org/multierr"
)
func Init(ctx context.Context, name string) (context.Context, func() error) {
stop := [3]func() error{
initLogger(name),
}
ctx, stop[1] = initMetrics(ctx, name)
ctx, stop[2] = initTracing(ctx, name)
reverse(stop[:])
return ctx, func() error {
log.Println("flushing logs...")
errs := make([]error, len(stop))
for i, fn := range stop {
errs[i] = fn()
}
log.Println("all stopped.")
return multierr.Combine(errs...)
}
}

111
internal/logz/logger.go Normal file
View File

@ -0,0 +1,111 @@
package logz
import (
"bytes"
"encoding/json"
"io"
"log"
"os"
"runtime/debug"
"strings"
"time"
"github.com/go-logr/stdr"
"github.com/logzio/logzio-go"
"go.opentelemetry.io/otel"
)
type logzwriter struct {
name string
pkg string
goversion string
hostname string
w io.Writer
}
func (l *logzwriter) Write(b []byte) (int, error) {
i := 0
for _, sp := range bytes.Split(b, []byte("\n")) {
msg := struct {
Message string `json:"message"`
Host string `json:"host"`
GoVersion string `json:"go_version"`
Package string `json:"pkg"`
App string `json:"app"`
}{
Message: strings.TrimSpace(string(sp)),
Host: l.hostname,
GoVersion: l.goversion,
Package: l.pkg,
App: l.name,
}
if msg.Message == "" || strings.HasPrefix(msg.Message, "#") {
continue
}
b, err := json.Marshal(msg)
if err != nil {
return 0, err
}
j, err := l.w.Write(b)
i += j
if err != nil {
return i, err
}
}
return i, nil
}
func initLogger(name string) func() error {
log.SetPrefix("[" + name + "] ")
log.SetFlags(log.LstdFlags&^(log.Ldate|log.Ltime) | log.Lshortfile)
token := env("LOGZIO_LOG_TOKEN", "")
if token == "" {
return nil
}
l, err := logzio.New(
token,
// logzio.SetDebug(os.Stderr),
logzio.SetUrl(env("LOGZIO_LOG_URL", "https://listener.logz.io:8071")),
logzio.SetDrainDuration(time.Second*5),
logzio.SetTempDirectory(env("LOGZIO_DIR", os.TempDir())),
logzio.SetCheckDiskSpace(true),
logzio.SetDrainDiskThreshold(70),
)
if err != nil {
return nil
}
w := io.MultiWriter(os.Stderr, lzw(l, name))
log.SetOutput(w)
otel.SetLogger(stdr.New(log.Default()))
return func() error {
defer log.Println("logger stopped")
log.SetOutput(os.Stderr)
l.Stop()
return nil
}
}
func lzw(l io.Writer, name string) io.Writer {
lz := &logzwriter{
name: name,
w: l,
}
if info, ok := debug.ReadBuildInfo(); ok {
lz.goversion = info.GoVersion
lz.pkg = info.Path
}
if hostname, err := os.Hostname(); err == nil {
lz.hostname = hostname
}
return lz
}

92
internal/logz/metric.go Normal file
View File

@ -0,0 +1,92 @@
package logz
import (
"context"
"log"
"net/http"
"os"
"runtime/debug"
"time"
// metricsExporter "github.com/logzio/go-metrics-sdk"
"go.opentelemetry.io/contrib/instrumentation/runtime"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/exporters/prometheus"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/metric/global"
"go.opentelemetry.io/otel/sdk/metric/aggregator/histogram"
"go.opentelemetry.io/otel/sdk/metric/controller/basic"
controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
"go.opentelemetry.io/otel/sdk/metric/export/aggregation"
processor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
selector "go.opentelemetry.io/otel/sdk/metric/selector/simple"
"go.opentelemetry.io/otel/sdk/resource"
semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
)
var meterKey = contextKey{"meter"}
var promHTTPKey = contextKey{"promHTTP"}
func Meter(ctx context.Context) metric.Meter {
if t := fromContext[contextKey, metric.Meter](ctx, tracerKey); t != nil {
return t
}
return global.Meter("")
}
func PromHTTP(ctx context.Context) http.Handler {
if t := fromContext[contextKey, *prometheus.Exporter](ctx, promHTTPKey); t != nil {
return t
}
return http.NotFoundHandler()
}
func initMetrics(ctx context.Context, name string) (context.Context, func() error) {
goversion := ""
pkg := ""
host := ""
if info, ok := debug.ReadBuildInfo(); ok {
goversion = info.GoVersion
pkg = info.Path
}
if h, err := os.Hostname(); err == nil {
host = h
}
config := prometheus.Config{}
cont := controller.New(
processor.NewFactory(
selector.NewWithHistogramDistribution(
histogram.WithExplicitBoundaries(config.DefaultHistogramBoundaries),
),
aggregation.CumulativeTemporalitySelector(),
processor.WithMemory(true),
),
basic.WithResource(
resource.NewWithAttributes(
semconv.SchemaURL,
attribute.String("app", name),
attribute.String("host", host),
attribute.String("go_version", goversion),
attribute.String("pkg", pkg),
),
),
)
ex, err := prometheus.New(config, cont)
if err != nil {
return ctx, nil
}
ctx = toContext(ctx, promHTTPKey, ex)
global.SetMeterProvider(cont)
m := cont.Meter(name)
ctx = toContext(ctx, meterKey, m)
runtime.Start()
return ctx, func() error {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
defer log.Println("metrics stopped")
return cont.Stop(ctx)
}
}

146
internal/logz/tracer.go Normal file
View File

@ -0,0 +1,146 @@
package logz
import (
"context"
"fmt"
"log"
"net/http"
"os"
"runtime"
"strconv"
"time"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/resource"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
"go.opentelemetry.io/otel/trace"
)
type contextKey struct {
name string
}
var tracerKey = contextKey{"tracer"}
func Tracer(ctx context.Context) trace.Tracer {
if t := fromContext[contextKey, trace.Tracer](ctx, tracerKey); t != nil {
return t
}
return otel.Tracer("")
}
func Span(ctx context.Context, opts ...trace.SpanStartOption) (context.Context, trace.Span) {
var attrs []attribute.KeyValue
var name string
if pc, file, line, ok := runtime.Caller(1); ok {
if fn := runtime.FuncForPC(pc); fn != nil {
name = fn.Name()
}
attrs = append(attrs,
attribute.String("pc", fmt.Sprintf("%v", pc)),
attribute.String("file", file),
attribute.Int("line", line),
attribute.String("name", name),
)
}
ctx, span := Tracer(ctx).Start(ctx, name, opts...)
span.SetAttributes(attrs...)
return ctx, span
}
func initTracing(ctx context.Context, name string) (context.Context, func() error) {
res, err := resource.New(ctx,
resource.WithAttributes(
semconv.ServiceNameKey.String(name),
),
)
if err != nil {
log.Println(wrap(err, "failed to create trace resource"))
return ctx, nil
}
traceExporter, err := otlptracehttp.New(ctx,
otlptracehttp.WithInsecure(),
otlptracehttp.WithEndpoint("localhost:4318"),
)
if err != nil {
log.Println(wrap(err, "failed to create trace exporter"))
return ctx, nil
}
bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
var sample sdktrace.TracerProviderOption
sampleRate := env("LOGZIO_TRACE_SAMPLE", "always")
switch sampleRate {
case "always":
sample = sdktrace.WithSampler(sdktrace.AlwaysSample())
case "never":
sample = sdktrace.WithSampler(sdktrace.NeverSample())
default:
if v, err := strconv.Atoi(sampleRate); err != nil {
sample = sdktrace.WithSampler(sdktrace.AlwaysSample())
} else {
sample = sdktrace.WithSampler(sdktrace.TraceIDRatioBased(float64(v) * 0.01))
}
}
tracerProvider := sdktrace.NewTracerProvider(
sample,
sdktrace.WithResource(res),
sdktrace.WithSpanProcessor(bsp),
)
otel.SetTracerProvider(tracerProvider)
otel.SetTextMapPropagator(propagation.TraceContext{})
ctx = toContext(ctx, tracerKey, otel.Tracer(name))
return ctx, func() error {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
defer log.Println("tracer stopped")
return wrap(tracerProvider.Shutdown(ctx), "failed to shutdown TracerProvider")
}
}
func wrap(err error, s string) error {
if err != nil {
return fmt.Errorf(s, err)
}
return nil
}
func reverse[T any](s []T) {
first, last := 0, len(s)-1
for first < last {
s[first], s[last] = s[last], s[first]
first++
last--
}
}
func env(name, defaultValue string) string {
if v := os.Getenv(name); v != "" {
log.Println("# ", name, " = ", v)
return v
}
return defaultValue
}
func Htrace(h http.Handler, name string) http.Handler {
return otelhttp.NewHandler(h, name)
}
func toContext[K comparable, V any](ctx context.Context, key K, value V) context.Context {
return context.WithValue(ctx, key, value)
}
func fromContext[K comparable, V any](ctx context.Context, key K) V {
var empty V
if v, ok := ctx.Value(key).(V); ok {
return v
}
return empty
}

256
logger.go
View File

@ -1,256 +0,0 @@
package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"runtime/debug"
"strings"
"time"
metricsExporter "github.com/logzio/go-metrics-sdk"
"github.com/logzio/logzio-go"
"go.uber.org/multierr"
"github.com/go-logr/stdr"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
"go.opentelemetry.io/contrib/instrumentation/runtime"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
"go.opentelemetry.io/otel/metric/global"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/metric/controller/basic"
"go.opentelemetry.io/otel/sdk/resource"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
"go.opentelemetry.io/otel/trace"
)
var tracer trace.Tracer
func Init(ctx context.Context) func() error {
stop := []func() error{
initLogger(),
initMetrics(),
initTracing(ctx),
}
tracer = otel.Tracer(app_name)
reverse(stop)
return func() error {
log.Println("flushing logs...")
errs := make([]error, len(stop))
for i, fn := range stop {
errs[i] = fn()
}
log.Println("all stopped.")
return multierr.Combine(errs...)
}
}
type logzwriter struct {
pkg string
goversion string
hostname string
w io.Writer
}
func (l *logzwriter) Write(b []byte) (int, error) {
i := 0
for _, sp := range bytes.Split(b, []byte("\n")) {
msg := struct {
Message string `json:"message"`
Host string `json:"host"`
GoVersion string `json:"go_version"`
Package string `json:"pkg"`
App string `json:"app"`
}{
Message: strings.TrimSpace(string(sp)),
Host: l.hostname,
GoVersion: l.goversion,
Package: l.pkg,
App: app_name,
}
if msg.Message == "" || strings.HasPrefix(msg.Message, "#") {
continue
}
b, err := json.Marshal(msg)
if err != nil {
return 0, err
}
j, err := l.w.Write(b)
i += j
if err != nil {
return i, err
}
}
return i, nil
}
func initLogger() func() error {
log.SetPrefix("[" + app_name + "] ")
log.SetFlags(log.LstdFlags&^(log.Ldate|log.Ltime) | log.Lshortfile)
token := env("LOGZIO_LOG_TOKEN", "")
if token == "" {
return nil
}
l, err := logzio.New(
token,
// logzio.SetDebug(os.Stderr),
logzio.SetUrl(env("LOGZIO_LOG_URL", "https://listener.logz.io:8071")),
logzio.SetDrainDuration(time.Second*5),
logzio.SetTempDirectory(env("LOGZIO_DIR", os.TempDir())),
logzio.SetCheckDiskSpace(true),
logzio.SetDrainDiskThreshold(70),
)
if err != nil {
return nil
}
w := io.MultiWriter(os.Stderr, lzw(l))
log.SetOutput(w)
otel.SetLogger(stdr.New(log.Default()))
return func() error {
defer log.Println("logger stopped")
log.SetOutput(os.Stderr)
l.Stop()
return nil
}
}
func lzw(l io.Writer) io.Writer {
lz := &logzwriter{
w: l,
}
if info, ok := debug.ReadBuildInfo(); ok {
lz.goversion = info.GoVersion
lz.pkg = info.Path
}
if hostname, err := os.Hostname(); err == nil {
lz.hostname = hostname
}
return lz
}
func initMetrics() func() error {
goversion := ""
pkg := ""
host := ""
if info, ok := debug.ReadBuildInfo(); ok {
goversion = info.GoVersion
pkg = info.Path
}
if h, err := os.Hostname(); err == nil {
host = h
}
config := metricsExporter.Config{
LogzioMetricsListener: env("LOGZIO_METRIC_URL", "https://listener.logz.io:8053"),
LogzioMetricsToken: env("LOGZIO_METRIC_TOKEN", ""),
RemoteTimeout: 30 * time.Second,
PushInterval: 5 * time.Second,
}
if config.LogzioMetricsToken == "" {
return nil
}
// Use the `config` instance from last step.
cont, err := metricsExporter.InstallNewPipeline(
config,
basic.WithCollectPeriod(30*time.Second),
basic.WithResource(
resource.NewWithAttributes(
semconv.SchemaURL,
attribute.String("app", app_name),
attribute.String("host", host),
attribute.String("go_version", goversion),
attribute.String("pkg", pkg),
),
),
)
if err != nil {
return nil
}
global.SetMeterProvider(cont)
runtime.Start()
return func() error {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
defer log.Println("metrics stopped")
return cont.Stop(ctx)
}
}
func initTracing(ctx context.Context) func() error {
res, err := resource.New(ctx,
resource.WithAttributes(
semconv.ServiceNameKey.String("sour.is-ev"),
),
)
if err != nil {
log.Println(wrap(err, "failed to create trace resource"))
return nil
}
traceExporter, err := otlptracehttp.New(ctx,
otlptracehttp.WithInsecure(),
otlptracehttp.WithEndpoint("localhost:4318"),
)
if err != nil {
log.Println(wrap(err, "failed to create trace exporter"))
return nil
}
bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
tracerProvider := sdktrace.NewTracerProvider(
sdktrace.WithSampler(sdktrace.AlwaysSample()),
sdktrace.WithResource(res),
sdktrace.WithSpanProcessor(bsp),
)
otel.SetTracerProvider(tracerProvider)
otel.SetTextMapPropagator(propagation.TraceContext{})
return func() error {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
defer log.Println("tracer stopped")
return wrap(tracerProvider.Shutdown(ctx), "failed to shutdown TracerProvider")
}
}
func wrap(err error, s string) error {
if err != nil {
return fmt.Errorf(s, err)
}
return nil
}
func reverse[T any](s []T) {
first, last := 0, len(s)-1
for first < last {
s[first], s[last] = s[last], s[first]
first++
last--
}
}
func htrace(h http.Handler, name string) http.Handler {
return otelhttp.NewHandler(h, name)
}

28
main.go
View File

@ -12,12 +12,12 @@ import (
"github.com/99designs/gqlgen/graphql/handler"
"github.com/ravilushqa/otelgqlgen"
"github.com/rs/cors"
"go.opentelemetry.io/otel/metric/global"
"golang.org/x/sync/errgroup"
"github.com/sour-is/ev/api/gql_ev"
"github.com/sour-is/ev/internal/graph"
"github.com/sour-is/ev/internal/graph/generated"
"github.com/sour-is/ev/internal/logz"
"github.com/sour-is/ev/pkg/es"
diskstore "github.com/sour-is/ev/pkg/es/driver/disk-store"
memstore "github.com/sour-is/ev/pkg/es/driver/mem-store"
@ -26,7 +26,7 @@ import (
"github.com/sour-is/ev/pkg/playground"
)
const app_name string = "sour.is-ev"
const AppName string = "sour.is-ev"
func main() {
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
@ -35,27 +35,20 @@ func main() {
defer cancel()
}()
stop := Init(ctx)
ctx, stop := logz.Init(ctx, AppName)
defer stop()
ctx, span := tracer.Start(ctx, "main")
defer span.End()
up, err := global.GetMeterProvider().Meter(app_name).NewFloat64UpDownCounter("up")
Mup, err := logz.Meter(ctx).SyncInt64().UpDownCounter("up")
if err != nil {
log.Fatal(err)
}
up.Add(ctx, 1.0)
Mup.Add(ctx, 1)
if err := run(ctx); err != nil {
log.Println(err)
}
}
func run(ctx context.Context) error {
ctx, span := tracer.Start(ctx, "run")
defer span.End()
diskstore.Init(ctx)
memstore.Init(ctx)
@ -69,7 +62,11 @@ func run(ctx context.Context) error {
return err
}
res := graph.New(gql_ev.New(es))
r, err := gql_ev.New(es)
if err != nil {
return err
}
res := graph.New(r)
gql := handler.NewDefaultServer(generated.NewExecutableSchema(generated.Config{Resolvers: res}))
gql.Use(otelgqlgen.Middleware())
@ -79,8 +76,9 @@ func run(ctx context.Context) error {
mux := http.NewServeMux()
mux.Handle("/", playground.Handler("GraphQL playground", "/gql"))
mux.Handle("/gql", htrace(res.ChainMiddlewares(gql), "gql"))
mux.Handle("/inbox/", htrace(http.StripPrefix("/inbox/", svc), "inbox"))
mux.Handle("/gql", logz.Htrace(res.ChainMiddlewares(gql), "gql"))
mux.Handle("/inbox/", logz.Htrace(http.StripPrefix("/inbox/", svc), "inbox"))
mux.Handle("/metrics", logz.PromHTTP(ctx))
wk := http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {

238
pkg/cache/cache.go vendored Normal file
View File

@ -0,0 +1,238 @@
package cache
import (
"context"
"sync"
)
const (
// DefaultEvictedBufferSize defines the default buffer size to store evicted key/val
DefaultEvictedBufferSize = 16
)
// Cache is a thread-safe fixed size LRU cache.
type Cache[K comparable, V any] struct {
lru *LRU[K, V]
evictedKeys []K
evictedVals []V
onEvictedCB func(ctx context.Context, k K, v V)
lock sync.RWMutex
}
// New creates an LRU of the given size.
func NewCache[K comparable, V any](size int) (*Cache[K, V], error) {
return NewWithEvict[K, V](size, nil)
}
// NewWithEvict constructs a fixed size cache with the given eviction
// callback.
func NewWithEvict[K comparable, V any](size int, onEvicted func(context.Context, K, V)) (c *Cache[K, V], err error) {
// create a cache with default settings
c = &Cache[K, V]{
onEvictedCB: onEvicted,
}
if onEvicted != nil {
c.initEvictBuffers()
onEvicted = c.onEvicted
}
c.lru, err = NewLRU(size, onEvicted)
return
}
func (c *Cache[K, V]) initEvictBuffers() {
c.evictedKeys = make([]K, 0, DefaultEvictedBufferSize)
c.evictedVals = make([]V, 0, DefaultEvictedBufferSize)
}
// onEvicted save evicted key/val and sent in externally registered callback
// outside of critical section
func (c *Cache[K, V]) onEvicted(ctx context.Context, k K, v V) {
c.evictedKeys = append(c.evictedKeys, k)
c.evictedVals = append(c.evictedVals, v)
}
// Purge is used to completely clear the cache.
func (c *Cache[K, V]) Purge(ctx context.Context) {
var ks []K
var vs []V
c.lock.Lock()
c.lru.Purge(ctx)
if c.onEvictedCB != nil && len(c.evictedKeys) > 0 {
ks, vs = c.evictedKeys, c.evictedVals
c.initEvictBuffers()
}
c.lock.Unlock()
// invoke callback outside of critical section
if c.onEvictedCB != nil {
for i := 0; i < len(ks); i++ {
c.onEvictedCB(ctx, ks[i], vs[i])
}
}
}
// Add adds a value to the cache. Returns true if an eviction occurred.
func (c *Cache[K, V]) Add(ctx context.Context, key K, value V) (evicted bool) {
var k K
var v V
c.lock.Lock()
evicted = c.lru.Add(ctx, key, value)
if c.onEvictedCB != nil && evicted {
k, v = c.evictedKeys[0], c.evictedVals[0]
c.evictedKeys, c.evictedVals = c.evictedKeys[:0], c.evictedVals[:0]
}
c.lock.Unlock()
if c.onEvictedCB != nil && evicted {
c.onEvictedCB(ctx, k, v)
}
return
}
// Get looks up a key's value from the cache.
func (c *Cache[K, V]) Get(key K) (value *V, ok bool) {
c.lock.Lock()
value, ok = c.lru.Get(key)
c.lock.Unlock()
return value, ok
}
// Contains checks if a key is in the cache, without updating the
// recent-ness or deleting it for being stale.
func (c *Cache[K, V]) Contains(key K) bool {
c.lock.RLock()
containKey := c.lru.Contains(key)
c.lock.RUnlock()
return containKey
}
// Peek returns the key value (or undefined if not found) without updating
// the "recently used"-ness of the key.
func (c *Cache[K, V]) Peek(key K) (value *V, ok bool) {
c.lock.RLock()
value, ok = c.lru.Peek(key)
c.lock.RUnlock()
return value, ok
}
// ContainsOrAdd checks if a key is in the cache without updating the
// recent-ness or deleting it for being stale, and if not, adds the value.
// Returns whether found and whether an eviction occurred.
func (c *Cache[K, V]) ContainsOrAdd(ctx context.Context, key K, value V) (ok, evicted bool) {
var k K
var v V
c.lock.Lock()
if c.lru.Contains(key) {
c.lock.Unlock()
return true, false
}
evicted = c.lru.Add(ctx, key, value)
if c.onEvictedCB != nil && evicted {
k, v = c.evictedKeys[0], c.evictedVals[0]
c.evictedKeys, c.evictedVals = c.evictedKeys[:0], c.evictedVals[:0]
}
c.lock.Unlock()
if c.onEvictedCB != nil && evicted {
c.onEvictedCB(ctx, k, v)
}
return false, evicted
}
// PeekOrAdd checks if a key is in the cache without updating the
// recent-ness or deleting it for being stale, and if not, adds the value.
// Returns whether found and whether an eviction occurred.
func (c *Cache[K, V]) PeekOrAdd(ctx context.Context, key K, value V) (previous interface{}, ok, evicted bool) {
var k K
var v V
c.lock.Lock()
previous, ok = c.lru.Peek(key)
if ok {
c.lock.Unlock()
return previous, true, false
}
evicted = c.lru.Add(ctx, key, value)
if c.onEvictedCB != nil && evicted {
k, v = c.evictedKeys[0], c.evictedVals[0]
c.evictedKeys, c.evictedVals = c.evictedKeys[:0], c.evictedVals[:0]
}
c.lock.Unlock()
if c.onEvictedCB != nil && evicted {
c.onEvictedCB(ctx, k, v)
}
return nil, false, evicted
}
// Remove removes the provided key from the cache.
func (c *Cache[K, V]) Remove(ctx context.Context, key K) (present bool) {
var k K
var v V
c.lock.Lock()
present = c.lru.Remove(ctx, key)
if c.onEvictedCB != nil && present {
k, v = c.evictedKeys[0], c.evictedVals[0]
c.evictedKeys, c.evictedVals = c.evictedKeys[:0], c.evictedVals[:0]
}
c.lock.Unlock()
if c.onEvictedCB != nil && present {
c.onEvicted(ctx, k, v)
}
return
}
// Resize changes the cache size.
func (c *Cache[K, V]) Resize(ctx context.Context, size int) (evicted int) {
var ks []K
var vs []V
c.lock.Lock()
evicted = c.lru.Resize(ctx, size)
if c.onEvictedCB != nil && evicted > 0 {
ks, vs = c.evictedKeys, c.evictedVals
c.initEvictBuffers()
}
c.lock.Unlock()
if c.onEvictedCB != nil && evicted > 0 {
for i := 0; i < len(ks); i++ {
c.onEvictedCB(ctx, ks[i], vs[i])
}
}
return evicted
}
// RemoveOldest removes the oldest item from the cache.
func (c *Cache[K, V]) RemoveOldest(ctx context.Context) (key *K, value *V, ok bool) {
var k K
var v V
c.lock.Lock()
key, value, ok = c.lru.RemoveOldest(ctx)
if c.onEvictedCB != nil && ok {
k, v = c.evictedKeys[0], c.evictedVals[0]
c.evictedKeys, c.evictedVals = c.evictedKeys[:0], c.evictedVals[:0]
}
c.lock.Unlock()
if c.onEvictedCB != nil && ok {
c.onEvictedCB(ctx, k, v)
}
return
}
// GetOldest returns the oldest entry
func (c *Cache[K, V]) GetOldest() (key *K, value *V, ok bool) {
c.lock.RLock()
key, value, ok = c.lru.GetOldest()
c.lock.RUnlock()
return
}
// Keys returns a slice of the keys in the cache, from oldest to newest.
func (c *Cache[K, V]) Keys() []K {
c.lock.RLock()
keys := c.lru.Keys()
c.lock.RUnlock()
return keys
}
// Len returns the number of items in the cache.
func (c *Cache[K, V]) Len() int {
c.lock.RLock()
length := c.lru.Len()
c.lock.RUnlock()
return length
}

235
pkg/cache/list.go vendored Normal file
View File

@ -0,0 +1,235 @@
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Package list implements a doubly linked list.
//
// To iterate over a list (where l is a *List):
//
// for e := l.Front(); e != nil; e = e.Next() {
// // do something with e.Value
// }
package cache
// Element is an element of a linked list.
type Element[V any] struct {
// Next and previous pointers in the doubly-linked list of elements.
// To simplify the implementation, internally a list l is implemented
// as a ring, such that &l.root is both the next element of the last
// list element (l.Back()) and the previous element of the first list
// element (l.Front()).
next, prev *Element[V]
// The list to which this element belongs.
list *List[V]
// The value stored with this element.
Value V
}
// Next returns the next list element or nil.
func (e *Element[V]) Next() *Element[V] {
if p := e.next; e.list != nil && p != &e.list.root {
return p
}
return nil
}
// Prev returns the previous list element or nil.
func (e *Element[V]) Prev() *Element[V] {
if p := e.prev; e.list != nil && p != &e.list.root {
return p
}
return nil
}
// List represents a doubly linked list.
// The zero value for List is an empty list ready to use.
type List[V any] struct {
root Element[V] // sentinel list element, only &root, root.prev, and root.next are used
len int // current list length excluding (this) sentinel element
}
// Init initializes or clears list l.
func (l *List[V]) Init() *List[V] {
l.root.next = &l.root
l.root.prev = &l.root
l.len = 0
return l
}
// NewList returns an initialized list.
func NewList[V any]() *List[V] { return new(List[V]).Init() }
// Len returns the number of elements of list l.
// The complexity is O(1).
func (l *List[V]) Len() int { return l.len }
// Front returns the first element of list l or nil if the list is empty.
func (l *List[V]) Front() *Element[V] {
if l.len == 0 {
return nil
}
return l.root.next
}
// Back returns the last element of list l or nil if the list is empty.
func (l *List[V]) Back() *Element[V] {
if l.len == 0 {
return nil
}
return l.root.prev
}
// lazyInit lazily initializes a zero List value.
func (l *List[V]) lazyInit() {
if l.root.next == nil {
l.Init()
}
}
// insert inserts e after at, increments l.len, and returns e.
func (l *List[V]) insert(e, at *Element[V]) *Element[V] {
e.prev = at
e.next = at.next
e.prev.next = e
e.next.prev = e
e.list = l
l.len++
return e
}
// insertValue is a convenience wrapper for insert(&Element{Value: v}, at).
func (l *List[V]) insertValue(v V, at *Element[V]) *Element[V] {
return l.insert(&Element[V]{Value: v}, at)
}
// remove removes e from its list, decrements l.len
func (l *List[V]) remove(e *Element[V]) {
e.prev.next = e.next
e.next.prev = e.prev
e.next = nil // avoid memory leaks
e.prev = nil // avoid memory leaks
e.list = nil
l.len--
}
// move moves e to next to at.
func (l *List[V]) move(e, at *Element[V]) {
if e == at {
return
}
e.prev.next = e.next
e.next.prev = e.prev
e.prev = at
e.next = at.next
e.prev.next = e
e.next.prev = e
}
// Remove removes e from l if e is an element of list l.
// It returns the element value e.Value.
// The element must not be nil.
func (l *List[V]) Remove(e *Element[V]) any {
if e.list == l {
// if e.list == l, l must have been initialized when e was inserted
// in l or l == nil (e is a zero Element) and l.remove will crash
l.remove(e)
}
return e.Value
}
// PushFront inserts a new element e with value v at the front of list l and returns e.
func (l *List[V]) PushFront(v V) *Element[V] {
l.lazyInit()
return l.insertValue(v, &l.root)
}
// PushBack inserts a new element e with value v at the back of list l and returns e.
func (l *List[V]) PushBack(v V) *Element[V] {
l.lazyInit()
return l.insertValue(v, l.root.prev)
}
// InsertBefore inserts a new element e with value v immediately before mark and returns e.
// If mark is not an element of l, the list is not modified.
// The mark must not be nil.
func (l *List[V]) InsertBefore(v V, mark *Element[V]) *Element[V] {
if mark.list != l {
return nil
}
// see comment in List.Remove about initialization of l
return l.insertValue(v, mark.prev)
}
// InsertAfter inserts a new element e with value v immediately after mark and returns e.
// If mark is not an element of l, the list is not modified.
// The mark must not be nil.
func (l *List[V]) InsertAfter(v V, mark *Element[V]) *Element[V] {
if mark.list != l {
return nil
}
// see comment in List.Remove about initialization of l
return l.insertValue(v, mark)
}
// MoveToFront moves element e to the front of list l.
// If e is not an element of l, the list is not modified.
// The element must not be nil.
func (l *List[V]) MoveToFront(e *Element[V]) {
if e.list != l || l.root.next == e {
return
}
// see comment in List.Remove about initialization of l
l.move(e, &l.root)
}
// MoveToBack moves element e to the back of list l.
// If e is not an element of l, the list is not modified.
// The element must not be nil.
func (l *List[V]) MoveToBack(e *Element[V]) {
if e.list != l || l.root.prev == e {
return
}
// see comment in List.Remove about initialization of l
l.move(e, l.root.prev)
}
// MoveBefore moves element e to its new position before mark.
// If e or mark is not an element of l, or e == mark, the list is not modified.
// The element and mark must not be nil.
func (l *List[V]) MoveBefore(e, mark *Element[V]) {
if e.list != l || e == mark || mark.list != l {
return
}
l.move(e, mark.prev)
}
// MoveAfter moves element e to its new position after mark.
// If e or mark is not an element of l, or e == mark, the list is not modified.
// The element and mark must not be nil.
func (l *List[V]) MoveAfter(e, mark *Element[V]) {
if e.list != l || e == mark || mark.list != l {
return
}
l.move(e, mark)
}
// PushBackList inserts a copy of another list at the back of list l.
// The lists l and other may be the same. They must not be nil.
func (l *List[V]) PushBackList(other *List[V]) {
l.lazyInit()
for i, e := other.Len(), other.Front(); i > 0; i, e = i-1, e.Next() {
l.insertValue(e.Value, l.root.prev)
}
}
// PushFrontList inserts a copy of another list at the front of list l.
// The lists l and other may be the same. They must not be nil.
func (l *List[V]) PushFrontList(other *List[V]) {
l.lazyInit()
for i, e := other.Len(), other.Back(); i > 0; i, e = i-1, e.Prev() {
l.insertValue(e.Value, &l.root)
}
}

175
pkg/cache/lru.go vendored Normal file
View File

@ -0,0 +1,175 @@
package cache
import (
"context"
"errors"
)
// EvictCallback is used to get a callback when a cache entry is evicted
type EvictCallback[K comparable, V any] func(context.Context, K, V)
// LRU implements a non-thread safe fixed size LRU cache
type LRU[K comparable, V any] struct {
size int
evictList *List[entry[K, V]]
items map[K]*Element[entry[K, V]]
onEvict EvictCallback[K, V]
}
// entry is used to hold a value in the evictList
type entry[K comparable, V any] struct {
key K
value V
}
// NewLRU constructs an LRU of the given size
func NewLRU[K comparable, V any](size int, onEvict EvictCallback[K, V]) (*LRU[K, V], error) {
if size <= 0 {
return nil, errors.New("must provide a positive size")
}
c := &LRU[K, V]{
size: size,
evictList: NewList[entry[K, V]](),
items: make(map[K]*Element[entry[K, V]]),
onEvict: onEvict,
}
return c, nil
}
// Purge is used to completely clear the cache.
func (c *LRU[K, V]) Purge(ctx context.Context) {
for k, v := range c.items {
if c.onEvict != nil {
c.onEvict(ctx, k, v.Value.value)
}
delete(c.items, k)
}
c.evictList.Init()
}
// Add adds a value to the cache. Returns true if an eviction occurred.
func (c *LRU[K, V]) Add(ctx context.Context, key K, value V) (evicted bool) {
// Check for existing item
if ent, ok := c.items[key]; ok {
c.evictList.MoveToFront(ent)
ent.Value.value = value
return false
}
// Add new item
entry := c.evictList.PushFront(entry[K, V]{key, value})
c.items[key] = entry
evict := c.evictList.Len() > c.size
// Verify size not exceeded
if evict {
c.removeOldest(ctx)
}
return evict
}
// Get looks up a key's value from the cache.
func (c *LRU[K, V]) Get(key K) (value *V, ok bool) {
if ent, ok := c.items[key]; ok {
c.evictList.MoveToFront(ent)
if ent == nil {
return nil, false
}
return &ent.Value.value, true
}
return
}
// Contains checks if a key is in the cache, without updating the recent-ness
// or deleting it for being stale.
func (c *LRU[K, V]) Contains(key K) (ok bool) {
_, ok = c.items[key]
return ok
}
// Peek returns the key value (or undefined if not found) without updating
// the "recently used"-ness of the key.
func (c *LRU[K, V]) Peek(key K) (value *V, ok bool) {
if ent, ok := c.items[key]; ok {
return &ent.Value.value, true
}
return nil, false
}
// Remove removes the provided key from the cache, returning if the
// key was contained.
func (c *LRU[K, V]) Remove(ctx context.Context, key K) (present bool) {
if ent, ok := c.items[key]; ok {
c.removeElement(ctx, ent)
return true
}
return false
}
// RemoveOldest removes the oldest item from the cache.
func (c *LRU[K, V]) RemoveOldest(ctx context.Context) (key *K, value *V, ok bool) {
ent := c.evictList.Back()
if ent != nil {
c.removeElement(ctx, ent)
kv := ent.Value
return &kv.key, &kv.value, true
}
return nil, nil, false
}
// GetOldest returns the oldest entry
func (c *LRU[K, V]) GetOldest() (key *K, value *V, ok bool) {
ent := c.evictList.Back()
if ent != nil {
kv := ent.Value
return &kv.key, &kv.value, true
}
return nil, nil, false
}
// Keys returns a slice of the keys in the cache, from oldest to newest.
func (c *LRU[K, V]) Keys() []K {
keys := make([]K, len(c.items))
i := 0
for ent := c.evictList.Back(); ent != nil; ent = ent.Prev() {
keys[i] = ent.Value.key
i++
}
return keys
}
// Len returns the number of items in the cache.
func (c *LRU[K, V]) Len() int {
return c.evictList.Len()
}
// Resize changes the cache size.
func (c *LRU[K, V]) Resize(ctx context.Context, size int) (evicted int) {
diff := c.Len() - size
if diff < 0 {
diff = 0
}
for i := 0; i < diff; i++ {
c.removeOldest(ctx)
}
c.size = size
return diff
}
// removeOldest removes the oldest item from the cache.
func (c *LRU[K, V]) removeOldest(ctx context.Context) {
ent := c.evictList.Back()
if ent != nil {
c.removeElement(ctx, ent)
}
}
// removeElement is used to remove a given list element from the cache
func (c *LRU[K, V]) removeElement(ctx context.Context, e *Element[entry[K, V]]) {
c.evictList.Remove(e)
kv := e.Value
delete(c.items, kv.key)
if c.onEvict != nil {
c.onEvict(ctx, kv.key, kv.value)
}
}

64
pkg/es/driver/disk-store/disk-store.go
View File

@ -9,7 +9,11 @@ import (
"strings"
"github.com/tidwall/wal"
"go.opentelemetry.io/otel/metric/instrument/syncint64"
"go.uber.org/multierr"
"github.com/sour-is/ev/internal/logz"
"github.com/sour-is/ev/pkg/cache"
"github.com/sour-is/ev/pkg/es"
"github.com/sour-is/ev/pkg/es/driver"
"github.com/sour-is/ev/pkg/es/event"
@ -17,25 +21,49 @@ import (
"github.com/sour-is/ev/pkg/math"
)
const CachSize = 1000
type lockedWal = locker.Locked[wal.Log]
type openlogs struct {
logs map[string]*locker.Locked[wal.Log]
logs *cache.Cache[string, *lockedWal]
}
type diskStore struct {
path string
openlogs *locker.Locked[openlogs]
Mdisk_open syncint64.Counter
Mdisk_evict syncint64.Counter
}
const AppendOnly = es.AppendOnly
const AllEvents = es.AllEvents
func Init(ctx context.Context) error {
es.Register(ctx, "file", &diskStore{})
return nil
m := logz.Meter(ctx)
var err, errs error
Mdisk_open, err := m.SyncInt64().Counter("disk_open")
errs = multierr.Append(errs, err)
Mdisk_evict, err := m.SyncInt64().Counter("disk_evict")
errs = multierr.Append(errs, err)
es.Register(ctx, "file", &diskStore{
Mdisk_open: Mdisk_open,
Mdisk_evict: Mdisk_evict,
})
return errs
}
var _ driver.Driver = (*diskStore)(nil)
func (diskStore) Open(_ context.Context, dsn string) (driver.Driver, error) {
func (d *diskStore) Open(ctx context.Context, dsn string) (driver.Driver, error) {
ctx, span := logz.Span(ctx)
defer span.End()
d.Mdisk_open.Add(ctx, 1)
scheme, path, ok := strings.Cut(dsn, ":")
if !ok {
return nil, fmt.Errorf("expected scheme")
@ -51,16 +79,34 @@ func (diskStore) Open(_ context.Context, dsn string) (driver.Driver, error) {
return nil, err
}
}
c, err := cache.NewWithEvict(CachSize, func(ctx context.Context, s string, l *lockedWal) {
l.Modify(ctx, func(w *wal.Log) error {
// logz.Mdisk_evict.Add(ctx, 1)
logs := &openlogs{logs: make(map[string]*locker.Locked[wal.Log])}
return &diskStore{path: path, openlogs: locker.New(logs)}, nil
err := w.Close()
if err != nil {
log.Print(err)
}
return nil
})
})
if err != nil {
return nil, err
}
logs := &openlogs{logs: c}
return &diskStore{
path: path,
openlogs: locker.New(logs),
Mdisk_open: d.Mdisk_open,
Mdisk_evict: d.Mdisk_evict,
}, nil
}
func (ds *diskStore) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) {
el := &eventLog{streamID: streamID}
return el, ds.openlogs.Modify(ctx, func(openlogs *openlogs) error {
if events, ok := openlogs.logs[streamID]; ok {
el.events = events
if events, ok := openlogs.logs.Get(streamID); ok {
el.events = *events
return nil
}
@ -69,7 +115,7 @@ func (ds *diskStore) EventLog(ctx context.Context, streamID string) (driver.Even
return err
}
el.events = locker.New(l)
openlogs.logs[streamID] = el.events
openlogs.logs.Add(ctx, streamID, el.events)
return nil
})
}

67
pkg/es/es.go
View File

@ -6,9 +6,12 @@ import (
"fmt"
"strings"
"github.com/sour-is/ev/internal/logz"
"github.com/sour-is/ev/pkg/es/driver"
"github.com/sour-is/ev/pkg/es/event"
"github.com/sour-is/ev/pkg/locker"
"go.opentelemetry.io/otel/metric/instrument/syncint64"
"go.uber.org/multierr"
)
type config struct {
@ -23,20 +26,56 @@ var (
)
func Register(ctx context.Context, name string, d driver.Driver) error {
return drivers.Modify(ctx, func(c *config) error {
ctx, span := logz.Span(ctx)
defer span.End()
m := logz.Meter(ctx)
var err, errs error
Mes_open, err = m.SyncInt64().Counter("es_open")
errs = multierr.Append(errs, err)
Mes_read, err = m.SyncInt64().Counter("es_read")
errs = multierr.Append(errs, err)
Mes_load, err = m.SyncInt64().Counter("es_load")
errs = multierr.Append(errs, err)
Mes_save, err = m.SyncInt64().Counter("es_save")
errs = multierr.Append(errs, err)
Mes_append, err = m.SyncInt64().Counter("es_append")
errs = multierr.Append(errs, err)
err = drivers.Modify(ctx, func(c *config) error {
if _, set := c.drivers[name]; set {
return fmt.Errorf("driver %s already set", name)
}
c.drivers[name] = d
return nil
})
return multierr.Append(errs, err)
}
type EventStore struct {
driver.Driver
}
var (
Mes_open syncint64.Counter
Mes_read syncint64.Counter
Mes_load syncint64.Counter
Mes_save syncint64.Counter
Mes_append syncint64.Counter
)
func Open(ctx context.Context, dsn string, options ...Option) (*EventStore, error) {
ctx, span := logz.Span(ctx)
defer span.End()
Mes_open.Add(ctx, 1)
name, _, ok := strings.Cut(dsn, ":")
if !ok {
return nil, fmt.Errorf("%w: no scheme", ErrNoDriver)
@ -68,6 +107,11 @@ type Option interface {
}
func (es *EventStore) Save(ctx context.Context, agg event.Aggregate) (uint64, error) {
ctx, span := logz.Span(ctx)
defer span.End()
Mes_save.Add(ctx, 1)
l, err := es.EventLog(ctx, agg.StreamID())
if err != nil {
return 0, err
@ -83,6 +127,11 @@ func (es *EventStore) Save(ctx context.Context, agg event.Aggregate) (uint64, er
return count, err
}
func (es *EventStore) Load(ctx context.Context, agg event.Aggregate) error {
ctx, span := logz.Span(ctx)
defer span.End()
Mes_load.Add(ctx, 1)
l, err := es.Driver.EventLog(ctx, agg.StreamID())
if err != nil {
return err
@ -97,6 +146,11 @@ func (es *EventStore) Load(ctx context.Context, agg event.Aggregate) error {
return nil
}
func (es *EventStore) Read(ctx context.Context, streamID string, pos, count int64) (event.Events, error) {
ctx, span := logz.Span(ctx)
defer span.End()
Mes_read.Add(ctx, 1)
l, err := es.Driver.EventLog(ctx, streamID)
if err != nil {
return nil, err
@ -104,6 +158,11 @@ func (es *EventStore) Read(ctx context.Context, streamID string, pos, count int6
return l.Read(ctx, pos, count)
}
func (es *EventStore) Append(ctx context.Context, streamID string, events event.Events) (uint64, error) {
ctx, span := logz.Span(ctx)
defer span.End()
Mes_append.Add(ctx, 1)
l, err := es.Driver.EventLog(ctx, streamID)
if err != nil {
return 0, err
@ -111,6 +170,9 @@ func (es *EventStore) Append(ctx context.Context, streamID string, events event.
return l.Append(ctx, events, AppendOnly)
}
func (es *EventStore) FirstIndex(ctx context.Context, streamID string) (uint64, error) {
ctx, span := logz.Span(ctx)
defer span.End()
l, err := es.Driver.EventLog(ctx, streamID)
if err != nil {
return 0, err
@ -118,6 +180,9 @@ func (es *EventStore) FirstIndex(ctx context.Context, streamID string) (uint64,
return l.FirstIndex(ctx)
}
func (es *EventStore) LastIndex(ctx context.Context, streamID string) (uint64, error) {
ctx, span := logz.Span(ctx)
defer span.End()
l, err := es.Driver.EventLog(ctx, streamID)
if err != nil {
return 0, err

15
pkg/msgbus/service.go
View File

@ -13,6 +13,7 @@ import (
"time"
"github.com/gorilla/websocket"
"github.com/sour-is/ev/internal/logz"
"github.com/sour-is/ev/pkg/es"
"github.com/sour-is/ev/pkg/es/event"
)
@ -22,6 +23,9 @@ type service struct {
}
func New(ctx context.Context, es *es.EventStore) (*service, error) {
ctx, span := logz.Span(ctx)
defer span.End()
if err := event.Register(ctx, &PostEvent{}); err != nil {
return nil, err
}
@ -36,6 +40,11 @@ var upgrader = websocket.Upgrader{
}
func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
ctx, span := logz.Span(ctx)
defer span.End()
r = r.WithContext(ctx)
switch r.Method {
case http.MethodGet:
if r.Header.Get("Upgrade") == "websocket" {
@ -52,6 +61,8 @@ func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
func (s *service) get(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
ctx, span := logz.Span(ctx)
defer span.End()
name, _, _ := strings.Cut(r.URL.Path, "/")
if name == "" {
@ -100,6 +111,8 @@ func (s *service) get(w http.ResponseWriter, r *http.Request) {
}
func (s *service) post(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
ctx, span := logz.Span(ctx)
defer span.End()
name, tags, _ := strings.Cut(r.URL.Path, "/")
if name == "" {
@ -162,6 +175,8 @@ func (s *service) post(w http.ResponseWriter, r *http.Request) {
}
func (s *service) websocket(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
ctx, span := logz.Span(ctx)
defer span.End()
name, _, _ := strings.Cut(r.URL.Path, "/")
if name == "" {