mya/okit
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Compare commits

base: mya:2653a917470a34a23c3e187e6467b0adda095c25
Branches Tags
mya:main
..
compare: mya:d3d7d271232d834183862bcc209341f793f38f00
Branches Tags
mya:main

No commits in common. "2653a917470a34a23c3e187e6467b0adda095c25" and "d3d7d271232d834183862bcc209341f793f38f00" have entirely different histories.
2653a91747 ... d3d7d27123

17 changed files with 344 additions and 547 deletions
Show Stats Expand all files Collapse all files

2
Makefile
View File

@ -1,7 +1,7 @@
CWD = $(shell pwd)
define HELP_TEXT
Welcome to okit!
Welcome to obkit!
Targets:
help provides help text

8
README.md
View File

@ -2,9 +2,6 @@
Short for "observability kit", `okit` aims to provide an all-in-one solution to application observability.
_**DISCLAIMER!** This is an early preview. There are significant portions of this repository that still need to be
implemented. Check out the [issues](https://github.com/mjpitz/okit/issues) to see how you can help._
## Why
Traditional approaches to observability treat logging, application metrics, and tracing as independent operations, with
@ -58,8 +55,9 @@ func main() {
okit.Emit("user_signup", tags...)
// Tracing
ctx := context.Background()
defer okit.Trace(&ctx, tags...).Done()
ctx, done := okit.Trace(context.Background(), tags...)
defer done()
_ = ctx.Err() // not needed, removes unused error
// Logging
okit.Debug("a message used for debugging", tags...)

104
api.go
View File

@ -22,61 +22,9 @@ package okit
import (
"context"
"fmt"
"strings"
)
const (
// LevelTrace is used to enable bookend events for tracing.
LevelTrace Level = iota
// LevelDebug enables debug level logging for the application.
LevelDebug
// LevelInfo is the default level of logging for applications.
LevelInfo
// LevelWarn decreases log verbosity by removing informational messages from the stream.
LevelWarn
// LevelError further decreases log verbosity by only surfacing errors encountered by the system.
LevelError
)
// Level allows the logging level to be configured.
type Level uint8
// Set allows a given log level to be set to a certain value.
func (l *Level) Set(val string) error {
switch val {
case "trace", "TRACE":
*l = LevelTrace
case "debug", "DEBUG":
*l = LevelTrace
case "info", "INFO":
*l = LevelTrace
case "warn", "WARN":
*l = LevelTrace
case "error", "ERROR":
*l = LevelTrace
}
return fmt.Errorf("unrecognized level")
}
// String returns the string representation of the Level.
func (l *Level) String() string {
switch *l {
case LevelTrace:
return "trace"
case LevelDebug:
return "debug"
case LevelInfo:
return "info"
case LevelWarn:
return "warn"
case LevelError:
return "error"
}
return "unknown"
}
// Logger defines common operations for writing log messages.
type Logger interface {
// Debug adds a debug message to the event stream, if configured.
@ -107,7 +55,7 @@ type DoneFunc func()
// Tracer allows method calls to be instrumented for debugging.
type Tracer interface {
// Trace captures the method calling context and returns a function that can be invoked to complete the trace.
Trace(ctx *context.Context, tags ...Tag) ActiveTrace
Trace(ctx context.Context, tags ...Tag) (context.Context, DoneFunc)
}
// Span defines an internal structure for tracking traces across an application. This structure is only used for
@ -135,8 +83,48 @@ type Interface[T any] interface {
Wither[T]
}
// ActiveTrace defines an abstraction that allows traces to be completed using a Done function.
type ActiveTrace interface {
// Done finishes the current tracing operation.
Done()
type Level int8
const (
UnknownLevel = 0
TraceLevel = 1
DebugLevel = 2
InfoLevel = 3
WarnLevel = 4
ErrorLevel = 5
)
func (l *Level) String() string {
switch *l {
case TraceLevel:
return "trace"
case DebugLevel:
return "debug"
case InfoLevel:
return "info"
case WarnLevel:
return "warn"
case ErrorLevel:
return "error"
}
return "unknown"
}
func (l *Level) Set(val string) error {
switch {
case strings.EqualFold("trace", val):
*l = TraceLevel
case strings.EqualFold("debug", val):
*l = DebugLevel
case strings.EqualFold("info", val):
*l = InfoLevel
case strings.EqualFold("warn", val):
*l = WarnLevel
case strings.EqualFold("error", val):
*l = ErrorLevel
}
*l = UnknownLevel
return nil
}

50
okit.go → api_impl.go
View File

@ -21,31 +21,19 @@
package okit
import (
"bufio"
"context"
"os"
)
var (
global chan Client
)
// DefaultClient provides a default client implementation.
var DefaultClient = NewClient()
func init() {
global = make(chan Client, 1)
global <- NewClient()
}
// DefaultFormat is used to format how data is written to stdout.
var DefaultFormat Format = TextFormat{}
// Replace updates the default Client to be the provided one.
func Replace(c Client) {
<-global
global <- c
}
// C returns a copy of the current default Client.
func C() Client {
c := <-global
global <- c
return c
}
// TODO: make this configurable
var sink = bufio.NewWriter(os.Stdout)
// ContextKey is a holder structure that allows data to be attached to contexts.
type ContextKey string
@ -55,40 +43,36 @@ var SpanKey = ContextKey("okit.span")
// With returns a client containing additional tags.
func With(tags ...Tag) Client {
return C().With(tags...)
return DefaultClient.WithCallerSkip(3).With(tags...)
}
// Emit emits an event with the provided tags.
func Emit(event string, tags ...Tag) {
C().WithCallerSkip(3).Emit(event, tags...)
DefaultClient.WithCallerSkip(3).Emit(event, tags...)
}
// Observe reports a metric value with the provided tags.
func Observe(metric string, value float64, tags ...Tag) {
C().WithCallerSkip(3).Observe(metric, value, tags...)
DefaultClient.WithCallerSkip(3).Observe(metric, value, tags...)
}
// Trace traces a function call. It implements distributed tracing and bookend events.
func Trace(ctxp *context.Context, tags ...Tag) ActiveTrace {
return C().WithCallerSkip(3).Trace(ctxp, tags...)
func Trace(ctx context.Context, tags ...Tag) (context.Context, DoneFunc) {
return DefaultClient.WithCallerSkip(3).Trace(ctx, tags...)
}
// Debug writes a message at a debug level.
func Debug(msg string, tags ...Tag) {
C().WithCallerSkip(3).Debug(msg, tags...)
DefaultClient.WithCallerSkip(3).Debug(msg, tags...)
}
// Info writes a message at a informational level.
func Info(msg string, tags ...Tag) {
C().WithCallerSkip(3).Info(msg, tags...)
DefaultClient.WithCallerSkip(3).Info(msg, tags...)
}
// Warn writes a message at a warning level.
func Warn(msg string, tags ...Tag) {
C().WithCallerSkip(3).Warn(msg, tags...)
DefaultClient.WithCallerSkip(3).Warn(msg, tags...)
}
// Error writes a message at a error level.
func Error(msg string, tags ...Tag) {
C().WithCallerSkip(3).Error(msg, tags...)
DefaultClient.WithCallerSkip(3).Error(msg, tags...)
}

219
client.go
View File

@ -26,17 +26,18 @@ import (
"encoding/base32"
"io"
"runtime"
"sync"
"time"
"go.pitz.tech/okit/observer"
"go.pitz.tech/okit/pb"
)
var (
encoding = base32.StdEncoding.WithPadding(base32.NoPadding)
var encoding = base32.StdEncoding.WithPadding(base32.NoPadding)
defaultUUID = func() string {
// NewClient produces a new default client implementation for emitting metrics.
func NewClient() Client {
return Client{
now: time.Now,
uuid: func() string {
buf := make([]byte, 16)
n, err := io.ReadFull(rand.Reader, buf)
if err != nil {
@ -44,94 +45,69 @@ var (
}
return encoding.EncodeToString(buf[:n])
}
)
// NewClient produces a new default client implementation for emitting metrics.
func NewClient() Client {
return Client{
observer: observer.Local(),
level: LevelInfo,
now: time.Now,
uuid: defaultUUID,
},
callerSkip: 2,
}
}
// Client provides a default implementation.
type Client struct {
observer observer.Observer
level Level
now func() time.Time
uuid func() string
tags []Tag
callerSkip int
}
// WithObservers configures the new client with the provided observers.
func (c Client) WithObservers(observers ...observer.Observer) Client {
c.observer = observer.Composite(observers)
return c
}
// WithLevel allows the current logging level to be tuned. This does not prevent traces from being emit, or metrics
// from being reported. It only modifies the behavior of logs.
func (c Client) WithLevel(level Level) Client {
c.level = level
return c
}
// WithNow configures the function that's used to obtain the current timestamp.
func (c Client) WithNow(now func() time.Time) Client {
c.now = now
return c
func (o Client) WithNow(now func() time.Time) Client {
o.now = now
return o
}
// WithUUID returns a new uuid that uniquely identifies traces, spans, and tags.
func (c Client) WithUUID(uuid func() string) Client {
c.uuid = uuid
return c
func (o Client) WithUUID(uuid func() string) Client {
o.uuid = uuid
return o
}
// WithCallerSkip is used to configure the number of frames to skip when determining the caller. Callers are
// predominantly used when performing traces.
func (c Client) WithCallerSkip(callerSkip int) Client {
c.callerSkip = callerSkip
return c
func (o Client) WithCallerSkip(callerSkip int) Client {
o.callerSkip = callerSkip
return o
}
// With appends tags to the current client that will automatically be added to all events, metrics, logs, and traces.
func (c Client) With(tags ...Tag) Client {
c.tags = append(c.tags, tags...)
return c
func (o Client) With(tags ...Tag) Client {
o.tags = append(o.tags, tags...)
return o
}
// Emit allows for the emission of an event which has now value and only associated tags.
func (c Client) Emit(event string, tags ...Tag) {
c.With(tags...).emit(&pb.Entry{
func (o Client) Emit(event string, tags ...Tag) {
o.With(tags...).emit(&pb.Entry{
Kind: pb.Kind_Event,
Scope: event,
})
}
// Observe reports a metric and it's associated value.
func (c Client) Observe(metric string, value float64, tags ...Tag) {
c.With(tags...).With(Float64("value", value)).emit(&pb.Entry{
func (o Client) Observe(metric string, value float64, tags ...Tag) {
o.With(tags...).emit(&pb.Entry{
Value: &pb.Entry_Double{Double: value},
Kind: pb.Kind_Metric,
Scope: metric,
})
}
// Trace allows a method to be traced, and it's execution time recorded and reported for viewing.
func (c Client) Trace(ctxp *context.Context, tags ...Tag) ActiveTrace {
ctx := *ctxp
name, _ := caller(c.callerSkip)
start := c.now()
func (o Client) Trace(ctx context.Context, tags ...Tag) (context.Context, DoneFunc) {
name, _ := caller(o.callerSkip)
start := o.now()
span := Span{
TraceID: c.uuid(),
ID: c.uuid(),
TraceID: o.uuid(),
ID: o.uuid(),
}
v := ctx.Value(SpanKey)
@ -144,142 +120,101 @@ func (c Client) Trace(ctxp *context.Context, tags ...Tag) ActiveTrace {
}
//goland:noinspection GoAssignmentToReceiver
c = c.
With(
o = o.With(
append(
[]Tag{
String("traceId", span.TraceID),
String("traceSpanId", span.ID),
String("traceParentId", span.ParentID),
).
With(tags...)
},
tags...,
)...,
)
if c.level <= LevelTrace {
c.With(String("bookend", "start"), level(LevelTrace)).
emit(&pb.Entry{
// bookend
o.With(String("bookend", "start")).emit(&pb.Entry{
Kind: pb.Kind_Log,
Scope: name,
Value: &pb.Entry_String_{String_: "trace"},
})
}
ctx = context.WithValue(ctx, SpanKey, span)
*ctxp = ctx
return context.WithValue(ctx, SpanKey, span), func() {
duration := o.now().Sub(start)
return &active{name, start, c, sync.Once{}}
}
type active struct {
name string
start time.Time
c Client
once sync.Once
}
func (a *active) Done() {
a.once.Do(func() {
duration := a.c.now().Sub(a.start)
//goland:noinspection GoAssignmentToReceiver
a.c = a.c.With(Duration("duration", duration))
if a.c.level <= LevelTrace {
a.c.With(String("bookend", "end"), level(LevelTrace)).
emit(
&pb.Entry{
// bookend
o.With(String("bookend", "end")).emit(&pb.Entry{
Kind: pb.Kind_Log,
Scope: a.name,
Scope: name,
Value: &pb.Entry_String_{String_: "trace"},
})
// trace
o.emit(
&pb.Entry{
Kind: pb.Kind_Trace,
Scope: name,
Value: &pb.Entry_Duration{Duration: pb.DurationPB(duration)},
},
)
}
// trace
a.c.emit(&pb.Entry{
Kind: pb.Kind_Trace,
Scope: a.name,
})
})
}
// Debug produces a debug log event.
func (c Client) Debug(msg string, tags ...Tag) {
if c.level > LevelDebug {
return
}
name, _ := caller(c.callerSkip)
c.With(tags...).
With(message(msg), level(LevelDebug)).
emit(&pb.Entry{
func (o Client) Debug(msg string, tags ...Tag) {
o.With(tags...).emit(&pb.Entry{
Kind: pb.Kind_Log,
Scope: name,
Scope: msg,
Value: &pb.Entry_String_{String_: "debug"},
})
}
// Info produces an information log event.
func (c Client) Info(msg string, tags ...Tag) {
if c.level > LevelInfo {
return
}
name, _ := caller(c.callerSkip)
c.With(tags...).
With(message(msg), level(LevelInfo)).
emit(&pb.Entry{
func (o Client) Info(msg string, tags ...Tag) {
o.With(tags...).emit(&pb.Entry{
Kind: pb.Kind_Log,
Scope: name,
Scope: msg,
Value: &pb.Entry_String_{String_: "info"},
})
}
// Warn produces a warning that is surfaced to operators.
func (c Client) Warn(msg string, tags ...Tag) {
if c.level > LevelWarn {
return
}
name, _ := caller(c.callerSkip)
c.With(tags...).
With(message(msg), level(LevelWarn)).
emit(&pb.Entry{
func (o Client) Warn(msg string, tags ...Tag) {
o.With(tags...).emit(&pb.Entry{
Kind: pb.Kind_Log,
Scope: name,
Scope: msg,
Value: &pb.Entry_String_{String_: "warn"},
})
}
// Error produces a message that communicates an error has occurred.
func (c Client) Error(msg string, tags ...Tag) {
if c.level > LevelError {
return
}
name, _ := caller(c.callerSkip)
c.With(tags...).
With(message(msg), level(LevelError)).
emit(&pb.Entry{
func (o Client) Error(msg string, tags ...Tag) {
o.With(tags...).emit(&pb.Entry{
Kind: pb.Kind_Log,
Scope: name,
Scope: msg,
Value: &pb.Entry_String_{String_: "error"},
})
}
func (c Client) emit(entries ...*pb.Entry) {
now := pb.TimestampPB(c.now())
func (o Client) emit(entries ...*pb.Entry) {
now := pb.TimestampPB(o.now())
tags := make([]*pb.Tag, 0, len(c.tags))
for _, tag := range c.tags {
tags := make([]*pb.Tag, 0, len(o.tags))
for _, tag := range o.tags {
tags = append(tags, tag.AsTagPB())
}
for _, entry := range entries {
entry.Timestamp = now
entry.Tags = tags
_ = DefaultFormat.Marshal(sink, entry)
}
c.observer.Observe(entries)
_ = sink.Flush()
}
// caller attempts to get method caller information. This information is used for tracing information across an
// applications source code. Need to look into the performance of this call. Parts of it may be able to be cached.
// applications source code.
func caller(skip int) (name string, line int) {
pctr, _, line, ok := runtime.Caller(skip)
if !ok {

13
examples/json/main.go
View File

@ -25,20 +25,11 @@ import (
"go.pitz.tech/okit"
okithttp "go.pitz.tech/okit/http"
"go.pitz.tech/okit/observer"
)
func main() {
client := okit.NewClient().
WithObservers(
observer.Local(
observer.Format("json"),
),
// TODO: add remote observers
)
okit.Replace(client)
// Enable JSON output
okit.DefaultFormat = okit.JSONFormat{}
// Instrument HTTP Clients
okithttp.InstrumentClient(http.DefaultClient)

8
examples/overview/main.go
View File

@ -27,9 +27,6 @@ import (
)
func main() {
client := okit.NewClient().WithLevel(okit.LevelTrace)
okit.Replace(client)
var tags []okit.Tag
// Metric emission
@ -42,8 +39,9 @@ func main() {
okit.Emit("user_signup", tags...)
// Tracing
ctx := context.Background()
defer okit.Trace(&ctx, tags...).Done()
ctx, done := okit.Trace(context.Background(), tags...)
defer done()
_ = ctx.Err() // not needed, removes unused error
// Logging
okit.Debug("a message used for debugging", tags...)

61
format.go Normal file
View File

@ -0,0 +1,61 @@
package okit
import (
"bufio"
"strconv"
"strings"
"time"
"github.com/golang/protobuf/jsonpb"
"go.pitz.tech/okit/pb"
)
// Format defines an abstraction for writing entries to stdout/stderr. This is predominantly used in report log, trace,
// and event information from the process.
type Format interface {
Marshal(writer *bufio.Writer, entry *pb.Entry) error
}
// JSONFormat writes entries using JSON encoded protocol buffers.
type JSONFormat struct {
Marshaler jsonpb.Marshaler
}
func (f JSONFormat) Marshal(writer *bufio.Writer, entry *pb.Entry) error {
f.Marshaler.Marshal(writer, entry)
return writer.WriteByte('\n')
}
// TextFormat writes entries using a custom text format.
type TextFormat struct{}
func (f TextFormat) Marshal(writer *bufio.Writer, entry *pb.Entry) error {
sink.WriteString(entry.Timestamp.AsTime().Format(time.RFC3339))
sink.WriteByte('\t')
sink.WriteString(strings.ToUpper(entry.Kind.String()))
sink.WriteByte('\t')
switch v := entry.GetValue().(type) {
case *pb.Entry_String_:
sink.WriteString(v.String_)
case *pb.Entry_Double:
sink.WriteString(strconv.FormatFloat(v.Double, 'f', 5, 64))
case *pb.Entry_Duration:
sink.WriteString(v.Duration.AsDuration().String())
}
sink.WriteByte('\t')
sink.WriteString(entry.Scope)
sink.WriteByte('\t')
for _, tag := range entry.Tags {
qs := tag.AsQueryString()
if qs == "" {
continue
}
sink.WriteString(qs)
sink.WriteByte('&')
}
return sink.WriteByte('\n')
}

71
format/format.go
View File

@ -1,71 +0,0 @@
// Copyright (C) 2022 The OKit Authors
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
// OR OTHER DEALINGS IN THE SOFTWARE.
package format
import (
"bufio"
"strings"
"time"
"github.com/golang/protobuf/jsonpb"
"go.pitz.tech/okit/pb"
)
// Marshaler defines an abstraction for writing entries to stdout/stderr. This is predominantly used in report log, trace,
// and event information from the process.
type Marshaler interface {
Marshal(writer *bufio.Writer, entry *pb.Entry) error
}
// JSON writes entries using JSON encoded protocol buffers.
type JSON struct {
Marshaler jsonpb.Marshaler
}
func (f JSON) Marshal(writer *bufio.Writer, entry *pb.Entry) error {
f.Marshaler.Marshal(writer, entry)
return writer.WriteByte('\n')
}
// Text writes entries using a custom text format. Entries are written using a TSV format and tags are url-encoded
// key-value pairs.
type Text struct{}
func (f Text) Marshal(writer *bufio.Writer, entry *pb.Entry) error {
writer.WriteString(entry.Timestamp.AsTime().Format(time.RFC3339))
writer.WriteByte('\t')
writer.WriteString(strings.ToUpper(entry.Kind.String()))
writer.WriteByte('\t')
writer.WriteString(entry.Scope)
writer.WriteByte('\t')
for _, tag := range entry.Tags {
qs := tag.AsQueryString()
if qs == "" {
continue
}
writer.WriteString(qs)
writer.WriteByte('&')
}
return writer.WriteByte('\n')
}

2
http/endpoint.go
View File

@ -38,7 +38,7 @@ func (e *Endpoint) Health(w http.ResponseWriter, r *http.Request) {
func (e *Endpoint) Metrics(w http.ResponseWriter, r *http.Request) {
// Render metrics endpoint
// Unlike prometheus, metrics are pushed and not scraped. So there's no need for a complicated protocol here.
// Unlike prometheus metrics are pushed, not scraped. So there's no need for complicated protocol here.
}
func (e *Endpoint) Trace(w http.ResponseWriter, r *http.Request) {

48
http/instrumentation.go
View File

@ -21,19 +21,11 @@
package http
import (
"context"
"net/http"
"go.pitz.tech/okit"
)
const (
// OkitTraceIDHeader is a string constant that defines a common key used to propagate an okit trace.
OkitTraceIDHeader = "x-okit-trace-id"
// OkitSpanIDHeader is a string constant that defines a common key used to propagate the parent span.
OkitSpanIDHeader = "x-okit-span-id"
)
// InstrumentClient updates the provided http.Client to use an instrumented http.RoundTripper. If the provided
// client.Transport is nil, then http.DefaultTransport is used.
func InstrumentClient(client *http.Client) {
@ -54,26 +46,16 @@ type roundTripper struct {
}
func (r *roundTripper) RoundTrip(req *http.Request) (resp *http.Response, err error) {
ctx := req.Context()
statusCode := 0
tags := []okit.Tag{
ctx, done := okit.Trace(req.Context(),
okit.String("host", req.URL.Host),
okit.String("method", req.Method),
okit.String("path", req.URL.Path),
okit.Intp("status", &statusCode),
}
defer okit.Trace(&ctx, tags...).Done()
// propagate to server
// TODO: support spec compliant propagation
if v := ctx.Value(okit.SpanKey); v != nil {
if span, ok := v.(okit.Span); ok {
req.Header.Set(OkitTraceIDHeader, span.TraceID)
req.Header.Set(OkitSpanIDHeader, span.ID)
}
}
okit.Errp(&err),
)
defer done()
resp, err = r.delegate.RoundTrip(req.WithContext(ctx))
if err != nil {
@ -97,27 +79,15 @@ type handler struct {
}
func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
// propagation from client
traceID := r.Header.Get(OkitTraceIDHeader)
spanID := r.Header.Get(OkitSpanIDHeader)
if traceID != "" && spanID != "" {
ctx = context.WithValue(ctx, okit.SpanKey, okit.Span{
TraceID: traceID,
ID: spanID,
})
}
statusCode := 200
tags := []okit.Tag{
ctx, done := okit.Trace(r.Context(),
okit.String("host", r.URL.Host),
okit.String("method", r.Method),
okit.String("path", r.URL.Path),
okit.Intp("status", &statusCode),
}
defer okit.Trace(&ctx, tags...).Done()
)
defer done()
h.delegate.ServeHTTP(
&response{

81
observer/local.go
View File

@ -1,81 +0,0 @@
// Copyright (C) 2022 The OKit Authors
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
// OR OTHER DEALINGS IN THE SOFTWARE.
package observer
import (
"bufio"
"io"
"os"
"go.pitz.tech/okit/format"
"go.pitz.tech/okit/pb"
)
// LocalOption defines a mechanism that allows properties of the observer to be overridden or configured.
type LocalOption func(l *local)
// Output configures the output of the local observer to point to the provided writer.
func Output(writer io.Writer) LocalOption {
return func(l *local) {
l.output = bufio.NewWriter(writer)
}
}
// Format configures how the information is written to the target writer.
func Format(fmt string) LocalOption {
return func(l *local) {
switch fmt {
case "json":
l.format = format.JSON{}
default:
l.format = format.Text{}
}
}
}
// Local creates an Observer that writes logs to an output stream using a configured format.
func Local(opts ...LocalOption) Observer {
l := &local{
output: bufio.NewWriter(os.Stdout),
format: format.Text{},
}
for _, o := range opts {
o(l)
}
return l
}
type local struct {
output *bufio.Writer
format format.Marshaler
}
func (l *local) Observe(entries []*pb.Entry) {
// TODO: how to handle internal errors
for _, entry := range entries {
_ = l.format.Marshal(l.output, entry)
}
_ = l.output.Flush()
}

39
observer/observer.go
View File

@ -1,39 +0,0 @@
// Copyright (C) 2022 The OKit Authors
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
// OR OTHER DEALINGS IN THE SOFTWARE.
package observer
import (
"go.pitz.tech/okit/pb"
)
// Observer provides a mechanism to subscribe to entries within the okit system.
type Observer interface {
Observe(entries []*pb.Entry)
}
// Composite provides a way for multiple observers to be plugged into the system.
type Composite []Observer
func (c Composite) Observe(entries []*pb.Entry) {
for _, observer := range c {
observer.Observe(entries)
}
}

119
pb/obkit.pb.go
View File

@ -385,6 +385,13 @@ type Entry struct {
Kind Kind `protobuf:"varint,3,opt,name=kind,proto3,enum=obkit.Kind" json:"kind,omitempty"`
// Tags is an optional field that contains a list of metadata associated with the entry.
Tags []*Tag `protobuf:"bytes,4,rep,name=tags,proto3" json:"tags,omitempty"`
// Value is an optional field that defines a value associated with the entry (e.g. in the case of a log, metric, or trace).
//
// Types that are assignable to Value:
// *Entry_String_
// *Entry_Double
// *Entry_Duration
Value isEntry_Value `protobuf_oneof:"value"`
}
func (x *Entry) Reset() {
@ -447,6 +454,56 @@ func (x *Entry) GetTags() []*Tag {
return nil
}
func (m *Entry) GetValue() isEntry_Value {
if m != nil {
return m.Value
}
return nil
}
func (x *Entry) GetString_() string {
if x, ok := x.GetValue().(*Entry_String_); ok {
return x.String_
}
return ""
}
func (x *Entry) GetDouble() float64 {
if x, ok := x.GetValue().(*Entry_Double); ok {
return x.Double
}
return 0
}
func (x *Entry) GetDuration() *Duration {
if x, ok := x.GetValue().(*Entry_Duration); ok {
return x.Duration
}
return nil
}
type isEntry_Value interface {
isEntry_Value()
}
type Entry_String_ struct {
String_ string `protobuf:"bytes,5,opt,name=string,proto3,oneof"` // used for logs
}
type Entry_Double struct {
Double float64 `protobuf:"fixed64,6,opt,name=double,proto3,oneof"` // used for metrics
}
type Entry_Duration struct {
Duration *Duration `protobuf:"bytes,7,opt,name=duration,proto3,oneof"` // used for traces
}
func (*Entry_String_) isEntry_Value() {}
func (*Entry_Double) isEntry_Value() {}
func (*Entry_Duration) isEntry_Value() {}
// ClientInfo is used to capture specific information about the client application that is emitting information.
type ClientInfo struct {
state protoimpl.MessageState
@ -590,7 +647,7 @@ var file_obkit_proto_rawDesc = []byte{
0x61, 0x6d, 0x70, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x6f, 0x62, 0x6b, 0x69,
0x74, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x48, 0x00, 0x52, 0x09, 0x74,
0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x42, 0x07, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75,
0x65, 0x22, 0x8e, 0x01, 0x0a, 0x05, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x2e, 0x0a, 0x09, 0x74,
0x65, 0x22, 0xfa, 0x01, 0x0a, 0x05, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x2e, 0x0a, 0x09, 0x74,
0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x10,
0x2e, 0x6f, 0x62, 0x6b, 0x69, 0x74, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x14, 0x0a, 0x05, 0x73,
@ -599,23 +656,29 @@ var file_obkit_proto_rawDesc = []byte{
0x0b, 0x2e, 0x6f, 0x62, 0x6b, 0x69, 0x74, 0x2e, 0x4b, 0x69, 0x6e, 0x64, 0x52, 0x04, 0x6b, 0x69,
0x6e, 0x64, 0x12, 0x1e, 0x0a, 0x04, 0x74, 0x61, 0x67, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b,
0x32, 0x0a, 0x2e, 0x6f, 0x62, 0x6b, 0x69, 0x74, 0x2e, 0x54, 0x61, 0x67, 0x52, 0x04, 0x74, 0x61,
0x67, 0x73, 0x22, 0x3c, 0x0a, 0x0a, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49, 0x6e, 0x66, 0x6f,
0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65,
0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65,
0x22, 0x5b, 0x0a, 0x06, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x29, 0x0a, 0x06, 0x63, 0x6c,
0x69, 0x65, 0x6e, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x11, 0x2e, 0x6f, 0x62, 0x6b,
0x69, 0x74, 0x2e, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x06, 0x63,
0x6c, 0x69, 0x65, 0x6e, 0x74, 0x12, 0x26, 0x0a, 0x07, 0x65, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73,
0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0c, 0x2e, 0x6f, 0x62, 0x6b, 0x69, 0x74, 0x2e, 0x45,
0x6e, 0x74, 0x72, 0x79, 0x52, 0x07, 0x65, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73, 0x2a, 0x3e, 0x0a,
0x04, 0x4b, 0x69, 0x6e, 0x64, 0x12, 0x0b, 0x0a, 0x07, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e,
0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x10, 0x01, 0x12, 0x0a, 0x0a,
0x06, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x10, 0x02, 0x12, 0x09, 0x0a, 0x05, 0x54, 0x72, 0x61,
0x63, 0x65, 0x10, 0x03, 0x12, 0x07, 0x0a, 0x03, 0x4c, 0x6f, 0x67, 0x10, 0x04, 0x42, 0x1c, 0x5a,
0x1a, 0x63, 0x6f, 0x64, 0x65, 0x2e, 0x70, 0x69, 0x74, 0x7a, 0x2e, 0x74, 0x65, 0x63, 0x68, 0x2f,
0x6d, 0x79, 0x61, 0x2f, 0x6f, 0x6b, 0x69, 0x74, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x33,
0x67, 0x73, 0x12, 0x18, 0x0a, 0x06, 0x73, 0x74, 0x72, 0x69, 0x6e, 0x67, 0x18, 0x05, 0x20, 0x01,
0x28, 0x09, 0x48, 0x00, 0x52, 0x06, 0x73, 0x74, 0x72, 0x69, 0x6e, 0x67, 0x12, 0x18, 0x0a, 0x06,
0x64, 0x6f, 0x75, 0x62, 0x6c, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x01, 0x48, 0x00, 0x52, 0x06,
0x64, 0x6f, 0x75, 0x62, 0x6c, 0x65, 0x12, 0x2d, 0x0a, 0x08, 0x64, 0x75, 0x72, 0x61, 0x74, 0x69,
0x6f, 0x6e, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x6f, 0x62, 0x6b, 0x69, 0x74,
0x2e, 0x44, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x48, 0x00, 0x52, 0x08, 0x64, 0x75, 0x72,
0x61, 0x74, 0x69, 0x6f, 0x6e, 0x42, 0x07, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x3c,
0x0a, 0x0a, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x12, 0x0a, 0x04,
0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65,
0x12, 0x1a, 0x0a, 0x08, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01,
0x28, 0x09, 0x52, 0x08, 0x69, 0x6e, 0x73, 0x74, 0x61, 0x6e, 0x63, 0x65, 0x22, 0x5b, 0x0a, 0x06,
0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x29, 0x0a, 0x06, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74,
0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x11, 0x2e, 0x6f, 0x62, 0x6b, 0x69, 0x74, 0x2e, 0x43,
0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x06, 0x63, 0x6c, 0x69, 0x65, 0x6e,
0x74, 0x12, 0x26, 0x0a, 0x07, 0x65, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03,
0x28, 0x0b, 0x32, 0x0c, 0x2e, 0x6f, 0x62, 0x6b, 0x69, 0x74, 0x2e, 0x45, 0x6e, 0x74, 0x72, 0x79,
0x52, 0x07, 0x65, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73, 0x2a, 0x3e, 0x0a, 0x04, 0x4b, 0x69, 0x6e,
0x64, 0x12, 0x0b, 0x0a, 0x07, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x10, 0x00, 0x12, 0x09,
0x0a, 0x05, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x4d, 0x65, 0x74,
0x72, 0x69, 0x63, 0x10, 0x02, 0x12, 0x09, 0x0a, 0x05, 0x54, 0x72, 0x61, 0x63, 0x65, 0x10, 0x03,
0x12, 0x07, 0x0a, 0x03, 0x4c, 0x6f, 0x67, 0x10, 0x04, 0x42, 0x1c, 0x5a, 0x1a, 0x63, 0x6f, 0x64,
0x65, 0x2e, 0x70, 0x69, 0x74, 0x7a, 0x2e, 0x74, 0x65, 0x63, 0x68, 0x2f, 0x6d, 0x79, 0x61, 0x2f,
0x6f, 0x6b, 0x69, 0x74, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
@ -647,13 +710,14 @@ var file_obkit_proto_depIdxs = []int32{
2, // 2: obkit.Entry.timestamp:type_name -> obkit.Timestamp
0, // 3: obkit.Entry.kind:type_name -> obkit.Kind
3, // 4: obkit.Entry.tags:type_name -> obkit.Tag
5, // 5: obkit.Packet.client:type_name -> obkit.ClientInfo
4, // 6: obkit.Packet.entries:type_name -> obkit.Entry
7, // [7:7] is the sub-list for method output_type
7, // [7:7] is the sub-list for method input_type
7, // [7:7] is the sub-list for extension type_name
7, // [7:7] is the sub-list for extension extendee
0, // [0:7] is the sub-list for field type_name
1, // 5: obkit.Entry.duration:type_name -> obkit.Duration
5, // 6: obkit.Packet.client:type_name -> obkit.ClientInfo
4, // 7: obkit.Packet.entries:type_name -> obkit.Entry
8, // [8:8] is the sub-list for method output_type
8, // [8:8] is the sub-list for method input_type
8, // [8:8] is the sub-list for extension type_name
8, // [8:8] is the sub-list for extension extendee
0, // [0:8] is the sub-list for field type_name
}
func init() { file_obkit_proto_init() }
@ -744,6 +808,11 @@ func file_obkit_proto_init() {
(*Tag_Duration)(nil),
(*Tag_Timestamp)(nil),
}
file_obkit_proto_msgTypes[3].OneofWrappers = []interface{}{
(*Entry_String_)(nil),
(*Entry_Double)(nil),
(*Entry_Duration)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{

6
pb/obkit.pbext.go
View File

@ -29,24 +29,20 @@ import (
//go:generate protoc --go_out=paths=source_relative:. -I=. obkit.proto
// AsDuration converts a protocol level Duration to a Golang time.Duration.
func (x *Duration) AsDuration() time.Duration {
return time.Duration(x.Nanos)
}
// DurationPB converts a time.Duration to a protocol level Duration.
func DurationPB(d time.Duration) *Duration {
return &Duration{
Nanos: int64(d),
}
}
// AsTime converts a protocol level Timestamp to a Golang time.Time.
func (x *Timestamp) AsTime() time.Time {
return time.Unix(x.Seconds, int64(x.Nanos))
}
// TimestampPB converts a time.Time to a protocol level Timestamp.
func TimestampPB(t time.Time) *Timestamp {
return &Timestamp{
Seconds: t.Unix(),
@ -54,7 +50,7 @@ func TimestampPB(t time.Time) *Timestamp {
}
}
// AsQueryString converts the associated tag to a URI query string.
// AsQueryString converts the associated tag to a URI query string
func (x *Tag) AsQueryString() string {
val := ""

14
pb/obkit.proto
View File

@ -79,6 +79,13 @@ message Entry {
// Tags is an optional field that contains a list of metadata associated with the entry.
repeated Tag tags = 4;
// Value is an optional field that defines a value associated with the entry (e.g. in the case of a log, metric, or trace).
oneof value {
string string = 5; // used for logs
double double = 6; // used for metrics
Duration duration = 7; // used for traces
}
}
// ClientInfo is used to capture specific information about the client application that is emitting information.
@ -93,12 +100,9 @@ message ClientInfo {
// Packet defines the set of data that is contained within each message sent to the server.
message Packet {
// Id provides an id for the packet so the server can deduplicate packets of information being sent to them.
bytes id = 1;
// Client contains information that identifies the emitting client.
ClientInfo client = 2;
ClientInfo client = 1;
// Entries are processed in batches, allowing local clients to buffer if necessary.
repeated Entry entries = 3;
repeated Entry entries = 2;
}

6
tags.ext.go
View File

@ -25,9 +25,3 @@ func Err(err error) Tag { return Tag{"err", err} }
// Errp returns a Tag whose value is read after submission.
func Errp(err *error) Tag { return Tag{"err", err} }
// level returns a tag that contains the string representation of a level.
func level(level Level) Tag { return Tag{"level", (&level).String()} }
// message returns a tag containing message string to index.
func message(msg string) Tag { return Tag{"msg", msg} }