-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: Refactor Go architecture (#43)
* refactor: Refactor Go architecture Refactoring Go SDK to be closer to the API and internal representation that we have with the Rust SDK. Just doing Datadog at the moment and seeking some feedback then i'll port the others and add docs and tests. * fix code formatting * fix formatting * combine new and init * remove vistigial parts of adapter struct * Add back and refactor other adapters * move some more out of trace_ctx * Add 128bit support to TelemetryId, add comments * fix datadog adapter * apply tid changes from comments * Fix otel adapter memory grow events * fix: include memory grow events as allocation attributes --------- Co-authored-by: Steve Manuel <[email protected]>
- Loading branch information
Showing
16 changed files
with
523 additions
and
543 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,125 +1,63 @@ | ||
| package observe | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "errors" | ||
| "fmt" | ||
| "math/rand" | ||
| "time" | ||
| "context" | ||
|
|
||
| "github.com/tetratelabs/wabin/leb128" | ||
| "github.com/tetratelabs/wabin/wasm" | ||
| "github.com/tetratelabs/wazero" | ||
| ) | ||
|
|
||
| // The primary interface that every Adapter needs to follow | ||
| // Start() and Stop() can just call the implementations on AdapterBase | ||
| // or provide some custom logic. HandleTraceEvent is called after | ||
| // an invocation of a wasm module is done and all events are collected. | ||
| type Adapter interface { | ||
| Start(collector *Collector, wasm []byte) error | ||
| Stop(collector *Collector) | ||
| Event(Event) | ||
| Start() | ||
| Stop() | ||
| HandleTraceEvent(TraceEvent) | ||
| } | ||
|
|
||
| type AdapterBase struct { | ||
| Collectors map[*Collector]chan bool | ||
| // The payload that contains all the Events | ||
| // from a single wasm module invocation | ||
| type TraceEvent struct { | ||
| Events []Event | ||
| TelemetryId TelemetryId | ||
| } | ||
|
|
||
| func checkVersion(m *wasm.Module) error { | ||
| var minorGlobal *wasm.Export = nil | ||
| var majorGlobal *wasm.Export = nil | ||
| for _, export := range m.ExportSection { | ||
| if export.Type != wasm.ExternTypeGlobal { | ||
| continue | ||
| } | ||
|
|
||
| if export.Name == "wasm_instr_version_minor" { | ||
| minorGlobal = export | ||
| } else if export.Name == "wasm_instr_version_major" { | ||
| majorGlobal = export | ||
| } | ||
| } | ||
|
|
||
| if minorGlobal == nil || majorGlobal == nil { | ||
| return errors.New("wasm_instr_version functions not found") | ||
| } | ||
|
|
||
| minor, _, err := leb128.DecodeUint32(bytes.NewReader(m.GlobalSection[minorGlobal.Index].Init.Data)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| major, _, err := leb128.DecodeUint32(bytes.NewReader(m.GlobalSection[majorGlobal.Index].Init.Data)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if major != wasmInstrVersionMajor || minor < wasmInstrVersionMinor { | ||
| return errors.New(fmt.Sprintf("Expected instrumentation version >= %d.%d but got %d.%d", wasmInstrVersionMajor, wasmInstrVersionMinor, major, minor)) | ||
| } | ||
|
|
||
| return nil | ||
| // Shared implementation for all Adapters | ||
| type AdapterBase struct { | ||
| TraceEvents chan TraceEvent | ||
| stop chan bool | ||
| } | ||
|
|
||
| func (a *AdapterBase) Wait(collector *Collector, timeout time.Duration, callback func()) { | ||
| for { | ||
| select { | ||
| case <-time.After(timeout): | ||
| if len(collector.Events) > 0 { | ||
| if callback != nil { | ||
| callback() | ||
| } | ||
| continue | ||
| } | ||
| a.RemoveCollector(collector) | ||
| return | ||
| } | ||
| func (a *AdapterBase) NewTraceCtx(ctx context.Context, r wazero.Runtime, wasm []byte, config *Config) (*TraceCtx, error) { | ||
| if config == nil { | ||
| config = NewDefaultConfig() | ||
| } | ||
| return newTraceCtx(ctx, a, r, wasm, config) | ||
| } | ||
|
|
||
| func NewAdapterBase() AdapterBase { | ||
| a := AdapterBase{ | ||
| Collectors: map[*Collector]chan bool{}, | ||
| } | ||
| return a | ||
| } | ||
|
|
||
| func (a *AdapterBase) Start(collector *Collector, wasm []byte) error { | ||
| a.Collectors[collector] = make(chan bool, 1) | ||
| return collector.GetNames(wasm) | ||
| } | ||
|
|
||
| func (a *AdapterBase) RemoveCollector(collector *Collector) { | ||
| delete(a.Collectors, collector) | ||
| } | ||
|
|
||
| func (a *AdapterBase) Stop(collector *Collector) { | ||
| stop, ok := a.Collectors[collector] | ||
| if ok { | ||
| stop <- true | ||
| a.RemoveCollector(collector) | ||
| return AdapterBase{ | ||
| // TODO set to some kind of max, add dump logic | ||
| TraceEvents: make(chan TraceEvent, 100), | ||
| } | ||
| } | ||
|
|
||
| func (a AdapterBase) StopChan(collector *Collector) chan bool { | ||
| return a.Collectors[collector] | ||
| } | ||
|
|
||
| type TelemetryId uint64 | ||
|
|
||
| var rng rand.Source | ||
| func (b *AdapterBase) Start(a Adapter) { | ||
| b.stop = make(chan bool) | ||
|
|
||
| func init() { | ||
| rng = rand.NewSource(time.Now().UnixNano()) | ||
| } | ||
|
|
||
| func NewTraceId() TelemetryId { | ||
| return TelemetryId(rng.Int63()) | ||
| } | ||
|
|
||
| func NewSpanId() TelemetryId { | ||
| return TelemetryId(rng.Int63()) | ||
| } | ||
|
|
||
| func (t TelemetryId) ToHex8() string { | ||
| return fmt.Sprintf("%016x", t) | ||
| go func() { | ||
| for { | ||
| select { | ||
| case event := <-b.TraceEvents: | ||
| a.HandleTraceEvent(event) | ||
| case <-b.stop: | ||
| return | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| func (t TelemetryId) ToHex16() string { | ||
| return fmt.Sprintf("%032x", t) | ||
| func (b *AdapterBase) Stop() { | ||
| b.stop <- true | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.