feat: add logs/metrics/tracing

.gitignore

@@ -1,2 +1,6 @@
 .vscode/
 data/
+.gitsecret/keys/random_seed
+!*.secret
+local.mk
+logzio.yml

.gitsecret/paths/mapping.cfg

@@ -0,0 +1,2 @@
+logzio.yml
+local.mk

Makefile

@@ -1,5 +1,6 @@
 export EV_DATA = mem:
 export EV_HTTP = :8080
+-include local.mk
 
 run: gen
 	go run .
@@ -20,7 +21,7 @@ endif
 	gqlgen
 
 load:
-	watch -n .1 "http POST localhost:8080/event/asdf/test a=b one=1 two:='{\"v\":2}' | jq"
+	watch -n .1 "http POST localhost:8080/inbox/asdf/test a=b one=1 two:='{\"v\":2}' | jq"
 
 bi:
 	go build .

go.mod

@@ -5,16 +5,53 @@ go 1.18
 require (
 	github.com/99designs/gqlgen v0.17.13
 	github.com/tidwall/wal v1.1.7
-	github.com/vektah/gqlparser/v2 v2.4.6
+	github.com/vektah/gqlparser/v2 v2.4.7
 	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
 )
 
 require (
 	github.com/agnivade/levenshtein v1.1.1 // indirect
+	github.com/beeker1121/goque v2.1.0+incompatible // indirect
+	github.com/cenkalti/backoff/v4 v4.1.3 // indirect
+	github.com/felixge/httpsnoop v1.0.3 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // 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/gorilla/websocket v1.5.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
-	github.com/mitchellh/mapstructure v1.3.1 // indirect
+	github.com/logzio/go-metrics-sdk v1.0.0 // indirect
+	github.com/logzio/logzio-go v1.0.6 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // 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/ravilushqa/otelgqlgen v0.9.0 // indirect
 	github.com/rs/cors v1.8.2 // 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/contrib/instrumentation/net/http/otelhttp v0.30.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/runtime v0.30.0 // indirect
+	go.opentelemetry.io/otel 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/otel/metric v0.27.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.9.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v0.27.0 // indirect
+	go.opentelemetry.io/otel/trace v1.9.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
+	golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9 // indirect
+	golang.org/x/text v0.3.7 // indirect
+	google.golang.org/genproto v0.0.0-20211118181313-81c1377c94b1 // indirect
+	google.golang.org/grpc v1.46.2 // indirect
+	google.golang.org/protobuf v1.28.0 // indirect
 )
 
 require (
@@ -24,4 +61,5 @@ require (
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/tidwall/tinylru v1.1.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.9.0
 )

logger.go

@@ -0,0 +1,232 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"runtime/debug"
+	"strings"
+	"time"
+
+	metricsExporter "github.com/logzio/go-metrics-sdk"
+	"github.com/logzio/logzio-go"
+
+	"github.com/go-logr/stdr"
+	"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"
+)
+
+func Init(ctx context.Context) {
+	stop := []func() error {
+		initLogger(),
+		initMetrics(),
+		initTracing(ctx),
+	}
+
+	reverse(stop)
+
+	go func() {
+		<-ctx.Done()
+		for _, fn := range stop {
+			if err := fn(); err != nil {
+				log.Println(err)
+			}
+		}
+	}()
+}
+
+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 { 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()
+		return cont.Stop(ctx)
+	}
+}
+
+func initTracing(ctx context.Context) func() error {
+	res, err := resource.New(ctx,
+		resource.WithAttributes(
+			semconv.ServiceNameKey.String("sour.is-ev"),
+		),
+	)
+	log.Println(wrap(err, "failed to create trace resource"))
+
+	traceExporter, err := otlptracehttp.New(ctx,
+		otlptracehttp.WithInsecure(),
+		otlptracehttp.WithEndpoint("localhost:4318"),
+	)
+	log.Println(wrap(err, "failed to create trace exporter"))
+
+	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()
+		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--
+	}
+}

main.go

@@ -10,7 +10,9 @@ import (
 	"time"
 
 	"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"
@@ -24,12 +26,21 @@ import (
 	"github.com/sour-is/ev/pkg/playground"
 )
 
+const app_name string = "sour.is-ev"
+
 func main() {
 	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
 	go func() {
 		<-ctx.Done()
 		defer cancel()
 	}()
+	Init(ctx)
+
+	up, err := global.GetMeterProvider().Meter(app_name).NewFloat64UpDownCounter("up")
+	if err != nil {
+		log.Fatal(err)
+	}
+	up.Add(ctx, 1.0)
 
 	if err := run(ctx); err != nil {
 		log.Fatal(err)
@@ -51,6 +62,7 @@ func run(ctx context.Context) error {
 
 	res := graph.New(gql_ev.New(es))
 	gql := handler.NewDefaultServer(generated.NewExecutableSchema(generated.Config{Resolvers: res}))
+	gql.Use(otelgqlgen.Middleware())
 
 	s := http.Server{
 		Addr: env("EV_HTTP", ":8080"),

pkg/es/driver/streamer/streamer.go

@@ -2,7 +2,6 @@ package streamer
 
 import (
 	"context"
-	"log"
 
 	"github.com/sour-is/ev/pkg/es"
 	"github.com/sour-is/ev/pkg/es/driver"
@@ -56,19 +55,15 @@ func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64)
 		size: es.AllEvents,
 	})
 	sub.unsub = s.delete(streamID, sub)
-	log.Println("start ", sub)
 
 	return sub, s.state.Modify(ctx, func(state *state) error {
 		state.subscribers[streamID] = append(state.subscribers[streamID], sub)
-		log.Println("add ", len(state.subscribers[streamID]))
 		return nil
 	})
 }
 func (s *streamer) Send(ctx context.Context, streamID string, events event.Events) error {
 	return s.state.Modify(ctx, func(state *state) error {
-		log.Println("trigger ", len(state.subscribers[streamID]))
 		for _, sub := range state.subscribers[streamID] {
-			log.Println("trigg ", sub)
 			err := sub.position.Modify(ctx, func(position *position) error {
 				position.size = int64(events.Last().EventMeta().Position - uint64(position.idx))
 
@@ -78,7 +73,9 @@ func (s *streamer) Send(ctx context.Context, streamID string, events event.Event
 				}
 				return nil
 			})
-			if err != nil { return err }
+			if err != nil {
+				return err
+			}
 		}
 		return nil
 	})
@@ -89,7 +86,6 @@ func (s *streamer) delete(streamID string, sub *subscription) func(context.Conte
 		if err := ctx.Err(); err != nil {
 			return err
 		}
-		log.Println("unsub ", s)
 		return s.state.Modify(ctx, func(state *state) error {
 			lis := state.subscribers[streamID]
 			for i := range lis {
@@ -150,8 +146,6 @@ type subscription struct {
 
 func (s *subscription) Recv(ctx context.Context) bool {
 	var wait func(context.Context) bool
-	log.Println("wait ", s)
-	defer log.Println("recv ", s)
 	err := s.position.Modify(ctx, func(position *position) error {
 		if position.size == es.AllEvents {
 			return nil
@@ -187,13 +181,10 @@ func (s *subscription) Events(ctx context.Context) (event.Events, error) {
 	var events event.Events
 	return events, s.position.Modify(ctx, func(position *position) error {
 		var err error
-		log.Println("pos=", position, s)
 		events, err = s.events.Read(ctx, position.idx, position.size)
 		if err != nil {
-			log.Println(err, s)
 			return err
 		}
-		log.Println("events=", len(events), s)
 		position.size = int64(len(events))
 		if len(events) > 0 {
 			position.idx = int64(events.First().EventMeta().Position - 1)

pkg/es/es_test.go

@@ -2,6 +2,7 @@ package es_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -15,6 +16,7 @@ import (
 
 func TestES(t *testing.T) {
 	is := is.New(t)
+
 	ctx := context.Background()
 
 	err := event.Register(ctx, &ValueSet{})
@@ -22,6 +24,12 @@ func TestES(t *testing.T) {
 
 	memstore.Init(ctx)
 
+	_, err = es.Open(ctx, "mem")
+	is.True(errors.Is(err, es.ErrNoDriver))
+
+	_, err = es.Open(ctx, "bogo:")
+	is.True(errors.Is(err, es.ErrNoDriver))
+
 	es, err := es.Open(ctx, "mem:")
 	is.NoErr(err)
 
@@ -40,12 +48,25 @@ func TestES(t *testing.T) {
 	t.Log(thing.StreamVersion(), thing.Name, thing.Value)
 	t.Log("Wrote: ", i)
 
+	i, err = es.Append(ctx, "thing-time", event.NewEvents(&ValueSet{Value: "xxx"}))
+	is.NoErr(err)
+	is.Equal(i, uint64(1))
+
 	events, err := es.Read(ctx, "thing-time", -1, -11)
 	is.NoErr(err)
 
 	for i, e := range events {
 		t.Logf("event %d %d - %v\n", i, e.EventMeta().Position, e)
 	}
+
+	first, err := es.FirstIndex(ctx, "thing-time")
+	is.NoErr(err)
+	is.Equal(first, uint64(1))
+
+	last, err := es.LastIndex(ctx, "thing-time")
+	is.NoErr(err)
+	is.Equal(last, uint64(2))
+
 }
 
 type Thing struct {